vfs: invalidate entry cache in readdir (#4453)

pkg/fuse/fuse.go

@@ -354,11 +354,12 @@ func (fs *fileSystem) ReadDir(cancel <-chan struct{}, in *fuse.ReadIn, out *fuse
 	defer releaseContext(ctx)
 	entries, _, err := fs.v.Readdir(ctx, Ino(in.NodeId), in.Size, int(in.Offset), in.Fh, false)
 	var de fuse.DirEntry
-	for _, e := range entries {
+	for i, e := range entries {
 		de.Ino = uint64(e.Inode)
 		de.Name = string(e.Name)
 		de.Mode = e.Attr.SMode()
 		if !out.AddDirEntry(de) {
+			fs.v.UpdateReaddirOffset(ctx, Ino(in.NodeId), in.Fh, int(in.Offset)+i)
 			break
 		}
 	}
@@ -371,12 +372,13 @@ func (fs *fileSystem) ReadDirPlus(cancel <-chan struct{}, in *fuse.ReadIn, out *
 	entries, readAt, err := fs.v.Readdir(ctx, Ino(in.NodeId), in.Size, int(in.Offset), in.Fh, true)
 	ctx.start = readAt
 	var de fuse.DirEntry
-	for _, e := range entries {
+	for i, e := range entries {
 		de.Ino = uint64(e.Inode)
 		de.Name = string(e.Name)
 		de.Mode = e.Attr.SMode()
 		eo := out.AddDirLookupEntry(de)
 		if eo == nil {
+			fs.v.UpdateReaddirOffset(ctx, Ino(in.NodeId), in.Fh, int(in.Offset)+i)
 			break
 		}
 		if e.Attr.Full {

pkg/vfs/handle.go

@@ -33,6 +33,8 @@ type handle struct {
 	// for dir
 	children []*meta.Entry
 	readAt   time.Time
+	readOff  int
+	index    map[string]int
 
 	// for file
 	locks      uint8
@@ -234,3 +236,35 @@ func (v *VFS) releaseFileHandle(ino Ino, fh uint64) {
 		h.Close()
 	}
 }
+
+func (v *VFS) invalidateDirHandle(parent Ino, name string, inode Ino, attr *Attr) {
+	v.hanleM.Lock()
+	hs := v.handles[parent]
+	v.hanleM.Unlock()
+	for _, h := range hs {
+		h.Lock()
+		if h.children != nil && h.index != nil {
+			if inode > 0 {
+				h.children = append(h.children, &meta.Entry{
+					Inode: inode,
+					Name:  []byte(name),
+					Attr:  attr,
+				})
+				h.index[name] = len(h.children) - 1
+			} else {
+				i, ok := h.index[name]
+				if ok {
+					delete(h.index, name)
+					h.children[i].Inode = 0 // invalid
+					if i >= h.readOff {
+						// not read yet, remove it
+						h.children[i] = h.children[len(h.children)-1]
+						h.index[string(h.children[i].Name)] = i
+						h.children = h.children[:len(h.children)-1]
+					}
+				}
+			}
+		}
+		h.Unlock()
+	}
+}

pkg/vfs/vfs.go

@@ -172,6 +172,7 @@ func (v *VFS) Mknod(ctx Context, parent Ino, name string, mode uint16, cumask ui
 	err = v.Meta.Mknod(ctx, parent, name, _type, mode&07777, cumask, rdev, "", &inode, attr)
 	if err == 0 {
 		entry = &meta.Entry{Inode: inode, Attr: attr}
+		v.invalidateDirHandle(parent, name, inode, attr)
 	}
 	return
 }
@@ -187,6 +188,9 @@ func (v *VFS) Unlink(ctx Context, parent Ino, name string) (err syscall.Errno) {
 		return
 	}
 	err = v.Meta.Unlink(ctx, parent, name)
+	if err == 0 {
+		v.invalidateDirHandle(parent, name, 0, nil)
+	}
 	return
 }
 
@@ -208,6 +212,7 @@ func (v *VFS) Mkdir(ctx Context, parent Ino, name string, mode uint16, cumask ui
 	err = v.Meta.Mkdir(ctx, parent, name, mode, cumask, 0, &inode, attr)
 	if err == 0 {
 		entry = &meta.Entry{Inode: inode, Attr: attr}
+		v.invalidateDirHandle(parent, name, inode, attr)
 	}
 	return
 }
@@ -219,6 +224,9 @@ func (v *VFS) Rmdir(ctx Context, parent Ino, name string) (err syscall.Errno) {
 		return
 	}
 	err = v.Meta.Rmdir(ctx, parent, name)
+	if err == 0 {
+		v.invalidateDirHandle(parent, name, 0, nil)
+	}
 	return
 }
 
@@ -240,6 +248,7 @@ func (v *VFS) Symlink(ctx Context, path string, parent Ino, name string) (entry
 	err = v.Meta.Symlink(ctx, parent, name, path, &inode, attr)
 	if err == 0 {
 		entry = &meta.Entry{Inode: inode, Attr: attr}
+		v.invalidateDirHandle(parent, name, inode, attr)
 	}
 	return
 }
@@ -267,7 +276,14 @@ func (v *VFS) Rename(ctx Context, parent Ino, name string, newparent Ino, newnam
 		return
 	}
 
-	err = v.Meta.Rename(ctx, parent, name, newparent, newname, flags, nil, nil)
+	var inode Ino
+	var attr = &Attr{}
+	err = v.Meta.Rename(ctx, parent, name, newparent, newname, flags, &inode, attr)
+	if err == 0 {
+		v.invalidateDirHandle(parent, name, 0, nil)
+		v.invalidateDirHandle(newparent, newname, 0, nil)
+		v.invalidateDirHandle(newparent, newname, inode, attr)
+	}
 	return
 }
 
@@ -292,6 +308,7 @@ func (v *VFS) Link(ctx Context, ino Ino, newparent Ino, newname string) (entry *
 	err = v.Meta.Link(ctx, ino, newparent, newname, attr)
 	if err == 0 {
 		entry = &meta.Entry{Inode: ino, Attr: attr}
+		v.invalidateDirHandle(newparent, newname, ino, attr)
 	}
 	return
 }
@@ -357,14 +374,31 @@ func (v *VFS) Readdir(ctx Context, ino Ino, size uint32, off int, fh uint64, plu
 				})
 			}
 		}
+		index := make(map[string]int)
+		for i, e := range inodes {
+			index[string(e.Name)] = i
+		}
+		h.index = index
 	}
 	if off < len(h.children) {
 		entries = h.children[off:]
+		// we don't know how much of them will be sent, assume all of them
+		h.readOff = len(h.children) - 1
 	}
 	readAt = h.readAt
 	return
 }
 
