A set's download reported as the whole set's, not the part being fetched
CI / test (push) Successful in 9m15s
CI / test (push) Successful in 9m15s
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
a5fefbdaaa
commit
d5dcf78836
4 files changed
+66
-8
No files matched your search
@@ -26,6 +26,14 @@ is only released once its **exit criterion** is met, not when its code lands.
|
||||
Until v1.0.0 the minor version is the breaking one: configuration formats and Go
|
||||
APIs change between minors without a deprecation period.
|
||||
|
||||
## Unreleased
|
||||
|
||||
### Fixed
|
||||
|
||||
- **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.
|
||||
|
||||
## v0.16.1 — 2026-10-30
|
||||
|
||||
The mirror page's Strata tab, Strata's ready-made packs served with Strata fetching them, and two fixes found
|
||||
|
||||
@@ -202,7 +202,7 @@ func (f *Fetcher) ensure(ctx context.Context, source string, allowPickle bool, w
|
||||
|
||||
// fetch downloads a source to a target, resuming and verifying.
|
||||
func (f *Fetcher) fetch(ctx context.Context, source, target string) (string, error) {
|
||||
return f.fetchAs(ctx, source, source, target)
|
||||
return f.fetchAs(ctx, source, source, target, setShare{})
|
||||
}
|
||||
|
||||
// fetchChecked is fetch for a file whose sha256 is known: a download that
|
||||
@@ -287,7 +287,7 @@ func ChecksumPath(file string) string { return checksumPath(file) }
|
||||
//
|
||||
// Each file is a "download" span, with how many bytes it took and where a
|
||||
// resumed one started from.
|
||||
func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string) (_ string, err error) {
|
||||
func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string, share setShare) (_ string, err error) {
|
||||
ctx, span := obs.Tracer(tracerName).Start(ctx, "download",
|
||||
trace.WithSpanKind(trace.SpanKindClient),
|
||||
trace.WithAttributes(attribute.String("llmgw.weights.source", redact(source))))
|
||||
@@ -351,8 +351,8 @@ func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string)
|
||||
}
|
||||
|
||||
total := existing + resp.ContentLength
|
||||
f.report(Progress{Source: reportAs, BytesDone: existing, BytesTotal: total})
|
||||
written, err := f.copyWithProgress(ctx, out, resp.Body, reportAs, existing, total)
|
||||
f.report(share.of(Progress{Source: reportAs, BytesDone: existing, BytesTotal: total}))
|
||||
written, err := f.copyWithProgress(ctx, out, resp.Body, reportAs, existing, total, share)
|
||||
span.SetAttributes(
|
||||
attribute.Int64("llmgw.weights.resumed_from", existing),
|
||||
attribute.Int64("llmgw.weights.bytes", written),
|
||||
@@ -376,7 +376,7 @@ func (f *Fetcher) fetchAs(ctx context.Context, source, reportAs, target string)
|
||||
}
|
||||
|
||||
func (f *Fetcher) copyWithProgress(ctx context.Context, dst io.Writer, src io.Reader,
|
||||
source string, done, total int64,
|
||||
source string, done, total int64, share setShare,
|
||||
) (int64, error) {
|
||||
buf := make([]byte, 4<<20)
|
||||
last := time.Now()
|
||||
@@ -394,12 +394,12 @@ func (f *Fetcher) copyWithProgress(ctx context.Context, dst io.Writer, src io.Re
|
||||
// hundred gigabytes is twenty-five thousand callbacks, and nobody
|
||||
// needs to be told that often.
|
||||
if time.Since(last) > 2*time.Second {
|
||||
f.report(Progress{Source: source, BytesDone: done, BytesTotal: total})
|
||||
f.report(share.of(Progress{Source: source, BytesDone: done, BytesTotal: total}))
|
||||
last = time.Now()
|
||||
}
|
||||
}
|
||||
if errors.Is(err, io.EOF) {
|
||||
f.report(Progress{Source: source, BytesDone: done, BytesTotal: total})
|
||||
f.report(share.of(Progress{Source: source, BytesDone: done, BytesTotal: total}))
|
||||
return done, nil
|
||||
}
|
||||
if err != nil {
|
||||
@@ -408,6 +408,18 @@ func (f *Fetcher) copyWithProgress(ctx context.Context, dst io.Writer, src io.Re
|
||||
}
|
||||
}
|
||||
|
||||
// setShare places one file's download within its set's: the bytes of the
|
||||
// files before it, and the whole set's. Zero for a file on its own.
|
||||
type setShare struct{ before, total int64 }
|
||||
|
||||
// of is a file's progress as the set's.
|
||||
func (s setShare) of(p Progress) Progress {
|
||||
if s.total > 0 {
|
||||
p.BytesDone, p.BytesTotal = s.before+p.BytesDone, s.total
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func (f *Fetcher) report(p Progress) {
|
||||
if f.OnProgress != nil {
|
||||
f.OnProgress(p)
|
||||
|
||||
@@ -89,6 +89,12 @@ func (f *Fetcher) ensureSet(ctx context.Context, source string, allowPickle bool
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return "", err
|
||||
}
|
||||
// One number for the whole set, as the shards waiting on it see it: the
|
||||
// file being fetched was shown alone, a 46 GB part of 188.
|
||||
var setTotal, before int64
|
||||
for _, file := range m.Files {
|
||||
setTotal += file.Bytes
|
||||
}
|
||||
for _, file := range m.Files {
|
||||
// Rejected rather than joined: a name with a separator in it would
|
||||
// write outside the directory the mirror said it was in, and the
|
||||
@@ -98,15 +104,17 @@ func (f *Fetcher) ensureSet(ctx context.Context, source string, allowPickle bool
|
||||
}
|
||||
target := filepath.Join(dir, file.Name)
|
||||
if st, err := os.Stat(target); err == nil && st.Size() == file.Bytes {
|
||||
before += file.Bytes
|
||||
continue
|
||||
}
|
||||
// Reported under the set's own address, because that is what the
|
||||
// shards are waiting on. Reusing fetch rather than writing a second
|
||||
// downloader keeps resume, the .part rename and the progress callback
|
||||
// in one place.
|
||||
if _, err := f.fetchAs(ctx, source+file.Name, source, target); err != nil {
|
||||
if _, err := f.fetchAs(ctx, source+file.Name, source, target, setShare{before: before, total: setTotal}); err != nil {
|
||||
return "", err
|
||||
}
|
||||
before += file.Bytes
|
||||
}
|
||||
|
||||
// Each file against the checksum the mirror took when it fetched it,
|
||||
|
||||
@@ -204,3 +204,33 @@ func TestASetStillDownloadingOnTheMirrorSaysSo(t *testing.T) {
|
||||
t.Fatalf("got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A set's download is one number climbing against the whole set, not each
|
||||
// part's own: a 46 GB part of 188 was shown as the download.
|
||||
func TestASetsProgressIsTheWholeSets(t *testing.T) {
|
||||
f, srv, _ := mirrorOf(t, map[string]string{
|
||||
"m-00001-of-00002.gguf": strings.Repeat("a", 600),
|
||||
"m-00002-of-00002.gguf": strings.Repeat("b", 400),
|
||||
})
|
||||
var seen []Progress
|
||||
f.OnProgress = func(p Progress) { seen = append(seen, p) }
|
||||
if _, err := f.Ensure(t.Context(), srv.URL+"/m/v/", false); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(seen) == 0 {
|
||||
t.Fatal("no progress reported")
|
||||
}
|
||||
last := int64(-1)
|
||||
for _, p := range seen {
|
||||
if p.BytesTotal != 1000 {
|
||||
t.Fatalf("reported against %d, not the set's 1000: %+v", p.BytesTotal, seen)
|
||||
}
|
||||
if p.BytesDone < last {
|
||||
t.Fatalf("went back: %+v", seen)
|
||||
}
|
||||
last = p.BytesDone
|
||||
}
|
||||
if last != 1000 {
|
||||
t.Errorf("ended at %d", last)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user