Hojun-Cho/got
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

httpdl: add segment work-stealing and --auto-split

Browse Source 
an idle worker now steals the back half of the busiest in-flight segment
instead of exiting, so one slow segment no longer drains alone over a single
connection. control file persists stolen tails; resume re-tiles.

add --auto-split (opt-in): derive the connection count from file size (one
per min-split-size, up to 16), overriding -x/-s. default stays at 1
connection for a single server.
This commit is contained in:
Hojun-Cho
2026-06-19 13:56:31 +09:00
parent eb8d9b9460
commit 554beb6b48
7 changed files with 407 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
httpdl/httpdl.go
View File

@@ -57,7 +57,8 @@ type Config struct {
Split int // --split: total connections across all mirrors
MaxConnPerServer int // --max-connection-per-server: per-host connection cap
MinSplit int64
Tries int // 0 = unlimited
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
Continue bool
@@ -119,7 +120,7 @@ type Config struct {
// connections.
type Download struct {
uris []string // mirror list; uris[0] is the primary (naming + resume key)
maxConns int // segment workers = min(split, len(uris)*max-connection-per-server)
maxConns int // manual segment-worker cap = min(split, len(uris)*max-connection-per-server); --auto-split derives its own count per file in segmented()
cfg Config
client *http.Client
limit []*rate.Limiter
@@ -214,10 +215,17 @@ 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: perHost,
MaxIdleConnsPerHost: perHost,
MaxConnsPerHost: connCap,
MaxIdleConnsPerHost: connCap,
ResponseHeaderTimeout: cfg.Timeout,
TLSHandshakeTimeout: connectTimeout,
DialContext: dialContext,
@@ -610,7 +618,13 @@ 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 {
segs := makeSegments(total, d.cfg.MinSplit, d.maxConns)
// --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)
if c := d.resumeControl(out, etag, lastmod); c != nil {
segs = segsFromControl(c)
} else if d.cfg.Continue && !fileExists(controlPath(out)) {
@@ -648,36 +662,31 @@ 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, segs, saveDone)
jobs := make(chan *seg)
go func() {
defer close(jobs)
for i := range segs {
if segs[i].done() {
continue
}
select {
case jobs <- &segs[i]:
case <-ctx.Done(): // stop feeding once we're tearing down
return
}
}
}()
go d.saveLoop(ctx, out, total, etag, lastmod, p, 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.
var (
wg sync.WaitGroup
errOnce sync.Once
runErr error
)
for w := 0; w < d.maxConns; w++ {
for w := 0; w < conns; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for s := range jobs {
if err := d.fetchSeg(ctx, f, s, total); err != nil {
for {
s := p.acquire()
if s == nil {
return
}
if err := d.fetchSeg(ctx, f, p, s, total); err != nil {
errOnce.Do(func() { runErr = err; cancel() })
return
}
@@ -691,7 +700,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
}
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
p.snapshot(d.primary(), total, etag, lastmod).save(out)
removeControl(out)
return nil
}
@@ -699,7 +708,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, s *seg, total int64) error {
func (d *Download) fetchSeg(ctx context.Context, f *os.File, p *pool, 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.
@@ -716,7 +725,7 @@ func (d *Download) fetchSeg(ctx context.Context, f *os.File, s *seg, total int64
if s.done() {
return nil
}
return d.fetchOnce(ctx, f, s, uri, total)
return d.fetchOnce(ctx, f, p, s, uri, total)
})
if err == nil {
return nil
@@ -726,10 +735,10 @@ func (d *Download) fetchSeg(ctx context.Context, f *os.File, s *seg, total int64
return lastErr
}
func (d *Download) fetchOnce(ctx context.Context, f *os.File, 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 {
reqCtx, cancel := context.WithCancel(ctx)
defer cancel()
req, err := d.request(reqCtx, uri, fmt.Sprintf("bytes=%d-%d", s.offset(), s.end))
req, err := d.request(reqCtx, uri, fmt.Sprintf("bytes=%d-%d", s.offset(), s.endOff()))
if err != nil {
return err
}
@@ -755,7 +764,7 @@ func (d *Download) fetchOnce(ctx context.Context, f *os.File, s *seg, uri string
}
body, stop := d.idleGuard(resp.Body, cancel)
defer stop()
return d.pump(ctx, f, s, body)
return d.pump(ctx, f, p, s, body)
}
// errIdleTimeout marks a read that stalled past the idle window, so callers can
@@ -901,8 +910,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.
func (d *Download) pump(ctx context.Context, f *os.File, s *seg, body io.Reader) error {
// 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 {
buf := make([]byte, readBuf)
for s.remaining() > 0 {
n := int64(len(buf))
@@ -914,7 +925,7 @@ func (d *Download) pump(ctx context.Context, f *os.File, s *seg, body io.Reader)
if _, werr := f.WriteAt(buf[:rd], s.offset()); werr != nil {
return werr
}
s.advance(int64(rd))
p.advance(s, int64(rd))
atomic.AddInt64(&d.completed, int64(rd))
d.throttle(ctx, rd)
}
@@ -1066,7 +1077,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, segs []seg, done chan<- struct{}) {
func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag, lastmod string, p *pool, done chan<- struct{}) {
defer close(done)
interval := d.cfg.AutoSaveInterval
if interval <= 0 {
@@ -1077,10 +1088,10 @@ func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag,
for {
select {
case <-ctx.Done():
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
p.snapshot(d.primary(), total, etag, lastmod).save(out)
return
case <-t.C:
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
p.snapshot(d.primary(), total, etag, lastmod).save(out)
}
}
}