From a88624747ef4481fe24b237b370dd8a72c673e95 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Mon, 2 Nov 2026 08:29:25 +0100 Subject: [PATCH] A set wanted by several replicas on one node is fetched once Single files were shared between the shards and replicas asking for them at the same time; sets were not, and four replicas on one node's four cards wrote the same .part files over one another until all but one failed. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 6 +++++ internal/agent/weights/fetch.go | 18 ++++++++++++--- internal/agent/weights/set_test.go | 37 ++++++++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5df569c..51daabc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -80,6 +80,12 @@ APIs change between minors without a deprecation period. ### Fixed +- **Several replicas of a multi-file model on one node fetched it once + each, over one another.** Four replicas placed on a node's four cards + wanted the same safetensors set at the same moment, each wrote the same + `.part` files, and all but one failed to start ("rename … no such file") + until a retry found the set finished. A set is now fetched once per node + and every replica waiting on it gets it, as single files always were. - **A model of several files showed one file's download as the whole.** A node fetching a set reported the part it was on — "59% of 46.1 GB" for a 188 GB model — instead of the set: it now reports the set's bytes against its whole size. diff --git a/internal/agent/weights/fetch.go b/internal/agent/weights/fetch.go index 686b5ff..f8005da 100644 --- a/internal/agent/weights/fetch.go +++ b/internal/agent/weights/fetch.go @@ -146,7 +146,11 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w // file, and a set has to be handled before any of it. if IsSet(source) { outcome(ctx, "set") - return f.ensureSet(ctx, source, allowPickle) + // Shared like a single file: four replicas placed on one node's + // four cards want the same set at the same instant, and each + // fetching it wrote the same .part files over one another — three + // failed renaming a file another had already renamed. + return f.shared(ctx, source, func() (string, error) { return f.ensureSet(ctx, source, allowPickle) }) } if !isRemote(source) { if !isPath(source) { @@ -178,6 +182,15 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w // Not the file the catalog names: fetched again below, and checked. } + return f.shared(ctx, source, func() (string, error) { + outcome(ctx, "downloaded") + return f.fetchChecked(ctx, source, target, want) + }) +} + +// shared runs fetch once for every caller asking for source at the same +// time; the others wait for its answer. +func (f *Fetcher) shared(ctx context.Context, source string, fetch func() (string, error)) (string, error) { f.mu.Lock() if f.inflight == nil { f.inflight = map[string]*fetchCall{} @@ -196,8 +209,7 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w f.inflight[source] = call f.mu.Unlock() - outcome(ctx, "downloaded") - call.path, call.err = f.fetchChecked(ctx, source, target, want) + call.path, call.err = fetch() close(call.done) f.mu.Lock() diff --git a/internal/agent/weights/set_test.go b/internal/agent/weights/set_test.go index 1c7e6a5..1a95ff0 100644 --- a/internal/agent/weights/set_test.go +++ b/internal/agent/weights/set_test.go @@ -10,7 +10,9 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "testing" + "time" ) // mirrorOf serves a manifest at "/m/v/" and the files beneath it. @@ -270,3 +272,38 @@ func TestASetBeingCheckedSaysSo(t *testing.T) { t.Errorf("%+v", checking) } } + +// Four replicas on one node's four cards start together and want the same +// set: it is fetched once, and every one of them gets it. Each fetching it +// alone wrote the same .part files over one another, and three failed. +func TestASetWantedByFourReplicasAtOnceIsFetchedOnce(t *testing.T) { + body := strings.Repeat("a", 1<<20) + var hits atomic.Int64 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/") { + fmt.Fprintf(w, `{"model":"m","version":"v","files":[{"name":"a.safetensors","bytes":%d},{"name":"b.safetensors","bytes":%d}]}`, + len(body), len(body)) + return + } + hits.Add(1) + time.Sleep(100 * time.Millisecond) // long enough for all four to ask + _, _ = w.Write([]byte(body)) + })) + t.Cleanup(srv.Close) + f := &Fetcher{CacheDir: t.TempDir()} + + var wg sync.WaitGroup + errs := make([]error, 4) + for i := range errs { + wg.Go(func() { _, errs[i] = f.Ensure(t.Context(), srv.URL+"/m/v/", false) }) + } + wg.Wait() + for i, err := range errs { + if err != nil { + t.Errorf("replica %d: %v", i, err) + } + } + if n := hits.Load(); n != 2 { + t.Errorf("%d downloads of a set of two files", n) + } +}