diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 10ae6974..c4378960 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -53,7 +53,7 @@ jobs: with: go-version: ${{ env.GO_VERSION }} - name: Run Unit tests. - run: go test $(go list ./... | grep -v /cmd/ | grep -v /bind/ ) -v -covermode=count -coverprofile=coverage.txt + run: go test $(go list ./... | grep -v /cmd/ | grep -v /bind/) -v -covermode=count -coverprofile=coverage.txt - uses: codecov/codecov-action@v4 with: files: ./coverage.txt diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 3c5e34a3..73e43c2a 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -51,6 +51,11 @@ type chunk struct { retryTimes int } +// get remain to download bytes +func (c *chunk) remain() int64 { + return c.End - c.Begin + 1 - c.Downloaded +} + func newChunk(begin int64, end int64) *chunk { return &chunk{ Begin: begin, @@ -319,7 +324,7 @@ func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) { // retry until all remain chunks failed for { // if chunk is completed, return - if f.meta.Res.Range && chunk.Downloaded >= chunk.End-chunk.Begin+1 { + if f.meta.Res.Range && chunk.remain() <= 0 { return nil } // if all chunks failed, return @@ -364,15 +369,30 @@ func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) { err = NewRequestError(resp.StatusCode, resp.Status) return err } + 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 := f.file.WriteAt(buf[:n], chunk.Begin+chunk.Downloaded) if err != nil { return err } chunk.Downloaded += int64(n) + + if finished { + return nil + } } if err != nil { if err == io.EOF { diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index e8673c55..89a43d4b 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -176,6 +176,14 @@ func TestFetcher_DownloadResponseBodyReadTimeout(t *testing.T) { downloadError(listener, 4, t) } +func TestFetcher_DownloadOnBugFileServer(t *testing.T) { + listener := test.StartTestRangeBugServer() + defer listener.Close() + + downloadNormal(listener, 1, t) + downloadNormal(listener, 4, t) +} + func TestFetcher_DownloadResume(t *testing.T) { listener := test.StartTestFileServer() defer listener.Close() diff --git a/internal/test/httptest.go b/internal/test/httptest.go index 85696748..54dec46e 100644 --- a/internal/test/httptest.go +++ b/internal/test/httptest.go @@ -180,68 +180,98 @@ func StartTestLimitServer(maxConnections int32, delay int64) net.Listener { writer.WriteHeader(403) return } - - r := request.Header.Get("Range") - if r == "" { - writer.Header().Set("Content-Length", fmt.Sprintf("%d", BuildSize)) - writer.WriteHeader(200) - (writer.(http.Flusher)).Flush() - - file, err := os.Open(BuildFile) - if err != nil { - panic(err) - } - defer file.Close() - slowCopy(sl, writer, file, delay) - } else { - // split range - s := strings.Split(r, "=") - if len(s) != 2 { - writer.WriteHeader(400) - return - } - s = strings.Split(s[1], "-") - if len(s) != 2 { - writer.WriteHeader(400) - return - } - start, err := strconv.ParseInt(s[0], 10, 64) - if err != nil { - writer.WriteHeader(400) - return - } - end, err := strconv.ParseInt(s[1], 10, 64) - if err != nil { - writer.WriteHeader(400) - return - } - if start < 0 || end < 0 || start > end { - writer.WriteHeader(400) - return - } - if end >= BuildSize { - end = BuildSize - 1 - } - writer.Header().Set("Content-Length", fmt.Sprintf("%d", end-start+1)) - writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, BuildSize)) - writer.Header().Set("Accept-Ranges", "bytes") - writer.WriteHeader(206) - (writer.(http.Flusher)).Flush() - - file, err := os.Open(BuildFile) - if err != nil { - writer.WriteHeader(500) - return - } - defer file.Close() - file.Seek(start, 0) - slowCopyN(sl, writer, file, end-start+1, delay) - } + rangeFileHandle( + writer, + request, + nil, + func(file *os.File, n int64) { + slowCopyN(sl, writer, file, n, delay) + }, + ) }) return mux }) } +// StartTestRangeBugServer simulate bug server: +// Don't follow Range request rules, always return more data than range, e.g. Range: bytes=0-100, return 150 bytes +func StartTestRangeBugServer() net.Listener { + return startTestServer(func(sl *shutdownListener) http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) { + rangeFileHandle( + writer, + request, + func(end int64) int64 { + var bugEnd = end + if end != 0 { + bugEnd = end + 50 + if bugEnd >= BuildSize { + bugEnd = end + } + } + return bugEnd + }, + func(file *os.File, n int64) { + io.CopyN(writer, file, n) + }, + ) + }) + return mux + }) +} + +func rangeFileHandle(writer http.ResponseWriter, request *http.Request, modifyEnd func(end int64) int64, iocpN func(file *os.File, n int64)) { + r := request.Header.Get("Range") + // split range + s := strings.Split(r, "=") + if len(s) != 2 { + writer.WriteHeader(400) + return + } + s = strings.Split(s[1], "-") + if len(s) != 2 { + writer.WriteHeader(400) + return + } + start, err := strconv.ParseInt(s[0], 10, 64) + if err != nil { + writer.WriteHeader(400) + return + } + end, err := strconv.ParseInt(s[1], 10, 64) + if err != nil { + writer.WriteHeader(400) + return + } + if start < 0 || end < 0 || start > end { + writer.WriteHeader(400) + return + } + if end >= BuildSize { + end = BuildSize - 1 + } + + if modifyEnd != nil { + end = modifyEnd(end) + } + + writer.Header().Set("Content-Length", fmt.Sprintf("%d", end-start+1)) + writer.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, BuildSize)) + writer.Header().Set("Accept-Ranges", "bytes") + writer.WriteHeader(206) + (writer.(http.Flusher)).Flush() + + file, err := os.Open(BuildFile) + if err != nil { + writer.WriteHeader(500) + return + } + defer file.Close() + file.Seek(start, 0) + iocpN(file, end-start+1) +} + // slowCopyN copies n bytes from src to dst, speed limit is bytes per second func slowCopy(sl *shutdownListener, dst io.Writer, src io.Reader, delay int64) (written int64, err error) { buf := make([]byte, 32*1024) diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 844396b8..c8c00e63 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -208,7 +208,7 @@ func (d *Downloader) Setup() error { if task.Status == base.DownloadStatusRunning { downloadDataChanged = current != task.Progress.Downloaded task.Progress.Used = task.timer.Used() - task.Progress.Speed = task.calcSpeed(task.speedArr, current-task.Progress.Downloaded, tick) + task.Progress.Speed = task.updateSpeed(current-task.Progress.Downloaded, tick) task.Progress.Downloaded = current } @@ -217,7 +217,7 @@ func (d *Downloader) Setup() error { uploader := task.fetcher.(fetcher.Uploader) currentUploaded := uploader.UploadedBytes() uploadDataChanged = currentUploaded != task.Progress.Uploaded - task.Progress.UploadSpeed = task.calcSpeed(task.uploadSpeedArr, currentUploaded-task.Progress.Uploaded, tick) + task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, tick) task.Progress.Uploaded = currentUploaded } d.emit(EventKeyProgress, task) diff --git a/pkg/download/model.go b/pkg/download/model.go index 085a20f0..3da050ca 100644 --- a/pkg/download/model.go +++ b/pkg/download/model.go @@ -100,16 +100,27 @@ func (t *Task) clone() *Task { return util.DeepClone(t) } -func (t *Task) calcSpeed(speedArr []int64, downloaded int64, usedTime float64) int64 { - speedArr = append(speedArr, downloaded) - if len(speedArr) > 6 { - speedArr = speedArr[1:] +func (t *Task) updateSpeed(downloaded int64, usedTime float64) int64 { + return calcSpeed(&t.speedArr, downloaded, usedTime) +} + +func (t *Task) updateUploadSpeed(downloaded int64, usedTime float64) int64 { + return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime) +} + +func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 { + *speedArr = append(*speedArr, downloaded) + // Record last 5 seconds of download speed to calculate the average speed + if len(*speedArr) > int(5.0/usedTime) { + *speedArr = (*speedArr)[1:] } + var total int64 - for _, v := range speedArr { + for _, v := range *speedArr { total += v } - return int64(float64(total) / float64(len(speedArr)) / usedTime) + + return int64(float64(total) / float64(len(*speedArr)) / usedTime) } type TaskFilter struct {