diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index 978b0675e..f29cbf2b2 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -117,13 +117,13 @@ func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileI return nil } -func fetchChunkRange(ctx context.Context, buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64) (int, error) { +func fetchChunkRange(ctx context.Context, buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64, refreshUrls util_http.RefreshUrlsFunc) (int, error) { urlStrings, err := lookupFileIdFn(ctx, fileId) if err != nil { glog.ErrorfCtx(ctx, "operation LookupFileId %s failed, err: %v", fileId, err) return 0, err } - return util_http.RetriedFetchChunkData(ctx, buffer, urlStrings, cipherKey, isGzipped, false, offset, fileId) + return util_http.RetriedFetchChunkData(ctx, buffer, urlStrings, cipherKey, isGzipped, false, offset, fileId, refreshUrls) } func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrings []string, jwt string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, size int) (written int64, err error) { diff --git a/weed/filer/reader_at.go b/weed/filer/reader_at.go index 5e8fd6154..d87198f54 100644 --- a/weed/filer/reader_at.go +++ b/weed/filer/reader_at.go @@ -324,7 +324,8 @@ func (c *ChunkReadAt) readChunkSliceAt(ctx context.Context, buffer []byte, chunk if n > 0 { return n, err } - return fetchChunkRange(ctx, buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset)) + return fetchChunkRange(ctx, buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset), + c.readerCache.refreshUrls(ctx, chunkView.FileId)) } shouldCache := (uint64(chunkView.ViewOffset) + chunkView.ChunkSize) <= c.readerCache.chunkCache.GetMaxFilePartSizeInCache() diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go index 055666fd8..f2a139c92 100644 --- a/weed/filer/reader_cache.go +++ b/weed/filer/reader_cache.go @@ -18,7 +18,7 @@ type CacheInvalidator interface { InvalidateCache(fileId string) } -type fetchChunkDataFnType func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string) (n int, err error) +type fetchChunkDataFnType func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (n int, err error) type ReaderCache struct { chunkCache chunk_cache.ChunkCache @@ -104,6 +104,25 @@ func (rc *ReaderCache) MaybeCache(chunkViews *Interval[*ChunkView], count int) { return } +// refreshUrls lets a fetch loop recover inside a single read: when every cached +// location for a chunk has failed, drop the cached entry and look it up again +// rather than spending the whole backoff ladder on locations that are gone. +// Nil when there is nothing to invalidate against. +func (rc *ReaderCache) refreshUrls(ctx context.Context, fileId string) util_http.RefreshUrlsFunc { + if rc.cacheInvalidator == nil || rc.lookupFileIdFn == nil { + return nil + } + return func() []string { + rc.cacheInvalidator.InvalidateCache(fileId) + urls, err := rc.lookupFileIdFn(ctx, fileId) + if err != nil { + glog.V(0).InfofCtx(ctx, "re-lookup chunk %s: %v", fileId, err) + return nil + } + return urls + } +} + func (rc *ReaderCache) ReadChunkAt(ctx context.Context, buffer []byte, fileId string, cipherKey []byte, isGzipped bool, offset int64, chunkSize int, shouldCache bool) (int, error) { rc.Lock() @@ -270,7 +289,7 @@ func (s *SingleChunkCacher) fetchChunkData(ctx context.Context, urlStrings []str // Allocate buffer and download without holding the lock. // This allows multiple downloads to proceed in parallel. data := mem.Allocate(s.chunkSize) - _, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId) + _, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId, s.parent.refreshUrls(ctx, s.chunkFileId)) if fetchErr != nil { mem.Free(data) return nil, fetchErr diff --git a/weed/filer/reader_cache_test.go b/weed/filer/reader_cache_test.go index a6771246c..b92b17c1b 100644 --- a/weed/filer/reader_cache_test.go +++ b/weed/filer/reader_cache_test.go @@ -3,6 +3,7 @@ package filer import ( "context" "fmt" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" "sync" "sync/atomic" "testing" @@ -100,7 +101,7 @@ func TestReaderCacheRetryAfterCacheInvalidation(t *testing.T) { } var fetchCount int32 - fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string) (int, error) { + fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string, _ util_http.RefreshUrlsFunc) (int, error) { if requestedFileId != fileId { return 0, fmt.Errorf("unexpected fetch file id %s", requestedFileId) } @@ -161,7 +162,7 @@ func TestReaderCacheRemovesFailedDownloader(t *testing.T) { } var fetchCount int32 - fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string) (int, error) { + fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, requestedFileId string, _ util_http.RefreshUrlsFunc) (int, error) { atomic.AddInt32(&fetchCount, 1) return 0, fmt.Errorf("fetch failed") } @@ -575,7 +576,7 @@ func TestReaderCacheDownloaderDedup(t *testing.T) { return []string{"http://volume/" + fileId}, nil } - fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string) (int, error) { + fetchFn := func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) { atomic.AddInt32(&fetchCount, 1) <-fetchGate return copy(buffer, testData), nil diff --git a/weed/filer/stream.go b/weed/filer/stream.go index ec855be0b..b06e4a55a 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -102,26 +102,6 @@ func PrepareStreamContent(masterClient wdclient.HasLookupFileIdFunction, jwtFunc type VolumeServerJwtFunction func(fileId string) string -// urlSlicesEqual checks if two URL slices contain the same URLs (order-independent) -func urlSlicesEqual(a, b []string) bool { - if len(a) != len(b) { - return false - } - // Create a map to count occurrences in first slice - counts := make(map[string]int) - for _, url := range a { - counts[url]++ - } - // Verify all URLs in second slice match - for _, url := range b { - if counts[url] == 0 { - return false - } - counts[url]-- - } - return true -} - // retryFetchWithFreshLocations is the shared self-heal for the read paths: when a chunk fetch // fails, invalidate the cached volume locations, re-lookup, and call refetch only when the // resolved locations actually changed (so we never retry against the same servers). originalErr @@ -143,7 +123,7 @@ func retryFetchWithFreshLocations(ctx context.Context, invalidator CacheInvalida glog.WarningfCtx(ctx, "re-lookup for chunk %s returned no locations, skipping retry", fileId) return fmt.Errorf("re-lookup chunk %s returned no locations", fileId) } - if urlSlicesEqual(oldUrls, newUrls) { + if util_http.SameUrls(oldUrls, newUrls) { glog.V(0).InfofCtx(ctx, "re-lookup returned same locations for chunk %s, skipping retry", fileId) return originalErr } diff --git a/weed/util/http/http_global_client_util.go b/weed/util/http/http_global_client_util.go index c57c0a489..cd48f2968 100644 --- a/weed/util/http/http_global_client_util.go +++ b/weed/util/http/http_global_client_util.go @@ -544,7 +544,52 @@ func (r *CountingReader) Read(p []byte) (n int, err error) { return n, err } -func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string) (n int, err error) { +// refreshedUrls asks for a fresh location list and reports whether it is worth +// retrying on: a list that comes back empty, or identical to the one that just +// failed everywhere, says the locations were never the problem. +func refreshedUrls(ctx context.Context, refreshUrls RefreshUrlsFunc, current []string, fileId string) ([]string, bool) { + if refreshUrls == nil { + return nil, false + } + fresh := refreshUrls() + if len(fresh) == 0 || SameUrls(current, fresh) { + return nil, false + } + glog.V(0).InfofCtx(ctx, "chunk %s failed on every known location, retrying on %d fresh ones", fileId, len(fresh)) + return fresh, true +} + +// SameUrls reports whether two location lists hold the same URLs, regardless of +// order: lookups shuffle the locations they return, so comparing positionally +// would read a reshuffle of the very same replicas as a fresh set. +func SameUrls(a, b []string) bool { + if len(a) != len(b) { + return false + } + counts := make(map[string]int, len(a)) + for _, url := range a { + counts[url]++ + } + for _, url := range b { + if counts[url] == 0 { + return false + } + counts[url]-- + } + return true +} + +// RefreshUrlsFunc supplies a fresh location list for a chunk. The retry loops +// call it once, after every location in the list they were given has failed, +// which is the point at which the list itself is the likely problem. Returning +// nil or the same list leaves the caller on the original locations. +type RefreshUrlsFunc func() []string + +// RetriedFetchChunkData reads a chunk, trying every location before backing off +// and trying them again. refreshUrls may be nil; when it is not, a pass in which +// every location failed is treated as a stale list rather than a slow cluster, +// and the fresh list is tried immediately instead of after the next backoff. +func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls RefreshUrlsFunc) (n int, err error) { loadJwtConfigOnce.Do(loadJwtConfig) var jwt security.EncodedJwt @@ -555,7 +600,7 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri // For unencrypted, non-gzipped full chunks, use direct buffer read // This avoids the 64KB intermediate buffer and callback overhead if cipherKey == nil && !isGzipped && isFullChunk { - return retriedFetchChunkDataDirect(ctx, buffer, urlStrings, string(jwt)) + return retriedFetchChunkDataDirect(ctx, buffer, urlStrings, string(jwt), fileId, refreshUrls) } var shouldRetry bool @@ -604,6 +649,11 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri } } if err != nil && shouldRetry { + if fresh, ok := refreshedUrls(ctx, refreshUrls, urlStrings, fileId); ok { + urlStrings, refreshUrls = fresh, nil + continue + } + refreshUrls = nil glog.V(0).InfofCtx(ctx, "retry reading in %v", waitTime) // Sleep with proper context cancellation and timer cleanup timer := time.NewTimer(waitTime) @@ -626,7 +676,7 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri // retriedFetchChunkDataDirect reads chunk data directly into the buffer without // intermediate buffering. This reduces memory copies and improves throughput // for large chunk reads. -func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings []string, jwt string) (n int, err error) { +func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings []string, jwt, fileId string, refreshUrls RefreshUrlsFunc) (n int, err error) { var shouldRetry bool for waitTime := time.Second; waitTime < util.RetryWaitTime; waitTime += waitTime / 2 { @@ -654,6 +704,11 @@ func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings } if err != nil && shouldRetry { + if fresh, ok := refreshedUrls(ctx, refreshUrls, urlStrings, fileId); ok { + urlStrings, refreshUrls = fresh, nil + continue + } + refreshUrls = nil glog.V(0).InfofCtx(ctx, "retry reading in %v", waitTime) timer := time.NewTimer(waitTime) select { diff --git a/weed/util/http/http_global_client_util_test.go b/weed/util/http/http_global_client_util_test.go index 74b103559..16426cb99 100644 --- a/weed/util/http/http_global_client_util_test.go +++ b/weed/util/http/http_global_client_util_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "testing" + "time" ) func TestAppendQueryParameter(t *testing.T) { @@ -132,3 +133,83 @@ func TestDeleteProxiedReturnsInvalidRequestErrorBeforeAddingAuth(t *testing.T) { t.Fatal("expected invalid request error") } } + +// TestRetriedFetchChunkDataRetriesFreshUrlsImmediately covers the case the +// refresh hook exists for: every location the caller knew about is gone, and +// the data is live somewhere the caller has not heard of yet. The read must +// land on the fresh location without first sitting through the backoff ladder. +func TestRetriedFetchChunkDataRetriesFreshUrlsImmediately(t *testing.T) { + payload := []byte("chunk contents") + live := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write(payload) + })) + defer live.Close() + + // A port nothing listens on: the address is well formed, the dial fails. + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + deadURL := dead.URL + dead.Close() + + refreshed := 0 + buffer := make([]byte, len(payload)) + start := time.Now() + n, err := RetriedFetchChunkData(context.Background(), buffer, []string{deadURL + "/3,abc"}, nil, false, true, 0, "3,abc", + func() []string { + refreshed++ + return []string{live.URL + "/3,abc"} + }) + elapsed := time.Since(start) + + if err != nil { + t.Fatalf("fetch with a refreshed location: %v", err) + } + if n != len(payload) || string(buffer[:n]) != string(payload) { + t.Fatalf("got %q, want %q", buffer[:n], payload) + } + if refreshed != 1 { + t.Fatalf("refresh called %d times, want exactly 1", refreshed) + } + // The first backoff is a full second; landing well under it is the point. + if elapsed > 500*time.Millisecond { + t.Fatalf("took %v, expected the retry to skip the backoff", elapsed) + } +} + +// TestRetriedFetchChunkDataKeepsBackoffWhenLocationsAreUnchanged makes sure a +// refresh that returns the same list is treated as "the locations were never +// the problem" rather than as a reason to spin. +func TestRetriedFetchChunkDataKeepsBackoffWhenLocationsAreUnchanged(t *testing.T) { + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + deadURL := dead.URL + dead.Close() + + urls := []string{deadURL + "/3,abc"} + refreshed := 0 + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + buffer := make([]byte, 4) + _, err := RetriedFetchChunkData(ctx, buffer, urls, nil, false, true, 0, "3,abc", func() []string { + refreshed++ + return urls + }) + if err == nil { + t.Fatal("expected the fetch to fail against a dead location") + } + if refreshed != 1 { + t.Fatalf("refresh called %d times, want exactly 1", refreshed) + } +} + +func TestSameUrlsIgnoresOrder(t *testing.T) { + a := []string{"http://a:8080/3,x", "http://b:8080/3,x"} + if !SameUrls(a, []string{a[1], a[0]}) { + t.Fatal("a reshuffle of the same locations should not count as fresh") + } + if SameUrls(a, []string{a[0], "http://c:8080/3,x"}) { + t.Fatal("a different location should count as fresh") + } + if SameUrls(a, a[:1]) { + t.Fatal("a shorter list should count as fresh") + } +}