Add some fix for e2e perf testing

pkg/cmd/client/main.go

@@ -17,11 +17,14 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
+	"velda.io/clfs/pkg/test"
 	"velda.io/clfs/pkg/vfs"
 )
 
 var debug = flag.Bool("debug", false, "Enable debug logging")
 var endpoint = flag.String("endpoint", "dns:///localhost:50055", "gRPC endpoint to connect to")
+var debugEndpoint = flag.String("debug-endpoint", "localhost:6070", "Endpoint for pprof debug server")
+var injectLatency = flag.Duration("inject-latency", 0, "Inject latency into gRPC client requests")
 
 type MountOptions func(*fs.Options)
 
@@ -33,6 +36,10 @@ func RunClient(endpoint string) vfs.ServerProtocol {
 
 	c := vfs.NewClient(conn)
 	err = c.Start(context.Background())
+	if *injectLatency > 0 {
+		log.Printf("Injecting %v latency into gRPC client requests", *injectLatency)
+		c.Stream = test.NewLatencyInjectedStream(c.Stream, *injectLatency)
+	}
 	if err != nil {
 		log.Fatalf("Client start error: %v", err)
 	}
@@ -87,7 +94,10 @@ func main() {
 	}
 
 	go func() {
-		lis, _ := net.Listen("tcp", ":6070")
+		if *debugEndpoint == "" {
+			return
+		}
+		lis, _ := net.Listen("tcp", *debugEndpoint)
 		http.Serve(lis, nil)
 	}()
 

pkg/server/claim_tracker.go

@@ -2,17 +2,20 @@ package server
 
 import (
 	"sync"
+
+	"velda.io/clfs/pkg/proto"
 )
 
+type pendingClaimCallback func(proto.ClaimStatus)
 type dentryClaimTracker struct {
 	sessions map[*session]bool
-	queue    []func()
+	queue    []pendingClaimCallback
 }
 type claimTracker struct {
 	mu       sync.Mutex
 	writer   *session
 	readers  map[*session]bool
-	queue    []func()
+	queue    []pendingClaimCallback
 	updater  claimUpdater
 	dentries map[string]*dentryClaimTracker
 }
@@ -64,19 +67,20 @@ func (t *claimTracker) AddDentryClaim(s *session, dentry string) {
 	}
 }
 
-func (t *claimTracker) Write(s *session, dentry string, callback func()) {
+func (t *claimTracker) Write(s *session, dentry string, callback pendingClaimCallback) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	noQueue := len(t.queue) == 0
 	// Currently holds the writer claim and not revoking
 	if t.writer == s {
-		callback()
+		callback(proto.ClaimStatus_CLAIM_STATUS_UNSPECIFIED)
 		return
 	}
 	if t.writer != nil && noQueue {
 		debugf("%p: Revoking writer %p, write from %p", t, t.writer, s)
 		t.updater.NotifyRevokeWriter(t.writer)
 	}
+	delete(t.readers, s)
 	if len(t.readers) > 0 && noQueue {
 		for reader := range t.readers {
 			t.updater.NotifyRevokeReader(reader)
@@ -88,7 +92,7 @@ func (t *claimTracker) Write(s *session, dentry string, callback func()) {
 			t.writer = s
 			s.AddWriterClaim(t)
 			t.dentries = nil
-			callback()
+			callback(proto.ClaimStatus_CLAIM_STATUS_EXCLUSIVE_WRITE_GRANTED)
 			return
 		}
 		dentryClaimed := true
@@ -111,13 +115,13 @@ func (t *claimTracker) Write(s *session, dentry string, callback func()) {
 				}
 			}
 		}
-		callback()
+		callback(proto.ClaimStatus_CLAIM_STATUS_UNSPECIFIED)
 		return
 	}
 	t.queue = append(t.queue, callback)
 }
 
