From d12122ea86db8df134875449119516fb10e2e79b Mon Sep 17 00:00:00 2001 From: Levi Date: Sun, 6 Sep 2026 11:23:27 +0800 Subject: [PATCH 1/3] feat: make desktop FFI invocation asynchronous --- bind/desktop/invoke_executor.go | 123 ++++++++++ bind/desktop/invoke_executor_test.go | 102 +++++++++ bind/desktop/main.go | 58 +++++ pkg/rest/invoke.go | 4 + pkg/rest/server.go | 2 +- .../docs/backend-transport-architecture.md | 8 +- ui/flutter/include/libgopeed.h | 39 +++- ui/flutter/lib/core/common/libgopeed_ffi.dart | 74 +----- .../lib/core/entry/libgopeed_boot_native.dart | 15 +- ui/flutter/lib/core/ffi/libgopeed_bind.dart | 57 ++++- ui/flutter/lib/core/ffi/libgopeed_worker.dart | 210 +++++++++++++++--- .../gopeed/entry/gopeed_transport_native.dart | 22 +- ui/flutter/pubspec.yaml | 1 + 13 files changed, 574 insertions(+), 141 deletions(-) create mode 100644 bind/desktop/invoke_executor.go create mode 100644 bind/desktop/invoke_executor_test.go diff --git a/bind/desktop/invoke_executor.go b/bind/desktop/invoke_executor.go new file mode 100644 index 00000000..7120f489 --- /dev/null +++ b/bind/desktop/invoke_executor.go @@ -0,0 +1,123 @@ +package main + +import ( + "fmt" + "runtime" + "sync" +) + +const ( + minInvokeConcurrency = 4 + maxInvokeConcurrency = 32 +) + +type invokeRequest struct { + method string + path string + query string + body string +} + +type invokeTask struct { + request invokeRequest + complete func(result string, err error) +} + +// invokeExecutor keeps native API work concurrent without ever rejecting a +// request because all workers are busy. submit only appends to the waiting +// queue; callers observe backpressure through the Future completed by the +// callback rather than by blocking the FFI entry point. +type invokeExecutor struct { + mu sync.Mutex + ready *sync.Cond + queue []invokeTask + closed bool + workers sync.WaitGroup + handler func(invokeRequest) string +} + +func invokeConcurrency(cpuCount int) int { + concurrency := cpuCount * 2 + if concurrency < minInvokeConcurrency { + return minInvokeConcurrency + } + 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} + 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 { + return false + } + e.queue = append(e.queue, task) + e.ready.Signal() + return true +} + +func (e *invokeExecutor) close() { + e.mu.Lock() + e.closed = true + e.ready.Broadcast() + e.mu.Unlock() + e.workers.Wait() +} + +func (e *invokeExecutor) run() { + defer e.workers.Done() + for { + task, ok := e.take() + if !ok { + return + } + result, err := e.execute(task.request) + task.complete(result, err) + } +} + +func (e *invokeExecutor) take() (invokeTask, bool) { + e.mu.Lock() + defer e.mu.Unlock() + for len(e.queue) == 0 && !e.closed { + e.ready.Wait() + } + 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 +} + +func (e *invokeExecutor) execute(request invokeRequest) (result string, err error) { + defer func() { + if recovered := recover(); recovered != nil { + err = fmt.Errorf("invoke panic: %v", recovered) + } + }() + return e.handler(request), nil +} diff --git a/bind/desktop/invoke_executor_test.go b/bind/desktop/invoke_executor_test.go new file mode 100644 index 00000000..1bdd1d43 --- /dev/null +++ b/bind/desktop/invoke_executor_test.go @@ -0,0 +1,102 @@ +package main + +import ( + "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 + ) + 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 + } + } + started <- struct{}{} + <-release + active.Add(-1) + return request.path + }) + + var completed sync.WaitGroup + completed.Add(taskCount) + for i := range taskCount { + accepted := executor.submit(invokeTask{ + request: invokeRequest{path: strconv.Itoa(i)}, + complete: func(string, error) { + completed.Done() + }, + }) + if !accepted { + t.Fatalf("task %d was rejected while all workers were busy", i) + } + } + + for range workerCount { + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("workers did not start queued tasks") + } + } + if got := maxActive.Load(); got != workerCount { + t.Fatalf("maximum active tasks = %d, want %d", got, workerCount) + } + + close(release) + completed.Wait() + executor.close() +} + +func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) { + executor := newInvokeExecutor(1, func(invokeRequest) string { + panic("boom") + }) + completed := make(chan error, 1) + executor.submit(invokeTask{complete: func(_ string, err error) { + completed <- err + }}) + select { + case err := <-completed: + if err == nil || err.Error() != "invoke panic: boom" { + t.Fatalf("unexpected error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("panic result was not completed") + } + executor.close() +} diff --git a/bind/desktop/main.go b/bind/desktop/main.go index cedbe38f..45cae19d 100644 --- a/bind/desktop/main.go +++ b/bind/desktop/main.go @@ -6,14 +6,21 @@ package main typedef void (*TaskEventCallback)(char* payload); +typedef void (*InvokeResultCallback)(uint64_t request_id, int success, char* payload); + static void callTaskEventCallback(uintptr_t callback, char* payload) { ((TaskEventCallback)callback)(payload); } + +static void callInvokeResultCallback(uintptr_t callback, uint64_t request_id, int success, char* payload) { + ((InvokeResultCallback)callback)(request_id, success, payload); +} */ import "C" import ( "encoding/json" "fmt" + "sync" "unsafe" "github.com/GopeedLab/gopeed/pkg/rest" @@ -22,6 +29,20 @@ import ( func main() {} +var ( + invokeExecutorOnce sync.Once + desktopInvoker *invokeExecutor +) + +func getInvokeExecutor() *invokeExecutor { + invokeExecutorOnce.Do(func() { + desktopInvoker = newDefaultInvokeExecutor(func(request invokeRequest) string { + return rest.Invoke(request.method, request.path, request.query, request.body) + }) + }) + return desktopInvoker +} + //export Start func Start(cfg *C.char) (int, *C.char) { var config model.StartConfig @@ -85,6 +106,43 @@ func Invoke(method *C.char, path *C.char, query *C.char, body *C.char) *C.char { )) } +//export InvokeAsync +func InvokeAsync( + method *C.char, + path *C.char, + query *C.char, + body *C.char, + requestID C.ulonglong, + callback C.uintptr_t, +) { + if callback == 0 { + return + } + request := invokeRequest{ + method: C.GoString(method), + path: C.GoString(path), + query: C.GoString(query), + body: C.GoString(body), + } + complete := func(result string, err error) { + success := C.int(1) + payload := result + if err != nil { + success = 0 + payload = err.Error() + } + C.callInvokeResultCallback( + callback, + C.uint64_t(requestID), + success, + C.CString(payload), + ) + } + if !getInvokeExecutor().submit(invokeTask{request: request, complete: complete}) { + complete("", fmt.Errorf("invoke executor is closed")) + } +} + //export SubscribeTaskEvents func SubscribeTaskEvents(mask C.ulonglong, callback C.uintptr_t) { if mask == 0 || callback == 0 { diff --git a/pkg/rest/invoke.go b/pkg/rest/invoke.go index 122d20c3..be801896 100644 --- a/pkg/rest/invoke.go +++ b/pkg/rest/invoke.go @@ -16,6 +16,8 @@ const ( // Invoke dispatches a Core API request without using an HTTP listener. It is // the shared entry point for Desktop FFI and gomobile bindings. func Invoke(method, path, rawQuery, body string) string { + runtimeMu.RLock() + defer runtimeMu.RUnlock() service := APIService if service == nil { return marshalInvokeResult(model.NewErrorResult("service not started")) @@ -29,6 +31,8 @@ func Invoke(method, path, rawQuery, body string) string { } func SubscribeTaskEvents(mask uint64, listener func(payload string)) { + runtimeMu.RLock() + defer runtimeMu.RUnlock() if APIService == nil { return } diff --git a/pkg/rest/server.go b/pkg/rest/server.go index a62562ba..69cbf2c8 100644 --- a/pkg/rest/server.go +++ b/pkg/rest/server.go @@ -29,7 +29,7 @@ import ( ) var ( - runtimeMu sync.Mutex + runtimeMu sync.RWMutex Downloader *download.Downloader APIService *goapi.Service diff --git a/ui/flutter/docs/backend-transport-architecture.md b/ui/flutter/docs/backend-transport-architecture.md index 2de7ab81..1d3879ee 100644 --- a/ui/flutter/docs/backend-transport-architecture.md +++ b/ui/flutter/docs/backend-transport-architecture.md @@ -260,6 +260,7 @@ StartAPIServer StopAPIServer RestartAPIServer Invoke +InvokeAsync FreeCString SubscribeTaskEvents ``` @@ -268,9 +269,12 @@ Requirements: - Use the cgo C-string ABI in the first phase. Reconsider pointer-and-length parameters only during a future ABI-breaking revision. - Memory allocated by Go must be released with `FreeCString`. -- Dart-owned input memory must be released after the synchronous native call returns. +- `InvokeAsync` copies every Dart-owned input string before returning, queues the request in a bounded-concurrency Go executor, and completes it through a `NativeCallable.listener` callback. +- The async executor uses `clamp(GOMAXPROCS * 2, 4, 32)` workers. Requests wait in FIFO order when every worker is busy; pool saturation is backpressure, not an API error. +- Dart-owned input memory must be released after the synchronous native entry call returns. Go-owned callback payloads remain valid until Dart copies them and calls `FreeCString`. - `Invoke` returns the same Result JSON as REST; business errors do not use a separate ABI error model. -- Run blocking Desktop FFI work outside the Flutter UI isolate. Mobile MethodChannel handlers likewise run blocking Go work on a background task queue. +- Keep `Invoke` as a compatibility entry point, but route Desktop application requests through `InvokeAsync` in the FFI bridge isolate. +- Run blocking Desktop FFI work outside the Flutter UI isolate. The bridge isolate owns the dynamic library and native callbacks, decodes callback JSON, and forwards sendable Dart values to the UI isolate. Mobile MethodChannel handlers likewise run blocking Go work on a background task queue. - Dart bindings are generated from `include/libgopeed.h` with `ffigen`. Every exported Desktop symbol must be added to the `ffigen.functions.include` allowlist. - Never hand-edit `lib/core/ffi/libgopeed_bind.dart`; regenerate it with `flutter pub run ffigen` after changing the C header. diff --git a/ui/flutter/include/libgopeed.h b/ui/flutter/include/libgopeed.h index a7656dd9..945955be 100644 --- a/ui/flutter/include/libgopeed.h +++ b/ui/flutter/include/libgopeed.h @@ -12,6 +12,8 @@ #ifndef GO_CGO_GOSTRING_TYPEDEF typedef struct { const char *p; ptrdiff_t n; } _GoString_; +extern size_t _GoStringLen(_GoString_ s); +extern const char *_GoStringPtr(_GoString_ s); #endif #endif @@ -19,6 +21,24 @@ typedef struct { const char *p; ptrdiff_t n; } _GoString_; /* Start of preamble from import "C" comments. */ +#line 3 "main.go" + +#include +#include + +typedef void (*TaskEventCallback)(char* payload); + +typedef void (*InvokeResultCallback)(uint64_t request_id, int success, char* payload); + +static void callTaskEventCallback(uintptr_t callback, char* payload) { + ((TaskEventCallback)callback)(payload); +} + +static void callInvokeResultCallback(uintptr_t callback, uint64_t request_id, int success, char* payload) { + ((InvokeResultCallback)callback)(request_id, success, payload); +} + +#line 1 "cgo-generated-wrapper" /* End of preamble from import "C" comments. */ @@ -44,10 +64,16 @@ typedef size_t GoUintptr; typedef float GoFloat32; typedef double GoFloat64; #ifdef _MSC_VER +#if !defined(__cplusplus) || _MSVC_LANG <= 201402L #include typedef _Fcomplex GoComplex64; typedef _Dcomplex GoComplex128; #else +#include +typedef std::complex GoComplex64; +typedef std::complex GoComplex128; +#endif +#else typedef float _Complex GoComplex64; typedef double _Complex GoComplex128; #endif @@ -81,13 +107,14 @@ struct Start_return { char* r1; }; extern struct Start_return Start(char* cfg); -extern void Stop(); -extern char* GetAPIServerState(); -extern char* StartAPIServer(); -extern char* StopAPIServer(); -extern char* RestartAPIServer(); +extern void Stop(void); +extern char* GetAPIServerState(void); +extern char* StartAPIServer(void); +extern char* StopAPIServer(void); +extern char* RestartAPIServer(void); extern char* Invoke(char* method, char* path, char* query, char* body); -extern void SubscribeTaskEvents(GoUint64 mask, GoUintptr callback); +extern void InvokeAsync(char* method, char* path, char* query, char* body, long long unsigned int requestID, uintptr_t callback); +extern void SubscribeTaskEvents(long long unsigned int mask, uintptr_t callback); extern void FreeCString(char* value); #ifdef __cplusplus diff --git a/ui/flutter/lib/core/common/libgopeed_ffi.dart b/ui/flutter/lib/core/common/libgopeed_ffi.dart index aae4b938..7081f34f 100644 --- a/ui/flutter/lib/core/common/libgopeed_ffi.dart +++ b/ui/flutter/lib/core/common/libgopeed_ffi.dart @@ -1,10 +1,6 @@ import 'dart:async'; import 'dart:convert'; -import 'dart:ffi'; -import 'package:ffi/ffi.dart'; - -import '../ffi/libgopeed_bind.dart'; import '../ffi/libgopeed_worker.dart'; import 'api_server_state.dart'; import 'libgopeed_interface.dart'; @@ -12,51 +8,22 @@ import 'start_config.dart'; import 'task_event.dart'; class LibgopeedFFi implements LibgopeedInterface { - late LibgopeedBind _libgopeed; + LibgopeedFFi() { + _worker.taskEvents.listen(_onTaskEvent, onError: _taskEvents.addError); + } + final _taskEvents = StreamController.broadcast(); final _worker = LibgopeedWorker(); - late final NativeCallable)> _taskEventCallback; - LibgopeedFFi(LibgopeedBind libgopeed) { - _libgopeed = libgopeed; - _taskEventCallback = NativeCallable)>.listener(_onTaskEvent); - } - - void _onTaskEvent(Pointer payload) { - try { - final decoded = jsonDecode(payload.cast().toDartString()) as Map; - _taskEvents.add(TaskEvent.fromJson(decoded)); - } finally { - _libgopeed.FreeCString(payload); - } + void _onTaskEvent(Map payload) { + _taskEvents.add(TaskEvent.fromJson(payload)); } @override - Future start(StartConfig cfg) { - var completer = Completer(); - final cfgPtr = jsonEncode(cfg).toNativeUtf8(); - try { - final result = _libgopeed.Start(cfgPtr.cast()); - if (result.r1 != nullptr) { - final message = result.r1.cast().toDartString(); - _libgopeed.FreeCString(result.r1); - completer.completeError(Exception(message)); - } else { - completer.complete(result.r0); - } - } finally { - malloc.free(cfgPtr); - } - return completer.future; - } + Future start(StartConfig cfg) => _worker.start(jsonEncode(cfg)); @override - Future stop() { - var completer = Completer(); - _libgopeed.Stop(); - completer.complete(); - return completer.future; - } + Future stop() => _worker.stop(); Future _apiServerOperation(String operation) async { final payload = await _worker.apiServer(operation); @@ -77,33 +44,12 @@ class LibgopeedFFi implements LibgopeedInterface { @override Future invoke(String method, String path, {String query = '', String body = ''}) async { - final methodPtr = method.toNativeUtf8(); - final pathPtr = path.toNativeUtf8(); - final queryPtr = query.toNativeUtf8(); - final bodyPtr = body.toNativeUtf8(); - Pointer? resultPtr; - try { - resultPtr = _libgopeed.Invoke(methodPtr.cast(), pathPtr.cast(), queryPtr.cast(), bodyPtr.cast()); - if (resultPtr == nullptr) { - throw StateError('Gopeed Invoke returned a null response'); - } - return resultPtr.cast().toDartString(); - } finally { - malloc.free(methodPtr); - malloc.free(pathPtr); - malloc.free(queryPtr); - malloc.free(bodyPtr); - if (resultPtr != null && resultPtr != nullptr) { - _libgopeed.FreeCString(resultPtr); - } - } + return jsonEncode(await _worker.invoke(method, path, query, body)); } @override Stream get taskEvents => _taskEvents.stream; @override - Future subscribeTaskEvents(Set events) async { - _libgopeed.SubscribeTaskEvents(events.mask, events.isEmpty ? 0 : _taskEventCallback.nativeFunction.address); - } + Future subscribeTaskEvents(Set events) => _worker.subscribeTaskEvents(events.mask); } diff --git a/ui/flutter/lib/core/entry/libgopeed_boot_native.dart b/ui/flutter/lib/core/entry/libgopeed_boot_native.dart index 1077c045..6c3a9e4e 100644 --- a/ui/flutter/lib/core/entry/libgopeed_boot_native.dart +++ b/ui/flutter/lib/core/entry/libgopeed_boot_native.dart @@ -1,6 +1,4 @@ import 'dart:async'; -import 'dart:ffi'; -import 'dart:io'; import '../../util/util.dart'; import '../common/libgopeed_channel.dart'; @@ -9,7 +7,6 @@ import '../common/libgopeed_interface.dart'; import '../common/api_server_state.dart'; import '../common/start_config.dart'; import '../common/task_event.dart'; -import '../ffi/libgopeed_bind.dart'; import '../libgopeed_boot.dart'; LibgopeedBoot create() => LibgopeedBootNative(); @@ -19,17 +16,7 @@ class LibgopeedBootNative implements LibgopeedBoot { LibgopeedBootNative() { if (Util.isDesktop()) { - var libName = "libgopeed."; - if (Platform.isWindows) { - libName += "dll"; - } - if (Platform.isMacOS) { - libName += "dylib"; - } - if (Platform.isLinux) { - libName += "so"; - } - _libgopeed = LibgopeedFFi(LibgopeedBind(DynamicLibrary.open(libName))); + _libgopeed = LibgopeedFFi(); } else { _libgopeed = LibgopeedChannel(); } diff --git a/ui/flutter/lib/core/ffi/libgopeed_bind.dart b/ui/flutter/lib/core/ffi/libgopeed_bind.dart index 347d416b..90a3aae9 100644 --- a/ui/flutter/lib/core/ffi/libgopeed_bind.dart +++ b/ui/flutter/lib/core/ffi/libgopeed_bind.dart @@ -30,22 +30,30 @@ class LibgopeedBind { late final _StopPtr = _lookup>('Stop'); late final _Stop = _StopPtr.asFunction(); - ffi.Pointer GetAPIServerState() => _GetAPIServerState(); + ffi.Pointer GetAPIServerState() { + return _GetAPIServerState(); + } late final _GetAPIServerStatePtr = _lookup Function()>>('GetAPIServerState'); late final _GetAPIServerState = _GetAPIServerStatePtr.asFunction Function()>(); - ffi.Pointer StartAPIServer() => _StartAPIServer(); + ffi.Pointer StartAPIServer() { + return _StartAPIServer(); + } late final _StartAPIServerPtr = _lookup Function()>>('StartAPIServer'); late final _StartAPIServer = _StartAPIServerPtr.asFunction Function()>(); - ffi.Pointer StopAPIServer() => _StopAPIServer(); + ffi.Pointer StopAPIServer() { + return _StopAPIServer(); + } late final _StopAPIServerPtr = _lookup Function()>>('StopAPIServer'); late final _StopAPIServer = _StopAPIServerPtr.asFunction Function()>(); - ffi.Pointer RestartAPIServer() => _RestartAPIServer(); + ffi.Pointer RestartAPIServer() { + return _RestartAPIServer(); + } late final _RestartAPIServerPtr = _lookup Function()>>('RestartAPIServer'); late final _RestartAPIServer = _RestartAPIServerPtr.asFunction Function()>(); @@ -80,13 +88,48 @@ class LibgopeedBind { ) >(); + void InvokeAsync( + ffi.Pointer method, + ffi.Pointer path, + ffi.Pointer query, + ffi.Pointer body, + int requestID, + int callback, + ) { + return _InvokeAsync(method, path, query, body, requestID, callback); + } + + late final _InvokeAsyncPtr = + _lookup< + ffi.NativeFunction< + ffi.Void Function( + ffi.Pointer, + ffi.Pointer, + ffi.Pointer, + ffi.Pointer, + ffi.UnsignedLongLong, + ffi.UintPtr, + ) + > + >('InvokeAsync'); + late final _InvokeAsync = + _InvokeAsyncPtr.asFunction< + void Function( + ffi.Pointer, + ffi.Pointer, + ffi.Pointer, + ffi.Pointer, + int, + int, + ) + >(); + void SubscribeTaskEvents(int mask, int callback) { return _SubscribeTaskEvents(mask, callback); } - late final _SubscribeTaskEventsPtr = _lookup>( - 'SubscribeTaskEvents', - ); + late final _SubscribeTaskEventsPtr = + _lookup>('SubscribeTaskEvents'); late final _SubscribeTaskEvents = _SubscribeTaskEventsPtr.asFunction(); void FreeCString(ffi.Pointer value) { diff --git a/ui/flutter/lib/core/ffi/libgopeed_worker.dart b/ui/flutter/lib/core/ffi/libgopeed_worker.dart index 72f60912..4ffeba1b 100644 --- a/ui/flutter/lib/core/ffi/libgopeed_worker.dart +++ b/ui/flutter/lib/core/ffi/libgopeed_worker.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:convert'; import 'dart:ffi'; import 'dart:io'; import 'dart:isolate'; @@ -7,38 +8,79 @@ import 'package:ffi/ffi.dart'; import 'libgopeed_bind.dart'; +typedef _InvokeResultNative = Void Function(Uint64 requestID, Int32 success, Pointer payload); +typedef _TaskEventNative = Void Function(Pointer payload); + class LibgopeedWorker { - LibgopeedWorker() { + factory LibgopeedWorker() => _instance; + + LibgopeedWorker._() { _receivePort.listen(_handleWorkerMessage); _errorPort.listen(_handleWorkerError); _isolateReady = Isolate.spawn(_workerMain, _receivePort.sendPort, onError: _errorPort.sendPort); } + static final LibgopeedWorker _instance = LibgopeedWorker._(); + final ReceivePort _receivePort = ReceivePort(); final ReceivePort _errorPort = ReceivePort(); - final Map> _pending = {}; + 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; + bool _stopped = false; - Future invoke(String method, String path, String query, String body) async { - await _isolateReady; - final sendPort = await _sendPort.future; - final requestID = _requestID++; - final completer = Completer(); - _pending[requestID] = completer; - sendPort.send(['invoke', requestID, method, path, query, body]); - return completer.future; + Stream> get taskEvents => _taskEvents.stream; + + Future start(String config) => _request('start', [config]); + + Future stop() => _stopFuture ??= _stop(); + + Future _stop() async { + _stopping = true; + if (_invokeRequests.isNotEmpty) { + _invokeDrain ??= Completer(); + await _invokeDrain!.future; + } + await _request('stop', const [], allowWhileStopping: true); + _stopped = true; + await _taskEvents.close(); + _receivePort.close(); + _errorPort.close(); } - Future apiServer(String operation) async { + Future invoke(String method, String path, String query, String body) { + return _request('invoke', [method, path, query, body], trackInvoke: true); + } + + Future apiServer(String operation) => _request('apiServer', [operation]); + + Future subscribeTaskEvents(int mask) async { + await _request('subscribeTaskEvents', [mask]); + } + + Future _request( + String type, + List arguments, { + bool trackInvoke = false, + bool allowWhileStopping = false, + }) async { + if (_stopped || (_stopping && !allowWhileStopping)) { + throw StateError('libgopeed worker is stopping'); + } await _isolateReady; final sendPort = await _sendPort.future; final requestID = _requestID++; - final completer = Completer(); + final completer = Completer(); _pending[requestID] = completer; - sendPort.send(['apiServer', requestID, operation]); - return completer.future; + if (trackInvoke) _invokeRequests.add(requestID); + sendPort.send([type, requestID, ...arguments]); + return (await completer.future) as T; } void _handleWorkerMessage(dynamic message) { @@ -47,13 +89,22 @@ class LibgopeedWorker { return; } final response = message as List; - final requestID = response[0] as int; - final completer = _pending.remove(requestID); - if (completer == null) return; - if (response[1] as bool) { - completer.complete(response[2] as String); - } else { - completer.completeError(StateError(response[2] as String)); + switch (response[0] as String) { + case 'response': + 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 { + completer.completeError(StateError(response[3] as String)); + } + case 'taskEvent': + _taskEvents.add((response[1] as Map).cast()); } } @@ -65,35 +116,120 @@ class LibgopeedWorker { completer.completeError(error); } _pending.clear(); + _invokeRequests.clear(); + _invokeDrain?.completeError(error); + _invokeDrain = null; + _taskEvents.addError(error); } static void _workerMain(SendPort mainSendPort) { final requests = ReceivePort(); final bindings = LibgopeedBind(DynamicLibrary.open(_libraryName())); + late final NativeCallable<_InvokeResultNative> invokeResultCallback; + late final NativeCallable<_TaskEventNative> taskEventCallback; + + invokeResultCallback = NativeCallable<_InvokeResultNative>.listener(( + int requestID, + int success, + Pointer payload, + ) { + if (payload == nullptr) { + mainSendPort.send(['response', requestID, false, 'Gopeed InvokeAsync returned a null response']); + return; + } + try { + final result = payload.cast().toDartString(); + mainSendPort.send(['response', requestID, success != 0, success != 0 ? jsonDecode(result) : result]); + } catch (error) { + mainSendPort.send(['response', requestID, false, error.toString()]); + } finally { + bindings.FreeCString(payload); + } + }); + + taskEventCallback = NativeCallable<_TaskEventNative>.listener((Pointer payload) { + if (payload == nullptr) return; + try { + final event = jsonDecode(payload.cast().toDartString()) as Map; + mainSendPort.send(['taskEvent', event]); + } finally { + bindings.FreeCString(payload); + } + }); + mainSendPort.send(requests.sendPort); requests.listen((dynamic message) { final request = message as List; final type = request[0] as String; final requestID = request[1] as int; - try { - final response = switch (type) { - 'invoke' => _invoke( + if (type == 'invoke') { + try { + _invokeAsync( bindings, + invokeResultCallback.nativeFunction.address, + requestID, request[2] as String, request[3] as String, request[4] as String, request[5] as String, - ), + ); + } catch (error) { + mainSendPort.send(['response', requestID, false, error.toString()]); + } + return; + } + + try { + final response = switch (type) { + 'start' => _start(bindings, request[2] as String), + 'stop' => _stopNative(bindings), 'apiServer' => _apiServer(bindings, request[2] as String), + 'subscribeTaskEvents' => _subscribeTaskEvents( + bindings, + taskEventCallback.nativeFunction.address, + request[2] as int, + ), _ => throw StateError('Unknown libgopeed worker request: $type'), }; - mainSendPort.send([requestID, true, response]); + mainSendPort.send(['response', requestID, true, response]); + if (type == 'stop') { + invokeResultCallback.close(); + taskEventCallback.close(); + requests.close(); + } } catch (error) { - mainSendPort.send([requestID, false, error.toString()]); + mainSendPort.send(['response', requestID, false, error.toString()]); } }); } + static int _start(LibgopeedBind bindings, String config) { + final configPtr = config.toNativeUtf8(); + try { + final result = bindings.Start(configPtr.cast()); + if (result.r1 != nullptr) { + try { + throw StateError(result.r1.cast().toDartString()); + } finally { + bindings.FreeCString(result.r1); + } + } + return result.r0; + } finally { + malloc.free(configPtr); + } + } + + static Object? _stopNative(LibgopeedBind bindings) { + bindings.Stop(); + return null; + } + + static Object? _subscribeTaskEvents(LibgopeedBind bindings, int callback, int mask) { + bindings.SubscribeTaskEvents(mask, mask == 0 ? 0 : callback); + return null; + } + static String _apiServer(LibgopeedBind bindings, String operation) { final resultPtr = switch (operation) { 'get' => bindings.GetAPIServerState(), @@ -112,26 +248,26 @@ class LibgopeedWorker { } } - static String _invoke(LibgopeedBind bindings, String method, String path, String query, String body) { + static void _invokeAsync( + LibgopeedBind bindings, + int callback, + int requestID, + String method, + String path, + String query, + String body, + ) { final methodPtr = method.toNativeUtf8(); final pathPtr = path.toNativeUtf8(); final queryPtr = query.toNativeUtf8(); final bodyPtr = body.toNativeUtf8(); - Pointer? resultPtr; try { - resultPtr = bindings.Invoke(methodPtr.cast(), pathPtr.cast(), queryPtr.cast(), bodyPtr.cast()); - if (resultPtr == nullptr) { - throw StateError('Gopeed Invoke returned a null response'); - } - return resultPtr.cast().toDartString(); + bindings.InvokeAsync(methodPtr.cast(), pathPtr.cast(), queryPtr.cast(), bodyPtr.cast(), requestID, callback); } finally { malloc.free(methodPtr); malloc.free(pathPtr); malloc.free(queryPtr); malloc.free(bodyPtr); - if (resultPtr != null && resultPtr != nullptr) { - bindings.FreeCString(resultPtr); - } } } diff --git a/ui/flutter/lib/core/network/gopeed/entry/gopeed_transport_native.dart b/ui/flutter/lib/core/network/gopeed/entry/gopeed_transport_native.dart index 36f3f3e5..faad20b7 100644 --- a/ui/flutter/lib/core/network/gopeed/entry/gopeed_transport_native.dart +++ b/ui/flutter/lib/core/network/gopeed/entry/gopeed_transport_native.dart @@ -2,9 +2,9 @@ import 'dart:convert'; import 'package:dio/dio.dart'; -import '../../../libgopeed_boot.dart'; -import '../../../ffi/libgopeed_worker.dart'; import '../../../../util/util.dart'; +import '../../../ffi/libgopeed_worker.dart'; +import '../../../libgopeed_boot.dart'; import '../gopeed_transport.dart'; GopeedTransport createTransport(GopeedTransportConfig config) => NativeGopeedTransport(); @@ -32,14 +32,16 @@ class NativeGopeedTransport implements GopeedTransport { }) async { final request = _RequestParts.from(path, queryParameters); final requestBody = data == null ? '' : jsonEncode(data); - final payload = _ffiWorker != null - ? await _ffiWorker.invoke(method.toUpperCase(), request.path, request.query, requestBody) - : await LibgopeedBoot.instance.invoke( - method.toUpperCase(), - request.path, - query: request.query, - body: requestBody, - ); + final worker = _ffiWorker; + if (worker != null) { + return worker.invoke(method.toUpperCase(), request.path, request.query, requestBody); + } + final payload = await LibgopeedBoot.instance.invoke( + method.toUpperCase(), + request.path, + query: request.query, + body: requestBody, + ); return jsonDecode(payload); } diff --git a/ui/flutter/pubspec.yaml b/ui/flutter/pubspec.yaml index 6bca2e53..60c93a08 100644 --- a/ui/flutter/pubspec.yaml +++ b/ui/flutter/pubspec.yaml @@ -209,6 +209,7 @@ ffigen: - "StopAPIServer" - "RestartAPIServer" - "Invoke" + - "InvokeAsync" - "SubscribeTaskEvents" - "FreeCString" structs: From 607d1729e6329813d90220e7923da752ef9e76f5 Mon Sep 17 00:00:00 2001 From: Levi Date: Sun, 6 Sep 2026 11:59:11 +0800 Subject: [PATCH 2/3] refactor: share async invoke bridge across native platforms --- bind/desktop/invoke_executor.go | 123 ----------- bind/desktop/main.go | 44 +--- bind/mobile/main.go | 25 ++- bind/native/invoke.go | 202 ++++++++++++++++++ .../invoke_test.go} | 96 ++++++++- pkg/rest/invoke.go | 7 +- pkg/rest/invoke_test.go | 8 +- .../kotlin/com/gopeed/gopeed/MainActivity.kt | 25 ++- .../docs/backend-transport-architecture.md | 25 ++- ui/flutter/include/libgopeed.h | 1 - ui/flutter/ios/Runner/AppDelegate.swift | 8 +- .../ios/Runner/GopeedTaskEventForwarder.h | 8 + .../ios/Runner/GopeedTaskEventForwarder.m | 51 +++++ .../lib/core/common/libgopeed_channel.dart | 3 + ui/flutter/lib/core/ffi/libgopeed_bind.dart | 30 --- ui/flutter/pubspec.yaml | 1 - .../core/common/libgopeed_channel_test.dart | 29 +++ 17 files changed, 464 insertions(+), 222 deletions(-) delete mode 100644 bind/desktop/invoke_executor.go create mode 100644 bind/native/invoke.go rename bind/{desktop/invoke_executor_test.go => native/invoke_test.go} (51%) create mode 100644 ui/flutter/test/core/common/libgopeed_channel_test.dart diff --git a/bind/desktop/invoke_executor.go b/bind/desktop/invoke_executor.go deleted file mode 100644 index 7120f489..00000000 --- a/bind/desktop/invoke_executor.go +++ /dev/null @@ -1,123 +0,0 @@ -package main - -import ( - "fmt" - "runtime" - "sync" -) - -const ( - minInvokeConcurrency = 4 - maxInvokeConcurrency = 32 -) - -type invokeRequest struct { - method string - path string - query string - body string -} - -type invokeTask struct { - request invokeRequest - complete func(result string, err error) -} - -// invokeExecutor keeps native API work concurrent without ever rejecting a -// request because all workers are busy. submit only appends to the waiting -// queue; callers observe backpressure through the Future completed by the -// callback rather than by blocking the FFI entry point. -type invokeExecutor struct { - mu sync.Mutex - ready *sync.Cond - queue []invokeTask - closed bool - workers sync.WaitGroup - handler func(invokeRequest) string -} - -func invokeConcurrency(cpuCount int) int { - concurrency := cpuCount * 2 - if concurrency < minInvokeConcurrency { - return minInvokeConcurrency - } - 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} - 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 { - return false - } - e.queue = append(e.queue, task) - e.ready.Signal() - return true -} - -func (e *invokeExecutor) close() { - e.mu.Lock() - e.closed = true - e.ready.Broadcast() - e.mu.Unlock() - e.workers.Wait() -} - -func (e *invokeExecutor) run() { - defer e.workers.Done() - for { - task, ok := e.take() - if !ok { - return - } - result, err := e.execute(task.request) - task.complete(result, err) - } -} - -func (e *invokeExecutor) take() (invokeTask, bool) { - e.mu.Lock() - defer e.mu.Unlock() - for len(e.queue) == 0 && !e.closed { - e.ready.Wait() - } - 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 -} - -func (e *invokeExecutor) execute(request invokeRequest) (result string, err error) { - defer func() { - if recovered := recover(); recovered != nil { - err = fmt.Errorf("invoke panic: %v", recovered) - } - }() - return e.handler(request), nil -} diff --git a/bind/desktop/main.go b/bind/desktop/main.go index 45cae19d..7054093b 100644 --- a/bind/desktop/main.go +++ b/bind/desktop/main.go @@ -20,29 +20,15 @@ import "C" import ( "encoding/json" "fmt" - "sync" "unsafe" + nativebridge "github.com/GopeedLab/gopeed/bind/native" "github.com/GopeedLab/gopeed/pkg/rest" "github.com/GopeedLab/gopeed/pkg/rest/model" ) func main() {} -var ( - invokeExecutorOnce sync.Once - desktopInvoker *invokeExecutor -) - -func getInvokeExecutor() *invokeExecutor { - invokeExecutorOnce.Do(func() { - desktopInvoker = newDefaultInvokeExecutor(func(request invokeRequest) string { - return rest.Invoke(request.method, request.path, request.query, request.body) - }) - }) - return desktopInvoker -} - //export Start func Start(cfg *C.char) (int, *C.char) { var config model.StartConfig @@ -56,11 +42,13 @@ func Start(cfg *C.char) (int, *C.char) { if err != nil { return 0, C.CString(err.Error()) } + nativebridge.ResumeInvokes() return realPort, nil } //export Stop func Stop() { + nativebridge.PauseInvokesAndWait() rest.Stop() } @@ -96,16 +84,6 @@ func apiServerResult(state *model.APIServerState, err error) string { return string(data) } -//export Invoke -func Invoke(method *C.char, path *C.char, query *C.char, body *C.char) *C.char { - return C.CString(rest.Invoke( - C.GoString(method), - C.GoString(path), - C.GoString(query), - C.GoString(body), - )) -} - //export InvokeAsync func InvokeAsync( method *C.char, @@ -118,12 +96,6 @@ func InvokeAsync( if callback == 0 { return } - request := invokeRequest{ - method: C.GoString(method), - path: C.GoString(path), - query: C.GoString(query), - body: C.GoString(body), - } complete := func(result string, err error) { success := C.int(1) payload := result @@ -138,9 +110,13 @@ func InvokeAsync( C.CString(payload), ) } - if !getInvokeExecutor().submit(invokeTask{request: request, complete: complete}) { - complete("", fmt.Errorf("invoke executor is closed")) - } + nativebridge.InvokeAsync( + C.GoString(method), + C.GoString(path), + C.GoString(query), + C.GoString(body), + complete, + ) } //export SubscribeTaskEvents diff --git a/bind/mobile/main.go b/bind/mobile/main.go index ab45ae32..2d4b8dae 100644 --- a/bind/mobile/main.go +++ b/bind/mobile/main.go @@ -5,6 +5,7 @@ import "C" import ( "encoding/json" + nativebridge "github.com/GopeedLab/gopeed/bind/native" "github.com/GopeedLab/gopeed/internal/webview/rpcprovider" "github.com/GopeedLab/gopeed/pkg/rest" "github.com/GopeedLab/gopeed/pkg/rest/model" @@ -18,10 +19,15 @@ func Start(cfg string) (int, error) { config.ProductionMode = true config.NativeMode = true applyWebViewProvider(&config) - return rest.Start(&config) + port, err := rest.Start(&config) + if err == nil { + nativebridge.ResumeInvokes() + } + return port, err } func Stop() { + nativebridge.PauseInvokesAndWait() rest.Stop() } @@ -50,8 +56,21 @@ func apiServerResult(state *model.APIServerState, err error) string { return string(data) } -func Invoke(method, path, query, body string) string { - return rest.Invoke(method, path, query, body) +type InvokeResultListener interface { + OnResult(requestID int64, success bool, payload string) +} + +func InvokeAsync(method, path, query, body string, requestID int64, listener InvokeResultListener) { + if listener == nil { + return + } + nativebridge.InvokeAsync(method, path, query, body, func(result string, err error) { + if err != nil { + listener.OnResult(requestID, false, err.Error()) + return + } + listener.OnResult(requestID, true, result) + }) } type TaskEventListener interface { diff --git a/bind/native/invoke.go b/bind/native/invoke.go new file mode 100644 index 00000000..27056551 --- /dev/null +++ b/bind/native/invoke.go @@ -0,0 +1,202 @@ +// Package nativebridge contains transport-neutral behavior shared by the +// Desktop C FFI and gomobile bridge adapters. +package nativebridge + +import ( + "fmt" + "runtime" + "sync" + + "github.com/GopeedLab/gopeed/pkg/rest" +) + +const ( + minInvokeConcurrency = 4 + maxInvokeConcurrency = 32 +) + +type invokeRequest struct { + method string + path string + query string + body string +} + +type invokeTask struct { + request invokeRequest + complete func(result string, err error) +} + +// InvokeAsync queues 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 { + return + } + task := invokeTask{ + request: invokeRequest{ + method: method, + path: path, + query: query, + body: body, + }, + complete: callback, + } + if !getInvokeExecutor().submit(task) { + completeCallback(callback, "", fmt.Errorf("native bridge is stopping")) + } +} + +// ResumeInvokes enables request submission after the native runtime starts. +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() +} + +var ( + invokeExecutorOnce sync.Once + sharedInvoker *invokeExecutor +) + +func getInvokeExecutor() *invokeExecutor { + invokeExecutorOnce.Do(func() { + sharedInvoker = newDefaultInvokeExecutor(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. +type invokeExecutor struct { + mu sync.Mutex + ready *sync.Cond + queue []invokeTask + closed bool + accepting bool + pending int + workers sync.WaitGroup + handler func(invokeRequest) string +} + +func invokeConcurrency(cpuCount int) int { + concurrency := cpuCount * 2 + if concurrency < minInvokeConcurrency { + return minInvokeConcurrency + } + 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 { + return false + } + e.queue = append(e.queue, task) + e.pending++ + e.ready.Signal() + return true +} + +func (e *invokeExecutor) resume() { + e.mu.Lock() + e.accepting = true + e.mu.Unlock() +} + +func (e *invokeExecutor) pauseAndWait() { + e.mu.Lock() + e.accepting = false + for e.pending > 0 { + e.ready.Wait() + } + e.mu.Unlock() +} + +func (e *invokeExecutor) close() { + e.mu.Lock() + e.closed = true + e.ready.Broadcast() + e.mu.Unlock() + e.workers.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) { + e.mu.Lock() + defer e.mu.Unlock() + for len(e.queue) == 0 && !e.closed { + e.ready.Wait() + } + 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 +} + +func (e *invokeExecutor) execute(request invokeRequest) (result string, err error) { + defer func() { + if recovered := recover(); recovered != nil { + err = fmt.Errorf("invoke panic: %v", recovered) + } + }() + return e.handler(request), nil +} + +func completeCallback(callback func(string, error), result string, err error) { + defer func() { + _ = recover() + }() + callback(result, err) +} diff --git a/bind/desktop/invoke_executor_test.go b/bind/native/invoke_test.go similarity index 51% rename from bind/desktop/invoke_executor_test.go rename to bind/native/invoke_test.go index 1bdd1d43..dc55bdb4 100644 --- a/bind/desktop/invoke_executor_test.go +++ b/bind/native/invoke_test.go @@ -1,6 +1,7 @@ -package main +package nativebridge import ( + "runtime" "strconv" "sync" "sync/atomic" @@ -100,3 +101,96 @@ func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) { } executor.close() } + +func TestInvokeExecutorSurvivesCallbackPanic(t *testing.T) { + executor := newInvokeExecutor(1, func(request invokeRequest) string { + return request.path + }) + executor.submit(invokeTask{complete: func(string, error) { + panic("callback failed") + }}) + + completed := make(chan string, 1) + executor.submit(invokeTask{ + request: invokeRequest{path: "next"}, + complete: func(result string, _ error) { + completed <- result + }, + }) + select { + case result := <-completed: + if result != "next" { + t.Fatalf("unexpected result: %q", result) + } + case <-time.After(time.Second): + t.Fatal("worker stopped after callback panic") + } + executor.close() +} + +func TestInvokeExecutorPausesAndDrains(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}) + executor := newInvokeExecutor(1, func(request invokeRequest) string { + if request.path == "running" { + close(started) + <-release + } + return request.path + }) + completed := make(chan struct{}) + if !executor.submit(invokeTask{ + request: invokeRequest{path: "running"}, + complete: func(string, error) { + close(completed) + }, + }) { + t.Fatal("initial task was rejected") + } + <-started + + drained := make(chan struct{}) + go func() { + executor.pauseAndWait() + close(drained) + }() + deadline := time.Now().Add(time.Second) + for { + executor.mu.Lock() + paused := !executor.accepting + executor.mu.Unlock() + if paused { + break + } + if time.Now().After(deadline) { + t.Fatal("executor did not pause") + } + runtime.Gosched() + } + select { + case <-drained: + t.Fatal("pause returned before the running request completed") + default: + } + if executor.submit(invokeTask{}) { + t.Fatal("paused executor accepted a new request") + } + + close(release) + <-completed + select { + case <-drained: + case <-time.After(time.Second): + t.Fatal("pause did not return after pending requests drained") + } + + executor.resume() + resumed := make(chan struct{}) + if !executor.submit(invokeTask{complete: func(string, error) { + close(resumed) + }}) { + t.Fatal("resumed executor rejected a request") + } + <-resumed + executor.close() +} diff --git a/pkg/rest/invoke.go b/pkg/rest/invoke.go index be801896..78c25291 100644 --- a/pkg/rest/invoke.go +++ b/pkg/rest/invoke.go @@ -13,9 +13,10 @@ const ( TaskEventError = uint64(goapi.TaskEventError) ) -// Invoke dispatches a Core API request without using an HTTP listener. It is -// the shared entry point for Desktop FFI and gomobile bindings. -func Invoke(method, path, rawQuery, body string) string { +// Dispatch executes one in-process API request. Native scheduling and callback +// behavior belong to bind/native; this function only bridges into the shared +// API Service without opening an HTTP listener. +func Dispatch(method, path, rawQuery, body string) string { runtimeMu.RLock() defer runtimeMu.RUnlock() service := APIService diff --git a/pkg/rest/invoke_test.go b/pkg/rest/invoke_test.go index 1d9dfe2e..d1121802 100644 --- a/pkg/rest/invoke_test.go +++ b/pkg/rest/invoke_test.go @@ -27,7 +27,7 @@ func TestNativeInvokeWithoutRESTListener(t *testing.T) { } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { @@ -106,7 +106,7 @@ func TestAPIServerLifecycleUsesPersistedConfigWithoutRestartingCore(t *testing.T } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { @@ -153,7 +153,7 @@ func TestRestartAPIServerFailsFastAndLeavesListenerStopped(t *testing.T) { } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { @@ -229,7 +229,7 @@ func TestNativeAPIServerAutoStartFailureKeepsCoreAvailable(t *testing.T) { } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { diff --git a/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt b/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt index 165070da..47b0e3b9 100644 --- a/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt +++ b/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt @@ -1,6 +1,7 @@ package com.gopeed.gopeed import com.gopeed.libgopeed.Libgopeed +import com.gopeed.libgopeed.InvokeResultListener import com.gopeed.libgopeed.TaskEventListener import io.flutter.embedding.android.FlutterActivity import io.flutter.embedding.engine.FlutterEngine @@ -38,13 +39,23 @@ class MainActivity : FlutterActivity() { "restartApiServer" -> result.success(Libgopeed.restartAPIServer()) "invoke" -> { try { - result.success( - Libgopeed.invoke( - call.argument("method") ?: "", - call.argument("path") ?: "", - call.argument("query") ?: "", - call.argument("body") ?: "", - ), + Libgopeed.invokeAsync( + call.argument("method") ?: "", + call.argument("path") ?: "", + call.argument("query") ?: "", + call.argument("body") ?: "", + call.argument("requestID")?.toLong() ?: 0L, + object : InvokeResultListener { + override fun onResult(requestID: Long, success: Boolean, payload: String?) { + runOnUiThread { + if (success) { + result.success(payload ?: "") + } else { + result.error("ERROR", payload ?: "InvokeAsync failed", null) + } + } + } + }, ) } catch (error: Exception) { result.error("ERROR", error.message, null) diff --git a/ui/flutter/docs/backend-transport-architecture.md b/ui/flutter/docs/backend-transport-architecture.md index 1d3879ee..a00985c1 100644 --- a/ui/flutter/docs/backend-transport-architecture.md +++ b/ui/flutter/docs/backend-transport-architecture.md @@ -259,7 +259,6 @@ GetAPIServerState StartAPIServer StopAPIServer RestartAPIServer -Invoke InvokeAsync FreeCString SubscribeTaskEvents @@ -269,12 +268,12 @@ Requirements: - Use the cgo C-string ABI in the first phase. Reconsider pointer-and-length parameters only during a future ABI-breaking revision. - Memory allocated by Go must be released with `FreeCString`. -- `InvokeAsync` copies every Dart-owned input string before returning, queues the request in a bounded-concurrency Go executor, and completes it through a `NativeCallable.listener` callback. -- The async executor uses `clamp(GOMAXPROCS * 2, 4, 32)` workers. Requests wait in FIFO order when every worker is busy; pool saturation is backpressure, not an API error. -- Dart-owned input memory must be released after the synchronous native entry call returns. Go-owned callback payloads remain valid until Dart copies them and calls `FreeCString`. -- `Invoke` returns the same Result JSON as REST; business errors do not use a separate ABI error model. -- Keep `Invoke` as a compatibility entry point, but route Desktop application requests through `InvokeAsync` in the FFI bridge isolate. -- Run blocking Desktop FFI work outside the Flutter UI isolate. The bridge isolate owns the dynamic library and native callbacks, decodes callback JSON, and forwards sendable Dart values to the UI isolate. Mobile MethodChannel handlers likewise run blocking Go work on a background task queue. +- `InvokeAsync` is the only native invocation entry point. Desktop and gomobile adapters both submit to the shared bounded-concurrency executor in `bind/native`. +- The shared executor uses `clamp(GOMAXPROCS * 2, 4, 32)` workers. Requests wait in FIFO order when every worker is busy; pool saturation is backpressure, not an API error. +- The Desktop adapter copies every Dart-owned input string before returning and completes through a `NativeCallable.listener` callback. +- Dart-owned input memory must be released after the non-blocking `InvokeAsync` native entry returns. Go-owned callback payloads remain valid until Dart copies them and calls `FreeCString`. +- `InvokeAsync` returns the same Result JSON as REST through its callback; business errors do not use a separate ABI error model. +- The Desktop bridge isolate owns the dynamic library and native callbacks, decodes callback JSON, and forwards sendable Dart values to the UI isolate. Mobile MethodChannel handlers run on background task queues, while invoke completion is delivered through the shared Go callback flow. - Dart bindings are generated from `include/libgopeed.h` with `ffigen`. Every exported Desktop symbol must be added to the `ffigen.functions.include` allowlist. - Never hand-edit `lib/core/ffi/libgopeed_bind.dart`; regenerate it with `flutter pub run ffigen` after changing the C header. @@ -289,14 +288,14 @@ GetAPIServerState() StartAPIServer() StopAPIServer() RestartAPIServer() -Invoke(method, path, query, body) +InvokeAsync(method, path, query, body, requestID, listener) SubscribeTaskEvents(mask, listener) ``` Flutter bridge rules: -- Normal calls use MethodChannel. -- Android performs normal calls on a background task queue. +- Normal calls use MethodChannel, while `InvokeAsync` completes the pending MethodChannel result through a gomobile listener. +- Android and iOS run the Libgopeed MethodChannel handler on a serial background task queue; callback adapters marshal completion back to the platform UI thread. - Events reuse the same MethodChannel through the reverse `taskEvent` message. - Kotlin and Swift contain no business logic. They only forward arguments, dispatch callbacks to the UI thread, and translate errors. @@ -593,8 +592,8 @@ Flutter converts all of them into `GopeedException`. Components do not handle Di ### 11.3 Platform smoke tests -- Windows, macOS, Linux: startup, Invoke, events, shutdown, and repeated startup. -- Android, iOS: gomobile Invoke, reverse MethodChannel events, and background/foreground transitions. +- Windows, macOS, Linux: startup, InvokeAsync, events, shutdown, and repeated startup. +- Android, iOS: gomobile InvokeAsync callbacks, reverse MethodChannel events, and background/foreground transitions. - Web: REST, login, proxy, and task-file downloads. - Native: enabling and disabling REST does not interrupt download tasks. @@ -609,7 +608,7 @@ Flutter converts all of them into `GopeedException`. Components do not handle Di ### Phase 2: Native in-process calls — implemented on this branch - Integrate the FFI Transport on Desktop. -- Extend gomobile Invoke and MethodChannel on Mobile. +- Extend gomobile InvokeAsync callbacks and MethodChannel on Mobile. - Move Flutter API calls to the unified Transport so each typed method is maintained once. - Do not create a socket listener by default on native clients. diff --git a/ui/flutter/include/libgopeed.h b/ui/flutter/include/libgopeed.h index 945955be..9eb93ac8 100644 --- a/ui/flutter/include/libgopeed.h +++ b/ui/flutter/include/libgopeed.h @@ -112,7 +112,6 @@ extern char* GetAPIServerState(void); extern char* StartAPIServer(void); extern char* StopAPIServer(void); extern char* RestartAPIServer(void); -extern char* Invoke(char* method, char* path, char* query, char* body); extern void InvokeAsync(char* method, char* path, char* query, char* body, long long unsigned int requestID, uintptr_t callback); extern void SubscribeTaskEvents(long long unsigned int mask, uintptr_t callback); extern void FreeCString(char* value); diff --git a/ui/flutter/ios/Runner/AppDelegate.swift b/ui/flutter/ios/Runner/AppDelegate.swift index aac051ad..e99ee6de 100644 --- a/ui/flutter/ios/Runner/AppDelegate.swift +++ b/ui/flutter/ios/Runner/AppDelegate.swift @@ -14,10 +14,13 @@ import UIKit func didInitializeImplicitFlutterEngine(_ engineBridge: FlutterImplicitEngineBridge) { GeneratedPluginRegistrant.register(with: engineBridge.pluginRegistry) let messenger = engineBridge.applicationRegistrar.messenger() + let libgopeedTaskQueue = messenger.makeBackgroundTaskQueue?() let libgopeedChannel = FlutterMethodChannel( name: "gopeed.com/libgopeed", - binaryMessenger: messenger + binaryMessenger: messenger, + codec: FlutterStandardMethodCodec.sharedInstance(), + taskQueue: libgopeedTaskQueue ) let taskEventForwarder = GopeedTaskEventForwarder(channel: libgopeedChannel) libgopeedChannel.setMethodCallHandler { call, result in @@ -56,7 +59,8 @@ import UIKit let path = arguments?["path"] as? String ?? "" let query = arguments?["query"] as? String ?? "" let body = arguments?["body"] as? String ?? "" - result(LibgopeedInvoke(method, path, query, body)) + let requestID = (arguments?["requestID"] as? NSNumber)?.int64Value ?? 0 + GopeedInvokeAsyncWithResult(method, path, query, body, requestID, result) case "subscribeTaskEvents": let arguments = call.arguments as? [String: Any] let mask = (arguments?["mask"] as? NSNumber)?.int64Value ?? 0 diff --git a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h index e12dd1fd..6f9869e6 100644 --- a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h +++ b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h @@ -13,4 +13,12 @@ FOUNDATION_EXPORT void GopeedSubscribeTaskEventsWithForwarder( int64_t mask, GopeedTaskEventForwarder * _Nullable listener); +FOUNDATION_EXPORT void GopeedInvokeAsyncWithResult( + NSString *method, + NSString *path, + NSString *query, + NSString *body, + int64_t requestID, + FlutterResult result); + NS_ASSUME_NONNULL_END diff --git a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m index c6bcee5e..e35bd5cf 100644 --- a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m +++ b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m @@ -8,6 +8,45 @@ @end +@interface GopeedInvokeResultForwarder : NSObject + +@property(nonatomic, copy, nullable) FlutterResult result; + +- (instancetype)initWithResult:(FlutterResult)result; + +@end + +@implementation GopeedInvokeResultForwarder + +- (instancetype)initWithResult:(FlutterResult)result { + self = [super init]; + if (self) { + _result = [result copy]; + } + return self; +} + +- (void)onResult:(int64_t)requestID + success:(BOOL)success + payload:(NSString * _Nullable)payload { + FlutterResult result = self.result; + self.result = nil; + if (result == nil) { + return; + } + dispatch_async(dispatch_get_main_queue(), ^{ + if (success) { + result(payload ?: @""); + } else { + result([FlutterError errorWithCode:@"ERROR" + message:payload ?: @"InvokeAsync failed" + details:nil]); + } + }); +} + +@end + @implementation GopeedTaskEventForwarder - (instancetype)initWithChannel:(FlutterMethodChannel *)channel { @@ -33,3 +72,15 @@ void GopeedSubscribeTaskEventsWithForwarder( GopeedTaskEventForwarder * _Nullable listener) { LibgopeedSubscribeTaskEvents(mask, listener); } + +void GopeedInvokeAsyncWithResult( + NSString *method, + NSString *path, + NSString *query, + NSString *body, + int64_t requestID, + FlutterResult result) { + GopeedInvokeResultForwarder *forwarder = + [[GopeedInvokeResultForwarder alloc] initWithResult:result]; + LibgopeedInvokeAsync(method, path, query, body, requestID, forwarder); +} diff --git a/ui/flutter/lib/core/common/libgopeed_channel.dart b/ui/flutter/lib/core/common/libgopeed_channel.dart index 7bc32778..a6e03c46 100644 --- a/ui/flutter/lib/core/common/libgopeed_channel.dart +++ b/ui/flutter/lib/core/common/libgopeed_channel.dart @@ -11,6 +11,7 @@ import 'task_event.dart'; class LibgopeedChannel implements LibgopeedInterface { static const _channel = MethodChannel('gopeed.com/libgopeed'); final _taskEvents = StreamController.broadcast(); + int _requestID = 0; LibgopeedChannel() { _channel.setMethodCallHandler((call) async { @@ -50,7 +51,9 @@ class LibgopeedChannel implements LibgopeedInterface { @override Future invoke(String method, String path, {String query = '', String body = ''}) async { + final requestID = _requestID++; final result = await _channel.invokeMethod('invoke', { + 'requestID': requestID, 'method': method, 'path': path, 'query': query, diff --git a/ui/flutter/lib/core/ffi/libgopeed_bind.dart b/ui/flutter/lib/core/ffi/libgopeed_bind.dart index 90a3aae9..93e9b2e5 100644 --- a/ui/flutter/lib/core/ffi/libgopeed_bind.dart +++ b/ui/flutter/lib/core/ffi/libgopeed_bind.dart @@ -58,36 +58,6 @@ class LibgopeedBind { late final _RestartAPIServerPtr = _lookup Function()>>('RestartAPIServer'); late final _RestartAPIServer = _RestartAPIServerPtr.asFunction Function()>(); - ffi.Pointer Invoke( - ffi.Pointer method, - ffi.Pointer path, - ffi.Pointer query, - ffi.Pointer body, - ) { - return _Invoke(method, path, query, body); - } - - late final _InvokePtr = - _lookup< - ffi.NativeFunction< - ffi.Pointer Function( - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ) - > - >('Invoke'); - late final _Invoke = - _InvokePtr.asFunction< - ffi.Pointer Function( - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ) - >(); - void InvokeAsync( ffi.Pointer method, ffi.Pointer path, diff --git a/ui/flutter/pubspec.yaml b/ui/flutter/pubspec.yaml index 60c93a08..056cf870 100644 --- a/ui/flutter/pubspec.yaml +++ b/ui/flutter/pubspec.yaml @@ -208,7 +208,6 @@ ffigen: - "StartAPIServer" - "StopAPIServer" - "RestartAPIServer" - - "Invoke" - "InvokeAsync" - "SubscribeTaskEvents" - "FreeCString" diff --git a/ui/flutter/test/core/common/libgopeed_channel_test.dart b/ui/flutter/test/core/common/libgopeed_channel_test.dart new file mode 100644 index 00000000..a3b44b8e --- /dev/null +++ b/ui/flutter/test/core/common/libgopeed_channel_test.dart @@ -0,0 +1,29 @@ +import 'package:flutter/services.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:gopeed/core/common/libgopeed_channel.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + const channel = MethodChannel('gopeed.com/libgopeed'); + final messenger = TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + + tearDown(() { + messenger.setMockMethodCallHandler(channel, null); + }); + + test('mobile invokes carry unique request IDs for async callbacks', () async { + final requestIDs = []; + messenger.setMockMethodCallHandler(channel, (call) async { + expect(call.method, 'invoke'); + final arguments = (call.arguments as Map).cast(); + requestIDs.add(arguments['requestID']! as int); + return '{"code":0}'; + }); + + final bridge = LibgopeedChannel(); + await Future.wait([bridge.invoke('GET', '/api/v1/info'), bridge.invoke('GET', '/api/v1/config')]); + + expect(requestIDs, [0, 1]); + }); +} From d51eeb3f2cfa5afa3ceb695a299e66b94101e25e Mon Sep 17 00:00:00 2001 From: Levi Date: Sun, 6 Sep 2026 21:49:11 +0800 Subject: [PATCH 3/3] fix: prevent native requests from blocking shutdown --- bind/desktop/main.go | 3 +- bind/mobile/main.go | 3 +- bind/native/invoke.go | 208 +++++++++++------- bind/native/invoke_test.go | 158 +++++++++---- pkg/rest/invoke.go | 2 +- pkg/rest/invoke_test.go | 72 ++++++ ui/flutter/lib/core/ffi/libgopeed_worker.dart | 30 +-- 7 files changed, 317 insertions(+), 159 deletions(-) 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); }