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

@@ -241,7 +241,7 @@ 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 {
+	if opCode != fusekernel.OpForget && opCode < 100 {
 		var cancel func()
 		ctx, cancel = context.WithCancel(ctx)
 		c.recordCancelFunc(fuseID, cancel)
@@ -411,6 +411,36 @@ func (c *Connection) ReadOp() (ctx context.Context, op interface{}, err error) {
 	}
 }
 
+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{},
@@ -498,6 +528,27 @@ func (c *Connection) Reply(ctx context.Context, opErr error) {
 	}
 }
 
+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

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

fuseops/ops.go

@@ -845,3 +845,24 @@ type SetXattrOp struct {
 	// simply replace the value if the attribute exists.
 	Flags uint32
 }
+type NotifyPollOp struct {
+}
+
+type NotifyInvalInodeOp struct {
+	Ino InodeID
+	Off int64
+	Len int64
+}
+type NotifyInvalEntryOp struct {
+	Parent InodeID
+	Name   string
+}
+type NotifyStoreOp struct {
+}
+type NotifyRetrieveOp struct {
+}
+type NotifyDeleteOp struct {
+	Parent InodeID
+	Child  InodeID
+	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,
@@ -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
+
+}

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 (
 	"fmt"
 	"os"
 
+	"github.com/jacobsa/fuse/fuseops"
 	"golang.org/x/net/context"
 )
 
@@ -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
@@ -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()

mounted_file_system.go

@@ -24,6 +24,7 @@ type MountedFileSystem struct {
 	// The result to return from Join. Not valid until the channel is closed.
 	joinStatus          error
 	joinStatusAvailable chan struct{}
+	Conn                interface{}
 }
 
 // Dir returns the directory on which the file system is mounted (or where we

samples/forgetfs/forget_fs.go

@@ -99,6 +99,23 @@ type ForgetFS struct {
 func (fs *ForgetFS) ServeOps(c *fuse.Connection) {
 	fs.server.ServeOps(c)
 }
+func (fs *ForgetFS) SetMfs(mfs *fuse.MountedFileSystem) {
+	fs.server.SetMfs(mfs)
+}
+func (fs *ForgetFS) GetMfs() *fuse.MountedFileSystem {
+	return fs.server.GetMfs()
+}
+func (fs *ForgetFS) InvalidateEntry(nodeid fuseops.InodeID, name string) error {
+	return nil
+}
+
+func (fs *ForgetFS) InvalidateInode(nodeid fuseops.InodeID, parent int64, nlookup int64) error {
+	return nil
+}
+
+func (fs *ForgetFS) NotifyDelete(nodeid fuseops.InodeID, parent fuseops.InodeID, name string) error {
+	return nil
+}
 
 // Panic if there are any inodes that have a non-zero reference count. For use
 // after unmounting.