httpdl: unify completed counter and dedupe mirror loop; engine: harden untrack
From a Rob Pike pass over the tree: - singleOnce adds deltas to d.completed like pump does, instead of storing absolute offsets, so the counter has one write discipline. - one overMirrors helper replaces the three copy-pasted mirror-fallover loops in probeRetry, fetchSeg and single. - failedDownload becomes a pointer, the only value-type Download impl, so Engine.untrack's == is always pointer identity and can't compare the fields of a value (keying on Stat().ID is out — a torrent's ID changes from source to infohash once metadata loads).
This commit is contained in:
@@ -91,6 +91,10 @@ func (e *Engine) track(d Download) {
|
|||||||
e.mu.Unlock()
|
e.mu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// untrack removes d from the active set. Every Download implementation is a
|
||||||
|
// pointer type, so the == compares identities — never the fields of a value,
|
||||||
|
// which could panic on a non-comparable one. (Keying on Stat().ID is not an
|
||||||
|
// option: a torrent's ID changes from source to infohash once metadata loads.)
|
||||||
func (e *Engine) untrack(d Download) {
|
func (e *Engine) untrack(d Download) {
|
||||||
e.mu.Lock()
|
e.mu.Lock()
|
||||||
defer e.mu.Unlock()
|
defer e.mu.Unlock()
|
||||||
|
|||||||
@@ -536,29 +536,43 @@ func (d *Download) withRetries(ctx context.Context, what string, attempt func()
|
|||||||
return retriesExhausted(what, tries, lastErr)
|
return retriesExhausted(what, tries, lastErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// overMirrors calls fn against each mirror URL in turn, starting at offset start
|
||||||
|
// so callers can spread work across mirrors (segment index) and rotate on retry.
|
||||||
|
// It returns nil on the first success, ctx.Err() if cancelled between mirrors, or
|
||||||
|
// the last error once every mirror is spent.
|
||||||
|
func (d *Download) overMirrors(ctx context.Context, start int, fn func(uri string) error) error {
|
||||||
|
var lastErr error
|
||||||
|
for i := range d.uris {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
if err := fn(d.mirror(start + i)); err != nil {
|
||||||
|
lastErr = err
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return lastErr
|
||||||
|
}
|
||||||
|
|
||||||
// probeRetry runs probe within the shared retry policy (item [18]). The probe
|
// probeRetry runs probe within the shared retry policy (item [18]). The probe
|
||||||
// result is captured through pr because withRetries only carries an error.
|
// result is captured through pr because withRetries only carries an error.
|
||||||
func (d *Download) probeRetry(ctx context.Context) (probeResult, error) {
|
func (d *Download) probeRetry(ctx context.Context) (probeResult, error) {
|
||||||
// Probe mirrors in order (primary first); the first that answers anchors the
|
// Probe mirrors in order (primary first); the first that answers anchors the
|
||||||
// size and validators for the whole download. A dead/404 primary falls over
|
// size and validators for the whole download. A dead/404 primary falls over
|
||||||
// to the next mirror.
|
// to the next mirror.
|
||||||
var lastErr error
|
var pr probeResult
|
||||||
for _, u := range d.uris {
|
err := d.overMirrors(ctx, 0, func(uri string) error {
|
||||||
if ctx.Err() != nil {
|
return d.withRetries(ctx, uri, func() error {
|
||||||
return probeResult{}, ctx.Err()
|
|
||||||
}
|
|
||||||
var pr probeResult
|
|
||||||
err := d.withRetries(ctx, u, func() error {
|
|
||||||
var perr error
|
var perr error
|
||||||
pr, perr = d.probe(ctx, u)
|
pr, perr = d.probe(ctx, uri)
|
||||||
return perr
|
return perr
|
||||||
})
|
})
|
||||||
if err == nil {
|
})
|
||||||
return pr, nil
|
if err != nil {
|
||||||
}
|
return probeResult{}, err
|
||||||
lastErr = err
|
|
||||||
}
|
}
|
||||||
return probeResult{}, lastErr
|
return pr, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// probe issues a one-byte ranged GET to learn the total size, whether the
|
// probe issues a one-byte ranged GET to learn the total size, whether the
|
||||||
@@ -715,24 +729,14 @@ func (d *Download) fetchSeg(ctx context.Context, f *os.File, p *pool, s *seg, to
|
|||||||
// The segment fails only once every mirror is spent. Starting at s.index
|
// The segment fails only once every mirror is spent. Starting at s.index
|
||||||
// spreads the segments across mirrors round-robin, and each attempt resumes
|
// spreads the segments across mirrors round-robin, and each attempt resumes
|
||||||
// from s.offset().
|
// from s.offset().
|
||||||
var lastErr error
|
return d.overMirrors(ctx, s.index, func(uri string) error {
|
||||||
for i := range d.uris {
|
return d.withRetries(ctx, fmt.Sprintf("segment %d", s.index), func() error {
|
||||||
if ctx.Err() != nil {
|
|
||||||
return ctx.Err()
|
|
||||||
}
|
|
||||||
uri := d.mirror(s.index + i)
|
|
||||||
err := d.withRetries(ctx, fmt.Sprintf("segment %d", s.index), func() error {
|
|
||||||
if s.done() {
|
if s.done() {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return d.fetchOnce(ctx, f, p, s, uri, total)
|
return d.fetchOnce(ctx, f, p, s, uri, total)
|
||||||
})
|
})
|
||||||
if err == nil {
|
})
|
||||||
return nil
|
|
||||||
}
|
|
||||||
lastErr = err
|
|
||||||
}
|
|
||||||
return lastErr
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Download) fetchOnce(ctx context.Context, f *os.File, p *pool, s *seg, uri string, total int64) error {
|
func (d *Download) fetchOnce(ctx context.Context, f *os.File, p *pool, s *seg, uri string, total int64) error {
|
||||||
@@ -816,10 +820,9 @@ const speedStartupGrace = 10 * time.Second
|
|||||||
|
|
||||||
// belowSpeed reports whether the bytes transferred between two cumulative
|
// belowSpeed reports whether the bytes transferred between two cumulative
|
||||||
// d.completed samples (prev -> now over one window) put the download at or
|
// d.completed samples (prev -> now over one window) put the download at or
|
||||||
// below limit. A negative delta is not a slow second: d.completed is not
|
// below limit. A negative delta is not a slow second: when a resumed Range is
|
||||||
// strictly monotonic — singleOnce rewrites it with an absolute StoreInt64 and a
|
// answered by a 200 the transfer restarts from offset 0, so d.completed jumps
|
||||||
// resumed Range answered by a 200 resets the offset to 0, so the counter can
|
// backward on that retry. Treating that backward jump as "below limit" would
|
||||||
// jump backward on a retry. Treating that backward jump as "below limit" would
|
|
||||||
// false-abort a healthy download, so a reset (delta < 0) reports false; the
|
// false-abort a healthy download, so a reset (delta < 0) reports false; the
|
||||||
// caller has already advanced its baseline, so the next full window measures the
|
// caller has already advanced its baseline, so the next full window measures the
|
||||||
// real speed. Only a genuine non-negative delta at or under the limit aborts.
|
// real speed. Only a genuine non-negative delta at or under the limit aborts.
|
||||||
@@ -964,23 +967,13 @@ func (d *Download) single(ctx context.Context, out string) error {
|
|||||||
first := true
|
first := true
|
||||||
// One connection at a time, but try each mirror: a mirror that fails its
|
// One connection at a time, but try each mirror: a mirror that fails its
|
||||||
// retry budget falls over to the next (resuming from what's already on disk).
|
// retry budget falls over to the next (resuming from what's already on disk).
|
||||||
var lastErr error
|
return d.overMirrors(ctx, 0, func(uri string) error {
|
||||||
for i := range d.uris {
|
return d.withRetries(ctx, uri, func() error {
|
||||||
if ctx.Err() != nil {
|
|
||||||
return ctx.Err()
|
|
||||||
}
|
|
||||||
uri := d.mirror(i)
|
|
||||||
err := d.withRetries(ctx, uri, func() error {
|
|
||||||
resumeFromDisk := !first || foreignResume
|
resumeFromDisk := !first || foreignResume
|
||||||
first = false
|
first = false
|
||||||
return d.singleOnce(ctx, out, uri, resumeFromDisk)
|
return d.singleOnce(ctx, out, uri, resumeFromDisk)
|
||||||
})
|
})
|
||||||
if err == nil {
|
})
|
||||||
return nil
|
|
||||||
}
|
|
||||||
lastErr = err
|
|
||||||
}
|
|
||||||
return lastErr
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// singleOnce performs one single-stream transfer attempt. When resumeFromDisk is
|
// singleOnce performs one single-stream transfer attempt. When resumeFromDisk is
|
||||||
@@ -1058,7 +1051,7 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
|
|||||||
}
|
}
|
||||||
off += int64(rd)
|
off += int64(rd)
|
||||||
got += int64(rd)
|
got += int64(rd)
|
||||||
atomic.StoreInt64(&d.completed, off)
|
atomic.AddInt64(&d.completed, int64(rd)) // add deltas, as pump does
|
||||||
d.throttle(ctx, rd)
|
d.throttle(ctx, rd)
|
||||||
}
|
}
|
||||||
if err == io.EOF {
|
if err == io.EOF {
|
||||||
|
|||||||
10
main.go
10
main.go
@@ -190,9 +190,9 @@ type failedDownload struct {
|
|||||||
err error
|
err error
|
||||||
}
|
}
|
||||||
|
|
||||||
func (f failedDownload) Name() string { return f.name }
|
func (f *failedDownload) Name() string { return f.name }
|
||||||
func (f failedDownload) Run(context.Context) error { return f.err }
|
func (f *failedDownload) Run(context.Context) error { return f.err }
|
||||||
func (f failedDownload) Stat() download.Stat {
|
func (f *failedDownload) Stat() download.Stat {
|
||||||
return download.Stat{Name: f.name, Status: download.Errored, Total: -1}
|
return download.Stat{Name: f.name, Status: download.Errored, Total: -1}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -281,7 +281,7 @@ func build(opts *cli.Options, targets []target) ([]job, *torrent.Client, error)
|
|||||||
u := t.uris[0] // primary: classifies the target and names the session source
|
u := t.uris[0] // primary: classifies the target and names the session source
|
||||||
src := strings.Join(t.uris, "\t")
|
src := strings.Join(t.uris, "\t")
|
||||||
if first, isDup := dups[u]; isDup {
|
if first, isDup := dups[u]; isDup {
|
||||||
out = append(out, job{source: src, dl: failedDownload{
|
out = append(out, job{source: src, dl: &failedDownload{
|
||||||
name: u,
|
name: u,
|
||||||
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
|
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
|
||||||
}})
|
}})
|
||||||
@@ -397,7 +397,7 @@ func followUps(jobs []job, client *torrent.Client, bo bt.Options) []job {
|
|||||||
// from under the other, stranding it at 0%. build() dedupes pass-1 sources
|
// from under the other, stranding it at 0%. build() dedupes pass-1 sources
|
||||||
// but cannot reach a not-yet-fetched HTTP .torrent, so dedupe here too.
|
// but cannot reach a not-yet-fetched HTTP .torrent, so dedupe here too.
|
||||||
if first, dup := markInfoHash(seen, path); dup {
|
if first, dup := markInfoHash(seen, path); dup {
|
||||||
out = append(out, job{source: path, dl: failedDownload{
|
out = append(out, job{source: path, dl: &failedDownload{
|
||||||
name: path,
|
name: path,
|
||||||
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
|
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
|
||||||
}})
|
}})
|
||||||
|
|||||||
Reference in New Issue
Block a user