Compare commits
3 Commits
db73b51e0d
...
81c44d9ec7
| Author | SHA1 | Date | |
|---|---|---|---|
| 81c44d9ec7 | |||
| 270812de3e | |||
| 65104ade92 |
@@ -18,9 +18,6 @@ Needs Go 1.25+.
|
||||
# HTTP download with 16 connections
|
||||
got -x16 -s16 https://example.com/big.iso
|
||||
|
||||
# let got pick the connection count from the file size
|
||||
got --auto-split https://example.com/big.iso
|
||||
|
||||
# resume an interrupted download
|
||||
got -c https://example.com/big.iso
|
||||
|
||||
@@ -47,7 +44,6 @@ Run `got --help` (or `--help=all`) for every option.
|
||||
| `-c, --continue` | resume a partial download | false |
|
||||
| `-x, --max-connection-per-server` | connections to one server (1–16) | 1 |
|
||||
| `-s, --split` | split a download into N connections | 5 |
|
||||
| `--auto-split` | choose connections from file size (overrides `-x`/`-s`) | false |
|
||||
| `-j, --max-concurrent-downloads` | downloads at once | 5 |
|
||||
| `--max-overall-download-limit` | global speed cap | 0 (off) |
|
||||
| `--checksum` | verify the finished file: `TYPE=DIGEST` (sha-256, sha-1, …) | |
|
||||
|
||||
26
bt/bt.go
26
bt/bt.go
@@ -68,10 +68,17 @@ func ParsePorts(spec string) []int {
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for i := a; i <= b; i++ {
|
||||
if i >= 1 && i <= 65535 {
|
||||
ports = append(ports, i)
|
||||
// Clamp to the valid port space before iterating: a parseable but oversized
|
||||
// spec like "1-9999999999" would otherwise spin for ~10^10 iterations and
|
||||
// hang startup, even though only [1,65535] can ever be appended.
|
||||
if a < 1 {
|
||||
a = 1
|
||||
}
|
||||
if b > 65535 {
|
||||
b = 65535
|
||||
}
|
||||
for i := a; i <= b; i++ {
|
||||
ports = append(ports, i)
|
||||
}
|
||||
}
|
||||
return ports
|
||||
@@ -306,14 +313,17 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := d.choose(t); err != nil {
|
||||
return err
|
||||
}
|
||||
// --check-integrity: re-hash the existing on-disk data BEFORE arming the
|
||||
// request loop, so already-good pieces are not re-requested from peers (aria2
|
||||
// verifies first, then fetches only what is missing).
|
||||
if d.opts.CheckIntegrity {
|
||||
if err := t.VerifyDataContext(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := d.choose(t); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := d.wait(ctx, t); err != nil {
|
||||
return err
|
||||
@@ -382,7 +392,9 @@ func (d *Download) awaitInfo(ctx context.Context, t *torrent.Torrent) error {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-deadline:
|
||||
return fmt.Errorf("timed out fetching metadata after %s", d.opts.MetaTimeout)
|
||||
// Wrap DeadlineExceeded so the exit-code mapping classifies it as a timeout
|
||||
// (2) via errors.Is, rather than matching on the message text.
|
||||
return fmt.Errorf("timed out fetching metadata after %s: %w", d.opts.MetaTimeout, context.DeadlineExceeded)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -97,7 +97,12 @@ func flagLabel(o *Opt) string {
|
||||
case Int:
|
||||
b.WriteString("=N")
|
||||
case Float:
|
||||
// seed-time is a count of minutes; only seed-ratio is an actual ratio.
|
||||
if o.Long == "seed-time" {
|
||||
b.WriteString("=MIN")
|
||||
} else {
|
||||
b.WriteString("=RATIO")
|
||||
}
|
||||
case Enum:
|
||||
b.WriteString("=" + strings.Join(o.Enum, "|"))
|
||||
default:
|
||||
|
||||
@@ -81,7 +81,6 @@ var options = []Opt{
|
||||
{Long: "max-connection-per-server", Short: 'x', Kind: Int, Default: "1", Min: 1, Max: 16, Help: "max connections to one server (1-16)", Tag: Basic},
|
||||
{Long: "split", Short: 's', Kind: Int, Default: "5", Min: 1, Help: "split a download into N connections; actual connections are min(max-connection-per-server, split), and -x defaults to 1", Tag: Basic},
|
||||
{Long: "min-split-size", Short: 'k', Kind: Size, Default: "20M", Min: 1 << 20, Max: 1 << 30, Help: "do not split a piece smaller than SIZE (1M-1024M)", Tag: Basic},
|
||||
{Long: "auto-split", Kind: Bool, Default: "false", Help: "got-only: pick the connection count from file size (one per min-split-size, up to 16); overrides -x/-s", Tag: HTTP},
|
||||
{Long: "force-sequential", Short: 'Z', Kind: Bool, Default: "false", Help: "download each command-line URI as its own file instead of mirroring them", Tag: HTTP},
|
||||
{Long: "max-tries", Short: 'm', Kind: Int, Default: "5", Min: 0, Help: "max retries per segment (0 = unlimited)", Tag: HTTP},
|
||||
{Long: "timeout", Short: 't', Kind: Int, Default: "60", Min: 1, Help: "connection timeout in seconds", Tag: HTTP},
|
||||
|
||||
@@ -2,6 +2,7 @@ package cli
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
@@ -105,5 +106,10 @@ func parseSize(s string) (int64, error) {
|
||||
if n < 0 {
|
||||
return 0, fmt.Errorf("negative size %q", s)
|
||||
}
|
||||
// Reject a value whose unit multiply would overflow int64 and silently wrap to
|
||||
// a bogus (positive or negative) byte count.
|
||||
if mult > 1 && n > math.MaxInt64/mult {
|
||||
return 0, fmt.Errorf("size %q too large", s)
|
||||
}
|
||||
return n * mult, nil
|
||||
}
|
||||
|
||||
@@ -46,6 +46,12 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
|
||||
go func(i int, d Download) {
|
||||
defer wg.Done()
|
||||
|
||||
// Track immediately so a download still queued for a concurrency slot
|
||||
// is visible to the progress reporter as Waiting, rather than hidden
|
||||
// until it starts running.
|
||||
e.track(d)
|
||||
defer e.untrack(d)
|
||||
|
||||
// Acquire a concurrency slot, or bail if we're shutting down
|
||||
// before this download ever started.
|
||||
select {
|
||||
@@ -56,9 +62,6 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
|
||||
}
|
||||
defer func() { <-slots }()
|
||||
|
||||
e.track(d)
|
||||
defer e.untrack(d)
|
||||
|
||||
err := d.Run(ctx)
|
||||
results <- Result{Name: d.Name(), Index: i, Err: err, Stat: d.Stat()}
|
||||
}(i, d)
|
||||
@@ -74,7 +77,8 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
|
||||
return out
|
||||
}
|
||||
|
||||
// Snapshot returns a Stat for every currently running download.
|
||||
// Snapshot returns a Stat for every tracked download — those running plus those
|
||||
// still queued for a concurrency slot (reported as Waiting).
|
||||
func (e *Engine) Snapshot() []Stat {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
@@ -27,10 +28,10 @@ type segState struct {
|
||||
|
||||
func controlPath(out string) string { return out + ".got" }
|
||||
|
||||
// snapshot builds a control record from the live segments. Callers hold the
|
||||
// pool lock (see pool.snapshot) so the slice and each segment's frontier are
|
||||
// stable for the read.
|
||||
func snapshot(url string, total int64, etag, lastmod string, segs []*seg) control {
|
||||
// snapshot builds a control record from the live segments. The segment slice is
|
||||
// fixed once the file is divided and each segment's frontier is owned by one
|
||||
// worker, so reading written atomically gives a consistent record with no lock.
|
||||
func snapshot(url string, total int64, etag, lastmod string, segs []seg) control {
|
||||
c := control{URL: url, Total: total, ETag: etag, LastModified: lastmod, Segs: make([]segState, len(segs))}
|
||||
for i := range segs {
|
||||
c.Segs[i] = segState{segs[i].start, segs[i].endOff(), atomic.LoadInt64(&segs[i].written)}
|
||||
@@ -75,7 +76,35 @@ func loadControl(out, url string, total int64, etag, lastmod string) *control {
|
||||
if (lastmod != "" || c.LastModified != "") && c.LastModified != lastmod {
|
||||
return nil
|
||||
}
|
||||
// Reject a control file whose segments do not exactly tile [0,total): a
|
||||
// truncated/corrupted/hand-edited sidecar that still parses as JSON could
|
||||
// otherwise mark a segment done() without its bytes on disk (inflated Written)
|
||||
// or leave an un-downloaded hole, both of which would be reported as a complete
|
||||
// file. We restart cleanly instead of trusting it.
|
||||
if !validSegs(c.Segs, c.Total) {
|
||||
return nil
|
||||
}
|
||||
return &c
|
||||
}
|
||||
|
||||
// validSegs reports whether segs cover [0,total) with no gap or overlap and a
|
||||
// sane written count for each. The recorded order is not sorted (a steal appends
|
||||
// a tail), so we check coverage on a sorted copy.
|
||||
func validSegs(segs []segState, total int64) bool {
|
||||
if total <= 0 || len(segs) == 0 {
|
||||
return false
|
||||
}
|
||||
sorted := append([]segState(nil), segs...)
|
||||
sort.Slice(sorted, func(i, j int) bool { return sorted[i].Start < sorted[j].Start })
|
||||
var next int64
|
||||
for _, s := range sorted {
|
||||
length := s.End - s.Start + 1
|
||||
if s.Start != next || s.End < s.Start || s.Written < 0 || s.Written > length {
|
||||
return false
|
||||
}
|
||||
next = s.End + 1
|
||||
}
|
||||
return next == total
|
||||
}
|
||||
|
||||
func removeControl(out string) { os.Remove(controlPath(out)) }
|
||||
|
||||
278
httpdl/httpdl.go
278
httpdl/httpdl.go
@@ -1,6 +1,6 @@
|
||||
// Package httpdl downloads a single HTTP(S) resource over one or more
|
||||
// connections. The model is deliberately flat: split the file into byte-range
|
||||
// segments, hand them to a pool of worker goroutines, and have each worker
|
||||
// segments, give each segment its own worker goroutine, and have each worker
|
||||
// stream its range straight to disk with WriteAt (safe for concurrent,
|
||||
// non-overlapping writes — no shared seek, no mutex). Goroutines blocking on
|
||||
// real I/O keep the model simple: no segment manager, no piece storage, no
|
||||
@@ -57,7 +57,6 @@ type Config struct {
|
||||
Split int // --split: total connections across all mirrors
|
||||
MaxConnPerServer int // --max-connection-per-server: per-host connection cap
|
||||
MinSplit int64
|
||||
AutoSplit bool // --auto-split (got-only): pick the connection count from file size
|
||||
Tries int // 0 = unlimited
|
||||
Timeout time.Duration
|
||||
FileAlloc string // none | prealloc | trunc | falloc
|
||||
@@ -120,19 +119,20 @@ type Config struct {
|
||||
// connections.
|
||||
type Download struct {
|
||||
uris []string // mirror list; uris[0] is the primary (naming + resume key)
|
||||
maxConns int // manual segment-worker cap = min(split, len(uris)*max-connection-per-server); --auto-split derives its own count per file in segmented()
|
||||
maxConns int // segment-worker cap = min(split, len(uris)*max-connection-per-server)
|
||||
cfg Config
|
||||
client *http.Client
|
||||
limit []*rate.Limiter
|
||||
jar *cookiejar.Jar // non-nil when cookies are in use, for save-cookies
|
||||
|
||||
// name and out are resolved in Run after the probe response is known (the
|
||||
// final post-redirect URL and any Content-Disposition can change them), then
|
||||
// never mutated again, so Stat can read name from another goroutine without a
|
||||
// lock once Run has set it. They are seeded in New so a Stat before Run still
|
||||
// has a sensible best-effort name.
|
||||
name string
|
||||
out string
|
||||
// final post-redirect URL and any Content-Disposition can change them). The
|
||||
// engine publishes a Download to the progress reporter before Run starts, so
|
||||
// Stat reads these from the reporter goroutine while resolveName writes them;
|
||||
// they are atomic.Pointer[string] (exactly like bt.Download.name) to keep that
|
||||
// read/write race-free. Seeded in New so a Stat before Run has a best-effort name.
|
||||
name atomic.Pointer[string]
|
||||
out atomic.Pointer[string]
|
||||
initErr error
|
||||
|
||||
// mu guards seenURLs, which records every URL we sent a request to so
|
||||
@@ -215,24 +215,31 @@ func New(uris []string, cfg Config) *Download {
|
||||
if perHost < 1 {
|
||||
perHost = 1
|
||||
}
|
||||
// The transport must allow as many concurrent connections per host as we may
|
||||
// actually open. --auto-split scales up to maxAutoConns per host once the file
|
||||
// size is known (in segmented), so raise the cap to that ceiling when it is on.
|
||||
connCap := perHost
|
||||
if cfg.AutoSplit {
|
||||
connCap = maxAutoConns
|
||||
}
|
||||
tr := &http.Transport{
|
||||
Proxy: http.ProxyFromEnvironment,
|
||||
MaxConnsPerHost: connCap,
|
||||
MaxIdleConnsPerHost: connCap,
|
||||
MaxConnsPerHost: perHost,
|
||||
MaxIdleConnsPerHost: perHost,
|
||||
ResponseHeaderTimeout: cfg.Timeout,
|
||||
TLSHandshakeTimeout: connectTimeout,
|
||||
DialContext: dialContext,
|
||||
TLSClientConfig: tlsCfg,
|
||||
}
|
||||
if cfg.Proxy != "" {
|
||||
if pu, err := url.Parse(cfg.Proxy); err == nil {
|
||||
pu, err := url.Parse(cfg.Proxy)
|
||||
switch {
|
||||
case err != nil:
|
||||
// A malformed proxy must fail loudly, not silently fall back to the
|
||||
// environment proxy (or a direct connection) and leak the real IP.
|
||||
if initErr == nil {
|
||||
initErr = fmt.Errorf("proxy %q: %w", cfg.Proxy, err)
|
||||
}
|
||||
case pu.Host == "":
|
||||
// "host:port" with no scheme parses to an opaque URL with no Host, which
|
||||
// http.ProxyURL would route nowhere; require an explicit scheme.
|
||||
if initErr == nil {
|
||||
initErr = fmt.Errorf("proxy %q: missing scheme (use scheme://host:port)", cfg.Proxy)
|
||||
}
|
||||
default:
|
||||
tr.Proxy = http.ProxyURL(pu)
|
||||
}
|
||||
}
|
||||
@@ -294,29 +301,47 @@ func New(uris []string, cfg Config) *Download {
|
||||
client.Jar = jar
|
||||
}
|
||||
|
||||
return &Download{
|
||||
d := &Download{
|
||||
uris: uris,
|
||||
maxConns: maxConns,
|
||||
cfg: cfg,
|
||||
client: client,
|
||||
limit: limit,
|
||||
jar: jar,
|
||||
name: name,
|
||||
out: out,
|
||||
total: -1,
|
||||
initErr: initErr,
|
||||
}
|
||||
d.setName(name)
|
||||
d.setOut(out)
|
||||
return d
|
||||
}
|
||||
|
||||
func (d *Download) Name() string { return d.name }
|
||||
func (d *Download) Name() string { return d.loadName() }
|
||||
|
||||
// Path returns the resolved output file path (used by --follow-torrent).
|
||||
func (d *Download) Path() string { return d.out }
|
||||
func (d *Download) Path() string { return d.loadOut() }
|
||||
|
||||
// name and out are read by Stat from the reporter goroutine while Run resolves
|
||||
// them, so they are stored atomically (mirroring bt.Download.name).
|
||||
func (d *Download) setName(s string) { d.name.Store(&s) }
|
||||
func (d *Download) setOut(s string) { d.out.Store(&s) }
|
||||
func (d *Download) loadName() string {
|
||||
if p := d.name.Load(); p != nil {
|
||||
return *p
|
||||
}
|
||||
return ""
|
||||
}
|
||||
func (d *Download) loadOut() string {
|
||||
if p := d.out.Load(); p != nil {
|
||||
return *p
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (d *Download) Stat() download.Stat {
|
||||
return download.Stat{
|
||||
Name: d.name,
|
||||
ID: d.out, // resolved output path; stable per process (CONTRACT)
|
||||
Name: d.loadName(),
|
||||
ID: d.loadOut(), // resolved output path; stable per process (CONTRACT)
|
||||
IsBT: false,
|
||||
Status: download.Status(atomic.LoadInt32(&d.status)),
|
||||
Total: atomic.LoadInt64(&d.total),
|
||||
@@ -400,15 +425,19 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
||||
// the limit past a startup grace. The watcher cancels xferCtx and sets a
|
||||
// flag so we can return the distinct "too slow" error rather than a bare
|
||||
// cancellation.
|
||||
out := d.loadOut()
|
||||
xferCtx, stopGuard, tooSlow := d.speedGuard(ctx)
|
||||
if !pr.ranged || pr.total <= 0 {
|
||||
err = d.single(xferCtx, d.out)
|
||||
err = d.single(xferCtx, out)
|
||||
} else {
|
||||
err = d.segmented(xferCtx, d.out, pr.total, pr.etag, pr.lastmod)
|
||||
err = d.segmented(xferCtx, out, pr.total, pr.etag, pr.lastmod)
|
||||
}
|
||||
stopGuard()
|
||||
if err != nil {
|
||||
if tooSlow() {
|
||||
// Only translate to ErrTooSlow when the transfer actually ended in the
|
||||
// guard's cancellation; a write error or mirror exhaustion that merely
|
||||
// raced the guard firing keeps its own (real) cause and exit code.
|
||||
if tooSlow() && errors.Is(err, context.Canceled) {
|
||||
return fmt.Errorf("%s: %w (<= %d bytes/sec)", d.primary(), ErrTooSlow, d.cfg.LowestSpeedLimit)
|
||||
}
|
||||
return err
|
||||
@@ -416,14 +445,14 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
||||
// --checksum: verify the finished file before declaring success, so a
|
||||
// corrupted or tampered download fails (--checksum, exit code 32).
|
||||
if newSum != nil {
|
||||
if err = verifyFile(d.out, newSum, sumWant); err != nil {
|
||||
if err = verifyFile(out, newSum, sumWant); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// -R / --remote-time: stamp the finished file with the server's mtime.
|
||||
if d.cfg.RemoteTime && pr.lastmod != "" {
|
||||
if t, err := http.ParseTime(pr.lastmod); err == nil {
|
||||
os.Chtimes(d.out, t, t)
|
||||
os.Chtimes(out, t, t)
|
||||
}
|
||||
}
|
||||
d.setStatus(download.Complete)
|
||||
@@ -435,6 +464,7 @@ func (d *Download) Run(ctx context.Context) (err error) {
|
||||
// the original URL. Overwrite / auto-rename is then resolved against the final
|
||||
// path, honouring -c so a resumable file is never renamed (items [2],[8],[13]).
|
||||
func (d *Download) resolveName(pr probeResult) error {
|
||||
out := d.loadOut()
|
||||
if d.cfg.Out == "" {
|
||||
name := ""
|
||||
if cd := pr.cdisp; cd != "" {
|
||||
@@ -446,8 +476,9 @@ func (d *Download) resolveName(pr probeResult) error {
|
||||
if name == "" {
|
||||
name = nameFromURL(d.primary())
|
||||
}
|
||||
d.name = name
|
||||
d.out = filepath.Join(d.cfg.Dir, name)
|
||||
out = filepath.Join(d.cfg.Dir, name)
|
||||
d.setName(name)
|
||||
d.setOut(out)
|
||||
}
|
||||
|
||||
// --dry-run writes nothing, so skip overwrite/auto-rename resolution; the name
|
||||
@@ -458,20 +489,20 @@ func (d *Download) resolveName(pr probeResult) error {
|
||||
|
||||
// -c resumes either our own .got sidecar or a foreign/browser partial file,
|
||||
// so an existing file is never auto-renamed under -c (item [2]).
|
||||
if d.cfg.Continue && fileExists(d.out) {
|
||||
if d.cfg.Continue && fileExists(out) {
|
||||
return nil
|
||||
}
|
||||
if fileExists(d.out) && !d.cfg.AllowOverwrite {
|
||||
if fileExists(out) && !d.cfg.AllowOverwrite {
|
||||
if d.cfg.AutoRename {
|
||||
u, err := uniqueName(d.out)
|
||||
u, err := uniqueName(out)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.out = u
|
||||
d.name = filepath.Base(u)
|
||||
d.setOut(u)
|
||||
d.setName(filepath.Base(u))
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("%s: %w (use --allow-overwrite or -c)", d.out, ErrOutputExists)
|
||||
return fmt.Errorf("%s: %w (use --allow-overwrite or -c)", out, ErrOutputExists)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -587,8 +618,8 @@ func (d *Download) probe(ctx context.Context, uri string) (probeResult, error) {
|
||||
// is already current. Only done when there is no .got control file (an
|
||||
// in-progress resume must not be short-circuited), using the local file's
|
||||
// mtime for If-Modified-Since.
|
||||
if d.cfg.ConditionalGet && !fileExists(controlPath(d.out)) {
|
||||
if fi, err := os.Stat(d.out); err == nil {
|
||||
if out := d.loadOut(); d.cfg.ConditionalGet && !fileExists(controlPath(out)) {
|
||||
if fi, err := os.Stat(out); err == nil {
|
||||
req.Header.Set("If-Modified-Since", fi.ModTime().UTC().Format(http.TimeFormat))
|
||||
}
|
||||
}
|
||||
@@ -632,13 +663,7 @@ func (d *Download) probe(ctx context.Context, uri string) (probeResult, error) {
|
||||
// segmented downloads total bytes over d.maxConns workers, with per-segment
|
||||
// resume from the control file.
|
||||
func (d *Download) segmented(ctx context.Context, out string, total int64, etag, lastmod string) error {
|
||||
// --auto-split picks the connection count from the file size now that total is
|
||||
// known; otherwise use the manual cap fixed at construction.
|
||||
conns := d.maxConns
|
||||
if d.cfg.AutoSplit {
|
||||
conns = autoConns(total, d.cfg.MinSplit)
|
||||
}
|
||||
segs := makeSegments(total, d.cfg.MinSplit, conns)
|
||||
segs := makeSegments(total, d.cfg.MinSplit, d.maxConns)
|
||||
if c := d.resumeControl(out, etag, lastmod); c != nil {
|
||||
segs = segsFromControl(c)
|
||||
} else if d.cfg.Continue && !fileExists(controlPath(out)) {
|
||||
@@ -676,36 +701,27 @@ func (d *Download) segmented(ctx context.Context, out string, total int64, etag,
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
p := newPool(segs, d.cfg.MinSplit)
|
||||
|
||||
// Periodically persist progress so a crash can resume.
|
||||
saveDone := make(chan struct{})
|
||||
go d.saveLoop(ctx, out, total, etag, lastmod, p, saveDone)
|
||||
go d.saveLoop(ctx, out, total, etag, lastmod, segs, saveDone)
|
||||
|
||||
// One worker per connection. A worker fetches segments until acquire runs
|
||||
// dry, which happens only once every remaining byte is owned; a worker that
|
||||
// finishes early steals the tail of a slower segment rather than sitting idle
|
||||
// while the last segment drains over a single connection.
|
||||
// One worker per segment. Segments are non-overlapping byte ranges written
|
||||
// with WriteAt, so the workers share no cursor and need no coordination — a
|
||||
// finished worker simply exits. The division (makeSegments / a resumed
|
||||
// control) fixes the parallelism up front; there is no rebalancing.
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
errOnce sync.Once
|
||||
runErr error
|
||||
)
|
||||
for w := 0; w < conns; w++ {
|
||||
for i := range segs {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
go func(s *seg) {
|
||||
defer wg.Done()
|
||||
for {
|
||||
s := p.acquire()
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
if err := d.fetchSeg(ctx, f, p, s, total); err != nil {
|
||||
if err := d.fetchSeg(ctx, f, s, total); err != nil {
|
||||
errOnce.Do(func() { runErr = err; cancel() })
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}(&segs[i])
|
||||
}
|
||||
wg.Wait()
|
||||
cancel()
|
||||
@@ -714,7 +730,7 @@ func (d *Download) segmented(ctx context.Context, out string, total int64, etag,
|
||||
if runErr != nil {
|
||||
return runErr // keep the control file for a later -c
|
||||
}
|
||||
p.snapshot(d.primary(), total, etag, lastmod).save(out)
|
||||
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
|
||||
removeControl(out)
|
||||
return nil
|
||||
}
|
||||
@@ -722,7 +738,7 @@ func (d *Download) segmented(ctx context.Context, out string, total int64, etag,
|
||||
// fetchSeg downloads one segment, retrying from its resume point on error.
|
||||
// Each attempt re-reads s.offset(), so a retry continues from the bytes already
|
||||
// written rather than restarting the range.
|
||||
func (d *Download) fetchSeg(ctx context.Context, f *os.File, p *pool, s *seg, total int64) error {
|
||||
func (d *Download) fetchSeg(ctx context.Context, f *os.File, s *seg, total int64) error {
|
||||
// Try each mirror in turn for this segment: a transient error retries the
|
||||
// same mirror under withRetries; an exhausted budget or a per-mirror permanent
|
||||
// error (a 404, a non-206, a length mismatch) falls over to the next mirror.
|
||||
@@ -734,12 +750,12 @@ func (d *Download) fetchSeg(ctx context.Context, f *os.File, p *pool, s *seg, to
|
||||
if s.done() {
|
||||
return nil
|
||||
}
|
||||
return d.fetchOnce(ctx, f, p, s, uri, total)
|
||||
return d.fetchOnce(ctx, f, s, uri, total)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
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, s *seg, uri string, total int64) error {
|
||||
reqCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
req, err := d.request(reqCtx, uri, fmt.Sprintf("bytes=%d-%d", s.offset(), s.endOff()))
|
||||
@@ -757,60 +773,95 @@ func (d *Download) fetchOnce(ctx context.Context, f *os.File, p *pool, s *seg, u
|
||||
atomic.AddInt32(&d.conns, -1)
|
||||
}()
|
||||
if resp.StatusCode != http.StatusPartialContent {
|
||||
return statusError(fmt.Sprintf("segment %d: expected 206, got %s", s.index, resp.Status), resp.StatusCode)
|
||||
err := statusError(fmt.Sprintf("segment %d: expected 206, got %s", s.index, resp.Status), resp.StatusCode)
|
||||
// A 200 means this mirror ignores Range entirely; retrying it only burns the
|
||||
// budget, so make it fatal and let fetchSeg fail over to the next mirror.
|
||||
if resp.StatusCode == http.StatusOK {
|
||||
err = fatal{err}
|
||||
}
|
||||
return err
|
||||
}
|
||||
// A mirror must serve the same file as the primary, or its bytes written at
|
||||
// this offset would corrupt the output. Reject a 206 whose total length
|
||||
// disagrees with the probe (validate the total length); fatal so withRetries
|
||||
// stops and fetchSeg falls over to the next mirror instead of writing it.
|
||||
if _, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); ok && t != total {
|
||||
return fatal{fmt.Errorf("segment %d: %s reports length %d, want %d", s.index, uri, t, total)}
|
||||
// this offset would corrupt the output. Require a Content-Range that confirms
|
||||
// both the offset we asked for and the total length the probe saw; a missing,
|
||||
// unparseable, or divergent range is fatal so withRetries stops and fetchSeg
|
||||
// falls over to the next mirror instead of writing unverified bytes.
|
||||
if start, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); !ok || t != total || start != s.offset() {
|
||||
return fatal{fmt.Errorf("segment %d: %s returned an unverifiable range %q (want start %d, length %d)",
|
||||
s.index, uri, resp.Header.Get("Content-Range"), s.offset(), total)}
|
||||
}
|
||||
body, stop := d.idleGuard(resp.Body, cancel)
|
||||
defer stop()
|
||||
return d.pump(ctx, f, p, s, body)
|
||||
return d.pump(ctx, f, s, body)
|
||||
}
|
||||
|
||||
// errIdleTimeout marks a read that stalled past the idle window, so callers can
|
||||
// report a distinct "timed out" instead of the bare "context canceled" that the
|
||||
// cancelled request would otherwise surface (item [39]).
|
||||
var errIdleTimeout = errors.New("idle timeout: no data received")
|
||||
// ErrTimeout marks a transfer that stalled past the idle window (--timeout): a
|
||||
// distinct "timed out" instead of the bare "context canceled" the cancelled
|
||||
// request would otherwise surface (item [39]). main maps it to exit code 2 via
|
||||
// errors.Is, so it must stay on the error chain (CONTRACT).
|
||||
var ErrTimeout = errors.New("idle timeout: no data received")
|
||||
|
||||
// idleReader resets a watchdog timer on every read; if a read stalls longer
|
||||
// than the idle window the timer fires and cancels the request, so the blocked
|
||||
// Read returns an error instead of hanging forever on a wedged connection.
|
||||
// idleReader records the time of the last byte received; a background watcher
|
||||
// (idleGuard) cancels the request once a full idle window passes with no
|
||||
// progress. Tracking progress by timestamp — rather than arming a per-read timer
|
||||
// — means a slow-but-advancing transfer is never aborted, and there is no
|
||||
// Reset/Stop race that could poison a connection that was still delivering data.
|
||||
type idleReader struct {
|
||||
r io.Reader
|
||||
timer *time.Timer
|
||||
idle time.Duration
|
||||
fired atomic.Bool // set when the watchdog cancelled the request
|
||||
last atomic.Int64 // UnixNano of the last byte received
|
||||
timedOut atomic.Bool // set when the watcher cancelled the request
|
||||
}
|
||||
|
||||
func (ir *idleReader) Read(p []byte) (int, error) {
|
||||
ir.timer.Reset(ir.idle)
|
||||
n, err := ir.r.Read(p)
|
||||
ir.timer.Stop() // only the Read itself counts as idle, not write/throttle gaps
|
||||
if err != nil && ir.fired.Load() {
|
||||
// The cancellation came from our watchdog, not the caller's ctx, so
|
||||
if n > 0 {
|
||||
ir.last.Store(time.Now().UnixNano())
|
||||
}
|
||||
if err != nil && ir.timedOut.Load() {
|
||||
// The cancellation came from our watcher, not the caller's ctx, so
|
||||
// translate the read error into the distinct idle-timeout sentinel.
|
||||
return n, fmt.Errorf("%w (after %s)", errIdleTimeout, ir.idle)
|
||||
return n, fmt.Errorf("%w (after %s)", ErrTimeout, ir.idle)
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
// idleGuard wraps body with an idle timeout of d.cfg.Timeout, cancelling the
|
||||
// request if no data arrives in that window. It returns the reader to use and a
|
||||
// stop func to call when the transfer is done.
|
||||
// watch cancels the request once idle passes with no byte received. It polls a
|
||||
// few times per window so a stall is caught within ~idle of the last byte,
|
||||
// without a shared timer that Read would have to Reset/Stop.
|
||||
func (ir *idleReader) watch(stop <-chan struct{}, cancel context.CancelFunc) {
|
||||
tick := ir.idle / 4
|
||||
if tick <= 0 {
|
||||
tick = ir.idle
|
||||
}
|
||||
t := time.NewTicker(tick)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
case now := <-t.C:
|
||||
if now.UnixNano()-ir.last.Load() >= int64(ir.idle) {
|
||||
ir.timedOut.Store(true)
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// idleGuard wraps body with an idle timeout of d.cfg.Timeout: a watcher goroutine
|
||||
// cancels the request when no byte arrives for that long. It returns the reader to
|
||||
// use and a stop func to call (exactly once) when the transfer is done.
|
||||
func (d *Download) idleGuard(body io.Reader, cancel context.CancelFunc) (io.Reader, func()) {
|
||||
if d.cfg.Timeout <= 0 {
|
||||
return body, func() {}
|
||||
}
|
||||
ir := &idleReader{r: body, idle: d.cfg.Timeout}
|
||||
ir.timer = time.AfterFunc(d.cfg.Timeout, func() {
|
||||
ir.fired.Store(true)
|
||||
cancel()
|
||||
})
|
||||
return ir, func() { ir.timer.Stop() }
|
||||
ir.last.Store(time.Now().UnixNano())
|
||||
stop := make(chan struct{})
|
||||
go ir.watch(stop, cancel)
|
||||
return ir, func() { close(stop) }
|
||||
}
|
||||
|
||||
// speedStartupGrace is how long a download may stay slow before the
|
||||
@@ -913,10 +964,10 @@ func statusError(ctxMsg string, code int) error {
|
||||
}
|
||||
|
||||
// pump copies the response body into the file at the segment's running offset,
|
||||
// stopping at the segment end and respecting rate limits. The loop reads
|
||||
// remaining() afresh each turn, so a steal that shrinks s mid-transfer simply
|
||||
// ends the loop early at the new end, leaving the stolen tail to its new worker.
|
||||
func (d *Download) pump(ctx context.Context, f *os.File, p *pool, s *seg, body io.Reader) error {
|
||||
// stopping at the segment end and respecting rate limits. Only this worker writes
|
||||
// s.written, so the advance is a plain atomic add (Stat and the snapshot read it
|
||||
// atomically); the loop ends when the range is filled.
|
||||
func (d *Download) pump(ctx context.Context, f *os.File, s *seg, body io.Reader) error {
|
||||
buf := make([]byte, readBuf)
|
||||
for s.remaining() > 0 {
|
||||
n := int64(len(buf))
|
||||
@@ -928,7 +979,7 @@ func (d *Download) pump(ctx context.Context, f *os.File, p *pool, s *seg, body i
|
||||
if _, werr := f.WriteAt(buf[:rd], s.offset()); werr != nil {
|
||||
return werr
|
||||
}
|
||||
p.advance(s, int64(rd))
|
||||
s.addWritten(int64(rd))
|
||||
atomic.AddInt64(&d.completed, int64(rd))
|
||||
d.throttle(ctx, rd)
|
||||
}
|
||||
@@ -1009,6 +1060,19 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
|
||||
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
|
||||
return statusError(fmt.Sprintf("%s: %s", uri, resp.Status), resp.StatusCode)
|
||||
}
|
||||
// On failover, resumeAt is the prefix a previous mirror wrote. If this mirror
|
||||
// honours the range, make sure it serves the same file before we append to that
|
||||
// prefix — a divergent total or start means different content, and writing it
|
||||
// past the existing bytes would corrupt the output. Mirror the segmented guard
|
||||
// (fatal, so we fall over rather than append unverified bytes). A same-mirror
|
||||
// retry compares against its own earlier total, so it never false-trips.
|
||||
if resp.StatusCode == http.StatusPartialContent && resumeAt > 0 {
|
||||
if prev := atomic.LoadInt64(&d.total); prev > 0 {
|
||||
if start, _, t, ok := parseContentRange(resp.Header.Get("Content-Range")); ok && (t != prev || start != resumeAt) {
|
||||
return fatal{fmt.Errorf("%s: returned a divergent range %q (want start %d, length %d)", uri, resp.Header.Get("Content-Range"), resumeAt, prev)}
|
||||
}
|
||||
}
|
||||
}
|
||||
// If we asked to resume but the server replied 200 (no range honoured), start
|
||||
// over from the top rather than appending past the existing bytes.
|
||||
if resp.StatusCode == http.StatusOK {
|
||||
@@ -1070,7 +1134,7 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag, lastmod string, p *pool, done chan<- struct{}) {
|
||||
func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag, lastmod string, segs []seg, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
interval := d.cfg.AutoSaveInterval
|
||||
if interval <= 0 {
|
||||
@@ -1081,10 +1145,10 @@ func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag,
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
p.snapshot(d.primary(), total, etag, lastmod).save(out)
|
||||
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
|
||||
return
|
||||
case <-t.C:
|
||||
p.snapshot(d.primary(), total, etag, lastmod).save(out)
|
||||
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,27 +1,22 @@
|
||||
package httpdl
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
import "sync/atomic"
|
||||
|
||||
// seg is one contiguous byte range of the output file, downloaded by a single
|
||||
// ranged GET. A segment IS a byte range — a plain HTTP downloader does not need
|
||||
// the Piece/Segment/block layering that exists only to share code with
|
||||
// BitTorrent. written and end are updated with atomic ops so Stat() and the
|
||||
// resume snapshot can read them while a worker advances written or a steal
|
||||
// shrinks end (see pool).
|
||||
// BitTorrent. start and end are fixed when the file is divided; written is
|
||||
// advanced (atomically) by the segment's one worker, so Stat() and the resume
|
||||
// snapshot can read its frontier while it downloads.
|
||||
type seg struct {
|
||||
index int
|
||||
start int64 // first byte offset, inclusive
|
||||
end int64 // last byte offset, inclusive (a steal may shrink this)
|
||||
end int64 // last byte offset, inclusive
|
||||
written int64 // bytes already written into this segment (the resume point)
|
||||
owned bool // a worker is fetching this segment; guarded by pool.mu
|
||||
}
|
||||
|
||||
func (s *seg) endOff() int64 { return atomic.LoadInt64(&s.end) }
|
||||
func (s *seg) setEnd(v int64) { atomic.StoreInt64(&s.end, v) }
|
||||
func (s *seg) length() int64 { return s.endOff() - s.start + 1 }
|
||||
func (s *seg) length() int64 { return s.end - s.start + 1 }
|
||||
func (s *seg) endOff() int64 { return s.end }
|
||||
func (s *seg) done() bool { return atomic.LoadInt64(&s.written) >= s.length() }
|
||||
func (s *seg) addWritten(n int64) { atomic.AddInt64(&s.written, n) }
|
||||
func (s *seg) progress() int64 { return atomic.LoadInt64(&s.written) }
|
||||
@@ -30,7 +25,9 @@ func (s *seg) remaining() int64 { return s.length() - atomic.LoadInt64(&s.writ
|
||||
|
||||
// makeSegments divides a file of total bytes into contiguous segments, using at
|
||||
// most conns of them and never splitting below minSplit. The remainder lands in
|
||||
// the last segment. With conns==1 (the default) this yields one segment.
|
||||
// the last segment. With conns==1 (the default) this yields one segment. Each
|
||||
// segment gets its own worker, so the division here fixes the parallelism: there
|
||||
// is no later rebalancing.
|
||||
func makeSegments(total, minSplit int64, conns int) []seg {
|
||||
if conns < 1 {
|
||||
conns = 1
|
||||
@@ -58,127 +55,3 @@ func makeSegments(total, minSplit int64, conns int) []seg {
|
||||
}
|
||||
return segs
|
||||
}
|
||||
|
||||
// maxAutoConns is the ceiling --auto-split scales to. aria2 caps
|
||||
// max-connection-per-server (-x) at 16, so a got-chosen count honours the same
|
||||
// per-server limit rather than inventing a looser one.
|
||||
const maxAutoConns = 16
|
||||
|
||||
// autoConns picks a connection count from the file size, used only when
|
||||
// --auto-split is set: one connection per min-split-size of content, at least 1
|
||||
// and at most maxAutoConns. It is the got-only stand-in for hand-tuning -x/-s,
|
||||
// and because the count never exceeds total/minSplit, makeSegments yields exactly
|
||||
// that many balanced segments.
|
||||
func autoConns(total, minSplit int64) int {
|
||||
if minSplit < 1 {
|
||||
minSplit = 1
|
||||
}
|
||||
n := total / minSplit
|
||||
if n < 1 {
|
||||
n = 1
|
||||
}
|
||||
if n > maxAutoConns {
|
||||
n = maxAutoConns
|
||||
}
|
||||
return int(n)
|
||||
}
|
||||
|
||||
// pool is the live set of segments for one segmented download. A worker takes a
|
||||
// segment with acquire and fetches it to completion; once no fresh segment is
|
||||
// left, an idle worker steals the back half of whichever in-flight segment has
|
||||
// the most still to download, so the last slow segment is shared across the idle
|
||||
// connections instead of draining alone. aria2 does the same on-demand split via
|
||||
// its SegmentMan; without it a fixed pre-division leaves connections idle while
|
||||
// one slow mirror finishes.
|
||||
//
|
||||
// The mutex serialises a steal against the byte accounting it splits: advance (a
|
||||
// worker committing a write) takes it too, so a steal reading a victim's frontier
|
||||
// can never race the owner advancing past the chosen split point. A steal fires
|
||||
// only when the victim still has at least 2*minSplit to go and cuts at the
|
||||
// midpoint, so both halves stay >= minSplit (--min-split-size) and the half the
|
||||
// owner keeps is far larger than one read buffer — the owner's in-flight write
|
||||
// therefore can never reach into the stolen tail.
|
||||
type pool struct {
|
||||
mu sync.Mutex
|
||||
segs []*seg
|
||||
minSplit int64
|
||||
}
|
||||
|
||||
// newPool wraps the pre-divided segments in a pool. Each is copied into its own
|
||||
// allocation so a later steal can append a tail without invalidating the pointers
|
||||
// workers already hold.
|
||||
//
|
||||
// minSplit is floored at readBuf: a steal leaves the owner the front half of the
|
||||
// split, which is at least minSplit, while the owner's in-flight write is at most
|
||||
// readBuf, so minSplit >= readBuf is exactly what keeps that write out of the
|
||||
// stolen tail. The CLI already holds --min-split-size well above readBuf (>= 1
|
||||
// MiB), so the floor only guards direct callers and tests — but it keeps the
|
||||
// no-overlap invariant inside this file rather than resting on a distant flag.
|
||||
func newPool(segs []seg, minSplit int64) *pool {
|
||||
if minSplit < readBuf {
|
||||
minSplit = readBuf
|
||||
}
|
||||
p := &pool{minSplit: minSplit, segs: make([]*seg, len(segs))}
|
||||
for i := range segs {
|
||||
s := segs[i]
|
||||
p.segs[i] = &s
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// acquire returns the next segment for a worker to fetch: a fresh one if any
|
||||
// remain, otherwise the tail split off the busiest in-flight segment. It returns
|
||||
// nil when every remaining byte is already owned, i.e. the download is finishing
|
||||
// and there is nothing left to steal.
|
||||
func (p *pool) acquire() *seg {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for _, s := range p.segs {
|
||||
if !s.owned && !s.done() {
|
||||
s.owned = true
|
||||
return s
|
||||
}
|
||||
}
|
||||
return p.steal()
|
||||
}
|
||||
|
||||
// steal splits the back half off the in-flight segment with the most remaining
|
||||
// and returns it, or nil if none has enough left to be worth splitting. The
|
||||
// caller holds p.mu, which keeps every owner's advance out so each victim's
|
||||
// frontier is stable while we choose and commit the split.
|
||||
func (p *pool) steal() *seg {
|
||||
var victim *seg
|
||||
for _, s := range p.segs {
|
||||
if s.owned && !s.done() && s.remaining() >= 2*p.minSplit {
|
||||
if victim == nil || s.remaining() > victim.remaining() {
|
||||
victim = s
|
||||
}
|
||||
}
|
||||
}
|
||||
if victim == nil {
|
||||
return nil
|
||||
}
|
||||
mid := victim.offset() + victim.remaining()/2
|
||||
// index only feeds the mirror round-robin (d.mirror) and the "segment N" log
|
||||
// label; a tail's index need not be contiguous or unique, so len is fine.
|
||||
tail := &seg{index: len(p.segs), start: mid, end: victim.endOff(), owned: true}
|
||||
victim.setEnd(mid - 1)
|
||||
p.segs = append(p.segs, tail)
|
||||
return tail
|
||||
}
|
||||
|
||||
// advance commits n freshly written bytes to s under the pool lock, so a
|
||||
// concurrent steal sees a stable frontier for the segment it may split.
|
||||
func (p *pool) advance(s *seg, n int64) {
|
||||
p.mu.Lock()
|
||||
s.addWritten(n)
|
||||
p.mu.Unlock()
|
||||
}
|
||||
|
||||
// snapshot builds the resume record from the live set, including any stolen
|
||||
// tails, under the lock so it never races a steal appending a segment.
|
||||
func (p *pool) snapshot(url string, total int64, etag, lastmod string) control {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return snapshot(url, total, etag, lastmod, p.segs)
|
||||
}
|
||||
|
||||
@@ -47,31 +47,6 @@ func TestMakeSegments(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestAutoConns(t *testing.T) {
|
||||
const m = int64(20 << 20) // 20 MiB min-split, aria2's default
|
||||
tests := []struct {
|
||||
name string
|
||||
total int64
|
||||
minSplit int64
|
||||
want int
|
||||
}{
|
||||
{"under one min-split is a single connection", 5 << 20, m, 1},
|
||||
{"exactly one min-split", m, m, 1},
|
||||
{"five min-splits", 100 << 20, m, 5},
|
||||
{"caps at maxAutoConns", 10 << 30, m, maxAutoConns},
|
||||
{"rounds down to whole pieces", m*3 + 1, m, 3},
|
||||
{"tiny min-split still capped", 1 << 30, 1 << 20, maxAutoConns},
|
||||
{"zero total is one connection", 0, m, 1},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := autoConns(tc.total, tc.minSplit); got != tc.want {
|
||||
t.Errorf("autoConns(%d, %d) = %d, want %d", tc.total, tc.minSplit, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSegProgress(t *testing.T) {
|
||||
s := seg{start: 100, end: 199} // length 100
|
||||
if s.length() != 100 {
|
||||
@@ -215,88 +190,46 @@ func TestParseContentRange(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// assertTiles checks that the pool's segments still cover [0,total) exactly:
|
||||
// sorted by start they must be contiguous with no gap and no overlap.
|
||||
func assertTiles(t *testing.T, p *pool, total int64) {
|
||||
// tilesCover checks that segs cover [0,total) exactly: sorted by start they must
|
||||
// be contiguous with no gap and no overlap.
|
||||
func tilesCover(t *testing.T, segs []seg, total int64) {
|
||||
t.Helper()
|
||||
p.mu.Lock()
|
||||
type rng struct{ start, end int64 }
|
||||
rs := make([]rng, len(p.segs))
|
||||
for i, s := range p.segs {
|
||||
rs[i] = rng{s.start, s.endOff()}
|
||||
}
|
||||
p.mu.Unlock()
|
||||
sort.Slice(rs, func(i, j int) bool { return rs[i].start < rs[j].start })
|
||||
sorted := append([]seg(nil), segs...)
|
||||
sort.Slice(sorted, func(i, j int) bool { return sorted[i].start < sorted[j].start })
|
||||
var next int64
|
||||
for _, r := range rs {
|
||||
if r.start != next {
|
||||
t.Fatalf("segment gap/overlap: next byte %d, got start %d (ranges %v)", next, r.start, rs)
|
||||
for _, s := range sorted {
|
||||
if s.start != next {
|
||||
t.Fatalf("segment gap/overlap: next byte %d, got start %d", next, s.start)
|
||||
}
|
||||
next = r.end + 1
|
||||
next = s.endOff() + 1
|
||||
}
|
||||
if next != total {
|
||||
t.Fatalf("segments cover %d bytes, want %d", next, total)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPoolAcquireThenSteal(t *testing.T) {
|
||||
// Sizes are in units of readBuf because that is what newPool floors minSplit
|
||||
// to and what the no-overlap invariant is stated against.
|
||||
const total = 8 * readBuf
|
||||
p := newPool(makeSegments(total, readBuf, 1), readBuf) // one segment [0,total)
|
||||
|
||||
first := p.acquire()
|
||||
if first == nil || first.start != 0 || first.endOff() != total-1 {
|
||||
t.Fatalf("first acquire = %+v, want the whole [0,%d] segment", first, total-1)
|
||||
}
|
||||
if p.acquire() == nil {
|
||||
// nothing fresh left, so this must steal first's back half
|
||||
t.Fatal("second acquire returned nil, want a stolen tail")
|
||||
}
|
||||
// first kept the front half, a tail took the back half; together they still tile.
|
||||
if first.endOff() != total/2-1 {
|
||||
t.Errorf("victim end = %d, want %d after midpoint split", first.endOff(), total/2-1)
|
||||
}
|
||||
assertTiles(t, p, total)
|
||||
}
|
||||
|
||||
func TestPoolNoStealBelowThreshold(t *testing.T) {
|
||||
// remaining is below 2*minSplit, so the lone segment is not worth splitting
|
||||
// and an idle worker is told there is nothing to do.
|
||||
p := newPool(makeSegments(readBuf+readBuf/2, readBuf, 1), readBuf)
|
||||
if s := p.acquire(); s == nil {
|
||||
t.Fatal("first acquire returned nil, want the only segment")
|
||||
}
|
||||
if s := p.acquire(); s != nil {
|
||||
t.Fatalf("second acquire = %+v, want nil (tail would be below min-split)", s)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPoolConcurrentCoverage drives the pool the way real workers do — acquire a
|
||||
// segment, copy it in small chunks, commit each with p.advance, repeat — across
|
||||
// more workers than initial segments so stealing is forced. The coverage array
|
||||
// proves the core invariant: every byte is written exactly once, so a steal
|
||||
// never overlaps the owner's writes and never leaves a gap.
|
||||
func TestPoolConcurrentCoverage(t *testing.T) {
|
||||
// TestSegmentsConcurrentCoverage drives the segments the way real workers do —
|
||||
// one worker per segment, each copying its range in small chunks and committing
|
||||
// with addWritten. The coverage array proves the core invariant: every byte is
|
||||
// written exactly once, with no gap or overlap between adjacent segments.
|
||||
func TestSegmentsConcurrentCoverage(t *testing.T) {
|
||||
const (
|
||||
total = int64(64 * readBuf) // 2 MiB
|
||||
minSplit = int64(readBuf) // steal threshold is 2*this
|
||||
chunk = int64(readBuf / 4) // one "read", well below minSplit
|
||||
workers = 8
|
||||
minSplit = int64(readBuf)
|
||||
chunk = int64(readBuf / 4) // one "read"
|
||||
conns = 8
|
||||
)
|
||||
p := newPool(makeSegments(total, minSplit, 1), minSplit) // start with a single segment
|
||||
segs := makeSegments(total, minSplit, conns)
|
||||
if len(segs) < 2 {
|
||||
t.Fatalf("expected several segments, got %d", len(segs))
|
||||
}
|
||||
cover := make([]uint32, total)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for w := 0; w < workers; w++ {
|
||||
for i := range segs {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
go func(s *seg) {
|
||||
defer wg.Done()
|
||||
for {
|
||||
s := p.acquire()
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
for s.remaining() > 0 {
|
||||
n := chunk
|
||||
if s.remaining() < n {
|
||||
@@ -306,10 +239,9 @@ func TestPoolConcurrentCoverage(t *testing.T) {
|
||||
for j := off; j < off+n; j++ {
|
||||
atomic.AddUint32(&cover[j], 1)
|
||||
}
|
||||
p.advance(s, n)
|
||||
s.addWritten(n)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}(&segs[i])
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
@@ -318,47 +250,46 @@ func TestPoolConcurrentCoverage(t *testing.T) {
|
||||
t.Fatalf("byte %d written %d times, want exactly 1", i, c)
|
||||
}
|
||||
}
|
||||
assertTiles(t, p, total)
|
||||
if len(p.segs) == 1 {
|
||||
t.Error("no stealing happened: still one segment after 8 workers drained it")
|
||||
}
|
||||
tilesCover(t, segs, total)
|
||||
}
|
||||
|
||||
// TestPoolSnapshotAfterStealRoundTrips checks that a steal which happens before a
|
||||
// crash survives the control file: the snapshot records the stolen tail, and a
|
||||
// resume rebuilds a segment set that still tiles the file and keeps the bytes
|
||||
// already written. This is the persistence path an interrupt-and-resume relies on
|
||||
// but cannot deterministically trigger from the outside.
|
||||
func TestPoolSnapshotAfterStealRoundTrips(t *testing.T) {
|
||||
const total = 8 * readBuf
|
||||
p := newPool(makeSegments(total, readBuf, 1), readBuf)
|
||||
a := p.acquire() // the whole file
|
||||
p.advance(a, readBuf) // owner makes some progress, then idles out
|
||||
b := p.acquire() // steals a's back half
|
||||
if b == nil {
|
||||
t.Fatal("expected a stolen tail")
|
||||
// TestSnapshotRoundTrips checks the persistence path an interrupt-and-resume
|
||||
// relies on: a snapshot of partially-written segments records each frontier, and
|
||||
// segsFromControl rebuilds a set that still tiles the file and keeps the bytes
|
||||
// already written.
|
||||
func TestSnapshotRoundTrips(t *testing.T) {
|
||||
const (
|
||||
total = int64(8 * readBuf)
|
||||
minSplit = int64(readBuf)
|
||||
)
|
||||
segs := makeSegments(total, minSplit, 4)
|
||||
if len(segs) < 2 {
|
||||
t.Fatalf("expected several segments, got %d", len(segs))
|
||||
}
|
||||
// Each segment makes some progress, so the snapshot has a real frontier to
|
||||
// carry across the round trip.
|
||||
var want int64
|
||||
for i := range segs {
|
||||
n := int64(readBuf)
|
||||
if n > segs[i].length() {
|
||||
n = segs[i].length()
|
||||
}
|
||||
segs[i].addWritten(n)
|
||||
want += n
|
||||
}
|
||||
p.advance(b, readBuf) // the tail's worker makes progress too
|
||||
|
||||
c := p.snapshot("http://x", total, "", "")
|
||||
if len(c.Segs) != 2 {
|
||||
t.Fatalf("snapshot has %d segments, want 2 (original + stolen tail)", len(c.Segs))
|
||||
c := snapshot("http://x", total, "", "", segs)
|
||||
if len(c.Segs) != len(segs) {
|
||||
t.Fatalf("snapshot has %d segments, want %d", len(c.Segs), len(segs))
|
||||
}
|
||||
|
||||
rebuilt := segsFromControl(&c)
|
||||
sort.Slice(rebuilt, func(i, j int) bool { return rebuilt[i].start < rebuilt[j].start })
|
||||
var next, written int64
|
||||
tilesCover(t, rebuilt, total)
|
||||
var written int64
|
||||
for _, s := range rebuilt {
|
||||
if s.start != next {
|
||||
t.Fatalf("rebuilt gap/overlap: next byte %d, got start %d", next, s.start)
|
||||
}
|
||||
next = s.endOff() + 1
|
||||
written += s.progress()
|
||||
}
|
||||
if next != total {
|
||||
t.Fatalf("rebuilt covers %d bytes, want %d", next, total)
|
||||
}
|
||||
if written != 2*readBuf {
|
||||
t.Errorf("rebuilt written = %d, want %d (bytes must survive the round trip)", written, 2*readBuf)
|
||||
if written != want {
|
||||
t.Errorf("rebuilt written = %d, want %d (bytes must survive the round trip)", written, want)
|
||||
}
|
||||
}
|
||||
|
||||
59
main.go
59
main.go
@@ -100,9 +100,15 @@ func run(args []string) int {
|
||||
eng := download.NewEngine(opts.Int("max-concurrent-downloads"))
|
||||
|
||||
uiCtx, uiCancel := context.WithCancel(context.Background())
|
||||
uiDone := make(chan struct{})
|
||||
if !opts.Bool("quiet") && !opts.Bool("dry-run") {
|
||||
rep := progress.New(eng.Snapshot, opts.Bool("human-readable"))
|
||||
go rep.Run(uiCtx)
|
||||
go func() {
|
||||
rep.Run(uiCtx)
|
||||
close(uiDone)
|
||||
}()
|
||||
} else {
|
||||
close(uiDone)
|
||||
}
|
||||
|
||||
jobsRef.set(jobs)
|
||||
@@ -129,7 +135,7 @@ func run(args []string) int {
|
||||
}
|
||||
|
||||
uiCancel()
|
||||
time.Sleep(20 * time.Millisecond) // let the renderer clear its line
|
||||
<-uiDone // wait for the renderer's final frame before printing the OK/FAIL lines
|
||||
|
||||
saveCookies(jobs)
|
||||
|
||||
@@ -524,7 +530,6 @@ func httpConfig(opts *cli.Options, single bool, overallDL *rate.Limiter) httpdl.
|
||||
Split: opts.Int("split"),
|
||||
MaxConnPerServer: opts.Int("max-connection-per-server"),
|
||||
MinSplit: opts.Size("min-split-size"),
|
||||
AutoSplit: opts.Bool("auto-split"),
|
||||
Tries: opts.Int("max-tries"),
|
||||
Timeout: time.Duration(opts.Int("timeout")) * time.Second,
|
||||
FileAlloc: opts.Str("file-allocation"),
|
||||
@@ -711,42 +716,29 @@ func exitCode(err error) int {
|
||||
|
||||
// isNameResolution reports whether err is (or wraps) a DNS lookup failure —
|
||||
// "name resolution failed" (19), which usually points at the user's own
|
||||
// network or resolver rather than a dead remote.
|
||||
// network or resolver rather than a dead remote. The stdlib reports every such
|
||||
// failure (no-such-host, server-misbehaving) as *net.DNSError.
|
||||
func isNameResolution(err error) bool {
|
||||
var de *net.DNSError
|
||||
if errors.As(err, &de) {
|
||||
return true
|
||||
}
|
||||
msg := strings.ToLower(err.Error())
|
||||
return strings.Contains(msg, "no such host") || strings.Contains(msg, "server misbehaving")
|
||||
return errors.As(err, &de)
|
||||
}
|
||||
|
||||
// isNetwork reports whether err is a connection-level failure — refused, reset,
|
||||
// or an unreachable host/network — a "network problem" (6).
|
||||
// or an unreachable host/network — a "network problem" (6). A dial error from
|
||||
// net/http wraps the syscall errno, so the typed check sees through it.
|
||||
func isNetwork(err error) bool {
|
||||
if errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.ECONNRESET) ||
|
||||
errors.Is(err, syscall.ENETUNREACH) || errors.Is(err, syscall.EHOSTUNREACH) {
|
||||
return true
|
||||
}
|
||||
msg := strings.ToLower(err.Error())
|
||||
return strings.Contains(msg, "connection refused") ||
|
||||
strings.Contains(msg, "connection reset") ||
|
||||
strings.Contains(msg, "network is unreachable") ||
|
||||
strings.Contains(msg, "no route to host")
|
||||
return errors.Is(err, syscall.ECONNREFUSED) || errors.Is(err, syscall.ECONNRESET) ||
|
||||
errors.Is(err, syscall.ENETUNREACH) || errors.Is(err, syscall.EHOSTUNREACH)
|
||||
}
|
||||
|
||||
// isTimeout reports whether err is (or wraps) a timeout: a deadline exceeded, a
|
||||
// net.Error that timed out, or an idle-read timeout reported as such.
|
||||
// net.Error that timed out, or our own idle-read timeout (httpdl.ErrTimeout).
|
||||
func isTimeout(err error) bool {
|
||||
if errors.Is(err, context.DeadlineExceeded) {
|
||||
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, httpdl.ErrTimeout) {
|
||||
return true
|
||||
}
|
||||
var ne net.Error
|
||||
if errors.As(err, &ne) && ne.Timeout() {
|
||||
return true
|
||||
}
|
||||
msg := strings.ToLower(err.Error())
|
||||
return strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out")
|
||||
return errors.As(err, &ne) && ne.Timeout()
|
||||
}
|
||||
|
||||
// saveSession writes the sources of every download that did not finish to path,
|
||||
@@ -831,6 +823,11 @@ func makeLimiter(bps int64) *rate.Limiter {
|
||||
return rate.NewLimiter(rate.Limit(bps), download.LimiterBurst(bps))
|
||||
}
|
||||
|
||||
// maxSelectIndex bounds how far a --select-file range is expanded: a torrent has
|
||||
// at most a few thousand files, so a degenerate spec like "1-9999999999" must not
|
||||
// spin or balloon the map. Indexes past the real file count are ignored anyway.
|
||||
const maxSelectIndex = 1 << 20
|
||||
|
||||
// parseSelect parses "1,3-5" into a set of 1-based file indexes.
|
||||
func parseSelect(s string) map[int]bool {
|
||||
s = strings.TrimSpace(s)
|
||||
@@ -848,10 +845,14 @@ func parseSelect(s string) map[int]bool {
|
||||
continue
|
||||
}
|
||||
b, _ := strconv.Atoi(strings.TrimSpace(hi))
|
||||
for i := a; i <= b; i++ {
|
||||
if i > 0 {
|
||||
out[i] = true
|
||||
if a < 1 {
|
||||
a = 1
|
||||
}
|
||||
if b > maxSelectIndex {
|
||||
b = maxSelectIndex
|
||||
}
|
||||
for i := a; i <= b; i++ {
|
||||
out[i] = true
|
||||
}
|
||||
}
|
||||
return out
|
||||
|
||||
@@ -69,16 +69,14 @@ func TestExitCode(t *testing.T) {
|
||||
err error
|
||||
want int
|
||||
}{
|
||||
{"idle timeout", errors.New("idle timeout: no data received"), 2},
|
||||
{"metadata timed out", errors.New("timed out fetching metadata after 3s"), 2},
|
||||
{"idle timeout", fmt.Errorf("segment 0: %w (after 1m0s)", httpdl.ErrTimeout), 2},
|
||||
{"metadata timed out", fmt.Errorf("timed out fetching metadata after 3s: %w", context.DeadlineExceeded), 2},
|
||||
{"deadline", context.DeadlineExceeded, 2},
|
||||
{"file exists", fmt.Errorf("/tmp/x: %w (use --allow-overwrite or -c)", httpdl.ErrOutputExists), 13},
|
||||
{"dns typed", dnsErr, 19},
|
||||
{"dns wrapped through retries", fmt.Errorf("http://x: %w (after 5 tries)", dnsErr), 19},
|
||||
{"dns by message", errors.New("dial tcp: lookup nope.invalid: no such host"), 19},
|
||||
{"refused typed", refused, 6},
|
||||
{"refused wrapped through retries", fmt.Errorf("http://x: %w (after 5 tries)", refused), 6},
|
||||
{"refused by message", errors.New("dial tcp 127.0.0.1:1: connect: connection refused"), 6},
|
||||
{"not found", fmt.Errorf("%w: http://x/y: 404 Not Found", httpdl.ErrNotFound), 3},
|
||||
{"too slow", fmt.Errorf("http://x: %w (<= %d bytes/sec)", httpdl.ErrTooSlow, 1024), 5},
|
||||
{"checksum", fmt.Errorf("/tmp/x: %w (want a, got b)", httpdl.ErrChecksum), 32},
|
||||
@@ -227,7 +225,7 @@ func TestFollowUpInfoHashDedup(t *testing.T) {
|
||||
func TestReportExit(t *testing.T) {
|
||||
canceled := download.Result{Name: "a", Err: context.Canceled}
|
||||
done := download.Result{Name: "b", Err: nil}
|
||||
timeout := download.Result{Name: "c", Err: errors.New("idle timeout")}
|
||||
timeout := download.Result{Name: "c", Err: httpdl.ErrTimeout}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
|
||||
@@ -153,8 +153,7 @@ func (r *Reporter) lineOne(s download.Stat) string {
|
||||
fmt.Fprintf(&b, " UL:%s", speed(ul, r.human))
|
||||
}
|
||||
if !seeding && s.Total > 0 && dl > 0 {
|
||||
eta := time.Duration(float64(s.Total-s.Completed)/float64(dl)) * time.Second
|
||||
fmt.Fprintf(&b, " ETA:%s", secfmt(eta))
|
||||
fmt.Fprintf(&b, " ETA:%s", secfmt(etaDuration(s.Total-s.Completed, dl)))
|
||||
}
|
||||
b.WriteByte(']')
|
||||
return b.String()
|
||||
@@ -215,16 +214,23 @@ func (r *Reporter) summary(stats []download.Stat) string {
|
||||
return ""
|
||||
}
|
||||
var totDL, totUL int64
|
||||
stalled := 0
|
||||
stalled, waiting := 0, 0
|
||||
for _, s := range stats {
|
||||
dl, ul := r.rates(s)
|
||||
totDL += dl
|
||||
totUL += ul
|
||||
if s.Status == download.Waiting {
|
||||
waiting++
|
||||
continue
|
||||
}
|
||||
if r.isStalled(s, dl) {
|
||||
stalled++
|
||||
}
|
||||
}
|
||||
line := fmt.Sprintf("%d active DL:%s UL:%s", len(stats), speed(totDL, r.human), speed(totUL, r.human))
|
||||
line := fmt.Sprintf("%d active DL:%s UL:%s", len(stats)-waiting, speed(totDL, r.human), speed(totUL, r.human))
|
||||
if waiting > 0 {
|
||||
line += fmt.Sprintf(" (%d waiting)", waiting)
|
||||
}
|
||||
if stalled > 0 {
|
||||
line += fmt.Sprintf(" (%d stalled)", stalled)
|
||||
}
|
||||
@@ -245,6 +251,8 @@ func (r *Reporter) tableRow(s download.Stat) string {
|
||||
// upload rate, or a "stalled" flag).
|
||||
func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string) {
|
||||
switch s.Status {
|
||||
case download.Waiting:
|
||||
return "queued", ""
|
||||
case download.Seeding:
|
||||
return "seeding", "UL:" + speed(ul, r.human)
|
||||
case download.Complete:
|
||||
@@ -255,8 +263,7 @@ func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string
|
||||
metric = "DL:" + speed(dl, r.human)
|
||||
switch {
|
||||
case dl > 0 && s.Total > 0:
|
||||
eta := time.Duration(float64(s.Total-s.Completed)/float64(dl)) * time.Second
|
||||
tail = secfmt(eta)
|
||||
tail = secfmt(etaDuration(s.Total-s.Completed, dl))
|
||||
case r.isStalled(s, dl):
|
||||
tail = "stalled"
|
||||
}
|
||||
@@ -264,6 +271,20 @@ func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string
|
||||
}
|
||||
}
|
||||
|
||||
// maxETASeconds caps an ETA before it becomes a Duration: a near-stalled download
|
||||
// yields an astronomically large seconds value, and time.Duration(secs) *
|
||||
// time.Second would overflow int64 and wrap to a bogus tiny ETA. secfmt renders
|
||||
// anything past 99h as "--", so clamping to 100h loses nothing.
|
||||
const maxETASeconds = 100 * 3600
|
||||
|
||||
func etaDuration(remaining, dl int64) time.Duration {
|
||||
secs := float64(remaining) / float64(dl)
|
||||
if secs > maxETASeconds {
|
||||
secs = maxETASeconds
|
||||
}
|
||||
return time.Duration(secs) * time.Second
|
||||
}
|
||||
|
||||
// isStalled reports an active download that has gone a while with no new bytes,
|
||||
// so the table can flag it rather than leave the eye to read DL:0B. A download
|
||||
// too young to have a rate yet is not stalled.
|
||||
|
||||
Reference in New Issue
Block a user