diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go index 66cbac1e3..c76ab4e1a 100644 --- a/weed/filer/reader_cache.go +++ b/weed/filer/reader_cache.go @@ -178,6 +178,10 @@ func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte, // startCaching downloads the chunk data in the background. // It does NOT hold the lock during the HTTP download to allow concurrent readers // to wait efficiently using the done channel. +// +// Concurrent downloads of the same chunk are already deduplicated by the +// ReaderCache.downloaders map (guarded by the ReaderCache mutex). Each fileId +// has at most one active SingleChunkCacher at any time. func (s *SingleChunkCacher) startCaching() { s.wg.Add(1) defer s.wg.Done() diff --git a/weed/filer/reader_cache_test.go b/weed/filer/reader_cache_test.go index 3d04b73c7..8f7f6ffc5 100644 --- a/weed/filer/reader_cache_test.go +++ b/weed/filer/reader_cache_test.go @@ -433,6 +433,46 @@ func TestSingleChunkCacherMultipleReadersWaitForDownload(t *testing.T) { } } +// TestReaderCacheDownloaderDedup tests that concurrent ReadChunkAt calls for +// the same fileId result in only one network fetch (lookup call), because +// the downloaders map deduplicates in-flight downloads. +func TestReaderCacheDownloaderDedup(t *testing.T) { + cache := newMockChunkCacheForReaderCache() + var lookupCount int32 + lookupGate := make(chan struct{}) + + lookupFn := func(ctx context.Context, fileId string) ([]string, error) { + atomic.AddInt32(&lookupCount, 1) + <-lookupGate + // Return an error so we don't need to mock the HTTP fetch. + return nil, fmt.Errorf("simulated lookup for %s", fileId) + } + + rc := NewReaderCache(10, cache, lookupFn) + defer rc.destroy() + + const numReaders = 10 + var wg sync.WaitGroup + wg.Add(numReaders) + + for i := 0; i < numReaders; i++ { + go func() { + defer wg.Done() + buffer := make([]byte, 100) + rc.ReadChunkAt(context.Background(), buffer, "dedup-file", nil, false, 0, 100, false) + }() + } + + // Allow downloads to proceed. + close(lookupGate) + wg.Wait() + + count := atomic.LoadInt32(&lookupCount) + if count != 1 { + t.Errorf("expected exactly 1 lookup call, got %d", count) + } +} + // TestSingleChunkCacherOneReaderCancelsOthersContinue tests that when one reader // cancels, other readers waiting on the same chunk continue to wait. func TestSingleChunkCacherOneReaderCancelsOthersContinue(t *testing.T) { diff --git a/weed/util/singleflight.go b/weed/util/singleflight.go new file mode 100644 index 000000000..89bc95ec3 --- /dev/null +++ b/weed/util/singleflight.go @@ -0,0 +1,55 @@ +package util + +import ( + "sync" +) + +// call represents an in-flight or completed function call. +type call struct { + wg sync.WaitGroup + val []byte + err error +} + +// SingleFlightGroup provides deduplication of concurrent function calls +// keyed by a string. If multiple goroutines call Do with the same key +// concurrently, only one executes the function; the others wait and +// receive the same result. +// +// After a call completes, the key is removed so that subsequent calls +// trigger a fresh execution. +type SingleFlightGroup struct { + mu sync.Mutex + m map[string]*call +} + +// Do executes fn once for a given key, even if called concurrently. +// All callers for the same key block until fn returns and then receive +// the same result. +func (g *SingleFlightGroup) Do(key string, fn func() ([]byte, error)) ([]byte, error) { + g.mu.Lock() + if g.m == nil { + g.m = make(map[string]*call) + } + if c, ok := g.m[key]; ok { + g.mu.Unlock() + c.wg.Wait() + return c.val, c.err + } + c := &call{} + c.wg.Add(1) + g.m[key] = c + g.mu.Unlock() + + // Use defer to ensure cleanup even if fn panics. This prevents + // waiters from hanging indefinitely and removes the stale key. + defer func() { + g.mu.Lock() + delete(g.m, key) + c.wg.Done() + g.mu.Unlock() + }() + + c.val, c.err = fn() + return c.val, c.err +} diff --git a/weed/util/singleflight_test.go b/weed/util/singleflight_test.go new file mode 100644 index 000000000..e985175c7 --- /dev/null +++ b/weed/util/singleflight_test.go @@ -0,0 +1,159 @@ +package util + +import ( + "errors" + "sync" + "sync/atomic" + "testing" +) + +func TestSingleFlightGroupDedup(t *testing.T) { + // Verify that two concurrent Do calls for the same key execute fn + // only once. The main goroutine signals right before calling Do, + // and a helper goroutine waits for that signal before closing + // the gate. Because the signal-to-Do gap is a single function + // call on the main goroutine and the key cannot be removed until + // the gate is closed, the helper's close(gate) cannot race ahead. + var g SingleFlightGroup + + expected := []byte("result") + var primaryCalls, secondaryCalls int32 + + fnRunning := make(chan struct{}) // closed by fn + aboutToDo := make(chan struct{}) // closed by main before calling Do + gate := make(chan struct{}) // closed by helper to release fn + + // Primary goroutine: executes fn, which blocks on gate. + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + g.Do("key1", func() ([]byte, error) { + atomic.AddInt32(&primaryCalls, 1) + close(fnRunning) + <-gate + return expected, nil + }) + }() + + // Wait for fn to be running (key is now in the map). + <-fnRunning + + // Helper goroutine: waits for the main goroutine to signal it is + // about to call Do, then closes gate. By the time close(gate) + // executes, the main goroutine has already entered Do (or is about + // to) -- and since fn is still blocked on gate until this very + // close, the key is still in the map. + go func() { + <-aboutToDo + close(gate) + }() + + // Signal the helper then immediately call Do. The key is in the + // map (fn is blocked on gate which has not been closed yet because + // the helper is waiting on aboutToDo which we are about to close). + close(aboutToDo) + v, err := g.Do("key1", func() ([]byte, error) { + atomic.AddInt32(&secondaryCalls, 1) + return nil, nil + }) + + wg.Wait() + + if atomic.LoadInt32(&primaryCalls) != 1 { + t.Errorf("expected primary fn called once, got %d", primaryCalls) + } + if atomic.LoadInt32(&secondaryCalls) != 0 { + t.Errorf("expected secondary fn never called, got %d", secondaryCalls) + } + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if string(v) != string(expected) { + t.Errorf("got %q, want %q", v, expected) + } +} + +func TestSingleFlightGroupDifferentKeys(t *testing.T) { + var g SingleFlightGroup + var calls int32 + + fn := func() ([]byte, error) { + atomic.AddInt32(&calls, 1) + return []byte("ok"), nil + } + + var wg sync.WaitGroup + wg.Add(3) + for _, key := range []string{"a", "b", "c"} { + go func(k string) { + defer wg.Done() + g.Do(k, fn) + }(key) + } + wg.Wait() + + if c := atomic.LoadInt32(&calls); c != 3 { + t.Errorf("expected 3 independent calls, got %d", c) + } +} + +func TestSingleFlightGroupErrorPropagation(t *testing.T) { + var g SingleFlightGroup + gate := make(chan struct{}) + testErr := errors.New("download failed") + + fn := func() ([]byte, error) { + <-gate + return nil, testErr + } + + const n = 5 + errs := make([]error, n) + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func(idx int) { + defer wg.Done() + _, errs[idx] = g.Do("errkey", fn) + }(i) + } + + close(gate) + wg.Wait() + + for i := 0; i < n; i++ { + if !errors.Is(errs[i], testErr) { + t.Errorf("caller %d: expected %v, got %v", i, testErr, errs[i]) + } + } +} + +func TestSingleFlightGroupFreshCallAfterCompletion(t *testing.T) { + var g SingleFlightGroup + var calls int32 + + fn := func() ([]byte, error) { + c := atomic.AddInt32(&calls, 1) + return []byte{byte(c)}, nil + } + + // First call. + v1, err1 := g.Do("key", fn) + if err1 != nil { + t.Fatalf("first call error: %v", err1) + } + + // Second call should trigger a fresh execution. + v2, err2 := g.Do("key", fn) + if err2 != nil { + t.Fatalf("second call error: %v", err2) + } + + if atomic.LoadInt32(&calls) != 2 { + t.Errorf("expected fn called twice for sequential calls, got %d", calls) + } + if v1[0] == v2[0] { + t.Errorf("expected different results for sequential calls, got %v and %v", v1, v2) + } +}