From 7a0d3f547895adc5182c684dabadfbfd20edf0f1 Mon Sep 17 00:00:00 2001 From: "Silas, Mariusz Grzybacz" Date: Fri, 2 Oct 2026 21:56:05 +0000 Subject: [PATCH 1/5] feat(http): adaptive connections Add an opt-in mode that sizes the connection count on measured throughput, like Internet Download Manager: start with one connection, add one while the task speed rises by more than 10%, stop at a plateau (under 5%), give connections back when the speed stays 15% below the best for two windows, and probe one more every 30 s below the ceiling. The `connections` option stays the ceiling. A 403 or 429 on a new connection sets the ceiling to the count that worked; a 429 or 503 on a connection that already had data parks it. - adaptive.go: a pure controller (2 s windows) with no I/O. - fetcher.go: a parked connection state (stops at its offset; its range goes back to the pool and is taken whole by helpOtherConnection or the next addition; not a failure), monotonic connection IDs, Patch of `connections` (live for adaptive tasks) and `adaptive`, and growth again after resume. - Options: `downloader.http.adaptive` (default false) and a per-task `opts.extra.adaptive` (*bool, omitempty; nil follows the global). - UI: a settings switch under HTTP and a per-task switch. With adaptive off the code path is today's: slow start still commits batches of 1, 2, 4, 9 at 16 connections (tested), and records saved by older versions resume unchanged (tested). Servers without Range stay sequential. --- internal/protocol/http/adaptive.go | 305 +++++++++ .../http/adaptive_integration_test.go | 622 +++++++++++++++++ internal/protocol/http/adaptive_test.go | 277 ++++++++ internal/protocol/http/config.go | 4 + internal/protocol/http/fetcher.go | 623 +++++++++++++++--- pkg/protocol/http/model.go | 8 + .../lib/api/model/downloader_config.dart | 3 +- .../lib/api/model/downloader_config.g.dart | 2 + ui/flutter/lib/api/model/options.dart | 2 + ui/flutter/lib/api/model/options.g.dart | 2 + .../presentation/pages/settings_page.dart | 9 + .../pages/create_task_window_page.dart | 16 + ui/flutter/lib/l10n/app_en.arb | 2 + 13 files changed, 1798 insertions(+), 77 deletions(-) create mode 100644 internal/protocol/http/adaptive.go create mode 100644 internal/protocol/http/adaptive_integration_test.go create mode 100644 internal/protocol/http/adaptive_test.go diff --git a/internal/protocol/http/adaptive.go b/internal/protocol/http/adaptive.go new file mode 100644 index 0000000..057074d --- /dev/null +++ b/internal/protocol/http/adaptive.go @@ -0,0 +1,305 @@ +package http + +import "time" + +// Adaptive connections: start with one connection, add one while the task +// speed rises, and give connections back when more of them makes it slower. +// The controller below is pure: it sees only the clock, the task speed and +// the number of running connections, and returns a decision that the fetcher +// carries out. Its numbers come from the design: +// +// - every window (adaptiveTickInterval) is one speed sample; +// - an addition is judged on the mean of the next adaptiveEvalWindows +// windows against the speed before it; +// - a gain above adaptiveGrowGain keeps the connection and adds another; +// - a gain of at least adaptivePlateauGain keeps it but stops growth; +// - a smaller gain is a plateau: the connection is given back; +// - adaptiveEvalWindows windows below (1 - adaptiveShrinkDrop) of the best +// speed park the slowest connection; +// - every adaptiveProbeInterval, below the ceiling, one more connection is +// tried and kept only if it helps. +var ( + adaptiveTickInterval = 2 * time.Second + adaptiveProbeInterval = 30 * time.Second +) + +const ( + adaptiveEvalWindows = 2 + adaptiveGrowGain = 0.10 + adaptivePlateauGain = 0.05 + adaptiveShrinkDrop = 0.15 +) + +type adaptiveDecision int + +const ( + holdConns adaptiveDecision = iota + addConn + parkSlowest +) + +func (d adaptiveDecision) String() string { + switch d { + case addConn: + return "addConn" + case parkSlowest: + return "parkSlowest" + default: + return "holdConns" + } +} + +type adaptivePhase int + +const ( + // phaseMeasure collects windows at the current count, then either grows + // (when grow is set) or settles. + phaseMeasure adaptivePhase = iota + // phaseAdded judges the connection that was just added. + phaseAdded + // phaseParked judges a connection parked because the speed dropped. + phaseParked + // phaseSteady watches for a drop and probes on a timer. + phaseSteady +) + +type adaptiveController struct { + ceiling int + phase adaptivePhase + grow bool // phaseMeasure: add a connection once measured + // keepBest (phaseMeasure): the count is one the controller has measured + // before, so a lower result is a slowdown to react to, not the new best. + keepBest bool + + // expect is the running count the controller's own decisions should + // produce. Any other count was caused from outside (a connection ran out + // of work, or an addition found nothing to split) and starts a new + // measurement instead of being read as a slowdown. + expect int + + samples []int64 // windows collected for the current judgement + before int64 // speed before the change being judged + best int64 // best mean speed at the current count + + lastWindow int64 // phaseSteady: the previous window, for a two-window mean + lowWindows int // phaseSteady: consecutive windows below the shrink line + lowSum int64 + + // probeAt is when the next probe may start. It is zero until the first + // settle, and a probe moves it a full interval on from its own start. + probeAt time.Time +} + +func newAdaptiveController(ceiling int, now time.Time) *adaptiveController { + if ceiling < 1 { + ceiling = 1 + } + return &adaptiveController{ + ceiling: ceiling, + phase: phaseMeasure, + grow: true, + expect: 1, + } +} + +// setCeiling changes the maximum, for example when the task's connections +// option is patched while it runs. Running connections above it are parked +// one per tick. +func (c *adaptiveController) setCeiling(n int) { + if n < 1 { + n = 1 + } + c.ceiling = n +} + +// tick is called once per window with the task speed (bytes/s) measured over +// that window and the number of running connections. +func (c *adaptiveController) tick(now time.Time, speed int64, active int) adaptiveDecision { + if active > c.ceiling && active > 1 { + // A lower ceiling makes a lower speed expected. + c.measure(active-1, false, false) + return parkSlowest + } + if active != c.expect { + // Not our doing: measure the new count before judging anything. + c.measure(active, false, false) + } + + if c.phase == phaseSteady { + return c.steady(now, speed, active) + } + + c.samples = append(c.samples, speed) + if len(c.samples) < adaptiveEvalWindows { + return holdConns + } + mean := meanOf(c.samples) + c.samples = c.samples[:0] + + switch c.phase { + case phaseMeasure: + if !c.keepBest || mean > c.best { + c.best = mean + } + if c.grow && active < c.ceiling { + return c.add(mean, active) + } + c.settle(now) + return holdConns + + case phaseAdded: + gain := gainOf(mean, c.before) + switch { + case gain > adaptiveGrowGain: + c.best = mean + if active < c.ceiling { + return c.add(mean, active) + } + c.settle(now) + return holdConns + case gain >= adaptivePlateauGain: + c.best = mean + c.settle(now) + return holdConns + default: + // Plateau: the new connection did not help. Give it back and + // measure the smaller count before settling. Its earlier speed + // stays the best, so a slowdown that began meanwhile still shows. + if active > 1 { + c.best = c.before + c.measure(active-1, false, true) + return parkSlowest + } + c.settle(now) + return holdConns + } + + case phaseParked: + if float64(mean) >= float64(c.before)*(1-adaptivePlateauGain) { + // The parked connection was not needed. Keep walking down while + // parking is free. + c.best = mean + if active > 1 { + return c.park(mean, active) + } + c.settle(now) + return holdConns + } + // This park cost speed: undo it and settle at the count before it. + c.best = c.before + if active < c.ceiling { + c.measure(active+1, false, true) + return addConn + } + c.settle(now) + return holdConns + } + return holdConns +} + +// limited reports that a connection got a limiting HTTP status. With +// newConn, a 403 or 429 on a connection that never got data, the server's +// limit is the count that worked, which becomes the ceiling. Otherwise a 429 +// or 503 hit a connection mid-stream; the fetcher has parked it, and the next +// probe waits a full interval. active is the running count after the park. +func (c *adaptiveController) limited(now time.Time, active int, newConn bool) { + if active < 1 { + active = 1 + } + if newConn && active < c.ceiling { + c.ceiling = active + } + c.probeAt = now.Add(adaptiveProbeInterval) + // A refused new connection changed nothing that runs, so the best stands. + // A connection parked mid-stream is the server asking for less: measure + // afresh rather than read the lower speed as a further slowdown. + c.measure(active, false, newConn) +} + +func (c *adaptiveController) steady(now time.Time, speed int64, active int) adaptiveDecision { + if c.lastWindow > 0 { + if pair := (c.lastWindow + speed) / 2; pair > c.best { + c.best = pair + } + } + prev := c.lastWindow + c.lastWindow = speed + + if float64(speed) < float64(c.best)*(1-adaptiveShrinkDrop) { + c.lowWindows++ + c.lowSum += speed + } else { + c.lowWindows, c.lowSum = 0, 0 + } + if c.lowWindows >= adaptiveEvalWindows { + low := c.lowSum / int64(c.lowWindows) + c.lowWindows, c.lowSum = 0, 0 + if active > 1 { + return c.park(low, active) + } + // One connection is already the minimum: accept the new speed. + c.best = low + return holdConns + } + + if active < c.ceiling && !now.Before(c.probeAt) { + c.probeAt = now.Add(adaptiveProbeInterval) + before := speed + if prev > 0 { + before = (prev + speed) / 2 + } + return c.add(before, active) + } + return holdConns +} + +func (c *adaptiveController) add(before int64, active int) adaptiveDecision { + c.phase = phaseAdded + c.before = before + c.expect = active + 1 + c.samples = c.samples[:0] + return addConn +} + +func (c *adaptiveController) park(before int64, active int) adaptiveDecision { + c.phase = phaseParked + c.before = before + c.expect = active - 1 + c.samples = c.samples[:0] + return parkSlowest +} + +func (c *adaptiveController) measure(expect int, grow, keepBest bool) { + c.phase = phaseMeasure + c.grow = grow + c.keepBest = keepBest + c.expect = expect + c.samples = c.samples[:0] +} + +func (c *adaptiveController) settle(now time.Time) { + c.phase = phaseSteady + c.lastWindow = 0 + c.lowWindows, c.lowSum = 0, 0 + if c.probeAt.IsZero() || c.probeAt.Before(now) { + c.probeAt = now.Add(adaptiveProbeInterval) + } +} + +func meanOf(xs []int64) int64 { + var sum int64 + for _, x := range xs { + sum += x + } + return sum / int64(len(xs)) +} + +func gainOf(after, before int64) float64 { + if before <= 0 { + if after > 0 { + return 1 + } + return 0 + } + return float64(after)/float64(before) - 1 +} diff --git a/internal/protocol/http/adaptive_integration_test.go b/internal/protocol/http/adaptive_integration_test.go new file mode 100644 index 0000000..28cfd08 --- /dev/null +++ b/internal/protocol/http/adaptive_integration_test.go @@ -0,0 +1,622 @@ +package http + +import ( + "bytes" + "encoding/json" + "fmt" + gohttp "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/GopeedLab/gopeed/internal/controller" + "github.com/GopeedLab/gopeed/internal/fetcher" + "github.com/GopeedLab/gopeed/pkg/base" + "github.com/GopeedLab/gopeed/pkg/protocol/http" +) + +// ============================================================================ +// Test server: per-request rate, total cap, and a connection limit with 429. +// ============================================================================ + +const adaptivePiece = 16 * 1024 + +type adaptiveTestServer struct { + data []byte + perConn int64 // bytes/s for each request + totalCap atomic.Int64 // bytes/s for all requests together, 0 means no cap + maxConns atomic.Int32 // concurrent requests above this get 429, 0 means no limit + noRange bool + + active atomic.Int32 + rejected atomic.Int32 + + mu sync.Mutex + nextFree time.Time // pacing clock shared by all requests for the total cap + + srv *httptest.Server +} + +func newAdaptiveTestServer(t *testing.T, size int, perConn, totalCap int64, maxConns int32, noRange bool) *adaptiveTestServer { + t.Helper() + s := &adaptiveTestServer{data: adaptiveTestData(size), perConn: perConn, noRange: noRange} + s.totalCap.Store(totalCap) + s.maxConns.Store(maxConns) + s.srv = httptest.NewServer(gohttp.HandlerFunc(s.serve)) + t.Cleanup(s.srv.Close) + return s +} + +func adaptiveTestData(size int) []byte { + data := make([]byte, size) + for i := range data { + data[i] = byte(i % 251) + } + return data +} + +func (s *adaptiveTestServer) url() string { + return s.srv.URL + "/adaptive.bin" +} + +// reserve books one piece on the shared clock and returns when it may be sent. +func (s *adaptiveTestServer) reserve(n int) time.Time { + limit := s.totalCap.Load() + now := time.Now() + if limit <= 0 { + return now + } + s.mu.Lock() + defer s.mu.Unlock() + at := s.nextFree + if at.Before(now) { + at = now + } + s.nextFree = at.Add(time.Duration(float64(n) / float64(limit) * float64(time.Second))) + return at +} + +func (s *adaptiveTestServer) serve(w gohttp.ResponseWriter, r *gohttp.Request) { + n := s.active.Add(1) + defer s.active.Add(-1) + if limit := s.maxConns.Load(); limit > 0 && n > limit { + s.rejected.Add(1) + w.WriteHeader(gohttp.StatusTooManyRequests) + return + } + + start, end := int64(0), int64(len(s.data)-1) + status := gohttp.StatusOK + if !s.noRange { + w.Header().Set("Accept-Ranges", "bytes") + if spec := r.Header.Get("Range"); spec != "" { + bounds := strings.SplitN(strings.TrimPrefix(spec, "bytes="), "-", 2) + start, _ = strconv.ParseInt(bounds[0], 10, 64) + if len(bounds) == 2 && bounds[1] != "" { + end, _ = strconv.ParseInt(bounds[1], 10, 64) + } + if end > int64(len(s.data)-1) { + end = int64(len(s.data) - 1) + } + status = gohttp.StatusPartialContent + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(s.data))) + } + } + w.Header().Set("Content-Length", strconv.FormatInt(end-start+1, 10)) + w.WriteHeader(status) + + connNext := time.Now() + for pos := start; pos <= end; { + size := min(int64(adaptivePiece), end-pos+1) + now := time.Now() + if connNext.Before(now) { + connNext = now // no burst credit for time spent waiting on the cap + } + at := connNext + if shared := s.reserve(int(size)); shared.After(at) { + at = shared + } + if d := time.Until(at); d > 0 { + time.Sleep(d) + } + if _, err := w.Write(s.data[pos : pos+size]); err != nil { + return + } + pos += size + connNext = at.Add(time.Duration(float64(size) / float64(s.perConn) * float64(time.Second))) + } +} + +// ============================================================================ +// Helpers +// ============================================================================ + +// useFastAdaptiveClock shortens the controller's windows so the integration +// tests converge in seconds. The ratios between window, evaluation and probe +// stay those of the real 2 s / 30 s clock closely enough for the behaviour. +func useFastAdaptiveClock(t *testing.T) { + t.Helper() + tick, probe := adaptiveTickInterval, adaptiveProbeInterval + adaptiveTickInterval, adaptiveProbeInterval = 300*time.Millisecond, 3*time.Second + t.Cleanup(func() { adaptiveTickInterval, adaptiveProbeInterval = tick, probe }) +} + +func boolPtr(v bool) *bool { return &v } + +func newAdaptiveFetcher(t *testing.T, cfg config, s *adaptiveTestServer, extra *http.OptsExtra) (*Fetcher, string) { + t.Helper() + dir := t.TempDir() + f := buildConfigFetcher(cfg).(*Fetcher) + var optsExtra any // a typed nil would read as "set" + if extra != nil { + optsExtra = extra + } + err := f.Resolve(&base.Request{URL: s.url()}, &base.Options{ + Name: "adaptive.bin", + Path: dir, + Extra: optsExtra, + }) + if err != nil { + t.Fatal(err) + } + return f, filepath.Join(dir, "adaptive.bin") +} + +func assertAdaptiveFile(t *testing.T, path string, want []byte) { + t.Helper() + got, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, want) { + t.Fatalf("downloaded file differs from the served data (got %d bytes, want %d)", len(got), len(want)) + } +} + +// connTimeline samples the server's concurrent requests every 100 ms. +type connTimeline struct { + mu sync.Mutex + samples []int + stop chan struct{} + done chan struct{} +} + +func sampleConnections(s *adaptiveTestServer) *connTimeline { + tl := &connTimeline{stop: make(chan struct{}), done: make(chan struct{})} + go func() { + defer close(tl.done) + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-tl.stop: + return + case <-ticker.C: + tl.mu.Lock() + tl.samples = append(tl.samples, int(s.active.Load())) + tl.mu.Unlock() + } + } + }() + return tl +} + +func (tl *connTimeline) finish() []int { + close(tl.stop) + <-tl.done + tl.mu.Lock() + defer tl.mu.Unlock() + return append([]int(nil), tl.samples...) +} + +// snapshot returns the samples taken so far. +func (tl *connTimeline) snapshot() []int { + tl.mu.Lock() + defer tl.mu.Unlock() + return append([]int(nil), tl.samples...) +} + +// settled is the most common connection count in samples[from:to]. +func settled(samples []int, from, to int) int { + if to > len(samples) { + to = len(samples) + } + if from >= to { + return -1 + } + counts := map[int]int{} + for _, v := range samples[from:to] { + counts[v]++ + } + keys := make([]int, 0, len(counts)) + for k := range counts { + keys = append(keys, k) + } + sort.Ints(keys) + best := -1 + for _, k := range keys { + if best < 0 || counts[k] > counts[best] { + best = k + } + } + return best +} + +// waitFor polls cond every 50 ms until it holds or the timeout passes. +func waitFor(t *testing.T, timeout time.Duration, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out after %v waiting for %s", timeout, what) + } + time.Sleep(50 * time.Millisecond) + } +} + +func waitDone(t *testing.T, f fetcher.Fetcher, timeout time.Duration) error { + t.Helper() + ch := make(chan error, 1) + go func() { ch <- f.Wait() }() + select { + case err := <-ch: + return err + case <-time.After(timeout): + t.Fatalf("download did not finish within %v", timeout) + return nil + } +} + +// ============================================================================ +// Tests +// ============================================================================ + +func TestAdaptiveIntegrationSettlesAtTotalCap(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) + + tl := sampleConnections(s) + started := time.Now() + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 60*time.Second); err != nil { + t.Fatal(err) + } + samples := tl.finish() + t.Logf("took %v; connections every 100 ms: %v", time.Since(started).Round(time.Millisecond), samples) + + // Skip the growth (first 4 s) and the tail where chunks run out (last 1.5 s). + got := settled(samples, 40, len(samples)-15) + t.Logf("settled at %d connections", got) + if got < 5 || got > 7 { + t.Fatalf("settled at %d connections, want 6 +- 1", got) + } + if f.adaptive == nil { + t.Fatal("adaptive controller was not used") + } + assertAdaptiveFile(t, path, s.data) +} + +func TestAdaptiveIntegrationConnectionLimit429(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 40<<20, 1<<20, 0, 4, false) + f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) + + tl := sampleConnections(s) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 60*time.Second); err != nil { + t.Fatalf("a 429 connection limit must not fail the task: %v", err) + } + samples := tl.finish() + t.Logf("connections every 100 ms: %v; rejected with 429: %d", samples, s.rejected.Load()) + + got := settled(samples, 30, len(samples)-15) + t.Logf("settled at %d connections", got) + if got != 4 { + t.Fatalf("settled at %d connections, want 4", got) + } + if s.rejected.Load() == 0 { + t.Fatal("the server never refused a connection, so the limit was not found") + } + for i, c := range f.Stats().Snapshot.(*http.Stats).Connections { + if c.Failed { + t.Fatalf("connection %d reports a failure; a parked connection is not a failure", i) + } + } + assertAdaptiveFile(t, path, s.data) +} + +func TestAdaptiveIntegrationSlowdownGivesBack(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) + + tl := sampleConnections(s) + if err := f.Start(); err != nil { + t.Fatal(err) + } + waitFor(t, 15*time.Second, "growth to 6 connections", func() bool { return s.active.Load() >= 6 }) + time.Sleep(time.Second) + + // The server now gives only 2 MB/s in total. + s.totalCap.Store(2 << 20) + mark := len(tl.snapshot()) + defer func() { + if t.Failed() { + t.Logf("after the slowdown: %v", tl.snapshot()[mark:]) + } + }() + waitFor(t, 10*time.Second, "connections given back", func() bool { + recent := tl.snapshot()[mark:] + return len(recent) >= 10 && settled(recent, len(recent)-10, len(recent)) <= 3 + }) + gaveBack := tl.snapshot()[mark:] + t.Logf("after the slowdown: %v", gaveBack) + + // Let the rest finish quickly. + s.totalCap.Store(0) + if err := waitDone(t, f, 60*time.Second); err != nil { + t.Fatal(err) + } + tl.finish() + assertAdaptiveFile(t, path, s.data) +} + +func TestAdaptiveIntegrationPauseAndResumeGrowsAgain(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) + if err := f.Start(); err != nil { + t.Fatal(err) + } + waitFor(t, 15*time.Second, "growth to 5 connections", func() bool { return s.active.Load() >= 5 }) + + // Store, pause and restore through JSON, as the downloader does. + fm := new(FetcherManager) + data, err := fm.Store(f) + if err != nil { + t.Fatal(err) + } + if err := f.Pause(); err != nil { + t.Fatal(err) + } + waitFor(t, 5*time.Second, "the server to see the pause", func() bool { return s.active.Load() == 0 }) + + rawData, _ := json.Marshal(data) + rawMeta, _ := json.Marshal(f.Meta()) + var meta fetcher.FetcherMeta + if err := json.Unmarshal(rawMeta, &meta); err != nil { + t.Fatal(err) + } + v, restore := fm.Restore() + if err := json.Unmarshal(rawData, v); err != nil { + t.Fatal(err) + } + resumed := restore(&meta, v).(*Fetcher) + ctl := controller.NewController() + ctl.GetConfig = func(v any) { + raw, _ := json.Marshal(config{Connections: 16, Adaptive: true}) + json.Unmarshal(raw, v) + } + resumed.Setup(ctl) + + tl := sampleConnections(s) + if err := resumed.Start(); err != nil { + t.Fatal(err) + } + waitFor(t, 15*time.Second, "growth after resume", func() bool { return s.active.Load() >= 5 }) + time.Sleep(200 * time.Millisecond) // let the sampler record it + early := tl.snapshot() + t.Logf("connections after resume: %v", early) + // The controller measures one connection for two 300 ms windows before it + // adds the second, so the first 500 ms of traffic show one request. (A + // restored fetcher first waits up to 10 s in stopPrefetchAndCopyData for a + // prefetch that never ran; that delay predates this change.) + first := 0 + for first < len(early) && early[first] == 0 { + first++ + } + for i := first; i < len(early) && i < first+5; i++ { + if early[i] > 1 { + t.Fatalf("resume ran %d connections %d ms after its first request, want growth from one", early[i], (i-first)*100) + } + } + if m := maxOf(early); m < 5 { + t.Fatalf("resume grew to %d connections, want at least 5", m) + } + if err := waitDone(t, resumed, 60*time.Second); err != nil { + t.Fatal(err) + } + tl.finish() + assertAdaptiveFile(t, path, s.data) +} + +func TestAdaptiveIntegrationNoRangeIsSequential(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 4<<20, 8<<20, 0, 0, true) + f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) + if f.Meta().Res.Range { + t.Fatal("setup: the server must not advertise Range") + } + tl := sampleConnections(s) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 30*time.Second); err != nil { + t.Fatalf("a server without Range must download without error: %v", err) + } + samples := tl.finish() + for _, n := range samples { + if n > 1 { + t.Fatalf("saw %d concurrent requests, want one sequential download", n) + } + } + if n := len(f.Stats().Snapshot.(*http.Stats).Connections); n != 1 { + t.Fatalf("got %d connections, want 1", n) + } + assertAdaptiveFile(t, path, s.data) +} + +// With adaptive off, the slow-start batches are today's: 1, 2, 4, then the +// rest up to the configured 16 (the batch size squares and is capped). +func TestAdaptiveOffKeepsSlowStartSequence(t *testing.T) { + cases := []struct { + name string + cfg config + extra *http.OptsExtra + }{ + {"global off", config{Connections: 16}, nil}, + {"task off overrides global on", config{Connections: 16, Adaptive: true}, &http.OptsExtra{Adaptive: boolPtr(false)}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s := newAdaptiveTestServer(t, 64<<20, 4<<20, 0, 0, false) + f, path := newAdaptiveFetcher(t, tc.cfg, s, tc.extra) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 60*time.Second); err != nil { + t.Fatal(err) + } + if f.adaptive != nil { + t.Fatal("adaptive controller ran although adaptive is off") + } + f.slowStart.mu.Lock() + got := append([]int(nil), f.slowStart.batchHistory...) + f.slowStart.mu.Unlock() + if fmt.Sprint(got) != fmt.Sprint([]int{1, 2, 4, 9}) { + t.Fatalf("slow-start batches = %v, want [1 2 4 9]", got) + } + conns := f.Stats().Snapshot.(*http.Stats).Connections + if len(conns) != 16 { + t.Fatalf("got %d connections, want 16", len(conns)) + } + for i, c := range f.connections { + if c.ID != i { + t.Fatalf("connection %d has ID %d, want IDs 0..15 in order", i, c.ID) + } + } + assertAdaptiveFile(t, path, s.data) + }) + } +} + +// A record written before adaptive connections existed has no new fields. +// It must resume, with adaptive on and off. +func TestAdaptiveResumesPreChangeStoreBlob(t *testing.T) { + for _, adaptive := range []bool{false, true} { + t.Run(fmt.Sprintf("adaptive=%v", adaptive), func(t *testing.T) { + useFastAdaptiveClock(t) + const size = 8 << 20 + s := newAdaptiveTestServer(t, size, 8<<20, 0, 0, false) + dir := t.TempDir() + path := filepath.Join(dir, "adaptive.bin") + + // Four ranges, each a quarter downloaded, as an older version left them. + partial := make([]byte, size) + quarter := int64(size / 4) + var conns []string + for i := int64(0); i < 4; i++ { + begin, end, done := i*quarter, (i+1)*quarter-1, quarter/4 + copy(partial[begin:begin+done], s.data[begin:begin+done]) + role := 2 + if i == 0 { + role = 1 + } + conns = append(conns, fmt.Sprintf(`{"ID":%d,"Role":%d,"State":2,"Chunk":{"Begin":%d,"End":%d,"Downloaded":%d},"Downloaded":%d,"Completed":false}`, + i, role, begin, end, done, done)) + } + if err := os.WriteFile(path, partial, 0o644); err != nil { + t.Fatal(err) + } + blob := `{"Connections":[` + strings.Join(conns, ",") + `],"RedirectURL":"","IfRange":"","RangeReprobeEligible":false,` + + `"RangeValidatorPinned":false,"SequentialSizeUnknown":false,"Range":true,"ResourceSize":` + strconv.Itoa(size) + + `,"FileSize":` + strconv.Itoa(size) + `}` + rawMeta := `{"req":{"url":"` + s.url() + `","extra":{"method":"","header":null,"body":""}},` + + `"res":{"name":"","size":` + strconv.Itoa(size) + `,"range":true,"files":[{"name":"adaptive.bin","path":"","size":` + strconv.Itoa(size) + `}]},` + + `"opts":{"name":"adaptive.bin","path":"` + dir + `","selectFiles":null,"extra":{"connections":4,"autoTorrent":null}}}` + + var meta fetcher.FetcherMeta + if err := json.Unmarshal([]byte(rawMeta), &meta); err != nil { + t.Fatal(err) + } + fm := new(FetcherManager) + v, restore := fm.Restore() + if err := json.Unmarshal([]byte(blob), v); err != nil { + t.Fatal(err) + } + f := restore(&meta, v).(*Fetcher) + ctl := controller.NewController() + ctl.GetConfig = func(v any) { + raw, _ := json.Marshal(config{Connections: 16, Adaptive: adaptive}) + json.Unmarshal(raw, v) + } + f.Setup(ctl) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 30*time.Second); err != nil { + t.Fatal(err) + } + if (f.adaptive != nil) != adaptive { + t.Fatalf("adaptive controller used = %v, want %v", f.adaptive != nil, adaptive) + } + assertAdaptiveFile(t, path, s.data) + }) + } +} + +func TestAdaptivePatchConnectionsLowersCeiling(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) + if err := f.Start(); err != nil { + t.Fatal(err) + } + waitFor(t, 15*time.Second, "growth to 5 connections", func() bool { return s.active.Load() >= 5 }) + + if err := f.Patch(nil, &base.Options{Extra: &http.OptsExtra{Connections: 2}}); err != nil { + t.Fatal(err) + } + if got := f.meta.Opts.Extra.(*http.OptsExtra).Connections; got != 2 { + t.Fatalf("connections option = %d after Patch, want 2", got) + } + tl := sampleConnections(s) + waitFor(t, 5*time.Second, "the lower ceiling", func() bool { + recent := tl.snapshot() + return len(recent) >= 10 && settled(recent, len(recent)-10, len(recent)) <= 2 + }) + s.totalCap.Store(0) + if err := waitDone(t, f, 60*time.Second); err != nil { + t.Fatal(err) + } + samples := tl.finish() + for i, n := range samples[len(samples)/2:] { + if n > 2 { + t.Fatalf("sample %d shows %d connections after the ceiling became 2", i, n) + } + } + assertAdaptiveFile(t, path, s.data) +} + +func TestAdaptivePatchWithoutOptionsKeepsConnections(t *testing.T) { + f := buildConfigFetcher(config{Connections: 8}).(*Fetcher) + f.meta.Opts = &base.Options{Extra: &http.OptsExtra{Connections: 8}} + if err := f.Patch(nil, &base.Options{Extra: &http.OptsExtra{}}); err != nil { + t.Fatal(err) + } + if got := f.meta.Opts.Extra.(*http.OptsExtra).Connections; got != 8 { + t.Fatalf("connections = %d after an empty Patch, want 8", got) + } +} diff --git a/internal/protocol/http/adaptive_test.go b/internal/protocol/http/adaptive_test.go new file mode 100644 index 0000000..5a82923 --- /dev/null +++ b/internal/protocol/http/adaptive_test.go @@ -0,0 +1,277 @@ +package http + +import ( + "testing" + "time" +) + +const mb = int64(1000 * 1000) + +// simServer models a server for the pure controller tests: each connection +// gets perConn bytes/s and the total is capped at totalCap (0 means no cap). +type simServer struct { + perConn int64 + totalCap int64 + maxConns int // a new connection above this count is rejected (0 means no limit) +} + +func (s simServer) speed(active int) int64 { + speed := int64(active) * s.perConn + if s.totalCap > 0 && speed > s.totalCap { + speed = s.totalCap + } + return speed +} + +// simRun drives the controller with a simulated clock. It applies every +// decision to the connection count the way the engine does and records the +// count after each tick. +type simRun struct { + c *adaptiveController + now time.Time + active int + timeline []int + adds []time.Time + parks []time.Time +} + +func newSimRun(ceiling int) *simRun { + now := time.Unix(1_700_000_000, 0) + return &simRun{c: newAdaptiveController(ceiling, now), now: now, active: 1} +} + +func (r *simRun) step(s simServer) adaptiveDecision { + r.now = r.now.Add(adaptiveTickInterval) + d := r.c.tick(r.now, s.speed(r.active), r.active) + switch d { + case addConn: + r.adds = append(r.adds, r.now) + if s.maxConns > 0 && r.active+1 > s.maxConns { + // The new connection gets 403/429 and is parked by the engine. + r.c.limited(r.now, r.active, true) + } else { + r.active++ + } + case parkSlowest: + r.parks = append(r.parks, r.now) + r.active-- + } + r.timeline = append(r.timeline, r.active) + return d +} + +func (r *simRun) run(s simServer, d time.Duration) { + for end := r.now.Add(d); r.now.Before(end); { + r.step(s) + } +} + +func maxOf(xs []int) int { + m := 0 + for _, x := range xs { + if x > m { + m = x + } + } + return m +} + +func TestAdaptiveStartsAtOneAndGrowsWhileFaster(t *testing.T) { + r := newSimRun(16) + if r.c.tick(r.now.Add(adaptiveTickInterval), mb, 1) != holdConns { + t.Fatal("the first window must only measure the single starting connection") + } + r.now = r.now.Add(adaptiveTickInterval) + r.run(simServer{perConn: mb, totalCap: 6 * mb}, 3*time.Minute) + t.Logf("timeline: %v", r.timeline) + if got := r.active; got < 6 || got > 7 { + t.Fatalf("settled at %d connections, want 6 or 7", got) + } + if m := maxOf(r.timeline); m > 16 { + t.Fatalf("reached %d connections, above the ceiling of 16", m) + } + if m := maxOf(r.timeline); m > 7 { + t.Fatalf("reached %d connections, want at most one probe above 6", m) + } +} + +func TestAdaptivePlateauStopsGrowth(t *testing.T) { + r := newSimRun(16) + // The server gives 1 MB/s in total, however many connections ask. + flat := simServer{perConn: mb, totalCap: mb} + r.run(flat, 20*time.Second) + t.Logf("timeline: %v", r.timeline) + if len(r.adds) != 1 { + t.Fatalf("got %d additions in the first 20 s, want exactly 1 (the one that found the plateau)", len(r.adds)) + } + if len(r.parks) != 1 { + t.Fatalf("got %d parks, want the useless connection given back once", len(r.parks)) + } + if r.active != 1 { + t.Fatalf("active = %d after the plateau, want 1", r.active) + } +} + +func TestAdaptiveShrinksOnDrop(t *testing.T) { + r := newSimRun(16) + fast := simServer{perConn: mb, totalCap: 4 * mb} + r.run(fast, 30*time.Second) + if r.active != 4 { + t.Fatalf("setup: active = %d, want 4 (timeline %v)", r.active, r.timeline) + } + // The throughput drops by 20% for two windows, well before the next probe. + slow := simServer{perConn: mb, totalCap: 4 * mb * 80 / 100} + if d := r.step(slow); d != holdConns { + t.Fatalf("first low window gave %v, want hold", d) + } + if d := r.step(slow); d != parkSlowest { + t.Fatalf("second low window gave %v, want parkSlowest", d) + } +} + +func TestAdaptiveShrinkWalksDownWhileFree(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 6 * mb}, 30*time.Second) + if r.active != 6 { + t.Fatalf("setup: active = %d, want 6 (timeline %v)", r.active, r.timeline) + } + // The server now gives only 3 MB/s in total. Parking costs nothing down to + // three connections, and the park that costs speed is undone. + r.timeline = nil + r.run(simServer{perConn: mb, totalCap: 3 * mb}, 25*time.Second) + t.Logf("timeline after the slowdown: %v", r.timeline) + if r.active != 3 { + t.Fatalf("active = %d after the slowdown, want 3", r.active) + } +} + +func TestAdaptiveCeilingFromLimited(t *testing.T) { + r := newSimRun(16) + limit := simServer{perConn: mb, maxConns: 4} + r.run(limit, 5*time.Minute) + t.Logf("timeline: %v", r.timeline) + if m := maxOf(r.timeline); m > 4 { + t.Fatalf("reached %d connections, want at most 4", m) + } + if r.active != 4 { + t.Fatalf("active = %d, want 4", r.active) + } + // After the first rejection, the controller never asks above 4 again. + if len(r.adds) != 4 { + t.Fatalf("got %d additions, want 4 (three that worked and the rejected fifth)", len(r.adds)) + } +} + +func TestAdaptiveProbesEvery30s(t *testing.T) { + r := newSimRun(16) + capped := simServer{perConn: mb, totalCap: 3 * mb} + r.run(capped, 30*time.Second) + if r.active != 3 { + t.Fatalf("setup: active = %d, want 3 (timeline %v)", r.active, r.timeline) + } + r.adds, r.timeline = nil, nil + r.run(capped, 5*time.Minute) + t.Logf("timeline: %v", r.timeline) + // One probe per 30 s, each given back because it does not help. + if n := len(r.adds); n < 9 || n > 11 { + t.Fatalf("got %d probes in 5 minutes, want 10 +- 1", n) + } + for i := 1; i < len(r.adds); i++ { + if gap := r.adds[i].Sub(r.adds[i-1]); gap < 30*time.Second { + t.Fatalf("probes %v apart, want at least 30 s", gap) + } + } + if r.active != 3 { + t.Fatalf("active = %d after useless probes, want 3", r.active) + } + if m := maxOf(r.timeline); m > 4 { + t.Fatalf("reached %d connections, want at most one probe at a time", m) + } + + // When the server opens up, a probe helps, stays, and growth resumes. + r.run(simServer{perConn: mb, totalCap: 5 * mb}, 60*time.Second) + if r.active != 5 { + t.Fatalf("active = %d after the server opened up, want 5", r.active) + } +} + +func TestAdaptiveNeverAboveCeiling(t *testing.T) { + r := newSimRun(3) + r.run(simServer{perConn: mb}, 3*time.Minute) + if m := maxOf(r.timeline); m > 3 { + t.Fatalf("reached %d connections, above the ceiling of 3", m) + } + if r.active != 3 { + t.Fatalf("active = %d, want 3", r.active) + } +} + +func TestAdaptiveLowerCeilingParks(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 6 * mb}, 30*time.Second) + if r.active != 6 { + t.Fatalf("setup: active = %d, want 6", r.active) + } + r.c.setCeiling(4) + r.run(simServer{perConn: mb, totalCap: 6 * mb}, 10*time.Second) + if r.active != 4 { + t.Fatalf("active = %d after lowering the ceiling, want 4", r.active) + } +} + +func TestAdaptiveFinishingConnectionsAreNotASlowdown(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 4 * mb}, 30*time.Second) + if r.active != 4 { + t.Fatalf("setup: active = %d, want 4", r.active) + } + // Two connections run out of work: the engine reports two fewer, and the + // task speed halves. That is not a reason to park anything. + r.active = 2 + r.parks = nil + for i := 0; i < 4; i++ { + r.step(simServer{perConn: mb}) + } + if len(r.parks) != 0 { + t.Fatalf("parked %d connections after others finished, want 0", len(r.parks)) + } +} + +func TestAdaptiveMidStreamLimitParksAndHoldsOff(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 4 * mb}, 30*time.Second) + if r.active != 4 { + t.Fatalf("setup: active = %d, want 4", r.active) + } + // One connection gets 429 on a later request; the engine parks it. + r.active = 3 + r.c.limited(r.now, 3, false) + r.adds = nil + r.run(simServer{perConn: mb, totalCap: 4 * mb}, 25*time.Second) + if len(r.adds) != 0 { + t.Fatalf("got %d additions within 25 s of a 429, want none before the probe", len(r.adds)) + } +} + +// A slowdown that starts while a probe is being judged must still shrink: +// giving the probe back must not make the slower speed the new best. +func TestAdaptiveSlowdownDuringProbeStillShrinks(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 6 * mb}, 30*time.Second) + if r.active != 6 { + t.Fatalf("setup: active = %d, want 6", r.active) + } + // Run until the next probe adds a seventh connection. + for i := 0; i < 100 && r.active != 7; i++ { + r.step(simServer{perConn: mb, totalCap: 6 * mb}) + } + if r.active != 7 { + t.Fatal("setup: no probe within 200 s") + } + r.timeline = nil + r.run(simServer{perConn: mb, totalCap: 2 * mb}, 30*time.Second) + t.Logf("timeline after the slowdown: %v", r.timeline) + if r.active > 3 { + t.Fatalf("active = %d after the server slowed to 2 MB/s, want connections given back (<= 3)", r.active) + } +} diff --git a/internal/protocol/http/config.go b/internal/protocol/http/config.go index 7a09e59..e3db6b1 100644 --- a/internal/protocol/http/config.go +++ b/internal/protocol/http/config.go @@ -7,4 +7,8 @@ type config struct { UserAgent string `json:"userAgent"` Connections int `json:"connections"` UseServerCtime bool `json:"useServerCtime"` + // Adaptive is the default for tasks that do not set OptsExtra.Adaptive: + // grow and shrink the connection count on measured throughput, with + // Connections as the ceiling. Off unless the user turns it on. + Adaptive bool `json:"adaptive"` } diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 9eca008..70bcae5 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -77,6 +77,10 @@ const ( connDownloading // HTTP response OK, downloading connCompleted // Completed connFailed // Failed + // connParked: stopped by adaptive connections at its current offset. Its + // remaining range is free for other connections to take; it is not a + // failure. Appended last because State is persisted as a number. + connParked ) type connectionRole int @@ -131,6 +135,11 @@ type connection struct { retryTimes int lastErr error + // Adaptive connections (runtime only, never persisted) + parked bool // asked to stop; its range is stealable once it has exited + running bool // a goroutine started by adaptiveLaunchLocked is live + gotData bool // got a 2xx response since it was last launched + // Speed tracking for work stealing decisions speed int64 // bytes per second lastSpeedCheck int64 // timestamp in nanoseconds @@ -153,6 +162,7 @@ type slowStartController struct { nextBatchSize int // Next batch size: 1, 2, 4, 16, 256... (capped) expansionCh chan struct{} // Signal to trigger next expansion paused bool // Pause expansion (e.g., on 429) + batchHistory []int // Committed batch sizes, in order } func newSlowStartController(maxConnections int) *slowStartController { @@ -229,6 +239,7 @@ func (s *slowStartController) commitBatch(count int) { defer s.mu.Unlock() s.totalLaunched += count + s.batchHistory = append(s.batchHistory, count) if s.nextBatchSize == 1 { // Bootstrap growth: squaring one would never increase the batch size. s.nextBatchSize = min(2, s.maxConnections) @@ -273,6 +284,17 @@ type Fetcher struct { // Slow start controller slowStart *slowStartController + // Adaptive controller; nil unless adaptive connections are on for this run. + // It is set by doStart before any connection starts and is not changed + // until the next doStart, after Pause has stopped every connection. + adaptive *adaptiveController + adaptiveMu sync.Mutex // serialises calls into adaptive + adaptiveWake chan struct{} // a connection goroutine exited + + // nextConnID is the next connection ID. IDs only grow, so an ID is never + // reused even if connections are removed. connMu guards it. + nextConnID int + // Max connection time for adaptive timeout (stored as int64 nanoseconds for atomic ops) maxConnTime atomic.Int64 @@ -808,6 +830,25 @@ func (f *Fetcher) doStart() error { maxConns := f.meta.Opts.Extra.(*fhttp.OptsExtra).Connections f.slowStart = newSlowStartController(maxConns) + // Adaptive connections need byte ranges and a known size; anything else + // keeps the slow-start path below unchanged. + var adaptive *adaptiveController + if f.adaptiveEnabled() && f.meta.Res.Range && f.meta.Res.Size > 0 { + adaptive = newAdaptiveController(maxConns, time.Now()) + f.adaptiveWake = make(chan struct{}, 1) + } else { + // Adaptive was switched off since the last run: parked connections + // resume like any other unfinished connection. + f.connMu.Lock() + for _, conn := range f.connections { + conn.parked = false + } + f.connMu.Unlock() + } + f.adaptiveMu.Lock() + f.adaptive = adaptive + f.adaptiveMu.Unlock() + // Create main context f.ctx, f.cancel = context.WithCancel(context.Background()) @@ -840,6 +881,15 @@ func (f *Fetcher) downloadLoop(done chan struct{}) { // Check if this is a resume or fresh start isResume := len(f.connections) > 0 + if f.adaptive != nil { + if !f.adaptiveLoop(ctx, isResume) { + return + } + // The origin dropped byte ranges after all; continue as today. + f.slowStartLoop(ctx) + return + } + if !isResume { // Capture the mode before launching the first connection. The server may // ignore that connection's Range request and switch the fetcher to a @@ -868,7 +918,10 @@ func (f *Fetcher) downloadLoop(done chan struct{}) { } } - // Slow start loop + f.slowStartLoop(ctx) +} + +func (f *Fetcher) slowStartLoop(ctx context.Context) { for { select { case <-ctx.Done(): @@ -954,43 +1007,12 @@ func (f *Fetcher) expandConnections() { // For first batch (no existing connections), allocate the remaining file to first connection if len(f.connections) == 0 { - // Check if we have prefetched data - prefetched := f.resolveDataPos.Load() - - // If prefetched all data, mark as done. A zero size means the extent is - // still unknown, so the first worker must discover it with bytes=start-. - if totalSize > 0 && prefetched >= totalSize { + conn := f.newFirstConnectionLocked(totalSize) + if conn == nil { f.connMu.Unlock() - - // Close the file before signaling completion - f.fileMu.Lock() - if f.file != nil { - f.file.Close() - f.file = nil - } - f.fileMu.Unlock() - - f.setState(stateDone) - f.doneCh <- nil + f.finishFromPrefetch() return } - - // First connection starts from prefetched position - connChunk := newOpenEndedChunk(prefetched) - if totalSize > 0 { - connChunk = newChunk(prefetched, totalSize-1) - } - conn := &connection{ - ID: 0, - Role: rolePrimary, - State: connNotStarted, - Chunk: connChunk, - } - // Mark prefetched bytes as already downloaded - conn.Chunk.Downloaded = 0 // Start fresh from prefetched position - conn.Downloaded = prefetched // Track total downloaded including prefetch - - conn.ctx, conn.cancel = context.WithCancel(f.ctx) f.connections = append(f.connections, conn) f.connMu.Unlock() @@ -1001,53 +1023,14 @@ func (f *Fetcher) expandConnections() { } // For subsequent batches, use "help other connection" strategy - // Find connections with enough remaining work to split - // During slow start, use fixed minimum size since speed is not yet stable - minSplitSize := int64(stealMinChunkSize) - newConns := make([]*connection, 0, batchSize) for i := 0; i < batchSize; i++ { - // Find the connection with most remaining work - var maxRemainConn *connection - var maxRemain int64 - - for _, conn := range f.connections { - if conn.Completed || conn.State == connFailed { - continue - } - if conn.Chunk.openEnded() { - continue - } - remain := conn.Chunk.remain() - // Only split if remaining work is at least 2x the minimum split size - if remain > maxRemain && remain > minSplitSize*2 { - maxRemainConn = conn - maxRemain = remain - } - } - - if maxRemainConn == nil { + conn := f.splitLargestLocked() + if conn == nil { // No connection has enough work to split break } - - // Split the work: new connection takes the latter half - chunkEnd := *maxRemainConn.Chunk.End - splitPoint := chunkEnd - maxRemainConn.Chunk.remain()/2 - newChunk := newChunk(splitPoint+1, chunkEnd) - maxRemainConn.Chunk.setEnd(splitPoint) - - connID := len(f.connections) - conn := &connection{ - ID: connID, - Role: roleWorker, - State: connNotStarted, - Chunk: newChunk, - } - conn.ctx, conn.cancel = context.WithCancel(f.ctx) - newConns = append(newConns, conn) - f.connections = append(f.connections, conn) } f.connMu.Unlock() @@ -1068,6 +1051,117 @@ func (f *Fetcher) expandConnections() { } } +// newFirstConnectionLocked allocates the rest of the file after the resolve +// prefetch to connection 0. It returns nil when the prefetch already holds the +// whole file; the caller then calls finishFromPrefetch after unlocking. +// The caller must hold connMu. +func (f *Fetcher) newFirstConnectionLocked(totalSize int64) *connection { + // Check if we have prefetched data + prefetched := f.resolveDataPos.Load() + + // If prefetched all data, mark as done. A zero size means the extent is + // still unknown, so the first worker must discover it with bytes=start-. + if totalSize > 0 && prefetched >= totalSize { + return nil + } + + // First connection starts from prefetched position + connChunk := newOpenEndedChunk(prefetched) + if totalSize > 0 { + connChunk = newChunk(prefetched, totalSize-1) + } + conn := &connection{ + ID: f.allocConnIDLocked(), + Role: rolePrimary, + State: connNotStarted, + Chunk: connChunk, + } + // Mark prefetched bytes as already downloaded + conn.Chunk.Downloaded = 0 // Start fresh from prefetched position + conn.Downloaded = prefetched // Track total downloaded including prefetch + + conn.ctx, conn.cancel = context.WithCancel(f.ctx) + return conn +} + +// finishFromPrefetch completes a task whose resolve prefetch already holds +// the whole file. The caller must not hold connMu. +func (f *Fetcher) finishFromPrefetch() { + // Close the file before signaling completion + f.fileMu.Lock() + if f.file != nil { + f.file.Close() + f.file = nil + } + f.fileMu.Unlock() + + f.setState(stateDone) + f.doneCh <- nil +} + +// splitLargestLocked gives the latter half of the largest splittable range to +// a new connection, appends it, and returns it without starting it. It returns +// nil when no range is big enough. The caller must hold connMu. +func (f *Fetcher) splitLargestLocked() *connection { + // Find connections with enough remaining work to split + // During slow start, use fixed minimum size since speed is not yet stable + minSplitSize := int64(stealMinChunkSize) + + // Find the connection with most remaining work + var maxRemainConn *connection + var maxRemain int64 + + for _, conn := range f.connections { + if conn.Completed || conn.State == connFailed || conn.parked { + continue + } + if conn.Chunk.openEnded() { + continue + } + remain := conn.Chunk.remain() + // Only split if remaining work is at least 2x the minimum split size + if remain > maxRemain && remain > minSplitSize*2 { + maxRemainConn = conn + maxRemain = remain + } + } + + if maxRemainConn == nil { + return nil + } + + // Split the work: new connection takes the latter half + chunkEnd := *maxRemainConn.Chunk.End + splitPoint := chunkEnd - maxRemainConn.Chunk.remain()/2 + newChunk := newChunk(splitPoint+1, chunkEnd) + maxRemainConn.Chunk.setEnd(splitPoint) + + conn := &connection{ + ID: f.allocConnIDLocked(), + Role: roleWorker, + State: connNotStarted, + Chunk: newChunk, + } + conn.ctx, conn.cancel = context.WithCancel(f.ctx) + f.connections = append(f.connections, conn) + return conn +} + +// allocConnIDLocked returns a connection ID that no connection has had. IDs +// grow from 0; restored records continue after their highest ID. While no +// connection is ever removed this equals len(f.connections), today's rule. +// The caller must hold connMu. +func (f *Fetcher) allocConnIDLocked() int { + next := f.nextConnID + for _, conn := range f.connections { + if conn != nil && conn.ID >= next { + next = conn.ID + 1 + } + } + f.nextConnID = next + 1 + return next +} + func (f *Fetcher) runConnection(conn *connection) { defer func() { // A range-advertising origin can fall back to one sequential response. @@ -1138,6 +1232,9 @@ func (f *Fetcher) runConnection(conn *connection) { if shouldUseResolveFallback && f.completeFromResolveFallback(conn) { return } + if requestErr != nil && f.adaptive != nil && f.adaptiveLimited(conn, requestErr.Code) { + return + } if isTerminalRangeError(err) { f.connMu.Lock() conn.lastErr = err @@ -1379,6 +1476,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ f.connMu.Lock() conn.State = connDownloading conn.failed = false + conn.gotData = true f.connMu.Unlock() if conn.Role == rolePrimary || conn.ID == 0 { @@ -2013,6 +2111,11 @@ func (f *Fetcher) helpOtherConnection(helper *connection) bool { f.connMu.Lock() defer f.connMu.Unlock() + // A parked range has nobody downloading it, so it goes first and whole. + if f.adaptive != nil && f.takeParkedRangeLocked(helper) { + return true + } + // Find the connection with longest remaining time var slowestConn *connection var maxRemainSeconds int64 @@ -2339,6 +2442,37 @@ func (f *Fetcher) Patch(req *base.Request, opts *base.Options) error { } } + // Patch options: a positive connections value and an explicit adaptive + // switch. Zero values mean "unchanged". + if opts != nil && opts.Extra != nil { + if err := base.ParseOptExtra[fhttp.OptsExtra](opts); err != nil { + return err + } + patchExtra := opts.Extra.(*fhttp.OptsExtra) + if f.meta.Opts == nil { + f.meta.Opts = &base.Options{} + } + if f.meta.Opts.Extra == nil { + f.meta.Opts.Extra = &fhttp.OptsExtra{} + } + existingExtra := f.meta.Opts.Extra.(*fhttp.OptsExtra) + if patchExtra.Connections > 0 { + existingExtra.Connections = patchExtra.Connections + // A running adaptive task takes the new ceiling at its next window; + // otherwise the value applies from the next start. + f.adaptiveMu.Lock() + if f.adaptive != nil { + f.adaptive.setCeiling(patchExtra.Connections) + } + f.adaptiveMu.Unlock() + } + if patchExtra.Adaptive != nil { + // Applies from the next start. + value := *patchExtra.Adaptive + existingExtra.Adaptive = &value + } + } + return nil } @@ -2469,6 +2603,7 @@ func (f *Fetcher) Stats() *fetcher.Stats { Completed: completed, Failed: connection.failed, RetryTimes: connection.retryTimes, + Parked: connection.parked && !completed, }) } if len(statsConnections) == 0 && prefetched > 0 { @@ -2531,3 +2666,339 @@ func (f *Fetcher) responseReader(resp *http.Response) io.Reader { } return NewTimeoutReader(resp.Body, readTimeout) } + +// ============================================================================ +// Adaptive connections +// ============================================================================ +// +// With adaptive connections on, downloadLoop runs adaptiveLoop instead of +// slow start. It starts one connection, samples the task speed every +// adaptiveTickInterval, and carries out the adaptiveController's decisions: +// add a connection (revive a parked range, else split the largest range) or +// park the slowest one. A parked connection stops at its current offset; once +// its goroutine has exited, its remaining range is taken whole by the next +// connection that runs out of work (helpOtherConnection), revived by the next +// addition, or revived when nothing else runs. + +// adaptiveEnabled reports whether this task uses adaptive connections: the +// task's own switch when set, else the global default. +func (f *Fetcher) adaptiveEnabled() bool { + if f.meta != nil && f.meta.Opts != nil { + if extra, ok := f.meta.Opts.Extra.(*fhttp.OptsExtra); ok && extra.Adaptive != nil { + return *extra.Adaptive + } + } + return f.config != nil && f.config.Adaptive +} + +// adaptiveLoop drives one Start cycle. It returns true only when the origin +// stopped honouring byte ranges and the slow-start loop must take over. +func (f *Fetcher) adaptiveLoop(ctx context.Context, isResume bool) bool { + if isResume { + f.adaptiveResume() + } else if !f.adaptiveStartFirst() { + return false + } + + ticker := time.NewTicker(adaptiveTickInterval) + defer ticker.Stop() + + measuring := false + var lastBytes int64 + var lastAt time.Time + + for { + var now time.Time + select { + case <-ctx.Done(): + // Paused or cancelled + return false + case <-f.adaptiveWake: + case now = <-ticker.C: + } + + f.connMu.Lock() + rangeMode := f.meta.Res.Range + f.connMu.Unlock() + if !rangeMode { + return true + } + if f.adaptiveIdle() { + f.waitForCompletion(ctx) + return false + } + if now.IsZero() { + continue // a wake-up only checks for completion + } + + active, ready, total := f.adaptiveSnapshot() + if !measuring { + // Measure only once a range response has been validated and the + // resolve prefetch has been handed over; the takeover adds its + // prefetched bytes at once and would read as a burst of speed. + if ready && !f.resolveFallback.Load() { + measuring = true + lastBytes, lastAt = total, now + } + continue + } + elapsed := now.Sub(lastAt).Seconds() + if elapsed <= 0 { + continue + } + speed := int64(float64(total-lastBytes) / elapsed) + lastBytes, lastAt = total, now + + f.adaptiveMu.Lock() + decision := f.adaptive.tick(now, speed, active) + f.adaptiveMu.Unlock() + + switch decision { + case addConn: + f.connMu.Lock() + f.adaptiveAddLocked() + f.connMu.Unlock() + case parkSlowest: + f.connMu.Lock() + f.adaptiveParkSlowestLocked() + f.connMu.Unlock() + } + } +} + +// adaptiveStartFirst starts connection 0 on a fresh start. It returns false +// when the resolve prefetch already completed the task. +func (f *Fetcher) adaptiveStartFirst() bool { + f.connMu.Lock() + conn := f.newFirstConnectionLocked(f.meta.Res.Size) + if conn == nil { + f.connMu.Unlock() + f.finishFromPrefetch() + return false + } + f.connections = append(f.connections, conn) + // Keep slow start's books as if it had launched this connection, so the + // slow-start loop can take over if the origin turns out to ignore ranges. + f.slowStart.commitBatch(1) + f.adaptiveLaunchLocked(conn) + f.connMu.Unlock() + return true +} + +// adaptiveResume parks every unfinished connection and revives the one with +// the most work left, so growth starts again from one connection. +func (f *Fetcher) adaptiveResume() { + f.connMu.Lock() + defer f.connMu.Unlock() + for _, conn := range f.connections { + if conn.Completed || conn.State == connCompleted { + continue + } + // Same permanent failures that resumeConnections leaves alone. + if conn.State == connFailed && conn.failed { + if re := extractRequestError(conn.lastErr); re != nil && re.Code == 403 { + continue + } + if conn.retryTimes >= 3 { + continue + } + } + conn.parked = true + conn.running = false + conn.State = connParked + conn.failed = false + } + if conn := f.largestParkedLocked(1); conn != nil { + f.reviveLocked(conn) + } +} + +// adaptiveLaunchLocked starts conn's goroutine and tracks it as running. The +// wrapper's bookkeeping is part of the WaitGroup, so Pause returns only after +// running is cleared. The caller must hold connMu. +func (f *Fetcher) adaptiveLaunchLocked(conn *connection) { + conn.running = true + conn.gotData = false + f.wg.Add(2) // one for runConnection, one for this wrapper + go func() { + defer f.wg.Done() + f.runConnection(conn) + + f.connMu.Lock() + conn.running = false + if conn.Completed || conn.State == connCompleted { + conn.parked = false + } + if conn.parked { + // Parking is not a failure: drop whatever the cancelled request + // recorded on its way out. + conn.State = connParked + conn.failed = false + conn.lastErr = nil + } + f.connMu.Unlock() + + select { + case f.adaptiveWake <- struct{}{}: + default: + } + }() +} + +// reviveLocked restarts a parked connection where it stopped. The caller must +// hold connMu, and conn must not be running. +func (f *Fetcher) reviveLocked(conn *connection) { + if conn.cancel != nil { + conn.cancel() + } + conn.ctx, conn.cancel = context.WithCancel(f.ctx) + conn.parked = false + conn.State = connNotStarted + conn.failed = false + conn.lastErr = nil + conn.retryTimes = 0 + conn.speed = 0 + conn.lastSpeedCheck = 0 + conn.lastSpeedDownload = 0 + f.adaptiveLaunchLocked(conn) +} + +// largestParkedLocked returns the stopped parked connection with the most +// work left, if that is at least minRemain bytes. The caller must hold connMu. +func (f *Fetcher) largestParkedLocked(minRemain int64) *connection { + var best *connection + var bestRemain int64 + for _, conn := range f.connections { + if !conn.parked || conn.running || conn.Completed || conn.Chunk == nil || conn.Chunk.openEnded() { + continue + } + if remain := conn.Chunk.remain(); remain >= minRemain && remain > bestRemain { + best, bestRemain = conn, remain + } + } + return best +} + +// adaptiveAddLocked carries out addConn. It returns false when there is no +// work to give a new connection; the controller notices that the count did +// not rise. The caller must hold connMu. +func (f *Fetcher) adaptiveAddLocked() bool { + if conn := f.largestParkedLocked(stealMinChunkSize); conn != nil { + f.reviveLocked(conn) + return true + } + conn := f.splitLargestLocked() + if conn == nil { + return false + } + f.adaptiveLaunchLocked(conn) + return true +} + +// adaptiveParkSlowestLocked carries out parkSlowest. It never parks the last +// running connection. The caller must hold connMu. +func (f *Fetcher) adaptiveParkSlowestLocked() bool { + var victim *connection + active := 0 + for _, conn := range f.connections { + if !conn.running || conn.parked { + continue + } + active++ + // The slowest; on a tie the newest, which is the one just added. + if victim == nil || conn.speed < victim.speed || (conn.speed == victim.speed && conn.ID > victim.ID) { + victim = conn + } + } + if active <= 1 || victim == nil { + return false + } + victim.parked = true + if victim.cancel != nil { + victim.cancel() + } + return true +} + +// takeParkedRangeLocked gives helper the whole remaining range of the stopped +// parked connection with the most work left. Unlike stealing from a running +// connection there is no minimum size: nobody else would fetch it. The caller +// must hold connMu. +func (f *Fetcher) takeParkedRangeLocked(helper *connection) bool { + victim := f.largestParkedLocked(1) + if victim == nil || victim == helper { + return false + } + pos := victim.Chunk.Begin + victim.Chunk.Downloaded + helper.Chunk.Begin = pos + helper.Chunk.setEnd(*victim.Chunk.End) + helper.Chunk.Downloaded = 0 + victim.Chunk.setEnd(pos - 1) + victim.parked = false + victim.Completed = true + victim.State = connCompleted + return true +} + +// adaptiveIdle reports whether the download is finished: nothing runs and +// no parked range is left. If nothing runs but a parked range is, it revives +// that connection and reports false. +func (f *Fetcher) adaptiveIdle() bool { + f.connMu.Lock() + defer f.connMu.Unlock() + for _, conn := range f.connections { + if conn.running { + return false + } + } + if conn := f.largestParkedLocked(1); conn != nil { + f.reviveLocked(conn) + return false + } + return true +} + +// adaptiveSnapshot returns the running, unparked connection count, whether +// one of them has a validated response, and the bytes downloaded so far. +func (f *Fetcher) adaptiveSnapshot() (active int, ready bool, total int64) { + f.connMu.Lock() + defer f.connMu.Unlock() + for _, conn := range f.connections { + total += conn.Downloaded + if conn.running && !conn.parked { + active++ + if conn.gotData { + ready = true + } + } + } + return active, ready, total +} + +// adaptiveLimited handles a limiting status on conn with adaptive on. A 403 +// or 429 before any data means the server allows no more connections; a 429 +// or 503 after data means it wants fewer. Either way, while another +// connection still runs, conn is parked (not failed) and the controller is +// told. It returns true when conn was parked and must stop. +func (f *Fetcher) adaptiveLimited(conn *connection, code int) bool { + f.connMu.Lock() + newConn := !conn.gotData + limiting := (newConn && (code == 403 || code == 429)) || (!newConn && (code == 429 || code == 503)) + others := 0 + for _, c := range f.connections { + if c != conn && c.running && !c.parked { + others++ + } + } + if !limiting || others == 0 { + f.connMu.Unlock() + return false + } + conn.parked = true + f.connMu.Unlock() + + f.adaptiveMu.Lock() + f.adaptive.limited(time.Now(), others, newConn) + f.adaptiveMu.Unlock() + return true +} diff --git a/pkg/protocol/http/model.go b/pkg/protocol/http/model.go index 9e96ffe..6a9243c 100644 --- a/pkg/protocol/http/model.go +++ b/pkg/protocol/http/model.go @@ -8,6 +8,10 @@ type ReqExtra struct { type OptsExtra struct { Connections int `json:"connections"` + // Adaptive grows and shrinks the connection count on measured throughput, + // with Connections as the ceiling. + // nil means use global config, true/false means explicit setting + Adaptive *bool `json:"adaptive,omitempty"` // AutoTorrent when task download complete, and it is a .torrent file, it will be auto create a new task for the torrent file // nil means use global config, true/false means explicit setting AutoTorrent *bool `json:"autoTorrent"` @@ -38,4 +42,8 @@ type StatsConnection struct { Completed bool `json:"completed"` Failed bool `json:"failed"` RetryTimes int `json:"retryTimes"` + // Parked is true for a connection that adaptive connections stopped + // because more connections did not make the task faster. It is not a + // failure; its remaining range is finished by the other connections. + Parked bool `json:"parked,omitempty"` } diff --git a/ui/flutter/lib/api/model/downloader_config.dart b/ui/flutter/lib/api/model/downloader_config.dart index dae6a10..06a962d 100644 --- a/ui/flutter/lib/api/model/downloader_config.dart +++ b/ui/flutter/lib/api/model/downloader_config.dart @@ -90,8 +90,9 @@ class HttpConfig { String userAgent; int connections; bool useServerCtime; + bool adaptive; - HttpConfig({this.userAgent = '', this.connections = 0, this.useServerCtime = false}); + HttpConfig({this.userAgent = '', this.connections = 0, this.useServerCtime = false, this.adaptive = false}); factory HttpConfig.fromJson(Map json) => _$HttpConfigFromJson(json); diff --git a/ui/flutter/lib/api/model/downloader_config.g.dart b/ui/flutter/lib/api/model/downloader_config.g.dart index 2ee3c2d..c354d78 100644 --- a/ui/flutter/lib/api/model/downloader_config.g.dart +++ b/ui/flutter/lib/api/model/downloader_config.g.dart @@ -84,12 +84,14 @@ HttpConfig _$HttpConfigFromJson(Map json) => HttpConfig( userAgent: json['userAgent'] as String? ?? '', connections: (json['connections'] as num?)?.toInt() ?? 0, useServerCtime: json['useServerCtime'] as bool? ?? false, + adaptive: json['adaptive'] as bool? ?? false, ); Map _$HttpConfigToJson(HttpConfig instance) => { 'userAgent': instance.userAgent, 'connections': instance.connections, 'useServerCtime': instance.useServerCtime, + 'adaptive': instance.adaptive, }; BtConfig _$BtConfigFromJson(Map json) => BtConfig( diff --git a/ui/flutter/lib/api/model/options.dart b/ui/flutter/lib/api/model/options.dart index 0c73073..a48140f 100644 --- a/ui/flutter/lib/api/model/options.dart +++ b/ui/flutter/lib/api/model/options.dart @@ -20,6 +20,7 @@ class Options { @JsonSerializable() class OptsExtraHttp { int connections; + bool? adaptive; bool? autoTorrent; bool? deleteTorrentAfterDownload; bool? autoExtract; @@ -28,6 +29,7 @@ class OptsExtraHttp { OptsExtraHttp({ this.connections = 0, + this.adaptive, this.autoTorrent, this.deleteTorrentAfterDownload, this.autoExtract, diff --git a/ui/flutter/lib/api/model/options.g.dart b/ui/flutter/lib/api/model/options.g.dart index 6e658bf..9bb7375 100644 --- a/ui/flutter/lib/api/model/options.g.dart +++ b/ui/flutter/lib/api/model/options.g.dart @@ -24,6 +24,7 @@ Map _$OptionsToJson(Options instance) => { OptsExtraHttp _$OptsExtraHttpFromJson(Map json) => OptsExtraHttp( connections: (json['connections'] as num?)?.toInt() ?? 0, + adaptive: json['adaptive'] as bool?, autoTorrent: json['autoTorrent'] as bool?, deleteTorrentAfterDownload: json['deleteTorrentAfterDownload'] as bool?, autoExtract: json['autoExtract'] as bool?, @@ -33,6 +34,7 @@ OptsExtraHttp _$OptsExtraHttpFromJson(Map json) => OptsExtraHtt Map _$OptsExtraHttpToJson(OptsExtraHttp instance) => { 'connections': instance.connections, + 'adaptive': ?instance.adaptive, 'autoTorrent': ?instance.autoTorrent, 'deleteTorrentAfterDownload': ?instance.deleteTorrentAfterDownload, 'autoExtract': ?instance.autoExtract, diff --git a/ui/flutter/lib/features/settings/presentation/pages/settings_page.dart b/ui/flutter/lib/features/settings/presentation/pages/settings_page.dart index c78a18a..95cdbd1 100644 --- a/ui/flutter/lib/features/settings/presentation/pages/settings_page.dart +++ b/ui/flutter/lib/features/settings/presentation/pages/settings_page.dart @@ -551,6 +551,15 @@ class _SettingsPageState extends ConsumerState { max: 256, ), ), + SettingsItem( + title: context.l10n.adaptiveConnections, + subtitle: context.l10n.adaptiveConnectionsDescription, + child: shad.Switch( + key: const ValueKey('http-adaptive-switch'), + value: config.protocolConfig.http.adaptive, + onChanged: (value) => _mutateConfig((next) => next.protocolConfig.http.adaptive = value), + ), + ), SettingsItem( title: context.l10n.useServerCtime, child: shad.Switch( diff --git a/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart b/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart index 921cfe8..554b07f 100644 --- a/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart +++ b/ui/flutter/lib/features/tasks/presentation/pages/create_task_window_page.dart @@ -75,6 +75,9 @@ class _CreateTaskWindowPageState extends ConsumerState { RequestProxyMode _proxyMode = RequestProxyMode.follow; String _proxyScheme = 'http'; bool _skipVerifyCert = false; + // null follows the global setting (_globalAdaptive) until the user toggles it + bool? _adaptive; + bool _globalAdaptive = false; bool? _autoTorrent; bool? _deleteTorrentAfterDownload; bool? _autoExtract; @@ -262,6 +265,7 @@ class _CreateTaskWindowPageState extends ConsumerState { if (extra.connections > 0) { _connectionsController.text = extra.connections.toString(); } + _adaptive = extra.adaptive; _autoTorrent = extra.autoTorrent; _deleteTorrentAfterDownload = extra.deleteTorrentAfterDownload; _autoExtract = extra.autoExtract; @@ -654,6 +658,16 @@ class _CreateTaskWindowPageState extends ConsumerState { ), ), const SizedBox(height: 14), + AppFormRow( + direction: stacked ? Axis.vertical : Axis.horizontal, + label: context.l10n.adaptiveConnections, + child: _OptionSwitch( + key: const ValueKey('create-task-adaptive'), + value: _adaptive ?? _globalAdaptive, + onChanged: (value) => setState(() => _adaptive = value), + ), + ), + const SizedBox(height: 14), AppFormRow( direction: stacked ? Axis.vertical : Axis.horizontal, label: context.l10n.autoTorrentEnable, @@ -806,6 +820,7 @@ class _CreateTaskWindowPageState extends ConsumerState { if (_asDefaultPath && _sameDirectory(_directoryController.text, _configuredDownloadDirectory)) { _asDefaultPath = false; } + _globalAdaptive = config.protocolConfig.http.adaptive; final connections = config.protocolConfig.http.connections; if (connections > 0 && (_connectionsController.text.isEmpty || _connectionsController.text == '16')) { _connectionsController.text = connections.toString(); @@ -1091,6 +1106,7 @@ class _CreateTaskWindowPageState extends ConsumerState { } else { extra = api_options.OptsExtraHttp( connections: connections, + adaptive: _adaptive, autoTorrent: _autoTorrent, deleteTorrentAfterDownload: _deleteTorrentAfterDownload, autoExtract: _autoExtract, diff --git a/ui/flutter/lib/l10n/app_en.arb b/ui/flutter/lib/l10n/app_en.arb index 11ce25e..ff810a9 100644 --- a/ui/flutter/lib/l10n/app_en.arb +++ b/ui/flutter/lib/l10n/app_en.arb @@ -42,6 +42,8 @@ "ftpTlsNone": "None", "ftpTlsExplicit": "Explicit TLS (FTPES)", "ftpTlsImplicit": "Implicit TLS (FTPS)", + "adaptiveConnections": "Adaptive connections", + "adaptiveConnectionsDescription": "Starts with one connection and adds more while the speed keeps rising, up to the Connections value. It closes connections that do not help.", "maxRetries": "Max retries", "useServerCtime": "Use server time for file creation", "maxRunning": "Concurrent Downloads", -- 2.55.0 From bba6119609c015e313aba599fccc74fd997da315 Mon Sep 17 00:00:00 2001 From: "Silas, Mariusz Grzybacz" Date: Fri, 2 Oct 2026 22:31:52 +0000 Subject: [PATCH 2/5] fix(http): ride out one 429 after a range split in adaptive mode When a connection's range is split, it stops reading its old response early and its follow-up request can reach the server before the server notices the old one is gone. A server that limits concurrent requests then answers 429 once. Adaptive mode parked the connection at once and held probes off for a full interval, one connection below the limit. - A 429/503 after data parks only when it repeats; the first goes through the normal retry. - A connection waiting to retry is not counted as active, so the controller re-measures instead of reading the pause as a slowdown. - parkSlowest treats a connection that is not downloading as speed 0. With a 4-connection limit the test server now sees 4 connections for 88% of the steady part, up from about 50%; the test asserts at least 70%. Also add tests for the handover to slow start when the origin ignores Range, and for the per-task switch turning adaptive on. --- .../http/adaptive_integration_test.go | 66 ++++++++++++++++++- internal/protocol/http/fetcher.go | 37 +++++++++-- 2 files changed, 95 insertions(+), 8 deletions(-) diff --git a/internal/protocol/http/adaptive_integration_test.go b/internal/protocol/http/adaptive_integration_test.go index 28cfd08..0be8f86 100644 --- a/internal/protocol/http/adaptive_integration_test.go +++ b/internal/protocol/http/adaptive_integration_test.go @@ -34,6 +34,9 @@ type adaptiveTestServer struct { totalCap atomic.Int64 // bytes/s for all requests together, 0 means no cap maxConns atomic.Int32 // concurrent requests above this get 429, 0 means no limit noRange bool + // ignoreRange advertises byte ranges but answers every request with the + // whole file and 200. + ignoreRange bool active atomic.Int32 rejected atomic.Int32 @@ -96,7 +99,7 @@ func (s *adaptiveTestServer) serve(w gohttp.ResponseWriter, r *gohttp.Request) { status := gohttp.StatusOK if !s.noRange { w.Header().Set("Accept-Ranges", "bytes") - if spec := r.Header.Get("Range"); spec != "" { + if spec := r.Header.Get("Range"); spec != "" && !s.ignoreRange { bounds := strings.SplitN(strings.TrimPrefix(spec, "bytes="), "-", 2) start, _ = strconv.ParseInt(bounds[0], 10, 64) if len(bounds) == 2 && bounds[1] != "" { @@ -326,6 +329,19 @@ func TestAdaptiveIntegrationConnectionLimit429(t *testing.T) { if got != 4 { t.Fatalf("settled at %d connections, want 4", got) } + // A 429 on a follow-up request after a range split is retried before + // anything is parked, so the count stays at the limit most of the time. + steady := samples[30 : len(samples)-15] + at4 := 0 + for _, n := range steady { + if n == 4 { + at4++ + } + } + t.Logf("at 4 connections for %d%% of the steady part", 100*at4/len(steady)) + if at4*10 < len(steady)*7 { + t.Fatalf("at 4 connections for only %d of %d samples, want at least 70%%", at4, len(steady)) + } if s.rejected.Load() == 0 { t.Fatal("the server never refused a connection, so the limit was not found") } @@ -620,3 +636,51 @@ func TestAdaptivePatchWithoutOptionsKeepsConnections(t *testing.T) { t.Fatalf("connections = %d after an empty Patch, want 8", got) } } + +// An origin that advertises Range but ignores it: the first request falls back +// to a sequential download, and the adaptive loop hands over to slow start. +func TestAdaptiveIntegrationIgnoredRangeHandsOverToSequential(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 4<<20, 8<<20, 0, 0, false) + s.ignoreRange = true + f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) + if !f.Meta().Res.Range { + t.Fatal("setup: the server must advertise Range") + } + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 30*time.Second); err != nil { + t.Fatalf("an ignored Range must fall back without error: %v", err) + } + if f.adaptive == nil { + t.Fatal("setup: adaptive must have started before the fallback") + } + if f.Meta().Res.Range { + t.Fatal("the fetcher did not fall back to a sequential download") + } + if n := len(f.Stats().Snapshot.(*http.Stats).Connections); n != 1 { + t.Fatalf("got %d connections, want 1", n) + } + assertAdaptiveFile(t, path, s.data) +} + +// The task switch turns adaptive on although the global default is off. +func TestAdaptiveTaskOnOverridesGlobalOff(t *testing.T) { + useFastAdaptiveClock(t) + s := newAdaptiveTestServer(t, 8<<20, 4<<20, 0, 0, false) + f, path := newAdaptiveFetcher(t, config{Connections: 16}, s, &http.OptsExtra{Adaptive: boolPtr(true)}) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 30*time.Second); err != nil { + t.Fatal(err) + } + if f.adaptive == nil { + t.Fatal("the task switch did not turn adaptive on") + } + if got := f.slowStart.batchHistory; fmt.Sprint(got) != "[1]" { + t.Fatalf("slow-start batches = %v, want only the first connection's book entry", got) + } + assertAdaptiveFile(t, path, s.data) +} diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 70bcae5..d23e81c 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -139,6 +139,10 @@ type connection struct { parked bool // asked to stop; its range is stealable once it has exited running bool // a goroutine started by adaptiveLaunchLocked is live gotData bool // got a 2xx response since it was last launched + // limitHits counts 429/503 responses in a row after data. The first is + // retried: after a range split, the server may still be serving this + // connection's previous request for a moment. + limitHits int // Speed tracking for work stealing decisions speed int64 // bytes per second @@ -1477,6 +1481,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [ conn.State = connDownloading conn.failed = false conn.gotData = true + conn.limitHits = 0 f.connMu.Unlock() if conn.Role == rolePrimary || conn.ID == 0 { @@ -2819,6 +2824,7 @@ func (f *Fetcher) adaptiveResume() { func (f *Fetcher) adaptiveLaunchLocked(conn *connection) { conn.running = true conn.gotData = false + conn.limitHits = 0 f.wg.Add(2) // one for runConnection, one for this wrapper go func() { defer f.wg.Done() @@ -2899,15 +2905,22 @@ func (f *Fetcher) adaptiveAddLocked() bool { // running connection. The caller must hold connMu. func (f *Fetcher) adaptiveParkSlowestLocked() bool { var victim *connection + var victimSpeed int64 active := 0 for _, conn := range f.connections { if !conn.running || conn.parked { continue } active++ + // A connection that is not downloading (connecting, or waiting to + // retry) delivers nothing now, whatever its last measured speed. + speed := conn.speed + if conn.State != connDownloading { + speed = 0 + } // The slowest; on a tie the newest, which is the one just added. - if victim == nil || conn.speed < victim.speed || (conn.speed == victim.speed && conn.ID > victim.ID) { - victim = conn + if victim == nil || speed < victimSpeed || (speed == victimSpeed && conn.ID > victim.ID) { + victim, victimSpeed = conn, speed } } if active <= 1 || victim == nil { @@ -2959,13 +2972,15 @@ func (f *Fetcher) adaptiveIdle() bool { } // adaptiveSnapshot returns the running, unparked connection count, whether -// one of them has a validated response, and the bytes downloaded so far. +// one of them has a validated response, and the bytes downloaded so far. A +// connection waiting to retry (State connFailed while running) is not counted: +// to the controller it is a change from outside, not a slowdown. func (f *Fetcher) adaptiveSnapshot() (active int, ready bool, total int64) { f.connMu.Lock() defer f.connMu.Unlock() for _, conn := range f.connections { total += conn.Downloaded - if conn.running && !conn.parked { + if conn.running && !conn.parked && conn.State != connFailed { active++ if conn.gotData { ready = true @@ -2977,9 +2992,9 @@ func (f *Fetcher) adaptiveSnapshot() (active int, ready bool, total int64) { // adaptiveLimited handles a limiting status on conn with adaptive on. A 403 // or 429 before any data means the server allows no more connections; a 429 -// or 503 after data means it wants fewer. Either way, while another -// connection still runs, conn is parked (not failed) and the controller is -// told. It returns true when conn was parked and must stop. +// or 503 after data, twice in a row, means it wants fewer. Either way, while +// another connection still runs, conn is parked (not failed) and the +// controller is told. It returns true when conn was parked and must stop. func (f *Fetcher) adaptiveLimited(conn *connection, code int) bool { f.connMu.Lock() newConn := !conn.gotData @@ -2994,6 +3009,14 @@ func (f *Fetcher) adaptiveLimited(conn *connection, code int) bool { f.connMu.Unlock() return false } + if !newConn { + conn.limitHits++ + if conn.limitHits < 2 { + // Let the normal retry wait and ask once more. + f.connMu.Unlock() + return false + } + } conn.parked = true f.connMu.Unlock() -- 2.55.0 From c2848a4c374b79d79325712c3fa0bb5477a4be43 Mon Sep 17 00:00:00 2001 From: "Silas, Mariusz Grzybacz" Date: Fri, 2 Oct 2026 23:04:57 +0000 Subject: [PATCH 3/5] fix(http): adaptive growth to 32, parked lanes in the UI, faster tests - Keep a new connection when it gains at least max(3%, 0.5/n), n being the count before it: each addition must bring half of its fair share. A fixed 10% bar stopped growth near 11 connections on a server that scales linearly, so a ceiling of 32 was unreachable. Below that bar, the plateau (under 5%) and shrink rules are unchanged. - A 403/429 on a new connection lowers the ceiling only after a second refusal within 10 s; one stray refusal is measured and retried. - Stats UI: parse `parked`, count parked connections as idle in the summary and draw their lanes greyed with an idle icon, instead of showing them as downloading. - Integration tests: smaller files and 250 ms windows (the full adaptive run takes about 45 s), the wall-clock convergence tests skip under -short, the Patch test asserts on the fetcher's own connection count, fetchers are paused on cleanup, and restored fetchers skip a known 10 s prefetch wait that has its own fix. - Patch doc comments mention the HTTP options. --- internal/protocol/http/adaptive.go | 73 +++++++++--- .../http/adaptive_integration_test.go | 108 ++++++++++++------ internal/protocol/http/adaptive_test.go | 56 ++++++++- internal/protocol/http/fetcher.go | 5 +- pkg/download/downloader.go | 2 +- ui/flutter/lib/api/model/task_stats.dart | 7 ++ .../task_statistics/task_statistics_tab.dart | 66 +++++++++-- ui/flutter/lib/l10n/app_en.arb | 1 + ui/flutter/test/widget_test.dart | 58 ++++++++++ 9 files changed, 311 insertions(+), 65 deletions(-) diff --git a/internal/protocol/http/adaptive.go b/internal/protocol/http/adaptive.go index 057074d..56f637a 100644 --- a/internal/protocol/http/adaptive.go +++ b/internal/protocol/http/adaptive.go @@ -11,13 +11,23 @@ import "time" // - every window (adaptiveTickInterval) is one speed sample; // - an addition is judged on the mean of the next adaptiveEvalWindows // windows against the speed before it; -// - a gain above adaptiveGrowGain keeps the connection and adds another; -// - a gain of at least adaptivePlateauGain keeps it but stops growth; +// - a gain of at least adaptiveKeepBar(n), where n is the count before the +// addition, keeps the connection and adds another: each new connection +// must bring at least half of its fair share (1/n), and never less than +// adaptiveMinKeepGain, so a server that scales linearly can be followed +// up to a ceiling of 32; +// - below that, a gain of at least adaptivePlateauGain keeps it but stops +// growth (this band exists only while n < 10); // - a smaller gain is a plateau: the connection is given back; // - adaptiveEvalWindows windows below (1 - adaptiveShrinkDrop) of the best // speed park the slowest connection; // - every adaptiveProbeInterval, below the ceiling, one more connection is -// tried and kept only if it helps. +// tried and kept only if it helps; +// - a 403 or 429 on a new connection, twice within adaptiveLimitWindow, +// sets the ceiling to the count that worked; a 429 or 503 on a +// connection that already had data parks that connection (the fetcher +// does this on the second such response in a row; while it waits to +// retry it delivers nothing, so it is the slowest). var ( adaptiveTickInterval = 2 * time.Second adaptiveProbeInterval = 30 * time.Second @@ -25,11 +35,22 @@ var ( const ( adaptiveEvalWindows = 2 - adaptiveGrowGain = 0.10 + adaptiveFairShare = 0.5 // a new connection must bring this part of 1/n + adaptiveMinKeepGain = 0.03 // and never less than this adaptivePlateauGain = 0.05 adaptiveShrinkDrop = 0.15 + adaptiveLimitWindow = 10 * time.Second // two refusals within it set the ceiling ) +// adaptiveKeepBar is the gain that keeps a connection added to n running +// ones and continues growth: max(3%, 0.5/n). +func adaptiveKeepBar(n int) float64 { + if n < 1 { + n = 1 + } + return max(adaptiveMinKeepGain, adaptiveFairShare/float64(n)) +} + type adaptiveDecision int const ( @@ -88,6 +109,10 @@ type adaptiveController struct { // probeAt is when the next probe may start. It is zero until the first // settle, and a probe moves it a full interval on from its own start. probeAt time.Time + + // strikeAt is when a new connection was last refused. A second refusal + // within adaptiveLimitWindow sets the ceiling; one alone does not. + strikeAt time.Time } func newAdaptiveController(ceiling int, now time.Time) *adaptiveController { @@ -150,7 +175,7 @@ func (c *adaptiveController) tick(now time.Time, speed int64, active int) adapti case phaseAdded: gain := gainOf(mean, c.before) switch { - case gain > adaptiveGrowGain: + case gain >= adaptiveKeepBar(active-1): c.best = mean if active < c.ceiling { return c.add(mean, active) @@ -197,23 +222,39 @@ func (c *adaptiveController) tick(now time.Time, speed int64, active int) adapti return holdConns } -// limited reports that a connection got a limiting HTTP status. With -// newConn, a 403 or 429 on a connection that never got data, the server's -// limit is the count that worked, which becomes the ceiling. Otherwise a 429 -// or 503 hit a connection mid-stream; the fetcher has parked it, and the next -// probe waits a full interval. active is the running count after the park. +// limited reports that a connection got a limiting HTTP status; the fetcher +// has parked it, and active is the running count after the park. +// +// With newConn, a 403 or 429 on a connection that never got data: the first +// refusal may be stray (for example a request the server has not yet seen +// end), so the controller measures and tries once more. A second refusal +// within adaptiveLimitWindow means the server's limit is the count that +// worked, which becomes the ceiling. +// +// Otherwise a 429 or 503 hit a connection mid-stream: the next probe waits a +// full interval, and the lower count is measured afresh rather than read as +// a further slowdown. func (c *adaptiveController) limited(now time.Time, active int, newConn bool) { if active < 1 { active = 1 } - if newConn && active < c.ceiling { - c.ceiling = active + if !newConn { + c.probeAt = now.Add(adaptiveProbeInterval) + c.measure(active, false, false) + return } - c.probeAt = now.Add(adaptiveProbeInterval) // A refused new connection changed nothing that runs, so the best stands. - // A connection parked mid-stream is the server asking for less: measure - // afresh rather than read the lower speed as a further slowdown. - c.measure(active, false, newConn) + if !c.strikeAt.IsZero() && now.Sub(c.strikeAt) <= adaptiveLimitWindow { + c.strikeAt = time.Time{} + if active < c.ceiling { + c.ceiling = active + } + c.probeAt = now.Add(adaptiveProbeInterval) + c.measure(active, false, true) + return + } + c.strikeAt = now + c.measure(active, true, true) } func (c *adaptiveController) steady(now time.Time, speed int64, active int) adaptiveDecision { diff --git a/internal/protocol/http/adaptive_integration_test.go b/internal/protocol/http/adaptive_integration_test.go index 0be8f86..4a2b460 100644 --- a/internal/protocol/http/adaptive_integration_test.go +++ b/internal/protocol/http/adaptive_integration_test.go @@ -26,7 +26,8 @@ import ( // Test server: per-request rate, total cap, and a connection limit with 429. // ============================================================================ -const adaptivePiece = 16 * 1024 +// Small pieces keep the pacing smooth enough for 250 ms windows at 512 KiB/s. +const adaptivePiece = 4 * 1024 type adaptiveTestServer struct { data []byte @@ -142,21 +143,45 @@ func (s *adaptiveTestServer) serve(w gohttp.ResponseWriter, r *gohttp.Request) { // ============================================================================ // useFastAdaptiveClock shortens the controller's windows so the integration -// tests converge in seconds. The ratios between window, evaluation and probe -// stay those of the real 2 s / 30 s clock closely enough for the behaviour. +// tests converge in seconds. The ratio of probe to window is the real one +// (30 s / 2 s). func useFastAdaptiveClock(t *testing.T) { t.Helper() tick, probe := adaptiveTickInterval, adaptiveProbeInterval - adaptiveTickInterval, adaptiveProbeInterval = 300*time.Millisecond, 3*time.Second + adaptiveTickInterval, adaptiveProbeInterval = 250*time.Millisecond, 3750*time.Millisecond t.Cleanup(func() { adaptiveTickInterval, adaptiveProbeInterval = tick, probe }) } func boolPtr(v bool) *bool { return &v } +// skipLongAdaptive skips the wall-clock convergence tests under -short. +func skipLongAdaptive(t *testing.T) { + t.Helper() + if testing.Short() { + t.Skip("wall-clock convergence test; run without -short") + } +} + +// engineActive is the fetcher's own count of running, unparked connections. +// Unlike the server's count, it has no requests lingering after a split. +func engineActive(f *Fetcher) int { + active, _, _ := f.adaptiveSnapshot() + return active +} + +// skipRestoredPrefetchWait works around an upstream delay: a restored fetcher +// waits up to 10 s in stopPrefetchAndCopyData for a prefetch that never ran +// (prefetchDone is never set). It has its own fix outside this change. +func skipRestoredPrefetchWait(f *Fetcher) { + f.prefetchDone.Store(true) +} + func newAdaptiveFetcher(t *testing.T, cfg config, s *adaptiveTestServer, extra *http.OptsExtra) (*Fetcher, string) { t.Helper() dir := t.TempDir() f := buildConfigFetcher(cfg).(*Fetcher) + // Stop the download if the test fails midway, before the clock is reset. + t.Cleanup(func() { f.Pause() }) var optsExtra any // a typed nil would read as "set" if extra != nil { optsExtra = extra @@ -282,8 +307,9 @@ func waitDone(t *testing.T, f fetcher.Fetcher, timeout time.Duration) error { // ============================================================================ func TestAdaptiveIntegrationSettlesAtTotalCap(t *testing.T) { + skipLongAdaptive(t) useFastAdaptiveClock(t) - s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + s := newAdaptiveTestServer(t, 16<<20, 512<<10, 3<<20, 0, false) f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) tl := sampleConnections(s) @@ -297,8 +323,8 @@ func TestAdaptiveIntegrationSettlesAtTotalCap(t *testing.T) { samples := tl.finish() t.Logf("took %v; connections every 100 ms: %v", time.Since(started).Round(time.Millisecond), samples) - // Skip the growth (first 4 s) and the tail where chunks run out (last 1.5 s). - got := settled(samples, 40, len(samples)-15) + // Skip the growth (first 3.5 s) and the tail where chunks run out (last 1.5 s). + got := settled(samples, 35, len(samples)-15) t.Logf("settled at %d connections", got) if got < 5 || got > 7 { t.Fatalf("settled at %d connections, want 6 +- 1", got) @@ -310,8 +336,9 @@ func TestAdaptiveIntegrationSettlesAtTotalCap(t *testing.T) { } func TestAdaptiveIntegrationConnectionLimit429(t *testing.T) { + skipLongAdaptive(t) useFastAdaptiveClock(t) - s := newAdaptiveTestServer(t, 40<<20, 1<<20, 0, 4, false) + s := newAdaptiveTestServer(t, 14<<20, 512<<10, 0, 4, false) f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) tl := sampleConnections(s) @@ -354,8 +381,9 @@ func TestAdaptiveIntegrationConnectionLimit429(t *testing.T) { } func TestAdaptiveIntegrationSlowdownGivesBack(t *testing.T) { + skipLongAdaptive(t) useFastAdaptiveClock(t) - s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + s := newAdaptiveTestServer(t, 13<<20, 512<<10, 3<<20, 0, false) f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) tl := sampleConnections(s) @@ -365,8 +393,8 @@ func TestAdaptiveIntegrationSlowdownGivesBack(t *testing.T) { waitFor(t, 15*time.Second, "growth to 6 connections", func() bool { return s.active.Load() >= 6 }) time.Sleep(time.Second) - // The server now gives only 2 MB/s in total. - s.totalCap.Store(2 << 20) + // The server now gives only 1 MiB/s in total: two connections' worth. + s.totalCap.Store(1 << 20) mark := len(tl.snapshot()) defer func() { if t.Failed() { @@ -390,8 +418,9 @@ func TestAdaptiveIntegrationSlowdownGivesBack(t *testing.T) { } func TestAdaptiveIntegrationPauseAndResumeGrowsAgain(t *testing.T) { + skipLongAdaptive(t) useFastAdaptiveClock(t) - s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + s := newAdaptiveTestServer(t, 14<<20, 512<<10, 3<<20, 0, false) f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) if err := f.Start(); err != nil { t.Fatal(err) @@ -426,6 +455,8 @@ func TestAdaptiveIntegrationPauseAndResumeGrowsAgain(t *testing.T) { json.Unmarshal(raw, v) } resumed.Setup(ctl) + skipRestoredPrefetchWait(resumed) + t.Cleanup(func() { resumed.Pause() }) tl := sampleConnections(s) if err := resumed.Start(); err != nil { @@ -435,15 +466,13 @@ func TestAdaptiveIntegrationPauseAndResumeGrowsAgain(t *testing.T) { time.Sleep(200 * time.Millisecond) // let the sampler record it early := tl.snapshot() t.Logf("connections after resume: %v", early) - // The controller measures one connection for two 300 ms windows before it - // adds the second, so the first 500 ms of traffic show one request. (A - // restored fetcher first waits up to 10 s in stopPrefetchAndCopyData for a - // prefetch that never ran; that delay predates this change.) + // The controller measures one connection for two 250 ms windows before it + // adds the second, so the first 500 ms of traffic show one request. first := 0 for first < len(early) && early[first] == 0 { first++ } - for i := first; i < len(early) && i < first+5; i++ { + for i := first; i < len(early) && i < first+4; i++ { if early[i] > 1 { t.Fatalf("resume ran %d connections %d ms after its first request, want growth from one", early[i], (i-first)*100) } @@ -497,7 +526,7 @@ func TestAdaptiveOffKeepsSlowStartSequence(t *testing.T) { } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - s := newAdaptiveTestServer(t, 64<<20, 4<<20, 0, 0, false) + s := newAdaptiveTestServer(t, 32<<20, 8<<20, 0, 0, false) f, path := newAdaptiveFetcher(t, tc.cfg, s, tc.extra) if err := f.Start(); err != nil { t.Fatal(err) @@ -579,6 +608,8 @@ func TestAdaptiveResumesPreChangeStoreBlob(t *testing.T) { json.Unmarshal(raw, v) } f.Setup(ctl) + skipRestoredPrefetchWait(f) + t.Cleanup(func() { f.Pause() }) if err := f.Start(); err != nil { t.Fatal(err) } @@ -594,13 +625,14 @@ func TestAdaptiveResumesPreChangeStoreBlob(t *testing.T) { } func TestAdaptivePatchConnectionsLowersCeiling(t *testing.T) { + skipLongAdaptive(t) useFastAdaptiveClock(t) - s := newAdaptiveTestServer(t, 64<<20, 1<<20, 6<<20, 0, false) + s := newAdaptiveTestServer(t, 12<<20, 1<<20, 6<<20, 0, false) f, path := newAdaptiveFetcher(t, config{Connections: 16, Adaptive: true}, s, nil) if err := f.Start(); err != nil { t.Fatal(err) } - waitFor(t, 15*time.Second, "growth to 5 connections", func() bool { return s.active.Load() >= 5 }) + waitFor(t, 15*time.Second, "growth to 5 connections", func() bool { return engineActive(f) >= 5 }) if err := f.Patch(nil, &base.Options{Extra: &http.OptsExtra{Connections: 2}}); err != nil { t.Fatal(err) @@ -608,21 +640,33 @@ func TestAdaptivePatchConnectionsLowersCeiling(t *testing.T) { if got := f.meta.Opts.Extra.(*http.OptsExtra).Connections; got != 2 { t.Fatalf("connections option = %d after Patch, want 2", got) } - tl := sampleConnections(s) - waitFor(t, 5*time.Second, "the lower ceiling", func() bool { - recent := tl.snapshot() - return len(recent) >= 10 && settled(recent, len(recent)-10, len(recent)) <= 2 - }) + // The outcome is the fetcher's own count. The server may briefly still + // serve a parked connection's request after the client let it go. + waitFor(t, 5*time.Second, "the lower ceiling", func() bool { return engineActive(f) <= 2 }) s.totalCap.Store(0) - if err := waitDone(t, f, 60*time.Second); err != nil { - t.Fatal(err) - } - samples := tl.finish() - for i, n := range samples[len(samples)/2:] { - if n > 2 { - t.Fatalf("sample %d shows %d connections after the ceiling became 2", i, n) + done := make(chan error, 1) + go func() { done <- f.Wait() }() + maxAfter := 0 + deadline := time.After(60 * time.Second) + for finished := false; !finished; { + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + finished = true + case <-time.After(50 * time.Millisecond): + if n := engineActive(f); n > maxAfter { + maxAfter = n + } + case <-deadline: + t.Fatal("download did not finish within 60 s") } } + t.Logf("most connections after the ceiling became 2: %d", maxAfter) + if maxAfter > 2 { + t.Fatalf("the fetcher ran %d connections after the ceiling became 2", maxAfter) + } assertAdaptiveFile(t, path, s.data) } diff --git a/internal/protocol/http/adaptive_test.go b/internal/protocol/http/adaptive_test.go index 5a82923..4515741 100644 --- a/internal/protocol/http/adaptive_test.go +++ b/internal/protocol/http/adaptive_test.go @@ -13,6 +13,9 @@ type simServer struct { perConn int64 totalCap int64 maxConns int // a new connection above this count is rejected (0 means no limit) + // strayAt rejects the first new connection that would reach this count, + // once, and accepts every later one (0 means never). + strayAt int } func (s simServer) speed(active int) int64 { @@ -33,6 +36,7 @@ type simRun struct { timeline []int adds []time.Time parks []time.Time + strayed bool } func newSimRun(ceiling int) *simRun { @@ -49,6 +53,9 @@ func (r *simRun) step(s simServer) adaptiveDecision { if s.maxConns > 0 && r.active+1 > s.maxConns { // The new connection gets 403/429 and is parked by the engine. r.c.limited(r.now, r.active, true) + } else if s.strayAt > 0 && r.active+1 == s.strayAt && !r.strayed { + r.strayed = true + r.c.limited(r.now, r.active, true) } else { r.active++ } @@ -156,9 +163,13 @@ func TestAdaptiveCeilingFromLimited(t *testing.T) { if r.active != 4 { t.Fatalf("active = %d, want 4", r.active) } - // After the first rejection, the controller never asks above 4 again. - if len(r.adds) != 4 { - t.Fatalf("got %d additions, want 4 (three that worked and the rejected fifth)", len(r.adds)) + // Two refusals within 10 s set the ceiling; after that the controller + // never asks above 4 again. + if len(r.adds) != 5 { + t.Fatalf("got %d additions, want 5 (three that worked and the fifth refused twice)", len(r.adds)) + } + if gap := r.adds[4].Sub(r.adds[3]); gap > 10*time.Second { + t.Fatalf("the second try came %v after the first refusal, want within 10 s", gap) } } @@ -275,3 +286,42 @@ func TestAdaptiveSlowdownDuringProbeStillShrinks(t *testing.T) { t.Fatalf("active = %d after the server slowed to 2 MB/s, want connections given back (<= 3)", r.active) } } + +// One stray refusal of a new connection does not lower the ceiling. +func TestAdaptiveOneStrayRefusalKeepsCeiling(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 6 * mb, strayAt: 5}, 2*time.Minute) + t.Logf("timeline: %v", r.timeline) + if !r.strayed { + t.Fatal("setup: the stray refusal never happened") + } + if r.active < 6 { + t.Fatalf("active = %d after one stray refusal, want 6 (the ceiling must not drop to 4)", r.active) + } +} + +// Each added connection must bring at least half of its fair share, so a +// server that scales linearly reaches a high ceiling. +func TestAdaptiveReachesHighCeilingOnLinearServer(t *testing.T) { + r := newSimRun(32) + r.run(simServer{perConn: mb, totalCap: 30 * mb}, 5*time.Minute) + t.Logf("timeline: %v", r.timeline) + if r.active < 28 { + t.Fatalf("active = %d with a server that scales to 30, want at least 28", r.active) + } + if m := maxOf(r.timeline); m > 32 { + t.Fatalf("reached %d connections, above the ceiling of 32", m) + } +} + +func TestAdaptiveKeepBar(t *testing.T) { + cases := []struct { + n int + want float64 + }{{1, 0.5}, {5, 0.1}, {10, 0.05}, {16, 0.03125}, {20, 0.03}, {30, 0.03}} + for _, tc := range cases { + if got := adaptiveKeepBar(tc.n); got != tc.want { + t.Errorf("adaptiveKeepBar(%d) = %v, want %v", tc.n, got, tc.want) + } + } +} diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index d23e81c..497b162 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -2396,7 +2396,10 @@ func (f *Fetcher) checkCompletion() bool { return false } -// Patch modifies the HTTP request information. +// Patch modifies the HTTP request information and options. A positive +// opts connections value is the new ceiling: live for a running adaptive +// task, otherwise from the next start. An explicit adaptive switch applies +// from the next start. func (f *Fetcher) Patch(req *base.Request, opts *base.Options) error { // Patch request info if req != nil { diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 00be30a..4cee23f 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -668,7 +668,7 @@ func (d *Downloader) CreateWithOptions(rrId string, opts *base.Options) (taskId } // Patch modifies task-specific data based on the protocol. -// For HTTP protocol, it can modify Request info. +// For HTTP protocol, it can modify Request info and the connections and adaptive options. // For BT protocol, it can modify SelectFiles. func (d *Downloader) Patch(id string, req *base.Request, opts *base.Options) error { task := d.GetTask(id) diff --git a/ui/flutter/lib/api/model/task_stats.dart b/ui/flutter/lib/api/model/task_stats.dart index c3e1536..9ad40b3 100644 --- a/ui/flutter/lib/api/model/task_stats.dart +++ b/ui/flutter/lib/api/model/task_stats.dart @@ -68,6 +68,7 @@ class HttpConnectionStats { required this.completed, required this.failed, required this.retryTimes, + this.parked = false, }); final int downloaded; @@ -76,6 +77,11 @@ class HttpConnectionStats { final bool failed; final int retryTimes; + /// Stopped by adaptive connections because more connections did not make + /// the task faster. It is idle, not failed; other connections finish its + /// range. + final bool parked; + factory HttpConnectionStats.fromJson(Map json) { return HttpConnectionStats( downloaded: _integer(json, 'downloaded'), @@ -83,6 +89,7 @@ class HttpConnectionStats { completed: json['completed'] == true, failed: json['failed'] == true, retryTimes: _integer(json, 'retryTimes'), + parked: json['parked'] == true, ); } } diff --git a/ui/flutter/lib/features/tasks/presentation/widgets/task_statistics/task_statistics_tab.dart b/ui/flutter/lib/features/tasks/presentation/widgets/task_statistics/task_statistics_tab.dart index 23a87cd..bb912d1 100644 --- a/ui/flutter/lib/features/tasks/presentation/widgets/task_statistics/task_statistics_tab.dart +++ b/ui/flutter/lib/features/tasks/presentation/widgets/task_statistics/task_statistics_tab.dart @@ -1,5 +1,3 @@ -import 'dart:math' as math; - import 'package:flutter/material.dart' show Icons; import 'package:flutter/widgets.dart'; import 'package:flutter_riverpod/flutter_riverpod.dart'; @@ -58,9 +56,7 @@ class _HttpStatistics extends StatelessWidget { @override Widget build(BuildContext context) { - final completed = stats.connections.where((connection) => connection.completed).length; - final failed = stats.connections.where((connection) => connection.failed).length; - final downloading = stats.connections.length - completed - failed; + final summary = HttpConnectionSummary.of(stats.connections); return Column( crossAxisAlignment: CrossAxisAlignment.start, children: [ @@ -68,9 +64,10 @@ class _HttpStatistics extends StatelessWidget { const SizedBox(height: 12), _SummaryGrid( items: [ - _SummaryItem(context.l10n.downloading, math.max(0, downloading).toString()), - _SummaryItem(context.l10n.completed, completed.toString()), - _SummaryItem(context.l10n.failed, failed.toString()), + _SummaryItem(context.l10n.downloading, summary.downloading.toString()), + if (summary.idle > 0) _SummaryItem(context.l10n.connectionIdle, summary.idle.toString()), + _SummaryItem(context.l10n.completed, summary.completed.toString()), + _SummaryItem(context.l10n.failed, summary.failed.toString()), ], ), const SizedBox(height: 18), @@ -80,6 +77,41 @@ class _HttpStatistics extends StatelessWidget { } } +/// Connection counts for the HTTP statistics summary. A parked connection is +/// idle: it neither downloads nor failed. +class HttpConnectionSummary { + const HttpConnectionSummary({ + required this.downloading, + required this.idle, + required this.completed, + required this.failed, + }); + + factory HttpConnectionSummary.of(List connections) { + var downloading = 0; + var idle = 0; + var completed = 0; + var failed = 0; + for (final connection in connections) { + if (connection.completed) { + completed++; + } else if (connection.failed) { + failed++; + } else if (connection.parked) { + idle++; + } else { + downloading++; + } + } + return HttpConnectionSummary(downloading: downloading, idle: idle, completed: completed, failed: failed); + } + + final int downloading; + final int idle; + final int completed; + final int failed; +} + class HttpConnectionLanes extends StatelessWidget { const HttpConnectionLanes({super.key, required this.connections, required this.taskDownloading}); @@ -116,14 +148,17 @@ class _ConnectionLane extends StatelessWidget { @override Widget build(BuildContext context) { final palette = AppPalette.of(context); - final retrying = !connection.completed && !connection.failed && connection.retryTimes > 0; - final downloading = taskDownloading && !retrying && !connection.completed && !connection.failed; + final idle = connection.parked && !connection.completed && !connection.failed; + final retrying = !idle && !connection.completed && !connection.failed && connection.retryTimes > 0; + final downloading = taskDownloading && !idle && !retrying && !connection.completed && !connection.failed; final hasTotal = connection.total > 0; final progress = hasTotal ? (connection.downloaded / connection.total).clamp(0.0, 1.0) : null; final status = connection.failed ? context.l10n.failed : connection.completed ? context.l10n.completed + : idle + ? context.l10n.connectionIdle : retrying ? context.l10n.retrying : context.l10n.downloading; @@ -131,6 +166,8 @@ class _ConnectionLane extends StatelessWidget { ? palette.error : connection.completed ? palette.success + : idle + ? palette.textMuted : palette.brand; return LayoutBuilder( builder: (context, constraints) { @@ -162,6 +199,7 @@ class _ConnectionLane extends StatelessWidget { failed: connection.failed, retrying: retrying, completed: connection.completed, + idle: idle, ), ), ), @@ -173,8 +211,8 @@ class _ConnectionLane extends StatelessWidget { indeterminate: downloading && !hasTotal, shimmer: downloading, height: 7, - trackColor: palette.brandTrack, - fillColor: palette.brandProgress, + trackColor: idle ? palette.progressTrack : palette.brandTrack, + fillColor: idle ? palette.textMuted : palette.brandProgress, highlightStartColor: palette.brandProgress, highlightEndColor: palette.brandProgress, ), @@ -206,6 +244,7 @@ class _ConnectionStatusIcon extends StatelessWidget { required this.failed, required this.retrying, required this.completed, + this.idle = false, }); final String status; @@ -214,6 +253,7 @@ class _ConnectionStatusIcon extends StatelessWidget { final bool failed; final bool retrying; final bool completed; + final bool idle; @override Widget build(BuildContext context) { @@ -226,6 +266,8 @@ class _ConnectionStatusIcon extends StatelessWidget { ? Icons.sync_problem_rounded : completed ? Icons.check_circle_rounded + : idle + ? Icons.pause_circle_outline_rounded : Icons.downloading_rounded, size: 15, color: color, diff --git a/ui/flutter/lib/l10n/app_en.arb b/ui/flutter/lib/l10n/app_en.arb index ff810a9..714fa12 100644 --- a/ui/flutter/lib/l10n/app_en.arb +++ b/ui/flutter/lib/l10n/app_en.arb @@ -43,6 +43,7 @@ "ftpTlsExplicit": "Explicit TLS (FTPES)", "ftpTlsImplicit": "Implicit TLS (FTPS)", "adaptiveConnections": "Adaptive connections", + "connectionIdle": "Idle", "adaptiveConnectionsDescription": "Starts with one connection and adds more while the speed keeps rising, up to the Connections value. It closes connections that do not help.", "maxRetries": "Max retries", "useServerCtime": "Use server time for file creation", diff --git a/ui/flutter/test/widget_test.dart b/ui/flutter/test/widget_test.dart index c9427a1..4137625 100644 --- a/ui/flutter/test/widget_test.dart +++ b/ui/flutter/test/widget_test.dart @@ -4964,6 +4964,64 @@ void main() { expect(tester.takeException(), isNull); }); + testWidgets('HTTP connection lanes show parked connections as idle', (WidgetTester tester) async { + await _setTestSize(tester, const Size(390, 320)); + await tester.pumpWidget( + shad.ShadcnApp( + theme: AppTheme.light(), + materialTheme: AppTheme.materialLight(), + home: const Padding( + padding: EdgeInsets.all(16), + child: HttpConnectionLanes( + taskDownloading: true, + connections: [ + HttpConnectionStats(downloaded: 1024, total: 2048, completed: false, failed: false, retryTimes: 0), + HttpConnectionStats( + downloaded: 512, + total: 2048, + completed: false, + failed: false, + retryTimes: 0, + parked: true, + ), + ], + ), + ), + ), + ); + await tester.pump(const Duration(milliseconds: 50)); + + final active = tester.widget(find.byKey(const ValueKey('http-connection-progress-0'))); + final parked = tester.widget(find.byKey(const ValueKey('http-connection-progress-1'))); + expect(active.shimmer, isTrue); + expect(parked.shimmer, isFalse); + expect(parked.indeterminate, isFalse); + expect(parked.value, 0.25); + expect(parked.fillColor, isNot(active.fillColor)); + final parkedStatus = find.byKey(const ValueKey('http-connection-status-1')); + expect( + find.descendant( + of: parkedStatus, + matching: find.byWidgetPredicate((widget) => widget is Semantics && widget.properties.label == 'Idle'), + ), + findsOneWidget, + ); + expect(tester.takeException(), isNull); + }); + + test('HTTP connection summary counts parked connections as idle, not downloading', () { + final summary = HttpConnectionSummary.of(const [ + HttpConnectionStats(downloaded: 1, total: 4, completed: false, failed: false, retryTimes: 0), + HttpConnectionStats(downloaded: 1, total: 4, completed: false, failed: false, retryTimes: 0, parked: true), + HttpConnectionStats(downloaded: 4, total: 4, completed: true, failed: false, retryTimes: 0), + HttpConnectionStats(downloaded: 1, total: 4, completed: false, failed: true, retryTimes: 3), + ]); + expect(summary.downloading, 1); + expect(summary.idle, 1); + expect(summary.completed, 1); + expect(summary.failed, 1); + }); + testWidgets('HTTP connections use indeterminate progress only when total is unknown', (WidgetTester tester) async { await _setTestSize(tester, const Size(390, 320)); await tester.pumpWidget( -- 2.55.0 From a81d92a954a96c5831af99a45a8585f3a252ca9a Mon Sep 17 00:00:00 2001 From: "Silas, Mariusz Grzybacz" Date: Fri, 2 Oct 2026 23:24:34 +0000 Subject: [PATCH 4/5] fix(http): adaptive keep bar 10% to 3%, prune test, strike window in windows - Keep bar max(3%, min(10%, 0.5/n)), n the count before the addition: 10% at the start (sublinear servers ramp fast again), half of the fair share later (linear servers still reach 32). A gain of at least the bar and 5% keeps the connection and grows; at least the smaller of the two keeps it but growth waits for the next probe interval. - A probe is judged on 2 windows after against 2 windows before. - Every 4th probe interval is a prune test: park the newest connection for 2 windows; if the speed falls by less than the keep bar it stays parked and the count may not grow past n-1 for a probe interval, otherwise it is resumed. Window noise no longer ratchets the count up: with +-5% jitter for 15 simulated minutes, caps of 6, 12 and 20 connections end at 6, 12-13 and 20-21 (before: 6-8, 15-18, 24-29). - The two-refusal ceiling window is 5 windows, not 10 s, and a connection that runs after a refusal clears the strike. --- internal/protocol/http/adaptive.go | 156 +++++++++++++----- .../http/adaptive_integration_test.go | 32 ++++ internal/protocol/http/adaptive_test.go | 116 +++++++++++-- internal/protocol/http/fetcher.go | 35 ++++ 4 files changed, 282 insertions(+), 57 deletions(-) diff --git a/internal/protocol/http/adaptive.go b/internal/protocol/http/adaptive.go index 56f637a..0f8bd6e 100644 --- a/internal/protocol/http/adaptive.go +++ b/internal/protocol/http/adaptive.go @@ -10,21 +10,28 @@ import "time" // // - every window (adaptiveTickInterval) is one speed sample; // - an addition is judged on the mean of the next adaptiveEvalWindows -// windows against the speed before it; -// - a gain of at least adaptiveKeepBar(n), where n is the count before the -// addition, keeps the connection and adds another: each new connection -// must bring at least half of its fair share (1/n), and never less than -// adaptiveMinKeepGain, so a server that scales linearly can be followed -// up to a ceiling of 32; -// - below that, a gain of at least adaptivePlateauGain keeps it but stops -// growth (this band exists only while n < 10); +// windows against the mean of the adaptiveEvalWindows windows before it; +// - the keep bar for a connection added to n running ones is +// adaptiveKeepBar(n) = max(3%, min(10%, 0.5/n)): 10% at the start, then +// half of the new connection's fair share (1/n), never under 3%, so a +// server that scales linearly can be followed up to a ceiling of 32; +// - a gain of at least the bar and at least adaptivePlateauGain (5%) keeps +// the connection and adds another; +// - a gain of at least the smaller of the two keeps the connection but +// stops growth until the next probe interval; // - a smaller gain is a plateau: the connection is given back; // - adaptiveEvalWindows windows below (1 - adaptiveShrinkDrop) of the best // speed park the slowest connection; // - every adaptiveProbeInterval, below the ceiling, one more connection is // tried and kept only if it helps; -// - a 403 or 429 on a new connection, twice within adaptiveLimitWindow, -// sets the ceiling to the count that worked; a 429 or 503 on a +// - every adaptivePruneEvery-th interval the newest connection is parked +// for adaptiveEvalWindows windows instead. If the speed falls by less than +// the keep bar, the connection did not help: it stays parked, and the +// count may not grow past n-1 until a probe interval has passed. +// Otherwise it is resumed. This gives back connections that window noise +// let in; +// - a 403 or 429 on a new connection, twice within adaptiveStrikeWindows +// windows, sets the ceiling to the count that worked; a 429 or 503 on a // connection that already had data parks that connection (the fetcher // does this on the second such response in a row; while it waits to // retry it delivers nothing, so it is the slowest). @@ -34,21 +41,23 @@ var ( ) const ( - adaptiveEvalWindows = 2 - adaptiveFairShare = 0.5 // a new connection must bring this part of 1/n - adaptiveMinKeepGain = 0.03 // and never less than this - adaptivePlateauGain = 0.05 - adaptiveShrinkDrop = 0.15 - adaptiveLimitWindow = 10 * time.Second // two refusals within it set the ceiling + adaptiveEvalWindows = 2 + adaptiveFairShare = 0.5 // a new connection must bring this part of 1/n + adaptiveMinKeepGain = 0.03 // the keep bar is never lower + adaptiveMaxKeepGain = 0.10 // nor higher + adaptivePlateauGain = 0.05 + adaptiveShrinkDrop = 0.15 + adaptiveStrikeWindows = 5 // two refusals within this many windows set the ceiling + adaptivePruneEvery = 4 // every 4th probe interval is a prune test ) -// adaptiveKeepBar is the gain that keeps a connection added to n running -// ones and continues growth: max(3%, 0.5/n). +// adaptiveKeepBar is the gain a connection added to n running ones must +// bring: max(3%, min(10%, 0.5/n)). func adaptiveKeepBar(n int) float64 { if n < 1 { n = 1 } - return max(adaptiveMinKeepGain, adaptiveFairShare/float64(n)) + return max(adaptiveMinKeepGain, min(adaptiveMaxKeepGain, adaptiveFairShare/float64(n))) } type adaptiveDecision int @@ -57,6 +66,7 @@ const ( holdConns adaptiveDecision = iota addConn parkSlowest + parkNewest ) func (d adaptiveDecision) String() string { @@ -65,6 +75,8 @@ func (d adaptiveDecision) String() string { return "addConn" case parkSlowest: return "parkSlowest" + case parkNewest: + return "parkNewest" default: return "holdConns" } @@ -80,6 +92,8 @@ const ( phaseAdded // phaseParked judges a connection parked because the speed dropped. phaseParked + // phasePruned judges the newest connection, parked by a prune test. + phasePruned // phaseSteady watches for a drop and probes on a timer. phaseSteady ) @@ -111,8 +125,21 @@ type adaptiveController struct { probeAt time.Time // strikeAt is when a new connection was last refused. A second refusal - // within adaptiveLimitWindow sets the ceiling; one alone does not. + // within strikeWindow sets the ceiling; one alone does not. A connection + // added after it that runs clears it. strikeAt time.Time + + // intervals counts probe intervals; every adaptivePruneEvery-th is a + // prune test instead of a probe. + intervals int + // holdCap is the count growth may not pass until holdUntil, after a + // prune test found a connection that did not help. + holdCap int + holdUntil time.Time + + // The clock, read once so a running controller never sees it change. + probeEvery time.Duration + strikeWindow time.Duration } func newAdaptiveController(ceiling int, now time.Time) *adaptiveController { @@ -120,11 +147,22 @@ func newAdaptiveController(ceiling int, now time.Time) *adaptiveController { ceiling = 1 } return &adaptiveController{ - ceiling: ceiling, - phase: phaseMeasure, - grow: true, - expect: 1, + ceiling: ceiling, + phase: phaseMeasure, + grow: true, + expect: 1, + probeEvery: adaptiveProbeInterval, + strikeWindow: adaptiveStrikeWindows * adaptiveTickInterval, + } +} + +// growLimit is the count growth may reach now: the ceiling, or less while a +// prune test's hold lasts. +func (c *adaptiveController) growLimit(now time.Time) int { + if c.holdCap > 0 && now.Before(c.holdUntil) && c.holdCap < c.ceiling { + return c.holdCap } + return c.ceiling } // setCeiling changes the maximum, for example when the task's connections @@ -166,23 +204,28 @@ func (c *adaptiveController) tick(now time.Time, speed int64, active int) adapti if !c.keepBest || mean > c.best { c.best = mean } - if c.grow && active < c.ceiling { + if c.grow && active < c.growLimit(now) { return c.add(mean, active) } c.settle(now) return holdConns case phaseAdded: + // The added connection ran for two windows, so a refusal streak is + // over. + c.strikeAt = time.Time{} gain := gainOf(mean, c.before) + bar := adaptiveKeepBar(active - 1) switch { - case gain >= adaptiveKeepBar(active-1): + case gain >= max(bar, adaptivePlateauGain): c.best = mean - if active < c.ceiling { + if active < c.growLimit(now) { return c.add(mean, active) } c.settle(now) return holdConns - case gain >= adaptivePlateauGain: + case gain >= min(bar, adaptivePlateauGain): + // Kept, but growth waits for the next probe interval. c.best = mean c.settle(now) return holdConns @@ -218,6 +261,26 @@ func (c *adaptiveController) tick(now time.Time, speed int64, active int) adapti } c.settle(now) return holdConns + + case phasePruned: + // active is n-1 now. The n-th connection's gain is measured as for an + // addition: the speed with it against the speed without it. Below the + // bar it would need to be added, it did not help. + if gainOf(c.before, mean) < adaptiveKeepBar(active) { + c.best = mean + c.holdCap = active + c.holdUntil = now.Add(c.probeEvery) + c.settle(now) + return holdConns + } + // It was needed: resume it. + c.best = c.before + if active < c.ceiling { + c.measure(active+1, false, true) + return addConn + } + c.settle(now) + return holdConns } return holdConns } @@ -228,8 +291,8 @@ func (c *adaptiveController) tick(now time.Time, speed int64, active int) adapti // With newConn, a 403 or 429 on a connection that never got data: the first // refusal may be stray (for example a request the server has not yet seen // end), so the controller measures and tries once more. A second refusal -// within adaptiveLimitWindow means the server's limit is the count that -// worked, which becomes the ceiling. +// within adaptiveStrikeWindows windows means the server's limit is the count +// that worked, which becomes the ceiling. // // Otherwise a 429 or 503 hit a connection mid-stream: the next probe waits a // full interval, and the lower count is measured afresh rather than read as @@ -239,17 +302,17 @@ func (c *adaptiveController) limited(now time.Time, active int, newConn bool) { active = 1 } if !newConn { - c.probeAt = now.Add(adaptiveProbeInterval) + c.probeAt = now.Add(c.probeEvery) c.measure(active, false, false) return } // A refused new connection changed nothing that runs, so the best stands. - if !c.strikeAt.IsZero() && now.Sub(c.strikeAt) <= adaptiveLimitWindow { + if !c.strikeAt.IsZero() && now.Sub(c.strikeAt) <= c.strikeWindow { c.strikeAt = time.Time{} if active < c.ceiling { c.ceiling = active } - c.probeAt = now.Add(adaptiveProbeInterval) + c.probeAt = now.Add(c.probeEvery) c.measure(active, false, true) return } @@ -283,17 +346,30 @@ func (c *adaptiveController) steady(now time.Time, speed int64, active int) adap return holdConns } - if active < c.ceiling && !now.Before(c.probeAt) { - c.probeAt = now.Add(adaptiveProbeInterval) - before := speed - if prev > 0 { - before = (prev + speed) / 2 + // Probe intervals. The speed before a test is the mean of the last two + // windows, so wait for a second window after settling. + if prev > 0 && !now.Before(c.probeAt) { + c.probeAt = now.Add(c.probeEvery) + c.intervals++ + before := (prev + speed) / 2 + if c.intervals%adaptivePruneEvery == 0 && active > 1 { + return c.prune(before, active) + } + if active < c.growLimit(now) { + return c.add(before, active) } - return c.add(before, active) } return holdConns } +func (c *adaptiveController) prune(before int64, active int) adaptiveDecision { + c.phase = phasePruned + c.before = before + c.expect = active - 1 + c.samples = c.samples[:0] + return parkNewest +} + func (c *adaptiveController) add(before int64, active int) adaptiveDecision { c.phase = phaseAdded c.before = before @@ -323,7 +399,7 @@ func (c *adaptiveController) settle(now time.Time) { c.lastWindow = 0 c.lowWindows, c.lowSum = 0, 0 if c.probeAt.IsZero() || c.probeAt.Before(now) { - c.probeAt = now.Add(adaptiveProbeInterval) + c.probeAt = now.Add(c.probeEvery) } } diff --git a/internal/protocol/http/adaptive_integration_test.go b/internal/protocol/http/adaptive_integration_test.go index 4a2b460..950a481 100644 --- a/internal/protocol/http/adaptive_integration_test.go +++ b/internal/protocol/http/adaptive_integration_test.go @@ -2,6 +2,7 @@ package http import ( "bytes" + "context" "encoding/json" "fmt" gohttp "net/http" @@ -728,3 +729,34 @@ func TestAdaptiveTaskOnOverridesGlobalOff(t *testing.T) { } assertAdaptiveFile(t, path, s.data) } + +// The prune test parks the most recently launched running connection, and +// never the last one. +func TestAdaptiveParkNewestPicksLatestLaunch(t *testing.T) { + f := &Fetcher{} + mk := func(id, seq int, running bool) *connection { + ctx, cancel := context.WithCancel(context.Background()) + return &connection{ID: id, launchSeq: seq, running: running, ctx: ctx, cancel: cancel} + } + // ID 5 was revived last, so it is newer than ID 7. + f.connections = []*connection{mk(0, 1, true), mk(5, 9, true), mk(7, 4, true), mk(8, 12, false)} + f.connMu.Lock() + ok := f.adaptiveParkNewestLocked() + f.connMu.Unlock() + if !ok || !f.connections[1].parked || f.connections[1].ctx.Err() == nil { + t.Fatalf("parked=%v, want connection 5 (the latest launch) parked and cancelled", ok) + } + for _, i := range []int{0, 2, 3} { + if f.connections[i].parked { + t.Fatalf("connection %d was parked too", f.connections[i].ID) + } + } + + single := &Fetcher{connections: []*connection{mk(0, 1, true)}} + single.connMu.Lock() + ok = single.adaptiveParkNewestLocked() + single.connMu.Unlock() + if ok || single.connections[0].parked { + t.Fatal("parked the last running connection") + } +} diff --git a/internal/protocol/http/adaptive_test.go b/internal/protocol/http/adaptive_test.go index 4515741..451d9b0 100644 --- a/internal/protocol/http/adaptive_test.go +++ b/internal/protocol/http/adaptive_test.go @@ -1,6 +1,9 @@ package http import ( + "math" + "math/rand" + "sort" "testing" "time" ) @@ -16,13 +19,24 @@ type simServer struct { // strayAt rejects the first new connection that would reach this count, // once, and accepts every later one (0 means never). strayAt int + // sqrt makes the speed grow with the square root of the count. + sqrt bool + // noise, when set, multiplies every window by 1 + U(-jitter, +jitter). + noise *rand.Rand + jitter float64 } func (s simServer) speed(active int) int64 { speed := int64(active) * s.perConn + if s.sqrt { + speed = int64(math.Sqrt(float64(active)) * float64(s.perConn)) + } if s.totalCap > 0 && speed > s.totalCap { speed = s.totalCap } + if s.noise != nil { + speed = int64(float64(speed) * (1 + s.jitter*(2*s.noise.Float64()-1))) + } return speed } @@ -34,8 +48,10 @@ type simRun struct { now time.Time active int timeline []int - adds []time.Time - parks []time.Time + adds []time.Time // every addConn + probes []time.Time // addConn that starts a judged addition (growth or probe) + parks []time.Time // every park decision + prunes []time.Time // parkNewest strayed bool } @@ -50,6 +66,9 @@ func (r *simRun) step(s simServer) adaptiveDecision { switch d { case addConn: r.adds = append(r.adds, r.now) + if r.c.phase == phaseAdded { + r.probes = append(r.probes, r.now) + } if s.maxConns > 0 && r.active+1 > s.maxConns { // The new connection gets 403/429 and is parked by the engine. r.c.limited(r.now, r.active, true) @@ -62,6 +81,10 @@ func (r *simRun) step(s simServer) adaptiveDecision { case parkSlowest: r.parks = append(r.parks, r.now) r.active-- + case parkNewest: + r.parks = append(r.parks, r.now) + r.prunes = append(r.prunes, r.now) + r.active-- } r.timeline = append(r.timeline, r.active) return d @@ -165,11 +188,11 @@ func TestAdaptiveCeilingFromLimited(t *testing.T) { } // Two refusals within 10 s set the ceiling; after that the controller // never asks above 4 again. - if len(r.adds) != 5 { - t.Fatalf("got %d additions, want 5 (three that worked and the fifth refused twice)", len(r.adds)) + if len(r.probes) != 5 { + t.Fatalf("got %d judged additions, want 5 (three that worked and the fifth refused twice)", len(r.probes)) } - if gap := r.adds[4].Sub(r.adds[3]); gap > 10*time.Second { - t.Fatalf("the second try came %v after the first refusal, want within 10 s", gap) + if gap := r.probes[4].Sub(r.probes[3]); gap > 5*adaptiveTickInterval { + t.Fatalf("the second try came %v after the first refusal, want within 5 windows", gap) } } @@ -180,15 +203,22 @@ func TestAdaptiveProbesEvery30s(t *testing.T) { if r.active != 3 { t.Fatalf("setup: active = %d, want 3 (timeline %v)", r.active, r.timeline) } - r.adds, r.timeline = nil, nil + r.probes, r.prunes, r.timeline = nil, nil, nil r.run(capped, 5*time.Minute) t.Logf("timeline: %v", r.timeline) - // One probe per 30 s, each given back because it does not help. - if n := len(r.adds); n < 9 || n > 11 { - t.Fatalf("got %d probes in 5 minutes, want 10 +- 1", n) - } - for i := 1; i < len(r.adds); i++ { - if gap := r.adds[i].Sub(r.adds[i-1]); gap < 30*time.Second { + // One test per 30 s: a probe that is given back because it does not + // help, or (every 4th interval) a prune that is undone because the + // connection is needed. + events := append(append([]time.Time(nil), r.probes...), r.prunes...) + sort.Slice(events, func(i, j int) bool { return events[i].Before(events[j]) }) + if n := len(events); n < 9 || n > 11 { + t.Fatalf("got %d probes and prunes in 5 minutes, want 10 +- 1", n) + } + if n := len(r.prunes); n < 2 || n > 3 { + t.Fatalf("got %d prunes in 10 intervals, want every 4th (2 or 3)", n) + } + for i := 1; i < len(events); i++ { + if gap := events[i].Sub(events[i-1]); gap < 30*time.Second { t.Fatalf("probes %v apart, want at least 30 s", gap) } } @@ -257,10 +287,10 @@ func TestAdaptiveMidStreamLimitParksAndHoldsOff(t *testing.T) { // One connection gets 429 on a later request; the engine parks it. r.active = 3 r.c.limited(r.now, 3, false) - r.adds = nil + r.probes = nil r.run(simServer{perConn: mb, totalCap: 4 * mb}, 25*time.Second) - if len(r.adds) != 0 { - t.Fatalf("got %d additions within 25 s of a 429, want none before the probe", len(r.adds)) + if len(r.probes) != 0 { + t.Fatalf("got %d additions within 25 s of a 429, want none before the probe", len(r.probes)) } } @@ -318,10 +348,62 @@ func TestAdaptiveKeepBar(t *testing.T) { cases := []struct { n int want float64 - }{{1, 0.5}, {5, 0.1}, {10, 0.05}, {16, 0.03125}, {20, 0.03}, {30, 0.03}} + }{{1, 0.1}, {5, 0.1}, {8, 0.0625}, {10, 0.05}, {16, 0.03125}, {20, 0.03}, {30, 0.03}} for _, tc := range cases { if got := adaptiveKeepBar(tc.n); got != tc.want { t.Errorf("adaptiveKeepBar(%d) = %v, want %v", tc.n, got, tc.want) } } } + +// A server whose speed grows with the square root of the count still gets +// several connections quickly: the keep bar starts at 10%, not 50%. +func TestAdaptiveSublinearServerGrowsFast(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, sqrt: true}, 30*time.Second) + t.Logf("timeline: %v", r.timeline) + if r.active < 5 { + t.Fatalf("active = %d after 30 s on a sqrt(n) server, want at least 5", r.active) + } +} + +// With +-5% noise on every window, a capped server does not ratchet the count +// upwards: useless connections that noise lets in are pruned again. +func TestAdaptiveJitterStaysNearCap(t *testing.T) { + for _, capConns := range []int{6, 12, 20} { + for seed := int64(1); seed <= 10; seed++ { + r := newSimRun(32) + s := simServer{perConn: mb, totalCap: int64(capConns) * mb, noise: rand.New(rand.NewSource(seed)), jitter: 0.05} + r.run(s, 15*time.Minute) + if r.active > capConns+2 || r.active < capConns-1 { + t.Errorf("cap %d, seed %d: final count %d, want %d..%d (max %d, %d probes, %d prunes)", + capConns, seed, r.active, capConns-1, capConns+2, maxOf(r.timeline), len(r.probes), len(r.prunes)) + } else { + t.Logf("cap %d, seed %d: final %d, max %d, %d prunes", capConns, seed, r.active, maxOf(r.timeline), len(r.prunes)) + } + } + } +} + +// A connection that does not help is found by the periodic prune, given +// back, and the count stays one lower for a probe interval. +func TestAdaptivePruneGivesBackUselessConnection(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 4 * mb}, 30*time.Second) + if r.active != 4 { + t.Fatalf("setup: active = %d, want 4", r.active) + } + // One extra connection slips in (as noise might let it): the server is + // capped, so the fifth adds nothing. + r.active = 5 + r.c.expect = 5 + r.prunes = nil + r.run(simServer{perConn: mb, totalCap: 4 * mb}, 3*time.Minute) + t.Logf("timeline: %v", r.timeline) + if len(r.prunes) == 0 { + t.Fatal("no prune in 3 minutes, want one every 4th probe interval") + } + if r.active != 4 { + t.Fatalf("active = %d, want the useless fifth connection given back (4)", r.active) + } +} diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 497b162..f024fb1 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -143,6 +143,7 @@ type connection struct { // retried: after a range split, the server may still be serving this // connection's previous request for a moment. limitHits int + launchSeq int // order of the last launch; the highest is the newest // Speed tracking for work stealing decisions speed int64 // bytes per second @@ -298,6 +299,9 @@ type Fetcher struct { // nextConnID is the next connection ID. IDs only grow, so an ID is never // reused even if connections are removed. connMu guards it. nextConnID int + // launchSeq numbers adaptive launches, so a prune test can find the + // newest connection. connMu guards it. + launchSeq int // Max connection time for adaptive timeout (stored as int64 nanoseconds for atomic ops) maxConnTime atomic.Int64 @@ -2770,6 +2774,10 @@ func (f *Fetcher) adaptiveLoop(ctx context.Context, isResume bool) bool { f.connMu.Lock() f.adaptiveParkSlowestLocked() f.connMu.Unlock() + case parkNewest: + f.connMu.Lock() + f.adaptiveParkNewestLocked() + f.connMu.Unlock() } } } @@ -2828,6 +2836,8 @@ func (f *Fetcher) adaptiveLaunchLocked(conn *connection) { conn.running = true conn.gotData = false conn.limitHits = 0 + f.launchSeq++ + conn.launchSeq = f.launchSeq f.wg.Add(2) // one for runConnection, one for this wrapper go func() { defer f.wg.Done() @@ -2936,6 +2946,31 @@ func (f *Fetcher) adaptiveParkSlowestLocked() bool { return true } +// adaptiveParkNewestLocked carries out parkNewest, the prune test: it parks +// the most recently launched running connection. It never parks the last +// one. The caller must hold connMu. +func (f *Fetcher) adaptiveParkNewestLocked() bool { + var victim *connection + active := 0 + for _, conn := range f.connections { + if !conn.running || conn.parked { + continue + } + active++ + if victim == nil || conn.launchSeq > victim.launchSeq { + victim = conn + } + } + if active <= 1 || victim == nil { + return false + } + victim.parked = true + if victim.cancel != nil { + victim.cancel() + } + return true +} + // takeParkedRangeLocked gives helper the whole remaining range of the stopped // parked connection with the most work left. Unlike stealing from a running // connection there is no minimum size: nobody else would fetch it. The caller -- 2.55.0 From d6afacbbd9d9a6860cb52d266b642f3c19befae1 Mon Sep 17 00:00:00 2001 From: "Silas, Mariusz Grzybacz" Date: Fri, 2 Oct 2026 23:52:03 +0000 Subject: [PATCH 5/5] fix(http): a refused prune resume is never a strike D6: after a prune test finds a connection needed, the request that resumes it may be refused because the server still counts the request just cancelled. That refusal (403, 429, 503, or a connection that could not be made) no longer counts as a strike and never lowers the ceiling; the connection stays parked until the next probe. Before, two such refusals per prune cycle lowered the ceiling by one each cycle, down to 1 connection in 15 simulated minutes. - Tests: refused resumes keep the ceiling at its starting value over 15 simulated minutes; a successful retry clears a strike (fails if the clear is removed); a prune decision parks a running connection in the fetcher (fails if the parkNewest dispatch is removed); the engine parks a refused resume without a strike. - The linear-to-30 test asserts at least 20 within 90 s and at least 28 within 7 minutes. - The prune hold is described as about two probe intervals (about 60 s). --- internal/protocol/http/adaptive.go | 32 +++++++- .../http/adaptive_integration_test.go | 59 +++++++++++++ internal/protocol/http/adaptive_test.go | 82 ++++++++++++++++++- internal/protocol/http/fetcher.go | 44 ++++++++++ 4 files changed, 211 insertions(+), 6 deletions(-) diff --git a/internal/protocol/http/adaptive.go b/internal/protocol/http/adaptive.go index 0f8bd6e..4829a5b 100644 --- a/internal/protocol/http/adaptive.go +++ b/internal/protocol/http/adaptive.go @@ -27,9 +27,14 @@ import "time" // - every adaptivePruneEvery-th interval the newest connection is parked // for adaptiveEvalWindows windows instead. If the speed falls by less than // the keep bar, the connection did not help: it stays parked, and the -// count may not grow past n-1 until a probe interval has passed. -// Otherwise it is resumed. This gives back connections that window noise -// let in; +// count may not grow past n-1 for about two probe intervals (the hold +// ends just after the next probe, so the one after that is the first +// that may add). Otherwise it is resumed. This gives back connections +// that window noise let in; +// - a refusal of the request that resumes a pruned connection is never a +// strike: the server may still count the request just cancelled. The +// connection stays parked until the next probe, and the ceiling is +// untouched; // - a 403 or 429 on a new connection, twice within adaptiveStrikeWindows // windows, sets the ceiling to the count that worked; a 429 or 503 on a // connection that already had data parks that connection (the fetcher @@ -133,10 +138,16 @@ type adaptiveController struct { // prune test instead of a probe. intervals int // holdCap is the count growth may not pass until holdUntil, after a - // prune test found a connection that did not help. + // prune test found a connection that did not help. holdUntil is one probe + // interval after the judgement, which is just after the next probe, so + // the hold lasts about two probe intervals. holdCap int holdUntil time.Time + // resuming is set while the controller brings back a connection that a + // prune test found needed. A refusal of that request is not a strike. + resuming bool + // The clock, read once so a running controller never sees it change. probeEvery time.Duration strikeWindow time.Duration @@ -204,6 +215,8 @@ func (c *adaptiveController) tick(now time.Time, speed int64, active int) adapti if !c.keepBest || mean > c.best { c.best = mean } + // A resumed connection that ran its measuring windows was not refused. + c.resuming = false if c.grow && active < c.growLimit(now) { return c.add(mean, active) } @@ -277,6 +290,7 @@ func (c *adaptiveController) tick(now time.Time, speed int64, active int) adapti c.best = c.before if active < c.ceiling { c.measure(active+1, false, true) + c.resuming = true return addConn } c.settle(now) @@ -306,6 +320,16 @@ func (c *adaptiveController) limited(now time.Time, active int, newConn bool) { c.measure(active, false, false) return } + if c.resuming { + // The request that resumes a pruned connection was refused. The server + // may still count the request just cancelled, so this is not a + // strike and the ceiling stays. The connection stays parked; the next + // probe brings it back. The lower count is measured afresh so it is + // not read as a slowdown. + c.resuming = false + c.measure(active, false, false) + return + } // A refused new connection changed nothing that runs, so the best stands. if !c.strikeAt.IsZero() && now.Sub(c.strikeAt) <= c.strikeWindow { c.strikeAt = time.Time{} diff --git a/internal/protocol/http/adaptive_integration_test.go b/internal/protocol/http/adaptive_integration_test.go index 950a481..89decbc 100644 --- a/internal/protocol/http/adaptive_integration_test.go +++ b/internal/protocol/http/adaptive_integration_test.go @@ -760,3 +760,62 @@ func TestAdaptiveParkNewestPicksLatestLaunch(t *testing.T) { t.Fatal("parked the last running connection") } } + +// D5 end to end: a prune decision parks a running connection in the +// fetcher, and the download still completes with the right bytes. +func TestAdaptiveIntegrationPruneParksAConnection(t *testing.T) { + skipLongAdaptive(t) + tick, probe := adaptiveTickInterval, adaptiveProbeInterval + // A short probe interval, so the 4th interval (the first prune) comes + // about 4 s after the count settles. At the ceiling of 3 no probe + // interrupts the steady state, so the intervals tick on time. + adaptiveTickInterval, adaptiveProbeInterval = 250*time.Millisecond, time.Second + t.Cleanup(func() { adaptiveTickInterval, adaptiveProbeInterval = tick, probe }) + s := newAdaptiveTestServer(t, 12<<20, 512<<10, 0, 0, false) + f, path := newAdaptiveFetcher(t, config{Connections: 3, Adaptive: true}, s, nil) + if err := f.Start(); err != nil { + t.Fatal(err) + } + if err := waitDone(t, f, 60*time.Second); err != nil { + t.Fatal(err) + } + f.connMu.Lock() + prunes := f.adaptivePrunes + f.connMu.Unlock() + t.Logf("connections parked by prune tests: %d", prunes) + if prunes == 0 { + t.Fatal("no prune test parked a connection") + } + assertAdaptiveFile(t, path, s.data) +} + +// D6 in the engine: while a pruned connection is being resumed, a refusal +// of its request (here a 503, which is otherwise retried) parks it and is +// not a strike. +func TestAdaptiveRefusedResumeParksWithoutStrike(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + running := &connection{ID: 0, running: true, gotData: true, ctx: ctx, cancel: cancel} + resumed := &connection{ID: 1, running: true, ctx: ctx, cancel: cancel} + f := &Fetcher{connections: []*connection{running, resumed}} + f.adaptive = newAdaptiveController(16, time.Now()) + f.adaptive.resuming = true + + if !f.adaptiveRefusedResume(resumed, NewRequestError(503)) { + t.Fatal("a 503 on the resume request was not handled as a refused resume") + } + if !resumed.parked { + t.Fatal("the refused connection was not parked") + } + if f.adaptive.ceiling != 16 || !f.adaptive.strikeAt.IsZero() || f.adaptive.resuming { + t.Fatalf("ceiling=%d strike=%v resuming=%v, want 16, no strike, resume over", + f.adaptive.ceiling, !f.adaptive.strikeAt.IsZero(), f.adaptive.resuming) + } + + // Outside a resume, the same status is left to the normal handling. + other := &connection{ID: 2, running: true, ctx: ctx, cancel: cancel} + f.connections = append(f.connections, other) + if f.adaptiveRefusedResume(other, NewRequestError(503)) || other.parked { + t.Fatal("a 503 outside a resume was treated as a refused resume") + } +} diff --git a/internal/protocol/http/adaptive_test.go b/internal/protocol/http/adaptive_test.go index 451d9b0..826e024 100644 --- a/internal/protocol/http/adaptive_test.go +++ b/internal/protocol/http/adaptive_test.go @@ -19,6 +19,13 @@ type simServer struct { // strayAt rejects the first new connection that would reach this count, // once, and accepts every later one (0 means never). strayAt int + // strays rejects the first new connection that would reach each listed + // count, once per count. + strays []int + // refuseResume models a server that still counts a pruned connection's + // cancelled request for 10 s: every request that would bring the count + // back to what it was before the prune is refused in that time. + refuseResume bool // sqrt makes the speed grow with the square root of the count. sqrt bool // noise, when set, multiplies every window by 1 + U(-jitter, +jitter). @@ -53,6 +60,10 @@ type simRun struct { parks []time.Time // every park decision prunes []time.Time // parkNewest strayed bool + strayHit map[int]bool + resumes int // requests refused while a pruned connection's slot lingered + pruneAt time.Time + pruneN int } func newSimRun(ceiling int) *simRun { @@ -75,6 +86,11 @@ func (r *simRun) step(s simServer) adaptiveDecision { } else if s.strayAt > 0 && r.active+1 == s.strayAt && !r.strayed { r.strayed = true r.c.limited(r.now, r.active, true) + } else if r.strayOnce(s, r.active+1) { + r.c.limited(r.now, r.active, true) + } else if s.refuseResume && r.active+1 >= r.pruneN && r.now.Sub(r.pruneAt) <= 10*time.Second { + r.resumes++ + r.c.limited(r.now, r.active, true) } else { r.active++ } @@ -84,12 +100,26 @@ func (r *simRun) step(s simServer) adaptiveDecision { case parkNewest: r.parks = append(r.parks, r.now) r.prunes = append(r.prunes, r.now) + r.pruneAt, r.pruneN = r.now, r.active r.active-- } r.timeline = append(r.timeline, r.active) return d } +func (r *simRun) strayOnce(s simServer, count int) bool { + for _, n := range s.strays { + if n == count && !r.strayHit[n] { + if r.strayHit == nil { + r.strayHit = map[int]bool{} + } + r.strayHit[n] = true + return true + } + } + return false +} + func (r *simRun) run(s simServer, d time.Duration) { for end := r.now.Add(d); r.now.Before(end); { r.step(s) @@ -332,12 +362,21 @@ func TestAdaptiveOneStrayRefusalKeepsCeiling(t *testing.T) { // Each added connection must bring at least half of its fair share, so a // server that scales linearly reaches a high ceiling. +// Up to about 21 connections each addition gains over 5% and growth chains, +// so the ramp is fast. Above that a linear gain (1/n) is under 5%: the +// connection is kept, but growth waits for the next probe interval (D4), and +// every 4th interval is a prune test (D5) instead of a probe. func TestAdaptiveReachesHighCeilingOnLinearServer(t *testing.T) { r := newSimRun(32) - r.run(simServer{perConn: mb, totalCap: 30 * mb}, 5*time.Minute) + server := simServer{perConn: mb, totalCap: 30 * mb} + r.run(server, 90*time.Second) + if r.active < 20 { + t.Fatalf("active = %d after 90 s with a server that scales to 30, want at least 20", r.active) + } + r.run(server, 7*time.Minute-90*time.Second) t.Logf("timeline: %v", r.timeline) if r.active < 28 { - t.Fatalf("active = %d with a server that scales to 30, want at least 28", r.active) + t.Fatalf("active = %d after 7 minutes with a server that scales to 30, want at least 28", r.active) } if m := maxOf(r.timeline); m > 32 { t.Fatalf("reached %d connections, above the ceiling of 32", m) @@ -407,3 +446,42 @@ func TestAdaptivePruneGivesBackUselessConnection(t *testing.T) { t.Fatalf("active = %d, want the useless fifth connection given back (4)", r.active) } } + +// A strike is cleared once a later connection has run. Then a single refusal +// shortly after is a new stray, not the second strike. +func TestAdaptiveSuccessfulRetryClearsStrike(t *testing.T) { + r := newSimRun(16) + // The first try at 5 and the first try at 6 are both refused once. The + // second refusal comes 8 s after the first, inside the 5-window strike + // window, but the successful retry at 5 in between cleared the strike. + r.run(simServer{perConn: mb, totalCap: 8 * mb, strays: []int{5, 6}}, 2*time.Minute) + t.Logf("timeline: %v", r.timeline) + if len(r.strayHit) != 2 { + t.Fatalf("setup: %d stray refusals happened, want 2", len(r.strayHit)) + } + if r.c.ceiling != 16 { + t.Fatalf("ceiling = %d, want 16 (two refusals with a successful retry between them are not a limit)", r.c.ceiling) + } + if r.active < 8 { + t.Fatalf("active = %d, want 8", r.active) + } +} + +// D6: the request that resumes a pruned connection may be refused because +// the server still counts the request just cancelled. That is never a +// strike: the ceiling stays, and the connection stays parked until the next +// probe brings it back. +func TestAdaptiveRefusedPruneResumeKeepsCeiling(t *testing.T) { + r := newSimRun(16) + r.run(simServer{perConn: mb, totalCap: 6 * mb, refuseResume: true}, 15*time.Minute) + t.Logf("timeline: %v", r.timeline) + if r.resumes < 3 { + t.Fatalf("setup: only %d resumes were refused, want several prune cycles", r.resumes) + } + if r.c.ceiling != 16 { + t.Fatalf("ceiling = %d after %d refused resumes, want 16 (unchanged)", r.c.ceiling, r.resumes) + } + if r.active < 5 { + t.Fatalf("active = %d, want the count to recover to 5 or 6 between prunes", r.active) + } +} diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index f024fb1..8f41561 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -302,6 +302,8 @@ type Fetcher struct { // launchSeq numbers adaptive launches, so a prune test can find the // newest connection. connMu guards it. launchSeq int + // adaptivePrunes counts connections parked by prune tests. connMu guards it. + adaptivePrunes int // Max connection time for adaptive timeout (stored as int64 nanoseconds for atomic ops) maxConnTime atomic.Int64 @@ -1240,6 +1242,9 @@ func (f *Fetcher) runConnection(conn *connection) { if shouldUseResolveFallback && f.completeFromResolveFallback(conn) { return } + if f.adaptive != nil && f.adaptiveRefusedResume(conn, err) { + return + } if requestErr != nil && f.adaptive != nil && f.adaptiveLimited(conn, requestErr.Code) { return } @@ -2968,6 +2973,7 @@ func (f *Fetcher) adaptiveParkNewestLocked() bool { if victim.cancel != nil { victim.cancel() } + f.adaptivePrunes++ return true } @@ -3028,6 +3034,44 @@ func (f *Fetcher) adaptiveSnapshot() (active int, ready bool, total int64) { return active, ready, total } +// adaptiveRefusedResume handles a refusal of the request that resumes a +// pruned connection: a 403, 429 or 503, or a connection that could not be +// made, before any data. The server may still count the request the prune +// just cancelled, so this is never a strike (the controller keeps its +// ceiling): conn is parked until the next probe. It returns true when conn +// was parked and must stop. +func (f *Fetcher) adaptiveRefusedResume(conn *connection, err error) bool { + f.adaptiveMu.Lock() + resuming := f.adaptive.resuming + f.adaptiveMu.Unlock() + if !resuming || isTerminalRangeError(err) { + return false + } + refused := true + if re := extractRequestError(err); re != nil { + refused = re.Code == 403 || re.Code == 429 || re.Code == 503 + } + + f.connMu.Lock() + others := 0 + for _, c := range f.connections { + if c != conn && c.running && !c.parked { + others++ + } + } + if conn.gotData || !refused || others == 0 { + f.connMu.Unlock() + return false + } + conn.parked = true + f.connMu.Unlock() + + f.adaptiveMu.Lock() + f.adaptive.limited(time.Now(), others, true) + f.adaptiveMu.Unlock() + return true +} + // adaptiveLimited handles a limiting status on conn with adaptive on. A 403 // or 429 before any data means the server allows no more connections; a 429 // or 503 after data, twice in a row, means it wants fewer. Either way, while -- 2.55.0