diff --git a/pkg/cmd/client/main.go b/pkg/cmd/client/main.go index ed34bd3..2772e67 100644 --- a/pkg/cmd/client/main.go +++ b/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) }() diff --git a/pkg/server/claim_tracker.go b/pkg/server/claim_tracker.go index 480766a..985bb19 100644 --- a/pkg/server/claim_tracker.go +++ b/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 } diff --git a/pkg/server/inode.go b/pkg/server/inode.go index 1102aa1..2addbc4 100644 --- a/pkg/server/inode.go +++ b/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,7 +164,7 @@ 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 @@ -166,7 +172,8 @@ func (n *ServerNode) GetAttr(s *session, req *proto.GetAttrRequest) (*proto.Oper 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 } diff --git a/pkg/vfs/cache.go b/pkg/vfs/cache.go index 540d121..46f6d22 100644 --- a/pkg/vfs/cache.go +++ b/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) } diff --git a/pkg/vfs/file_node.go b/pkg/vfs/file_node.go index e7980e4..0179f45 100644 --- a/pkg/vfs/file_node.go +++ b/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)