From 39202b44a19ef37178ee3b90cc891228d1f2bcef Mon Sep 17 00:00:00 2001 From: Levi Date: Fri, 20 Oct 2023 12:15:21 +0800 Subject: [PATCH] fix: http retry strategy (#240) --- .github/workflows/build.yml | 2 +- internal/protocol/http/fetcher.go | 64 +++++++++++++++------ internal/protocol/http/fetcher_test.go | 21 +++++-- internal/test/httptest.go | 78 +++++++++++++++++++++++++- 4 files changed, 139 insertions(+), 26 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 846cd0f0..a10e7a9c 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -44,7 +44,7 @@ jobs: go build -tags nosqlite -ldflags="-w -s" -buildmode=c-shared -o ui/flutter/windows/libgopeed.dll github.com/GopeedLab/gopeed/bind/desktop cd ui/flutter flutter build windows - $mingw = "C:\ProgramData\Chocolatey\lib\mingw\tools\install\mingw64\bin" + $mingw = "C:\Program Files\Git\mingw64\bin" $system = "C:\Windows\System32" $release = "build\windows\runner\Release\" cp $mingw\libstdc++-6.dll $release diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index acf10403..03f28b4b 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -18,6 +18,7 @@ import ( "os" "path" "strconv" + "time" ) type RequestError struct { @@ -37,6 +38,8 @@ type chunk struct { Begin int64 End int64 Downloaded int64 + + retryTimes int } func newChunk(begin int64, end int64) *chunk { @@ -255,24 +258,43 @@ func (f *Fetcher) fetch() { func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) { chunk := f.chunks[index] + chunk.retryTimes = 0 httpReq, err := buildRequest(ctx, f.meta.Req) if err != nil { return err } var ( - client = buildClient() - buf = make([]byte, 8192) + client = buildClient() + buf = make([]byte, 8192) + maxRetries = 3 ) - // 重试5次 - for i := 0; i < 5; i++ { + // retry until all remain chunks failed + for true { // if chunk is completed, return if f.meta.Res.Range && chunk.Downloaded >= chunk.End-chunk.Begin+1 { return } + + if chunk.retryTimes >= maxRetries { + if !f.meta.Res.Range { + return + } + // check if all failed + allFailed := true + for _, c := range f.chunks { + if chunk.Downloaded < chunk.End-chunk.Begin+1 && c.retryTimes < maxRetries { + allFailed = false + break + } + } + if allFailed { + return + } + } + var ( - resp *http.Response - retry bool + resp *http.Response ) if f.meta.Res.Range { httpReq.Header.Set(base.HttpHeaderRange, @@ -292,35 +314,41 @@ func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) { return nil }() if err != nil { - // retry request + // retry request after 1 second + chunk.retryTimes = chunk.retryTimes + 1 + time.Sleep(time.Second) continue } - // 请求成功就重置错误次数，连续失败5次才终止 - i = 0 - retry, err = func() (bool, error) { + // Http request success, reset retry times + chunk.retryTimes = 0 + err = func() error { defer resp.Body.Close() for { n, err := resp.Body.Read(buf) if n > 0 { _, err := f.file.WriteAt(buf[:n], chunk.Begin+chunk.Downloaded) if err != nil { - return false, err + return err } chunk.Downloaded += int64(n) } if err != nil { - if err != io.EOF { - return true, err + if err == io.EOF { + return nil } - break + return err } } - return false, nil }() - if !retry { - // 下载成功，跳出重试 - break + // if chunk is completed, reset other not complete chunks retry times + if err == nil { + for _, c := range f.chunks { + if c != chunk && c.retryTimes > 0 { + c.retryTimes = 0 + } + } } + break } return } diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index 7704f0fb..f0408f0e 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -76,7 +76,7 @@ func testResolve(startTestServer func() net.Listener, path string, want *base.Re func TestFetcher_DownloadNormal(t *testing.T) { listener := test.StartTestFileServer() defer listener.Close() - // 正常下载 + downloadNormal(listener, 1, t) downloadNormal(listener, 5, t) downloadNormal(listener, 8, t) @@ -86,7 +86,7 @@ func TestFetcher_DownloadNormal(t *testing.T) { func TestFetcher_DownloadContinue(t *testing.T) { listener := test.StartTestFileServer() defer listener.Close() - // 暂停继续 + downloadContinue(listener, 1, t) downloadContinue(listener, 5, t) downloadContinue(listener, 8, t) @@ -96,7 +96,7 @@ func TestFetcher_DownloadContinue(t *testing.T) { func TestFetcher_DownloadChunked(t *testing.T) { listener := test.StartTestCustomServer() defer listener.Close() - // chunked编码下载 + downloadNormal(listener, 1, t) downloadNormal(listener, 2, t) } @@ -104,24 +104,33 @@ func TestFetcher_DownloadChunked(t *testing.T) { func TestFetcher_DownloadPost(t *testing.T) { listener := test.StartTestPostServer() defer listener.Close() - // post下载 + downloadPost(listener, 1, t) } func TestFetcher_DownloadRetry(t *testing.T) { listener := test.StartTestRetryServer() defer listener.Close() - // chunked编码下载 + downloadNormal(listener, 1, t) } func TestFetcher_DownloadError(t *testing.T) { listener := test.StartTestErrorServer() defer listener.Close() - // chunked编码下载 + downloadError(listener, 1, t) } +func TestFetcher_DownloadLimit(t *testing.T) { + listener := test.StartTestLimitServer() + defer listener.Close() + + downloadNormal(listener, 1, t) + downloadNormal(listener, 2, t) + downloadNormal(listener, 8, 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 07fb0804..e00f6c3d 100644 --- a/internal/test/httptest.go +++ b/internal/test/httptest.go @@ -9,6 +9,9 @@ import ( "net" "net/http" "os" + "strconv" + "strings" + "sync/atomic" "time" ) @@ -81,7 +84,7 @@ func StartTestRetryServer() net.Listener { mux := http.NewServeMux() mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) { counter++ - if counter != 1 && counter < 5 { + if counter != 1 && counter < 2 { writer.WriteHeader(500) return } @@ -134,6 +137,79 @@ func StartTestErrorServer() net.Listener { }) } +// StartTestLimitServer connections limit server +func StartTestLimitServer() net.Listener { + var connections atomic.Int32 + + return startTestServer(func() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) { + defer func() { + connections.Add(-1) + }() + connections.Add(1) + if connections.Load() > 4 { + writer.WriteHeader(403) + return + } + + r := request.Header.Get("Range") + if r == "" { + writer.Header().Set("Content-Length", fmt.Sprintf("%d", BuildSize)) + writer.WriteHeader(200) + file, err := os.Open(BuildFile) + if err != nil { + panic(err) + } + defer file.Close() + io.Copy(writer, file) + } 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) + file, err := os.Open(BuildFile) + if err != nil { + writer.WriteHeader(500) + return + } + defer file.Close() + file.Seek(start, 0) + io.CopyN(writer, file, end-start+1) + } + }) + return mux + }) +} + func startTestServer(serverHandle func() http.Handler) net.Listener { listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil {