Skip to content

Commit

Permalink
Add notification messages framework
Browse files Browse the repository at this point in the history
Now we  only impl the NotifyInvalInode,NotifyInvalEntry,
NotifyDelete.

Notify funcs allow the filesystem to invalidate VFS caches.

The statement that 'In a networked fuse file-system, the local
fuse instance may receive changes from a remote system that do not
pass through the local kernel vfs.
When those changes are made the local fuse instance must notify the
local kernel of the change so as to keep the local kernel's cache
in sync. This is accomplished via notify delete or notify
invalidate.
In other words, these notifications only apply if fuse file-system
makes changes to the file-system that are not initiated by the
local kernel: changes initiated by the local kernel must not
involve notify delete or notify invalidate (will deadlock), and
are not required.' is quoted from Jhon Muir.

About fuse  notify func history ,please ref link
http://lkml.iu.edu/hypermail/linux/kernel/0906.0/01120.html .
  • Loading branch information
inevity committed Nov 5, 2017
1 parent 88e3bc5 commit 3a29c55
Show file tree
Hide file tree
Showing 8 changed files with 321 additions and 2 deletions.
69 changes: 68 additions & 1 deletion connection.go
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,19 @@ func (c *Connection) beginOp(
// should not record any state keyed on their ID.
//
// Cf. https://github.com/osxfuse/osxfuse/issues/208
if opCode != fusekernel.OpForget {

// Special case: For notify ops ,it is issued by the userspace filesystem to
// fuse kernel. The ops's kernel opCode is 0 in kernel' viewpoint or kernel not
// care whether what the iopcode is .But the ops have the notifycode from 1
// to 6 now. For adapt the notfiy ops to the BeginOP ,we fake an opcode as
// notfiycode + 100. Now opcode range of ops from kernel including origin
// from userspace and just from kernel is < 100. The notify ops's opcode
// range is [101-106].
// Then we can process all ops consistently.
// Cf. func (c *Connection) SetNotifyContext(op interface{})
// (context.Context, error) {

if opCode != fusekernel.OpForget && opCode < 100 {
var cancel func()
ctx, cancel = context.WithCancel(ctx)
c.recordCancelFunc(fuseID, cancel)
Expand Down Expand Up @@ -411,6 +423,38 @@ func (c *Connection) ReadOp() (ctx context.Context, op interface{}, err error) {
}
}

// The SetNotifyContext set context according with value of the notify op's
// fuseops.Notify*Op.
func (c *Connection) SetNotifyContext(op interface{}) (context.Context, error) {

outMsg := c.getOutMessage()

err := c.buildNotify(outMsg, op)
if err != nil {
return nil, err
}

ctx := context.Background()

switch op.(type) {
case *fuseops.NotifyInvalInodeOp:
ctx = c.beginOp(100+uint32(fusekernel.NotifyCodeInvalInode), 0)

case *fuseops.NotifyInvalEntryOp:
ctx = c.beginOp(100+uint32(fusekernel.NotifyCodeInvalEntry), 0)

case *fuseops.NotifyDeleteOp:
ctx = c.beginOp(100+uint32(fusekernel.NotifyCodeDelete), 0)

default:
panic(fmt.Sprintf("Unexpected op: %#v", op))
}

ctx = context.WithValue(ctx, contextKey, opState{nil, outMsg, op})
return ctx, nil

}

// Skip errors that happen as a matter of course, since they spook users.
func (c *Connection) shouldLogError(
op interface{},
Expand Down Expand Up @@ -498,6 +542,29 @@ func (c *Connection) Reply(ctx context.Context, opErr error) {
}
}

// The NotifyKernel is same as Reply func of Connection.But the diff is
// that the func only send to kernel.
func (c *Connection) NotifyKernel(ctx context.Context) {

// we should get outmsg from context
var key interface{} = contextKey
foo := ctx.Value(key)
state, ok := foo.(opState)
if !ok {
panic(fmt.Sprintf("Reply called with invalid context: %#v", ctx))
}

outMsg := state.outMsg
defer c.putOutMessage(outMsg)

c.debugLogger.Println("dev fd is:unique:notifycode ", c.dev.Fd(), outMsg.OutHeader().Unique, outMsg.OutHeader().Error)
err := c.writeMessage(outMsg.Bytes())
if err != nil && c.errorLogger != nil {
c.errorLogger.Printf("writeMessage: %v %v", err, outMsg.Bytes())
}

}

// Close the connection. Must not be called until operations that were read
// from the connection have been responded to.
func (c *Connection) close() (err error) {
Expand Down
61 changes: 61 additions & 0 deletions conversions.go
Original file line number Diff line number Diff line change
Expand Up @@ -597,6 +597,59 @@ func (c *Connection) kernelResponse(
return
}

func (c *Connection) buildNotify(
m *buffer.OutMessage,
op interface{}) error {

h := m.OutHeader()
h.Unique = 0
// Create the appropriate output message
switch o := op.(type) {
case *fuseops.NotifyInvalInodeOp:
h.Error = fusekernel.NotifyCodeInvalInode
size := fusekernel.NotifyInvalInodeOutSize
out := (*fusekernel.NotifyInvalInodeOut)(m.Grow(size))
out.Ino = uint64(o.Ino)
out.Off = int64(o.Off)
out.Len = int64(o.Len)

case *fuseops.NotifyInvalEntryOp:
err := checkName(o.Name)
if err != nil {
return err
}
h.Error = fusekernel.NotifyCodeInvalEntry
size := fusekernel.NotifyInvalEntryOutSize
out := (*fusekernel.NotifyInvalEntryOut)(m.Grow(size))
out.Parent = uint64(o.Parent)
out.Namelen = uint32(len(o.Name))
m.Append([]byte(o.Name))
b := []byte{'\x00'}
m.Append(b)

case *fuseops.NotifyDeleteOp:
err := checkName(o.Name)
if err != nil {
return err
}
h.Error = fusekernel.NotifyCodeDelete
size := fusekernel.NotifyDeleteOutSize
out := (*fusekernel.NotifyDeleteOut)(m.Grow(size))
out.Parent = uint64(o.Parent)
out.Child = uint64(o.Child)
out.Namelen = uint32(len(o.Name))
m.Append([]byte(o.Name))
b := []byte{'\x00'}
m.Append(b)

default:
return errors.New("unexpectedop")
}
h.Len = uint32(m.Len())

return nil
}

// Like kernelResponse, but assumes the user replied with a nil error to the
// op.
func (c *Connection) kernelResponseForOp(
Expand Down Expand Up @@ -888,3 +941,11 @@ func writeXattrSize(m *buffer.OutMessage, size uint32) {
out := (*fusekernel.GetxattrOut)(m.Grow(int(unsafe.Sizeof(fusekernel.GetxattrOut{}))))
out.Size = size
}
func checkName(name string) error {
const maxUint32 = ^uint32(0)
if uint64(len(name)) > uint64(maxUint32) {
// very unlikely, but we don't want to silently truncate
return syscall.ENAMETOOLONG
}
return nil
}
59 changes: 59 additions & 0 deletions fuseops/ops.go
Original file line number Diff line number Diff line change
Expand Up @@ -845,3 +845,62 @@ type SetXattrOp struct {
// simply replace the value if the attribute exists.
Flags uint32
}

// Notify IO readiness event
type NotifyPollOp struct {
}

// Notify to invalidate cache for an inode.
// Added in FUSE protocol version 7.12. If the kernel does not support this
// (or a newer) version, the op will return -ENOSYS and do nothing
type NotifyInvalInodeOp struct {
// inode to invalidatej
Ino InodeID

// the offset in the inode where to start invalidating or negative to invalidate attributes only
Off int64

// the amount of cache to invalidate or 0 for all
Len int64
}

// Notify to invalidate parent attributes and the dentry matching parent/name
// Added in FUSE protocol version 7.12. If the kernel does not support this
// (or a newer) version, the op will return -ENOSYS and do nothing
type NotifyInvalEntryOp struct {
// the inode number
Parent InodeID

// the child entry file name
Name string
}

// Store data to the kernel buffers
// Cf:http://libfuse.github.io/doxygen/fuse__lowlevel_8h.html#a9cb974af9745294ff446d11cba2422f1
type NotifyStoreOp struct {
}

// Retrieve data from the kernel buffers
type NotifyRetrieveOp struct {
}

// This op behaves like NotifyInvalEntryOp with the following additional
// effect (at least as of Linux kernel 4.8):

// If the provided child inode matches the inode that is currently
// associated with the cached dentry, and if there are any inotify
// watches registered for the dentry, then the watchers are informed
// that the dentry has been deleted.
// Added in FUSE protocol version 7.18. If the kernel does not
// support this (or a newer) version, op will return -ENOSYS and do nothing.

type NotifyDeleteOp struct {
// the inode number
Parent InodeID

// the child entry's inode
Child InodeID

// the child entry file name
Name string
}
64 changes: 64 additions & 0 deletions fuseutil/file_system.go
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,8 @@ func NewFileSystemServer(fs FileSystem) fuse.Server {
type fileSystemServer struct {
fs FileSystem
opsInFlight sync.WaitGroup
//use set/get to use mfs not Mfs
Mfs *fuse.MountedFileSystem
}

func (s *fileSystemServer) ServeOps(c *fuse.Connection) {
Expand Down Expand Up @@ -123,6 +125,59 @@ func (s *fileSystemServer) ServeOps(c *fuse.Connection) {
}
}

func (s *fileSystemServer) InvalidateEntry(parent fuseops.InodeID, name string) error {
c := s.GetMfs().Conn.(*fuse.Connection)

op := &fuseops.NotifyInvalEntryOp{
Parent: fuseops.InodeID(parent),
Name: string(name),
}
ctx, _ := c.SetNotifyContext(op)
s.opsInFlight.Add(1)
go func(ctx context.Context) {
defer s.opsInFlight.Done()
c.NotifyKernel(ctx)
}(ctx)
return nil
}
func (s *fileSystemServer) NotifyDelete(
parent fuseops.InodeID,
child fuseops.InodeID,
name string) error {
c := s.GetMfs().Conn.(*fuse.Connection)
op := &fuseops.NotifyDeleteOp{
Parent: fuseops.InodeID(parent),
Child: fuseops.InodeID(child),
Name: string(name),
}
ctx, _ := c.SetNotifyContext(op)
s.opsInFlight.Add(1)
go func(ctx context.Context) {
defer s.opsInFlight.Done()
c.NotifyKernel(ctx)
}(ctx)
return nil

}
func (s *fileSystemServer) InvalidateInode(
ino fuseops.InodeID,
off int64,
len int64) error {
c := s.GetMfs().Conn.(*fuse.Connection)
op := &fuseops.NotifyInvalInodeOp{
Ino: fuseops.InodeID(ino),
Off: off,
Len: len,
}
ctx, _ := c.SetNotifyContext(op)
s.opsInFlight.Add(1)
go func(ctx context.Context) {
defer s.opsInFlight.Done()
c.NotifyKernel(ctx)
}(ctx)
return nil

}
func (s *fileSystemServer) handleOp(
c *fuse.Connection,
ctx context.Context,
Expand Down Expand Up @@ -216,3 +271,12 @@ func (s *fileSystemServer) handleOp(

c.Reply(ctx, err)
}

func (s *fileSystemServer) GetMfs() *fuse.MountedFileSystem {
return s.Mfs

}
func (s *fileSystemServer) SetMfs(mfs *fuse.MountedFileSystem) {
s.Mfs = mfs

}
41 changes: 41 additions & 0 deletions internal/fusekernel/fuse_kernel.go
Original file line number Diff line number Diff line change
Expand Up @@ -756,6 +756,9 @@ const (
NotifyCodePoll int32 = 1
NotifyCodeInvalInode int32 = 2
NotifyCodeInvalEntry int32 = 3
NotifyCodeStore int32 = 4
NotifyCodeRetrieve int32 = 5
NotifyCodeDelete int32 = 6
)

type NotifyInvalInodeOut struct {
Expand All @@ -764,8 +767,46 @@ type NotifyInvalInodeOut struct {
Len int64
}

const NotifyInvalInodeOutSize = int(unsafe.Sizeof(NotifyInvalInodeOut{}))

type NotifyInvalEntryOut struct {
Parent uint64
Namelen uint32
padding uint32
}

const NotifyInvalEntryOutSize = int(unsafe.Sizeof(NotifyInvalEntryOut{}))

type NotifyDeleteOut struct {
Parent uint64
Child uint64
Namelen uint32
padding uint32
}

const NotifyDeleteOutSize = int(unsafe.Sizeof(NotifyDeleteOut{}))

type NotifyStoreOut struct {
Nodeid uint64
Offset uint64
Size uint32
padding uint32
}

type NotifyRetrieveOut struct {
NotifyUnique uint64
Nodeid uint64
Offset uint64
Size uint32
padding uint32
}

/* Matches the size of fuse_write_in */
type NotifyRetrieveIn struct {
Dummy1 uint64
Offset uint64
Size uint32
Dummy2 uint32
Dummy3 uint64
Dummy4 uint64
}
11 changes: 10 additions & 1 deletion mount.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
"fmt"
"os"

"github.com/jacobsa/fuse/fuseops"
"golang.org/x/net/context"
)

Expand All @@ -28,6 +29,11 @@ type Server interface {
// until all operations have been responded to. Must not be called more than
// once.
ServeOps(*Connection)
InvalidateEntry(fuseops.InodeID, string) error
InvalidateInode(fuseops.InodeID, int64, int64) error
NotifyDelete(fuseops.InodeID, fuseops.InodeID, string) error
SetMfs(*MountedFileSystem)
GetMfs() *MountedFileSystem
}

// Mount attempts to mount a file system on the given directory, using the
Expand Down Expand Up @@ -85,7 +91,10 @@ func Mount(
return
}

// Serve the connection in the background. When done, set the join status.
mfs.Conn = connection
server.SetMfs(mfs)

// Serve the connection in the background. When done, set the join status
go func() {
server.ServeOps(connection)
mfs.joinStatus = connection.close()
Expand Down
Loading

0 comments on commit 3a29c55

Please sign in to comment.