-func (t *claimTracker) Read(s *session, callback func()) {
+func (t *claimTracker) Read(s *session, callback func(proto.ClaimStatus)) {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 
@@ -126,7 +130,7 @@ func (t *claimTracker) Read(s *session, callback func()) {
 	if t.writer != nil {
 		if noQueue {
 			if t.writer == s {
-				callback()
+				callback(proto.ClaimStatus_CLAIM_STATUS_UNSPECIFIED)
 				return
 			}
 			debugf("%p: Revoking writer %p, read from %p", t, t.writer, s)
@@ -147,11 +151,14 @@ func (t *claimTracker) Read(s *session, callback func()) {
 	if _, exists := t.readers[s]; !exists {
 		if t.updater.ClaimReader(s) {
 			t.readers[s] = true
+			debugf("%p: Reader %p claimed", t, s)
 			s.AddReaderClaim(t)
 			t.dentries = nil
+			callback(proto.ClaimStatus_CLAIM_STATUS_LOCK_READ_GRANTED)
+			return
 		}
 	}
-	callback()
+	callback(proto.ClaimStatus_CLAIM_STATUS_UNSPECIFIED)
 }
 
 // Caller needs to remove it from session's live claims.
@@ -188,7 +195,7 @@ func (t *claimTracker) RevokedDentry(s *session, dentry string) {
 			queue := dentryClaims.queue
 			delete(t.dentries, dentry)
 			for _, callback := range queue {
-				callback()
+				callback(proto.ClaimStatus_CLAIM_STATUS_UNSPECIFIED)
 			}
 		}
 	} else {
@@ -200,7 +207,7 @@ func (t *claimTracker) checkQueues() {
 	if t.writer == nil && len(t.readers) == 0 {
 		debugf("%v: No active claims, processing queue", t)
 		for _, callback := range t.queue {
-			callback()
+			callback(proto.ClaimStatus_CLAIM_STATUS_UNSPECIFIED)
 		}
 		t.queue = nil
 	}

pkg/server/inode.go

@@ -39,8 +39,8 @@ type HandleCallback func(*proto.OperationResponse, error)
 
 func (n *ServerNode) Handle(s *session, req *proto.OperationRequest, callback HandleCallback) {
 
-	do := func() {
-		resp, err := n.handle(s, req)
+	do := func(claimUpdate proto.ClaimStatus) {
+		resp, err := n.handle(s, req, claimUpdate)
 		callback(resp, err)
 	}
 	var dentry string
@@ -59,11 +59,18 @@ func (n *ServerNode) Handle(s *session, req *proto.OperationRequest, callback Ha
 		return
 	// Readonly
 	case *proto.OperationRequest_Lookup:
+		readonly = true
 	case *proto.OperationRequest_GetAttr:
+		readonly = true
 	case *proto.OperationRequest_Readlink:
+		readonly = true
 	case *proto.OperationRequest_ScanDir:
+		readonly = true
 	case *proto.OperationRequest_Read:
 		readonly = true
+	case *proto.OperationRequest_Close:
+		// Should already have the claim if needed.
+		readonly = true
 
 	case *proto.OperationRequest_Mknod:
 		dentry = op.Mknod.Name
@@ -79,7 +86,6 @@ func (n *ServerNode) Handle(s *session, req *proto.OperationRequest, callback Ha
 		dentry = op.Create.Name
 	case *proto.OperationRequest_SetAttr:
 	case *proto.OperationRequest_Write:
-	case *proto.OperationRequest_Close:
 	case *proto.OperationRequest_Mount:
 		readonly = false
 
@@ -97,12 +103,12 @@ func (n *ServerNode) Handle(s *session, req *proto.OperationRequest, callback Ha
 }
 
 // Handle processes a single operation request for this node.
-func (n *ServerNode) handle(s *session, req *proto.OperationRequest) (*proto.OperationResponse, error) {
+func (n *ServerNode) handle(s *session, req *proto.OperationRequest, claimUpdate proto.ClaimStatus) (*proto.OperationResponse, error) {
 	switch op := req.Operation.(type) {
 	case *proto.OperationRequest_Lookup:
 		return n.Lookup(s, op.Lookup)
 	case *proto.OperationRequest_GetAttr:
-		return n.GetAttr(s, op.GetAttr)
+		return n.GetAttr(s, op.GetAttr, claimUpdate)
 	case *proto.OperationRequest_SetAttr:
 		return n.SetAttr(s, op.SetAttr)
 	case *proto.OperationRequest_Mknod:
@@ -122,7 +128,7 @@ func (n *ServerNode) handle(s *session, req *proto.OperationRequest) (*proto.Ope
 	case *proto.OperationRequest_Create:
 		return n.Create(s, op.Create)
 	case *proto.OperationRequest_Open:
-		return n.Open(s, op.Open)
+		return n.Open(s, op.Open, claimUpdate)
 	case *proto.OperationRequest_Read:
 		return n.Read(s, op.Read)
 	case *proto.OperationRequest_Write:
@@ -158,15 +164,16 @@ func (n *ServerNode) Lookup(s *session, req *proto.LookupRequest) (*proto.Operat
 	}, nil
 }
 
-func (n *ServerNode) GetAttr(s *session, req *proto.GetAttrRequest) (*proto.OperationResponse, error) {
+func (n *ServerNode) GetAttr(s *session, req *proto.GetAttrRequest, claimUpdate proto.ClaimStatus) (*proto.OperationResponse, error) {
 	var stat unix.Stat_t
 	if err := unix.Fstat(n.fd, &stat); err != nil {
 		return nil, err
 	}
 	return &proto.OperationResponse{
 		Response: &proto.OperationResponse_GetAttr{
 			GetAttr: &proto.GetAttrResponse{
-				Stat: StatProtoFromSysStat(&stat),
+				Stat:        StatProtoFromSysStat(&stat),
+				ClaimUpdate: claimUpdate,
 			},
 		},
 	}, nil
@@ -232,8 +239,16 @@ func (n *ServerNode) SetAttr(s *session, req *proto.SetAttrRequest) (*proto.Oper
 	}
 	if req.Stat.Valid&FATTR_ATIME != 0 || req.Stat.Valid&FATTR_MTIME != 0 {
 		var tv [2]unix.Timeval
-		tv[0] = unix.NsecToTimeval(req.Stat.Atime.Seconds*1e9 + int64(req.Stat.Atime.Nanos))
-		tv[1] = unix.NsecToTimeval(req.Stat.Mtime.Seconds*1e9 + int64(req.Stat.Mtime.Nanos))
+		if req.Stat.Valid&FATTR_ATIME != 0 && req.Stat.Atime != nil {
+			tv[0] = unix.NsecToTimeval(req.Stat.Atime.Seconds*1e9 + int64(req.Stat.Atime.Nanos))
+		} else {
+			tv[0] = unix.NsecToTimeval(0) // Use current time if not set
+		}
+		if req.Stat.Valid&FATTR_MTIME != 0 && req.Stat.Mtime != nil {
+			tv[1] = unix.NsecToTimeval(req.Stat.Mtime.Seconds*1e9 + int64(req.Stat.Mtime.Nanos))
+		} else {
+			tv[1] = unix.NsecToTimeval(0) // Use current time if not set
+		}
 		if err := unix.Futimes(updateFd, tv[:]); err != nil {
 			return nil, err
 		}
@@ -504,7 +519,7 @@ func (n *ServerNode) Create(s *session, req *proto.CreateRequest) (*proto.Operat
 	}, nil
 }
 
-func (n *ServerNode) Open(s *session, req *proto.OpenRequest) (*proto.OperationResponse, error) {
+func (n *ServerNode) Open(s *session, req *proto.OpenRequest, claim proto.ClaimStatus) (*proto.OperationResponse, error) {
 	fd, err := unix.Open(n.fdPath(), int(req.Flags), 0)
 	if err != nil {
 		return nil, err
@@ -514,6 +529,7 @@ func (n *ServerNode) Open(s *session, req *proto.OpenRequest) (*proto.OperationR
 		Response: &proto.OperationResponse_Open{
 			Open: &proto.OpenResponse{
 				FileHandle: fh,
+				Claim:      claim,
 			},
 		},
 	}, nil
@@ -571,6 +587,7 @@ func (n *ServerNode) fdPath() string {
 }
 
 func (n *ServerNode) NotifyRevokeWriter(s *session) {
+	debugf("%p: Notifying revoke writer for session %p", n, s)
 	s.SendNotify(n, &proto.OperationResponse{
 		ServerRequest: &proto.OperationResponse_ClaimUpdate{
 			ClaimUpdate: &proto.ClaimUpdateServerRequest{
@@ -581,6 +598,7 @@ func (n *ServerNode) NotifyRevokeWriter(s *session) {
 }
 
 func (n *ServerNode) NotifyRevokeReader(s *session) {
+	debugf("%p: Notifying revoke reader for session %p", n, s)
 	s.SendNotify(n, &proto.OperationResponse{
 		ServerRequest: &proto.OperationResponse_ClaimUpdate{
 			ClaimUpdate: &proto.ClaimUpdateServerRequest{
@@ -591,6 +609,7 @@ func (n *ServerNode) NotifyRevokeReader(s *session) {
 }
 
 func (n *ServerNode) NotifyRevokeDentry(s *session, dentry string) {
+	debugf("%p: Notifying revoke dentry '%s' for session %p", n, dentry, s)
 	s.SendNotify(n, &proto.OperationResponse{
 		ServerRequest: &proto.OperationResponse_DentryInvalidation{
 			DentryInvalidation: &proto.DentryInvalidationServerRequest{
@@ -603,9 +622,9 @@ func (n *ServerNode) NotifyRevokeDentry(s *session, dentry string) {
 }
 
 func (n *ServerNode) ClaimWriter(s *session) bool {
-	return false
+	return true
 }
 
 func (n *ServerNode) ClaimReader(s *session) bool {
-	return false
+	return true
 }

pkg/vfs/cache.go

@@ -2,6 +2,7 @@ package vfs
 
 import (
 	"sort"
+	"sync"
 )
 
 // Interval represents a cache interval with data starting at a specific offset
@@ -12,6 +13,7 @@ type Interval struct {
 
 // Cache stores data as sorted intervals (offset, data)
 type Cache struct {
+	mu        sync.RWMutex
 	intervals []Interval
 }
 
@@ -25,6 +27,9 @@ func NewCache() *Cache {
 // Insert adds or updates data at the specified offset
 // It handles merging intervals when they overlap or connect
 func (c *Cache) Insert(offset int64, data []byte) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	if len(data) == 0 {
 		return
 	}
@@ -70,6 +75,9 @@ func (c *Cache) Insert(offset int64, data []byte) {
 // Read retrieves data from the cache starting at the specified offset
 // with the specified length
 func (c *Cache) Read(offset int64, length int) []byte {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	if length <= 0 {
 		return nil
 	}
@@ -91,11 +99,17 @@ func (c *Cache) Read(offset int64, length int) []byte {
 
 // Clear removes all intervals from the cache
 func (c *Cache) Clear() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	c.intervals = []Interval{}
 }
 
 // Size returns the number of intervals in the cache
 func (c *Cache) Size() int {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	return len(c.intervals)
 }
 

pkg/vfs/file_node.go

@@ -58,11 +58,14 @@ func (n *FileInode) Open(ctx context.Context, flags uint32) (fs.FileHandle, uint
 		switch resp := response.Response.(type) {
 		case *proto.OperationResponse_Open:
 			fh.SetHandle(resp.Open.FileHandle)
+			if resp.Open.Claim != proto.ClaimStatus_CLAIM_STATUS_UNSPECIFIED {
+				n.handleClaimUpdate(resp.Open.Claim)
+			}
 		default:
 			debugf("Open: Unexpected response type: %T", resp)
 		}
 	})
-	return fh, 0, errno
+	return fh, fuse.FOPEN_NOFLUSH, errno
 }
 
 func (n *FileInode) Read(ctx context.Context, fh fs.FileHandle, dest []byte, offset int64) (fuse.ReadResult, syscall.Errno) {
@@ -92,6 +95,7 @@ func (n *FileInode) Read(ctx context.Context, fh fs.FileHandle, dest []byte, off
 	}
 	switch resp := response.Response.(type) {
 	case *proto.OperationResponse_Read:
+		n.cache.Insert(offset, resp.Read.Data)
 		return fuse.ReadResultData(resp.Read.Data), fs.OK
 	default:
 		debugf("Read: Unexpected response type: %T", resp)