diff --git a/bind/desktop/main.go b/bind/desktop/main.go index 7054093b..b756a1dd 100644 --- a/bind/desktop/main.go +++ b/bind/desktop/main.go @@ -48,8 +48,7 @@ func Start(cfg *C.char) (int, *C.char) { //export Stop func Stop() { - nativebridge.PauseInvokesAndWait() - rest.Stop() + nativebridge.Stop() } //export GetAPIServerState diff --git a/bind/mobile/main.go b/bind/mobile/main.go index 2d4b8dae..741335c7 100644 --- a/bind/mobile/main.go +++ b/bind/mobile/main.go @@ -27,8 +27,7 @@ func Start(cfg string) (int, error) { } func Stop() { - nativebridge.PauseInvokesAndWait() - rest.Stop() + nativebridge.Stop() } func GetAPIServerState() string { diff --git a/bind/native/invoke.go b/bind/native/invoke.go index 27056551..55dc1a62 100644 --- a/bind/native/invoke.go +++ b/bind/native/invoke.go @@ -3,17 +3,18 @@ package nativebridge import ( + "errors" "fmt" - "runtime" "sync" + "time" "github.com/GopeedLab/gopeed/pkg/rest" ) -const ( - minInvokeConcurrency = 4 - maxInvokeConcurrency = 32 -) +// StopTimeout bounds native shutdown even when an in-process request never returns. +const StopTimeout = 3 * time.Second + +var errNativeBridgeStopping = errors.New("native bridge is stopping") type invokeRequest struct { method string @@ -23,11 +24,19 @@ type invokeRequest struct { } type invokeTask struct { + id uint64 request invokeRequest complete func(result string, err error) + once sync.Once } -// InvokeAsync queues an in-process API request and completes it through the +func (t *invokeTask) finish(result string, err error) { + t.once.Do(func() { + completeCallback(t.complete, result, err) + }) +} + +// InvokeAsync starts an in-process API request and completes it through the // supplied callback. Desktop and mobile adapters share this executor. func InvokeAsync(method, path, query, body string, callback func(result string, err error)) { if callback == nil { @@ -43,7 +52,7 @@ func InvokeAsync(method, path, query, body string, callback func(result string, complete: callback, } if !getInvokeExecutor().submit(task) { - completeCallback(callback, "", fmt.Errorf("native bridge is stopping")) + completeCallback(callback, "", errNativeBridgeStopping) } } @@ -52,10 +61,41 @@ func ResumeInvokes() { getInvokeExecutor().resume() } -// PauseInvokesAndWait prevents new submissions and waits for every queued or -// running request to finish before a native runtime is stopped. -func PauseInvokesAndWait() { - getInvokeExecutor().pauseAndWait() +// PauseInvokesAndWait prevents new submissions and gives running requests a +// bounded opportunity to finish before a native runtime is stopped. +// Requests still pending after StopTimeout are failed and detached so shutdown +// can continue without invoking their callbacks a second time. +func PauseInvokesAndWait() bool { + return getInvokeExecutor().pauseAndWait(StopTimeout) +} + +// Stop pauses native requests and shuts down the shared runtime, but never +// keeps the host application waiting longer than StopTimeout. Cleanup that +// cannot finish within the deadline is allowed to continue in the background. +func Stop() { + stopWithin(getInvokeExecutor(), rest.Stop, StopTimeout) +} + +func stopWithin(executor *invokeExecutor, cleanup func(), timeout time.Duration) { + deadline := time.Now().Add(timeout) + executor.pauseAndWait(time.Until(deadline)) + + stopped := make(chan struct{}) + go func() { + defer close(stopped) + cleanup() + }() + + remaining := time.Until(deadline) + if remaining <= 0 { + return + } + timer := time.NewTimer(remaining) + defer timer.Stop() + select { + case <-stopped: + case <-timer.C: + } } var ( @@ -65,64 +105,56 @@ var ( func getInvokeExecutor() *invokeExecutor { invokeExecutorOnce.Do(func() { - sharedInvoker = newDefaultInvokeExecutor(func(request invokeRequest) string { + sharedInvoker = newInvokeExecutor(func(request invokeRequest) string { return rest.Dispatch(request.method, request.path, request.query, request.body) }) }) return sharedInvoker } -// invokeExecutor keeps native API work concurrent without rejecting requests -// when every worker is busy. Additional requests wait in FIFO order and are -// completed through their callback. +// invokeExecutor starts every native API request in its own goroutine, matching +// net/http's request concurrency. It only tracks them to support bounded drain +// during shutdown; one stuck request must never prevent another from starting. type invokeExecutor struct { mu sync.Mutex - ready *sync.Cond - queue []invokeTask closed bool accepting bool - pending int - workers sync.WaitGroup + nextID uint64 + pending map[uint64]*invokeTask + drained chan struct{} + running sync.WaitGroup handler func(invokeRequest) string } -func invokeConcurrency(cpuCount int) int { - concurrency := cpuCount * 2 - if concurrency < minInvokeConcurrency { - return minInvokeConcurrency +func newInvokeExecutor(handler func(invokeRequest) string) *invokeExecutor { + drained := make(chan struct{}) + close(drained) + return &invokeExecutor{ + handler: handler, + accepting: true, + pending: make(map[uint64]*invokeTask), + drained: drained, } - if concurrency > maxInvokeConcurrency { - return maxInvokeConcurrency - } - return concurrency -} - -func newInvokeExecutor(workerCount int, handler func(invokeRequest) string) *invokeExecutor { - if workerCount < 1 { - workerCount = 1 - } - executor := &invokeExecutor{handler: handler, accepting: true} - executor.ready = sync.NewCond(&executor.mu) - executor.workers.Add(workerCount) - for range workerCount { - go executor.run() - } - return executor -} - -func newDefaultInvokeExecutor(handler func(invokeRequest) string) *invokeExecutor { - return newInvokeExecutor(invokeConcurrency(runtime.GOMAXPROCS(0)), handler) } func (e *invokeExecutor) submit(task invokeTask) bool { e.mu.Lock() - defer e.mu.Unlock() if e.closed || !e.accepting { + e.mu.Unlock() return false } - e.queue = append(e.queue, task) - e.pending++ - e.ready.Signal() + if len(e.pending) == 0 { + e.drained = make(chan struct{}) + } + taskID := e.nextID + e.nextID++ + task.id = taskID + taskPtr := &task + e.pending[taskID] = taskPtr + e.running.Add(1) + e.mu.Unlock() + + go e.run(taskPtr) return true } @@ -132,57 +164,63 @@ func (e *invokeExecutor) resume() { e.mu.Unlock() } -func (e *invokeExecutor) pauseAndWait() { +func (e *invokeExecutor) pauseAndWait(timeout time.Duration) bool { e.mu.Lock() e.accepting = false - for e.pending > 0 { - e.ready.Wait() + if len(e.pending) == 0 { + e.mu.Unlock() + return true } + drained := e.drained e.mu.Unlock() + + timer := time.NewTimer(timeout) + defer timer.Stop() + select { + case <-drained: + return true + case <-timer.C: + } + + e.mu.Lock() + if len(e.pending) == 0 { + e.mu.Unlock() + return true + } + pending := make([]*invokeTask, 0, len(e.pending)) + for _, task := range e.pending { + pending = append(pending, task) + } + e.pending = make(map[uint64]*invokeTask) + close(e.drained) + e.mu.Unlock() + + for _, task := range pending { + task.finish("", errNativeBridgeStopping) + } + return false } func (e *invokeExecutor) close() { e.mu.Lock() e.closed = true - e.ready.Broadcast() + e.accepting = false e.mu.Unlock() - e.workers.Wait() + e.running.Wait() } -func (e *invokeExecutor) run() { - defer e.workers.Done() - for { - task, ok := e.take() - if !ok { - return - } - result, err := e.execute(task.request) - completeCallback(task.complete, result, err) - e.mu.Lock() - e.pending-- - if e.pending == 0 { - e.ready.Broadcast() - } - e.mu.Unlock() - } -} - -func (e *invokeExecutor) take() (invokeTask, bool) { +func (e *invokeExecutor) run(task *invokeTask) { + defer e.running.Done() + result, err := e.execute(task.request) + task.finish(result, err) e.mu.Lock() - defer e.mu.Unlock() - for len(e.queue) == 0 && !e.closed { - e.ready.Wait() + if _, exists := e.pending[task.id]; exists { + delete(e.pending, task.id) + if len(e.pending) == 0 { + close(e.drained) + } } - if len(e.queue) == 0 { - return invokeTask{}, false - } - task := e.queue[0] - e.queue[0] = invokeTask{} - e.queue = e.queue[1:] - if len(e.queue) == 0 { - e.queue = nil - } - return task, true + e.mu.Unlock() } func (e *invokeExecutor) execute(request invokeRequest) (result string, err error) { diff --git a/bind/native/invoke_test.go b/bind/native/invoke_test.go index dc55bdb4..649bc033 100644 --- a/bind/native/invoke_test.go +++ b/bind/native/invoke_test.go @@ -4,52 +4,17 @@ import ( "runtime" "strconv" "sync" - "sync/atomic" "testing" "time" ) -func TestInvokeConcurrency(t *testing.T) { - tests := []struct { - cpus int - want int - }{ - {cpus: 1, want: 4}, - {cpus: 2, want: 4}, - {cpus: 4, want: 8}, - {cpus: 8, want: 16}, - {cpus: 16, want: 32}, - {cpus: 64, want: 32}, - } - for _, test := range tests { - t.Run(strconv.Itoa(test.cpus), func(t *testing.T) { - if got := invokeConcurrency(test.cpus); got != test.want { - t.Fatalf("invokeConcurrency(%d) = %d, want %d", test.cpus, got, test.want) - } - }) - } -} - -func TestInvokeExecutorWaitsWithoutRejecting(t *testing.T) { - const ( - workerCount = 2 - taskCount = 6 - ) +func TestInvokeExecutorStartsEveryRequestIndependently(t *testing.T) { + const taskCount = 64 release := make(chan struct{}) started := make(chan struct{}, taskCount) - var active atomic.Int32 - var maxActive atomic.Int32 - executor := newInvokeExecutor(workerCount, func(request invokeRequest) string { - current := active.Add(1) - for { - maximum := maxActive.Load() - if current <= maximum || maxActive.CompareAndSwap(maximum, current) { - break - } - } + executor := newInvokeExecutor(func(request invokeRequest) string { started <- struct{}{} <-release - active.Add(-1) return request.path }) @@ -63,20 +28,19 @@ func TestInvokeExecutorWaitsWithoutRejecting(t *testing.T) { }, }) if !accepted { - t.Fatalf("task %d was rejected while all workers were busy", i) + t.Fatalf("request %d was rejected", i) } } - for range workerCount { + // Every handler must start before any of them is released. A worker pool + // would stall here once all of its workers were occupied. + for range taskCount { select { case <-started: case <-time.After(time.Second): - t.Fatal("workers did not start queued tasks") + t.Fatal("a request waited behind other blocked requests") } } - if got := maxActive.Load(); got != workerCount { - t.Fatalf("maximum active tasks = %d, want %d", got, workerCount) - } close(release) completed.Wait() @@ -84,7 +48,7 @@ func TestInvokeExecutorWaitsWithoutRejecting(t *testing.T) { } func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) { - executor := newInvokeExecutor(1, func(invokeRequest) string { + executor := newInvokeExecutor(func(invokeRequest) string { panic("boom") }) completed := make(chan error, 1) @@ -103,7 +67,7 @@ func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) { } func TestInvokeExecutorSurvivesCallbackPanic(t *testing.T) { - executor := newInvokeExecutor(1, func(request invokeRequest) string { + executor := newInvokeExecutor(func(request invokeRequest) string { return request.path }) executor.submit(invokeTask{complete: func(string, error) { @@ -131,7 +95,7 @@ func TestInvokeExecutorSurvivesCallbackPanic(t *testing.T) { func TestInvokeExecutorPausesAndDrains(t *testing.T) { release := make(chan struct{}) started := make(chan struct{}) - executor := newInvokeExecutor(1, func(request invokeRequest) string { + executor := newInvokeExecutor(func(request invokeRequest) string { if request.path == "running" { close(started) <-release @@ -151,7 +115,7 @@ func TestInvokeExecutorPausesAndDrains(t *testing.T) { drained := make(chan struct{}) go func() { - executor.pauseAndWait() + executor.pauseAndWait(time.Second) close(drained) }() deadline := time.Now().Add(time.Second) @@ -194,3 +158,101 @@ func TestInvokeExecutorPausesAndDrains(t *testing.T) { <-resumed executor.close() } + +func TestInvokeExecutorPauseTimesOutAndFailsPendingRequests(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}) + executor := newInvokeExecutor(func(request invokeRequest) string { + started <- struct{}{} + <-release + return request.path + }) + + errors := make(chan error, 2) + for _, path := range []string{"first", "second"} { + if !executor.submit(invokeTask{ + request: invokeRequest{path: path}, + complete: func(_ string, err error) { + errors <- err + }, + }) { + t.Fatalf("request %q was rejected", path) + } + } + for range 2 { + <-started + } + + const timeout = 30 * time.Millisecond + startedAt := time.Now() + if executor.pauseAndWait(timeout) { + t.Fatal("pause unexpectedly reported a complete drain") + } + if elapsed := time.Since(startedAt); elapsed < timeout || elapsed > 10*timeout { + t.Fatalf("pause returned after %s, want close to %s", elapsed, timeout) + } + for range 2 { + select { + case err := <-errors: + if err == nil || err.Error() != errNativeBridgeStopping.Error() { + t.Fatalf("unexpected timeout error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("pending request was not failed after the timeout") + } + } + + close(release) + executor.close() + select { + case err := <-errors: + t.Fatalf("request callback ran more than once: %v", err) + default: + } +} + +func TestStopWithinUsesOneDeadlineForRequestsAndCleanup(t *testing.T) { + releaseRequest := make(chan struct{}) + requestStarted := make(chan struct{}) + executor := newInvokeExecutor(func(invokeRequest) string { + close(requestStarted) + <-releaseRequest + return "done" + }) + requestResult := make(chan error, 1) + executor.submit(invokeTask{complete: func(_ string, err error) { + requestResult <- err + }}) + <-requestStarted + + releaseCleanup := make(chan struct{}) + cleanupStarted := make(chan struct{}) + cleanup := func() { + close(cleanupStarted) + <-releaseCleanup + } + + const timeout = 40 * time.Millisecond + startedAt := time.Now() + stopWithin(executor, cleanup, timeout) + if elapsed := time.Since(startedAt); elapsed < timeout || elapsed > 10*timeout { + t.Fatalf("stop returned after %s, want close to %s", elapsed, timeout) + } + select { + case <-cleanupStarted: + case <-time.After(time.Second): + t.Fatal("cleanup was not started after the request deadline") + } + select { + case err := <-requestResult: + if err == nil || err.Error() != errNativeBridgeStopping.Error() { + t.Fatalf("unexpected request result: %v", err) + } + case <-time.After(time.Second): + t.Fatal("timed-out request was not failed") + } + + close(releaseRequest) + close(releaseCleanup) + executor.close() +} diff --git a/pkg/rest/invoke.go b/pkg/rest/invoke.go index 78c25291..90ef4f36 100644 --- a/pkg/rest/invoke.go +++ b/pkg/rest/invoke.go @@ -18,8 +18,8 @@ const ( // API Service without opening an HTTP listener. func Dispatch(method, path, rawQuery, body string) string { runtimeMu.RLock() - defer runtimeMu.RUnlock() service := APIService + runtimeMu.RUnlock() if service == nil { return marshalInvokeResult(model.NewErrorResult("service not started")) } diff --git a/pkg/rest/invoke_test.go b/pkg/rest/invoke_test.go index d1121802..5981ef02 100644 --- a/pkg/rest/invoke_test.go +++ b/pkg/rest/invoke_test.go @@ -4,9 +4,12 @@ import ( "encoding/json" "fmt" "net/http" + "net/http/httptest" "path/filepath" "strings" + "sync" "testing" + "time" "github.com/GopeedLab/gopeed/pkg/base" "github.com/GopeedLab/gopeed/pkg/rest/model" @@ -35,6 +38,75 @@ func TestNativeInvokeWithoutRESTListener(t *testing.T) { } } +func TestStartAPIServerIsNotBlockedByPendingDispatch(t *testing.T) { + requestStarted := make(chan struct{}) + releaseRequest := make(chan struct{}) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseRequest) }) } + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + close(requestStarted) + <-releaseRequest + w.Header().Set("Content-Length", "0") + })) + t.Cleanup(func() { + release() + upstream.Close() + Stop() + }) + + if _, err := Start(&model.StartConfig{NativeMode: true, Storage: model.StorageMem}); err != nil { + t.Fatal(err) + } + requestBody, err := json.Marshal(&model.ResolveTask{Req: &base.Request{URL: upstream.URL}}) + if err != nil { + t.Fatal(err) + } + dispatchDone := make(chan struct{}) + go func() { + defer close(dispatchDone) + Dispatch(http.MethodPost, "/api/v1/resolve", "", string(requestBody)) + }() + select { + case <-requestStarted: + case <-time.After(time.Second): + t.Fatal("resolve request did not start") + } + + config, err := Downloader.GetConfig() + if err != nil { + t.Fatal(err) + } + config.API = &base.APIServerConfig{Enable: true, Network: "tcp", Address: "127.0.0.1:0"} + if err := Downloader.PutConfig(config); err != nil { + t.Fatal(err) + } + started := make(chan *model.APIServerState, 1) + startErrors := make(chan error, 1) + go func() { + state, err := StartAPIServer() + started <- state + startErrors <- err + }() + select { + case state := <-started: + if err := <-startErrors; err != nil { + t.Fatalf("start API server failed: %v", err) + } + if !state.Running || state.RunningPort == 0 { + t.Fatalf("API server did not start: %#v", state) + } + case <-time.After(time.Second): + t.Fatal("pending Resolve blocked API server startup") + } + + release() + select { + case <-dispatchDone: + case <-time.After(time.Second): + t.Fatal("resolve request did not finish after release") + } +} + func TestAPIServerLifecycleUsesPersistedConfigWithoutRestartingCore(t *testing.T) { port, err := Start(&model.StartConfig{NativeMode: true, Storage: model.StorageMem}) if err != nil { diff --git a/ui/flutter/lib/core/ffi/libgopeed_worker.dart b/ui/flutter/lib/core/ffi/libgopeed_worker.dart index 4ffeba1b..a056b077 100644 --- a/ui/flutter/lib/core/ffi/libgopeed_worker.dart +++ b/ui/flutter/lib/core/ffi/libgopeed_worker.dart @@ -26,10 +26,8 @@ class LibgopeedWorker { final ReceivePort _errorPort = ReceivePort(); final Map> _pending = {}; final StreamController> _taskEvents = StreamController>.broadcast(); - final Set _invokeRequests = {}; late final Future _isolateReady; final Completer _sendPort = Completer(); - Completer? _invokeDrain; Future? _stopFuture; int _requestID = 0; bool _stopping = false; @@ -43,19 +41,22 @@ class LibgopeedWorker { Future _stop() async { _stopping = true; - if (_invokeRequests.isNotEmpty) { - _invokeDrain ??= Completer(); - await _invokeDrain!.future; - } + // The native bridge owns the bounded drain. Sending Stop immediately is + // important because an InvokeAsync handler such as Resolve may never return. await _request('stop', const [], allowWhileStopping: true); _stopped = true; + final stoppingError = StateError('libgopeed worker stopped before the request completed'); + for (final completer in _pending.values) { + completer.completeError(stoppingError); + } + _pending.clear(); await _taskEvents.close(); _receivePort.close(); _errorPort.close(); } Future invoke(String method, String path, String query, String body) { - return _request('invoke', [method, path, query, body], trackInvoke: true); + return _request('invoke', [method, path, query, body]); } Future apiServer(String operation) => _request('apiServer', [operation]); @@ -64,12 +65,7 @@ class LibgopeedWorker { await _request('subscribeTaskEvents', [mask]); } - Future _request( - String type, - List arguments, { - bool trackInvoke = false, - bool allowWhileStopping = false, - }) async { + Future _request(String type, List arguments, {bool allowWhileStopping = false}) async { if (_stopped || (_stopping && !allowWhileStopping)) { throw StateError('libgopeed worker is stopping'); } @@ -78,7 +74,6 @@ class LibgopeedWorker { final requestID = _requestID++; final completer = Completer(); _pending[requestID] = completer; - if (trackInvoke) _invokeRequests.add(requestID); sendPort.send([type, requestID, ...arguments]); return (await completer.future) as T; } @@ -94,10 +89,6 @@ class LibgopeedWorker { final requestID = response[1] as int; final completer = _pending.remove(requestID); if (completer == null) return; - if (_invokeRequests.remove(requestID) && _invokeRequests.isEmpty) { - _invokeDrain?.complete(); - _invokeDrain = null; - } if (response[2] as bool) { completer.complete(response[3]); } else { @@ -116,9 +107,6 @@ class LibgopeedWorker { completer.completeError(error); } _pending.clear(); - _invokeRequests.clear(); - _invokeDrain?.completeError(error); - _invokeDrain = null; _taskEvents.addError(error); }