From bce76e6e219424c03a683414aaccebd0c58278c7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 23 May 2026 14:22:42 -0700 Subject: [PATCH] filer: serialize same-path mutations with a per-path lock (#9639) CreateEntry is a FindEntry-then-write with no lock, so concurrent creates to the same path race: OExcl can admit two creators, and a conditional check-then-act has no atomicity. Add a per-path exclusive lock (util.LockTable, which evicts idle keys so it stays bounded) on the FilerServer and take it in CreateEntry, so the existence check and the write are atomic on this filer. This is the local serialization point that lets callers route a key's writes to its owner filer and drop the distributed lock for that key. AppendToEntry keeps its distributed lock for now; it can move to the per-path lock once its callers route to the owner. --- weed/server/filer_grpc_server.go | 8 ++- .../filer_grpc_server_create_lock_test.go | 68 +++++++++++++++++++ weed/server/filer_grpc_server_rename_test.go | 4 ++ weed/server/filer_server.go | 9 +++ 4 files changed, 88 insertions(+), 1 deletion(-) create mode 100644 weed/server/filer_grpc_server_create_lock_test.go diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index 76dcb4dee..7e90eb09e 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -10,7 +10,6 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/cluster" - "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" @@ -186,6 +185,13 @@ func (fs *FilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntr newEntry.TtlSec = 0 } + // Serialize concurrent mutations to the same path on this filer so the + // read (existence/condition) and the write are atomic. Callers route a + // key's writes to this owner filer, making this local lock sufficient. + fullpath := newEntry.FullPath + pathLock := fs.entryLockTable.AcquireLock("CreateEntry", fullpath, util.ExclusiveLock) + defer fs.entryLockTable.ReleaseLock(fullpath, pathLock) + ctx, eventSink := filer.WithMetadataEventSink(ctx) createErr := fs.filer.CreateEntry(ctx, newEntry, req.OExcl, req.IsFromOtherCluster, req.Signatures, req.SkipCheckParentDirectory, so.MaxFileNameLength) diff --git a/weed/server/filer_grpc_server_create_lock_test.go b/weed/server/filer_grpc_server_create_lock_test.go new file mode 100644 index 000000000..ecd1f8ffb --- /dev/null +++ b/weed/server/filer_grpc_server_create_lock_test.go @@ -0,0 +1,68 @@ +package weed_server + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// Concurrent OExcl creates for the same path must yield exactly one winner. The +// filer's CreateEntry is a FindEntry-then-Insert; without the per-path lock both +// racers observe "not found" and both insert. The exclusive entry lock makes the +// check-then-act atomic so the losers see ErrEntryAlreadyExists. +func TestCreateEntryOExclSerialized(t *testing.T) { + store := newRenameTestStore() + store.findDelay = 5 * time.Millisecond + f := newRenameTestFiler(store) + f.DirBucketsPath = "/buckets" + + fs := &FilerServer{ + filer: f, + option: &FilerOption{}, + entryLockTable: util.NewLockTable[util.FullPath](), + } + + const racers = 8 + var success, alreadyExists, unexpected int32 + var wg sync.WaitGroup + start := make(chan struct{}) + for i := 0; i < racers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + resp, err := fs.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{ + Directory: "/test", + OExcl: true, + SkipCheckParentDirectory: true, + Entry: &filer_pb.Entry{ + Name: "obj", + Attributes: &filer_pb.FuseAttributes{Mtime: 1700000000, FileMode: 0644, Inode: 1}, + }, + }) + switch { + case err != nil: + atomic.AddInt32(&unexpected, 1) + case resp.Error == "": + atomic.AddInt32(&success, 1) + case resp.ErrorCode == filer_pb.FilerError_ENTRY_ALREADY_EXISTS: + atomic.AddInt32(&alreadyExists, 1) + default: + atomic.AddInt32(&unexpected, 1) + } + }() + } + close(start) + wg.Wait() + + // Exactly one winner; every loser fails with ENTRY_ALREADY_EXISTS and nothing + // else, so an unrelated failure can't masquerade as a passing test. + if success != 1 || alreadyExists != racers-1 || unexpected != 0 { + t.Fatalf("winners=%d already_exists=%d unexpected=%d (racers=%d)", success, alreadyExists, unexpected, racers) + } +} diff --git a/weed/server/filer_grpc_server_rename_test.go b/weed/server/filer_grpc_server_rename_test.go index aec0c0740..3fe9392a1 100644 --- a/weed/server/filer_grpc_server_rename_test.go +++ b/weed/server/filer_grpc_server_rename_test.go @@ -29,6 +29,7 @@ type renameTestStore struct { findCalls map[string]int commitErr error deleteErr error + findDelay time.Duration // optional: widen check-then-act windows in tests } func newRenameTestStore() *renameTestStore { @@ -69,6 +70,9 @@ func (s *renameTestStore) UpdateEntry(_ context.Context, entry *filer.Entry) err } func (s *renameTestStore) FindEntry(_ context.Context, p util.FullPath) (*filer.Entry, error) { + if s.findDelay > 0 { + time.Sleep(s.findDelay) + } s.mu.Lock() defer s.mu.Unlock() s.findCalls[string(p)]++ diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 52b42be3a..6cf2f87c3 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -124,6 +124,14 @@ type FilerServer struct { // mountPeerRegistry backs the MountRegister / MountList RPCs for peer // chunk sharing (tier 1). Always populated. mountPeerRegistry *filer.MountPeerRegistry + + // entryLockTable serializes mutations to the same entry path on this filer. + // CreateEntry takes it today; UpdateEntry and DeleteEntry are intended to take + // it too as their callers route a key's writes to this node, making it the + // local serialization point for read-modify-write operations that replaces + // the distributed lock for that key. Idle keys are evicted automatically, so + // the table stays bounded. + entryLockTable *util.LockTable[util.FullPath] } func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) { @@ -162,6 +170,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)), recentCopyRequests: make(map[string]recentCopyRequest), CredentialManager: option.CredentialManager, + entryLockTable: util.NewLockTable[util.FullPath](), } fs.mountPeerRegistry = filer.NewMountPeerRegistry() go fs.runMountPeerRegistrySweeper()