From eae8f33db5b870d7f9adda5d540f9462702bf0a8 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 21 May 2026 02:40:04 -0700 Subject: [PATCH] fix(filersink): return lock-free snapshot from ActiveTransfers (#9604) ChunkTransferStatus embeds a sync.RWMutex, so returning a slice of it made callers copy the lock when ranging. Split out a copyable ChunkTransferSnapshot holding the data fields and return that instead. --- .../replication/sink/filersink/fetch_write.go | 8 +++-- weed/replication/sink/filersink/filer_sink.go | 30 +++++++++---------- 2 files changed, 20 insertions(+), 18 deletions(-) diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 0399542f1..dd1c3ffb5 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -242,9 +242,11 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, } transferStatus := &ChunkTransferStatus{ - ChunkFileId: sourceChunk.GetFileIdString(), - Path: path, - Status: "downloading", + ChunkTransferSnapshot: ChunkTransferSnapshot{ + ChunkFileId: sourceChunk.GetFileIdString(), + Path: path, + Status: "downloading", + }, } fs.activeTransfers.Store(sourceChunk.GetFileIdString(), transferStatus) defer fs.activeTransfers.Delete(sourceChunk.GetFileIdString()) diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 50aaab449..c93942fe9 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -22,12 +22,9 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) -// ChunkTransferStatus tracks the progress of a single chunk being replicated. -// Fields are guarded by mu: ChunkFileId and Path are immutable after creation, -// while BytesReceived, Status, and LastErr are updated by fetchAndWrite and -// read by ActiveTransfers. -type ChunkTransferStatus struct { - mu sync.RWMutex +// ChunkTransferSnapshot is a lock-free, copyable view of a chunk transfer's +// progress, returned by ActiveTransfers. +type ChunkTransferSnapshot struct { ChunkFileId string Path string BytesReceived int64 @@ -35,6 +32,15 @@ type ChunkTransferStatus struct { LastErr string } +// ChunkTransferStatus tracks the progress of a single chunk being replicated. +// Fields are guarded by mu: ChunkFileId and Path are immutable after creation, +// while BytesReceived, Status, and LastErr are updated by fetchAndWrite and +// read by ActiveTransfers. +type ChunkTransferStatus struct { + mu sync.RWMutex + ChunkTransferSnapshot +} + type FilerSink struct { filerSource *source.FilerSource grpcAddress string @@ -134,18 +140,12 @@ func (fs *FilerSink) SetChunkConcurrency(concurrency int) { } // ActiveTransfers returns an immutable snapshot of all in-progress chunk transfers. -func (fs *FilerSink) ActiveTransfers() []ChunkTransferStatus { - var transfers []ChunkTransferStatus +func (fs *FilerSink) ActiveTransfers() []ChunkTransferSnapshot { + var transfers []ChunkTransferSnapshot fs.activeTransfers.Range(func(key, value any) bool { t := value.(*ChunkTransferStatus) t.mu.RLock() - transfers = append(transfers, ChunkTransferStatus{ - ChunkFileId: t.ChunkFileId, - Path: t.Path, - BytesReceived: t.BytesReceived, - Status: t.Status, - LastErr: t.LastErr, - }) + transfers = append(transfers, t.ChunkTransferSnapshot) t.mu.RUnlock() return true })