Add notification messages framework

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 .

Author: inevity

connection.go

@@ -240,7 +240,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)
@@ -410,6 +422,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{},
@@ -497,6 +541,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) {

conversions.go

@@ -623,6 +623,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(
@@ -922,3 +975,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
+}

fuseops/ops.go

@@ -865,3 +865,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
+}

fuseutil/file_system.go

@@ -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) {
@@ -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,
@@ -219,3 +274,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
+
+}

internal/fusekernel/fuse_kernel.go

@@ -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 {
@@ -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
+}

mount.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"github.com/jacobsa/fuse/fuseops"
 )
 
 // Server is an interface for any type that knows how to serve ops read from a
@@ -27,6 +28,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
@@ -84,7 +90,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()