diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index 57cef8fa7..83bc90135 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -110,7 +110,7 @@ func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileI return err } jwt := JwtForVolumeServer(fileId) - if _, err = retriedStreamFetchChunkData(ctx, bytesBuffer, urlStrings, jwt, cipherKey, isGzipped, true, 0, 0); err == nil { + if _, err = retriedStreamFetchChunkData(ctx, bytesBuffer, urlStrings, jwt, cipherKey, isGzipped, true, 0, 0, refreshUrls(ctx, invalidator, lookupFileIdFn, fileId)); err == nil { return nil } if ctxErr := ctx.Err(); ctxErr != nil { @@ -121,7 +121,7 @@ func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileI return retryFetchWithFreshLocations(ctx, invalidator, lookupFileIdFn, fileId, urlStrings, err, func(newUrls []string) error { // the failed attempt may have streamed a partial prefix into the buffer bytesBuffer.Reset() - _, retryErr := retriedStreamFetchChunkData(ctx, bytesBuffer, newUrls, jwt, cipherKey, isGzipped, true, 0, 0) + _, retryErr := retriedStreamFetchChunkData(ctx, bytesBuffer, newUrls, jwt, cipherKey, isGzipped, true, 0, 0, nil) return retryErr }) } @@ -135,7 +135,10 @@ func fetchChunkRange(ctx context.Context, buffer []byte, lookupFileIdFn wdclient 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) { +// retriedStreamFetchChunkData streams a chunk from the first location that +// answers. refreshUrls may be nil; when a location failed and a later one +// answered, it is called so the reads that follow start from a fresh list. +func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrings []string, jwt string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, size int, refreshUrls util_http.RefreshUrlsFunc) (written int64, err error) { var shouldRetry bool var totalWritten int @@ -149,7 +152,8 @@ func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrin } retriedCnt := 0 - for _, urlString := range urlStrings { + var failed bool + for _, urlString := range util_http.ReachableFirst(urlStrings) { // Check for context cancellation before each volume server request select { case <-ctx.Done(): @@ -191,11 +195,15 @@ func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrin break } if err != nil { + failed = true glog.V(0).InfofCtx(ctx, "read %s failed, err: %v", urlString, err) } else { break } } + if err == nil && failed && refreshUrls != nil { + refreshUrls() + } // all nodes have tried it if retriedCnt == len(urlStrings) { break diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index e9243be69..9e51722f4 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -593,6 +593,35 @@ func TestFetchWholeChunkRetriesFreshLocations(t *testing.T) { }, decoded.Chunks) } +// TestFetchWholeChunkRefreshesLocationsAfterPartialFailure is the manifest read +// whose cached list still names a dead replica ahead of a live one: the read +// succeeds, and the cached entry must still be dropped and looked up again. +func TestFetchWholeChunkRefreshesLocationsAfterPartialFailure(t *testing.T) { + manifestBytes, err := proto.Marshal(&filer_pb.FileChunkManifest{ + Chunks: []*filer_pb.FileChunk{{FileId: "100,abc", Offset: 0, Size: 8}}, + }) + assert.NoError(t, err) + + liveURL := manifestServer(t, manifestBytes).URL + "/5,abc" + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + deadURL := dead.URL + dead.Close() + + lookup := &stagedLookup{ + staleUrls: []string{deadURL + "/5,abc", liveURL}, + freshUrls: []string{liveURL}, + } + inv := &countingInvalidator{} + bytesBuffer := fetchManifestBuffer(t) + + assert.NoError(t, fetchWholeChunk(context.Background(), bytesBuffer, lookup.lookup, "5,abc", nil, false, inv)) + assert.Equal(t, int32(1), inv.invalidations.Load()) + assert.Equal(t, int32(2), lookup.calls.Load()) + decoded := &filer_pb.FileChunkManifest{} + assert.NoError(t, proto.Unmarshal(bytesBuffer.Bytes(), decoded)) + assertEqualChunks(t, []*filer_pb.FileChunk{{FileId: "100,abc", Offset: 0, Size: 8}}, decoded.Chunks) +} + // TestFetchWholeChunkWithoutInvalidator keeps the old behavior for callers whose // lookup function has no cache to drop: the fetch error surfaces as it is. func TestFetchWholeChunkWithoutInvalidator(t *testing.T) { diff --git a/weed/filer/reader_at.go b/weed/filer/reader_at.go index 8ba08d489..9e31228a7 100644 --- a/weed/filer/reader_at.go +++ b/weed/filer/reader_at.go @@ -346,7 +346,7 @@ func (c *ChunkReadAt) readChunkSliceAt(ctx context.Context, buffer []byte, chunk return n, err } return fetchChunkRange(ctx, buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset), - c.readerCache.refreshUrls(ctx, chunkView.FileId)) + refreshUrls(ctx, c.readerCache.cacheInvalidator, c.readerCache.lookupFileIdFn, 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 f2a139c92..577d6776e 100644 --- a/weed/filer/reader_cache.go +++ b/weed/filer/reader_cache.go @@ -104,25 +104,6 @@ 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() @@ -289,7 +270,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, s.parent.refreshUrls(ctx, s.chunkFileId)) + _, fetchErr := s.parent.fetchChunkDataFn(ctx, data, urlStrings, s.cipherKey, s.isGzipped, true, 0, s.chunkFileId, refreshUrls(ctx, s.parent.cacheInvalidator, s.parent.lookupFileIdFn, 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 b92b17c1b..585f65bbb 100644 --- a/weed/filer/reader_cache_test.go +++ b/weed/filer/reader_cache_test.go @@ -3,11 +3,14 @@ package filer import ( "context" "fmt" - util_http "github.com/seaweedfs/seaweedfs/weed/util/http" + "net/http" + "net/http/httptest" "sync" "sync/atomic" "testing" "time" + + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) // mockChunkCacheForReaderCache implements chunk cache for testing @@ -147,6 +150,47 @@ func TestReaderCacheRetryAfterCacheInvalidation(t *testing.T) { } } +// TestReaderCacheRefreshesLocationsAfterPartialFailure is the mount reading +// through a cached location list that still names a replica the master has +// dropped: the read succeeds on the other replica, and the cached entry must +// be dropped and looked up again so the next read starts without the dead one. +func TestReaderCacheRefreshesLocationsAfterPartialFailure(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() + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + deadURL := dead.URL + dead.Close() + + fileId := "3,abc" + invalidator := &mockCacheInvalidatorForReaderCache{} + var lookupCount int32 + lookupFn := func(ctx context.Context, requestedFileId string) ([]string, error) { + atomic.AddInt32(&lookupCount, 1) + if atomic.LoadInt32(&invalidator.calls) == 0 { + return []string{deadURL + "/" + fileId, live.URL + "/" + fileId}, nil + } + return []string{live.URL + "/" + fileId}, nil + } + + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), lookupFn, invalidator) + defer rc.destroy() + + buffer := make([]byte, len(payload)) + n, err := rc.ReadChunkAt(context.Background(), buffer, fileId, nil, false, 0, len(payload), false) + if err != nil || string(buffer[:n]) != string(payload) { + t.Fatalf("got %q, %v; want %q", buffer[:n], err, payload) + } + if got := atomic.LoadInt32(&invalidator.calls); got != 1 { + t.Fatalf("expected the stale entry to be invalidated once, got %d", got) + } + if got := atomic.LoadInt32(&lookupCount); got != 2 { + t.Fatalf("expected a lookup before the read and one after invalidation, got %d", got) + } +} + func TestReaderCacheRemovesFailedDownloader(t *testing.T) { cache := newMockChunkCacheForReaderCache() fileId := "425141,failed" diff --git a/weed/filer/stream.go b/weed/filer/stream.go index 5b4d4cbd5..096f46b17 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -102,6 +102,25 @@ func PrepareStreamContent(masterClient wdclient.HasLookupFileIdFunction, jwtFunc type VolumeServerJwtFunction func(fileId string) string +// refreshUrls lets a fetch loop relearn a chunk's locations inside a single +// read: drop the cached entry and look it up again, whether every location +// failed or one did while another answered. Nil when there is nothing to +// invalidate against. +func refreshUrls(ctx context.Context, invalidator CacheInvalidator, lookupFn wdclient.LookupFileIdFunctionType, fileId string) util_http.RefreshUrlsFunc { + if invalidator == nil || lookupFn == nil { + return nil + } + return func() []string { + invalidator.InvalidateCache(fileId) + urls, err := lookupFn(ctx, fileId) + if err != nil { + glog.V(0).InfofCtx(ctx, "re-lookup chunk %s: %v", fileId, err) + return nil + } + return urls + } +} + // 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 @@ -199,7 +218,8 @@ func PrepareStreamContentWithThrottler(ctx context.Context, masterClient wdclien urlStrings := fileId2Url[chunkView.FileId] start := time.Now() jwt := jwtFunc(chunkView.FileId) - written, err := retriedStreamFetchChunkData(ctx, writer, urlStrings, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.OffsetInChunk, int(chunkView.ViewSize)) + invalidator, _ := masterClient.(CacheInvalidator) + written, err := retriedStreamFetchChunkData(ctx, writer, urlStrings, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.OffsetInChunk, int(chunkView.ViewSize), refreshUrls(ctx, invalidator, masterClient.GetLookupFileIdFunction(), chunkView.FileId)) if err != nil && ctx.Err() != nil { return ctx.Err() @@ -207,9 +227,8 @@ func PrepareStreamContentWithThrottler(ctx context.Context, masterClient wdclien // If read failed, try to invalidate cache and re-lookup if err != nil && written == 0 { - invalidator, _ := masterClient.(CacheInvalidator) err = retryFetchWithFreshLocations(ctx, invalidator, masterClient.GetLookupFileIdFunction(), chunkView.FileId, urlStrings, err, func(newUrls []string) error { - _, refetchErr := retriedStreamFetchChunkData(ctx, writer, newUrls, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.OffsetInChunk, int(chunkView.ViewSize)) + _, refetchErr := retriedStreamFetchChunkData(ctx, writer, newUrls, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.OffsetInChunk, int(chunkView.ViewSize), nil) if refetchErr == nil { // Update the map so subsequent references use fresh URLs fileId2Url[chunkView.FileId] = newUrls diff --git a/weed/filer/stream_prefetch.go b/weed/filer/stream_prefetch.go index 33d01be25..3d6bed09f 100644 --- a/weed/filer/stream_prefetch.go +++ b/weed/filer/stream_prefetch.go @@ -10,6 +10,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/util" + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" "github.com/seaweedfs/seaweedfs/weed/util/mem" "github.com/seaweedfs/seaweedfs/weed/wdclient" ) @@ -46,6 +47,7 @@ func streamChunksPrefetched( prefetchAhead int, ) error { downloadThrottler := util.NewWriteThrottler(downloadMaxBytesPs) + invalidator, _ := masterClient.(CacheInvalidator) // Create a local cancellable context so the consumer can stop the producer // and all in-flight fetch goroutines on error (e.g., client disconnect). @@ -98,14 +100,14 @@ func streamChunksPrefetched( } // Launch fetch goroutine - go func(cv *ChunkView, urls []string, jwt string, pw *io.PipeWriter, res *chunkPipeResult) { + go func(cv *ChunkView, urls []string, jwt string, pw *io.PipeWriter, res *chunkPipeResult, refresh util_http.RefreshUrlsFunc) { defer func() { <-sem }() // release semaphore defer close(res.done) written, err := retriedStreamFetchChunkData( localCtx, pw, urls, jwt, cv.CipherKey, cv.IsGzipped, cv.IsFullChunk(), - cv.OffsetInChunk, int(cv.ViewSize), + cv.OffsetInChunk, int(cv.ViewSize), refresh, ) res.written = written res.fetchErr = err @@ -115,7 +117,7 @@ func streamChunksPrefetched( } else { pw.Close() } - }(chunkView, urlStrings, jwt, pw, result) + }(chunkView, urlStrings, jwt, pw, result, refreshUrls(localCtx, invalidator, masterClient.GetLookupFileIdFunction(), chunkView.FileId)) // Send result to consumer (blocks if channel full, back-pressuring producer) select { @@ -247,7 +249,7 @@ func retryWithCacheInvalidation( _, err := retriedStreamFetchChunkData( ctx, writer, newUrls, jwt, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), - chunkView.OffsetInChunk, int(chunkView.ViewSize), + chunkView.OffsetInChunk, int(chunkView.ViewSize), nil, ) return err }) diff --git a/weed/util/http/http_global_client_util.go b/weed/util/http/http_global_client_util.go index cd48f2968..23794e45e 100644 --- a/weed/util/http/http_global_client_util.go +++ b/weed/util/http/http_global_client_util.go @@ -111,8 +111,10 @@ func GetAuthenticated(url, jwt string) ([]byte, bool, error) { response, err := GetGlobalHttpClient().Do(request) if err != nil { + recordUnreachable(request.URL.Host) return nil, true, err } + recordReachable(request.URL.Host) defer CloseResponse(response) var reader io.ReadCloser @@ -392,8 +394,12 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte, r, err := GetGlobalHttpClient().Do(req) if err != nil { + if ctx.Err() == nil { + recordUnreachable(req.URL.Host) + } return true, err } + recordReachable(req.URL.Host) defer CloseResponse(r) if r.StatusCode >= 400 { if r.StatusCode == http.StatusNotFound { @@ -580,15 +586,19 @@ func SameUrls(a, b []string) bool { } // 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. +// call it at most once: after every location in the list failed, to retry on +// the fresh list at once, or after a later location answered for one that +// failed, so the next read starts from what the cluster knows now instead of +// trying the same dead replica again. 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. +// A pass that failed on one location and succeeded on another refreshes the +// list for the reads that follow. 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) @@ -613,7 +623,8 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri default: } - for _, urlString := range urlStrings { + var failed bool + for _, urlString := range ReachableFirst(urlStrings) { // Check for context cancellation before each volume server request select { case <-ctx.Done(): @@ -643,11 +654,15 @@ func RetriedFetchChunkData(ctx context.Context, buffer []byte, urlStrings []stri break } if err != nil { + failed = true glog.V(0).InfofCtx(ctx, "read %s failed, err: %v", urlString, err) } else { break } } + if err == nil && failed && refreshUrls != nil { + refreshUrls() + } if err != nil && shouldRetry { if fresh, ok := refreshedUrls(ctx, refreshUrls, urlStrings, fileId); ok { urlStrings, refreshUrls = fresh, nil @@ -686,7 +701,8 @@ func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings default: } - for _, urlString := range urlStrings { + var failed bool + for _, urlString := range ReachableFirst(urlStrings) { select { case <-ctx.Done(): return 0, ctx.Err() @@ -695,11 +711,15 @@ func retriedFetchChunkDataDirect(ctx context.Context, buffer []byte, urlStrings n, shouldRetry, err = readUrlDirectToBuffer(ctx, AppendQueryParameter(urlString, "readDeleted", "true"), jwt, buffer) if err == nil { + if failed && refreshUrls != nil { + refreshUrls() + } return n, nil } if !shouldRetry { break } + failed = true glog.V(0).InfofCtx(ctx, "read %s failed, err: %v", urlString, err) } @@ -737,8 +757,12 @@ func readUrlDirectToBuffer(ctx context.Context, fileUrl, jwt string, buffer []by r, err := GetGlobalHttpClient().Do(req) if err != nil { + if ctx.Err() == nil { + recordUnreachable(req.URL.Host) + } return 0, true, err } + recordReachable(req.URL.Host) defer CloseResponse(r) if r.StatusCode >= 400 { diff --git a/weed/util/http/http_global_client_util_test.go b/weed/util/http/http_global_client_util_test.go index 16426cb99..bd3d98d6c 100644 --- a/weed/util/http/http_global_client_util_test.go +++ b/weed/util/http/http_global_client_util_test.go @@ -201,6 +201,50 @@ func TestRetriedFetchChunkDataKeepsBackoffWhenLocationsAreUnchanged(t *testing.T } } +// TestRetriedFetchChunkDataRefreshesLocationsAfterPartialFailure covers a read +// that fails on a cached dead replica and succeeds on the next one: the read +// itself is fine, but the list it came from is stale and must be refreshed so +// later reads do not start with the dead replica again. Both the direct and +// the streaming read paths take this branch. +func TestRetriedFetchChunkDataRefreshesLocationsAfterPartialFailure(t *testing.T) { + forgetUnreachable(t) + payload := []byte("chunk contents") + live := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write(payload) + })) + defer live.Close() + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + deadURL := dead.URL + dead.Close() + + for _, isFullChunk := range []bool{true, false} { + forgetUnreachable(t) + refreshed := 0 + refresh := func() []string { + refreshed++ + return []string{live.URL + "/3,abc"} + } + buffer := make([]byte, len(payload)) + n, err := RetriedFetchChunkData(context.Background(), buffer, []string{deadURL + "/3,abc", live.URL + "/3,abc"}, nil, false, isFullChunk, 0, "3,abc", refresh) + if err != nil || string(buffer[:n]) != string(payload) { + t.Fatalf("fullChunk=%v: got %q, %v; want %q", isFullChunk, buffer[:n], err, payload) + } + if refreshed != 1 { + t.Fatalf("fullChunk=%v: refresh called %d times after a partial failure, want exactly 1", isFullChunk, refreshed) + } + + // a read answered by the first location leaves the list alone + refreshed = 0 + n, err = RetriedFetchChunkData(context.Background(), buffer, []string{live.URL + "/3,abc", deadURL + "/3,abc"}, nil, false, isFullChunk, 0, "3,abc", refresh) + if err != nil || string(buffer[:n]) != string(payload) { + t.Fatalf("fullChunk=%v: got %q, %v; want %q", isFullChunk, buffer[:n], err, payload) + } + if refreshed != 0 { + t.Fatalf("fullChunk=%v: refresh called %d times without a failure, want none", isFullChunk, 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]}) { diff --git a/weed/util/http/reachability.go b/weed/util/http/reachability.go new file mode 100644 index 000000000..ad277b3ef --- /dev/null +++ b/weed/util/http/reachability.go @@ -0,0 +1,60 @@ +package http + +import ( + "net/url" + "sync" + "time" +) + +// unreachableRetryInterval is how long a server that failed to answer is tried +// last before a read probes it again. +const unreachableRetryInterval = 30 * time.Second + +// unreachable maps a host to when it last failed to answer a request. A dead +// replica sits in the location list of every volume it held, so this is kept +// per host rather than per volume. +var unreachable sync.Map + +func recordUnreachable(host string) { + unreachable.Store(host, time.Now()) +} + +func recordReachable(host string) { + unreachable.Delete(host) +} + +// ReachableFirst orders urls so that hosts which recently failed to answer +// come last, keeping the order within each group. Once the retry interval has +// passed, the first read to ask claims the probe and tries that host first; +// the others keep it last until the probe settles. A read probes at most one +// host, so a claim is always followed by an attempt and several expired hosts +// are probed by successive reads. +func ReachableFirst(urls []string) []string { + var probe, reachable, unanswered []string + for _, u := range urls { + host := hostOf(u) + failedAt, failed := unreachable.Load(host) + switch { + case !failed: + reachable = append(reachable, u) + case time.Since(failedAt.(time.Time)) < unreachableRetryInterval: + unanswered = append(unanswered, u) + case probe == nil && unreachable.CompareAndSwap(host, failedAt, time.Now()): + probe = append(probe, u) + default: + unanswered = append(unanswered, u) + } + } + if probe == nil && unanswered == nil { + return urls + } + return append(append(probe, reachable...), unanswered...) +} + +func hostOf(rawUrl string) string { + parsed, err := url.Parse(rawUrl) + if err != nil { + return "" + } + return parsed.Host +} diff --git a/weed/util/http/reachability_test.go b/weed/util/http/reachability_test.go new file mode 100644 index 000000000..4901ede9b --- /dev/null +++ b/weed/util/http/reachability_test.go @@ -0,0 +1,112 @@ +package http + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +func forgetUnreachable(t *testing.T) { + t.Helper() + forget := func() { + unreachable.Range(func(host, _ any) bool { + unreachable.Delete(host) + return true + }) + } + forget() + t.Cleanup(forget) +} + +func assertOrder(t *testing.T, got []string, want ...string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("got %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("got %v, want %v", got, want) + } + } +} + +func TestReachableFirstTriesUnansweringHostLast(t *testing.T) { + forgetUnreachable(t) + urls := []string{"http://a:8080/3,x", "http://b:8080/3,x", "http://c:8080/3,x"} + + assertOrder(t, ReachableFirst(urls), urls...) + recordUnreachable("a:8080") + assertOrder(t, ReachableFirst(urls), urls[1], urls[2], urls[0]) + recordReachable("a:8080") + assertOrder(t, ReachableFirst(urls), urls...) +} + +func TestReachableFirstProbesOnceAfterRetryInterval(t *testing.T) { + forgetUnreachable(t) + urls := []string{"http://b:8080/3,x", "http://a:8080/3,x"} + unreachable.Store("a:8080", time.Now().Add(-unreachableRetryInterval)) + + // the first read to come by probes a, the next keeps it last until that settles + assertOrder(t, ReachableFirst(urls), urls[1], urls[0]) + assertOrder(t, ReachableFirst(urls), urls...) + recordReachable("a:8080") + assertOrder(t, ReachableFirst(urls), urls...) +} + +func TestReachableFirstProbesOneExpiredHostPerRead(t *testing.T) { + forgetUnreachable(t) + urls := []string{"http://a:8080/3,x", "http://b:8080/3,x", "http://c:8080/3,x"} + expired := time.Now().Add(-unreachableRetryInterval) + unreachable.Store("a:8080", expired) + unreachable.Store("c:8080", expired) + + assertOrder(t, ReachableFirst(urls), urls[0], urls[1], urls[2]) + assertOrder(t, ReachableFirst(urls), urls[2], urls[1], urls[0]) + assertOrder(t, ReachableFirst(urls), urls[1], urls[0], urls[2]) +} + +// hangupServer accepts the connection and drops it without answering, the way +// a replica behind a broken network does, and counts how often that happened. +func hangupServer(t *testing.T) (*httptest.Server, *int32) { + t.Helper() + var hangups int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&hangups, 1) + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + conn.Close() + })) + t.Cleanup(srv.Close) + return srv, &hangups +} + +func TestRetriedFetchChunkDataTriesUnansweringServerLast(t *testing.T) { + forgetUnreachable(t) + payload := []byte("chunk contents") + live := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write(payload) + })) + defer live.Close() + hangup, hangups := hangupServer(t) + + urls := []string{hangup.URL + "/3,abc", live.URL + "/3,abc"} + for i := 0; i < 3; i++ { + buffer := make([]byte, len(payload)) + n, err := RetriedFetchChunkData(context.Background(), buffer, urls, nil, false, true, 0, "3,abc", nil) + if err != nil { + t.Fatalf("read %d: %v", i, err) + } + if string(buffer[:n]) != string(payload) { + t.Fatalf("read %d got %q, want %q", i, buffer[:n], payload) + } + } + if got := atomic.LoadInt32(hangups); got != 1 { + t.Fatalf("the server that hung up was tried %d times, want once", got) + } +}