Compare commits
7 Commits
master
...
db73b51e0d
| Author | SHA1 | Date | |
|---|---|---|---|
| db73b51e0d | |||
| 27a5cece41 | |||
| d9de63f451 | |||
| 11734af7fb | |||
| 16cbb97748 | |||
| ee66235c36 | |||
| 3d9ea7ce8b |
10
.gitignore
vendored
10
.gitignore
vendored
@@ -7,3 +7,13 @@ ref/
|
|||||||
bin/
|
bin/
|
||||||
.torrent.db*
|
.torrent.db*
|
||||||
*.part
|
*.part
|
||||||
|
|
||||||
|
# media downloaded by running got in the repo tree (never commit content)
|
||||||
|
*.mkv
|
||||||
|
*.mp4
|
||||||
|
*.avi
|
||||||
|
*.mov
|
||||||
|
*.webm
|
||||||
|
*.iso
|
||||||
|
*.mp3
|
||||||
|
*.flac
|
||||||
|
|||||||
154
README.md
154
README.md
@@ -1,13 +1,8 @@
|
|||||||
# got
|
# got
|
||||||
|
|
||||||
A small, Rob-Pike-style command-line download tool in Go: one program that
|
A command-line download tool in Go. It fetches HTTP(S) files over multiple
|
||||||
downloads HTTP(S) URLs over several connections and BitTorrent
|
connections and downloads BitTorrent magnets and `.torrent` files — one binary,
|
||||||
magnets/torrents, with a focused set of options.
|
familiar flags.
|
||||||
|
|
||||||
The design leans on Go's runtime instead of a hand-rolled reactor. There is no
|
|
||||||
single-threaded event-poll loop: one goroutine per download blocks on real I/O,
|
|
||||||
a buffered channel of slots bounds concurrency, and `context` carries shutdown.
|
|
||||||
What remains is the data — and, per Pike, the data is the design.
|
|
||||||
|
|
||||||
## Build
|
## Build
|
||||||
|
|
||||||
@@ -15,148 +10,67 @@ What remains is the data — and, per Pike, the data is the design.
|
|||||||
go build -o got .
|
go build -o got .
|
||||||
```
|
```
|
||||||
|
|
||||||
Requires Go 1.25+. The BitTorrent engine is
|
Needs Go 1.25+.
|
||||||
[`anacrolix/torrent`](https://github.com/anacrolix/torrent); everything else is
|
|
||||||
the standard library plus `golang.org/x/{term,time}`.
|
|
||||||
|
|
||||||
## Use
|
## Usage
|
||||||
|
|
||||||
```sh
|
```sh
|
||||||
# segmented HTTP download, 16 connections, into ./downloads
|
# HTTP download with 16 connections
|
||||||
got -x16 -s16 -d downloads https://example.com/big.iso
|
got -x16 -s16 https://example.com/big.iso
|
||||||
|
|
||||||
|
# let got pick the connection count from the file size
|
||||||
|
got --auto-split https://example.com/big.iso
|
||||||
|
|
||||||
# resume an interrupted download
|
# resume an interrupted download
|
||||||
got -c -x16 https://example.com/big.iso
|
got -c https://example.com/big.iso
|
||||||
|
|
||||||
# the same file from several mirrors at once (one file, connections spread
|
# one file from several mirrors (connections spread across servers,
|
||||||
# across servers, a dead mirror falls over); -Z downloads them separately instead
|
# a dead mirror falls over); -Z downloads them as separate files instead
|
||||||
got -x4 https://a.example/big.iso https://b.example/big.iso
|
got -x4 https://a.example/big.iso https://b.example/big.iso
|
||||||
|
|
||||||
# a torrent or magnet; seed for 30 minutes after finishing
|
# a torrent or magnet; seed for 30 minutes after finishing
|
||||||
got --seed-time=30 ubuntu.torrent
|
got --seed-time=30 ubuntu.torrent
|
||||||
got 'magnet:?xt=urn:btih:...'
|
got 'magnet:?xt=urn:btih:...'
|
||||||
|
|
||||||
# list the files in a torrent without downloading
|
# many downloads, two at a time
|
||||||
got -S -T some.torrent
|
|
||||||
|
|
||||||
# several downloads at once, two running in parallel
|
|
||||||
got -j2 -i urls.txt
|
got -j2 -i urls.txt
|
||||||
```
|
```
|
||||||
|
|
||||||
Run `got --help` (or `--help=http`, `--help=bittorrent`, ...) for the full
|
Run `got --help` (or `--help=all`) for every option.
|
||||||
option list.
|
|
||||||
|
|
||||||
### Options
|
## Common options
|
||||||
|
|
||||||
A focused subset of options, with familiar short flags and sensible defaults.
|
|
||||||
|
|
||||||
| flag | meaning | default |
|
| flag | meaning | default |
|
||||||
|------|---------|---------|
|
|------|---------|---------|
|
||||||
| `-d, --dir` | output directory | `.` |
|
| `-d, --dir` | output directory | `.` |
|
||||||
| `-o, --out` | output filename (single download) | from URL |
|
| `-o, --out` | output filename | from URL |
|
||||||
| `-i, --input-file` | read downloads line by line (TAB-separated URLs = mirrors; `-` = stdin) | |
|
|
||||||
| `-c, --continue` | resume a partial download | false |
|
| `-c, --continue` | resume a partial download | false |
|
||||||
| `-x, --max-connection-per-server` | connections to one server (1–16) | 1 |
|
| `-x, --max-connection-per-server` | connections to one server (1–16) | 1 |
|
||||||
| `-s, --split` | split a download into N connections | 5 |
|
| `-s, --split` | split a download into N connections | 5 |
|
||||||
| `-Z, --force-sequential` | download each command-line URL separately, not as mirrors | false |
|
| `--auto-split` | choose connections from file size (overrides `-x`/`-s`) | false |
|
||||||
| `-k, --min-split-size` | do not split below this size | 20M |
|
| `-j, --max-concurrent-downloads` | downloads at once | 5 |
|
||||||
| `-j, --max-concurrent-downloads` | parallel downloads | 5 |
|
|
||||||
| `--max-overall-download-limit` | global speed cap | 0 (off) |
|
| `--max-overall-download-limit` | global speed cap | 0 (off) |
|
||||||
| `--max-download-limit` | per-download speed cap | 0 (off) |
|
| `--checksum` | verify the finished file: `TYPE=DIGEST` (sha-256, sha-1, …) | |
|
||||||
| `--retry-wait` | seconds to wait between retries | 0 |
|
| `--seed-time` | minutes to seed after a torrent finishes | by ratio |
|
||||||
| `--load-cookies` / `--save-cookies` | read/write a Netscape `cookies.txt` | |
|
|
||||||
| `--conditional-get` | skip the download if the local file is up to date | false |
|
|
||||||
| `--remote-time` | set the file's mtime from the server | false |
|
|
||||||
| `--checksum` | verify the finished file: `TYPE=DIGEST` (sha-256, sha-1, md5, …) | |
|
|
||||||
| `--dry-run` | check the file is available but do not download it | false |
|
|
||||||
| `--ca-certificate` | verify HTTPS against the CA certificates in FILE (PEM) | |
|
|
||||||
| `--stop` | stop the program after N seconds | 0 (off) |
|
|
||||||
| `--disable-ipv6` | force IPv4-only connections | false |
|
|
||||||
| `-T, --torrent-file` | a `.torrent` file | |
|
|
||||||
| `--seed-time` | minutes to seed after finishing | seed by ratio |
|
|
||||||
| `--seed-ratio` | stop seeding at this ratio | 1.0 |
|
| `--seed-ratio` | stop seeding at this ratio | 1.0 |
|
||||||
| `--listen-port` | port (range) for incoming peers | 6881-6999 |
|
|
||||||
| `--enable-dht` | use the BitTorrent DHT | true |
|
|
||||||
| `--select-file` | fetch only these file indexes (`1,3-5`) | all |
|
|
||||||
| `-S, --show-files` | list torrent files and exit | |
|
| `-S, --show-files` | list torrent files and exit | |
|
||||||
| `--bt-stop-timeout` | give up if no download progress for N s | 0 (off) |
|
| `-q, --quiet` | no progress output | false |
|
||||||
| `--bt-metadata-timeout` | give up if a magnet can't fetch metadata in N s | 60 |
|
|
||||||
| `--save-session` | on exit, write unfinished downloads to FILE | |
|
|
||||||
| `-q, --quiet` | no progress readout | false |
|
|
||||||
|
|
||||||
### Resuming across runs
|
## Resume
|
||||||
|
|
||||||
There is no daemon — resume is a plain file. Point
|
HTTP downloads resume from a small `<file>.got` sidecar — just re-run with `-c`.
|
||||||
`--save-session` and `-i` at the same file and add `-c`: unfinished downloads
|
Torrents resume from the data already on disk. To resume across runs, point
|
||||||
are loaded at start and the still-unfinished ones are written back on exit.
|
`--save-session` and `-i` at the same file:
|
||||||
|
|
||||||
```sh
|
```sh
|
||||||
got -c --save-session=got.session -i got.session <new urls/magnets...>
|
got -c --save-session=got.session -i got.session <urls/magnets...>
|
||||||
```
|
```
|
||||||
|
|
||||||
On the first run the session file need not exist. HTTP downloads resume from
|
## Notes
|
||||||
their `.got` control file; torrents resume from the data already on disk
|
|
||||||
(re-checked by the engine). A magnet with no reachable peers no longer hangs:
|
|
||||||
`--bt-metadata-timeout` bounds the metadata fetch and defaults to 60s (set it to
|
|
||||||
0 to wait forever).
|
|
||||||
|
|
||||||
## Design
|
- BitTorrent uses [`anacrolix/torrent`](https://github.com/anacrolix/torrent);
|
||||||
|
HTTP uses the standard library.
|
||||||
```
|
- Not implemented: FTP/SFTP, Metalink, and the JSON-RPC server.
|
||||||
main classify URIs, wire options, run, report exit code
|
- When several torrents run at once, `--max-download-limit` /
|
||||||
└─ download the Download interface + the scheduler (-j semaphore, ctx)
|
`--max-upload-limit` apply per run rather than per torrent (`--max-overall-*`
|
||||||
├─ httpdl []Segment + worker pool + WriteAt + JSON resume sidecar
|
are exact).
|
||||||
└─ bt anacrolix/torrent behind the Download interface
|
|
||||||
├─ cli one flat option table -> hand parser -> layered config
|
|
||||||
└─ progress one ticker goroutine, pull-snapshot, \r line redraw
|
|
||||||
```
|
|
||||||
|
|
||||||
The contract is one small interface:
|
|
||||||
|
|
||||||
```go
|
|
||||||
type Download interface {
|
|
||||||
Name() string
|
|
||||||
Run(ctx context.Context) error // blocks in its own goroutine
|
|
||||||
Stat() Stat // a value snapshot, lock-free to read
|
|
||||||
}
|
|
||||||
```
|
|
||||||
|
|
||||||
- **HTTP** splits the file into byte-range `Segment`s handed to a pool of
|
|
||||||
workers. Each worker streams its range straight to one shared file with
|
|
||||||
`os.File.WriteAt` — safe for concurrent non-overlapping writes, so there is no
|
|
||||||
shared seek offset and no mutex. Resume is a small JSON sidecar
|
|
||||||
(`<file>.got`) holding per-segment progress plus validators (length +
|
|
||||||
ETag/Last-Modified) so a stale file is never trusted.
|
|
||||||
- **BitTorrent** is `anacrolix/torrent` configured from the CLI options and
|
|
||||||
driven to completion, then seeded for `--seed-time` minutes or up to
|
|
||||||
`--seed-ratio`.
|
|
||||||
- **Options** are one flat `[]Opt` table — the single source of truth for
|
|
||||||
parsing, validation, defaults and `--help`. Layers apply in order:
|
|
||||||
built-in defaults, config file, proxy environment, command line.
|
|
||||||
- **Progress** is a single goroutine that ticks once a second, pulls a snapshot
|
|
||||||
of the running downloads, derives speeds from the change since the last tick,
|
|
||||||
and redraws one line with `\r` + erase-to-end-of-line.
|
|
||||||
|
|
||||||
## Status
|
|
||||||
|
|
||||||
Implemented: segmented HTTP(S) with resume — including a foreign/browser-started
|
|
||||||
partial file (`-c`) — multi-mirror downloads (several URLs, or TAB-grouped `-i`
|
|
||||||
lines, fetch one file with connections spread across servers and fallover; `-Z`
|
|
||||||
to download them separately), retries, rate limits, auto-renaming,
|
|
||||||
`Content-Disposition` naming, conditional GET, cookies
|
|
||||||
(`--load-cookies`/`--save-cookies`), HTTP basic auth, `--remote-time`, whole-file
|
|
||||||
checksum verification (`--checksum`), and a `--dry-run` availability check;
|
|
||||||
BitTorrent download + seeding (magnet and `.torrent`,
|
|
||||||
DHT, trackers, file selection, show-files, metadata-fetch timeout), seeding that
|
|
||||||
stops on whichever of `--seed-time`/`--seed-ratio` comes first, and a listen port
|
|
||||||
chosen from the whole range so a busy port no longer disables BitTorrent;
|
|
||||||
a command-line interface with documented exit codes, config file, live progress,
|
|
||||||
and session save/reload for resume across runs.
|
|
||||||
|
|
||||||
Deferred (not implemented yet): FTP/SFTP, Metalink, and the
|
|
||||||
JSON-RPC server. The architecture leaves room for
|
|
||||||
each — a new protocol is just another `Download`. One known limitation: the
|
|
||||||
BitTorrent engine (`anacrolix/torrent`) applies download/upload rate limits
|
|
||||||
client-wide, so `--max-upload-limit`/`--max-download-limit` act per-run rather
|
|
||||||
than per-torrent when several torrents run at once (`--max-overall-*` are exact);
|
|
||||||
DHT/PEX likewise cannot yet be disabled per private torrent.
|
|
||||||
|
|||||||
17
bt/bt.go
17
bt/bt.go
@@ -9,12 +9,16 @@ package bt
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"log/slog"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
alog "github.com/anacrolix/log"
|
||||||
missinggo "github.com/anacrolix/missinggo/v2"
|
missinggo "github.com/anacrolix/missinggo/v2"
|
||||||
"github.com/anacrolix/torrent"
|
"github.com/anacrolix/torrent"
|
||||||
"github.com/anacrolix/torrent/metainfo"
|
"github.com/anacrolix/torrent/metainfo"
|
||||||
@@ -73,6 +77,14 @@ func ParsePorts(spec string) []int {
|
|||||||
return ports
|
return ports
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// quietSlogger filters anacrolix's structured logging down to Error and above,
|
||||||
|
// so the routine webseed/peer/tracker warning chatter (the 403/429 noise) no
|
||||||
|
// longer interleaves with — and corrupts — the live progress display. Genuine
|
||||||
|
// errors still reach w.
|
||||||
|
func quietSlogger(w io.Writer) *slog.Logger {
|
||||||
|
return slog.New(slog.NewTextHandler(w, &slog.HandlerOptions{Level: slog.LevelError}))
|
||||||
|
}
|
||||||
|
|
||||||
// NewClient builds the shared anacrolix client from the run-wide config. When a
|
// NewClient builds the shared anacrolix client from the run-wide config. When a
|
||||||
// listen-port spec is given it tries each candidate port in turn, advancing past
|
// listen-port spec is given it tries each candidate port in turn, advancing past
|
||||||
// any that is already in use, and falls back to an OS-chosen port if none bind,
|
// any that is already in use, and falls back to an OS-chosen port if none bind,
|
||||||
@@ -91,6 +103,11 @@ func ParsePorts(spec string) []int {
|
|||||||
// prevention must come from a run-wide setting (e.g. --enable-dht=false).
|
// prevention must come from a run-wide setting (e.g. --enable-dht=false).
|
||||||
func NewClient(cfg ClientConfig) (*torrent.Client, error) {
|
func NewClient(cfg ClientConfig) (*torrent.Client, error) {
|
||||||
c := torrent.NewDefaultClientConfig()
|
c := torrent.NewDefaultClientConfig()
|
||||||
|
// Keep the library's own logging out of the readout. Webseed/peer warnings go
|
||||||
|
// through slog, the rest through the legacy analog logger, so filter both to
|
||||||
|
// Error and above (the analog default is Warning, which is what leaks today).
|
||||||
|
c.Slogger = quietSlogger(os.Stderr)
|
||||||
|
c.Logger = alog.Default.FilterLevel(alog.Error)
|
||||||
c.DataDir = cfg.Dir
|
c.DataDir = cfg.Dir
|
||||||
if c.DataDir == "" {
|
if c.DataDir == "" {
|
||||||
c.DataDir = "."
|
c.DataDir = "."
|
||||||
|
|||||||
@@ -1,10 +1,38 @@
|
|||||||
package bt
|
package bt
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"log/slog"
|
||||||
"reflect"
|
"reflect"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// TestQuietSloggerDropsWarnings: the library's warning chatter (the webseed
|
||||||
|
// 403/429 noise) is filtered out so it can't corrupt the live display, while a
|
||||||
|
// genuine error still gets through.
|
||||||
|
func TestQuietSloggerDropsWarnings(t *testing.T) {
|
||||||
|
var buf bytes.Buffer
|
||||||
|
lg := quietSlogger(&buf)
|
||||||
|
|
||||||
|
lg.Warn("webseed request error", "err", "429 Too Many Requests")
|
||||||
|
if buf.Len() != 0 {
|
||||||
|
t.Errorf("warning leaked into output: %q", buf.String())
|
||||||
|
}
|
||||||
|
// anacrolix's webseed code logs via Log(ctx, level, ...); confirm that path is
|
||||||
|
// filtered too, not just the Warn helper.
|
||||||
|
lg.Log(context.Background(), slog.LevelWarn, "another warning")
|
||||||
|
if buf.Len() != 0 {
|
||||||
|
t.Errorf("Log-at-Warn leaked: %q", buf.String())
|
||||||
|
}
|
||||||
|
|
||||||
|
lg.Error("real failure")
|
||||||
|
if !strings.Contains(buf.String(), "real failure") {
|
||||||
|
t.Errorf("error was suppressed, want it kept: %q", buf.String())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestParsePorts(t *testing.T) {
|
func TestParsePorts(t *testing.T) {
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
spec string
|
spec string
|
||||||
|
|||||||
@@ -78,9 +78,10 @@ var options = []Opt{
|
|||||||
{Long: "human-readable", Kind: Bool, Default: "true", Help: "show sizes as Ki/Mi/Gi", Tag: Basic},
|
{Long: "human-readable", Kind: Bool, Default: "true", Help: "show sizes as Ki/Mi/Gi", Tag: Basic},
|
||||||
|
|
||||||
// --- http ---
|
// --- http ---
|
||||||
{Long: "max-connection-per-server", Short: 'x', Kind: Int, Default: "1", Min: 1, Max: 16, Help: "max connections to one server (1-16)", Tag: HTTP},
|
{Long: "max-connection-per-server", Short: 'x', Kind: Int, Default: "1", Min: 1, Max: 16, Help: "max connections to one server (1-16)", Tag: Basic},
|
||||||
{Long: "split", Short: 's', Kind: Int, Default: "5", Min: 1, Help: "split a download into N connections; actual connections are min(max-connection-per-server, split), and -x defaults to 1", Tag: HTTP},
|
{Long: "split", Short: 's', Kind: Int, Default: "5", Min: 1, Help: "split a download into N connections; actual connections are min(max-connection-per-server, split), and -x defaults to 1", Tag: Basic},
|
||||||
{Long: "min-split-size", Short: 'k', Kind: Size, Default: "20M", Min: 1 << 20, Max: 1 << 30, Help: "do not split a piece smaller than SIZE (1M-1024M)", Tag: HTTP},
|
{Long: "min-split-size", Short: 'k', Kind: Size, Default: "20M", Min: 1 << 20, Max: 1 << 30, Help: "do not split a piece smaller than SIZE (1M-1024M)", Tag: Basic},
|
||||||
|
{Long: "auto-split", Kind: Bool, Default: "false", Help: "got-only: pick the connection count from file size (one per min-split-size, up to 16); overrides -x/-s", Tag: HTTP},
|
||||||
{Long: "force-sequential", Short: 'Z', Kind: Bool, Default: "false", Help: "download each command-line URI as its own file instead of mirroring them", Tag: HTTP},
|
{Long: "force-sequential", Short: 'Z', Kind: Bool, Default: "false", Help: "download each command-line URI as its own file instead of mirroring them", Tag: HTTP},
|
||||||
{Long: "max-tries", Short: 'm', Kind: Int, Default: "5", Min: 0, Help: "max retries per segment (0 = unlimited)", Tag: HTTP},
|
{Long: "max-tries", Short: 'm', Kind: Int, Default: "5", Min: 0, Help: "max retries per segment (0 = unlimited)", Tag: HTTP},
|
||||||
{Long: "timeout", Short: 't', Kind: Int, Default: "60", Min: 1, Help: "connection timeout in seconds", Tag: HTTP},
|
{Long: "timeout", Short: 't', Kind: Int, Default: "60", Min: 1, Help: "connection timeout in seconds", Tag: HTTP},
|
||||||
|
|||||||
@@ -91,6 +91,10 @@ func (e *Engine) track(d Download) {
|
|||||||
e.mu.Unlock()
|
e.mu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// untrack removes d from the active set. Every Download implementation is a
|
||||||
|
// pointer type, so the == compares identities — never the fields of a value,
|
||||||
|
// which could panic on a non-comparable one. (Keying on Stat().ID is not an
|
||||||
|
// option: a torrent's ID changes from source to infohash once metadata loads.)
|
||||||
func (e *Engine) untrack(d Download) {
|
func (e *Engine) untrack(d Download) {
|
||||||
e.mu.Lock()
|
e.mu.Lock()
|
||||||
defer e.mu.Unlock()
|
defer e.mu.Unlock()
|
||||||
|
|||||||
2
go.mod
2
go.mod
@@ -3,6 +3,7 @@ module github.com/hanbok/got
|
|||||||
go 1.25
|
go 1.25
|
||||||
|
|
||||||
require (
|
require (
|
||||||
|
github.com/anacrolix/log v0.17.1-0.20251118025802-918f1157b7bb
|
||||||
github.com/anacrolix/missinggo/v2 v2.10.0
|
github.com/anacrolix/missinggo/v2 v2.10.0
|
||||||
github.com/anacrolix/torrent v1.61.0
|
github.com/anacrolix/torrent v1.61.0
|
||||||
golang.org/x/sys v0.38.0
|
golang.org/x/sys v0.38.0
|
||||||
@@ -19,7 +20,6 @@ require (
|
|||||||
github.com/anacrolix/envpprof v1.4.0 // indirect
|
github.com/anacrolix/envpprof v1.4.0 // indirect
|
||||||
github.com/anacrolix/generics v0.1.1-0.20251125230353-15d98d46693b // indirect
|
github.com/anacrolix/generics v0.1.1-0.20251125230353-15d98d46693b // indirect
|
||||||
github.com/anacrolix/go-libutp v1.3.2 // indirect
|
github.com/anacrolix/go-libutp v1.3.2 // indirect
|
||||||
github.com/anacrolix/log v0.17.1-0.20251118025802-918f1157b7bb // indirect
|
|
||||||
github.com/anacrolix/missinggo v1.3.0 // indirect
|
github.com/anacrolix/missinggo v1.3.0 // indirect
|
||||||
github.com/anacrolix/missinggo/perf v1.0.0 // indirect
|
github.com/anacrolix/missinggo/perf v1.0.0 // indirect
|
||||||
github.com/anacrolix/mmsg v1.0.1 // indirect
|
github.com/anacrolix/mmsg v1.0.1 // indirect
|
||||||
|
|||||||
@@ -27,11 +27,13 @@ type segState struct {
|
|||||||
|
|
||||||
func controlPath(out string) string { return out + ".got" }
|
func controlPath(out string) string { return out + ".got" }
|
||||||
|
|
||||||
// snapshot builds a control record from the live segments.
|
// snapshot builds a control record from the live segments. Callers hold the
|
||||||
func snapshot(url string, total int64, etag, lastmod string, segs []seg) control {
|
// pool lock (see pool.snapshot) so the slice and each segment's frontier are
|
||||||
|
// stable for the read.
|
||||||
|
func snapshot(url string, total int64, etag, lastmod string, segs []*seg) control {
|
||||||
c := control{URL: url, Total: total, ETag: etag, LastModified: lastmod, Segs: make([]segState, len(segs))}
|
c := control{URL: url, Total: total, ETag: etag, LastModified: lastmod, Segs: make([]segState, len(segs))}
|
||||||
for i := range segs {
|
for i := range segs {
|
||||||
c.Segs[i] = segState{segs[i].start, segs[i].end, atomic.LoadInt64(&segs[i].written)}
|
c.Segs[i] = segState{segs[i].start, segs[i].endOff(), atomic.LoadInt64(&segs[i].written)}
|
||||||
}
|
}
|
||||||
return c
|
return c
|
||||||
}
|
}
|
||||||
|
|||||||
160
httpdl/httpdl.go
160
httpdl/httpdl.go
@@ -57,6 +57,7 @@ type Config struct {
|
|||||||
Split int // --split: total connections across all mirrors
|
Split int // --split: total connections across all mirrors
|
||||||
MaxConnPerServer int // --max-connection-per-server: per-host connection cap
|
MaxConnPerServer int // --max-connection-per-server: per-host connection cap
|
||||||
MinSplit int64
|
MinSplit int64
|
||||||
|
AutoSplit bool // --auto-split (got-only): pick the connection count from file size
|
||||||
Tries int // 0 = unlimited
|
Tries int // 0 = unlimited
|
||||||
Timeout time.Duration
|
Timeout time.Duration
|
||||||
FileAlloc string // none | prealloc | trunc | falloc
|
FileAlloc string // none | prealloc | trunc | falloc
|
||||||
@@ -119,7 +120,7 @@ type Config struct {
|
|||||||
// connections.
|
// connections.
|
||||||
type Download struct {
|
type Download struct {
|
||||||
uris []string // mirror list; uris[0] is the primary (naming + resume key)
|
uris []string // mirror list; uris[0] is the primary (naming + resume key)
|
||||||
maxConns int // segment workers = min(split, len(uris)*max-connection-per-server)
|
maxConns int // manual segment-worker cap = min(split, len(uris)*max-connection-per-server); --auto-split derives its own count per file in segmented()
|
||||||
cfg Config
|
cfg Config
|
||||||
client *http.Client
|
client *http.Client
|
||||||
limit []*rate.Limiter
|
limit []*rate.Limiter
|
||||||
@@ -214,10 +215,17 @@ func New(uris []string, cfg Config) *Download {
|
|||||||
if perHost < 1 {
|
if perHost < 1 {
|
||||||
perHost = 1
|
perHost = 1
|
||||||
}
|
}
|
||||||
|
// The transport must allow as many concurrent connections per host as we may
|
||||||
|
// actually open. --auto-split scales up to maxAutoConns per host once the file
|
||||||
|
// size is known (in segmented), so raise the cap to that ceiling when it is on.
|
||||||
|
connCap := perHost
|
||||||
|
if cfg.AutoSplit {
|
||||||
|
connCap = maxAutoConns
|
||||||
|
}
|
||||||
tr := &http.Transport{
|
tr := &http.Transport{
|
||||||
Proxy: http.ProxyFromEnvironment,
|
Proxy: http.ProxyFromEnvironment,
|
||||||
MaxConnsPerHost: perHost,
|
MaxConnsPerHost: connCap,
|
||||||
MaxIdleConnsPerHost: perHost,
|
MaxIdleConnsPerHost: connCap,
|
||||||
ResponseHeaderTimeout: cfg.Timeout,
|
ResponseHeaderTimeout: cfg.Timeout,
|
||||||
TLSHandshakeTimeout: connectTimeout,
|
TLSHandshakeTimeout: connectTimeout,
|
||||||
DialContext: dialContext,
|
DialContext: dialContext,
|
||||||
@@ -528,29 +536,43 @@ func (d *Download) withRetries(ctx context.Context, what string, attempt func()
|
|||||||
return retriesExhausted(what, tries, lastErr)
|
return retriesExhausted(what, tries, lastErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// overMirrors calls fn against each mirror URL in turn, starting at offset start
|
||||||
|
// so callers can spread work across mirrors (segment index) and rotate on retry.
|
||||||
|
// It returns nil on the first success, ctx.Err() if cancelled between mirrors, or
|
||||||
|
// the last error once every mirror is spent.
|
||||||
|
func (d *Download) overMirrors(ctx context.Context, start int, fn func(uri string) error) error {
|
||||||
|
var lastErr error
|
||||||
|
for i := range d.uris {
|
||||||
|
if ctx.Err() != nil {
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
if err := fn(d.mirror(start + i)); err != nil {
|
||||||
|
lastErr = err
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return lastErr
|
||||||
|
}
|
||||||
|
|
||||||
// probeRetry runs probe within the shared retry policy (item [18]). The probe
|
// probeRetry runs probe within the shared retry policy (item [18]). The probe
|
||||||
// result is captured through pr because withRetries only carries an error.
|
// result is captured through pr because withRetries only carries an error.
|
||||||
func (d *Download) probeRetry(ctx context.Context) (probeResult, error) {
|
func (d *Download) probeRetry(ctx context.Context) (probeResult, error) {
|
||||||
// Probe mirrors in order (primary first); the first that answers anchors the
|
// Probe mirrors in order (primary first); the first that answers anchors the
|
||||||
// size and validators for the whole download. A dead/404 primary falls over
|
// size and validators for the whole download. A dead/404 primary falls over
|
||||||
// to the next mirror.
|
// to the next mirror.
|
||||||
var lastErr error
|
|
||||||
for _, u := range d.uris {
|
|
||||||
if ctx.Err() != nil {
|
|
||||||
return probeResult{}, ctx.Err()
|
|
||||||
}
|
|
||||||
var pr probeResult
|
var pr probeResult
|
||||||
err := d.withRetries(ctx, u, func() error {
|
err := d.overMirrors(ctx, 0, func(uri string) error {
|
||||||
|
return d.withRetries(ctx, uri, func() error {
|
||||||
var perr error
|
var perr error
|
||||||
pr, perr = d.probe(ctx, u)
|
pr, perr = d.probe(ctx, uri)
|
||||||
return perr
|
return perr
|
||||||
})
|
})
|
||||||
if err == nil {
|
})
|
||||||
|
if err != nil {
|
||||||
|
return probeResult{}, err
|
||||||
|
}
|
||||||
return pr, nil
|
return pr, nil
|
||||||
}
|
|
||||||
lastErr = err
|
|
||||||
}
|
|
||||||
return probeResult{}, lastErr
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// probe issues a one-byte ranged GET to learn the total size, whether the
|
// probe issues a one-byte ranged GET to learn the total size, whether the
|
||||||
@@ -610,7 +632,13 @@ func (d *Download) probe(ctx context.Context, uri string) (probeResult, error) {
|
|||||||
// segmented downloads total bytes over d.maxConns workers, with per-segment
|
// segmented downloads total bytes over d.maxConns workers, with per-segment
|
||||||
// resume from the control file.
|
// resume from the control file.
|
||||||
func (d *Download) segmented(ctx context.Context, out string, total int64, etag, lastmod string) error {
|
func (d *Download) segmented(ctx context.Context, out string, total int64, etag, lastmod string) error {
|
||||||
segs := makeSegments(total, d.cfg.MinSplit, d.maxConns)
|
// --auto-split picks the connection count from the file size now that total is
|
||||||
|
// known; otherwise use the manual cap fixed at construction.
|
||||||
|
conns := d.maxConns
|
||||||
|
if d.cfg.AutoSplit {
|
||||||
|
conns = autoConns(total, d.cfg.MinSplit)
|
||||||
|
}
|
||||||
|
segs := makeSegments(total, d.cfg.MinSplit, conns)
|
||||||
if c := d.resumeControl(out, etag, lastmod); c != nil {
|
if c := d.resumeControl(out, etag, lastmod); c != nil {
|
||||||
segs = segsFromControl(c)
|
segs = segsFromControl(c)
|
||||||
} else if d.cfg.Continue && !fileExists(controlPath(out)) {
|
} else if d.cfg.Continue && !fileExists(controlPath(out)) {
|
||||||
@@ -648,36 +676,31 @@ func (d *Download) segmented(ctx context.Context, out string, total int64, etag,
|
|||||||
ctx, cancel := context.WithCancel(ctx)
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
|
p := newPool(segs, d.cfg.MinSplit)
|
||||||
|
|
||||||
// Periodically persist progress so a crash can resume.
|
// Periodically persist progress so a crash can resume.
|
||||||
saveDone := make(chan struct{})
|
saveDone := make(chan struct{})
|
||||||
go d.saveLoop(ctx, out, total, etag, lastmod, segs, saveDone)
|
go d.saveLoop(ctx, out, total, etag, lastmod, p, saveDone)
|
||||||
|
|
||||||
jobs := make(chan *seg)
|
|
||||||
go func() {
|
|
||||||
defer close(jobs)
|
|
||||||
for i := range segs {
|
|
||||||
if segs[i].done() {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
select {
|
|
||||||
case jobs <- &segs[i]:
|
|
||||||
case <-ctx.Done(): // stop feeding once we're tearing down
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
|
// One worker per connection. A worker fetches segments until acquire runs
|
||||||
|
// dry, which happens only once every remaining byte is owned; a worker that
|
||||||
|
// finishes early steals the tail of a slower segment rather than sitting idle
|
||||||
|
// while the last segment drains over a single connection.
|
||||||
var (
|
var (
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
errOnce sync.Once
|
errOnce sync.Once
|
||||||
runErr error
|
runErr error
|
||||||
)
|
)
|
||||||
for w := 0; w < d.maxConns; w++ {
|
for w := 0; w < conns; w++ {
|
||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
for s := range jobs {
|
for {
|
||||||
if err := d.fetchSeg(ctx, f, s, total); err != nil {
|
s := p.acquire()
|
||||||
|
if s == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := d.fetchSeg(ctx, f, p, s, total); err != nil {
|
||||||
errOnce.Do(func() { runErr = err; cancel() })
|
errOnce.Do(func() { runErr = err; cancel() })
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -691,7 +714,7 @@ func (d *Download) segmented(ctx context.Context, out string, total int64, etag,
|
|||||||
if runErr != nil {
|
if runErr != nil {
|
||||||
return runErr // keep the control file for a later -c
|
return runErr // keep the control file for a later -c
|
||||||
}
|
}
|
||||||
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
|
p.snapshot(d.primary(), total, etag, lastmod).save(out)
|
||||||
removeControl(out)
|
removeControl(out)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -699,37 +722,27 @@ func (d *Download) segmented(ctx context.Context, out string, total int64, etag,
|
|||||||
// fetchSeg downloads one segment, retrying from its resume point on error.
|
// fetchSeg downloads one segment, retrying from its resume point on error.
|
||||||
// Each attempt re-reads s.offset(), so a retry continues from the bytes already
|
// Each attempt re-reads s.offset(), so a retry continues from the bytes already
|
||||||
// written rather than restarting the range.
|
// written rather than restarting the range.
|
||||||
func (d *Download) fetchSeg(ctx context.Context, f *os.File, s *seg, total int64) error {
|
func (d *Download) fetchSeg(ctx context.Context, f *os.File, p *pool, s *seg, total int64) error {
|
||||||
// Try each mirror in turn for this segment: a transient error retries the
|
// Try each mirror in turn for this segment: a transient error retries the
|
||||||
// same mirror under withRetries; an exhausted budget or a per-mirror permanent
|
// same mirror under withRetries; an exhausted budget or a per-mirror permanent
|
||||||
// error (a 404, a non-206, a length mismatch) falls over to the next mirror.
|
// error (a 404, a non-206, a length mismatch) falls over to the next mirror.
|
||||||
// The segment fails only once every mirror is spent. Starting at s.index
|
// The segment fails only once every mirror is spent. Starting at s.index
|
||||||
// spreads the segments across mirrors round-robin, and each attempt resumes
|
// spreads the segments across mirrors round-robin, and each attempt resumes
|
||||||
// from s.offset().
|
// from s.offset().
|
||||||
var lastErr error
|
return d.overMirrors(ctx, s.index, func(uri string) error {
|
||||||
for i := range d.uris {
|
return d.withRetries(ctx, fmt.Sprintf("segment %d", s.index), func() error {
|
||||||
if ctx.Err() != nil {
|
|
||||||
return ctx.Err()
|
|
||||||
}
|
|
||||||
uri := d.mirror(s.index + i)
|
|
||||||
err := d.withRetries(ctx, fmt.Sprintf("segment %d", s.index), func() error {
|
|
||||||
if s.done() {
|
if s.done() {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return d.fetchOnce(ctx, f, s, uri, total)
|
return d.fetchOnce(ctx, f, p, s, uri, total)
|
||||||
|
})
|
||||||
})
|
})
|
||||||
if err == nil {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
lastErr = err
|
|
||||||
}
|
|
||||||
return lastErr
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Download) fetchOnce(ctx context.Context, f *os.File, s *seg, uri string, total int64) error {
|
func (d *Download) fetchOnce(ctx context.Context, f *os.File, p *pool, s *seg, uri string, total int64) error {
|
||||||
reqCtx, cancel := context.WithCancel(ctx)
|
reqCtx, cancel := context.WithCancel(ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
req, err := d.request(reqCtx, uri, fmt.Sprintf("bytes=%d-%d", s.offset(), s.end))
|
req, err := d.request(reqCtx, uri, fmt.Sprintf("bytes=%d-%d", s.offset(), s.endOff()))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -755,7 +768,7 @@ func (d *Download) fetchOnce(ctx context.Context, f *os.File, s *seg, uri string
|
|||||||
}
|
}
|
||||||
body, stop := d.idleGuard(resp.Body, cancel)
|
body, stop := d.idleGuard(resp.Body, cancel)
|
||||||
defer stop()
|
defer stop()
|
||||||
return d.pump(ctx, f, s, body)
|
return d.pump(ctx, f, p, s, body)
|
||||||
}
|
}
|
||||||
|
|
||||||
// errIdleTimeout marks a read that stalled past the idle window, so callers can
|
// errIdleTimeout marks a read that stalled past the idle window, so callers can
|
||||||
@@ -807,10 +820,9 @@ const speedStartupGrace = 10 * time.Second
|
|||||||
|
|
||||||
// belowSpeed reports whether the bytes transferred between two cumulative
|
// belowSpeed reports whether the bytes transferred between two cumulative
|
||||||
// d.completed samples (prev -> now over one window) put the download at or
|
// d.completed samples (prev -> now over one window) put the download at or
|
||||||
// below limit. A negative delta is not a slow second: d.completed is not
|
// below limit. A negative delta is not a slow second: when a resumed Range is
|
||||||
// strictly monotonic — singleOnce rewrites it with an absolute StoreInt64 and a
|
// answered by a 200 the transfer restarts from offset 0, so d.completed jumps
|
||||||
// resumed Range answered by a 200 resets the offset to 0, so the counter can
|
// backward on that retry. Treating that backward jump as "below limit" would
|
||||||
// jump backward on a retry. Treating that backward jump as "below limit" would
|
|
||||||
// false-abort a healthy download, so a reset (delta < 0) reports false; the
|
// false-abort a healthy download, so a reset (delta < 0) reports false; the
|
||||||
// caller has already advanced its baseline, so the next full window measures the
|
// caller has already advanced its baseline, so the next full window measures the
|
||||||
// real speed. Only a genuine non-negative delta at or under the limit aborts.
|
// real speed. Only a genuine non-negative delta at or under the limit aborts.
|
||||||
@@ -901,8 +913,10 @@ func statusError(ctxMsg string, code int) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// pump copies the response body into the file at the segment's running offset,
|
// pump copies the response body into the file at the segment's running offset,
|
||||||
// stopping at the segment end and respecting rate limits.
|
// stopping at the segment end and respecting rate limits. The loop reads
|
||||||
func (d *Download) pump(ctx context.Context, f *os.File, s *seg, body io.Reader) error {
|
// remaining() afresh each turn, so a steal that shrinks s mid-transfer simply
|
||||||
|
// ends the loop early at the new end, leaving the stolen tail to its new worker.
|
||||||
|
func (d *Download) pump(ctx context.Context, f *os.File, p *pool, s *seg, body io.Reader) error {
|
||||||
buf := make([]byte, readBuf)
|
buf := make([]byte, readBuf)
|
||||||
for s.remaining() > 0 {
|
for s.remaining() > 0 {
|
||||||
n := int64(len(buf))
|
n := int64(len(buf))
|
||||||
@@ -914,7 +928,7 @@ func (d *Download) pump(ctx context.Context, f *os.File, s *seg, body io.Reader)
|
|||||||
if _, werr := f.WriteAt(buf[:rd], s.offset()); werr != nil {
|
if _, werr := f.WriteAt(buf[:rd], s.offset()); werr != nil {
|
||||||
return werr
|
return werr
|
||||||
}
|
}
|
||||||
s.advance(int64(rd))
|
p.advance(s, int64(rd))
|
||||||
atomic.AddInt64(&d.completed, int64(rd))
|
atomic.AddInt64(&d.completed, int64(rd))
|
||||||
d.throttle(ctx, rd)
|
d.throttle(ctx, rd)
|
||||||
}
|
}
|
||||||
@@ -953,23 +967,13 @@ func (d *Download) single(ctx context.Context, out string) error {
|
|||||||
first := true
|
first := true
|
||||||
// One connection at a time, but try each mirror: a mirror that fails its
|
// One connection at a time, but try each mirror: a mirror that fails its
|
||||||
// retry budget falls over to the next (resuming from what's already on disk).
|
// retry budget falls over to the next (resuming from what's already on disk).
|
||||||
var lastErr error
|
return d.overMirrors(ctx, 0, func(uri string) error {
|
||||||
for i := range d.uris {
|
return d.withRetries(ctx, uri, func() error {
|
||||||
if ctx.Err() != nil {
|
|
||||||
return ctx.Err()
|
|
||||||
}
|
|
||||||
uri := d.mirror(i)
|
|
||||||
err := d.withRetries(ctx, uri, func() error {
|
|
||||||
resumeFromDisk := !first || foreignResume
|
resumeFromDisk := !first || foreignResume
|
||||||
first = false
|
first = false
|
||||||
return d.singleOnce(ctx, out, uri, resumeFromDisk)
|
return d.singleOnce(ctx, out, uri, resumeFromDisk)
|
||||||
})
|
})
|
||||||
if err == nil {
|
})
|
||||||
return nil
|
|
||||||
}
|
|
||||||
lastErr = err
|
|
||||||
}
|
|
||||||
return lastErr
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// singleOnce performs one single-stream transfer attempt. When resumeFromDisk is
|
// singleOnce performs one single-stream transfer attempt. When resumeFromDisk is
|
||||||
@@ -1047,7 +1051,7 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
|
|||||||
}
|
}
|
||||||
off += int64(rd)
|
off += int64(rd)
|
||||||
got += int64(rd)
|
got += int64(rd)
|
||||||
atomic.StoreInt64(&d.completed, off)
|
atomic.AddInt64(&d.completed, int64(rd)) // add deltas, as pump does
|
||||||
d.throttle(ctx, rd)
|
d.throttle(ctx, rd)
|
||||||
}
|
}
|
||||||
if err == io.EOF {
|
if err == io.EOF {
|
||||||
@@ -1066,7 +1070,7 @@ func (d *Download) singleOnce(ctx context.Context, out, uri string, resumeFromDi
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag, lastmod string, segs []seg, done chan<- struct{}) {
|
func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag, lastmod string, p *pool, done chan<- struct{}) {
|
||||||
defer close(done)
|
defer close(done)
|
||||||
interval := d.cfg.AutoSaveInterval
|
interval := d.cfg.AutoSaveInterval
|
||||||
if interval <= 0 {
|
if interval <= 0 {
|
||||||
@@ -1077,10 +1081,10 @@ func (d *Download) saveLoop(ctx context.Context, out string, total int64, etag,
|
|||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
|
p.snapshot(d.primary(), total, etag, lastmod).save(out)
|
||||||
return
|
return
|
||||||
case <-t.C:
|
case <-t.C:
|
||||||
snapshot(d.primary(), total, etag, lastmod, segs).save(out)
|
p.snapshot(d.primary(), total, etag, lastmod).save(out)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,22 +1,29 @@
|
|||||||
package httpdl
|
package httpdl
|
||||||
|
|
||||||
import "sync/atomic"
|
import (
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
)
|
||||||
|
|
||||||
// seg is one contiguous byte range of the output file, downloaded by a single
|
// seg is one contiguous byte range of the output file, downloaded by a single
|
||||||
// ranged GET. A segment IS a byte range — a plain HTTP downloader does not need
|
// ranged GET. A segment IS a byte range — a plain HTTP downloader does not need
|
||||||
// the Piece/Segment/block layering that exists only to share code with
|
// the Piece/Segment/block layering that exists only to share code with
|
||||||
// BitTorrent. written is updated with atomic ops so Stat() can read it while a
|
// BitTorrent. written and end are updated with atomic ops so Stat() and the
|
||||||
// worker advances it.
|
// resume snapshot can read them while a worker advances written or a steal
|
||||||
|
// shrinks end (see pool).
|
||||||
type seg struct {
|
type seg struct {
|
||||||
index int
|
index int
|
||||||
start int64 // first byte offset, inclusive
|
start int64 // first byte offset, inclusive
|
||||||
end int64 // last byte offset, inclusive
|
end int64 // last byte offset, inclusive (a steal may shrink this)
|
||||||
written int64 // bytes already written into this segment (the resume point)
|
written int64 // bytes already written into this segment (the resume point)
|
||||||
|
owned bool // a worker is fetching this segment; guarded by pool.mu
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *seg) length() int64 { return s.end - s.start + 1 }
|
func (s *seg) endOff() int64 { return atomic.LoadInt64(&s.end) }
|
||||||
|
func (s *seg) setEnd(v int64) { atomic.StoreInt64(&s.end, v) }
|
||||||
|
func (s *seg) length() int64 { return s.endOff() - s.start + 1 }
|
||||||
func (s *seg) done() bool { return atomic.LoadInt64(&s.written) >= s.length() }
|
func (s *seg) done() bool { return atomic.LoadInt64(&s.written) >= s.length() }
|
||||||
func (s *seg) advance(n int64) { atomic.AddInt64(&s.written, n) }
|
func (s *seg) addWritten(n int64) { atomic.AddInt64(&s.written, n) }
|
||||||
func (s *seg) progress() int64 { return atomic.LoadInt64(&s.written) }
|
func (s *seg) progress() int64 { return atomic.LoadInt64(&s.written) }
|
||||||
func (s *seg) offset() int64 { return s.start + atomic.LoadInt64(&s.written) }
|
func (s *seg) offset() int64 { return s.start + atomic.LoadInt64(&s.written) }
|
||||||
func (s *seg) remaining() int64 { return s.length() - atomic.LoadInt64(&s.written) }
|
func (s *seg) remaining() int64 { return s.length() - atomic.LoadInt64(&s.written) }
|
||||||
@@ -51,3 +58,127 @@ func makeSegments(total, minSplit int64, conns int) []seg {
|
|||||||
}
|
}
|
||||||
return segs
|
return segs
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// maxAutoConns is the ceiling --auto-split scales to. aria2 caps
|
||||||
|
// max-connection-per-server (-x) at 16, so a got-chosen count honours the same
|
||||||
|
// per-server limit rather than inventing a looser one.
|
||||||
|
const maxAutoConns = 16
|
||||||
|
|
||||||
|
// autoConns picks a connection count from the file size, used only when
|
||||||
|
// --auto-split is set: one connection per min-split-size of content, at least 1
|
||||||
|
// and at most maxAutoConns. It is the got-only stand-in for hand-tuning -x/-s,
|
||||||
|
// and because the count never exceeds total/minSplit, makeSegments yields exactly
|
||||||
|
// that many balanced segments.
|
||||||
|
func autoConns(total, minSplit int64) int {
|
||||||
|
if minSplit < 1 {
|
||||||
|
minSplit = 1
|
||||||
|
}
|
||||||
|
n := total / minSplit
|
||||||
|
if n < 1 {
|
||||||
|
n = 1
|
||||||
|
}
|
||||||
|
if n > maxAutoConns {
|
||||||
|
n = maxAutoConns
|
||||||
|
}
|
||||||
|
return int(n)
|
||||||
|
}
|
||||||
|
|
||||||
|
// pool is the live set of segments for one segmented download. A worker takes a
|
||||||
|
// segment with acquire and fetches it to completion; once no fresh segment is
|
||||||
|
// left, an idle worker steals the back half of whichever in-flight segment has
|
||||||
|
// the most still to download, so the last slow segment is shared across the idle
|
||||||
|
// connections instead of draining alone. aria2 does the same on-demand split via
|
||||||
|
// its SegmentMan; without it a fixed pre-division leaves connections idle while
|
||||||
|
// one slow mirror finishes.
|
||||||
|
//
|
||||||
|
// The mutex serialises a steal against the byte accounting it splits: advance (a
|
||||||
|
// worker committing a write) takes it too, so a steal reading a victim's frontier
|
||||||
|
// can never race the owner advancing past the chosen split point. A steal fires
|
||||||
|
// only when the victim still has at least 2*minSplit to go and cuts at the
|
||||||
|
// midpoint, so both halves stay >= minSplit (--min-split-size) and the half the
|
||||||
|
// owner keeps is far larger than one read buffer — the owner's in-flight write
|
||||||
|
// therefore can never reach into the stolen tail.
|
||||||
|
type pool struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
segs []*seg
|
||||||
|
minSplit int64
|
||||||
|
}
|
||||||
|
|
||||||
|
// newPool wraps the pre-divided segments in a pool. Each is copied into its own
|
||||||
|
// allocation so a later steal can append a tail without invalidating the pointers
|
||||||
|
// workers already hold.
|
||||||
|
//
|
||||||
|
// minSplit is floored at readBuf: a steal leaves the owner the front half of the
|
||||||
|
// split, which is at least minSplit, while the owner's in-flight write is at most
|
||||||
|
// readBuf, so minSplit >= readBuf is exactly what keeps that write out of the
|
||||||
|
// stolen tail. The CLI already holds --min-split-size well above readBuf (>= 1
|
||||||
|
// MiB), so the floor only guards direct callers and tests — but it keeps the
|
||||||
|
// no-overlap invariant inside this file rather than resting on a distant flag.
|
||||||
|
func newPool(segs []seg, minSplit int64) *pool {
|
||||||
|
if minSplit < readBuf {
|
||||||
|
minSplit = readBuf
|
||||||
|
}
|
||||||
|
p := &pool{minSplit: minSplit, segs: make([]*seg, len(segs))}
|
||||||
|
for i := range segs {
|
||||||
|
s := segs[i]
|
||||||
|
p.segs[i] = &s
|
||||||
|
}
|
||||||
|
return p
|
||||||
|
}
|
||||||
|
|
||||||
|
// acquire returns the next segment for a worker to fetch: a fresh one if any
|
||||||
|
// remain, otherwise the tail split off the busiest in-flight segment. It returns
|
||||||
|
// nil when every remaining byte is already owned, i.e. the download is finishing
|
||||||
|
// and there is nothing left to steal.
|
||||||
|
func (p *pool) acquire() *seg {
|
||||||
|
p.mu.Lock()
|
||||||
|
defer p.mu.Unlock()
|
||||||
|
for _, s := range p.segs {
|
||||||
|
if !s.owned && !s.done() {
|
||||||
|
s.owned = true
|
||||||
|
return s
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return p.steal()
|
||||||
|
}
|
||||||
|
|
||||||
|
// steal splits the back half off the in-flight segment with the most remaining
|
||||||
|
// and returns it, or nil if none has enough left to be worth splitting. The
|
||||||
|
// caller holds p.mu, which keeps every owner's advance out so each victim's
|
||||||
|
// frontier is stable while we choose and commit the split.
|
||||||
|
func (p *pool) steal() *seg {
|
||||||
|
var victim *seg
|
||||||
|
for _, s := range p.segs {
|
||||||
|
if s.owned && !s.done() && s.remaining() >= 2*p.minSplit {
|
||||||
|
if victim == nil || s.remaining() > victim.remaining() {
|
||||||
|
victim = s
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if victim == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
mid := victim.offset() + victim.remaining()/2
|
||||||
|
// index only feeds the mirror round-robin (d.mirror) and the "segment N" log
|
||||||
|
// label; a tail's index need not be contiguous or unique, so len is fine.
|
||||||
|
tail := &seg{index: len(p.segs), start: mid, end: victim.endOff(), owned: true}
|
||||||
|
victim.setEnd(mid - 1)
|
||||||
|
p.segs = append(p.segs, tail)
|
||||||
|
return tail
|
||||||
|
}
|
||||||
|
|
||||||
|
// advance commits n freshly written bytes to s under the pool lock, so a
|
||||||
|
// concurrent steal sees a stable frontier for the segment it may split.
|
||||||
|
func (p *pool) advance(s *seg, n int64) {
|
||||||
|
p.mu.Lock()
|
||||||
|
s.addWritten(n)
|
||||||
|
p.mu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// snapshot builds the resume record from the live set, including any stolen
|
||||||
|
// tails, under the lock so it never races a steal appending a segment.
|
||||||
|
func (p *pool) snapshot(url string, total int64, etag, lastmod string) control {
|
||||||
|
p.mu.Lock()
|
||||||
|
defer p.mu.Unlock()
|
||||||
|
return snapshot(url, total, etag, lastmod, p.segs)
|
||||||
|
}
|
||||||
|
|||||||
@@ -3,6 +3,9 @@ package httpdl
|
|||||||
import (
|
import (
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"sort"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -44,6 +47,31 @@ func TestMakeSegments(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAutoConns(t *testing.T) {
|
||||||
|
const m = int64(20 << 20) // 20 MiB min-split, aria2's default
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
total int64
|
||||||
|
minSplit int64
|
||||||
|
want int
|
||||||
|
}{
|
||||||
|
{"under one min-split is a single connection", 5 << 20, m, 1},
|
||||||
|
{"exactly one min-split", m, m, 1},
|
||||||
|
{"five min-splits", 100 << 20, m, 5},
|
||||||
|
{"caps at maxAutoConns", 10 << 30, m, maxAutoConns},
|
||||||
|
{"rounds down to whole pieces", m*3 + 1, m, 3},
|
||||||
|
{"tiny min-split still capped", 1 << 30, 1 << 20, maxAutoConns},
|
||||||
|
{"zero total is one connection", 0, m, 1},
|
||||||
|
}
|
||||||
|
for _, tc := range tests {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
if got := autoConns(tc.total, tc.minSplit); got != tc.want {
|
||||||
|
t.Errorf("autoConns(%d, %d) = %d, want %d", tc.total, tc.minSplit, got, tc.want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestSegProgress(t *testing.T) {
|
func TestSegProgress(t *testing.T) {
|
||||||
s := seg{start: 100, end: 199} // length 100
|
s := seg{start: 100, end: 199} // length 100
|
||||||
if s.length() != 100 {
|
if s.length() != 100 {
|
||||||
@@ -52,14 +80,14 @@ func TestSegProgress(t *testing.T) {
|
|||||||
if s.done() {
|
if s.done() {
|
||||||
t.Fatalf("new segment should not be done")
|
t.Fatalf("new segment should not be done")
|
||||||
}
|
}
|
||||||
s.advance(60)
|
s.addWritten(60)
|
||||||
if s.offset() != 160 {
|
if s.offset() != 160 {
|
||||||
t.Errorf("offset = %d, want 160", s.offset())
|
t.Errorf("offset = %d, want 160", s.offset())
|
||||||
}
|
}
|
||||||
if s.remaining() != 40 {
|
if s.remaining() != 40 {
|
||||||
t.Errorf("remaining = %d, want 40", s.remaining())
|
t.Errorf("remaining = %d, want 40", s.remaining())
|
||||||
}
|
}
|
||||||
s.advance(40)
|
s.addWritten(40)
|
||||||
if !s.done() {
|
if !s.done() {
|
||||||
t.Errorf("segment should be done after writing full length")
|
t.Errorf("segment should be done after writing full length")
|
||||||
}
|
}
|
||||||
@@ -186,3 +214,151 @@ func TestParseContentRange(t *testing.T) {
|
|||||||
t.Errorf("parseContentRange(garbage) ok = true, want false")
|
t.Errorf("parseContentRange(garbage) ok = true, want false")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// assertTiles checks that the pool's segments still cover [0,total) exactly:
|
||||||
|
// sorted by start they must be contiguous with no gap and no overlap.
|
||||||
|
func assertTiles(t *testing.T, p *pool, total int64) {
|
||||||
|
t.Helper()
|
||||||
|
p.mu.Lock()
|
||||||
|
type rng struct{ start, end int64 }
|
||||||
|
rs := make([]rng, len(p.segs))
|
||||||
|
for i, s := range p.segs {
|
||||||
|
rs[i] = rng{s.start, s.endOff()}
|
||||||
|
}
|
||||||
|
p.mu.Unlock()
|
||||||
|
sort.Slice(rs, func(i, j int) bool { return rs[i].start < rs[j].start })
|
||||||
|
var next int64
|
||||||
|
for _, r := range rs {
|
||||||
|
if r.start != next {
|
||||||
|
t.Fatalf("segment gap/overlap: next byte %d, got start %d (ranges %v)", next, r.start, rs)
|
||||||
|
}
|
||||||
|
next = r.end + 1
|
||||||
|
}
|
||||||
|
if next != total {
|
||||||
|
t.Fatalf("segments cover %d bytes, want %d", next, total)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPoolAcquireThenSteal(t *testing.T) {
|
||||||
|
// Sizes are in units of readBuf because that is what newPool floors minSplit
|
||||||
|
// to and what the no-overlap invariant is stated against.
|
||||||
|
const total = 8 * readBuf
|
||||||
|
p := newPool(makeSegments(total, readBuf, 1), readBuf) // one segment [0,total)
|
||||||
|
|
||||||
|
first := p.acquire()
|
||||||
|
if first == nil || first.start != 0 || first.endOff() != total-1 {
|
||||||
|
t.Fatalf("first acquire = %+v, want the whole [0,%d] segment", first, total-1)
|
||||||
|
}
|
||||||
|
if p.acquire() == nil {
|
||||||
|
// nothing fresh left, so this must steal first's back half
|
||||||
|
t.Fatal("second acquire returned nil, want a stolen tail")
|
||||||
|
}
|
||||||
|
// first kept the front half, a tail took the back half; together they still tile.
|
||||||
|
if first.endOff() != total/2-1 {
|
||||||
|
t.Errorf("victim end = %d, want %d after midpoint split", first.endOff(), total/2-1)
|
||||||
|
}
|
||||||
|
assertTiles(t, p, total)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPoolNoStealBelowThreshold(t *testing.T) {
|
||||||
|
// remaining is below 2*minSplit, so the lone segment is not worth splitting
|
||||||
|
// and an idle worker is told there is nothing to do.
|
||||||
|
p := newPool(makeSegments(readBuf+readBuf/2, readBuf, 1), readBuf)
|
||||||
|
if s := p.acquire(); s == nil {
|
||||||
|
t.Fatal("first acquire returned nil, want the only segment")
|
||||||
|
}
|
||||||
|
if s := p.acquire(); s != nil {
|
||||||
|
t.Fatalf("second acquire = %+v, want nil (tail would be below min-split)", s)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPoolConcurrentCoverage drives the pool the way real workers do — acquire a
|
||||||
|
// segment, copy it in small chunks, commit each with p.advance, repeat — across
|
||||||
|
// more workers than initial segments so stealing is forced. The coverage array
|
||||||
|
// proves the core invariant: every byte is written exactly once, so a steal
|
||||||
|
// never overlaps the owner's writes and never leaves a gap.
|
||||||
|
func TestPoolConcurrentCoverage(t *testing.T) {
|
||||||
|
const (
|
||||||
|
total = int64(64 * readBuf) // 2 MiB
|
||||||
|
minSplit = int64(readBuf) // steal threshold is 2*this
|
||||||
|
chunk = int64(readBuf / 4) // one "read", well below minSplit
|
||||||
|
workers = 8
|
||||||
|
)
|
||||||
|
p := newPool(makeSegments(total, minSplit, 1), minSplit) // start with a single segment
|
||||||
|
cover := make([]uint32, total)
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for w := 0; w < workers; w++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
for {
|
||||||
|
s := p.acquire()
|
||||||
|
if s == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
for s.remaining() > 0 {
|
||||||
|
n := chunk
|
||||||
|
if s.remaining() < n {
|
||||||
|
n = s.remaining()
|
||||||
|
}
|
||||||
|
off := s.offset()
|
||||||
|
for j := off; j < off+n; j++ {
|
||||||
|
atomic.AddUint32(&cover[j], 1)
|
||||||
|
}
|
||||||
|
p.advance(s, n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
for i, c := range cover {
|
||||||
|
if c != 1 {
|
||||||
|
t.Fatalf("byte %d written %d times, want exactly 1", i, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assertTiles(t, p, total)
|
||||||
|
if len(p.segs) == 1 {
|
||||||
|
t.Error("no stealing happened: still one segment after 8 workers drained it")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPoolSnapshotAfterStealRoundTrips checks that a steal which happens before a
|
||||||
|
// crash survives the control file: the snapshot records the stolen tail, and a
|
||||||
|
// resume rebuilds a segment set that still tiles the file and keeps the bytes
|
||||||
|
// already written. This is the persistence path an interrupt-and-resume relies on
|
||||||
|
// but cannot deterministically trigger from the outside.
|
||||||
|
func TestPoolSnapshotAfterStealRoundTrips(t *testing.T) {
|
||||||
|
const total = 8 * readBuf
|
||||||
|
p := newPool(makeSegments(total, readBuf, 1), readBuf)
|
||||||
|
a := p.acquire() // the whole file
|
||||||
|
p.advance(a, readBuf) // owner makes some progress, then idles out
|
||||||
|
b := p.acquire() // steals a's back half
|
||||||
|
if b == nil {
|
||||||
|
t.Fatal("expected a stolen tail")
|
||||||
|
}
|
||||||
|
p.advance(b, readBuf) // the tail's worker makes progress too
|
||||||
|
|
||||||
|
c := p.snapshot("http://x", total, "", "")
|
||||||
|
if len(c.Segs) != 2 {
|
||||||
|
t.Fatalf("snapshot has %d segments, want 2 (original + stolen tail)", len(c.Segs))
|
||||||
|
}
|
||||||
|
|
||||||
|
rebuilt := segsFromControl(&c)
|
||||||
|
sort.Slice(rebuilt, func(i, j int) bool { return rebuilt[i].start < rebuilt[j].start })
|
||||||
|
var next, written int64
|
||||||
|
for _, s := range rebuilt {
|
||||||
|
if s.start != next {
|
||||||
|
t.Fatalf("rebuilt gap/overlap: next byte %d, got start %d", next, s.start)
|
||||||
|
}
|
||||||
|
next = s.endOff() + 1
|
||||||
|
written += s.progress()
|
||||||
|
}
|
||||||
|
if next != total {
|
||||||
|
t.Fatalf("rebuilt covers %d bytes, want %d", next, total)
|
||||||
|
}
|
||||||
|
if written != 2*readBuf {
|
||||||
|
t.Errorf("rebuilt written = %d, want %d (bytes must survive the round trip)", written, 2*readBuf)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
58
main.go
58
main.go
@@ -190,9 +190,9 @@ type failedDownload struct {
|
|||||||
err error
|
err error
|
||||||
}
|
}
|
||||||
|
|
||||||
func (f failedDownload) Name() string { return f.name }
|
func (f *failedDownload) Name() string { return f.name }
|
||||||
func (f failedDownload) Run(context.Context) error { return f.err }
|
func (f *failedDownload) Run(context.Context) error { return f.err }
|
||||||
func (f failedDownload) Stat() download.Stat {
|
func (f *failedDownload) Stat() download.Stat {
|
||||||
return download.Stat{Name: f.name, Status: download.Errored, Total: -1}
|
return download.Stat{Name: f.name, Status: download.Errored, Total: -1}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -224,6 +224,24 @@ func dedupeTorrents(uris []string, follow bool) map[string]string {
|
|||||||
return dup
|
return dup
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// markInfoHash records the v1 infohash of the .torrent at path in seen and
|
||||||
|
// reports whether it repeats one already seen, returning the path first recorded
|
||||||
|
// under that infohash. Callers fail the duplicate rather than let two jobs share
|
||||||
|
// one torrent on the client (see bt.SourceInfoHash for why that aliasing is
|
||||||
|
// unsafe). A v2-only or unreadable infohash is never a duplicate — the dedup
|
||||||
|
// can't see it, so it runs.
|
||||||
|
func markInfoHash(seen map[metainfo.Hash]string, path string) (first string, dup bool) {
|
||||||
|
h, ok := bt.SourceInfoHash(path, true)
|
||||||
|
if !ok {
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
if first, dup = seen[h]; dup {
|
||||||
|
return first, true
|
||||||
|
}
|
||||||
|
seen[h] = path
|
||||||
|
return "", false
|
||||||
|
}
|
||||||
|
|
||||||
func build(opts *cli.Options, targets []target) ([]job, *torrent.Client, error) {
|
func build(opts *cli.Options, targets []target) ([]job, *torrent.Client, error) {
|
||||||
flat := allURIs(targets)
|
flat := allURIs(targets)
|
||||||
// Two sources naming the same torrent (a magnet and its own .torrent, say)
|
// Two sources naming the same torrent (a magnet and its own .torrent, say)
|
||||||
@@ -263,7 +281,7 @@ func build(opts *cli.Options, targets []target) ([]job, *torrent.Client, error)
|
|||||||
u := t.uris[0] // primary: classifies the target and names the session source
|
u := t.uris[0] // primary: classifies the target and names the session source
|
||||||
src := strings.Join(t.uris, "\t")
|
src := strings.Join(t.uris, "\t")
|
||||||
if first, isDup := dups[u]; isDup {
|
if first, isDup := dups[u]; isDup {
|
||||||
out = append(out, job{source: src, dl: failedDownload{
|
out = append(out, job{source: src, dl: &failedDownload{
|
||||||
name: u,
|
name: u,
|
||||||
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
|
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
|
||||||
}})
|
}})
|
||||||
@@ -358,6 +376,7 @@ func runJobs(ctx context.Context, eng *download.Engine, jobs []job) []download.R
|
|||||||
// so a regular download that merely ends in ".torrent" is left alone.
|
// so a regular download that merely ends in ".torrent" is left alone.
|
||||||
func followUps(jobs []job, client *torrent.Client, bo bt.Options) []job {
|
func followUps(jobs []job, client *torrent.Client, bo bt.Options) []job {
|
||||||
var out []job
|
var out []job
|
||||||
|
seen := map[metainfo.Hash]string{}
|
||||||
for _, j := range jobs {
|
for _, j := range jobs {
|
||||||
h, ok := j.dl.(*httpdl.Download)
|
h, ok := j.dl.(*httpdl.Download)
|
||||||
if !ok || j.dl.Stat().Status != download.Complete {
|
if !ok || j.dl.Stat().Status != download.Complete {
|
||||||
@@ -373,6 +392,17 @@ func followUps(jobs []job, client *torrent.Client, bo bt.Options) []job {
|
|||||||
if _, err := bt.Files(path); err != nil {
|
if _, err := bt.Files(path); err != nil {
|
||||||
continue // downloaded file is not a valid torrent
|
continue // downloaded file is not a valid torrent
|
||||||
}
|
}
|
||||||
|
// Two fetched .torrent files that name the same torrent would share one
|
||||||
|
// torrent on the refcount-less client: the first to finish Drops it out
|
||||||
|
// from under the other, stranding it at 0%. build() dedupes pass-1 sources
|
||||||
|
// but cannot reach a not-yet-fetched HTTP .torrent, so dedupe here too.
|
||||||
|
if first, dup := markInfoHash(seen, path); dup {
|
||||||
|
out = append(out, job{source: path, dl: &failedDownload{
|
||||||
|
name: path,
|
||||||
|
err: fmt.Errorf("%w: same torrent as %s", errDuplicate, first),
|
||||||
|
}})
|
||||||
|
continue
|
||||||
|
}
|
||||||
out = append(out, job{source: path, dl: bt.New(client, path, true, bo)})
|
out = append(out, job{source: path, dl: bt.New(client, path, true, bo)})
|
||||||
}
|
}
|
||||||
return out
|
return out
|
||||||
@@ -408,22 +438,23 @@ func allURIs(ts []target) []string {
|
|||||||
// gatherURIs collects download targets from the command line, --torrent-file,
|
// gatherURIs collects download targets from the command line, --torrent-file,
|
||||||
// and --input-file. Plain command-line http/https URIs are grouped into ONE
|
// and --input-file. Plain command-line http/https URIs are grouped into ONE
|
||||||
// mirrored download by default (every command-line URI names the same file);
|
// mirrored download by default (every command-line URI names the same file);
|
||||||
// -Z/--force-sequential makes each its own download. Torrents and magnets
|
// -Z/--force-sequential makes each its own download. Torrents, magnets, and
|
||||||
// are always one target each; --input-file groups TAB-separated URIs per line.
|
// .torrent URLs (fetched over HTTP then followed) are always one target each,
|
||||||
|
// since distinct torrents are not mirrors of one another; --input-file groups
|
||||||
|
// TAB-separated URIs per line.
|
||||||
func gatherURIs(opts *cli.Options, cmdURIs []string) []target {
|
func gatherURIs(opts *cli.Options, cmdURIs []string) []target {
|
||||||
var ts []target
|
var ts []target
|
||||||
follow := opts.Bool("follow-torrent")
|
follow := opts.Bool("follow-torrent")
|
||||||
seq := opts.Bool("force-sequential")
|
seq := opts.Bool("force-sequential")
|
||||||
var httpGroup []string
|
var httpGroup []string
|
||||||
for _, u := range cmdURIs {
|
for _, u := range cmdURIs {
|
||||||
switch uriKind(u, follow) {
|
// Only plain HTTP URLs collapse into one mirror download (alternate sources
|
||||||
case kindHTTP, kindFollow:
|
// for the same file); -Z opts out. A .torrent URL (kindFollow), like a
|
||||||
if seq {
|
// torrent or magnet, is its own download — distinct torrents are never
|
||||||
ts = append(ts, target{uris: []string{u}})
|
// mirrors of one another.
|
||||||
} else {
|
if uriKind(u, follow) == kindHTTP && !seq {
|
||||||
httpGroup = append(httpGroup, u)
|
httpGroup = append(httpGroup, u)
|
||||||
}
|
} else {
|
||||||
default:
|
|
||||||
ts = append(ts, target{uris: []string{u}})
|
ts = append(ts, target{uris: []string{u}})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -493,6 +524,7 @@ func httpConfig(opts *cli.Options, single bool, overallDL *rate.Limiter) httpdl.
|
|||||||
Split: opts.Int("split"),
|
Split: opts.Int("split"),
|
||||||
MaxConnPerServer: opts.Int("max-connection-per-server"),
|
MaxConnPerServer: opts.Int("max-connection-per-server"),
|
||||||
MinSplit: opts.Size("min-split-size"),
|
MinSplit: opts.Size("min-split-size"),
|
||||||
|
AutoSplit: opts.Bool("auto-split"),
|
||||||
Tries: opts.Int("max-tries"),
|
Tries: opts.Int("max-tries"),
|
||||||
Timeout: time.Duration(opts.Int("timeout")) * time.Second,
|
Timeout: time.Duration(opts.Int("timeout")) * time.Second,
|
||||||
FileAlloc: opts.Str("file-allocation"),
|
FileAlloc: opts.Str("file-allocation"),
|
||||||
|
|||||||
76
main_test.go
76
main_test.go
@@ -11,6 +11,9 @@ import (
|
|||||||
"syscall"
|
"syscall"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/anacrolix/torrent/bencode"
|
||||||
|
"github.com/anacrolix/torrent/metainfo"
|
||||||
|
"github.com/hanbok/got/bt"
|
||||||
"github.com/hanbok/got/cli"
|
"github.com/hanbok/got/cli"
|
||||||
"github.com/hanbok/got/download"
|
"github.com/hanbok/got/download"
|
||||||
"github.com/hanbok/got/httpdl"
|
"github.com/hanbok/got/httpdl"
|
||||||
@@ -112,8 +115,8 @@ func TestDedupeTorrents(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestGatherURIsGrouping checks the command-line grouping: plain http URIs
|
// TestGatherURIsGrouping checks the command-line grouping: plain http URIs
|
||||||
// become one mirrored download by default, -Z splits them, and torrents/magnets
|
// become one mirrored download by default, -Z splits them, and torrents,
|
||||||
// stay separate.
|
// magnets, and .torrent URLs stay separate.
|
||||||
func TestGatherURIsGrouping(t *testing.T) {
|
func TestGatherURIsGrouping(t *testing.T) {
|
||||||
parse := func(args ...string) []target {
|
parse := func(args ...string) []target {
|
||||||
res, err := cli.Parse(append([]string{"--no-conf"}, args...))
|
res, err := cli.Parse(append([]string{"--no-conf"}, args...))
|
||||||
@@ -131,6 +134,16 @@ func TestGatherURIsGrouping(t *testing.T) {
|
|||||||
if ts := parse("http://a/x", "magnet:?xt=urn:btih:abc"); len(ts) != 2 {
|
if ts := parse("http://a/x", "magnet:?xt=urn:btih:abc"); len(ts) != 2 {
|
||||||
t.Errorf("http + magnet: got %d targets, want 2 (magnet never mirrors)", len(ts))
|
t.Errorf("http + magnet: got %d targets, want 2 (magnet never mirrors)", len(ts))
|
||||||
}
|
}
|
||||||
|
// Distinct .torrent URLs are separate downloads, not mirrors of one file
|
||||||
|
// (follow-torrent defaults on, so each is fetched then followed).
|
||||||
|
if ts := parse("http://a/x.torrent", "http://b/y.torrent", "http://c/z.torrent"); len(ts) != 3 {
|
||||||
|
t.Errorf(".torrent URLs: got %d targets, want 3 separate downloads", len(ts))
|
||||||
|
}
|
||||||
|
// A plain http URL still mirror-groups even alongside a .torrent URL: one
|
||||||
|
// http group of 1 + one torrent target = 2.
|
||||||
|
if ts := parse("http://a/x", "http://b/y.torrent"); len(ts) != 2 {
|
||||||
|
t.Errorf("http + .torrent: got %d targets, want 2", len(ts))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestReadInputFileTAB checks the input-file grouping: TAB-separated URIs on
|
// TestReadInputFileTAB checks the input-file grouping: TAB-separated URIs on
|
||||||
@@ -152,6 +165,65 @@ func TestReadInputFileTAB(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// writeTorrent creates a .torrent file at dir/name whose info dict is built from
|
||||||
|
// a file of the given content and internal name. Two calls with the same
|
||||||
|
// internalName and content produce the same infohash regardless of the output
|
||||||
|
// filename, so the test can make two .torrent files that name one torrent.
|
||||||
|
func writeTorrent(t *testing.T, dir, name, internalName string, content []byte) string {
|
||||||
|
t.Helper()
|
||||||
|
data := filepath.Join(dir, internalName)
|
||||||
|
if err := os.WriteFile(data, content, 0o644); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
info := metainfo.Info{PieceLength: 32 * 1024}
|
||||||
|
if err := info.BuildFromFilePath(data); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
b, err := bencode.Marshal(info)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
tp := filepath.Join(dir, name)
|
||||||
|
f, err := os.Create(tp)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer f.Close()
|
||||||
|
if err := (&metainfo.MetaInfo{InfoBytes: b}).Write(f); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return tp
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestFollowUpInfoHashDedup covers the fix for two fetched .torrent files that
|
||||||
|
// name the same torrent: the second must be flagged a duplicate so followUps
|
||||||
|
// fails it rather than letting both share — and Drop — one torrent on the client.
|
||||||
|
func TestFollowUpInfoHashDedup(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
a := writeTorrent(t, dir, "a.torrent", "same", []byte("AAAAAAAA"))
|
||||||
|
b := writeTorrent(t, dir, "b.torrent", "same", []byte("AAAAAAAA")) // same infohash as a
|
||||||
|
c := writeTorrent(t, dir, "c.torrent", "diff", []byte("BBBBBBBB")) // a distinct torrent
|
||||||
|
|
||||||
|
// Sanity-check the fixture: a and b really share an infohash, c differs.
|
||||||
|
ha, _ := bt.SourceInfoHash(a, true)
|
||||||
|
hb, _ := bt.SourceInfoHash(b, true)
|
||||||
|
hc, _ := bt.SourceInfoHash(c, true)
|
||||||
|
if ha != hb || ha == hc {
|
||||||
|
t.Fatalf("fixture: want a==b!=c infohashes, got a=%v b=%v c=%v", ha, hb, hc)
|
||||||
|
}
|
||||||
|
|
||||||
|
seen := map[metainfo.Hash]string{}
|
||||||
|
if first, dup := markInfoHash(seen, a); dup {
|
||||||
|
t.Fatalf("a is the first occurrence, wrongly flagged a duplicate of %s", first)
|
||||||
|
}
|
||||||
|
if first, dup := markInfoHash(seen, b); !dup || first != a {
|
||||||
|
t.Errorf("b should duplicate a: got first=%q dup=%v, want %q true", first, dup, a)
|
||||||
|
}
|
||||||
|
if _, dup := markInfoHash(seen, c); dup {
|
||||||
|
t.Errorf("c is a distinct torrent, wrongly flagged a duplicate")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestReportExit(t *testing.T) {
|
func TestReportExit(t *testing.T) {
|
||||||
canceled := download.Result{Name: "a", Err: context.Canceled}
|
canceled := download.Result{Name: "a", Err: context.Canceled}
|
||||||
done := download.Result{Name: "b", Err: nil}
|
done := download.Result{Name: "b", Err: nil}
|
||||||
|
|||||||
@@ -25,8 +25,10 @@ type Reporter struct {
|
|||||||
w io.Writer
|
w io.Writer
|
||||||
tty bool
|
tty bool
|
||||||
width int
|
width int
|
||||||
|
height int
|
||||||
|
|
||||||
win map[string]*speedWindow // keyed by Stat.ID
|
win map[string]*speedWindow // keyed by Stat.ID
|
||||||
|
prevLines int // lines drawn in the previous TTY frame, redrawn in place
|
||||||
lastLog time.Time // last non-TTY line emitted
|
lastLog time.Time // last non-TTY line emitted
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -62,6 +64,7 @@ func New(snap func() []download.Stat, human bool) *Reporter {
|
|||||||
w: os.Stdout,
|
w: os.Stdout,
|
||||||
tty: term.IsTerminal(int(os.Stdout.Fd())),
|
tty: term.IsTerminal(int(os.Stdout.Fd())),
|
||||||
width: 80,
|
width: 80,
|
||||||
|
height: 24,
|
||||||
win: map[string]*speedWindow{},
|
win: map[string]*speedWindow{},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -103,29 +106,21 @@ func (r *Reporter) render(final bool) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
var line string
|
|
||||||
switch len(stats) {
|
|
||||||
case 0:
|
|
||||||
if !final && r.tty {
|
|
||||||
return // nothing active; leave the line as-is
|
|
||||||
}
|
|
||||||
case 1:
|
|
||||||
line = r.lineOne(stats[0])
|
|
||||||
default:
|
|
||||||
line = r.lineMany(stats)
|
|
||||||
}
|
|
||||||
|
|
||||||
if r.tty {
|
if r.tty {
|
||||||
r.width = termWidth(r.width)
|
r.width = termWidth(r.width)
|
||||||
line = clampRunes(line, r.width)
|
r.height = termHeight(r.height)
|
||||||
fmt.Fprint(r.w, "\r"+line+"\x1b[K")
|
r.redraw(r.block(stats), final)
|
||||||
if final {
|
|
||||||
fmt.Fprintln(r.w)
|
|
||||||
}
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Not a TTY: can't redraw one line, so print sparingly (and always the
|
// Not a TTY: no cursor control, so emit a single line sparingly (and always
|
||||||
// final line) to keep redirected output and logs readable.
|
// the final one) to keep redirected output and logs readable.
|
||||||
|
var line string
|
||||||
|
switch {
|
||||||
|
case len(stats) == 1:
|
||||||
|
line = r.lineOne(stats[0])
|
||||||
|
case len(stats) > 1:
|
||||||
|
line = r.summary(stats)
|
||||||
|
}
|
||||||
if line != "" && (final || now.Sub(r.lastLog) >= logEvery) {
|
if line != "" && (final || now.Sub(r.lastLog) >= logEvery) {
|
||||||
fmt.Fprintln(r.w, line)
|
fmt.Fprintln(r.w, line)
|
||||||
r.lastLog = now
|
r.lastLog = now
|
||||||
@@ -165,27 +160,161 @@ func (r *Reporter) lineOne(s download.Stat) string {
|
|||||||
return b.String()
|
return b.String()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *Reporter) lineMany(stats []download.Stat) string {
|
// nameW is the fixed rune width of the name column in the table, matching
|
||||||
|
// shortName's cap so a padded name lines the columns up.
|
||||||
|
const nameW = 20
|
||||||
|
|
||||||
|
// block builds the lines of the live display: the single-download dashboard for
|
||||||
|
// one download, a name/bar/speed/ETA table for several, or a one-line summary
|
||||||
|
// when the table would not fit the window.
|
||||||
|
func (r *Reporter) block(stats []download.Stat) []string {
|
||||||
|
switch {
|
||||||
|
case len(stats) == 0:
|
||||||
|
return nil
|
||||||
|
case len(stats) == 1:
|
||||||
|
return []string{r.lineOne(stats[0])}
|
||||||
|
case len(stats)+1 > r.height: // header + one row each would overflow the screen
|
||||||
|
return []string{r.summary(stats)}
|
||||||
|
default:
|
||||||
|
out := make([]string, 0, len(stats)+1)
|
||||||
|
out = append(out, r.summary(stats))
|
||||||
|
for _, s := range stats {
|
||||||
|
out = append(out, r.tableRow(s))
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// redraw repaints the block in place: move to the top of the previous frame,
|
||||||
|
// clear it, and print the new lines clamped to the terminal width. A stray line
|
||||||
|
// from stderr that lands in the block is painted over on the next tick.
|
||||||
|
func (r *Reporter) redraw(lines []string, final bool) {
|
||||||
|
if r.prevLines > 0 {
|
||||||
|
if r.prevLines > 1 {
|
||||||
|
fmt.Fprintf(r.w, "\x1b[%dA", r.prevLines-1) // up to the first line
|
||||||
|
}
|
||||||
|
fmt.Fprint(r.w, "\r\x1b[J") // column 0, clear to end of screen
|
||||||
|
}
|
||||||
|
for i, ln := range lines {
|
||||||
|
if i > 0 {
|
||||||
|
fmt.Fprint(r.w, "\n")
|
||||||
|
}
|
||||||
|
fmt.Fprint(r.w, clampRunes(ln, r.width))
|
||||||
|
}
|
||||||
|
r.prevLines = len(lines)
|
||||||
|
if final && len(lines) > 0 {
|
||||||
|
fmt.Fprintln(r.w) // leave the cursor below the block for the OK/FAIL lines
|
||||||
|
r.prevLines = 0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// summary is the header/aggregate line: the active count and total down/up
|
||||||
|
// speed, plus a stalled count when any download has gone quiet.
|
||||||
|
func (r *Reporter) summary(stats []download.Stat) string {
|
||||||
|
if len(stats) == 0 {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
var totDL, totUL int64
|
var totDL, totUL int64
|
||||||
|
stalled := 0
|
||||||
for _, s := range stats {
|
for _, s := range stats {
|
||||||
dl, ul := r.rates(s)
|
dl, ul := r.rates(s)
|
||||||
totDL += dl
|
totDL += dl
|
||||||
totUL += ul
|
totUL += ul
|
||||||
}
|
if r.isStalled(s, dl) {
|
||||||
var b strings.Builder
|
stalled++
|
||||||
fmt.Fprintf(&b, "[%d active DL:%s UL:%s]", len(stats), speed(totDL, r.human), speed(totUL, r.human))
|
|
||||||
for i, s := range stats {
|
|
||||||
if i >= 4 {
|
|
||||||
fmt.Fprintf(&b, "[+%d]", len(stats)-i)
|
|
||||||
break
|
|
||||||
}
|
|
||||||
if s.Total > 0 {
|
|
||||||
fmt.Fprintf(&b, "[%s %d%%]", shortName(s.Name), percent(s.Completed, s.Total))
|
|
||||||
} else {
|
|
||||||
fmt.Fprintf(&b, "[%s %s]", shortName(s.Name), humanSize(s.Completed, r.human))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return b.String()
|
line := fmt.Sprintf("%d active DL:%s UL:%s", len(stats), speed(totDL, r.human), speed(totUL, r.human))
|
||||||
|
if stalled > 0 {
|
||||||
|
line += fmt.Sprintf(" (%d stalled)", stalled)
|
||||||
|
}
|
||||||
|
return line
|
||||||
|
}
|
||||||
|
|
||||||
|
// tableRow renders one download as "name |bar| pct metric tail": the bar and
|
||||||
|
// percent for progress, then the field that answers "stalled or fast" (DL speed
|
||||||
|
// or seeding) and a tail (ETA, upload rate, or a stalled flag).
|
||||||
|
func (r *Reporter) tableRow(s download.Stat) string {
|
||||||
|
dl, ul := r.rates(s)
|
||||||
|
metric, tail := r.rowMetric(s, dl, ul)
|
||||||
|
return fmt.Sprintf(" %s %s %s %-9s %s", padRune(shortName(s.Name), nameW), barOf(s), pctField(s), metric, tail)
|
||||||
|
}
|
||||||
|
|
||||||
|
// rowMetric returns a table row's two trailing fields for the download's state:
|
||||||
|
// the primary metric (DL speed, "seeding", "done", "failed") and a tail (ETA,
|
||||||
|
// upload rate, or a "stalled" flag).
|
||||||
|
func (r *Reporter) rowMetric(s download.Stat, dl, ul int64) (metric, tail string) {
|
||||||
|
switch s.Status {
|
||||||
|
case download.Seeding:
|
||||||
|
return "seeding", "UL:" + speed(ul, r.human)
|
||||||
|
case download.Complete:
|
||||||
|
return "done", ""
|
||||||
|
case download.Errored:
|
||||||
|
return "failed", ""
|
||||||
|
default:
|
||||||
|
metric = "DL:" + speed(dl, r.human)
|
||||||
|
switch {
|
||||||
|
case dl > 0 && s.Total > 0:
|
||||||
|
eta := time.Duration(float64(s.Total-s.Completed)/float64(dl)) * time.Second
|
||||||
|
tail = secfmt(eta)
|
||||||
|
case r.isStalled(s, dl):
|
||||||
|
tail = "stalled"
|
||||||
|
}
|
||||||
|
return metric, tail
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// isStalled reports an active download that has gone a while with no new bytes,
|
||||||
|
// so the table can flag it rather than leave the eye to read DL:0B. A download
|
||||||
|
// too young to have a rate yet is not stalled.
|
||||||
|
func (r *Reporter) isStalled(s download.Stat, dl int64) bool {
|
||||||
|
if s.Status != download.Active || dl > 0 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
w := r.win[s.ID]
|
||||||
|
if w == nil || len(w.samples) < 2 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return w.samples[len(w.samples)-1].t.Sub(w.samples[0].t) >= 4*time.Second
|
||||||
|
}
|
||||||
|
|
||||||
|
// barOf renders a fixed-width progress bar. A seeding or complete download is
|
||||||
|
// full; an unknown total (no Content-Length, or pre-metadata) shows empty.
|
||||||
|
func barOf(s download.Stat) string {
|
||||||
|
const cells = 12
|
||||||
|
k := 0
|
||||||
|
switch {
|
||||||
|
case s.Status == download.Seeding || s.Status == download.Complete:
|
||||||
|
k = cells
|
||||||
|
case s.Total > 0:
|
||||||
|
k = int(int64(cells) * s.Completed / s.Total)
|
||||||
|
if k > cells {
|
||||||
|
k = cells
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return "|" + strings.Repeat("#", k) + strings.Repeat("-", cells-k) + "|"
|
||||||
|
}
|
||||||
|
|
||||||
|
// pctField is the percent column: a number, full for seeding/complete, or "--"
|
||||||
|
// when the total is unknown.
|
||||||
|
func pctField(s download.Stat) string {
|
||||||
|
switch {
|
||||||
|
case s.Status == download.Seeding || s.Status == download.Complete:
|
||||||
|
return "100%"
|
||||||
|
case s.Total > 0:
|
||||||
|
return fmt.Sprintf("%3d%%", percent(s.Completed, s.Total))
|
||||||
|
default:
|
||||||
|
return " --"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// padRune right-pads s with spaces to w runes (it is never longer, having been
|
||||||
|
// through shortName) so table columns line up on rune count.
|
||||||
|
func padRune(s string, w int) string {
|
||||||
|
if n := utf8.RuneCountInString(s); n < w {
|
||||||
|
return s + strings.Repeat(" ", w-n)
|
||||||
|
}
|
||||||
|
return s
|
||||||
}
|
}
|
||||||
|
|
||||||
// add records a sample and drops any that have fallen out of the window. A
|
// add records a sample and drops any that have fallen out of the window. A
|
||||||
@@ -272,3 +401,13 @@ func termWidth(prev int) int {
|
|||||||
}
|
}
|
||||||
return 80
|
return 80
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func termHeight(prev int) int {
|
||||||
|
if _, h, err := term.GetSize(int(os.Stdout.Fd())); err == nil && h > 0 {
|
||||||
|
return h
|
||||||
|
}
|
||||||
|
if prev > 0 {
|
||||||
|
return prev
|
||||||
|
}
|
||||||
|
return 24
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package progress
|
package progress
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -112,6 +113,100 @@ func TestLineOneCNSD(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestBarOf(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
s download.Stat
|
||||||
|
want string
|
||||||
|
}{
|
||||||
|
{"empty", download.Stat{Status: download.Active, Total: 100, Completed: 0}, "|------------|"},
|
||||||
|
{"half", download.Stat{Status: download.Active, Total: 100, Completed: 50}, "|######------|"},
|
||||||
|
{"full by complete", download.Stat{Status: download.Complete}, "|############|"},
|
||||||
|
{"full by seeding", download.Stat{Status: download.Seeding}, "|############|"},
|
||||||
|
{"unknown total is empty", download.Stat{Status: download.Active, Total: -1, Completed: 9}, "|------------|"},
|
||||||
|
}
|
||||||
|
for _, tc := range tests {
|
||||||
|
if got := barOf(tc.s); got != tc.want {
|
||||||
|
t.Errorf("%s: barOf = %q, want %q", tc.name, got, tc.want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPctField(t *testing.T) {
|
||||||
|
if got := pctField(download.Stat{Status: download.Active, Total: 100, Completed: 50}); got != " 50%" {
|
||||||
|
t.Errorf("active 50%% = %q, want ' 50%%'", got)
|
||||||
|
}
|
||||||
|
if got := pctField(download.Stat{Status: download.Seeding}); got != "100%" {
|
||||||
|
t.Errorf("seeding = %q, want 100%%", got)
|
||||||
|
}
|
||||||
|
if got := pctField(download.Stat{Status: download.Active, Total: -1}); got != " --" {
|
||||||
|
t.Errorf("unknown total = %q, want ' --'", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRowMetric: the two trailing fields per download state.
|
||||||
|
func TestRowMetric(t *testing.T) {
|
||||||
|
r := newReporter(true)
|
||||||
|
if m, tail := r.rowMetric(download.Stat{Status: download.Active, Total: 100, Completed: 50}, 10, 0); !strings.HasPrefix(m, "DL:") || tail == "" {
|
||||||
|
t.Errorf("active downloading: metric=%q tail=%q, want DL:… + an ETA", m, tail)
|
||||||
|
}
|
||||||
|
if m, tail := r.rowMetric(download.Stat{Status: download.Seeding}, 0, 128<<10); m != "seeding" || !strings.HasPrefix(tail, "UL:") {
|
||||||
|
t.Errorf("seeding: metric=%q tail=%q, want seeding + UL:…", m, tail)
|
||||||
|
}
|
||||||
|
if m, _ := r.rowMetric(download.Stat{Status: download.Complete}, 0, 0); m != "done" {
|
||||||
|
t.Errorf("complete metric=%q, want done", m)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestBlockTableAndFallback: several downloads render as header+rows when they
|
||||||
|
// fit, and collapse to a single summary line when the window is too short.
|
||||||
|
func TestBlockTableAndFallback(t *testing.T) {
|
||||||
|
r := newReporter(true)
|
||||||
|
stats := []download.Stat{
|
||||||
|
{ID: "a", Name: "a", Status: download.Active, Total: 100, Completed: 50},
|
||||||
|
{ID: "b", Name: "b", Status: download.Active, Total: 100, Completed: 10},
|
||||||
|
}
|
||||||
|
r.height = 24
|
||||||
|
if got := r.block(stats); len(got) != 3 { // header + 2 rows
|
||||||
|
t.Errorf("table block = %d lines, want 3:\n%v", len(got), got)
|
||||||
|
}
|
||||||
|
r.height = 2 // header + 2 rows = 3 > 2
|
||||||
|
if got := r.block(stats); len(got) != 1 {
|
||||||
|
t.Errorf("overflow block = %d lines, want 1 summary:\n%v", len(got), got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestRenderRedraw: the TTY path draws the block, then on the next frame moves
|
||||||
|
// the cursor up over the previous frame and clears it before repainting.
|
||||||
|
func TestRenderRedraw(t *testing.T) {
|
||||||
|
snap := []download.Stat{
|
||||||
|
{ID: "a", Name: "alpha", IsBT: true, Status: download.Active, Total: 100, Completed: 50},
|
||||||
|
{ID: "b", Name: "beta", IsBT: true, Status: download.Seeding, Total: 100, Completed: 100, Uploaded: 200},
|
||||||
|
}
|
||||||
|
var buf bytes.Buffer
|
||||||
|
r := &Reporter{
|
||||||
|
snap: func() []download.Stat { return snap },
|
||||||
|
human: true, w: &buf, tty: true, width: 200, height: 24,
|
||||||
|
win: map[string]*speedWindow{},
|
||||||
|
}
|
||||||
|
r.render(false)
|
||||||
|
first := buf.String()
|
||||||
|
for _, want := range []string{"2 active", "alpha", "seeding", "|"} {
|
||||||
|
if !strings.Contains(first, want) {
|
||||||
|
t.Fatalf("first frame missing %q:\n%q", want, first)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
buf.Reset()
|
||||||
|
r.render(false)
|
||||||
|
second := buf.String()
|
||||||
|
if !strings.Contains(second, "\x1b[2A") { // up prevLines-1 = 3-1
|
||||||
|
t.Errorf("second frame should move cursor up 2 lines:\n%q", second)
|
||||||
|
}
|
||||||
|
if !strings.Contains(second, "\x1b[J") {
|
||||||
|
t.Errorf("second frame should clear to end of screen:\n%q", second)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestShortNameRune ensures CJK names are truncated on runes, not bytes.
|
// TestShortNameRune ensures CJK names are truncated on runes, not bytes.
|
||||||
func TestShortNameRune(t *testing.T) {
|
func TestShortNameRune(t *testing.T) {
|
||||||
name := strings.Repeat("あ", 30) // 30 runes, 90 bytes
|
name := strings.Repeat("あ", 30) // 30 runes, 90 bytes
|
||||||
|
|||||||
Reference in New Issue
Block a user