+func (v *VFS) UpdateReaddirOffset(ctx Context, ino Ino, fh uint64, off int) {
+	h := v.findHandle(ino, fh)
+	if h == nil {
+		return
+	}
+	h.Lock()
+	defer h.Unlock()
+	h.readOff = off
+}
+
 func (v *VFS) Releasedir(ctx Context, ino Ino, fh uint64) int {
 	h := v.findHandle(ino, fh)
 	if h == nil {
@@ -398,6 +432,7 @@ func (v *VFS) Create(ctx Context, parent Ino, name string, mode uint16, cumask u
 		v.UpdateLength(inode, attr)
 		fh = v.newFileHandle(inode, attr.Length, flags)
 		entry = &meta.Entry{Inode: inode, Attr: attr}
+		v.invalidateDirHandle(parent, name, inode, attr)
 	}
 	return
 }

pkg/vfs/vfs_test.go

@@ -860,3 +860,73 @@ func TestInternalFile(t *testing.T) {
 		t.Fatalf("result: %s", string(resp[:n]))
 	}
 }
+
+func TestReaddirCache(t *testing.T) {
+	v, _ := createTestVFS()
+	ctx := NewLogContext(meta.Background)
+	entry, st := v.Mkdir(ctx, 1, "testdir", 0777, 022)
+	if st != 0 {
+		t.Fatalf("mkdir testdir: %s", st)
+	}
+	parent := entry.Inode
+	for i := 0; i < 100; i++ {
+		_, _ = v.Mkdir(ctx, parent, fmt.Sprintf("d%d", i), 0777, 022)
+	}
+	fh, _ := v.Opendir(ctx, parent, 0)
+	_, _ = v.Mkdir(ctx, parent, fmt.Sprintf("d%d", 100), 0777, 022)
+	defer v.Releasedir(ctx, parent, fh)
+	var off = 20
+	var files = make(map[string]bool)
+	// read first 20
+	entries, _, _ := v.Readdir(ctx, parent, 20, 0, fh, true)
+	for _, e := range entries[:off] {
+		files[string(e.Name)] = true
+	}
+	v.UpdateReaddirOffset(ctx, parent, fh, off)
+	for i := 0; i < 100; i += 10 {
+		name := fmt.Sprintf("d%d", i)
+		_ = v.Rmdir(ctx, parent, name)
+		delete(files, name)
+	}
+	for i := 100; i < 110; i++ {
+		_, _ = v.Mkdir(ctx, parent, fmt.Sprintf("d%d", i), 0777, 022)
+		_ = v.Rename(ctx, parent, fmt.Sprintf("d%d", i), parent, fmt.Sprintf("d%d", i+10), 0)
+		delete(files, fmt.Sprintf("d%d", i))
+	}
+	for {
+		entries, _, _ := v.Readdir(ctx, parent, 20, off, fh, true)
+		if len(entries) == 0 {
+			break
+		}
+		if len(entries) > 20 {
+			entries = entries[:20]
+		}
+		for _, e := range entries {
+			if e.Inode > 0 {
+				files[string(e.Name)] = true
+			} else {
+				t.Logf("invalid entry %s", e.Name)
+			}
+		}
+		off += len(entries)
+		v.UpdateReaddirOffset(ctx, parent, fh, off)
+	}
+	for i := 0; i < 100; i += 10 {
+		name := fmt.Sprintf("d%d", i)
+		if _, ok := files[name]; ok {
+			t.Fatalf("dir %s should be deleted", name)
+		}
+	}
+	for i := 100; i < 110; i++ {
+		name := fmt.Sprintf("d%d", i)
+		if _, ok := files[name]; ok {
+			t.Fatalf("dir %s should be deleted", name)
+		}
+	}
+	for i := 110; i < 120; i++ {
+		name := fmt.Sprintf("d%d", i)
+		if _, ok := files[name]; !ok {
+			t.Fatalf("dir %s should be added", name)
+		}
+	}
+}