diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 73e43c2a..d464534e 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -16,6 +16,7 @@ import ( "path" "strconv" "strings" + "sync" "time" "github.com/GopeedLab/gopeed/internal/controller" @@ -26,8 +27,11 @@ import ( "golang.org/x/sync/errgroup" ) -const connectTimeout = 15 * time.Second -const readTimeout = 15 * time.Second +const ( + connectTimeout = 15 * time.Second + readTimeout = 15 * time.Second + helpMinSize = 1 * 1024 * 1024 +) type RequestError struct { Code int @@ -46,6 +50,12 @@ type chunk struct { Begin int64 End int64 Downloaded int64 +} + +type connection struct { + Chunk *chunk + Downloaded int64 + Completed bool failed bool retryTimes int @@ -68,8 +78,11 @@ type Fetcher struct { config *config doneCh chan error - meta *fetcher.FetcherMeta - chunks []*chunk + meta *fetcher.FetcherMeta + connections []*connection + helpLock sync.Mutex + redirectURL string + redirectLock sync.Mutex file *os.File cancel context.CancelFunc @@ -222,9 +235,10 @@ func (f *Fetcher) Start() (err error) { return err } - if f.chunks == nil { - f.chunks = f.splitChunk() + if f.connections == nil { + f.connections = f.splitConnection() } + f.redirectURL = "" f.fetch() return } @@ -251,16 +265,26 @@ func (f *Fetcher) Meta() *fetcher.FetcherMeta { } func (f *Fetcher) Stats() any { - // todo http download health - return &fhttp.Stats{} + statsConnections := make([]*fhttp.StatsConnection, 0) + for _, connection := range f.connections { + statsConnections = append(statsConnections, &fhttp.StatsConnection{ + Downloaded: connection.Downloaded, + Completed: connection.Completed, + Failed: connection.failed, + RetryTimes: connection.retryTimes, + }) + } + return &fhttp.Stats{ + Connections: statsConnections, + } } func (f *Fetcher) Progress() fetcher.Progress { p := make(fetcher.Progress, 0) - if len(f.chunks) > 0 { + if len(f.connections) > 0 { total := int64(0) - for _, chunk := range f.chunks { - total += chunk.Downloaded + for _, connection := range f.connections { + total += connection.Downloaded } p = append(p, total) } @@ -275,16 +299,16 @@ func (f *Fetcher) fetch() { var ctx context.Context ctx, f.cancel = context.WithCancel(context.Background()) f.eg, _ = errgroup.WithContext(ctx) - chunkErrs := make([]error, len(f.chunks)) - for i := 0; i < len(f.chunks); i++ { + connectionErrs := make([]error, len(f.connections)) + for i := 0; i < len(f.connections); i++ { i := i f.eg.Go(func() error { - err := f.fetchChunk(i, ctx) + err := f.run(i, ctx) // if canceled, fail fast if errors.Is(err, context.Canceled) { return err } - chunkErrs[i] = err + connectionErrs[i] = err return nil }) } @@ -296,7 +320,7 @@ func (f *Fetcher) fetch() { return } // check all fetch results, if any error, return - for _, chunkErr := range chunkErrs { + for _, chunkErr := range connectionErrs { if chunkErr != nil { err = chunkErr break @@ -312,122 +336,192 @@ func (f *Fetcher) fetch() { }() } -func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) { - chunk := f.chunks[index] - chunk.failed = false - chunk.retryTimes = 0 - +func (f *Fetcher) run(index int, ctx context.Context) (err error) { + connection := f.connections[index] + connection.failed = false + connection.retryTimes = 0 var ( client = f.buildClient() buf = make([]byte, 8192) ) - // retry until all remain chunks failed - for { - // if chunk is completed, return - if f.meta.Res.Range && chunk.remain() <= 0 { - return nil - } - // if all chunks failed, return - if chunk.failed { - allFailed := true - for _, c := range f.chunks { - if !c.failed { - allFailed = false - break - } - } - if allFailed { - if chunk.retryTimes >= 3 { - return - } else { - chunk.retryTimes++ - } - } - } - var ( - httpReq *http.Request - resp *http.Response - ) - httpReq, err = f.buildRequest(ctx, f.meta.Req) - if err != nil { - return - } - if f.meta.Res.Range { - httpReq.Header.Set(base.HttpHeaderRange, - fmt.Sprintf(base.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End)) - } else { - chunk.Downloaded = 0 - } - err = func() error { - resp, err = client.Do(httpReq) - if err != nil { - return err + downloadChunk := func(chunk *chunk) (err error) { + // retry until all remain chunks failed + for { + // if chunk is completed, return + if f.meta.Res.Range && chunk.remain() <= 0 { + return nil } - defer resp.Body.Close() - if resp.StatusCode != base.HttpCodeOK && resp.StatusCode != base.HttpCodePartialContent { - err = NewRequestError(resp.StatusCode, resp.Status) - return err + // if all chunks failed, return + if connection.failed { + allFailed := true + for _, c := range f.connections { + if c.Completed { + continue + } + if !c.failed { + allFailed = false + break + } + } + if allFailed { + if connection.retryTimes >= 3 { + return + } else { + connection.retryTimes++ + } + } } - chunk.failed = false - reader := NewTimeoutReader(resp.Body, readTimeout) - for { - n, err := reader.Read(buf) - if n > 0 { - finished := false - if f.meta.Res.Range { - remain := chunk.remain() - // If downloaded bytes exceed the remain bytes, only write remain bytes - if remain < int64(n) { - n = int(remain) - finished = true + + err = func() error { + var ( + httpReq *http.Request + resp *http.Response + ) + f.redirectLock.Lock() + if f.redirectURL != "" { + f.redirectLock.Unlock() + } + err = func() (err error) { + defer func() { + if f.redirectURL == "" { + if err == nil { + f.redirectURL = resp.Request.URL.String() + } + f.redirectLock.Unlock() } - } + }() - _, err := f.file.WriteAt(buf[:n], chunk.Begin+chunk.Downloaded) + httpReq, err = f.buildRequest(ctx, f.meta.Req) if err != nil { - return err + return } - chunk.Downloaded += int64(n) - - if finished { - return nil + if f.meta.Res.Range { + httpReq.Header.Set(base.HttpHeaderRange, + fmt.Sprintf(base.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End)) + } else { + chunk.Downloaded = 0 } - } + resp, err = client.Do(httpReq) + if err != nil { + return + } + return + }() if err != nil { - if err == io.EOF { - return nil - } return err } + + defer resp.Body.Close() + if resp.StatusCode != base.HttpCodeOK && resp.StatusCode != base.HttpCodePartialContent { + err = NewRequestError(resp.StatusCode, resp.Status) + return err + } + connection.failed = false + reader := NewTimeoutReader(resp.Body, readTimeout) + for { + n, err := reader.Read(buf) + if n > 0 { + finished := false + if f.meta.Res.Range { + remain := chunk.remain() + // If downloaded bytes exceed the remain bytes, only write remain bytes + if remain < int64(n) { + n = int(remain) + finished = true + } + } + + _, err := f.file.WriteAt(buf[:n], chunk.Begin+chunk.Downloaded) + if err != nil { + return err + } + chunk.Downloaded += int64(n) + connection.Downloaded += int64(n) + + if finished { + return nil + } + } + if err != nil { + if err == io.EOF { + return nil + } + return err + } + } + }() + if err != nil { + // If canceled, do not retry + if errors.Is(err, context.Canceled) { + return + } + // retry request after 1 second + connection.failed = true + time.Sleep(time.Second) + continue } - }() - if err != nil { - // If canceled, do not retry - if errors.Is(err, context.Canceled) { - return - } - // retry request after 1 second - chunk.failed = true - time.Sleep(time.Second) + break + } + return + } + + for { + if err = downloadChunk(connection.Chunk); err != nil { + return + } + + // check this connection is completed + if !f.meta.Res.Range || !f.helpOtherConnection(connection) { + connection.Completed = true + return + } + } +} + +func (f *Fetcher) helpOtherConnection(helper *connection) bool { + f.helpLock.Lock() + defer f.helpLock.Unlock() + + // find the slowest connection + var maxRemainConnection *connection + var maxRemain int64 + for _, r := range f.connections { + if r == helper || r.Completed { continue } - break + + remain := r.Chunk.remain() + if remain > maxRemain && remain > helpMinSize { + maxRemainConnection = r + maxRemain = remain + } } - return + + if maxRemainConnection == nil { + return false + } + + // re-calculate the chunk range + helper.Chunk.Begin = maxRemainConnection.Chunk.End - maxRemainConnection.Chunk.remain()/2 + helper.Chunk.End = maxRemainConnection.Chunk.End + helper.Chunk.Downloaded = 0 + maxRemainConnection.Chunk.End = helper.Chunk.Begin - 1 + return true } func (f *Fetcher) buildRequest(ctx context.Context, req *base.Request) (httpReq *http.Request, err error) { - url, err := url.Parse(req.URL) - if err != nil { - return + var reqUrl string + if f.redirectURL != "" { + reqUrl = f.redirectURL + } else { + reqUrl = req.URL } var ( method string body io.Reader ) - headers := http.Header{} if req.Extra == nil { method = http.MethodGet @@ -452,9 +546,9 @@ func (f *Fetcher) buildRequest(ctx context.Context, req *base.Request) (httpReq } if ctx != nil { - httpReq, err = http.NewRequestWithContext(ctx, method, url.String(), body) + httpReq, err = http.NewRequestWithContext(ctx, method, reqUrl, body) } else { - httpReq, err = http.NewRequest(method, url.String(), body) + httpReq, err = http.NewRequest(method, reqUrl, body) } if err != nil { return @@ -467,30 +561,33 @@ func (f *Fetcher) buildRequest(ctx context.Context, req *base.Request) (httpReq return httpReq, nil } -func (f *Fetcher) splitChunk() (chunks []*chunk) { +func (f *Fetcher) splitConnection() (connections []*connection) { if f.meta.Res.Range { - connections := f.meta.Opts.Extra.(*fhttp.OptsExtra).Connections + optConnections := f.meta.Opts.Extra.(*fhttp.OptsExtra).Connections // 每个连接平均需要下载的分块大小 - chunkSize := f.meta.Res.Size / int64(connections) - chunks = make([]*chunk, connections) - for i := 0; i < connections; i++ { + chunkSize := f.meta.Res.Size / int64(optConnections) + connections = make([]*connection, optConnections) + for i := 0; i < optConnections; i++ { var ( begin = chunkSize * int64(i) end int64 ) - if i == connections-1 { + if i == optConnections-1 { // 最后一个分块需要保证把文件下载完 end = f.meta.Res.Size - 1 } else { end = begin + chunkSize - 1 } - chunk := newChunk(begin, end) - chunks[i] = chunk + connections[i] = &connection{ + Chunk: newChunk(begin, end), + } } } else { // 只支持单连接下载 - chunks = make([]*chunk, 1) - chunks[0] = newChunk(0, 0) + connections = make([]*connection, 1) + connections[0] = &connection{ + Chunk: newChunk(0, 0), + } } return } @@ -514,7 +611,7 @@ func (f *Fetcher) buildClient() *http.Client { } type fetcherData struct { - Chunks []*chunk + Connections []*connection } type FetcherManager struct { @@ -570,7 +667,7 @@ func (fm *FetcherManager) DefaultConfig() any { func (fm *FetcherManager) Store(f fetcher.Fetcher) (data any, err error) { _f := f.(*Fetcher) return &fetcherData{ - Chunks: _f.chunks, + Connections: _f.connections, }, nil } @@ -582,8 +679,8 @@ func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v fetcher.meta = meta base.ParseReqExtra[fhttp.ReqExtra](fetcher.meta.Req) base.ParseOptsExtra[fhttp.OptsExtra](fetcher.meta.Opts) - if len(fd.Chunks) > 0 { - fetcher.chunks = fd.Chunks + if len(fd.Connections) > 0 { + fetcher.connections = fd.Connections } return fetcher } diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index 89a43d4b..ea40f344 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -246,6 +246,33 @@ func TestFetcher_ConfigUseServerCtime(t *testing.T) { } } +func TestFetcher_Stats(t *testing.T) { + listener := test.StartTestFileServer() + defer listener.Close() + fetcher := doDownloadReady(buildConfigFetcher(config{ + Connections: 16, + }), listener, 0, t) + err := fetcher.Start() + if err != nil { + t.Fatal(err) + } + err = fetcher.Wait() + if err != nil { + t.Fatal(err) + } + stats := fetcher.Stats().(*http.Stats) + if len(stats.Connections) != 16 { + t.Errorf("Stats() got = %v, want %v", len(stats.Connections), 16) + } + totalDownloaded := int64(0) + for _, conn := range stats.Connections { + totalDownloaded += conn.Downloaded + } + if totalDownloaded != test.BuildSize { + t.Errorf("Stats() got = %v, want %v", totalDownloaded, test.BuildSize) + } +} + func TestFetcherManager_ParseName(t *testing.T) { type args struct { u string diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index c8c00e63..5ca70e62 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -148,6 +148,7 @@ func (d *Downloader) Setup() error { tasks = append(tasks[:i], tasks[i+1:]...) continue } + d.assignFetcherManager(task) initTask(task) if task.Status != base.DownloadStatusDone && task.Status != base.DownloadStatusError { task.Status = base.DownloadStatusPause @@ -892,18 +893,7 @@ func (d *Downloader) restoreTask(task *Task) error { } func (d *Downloader) restoreFetcher(task *Task) error { - var fm fetcher.FetcherManager - for _, f := range d.cfg.FetchManagers { - if f.Name() == task.Protocol { - fm = f - break - } - } - if fm == nil { - return ErrUnSupportedProtocol - } - task.fetcherManager = fm - v, f := fm.Restore() + v, f := task.fetcherManager.Restore() if v != nil { err := d.storage.Pop(bucketSave, task.ID, v) if err != nil { @@ -1145,6 +1135,15 @@ func logPanic(logDir string) { debug.SetCrashOutput(f, debug.CrashOptions{}) } +func (d *Downloader) assignFetcherManager(task *Task) error { + fm, err := d.parseFm(task.Meta.Req.URL) + if err != nil { + return err + } + task.fetcherManager = fm + return nil +} + func (d *Downloader) buildFetcher(url string) (fetcher.Fetcher, error) { fm, err := d.parseFm(url) if err != nil { diff --git a/pkg/protocol/http/model.go b/pkg/protocol/http/model.go index 0115b2e1..d642d20a 100644 --- a/pkg/protocol/http/model.go +++ b/pkg/protocol/http/model.go @@ -14,6 +14,12 @@ type OptsExtra struct { // Stats for download type Stats struct { - // http stats - // health indicators of http + Connections []*StatsConnection `json:"connections"` +} + +type StatsConnection struct { + Downloaded int64 `json:"downloaded"` + Completed bool `json:"completed"` + Failed bool `json:"failed"` + RetryTimes int `json:"retryTimes"` }