From 05b34a4b11c5d6b1489fe0a85dd6f158f1e84fd0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Fri, 15 May 2026 15:00:59 +0800 Subject: [PATCH] usbip: drain darwin watcher callbacks and wait for active sessions on close `darwinExportHost.Events` closed its producer channel after destroying the IOKit notification port, but `IONotificationPortDestroy` does not synchronously drain the dispatch queue: a callback block already executing or enqueued could still call `signal()` on the closed channel, or resolve the `cgo.Handle` after `ref.Delete()`. Retain the serial dispatch queue inside the watcher and `dispatch_sync` an empty block before releasing it, so no native callback is in flight when the Go side proceeds. `ServerService.Close` cancelled the context and immediately closed the `ExportHost`. On Darwin that calls `IOUSBHostDevice destroy` for every capture while `darwinServerDataSession.handleSubmit` goroutines may still be inside `IOUSBHostPipe` requests on the same device. Track active sessions, close them explicitly, and wait for the per-session cleanup goroutines via a WaitGroup before tearing down the host. --- service/usbip/server.go | 54 +++++++++++++++++++++++++++++----- service/usbip/usbhost_darwin.m | 11 +++++++ 2 files changed, 58 insertions(+), 7 deletions(-) diff --git a/service/usbip/server.go b/service/usbip/server.go index 80a39044e..553c521e0 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -31,6 +31,11 @@ type ServerService struct { ledger *exportLedger reconcileAccess sync.Mutex + + sessionsAccess sync.Mutex + sessions map[DataSession]struct{} + sessionsClosed bool + sessionsWG sync.WaitGroup } func NewServerService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPServerServiceOptions) (adapter.Service, error) { @@ -48,13 +53,14 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string, } ctx, cancel := context.WithCancel(ctx) return &ServerService{ - Adapter: boxService.NewAdapter(C.TypeUSBIPServer, tag), - ctx: ctx, - cancel: cancel, - logger: logger, - matches: options.Devices, - host: host, - ledger: newExportLedger(logger, importLeaseTTL, time.Now), + Adapter: boxService.NewAdapter(C.TypeUSBIPServer, tag), + ctx: ctx, + cancel: cancel, + logger: logger, + matches: options.Devices, + host: host, + ledger: newExportLedger(logger, importLeaseTTL, time.Now), + sessions: make(map[DataSession]struct{}), listener: listener.New(listener.Options{ Context: ctx, Logger: logger, @@ -91,10 +97,25 @@ func (s *ServerService) Close() error { if s.cancel != nil { s.cancel() } + + s.sessionsAccess.Lock() + s.sessionsClosed = true + sessions := make([]DataSession, 0, len(s.sessions)) + for session := range s.sessions { + sessions = append(sessions, session) + } + s.sessionsAccess.Unlock() + for _, conn := range s.ledger.CloseAllSubscribers() { _ = conn.Close() } err := common.Close(common.PtrOrNil(s.listener)) + + for _, session := range sessions { + _ = session.Close() + } + s.sessionsWG.Wait() + s.reconcileAccess.Lock() defer s.reconcileAccess.Unlock() _ = s.host.Close() @@ -301,15 +322,34 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended s.tearDownPreparedSession(busid, session) return false } + + s.sessionsAccess.Lock() + if s.sessionsClosed { + s.sessionsAccess.Unlock() + s.tearDownPreparedSession(busid, session) + return false + } + s.sessions[session] = struct{}{} + s.sessionsWG.Add(1) + s.sessionsAccess.Unlock() + err = session.Start() if err != nil { + s.sessionsAccess.Lock() + delete(s.sessions, session) + s.sessionsAccess.Unlock() + s.sessionsWG.Done() s.logger.Warn("start data session ", busid, ": ", err) s.tearDownPreparedSession(busid, session) return false } s.logger.Info("attached ", busid, " to remote ", conn.RemoteAddr()) go func() { + defer s.sessionsWG.Done() <-session.Done() + s.sessionsAccess.Lock() + delete(s.sessions, session) + s.sessionsAccess.Unlock() released, err := s.host.FinishImport(s.ctx, busid) if err != nil { s.logger.Debug("finish import ", busid, ": ", err) diff --git a/service/usbip/usbhost_darwin.m b/service/usbip/usbhost_darwin.m index 54117b43d..589d8c4a4 100644 --- a/service/usbip/usbhost_darwin.m +++ b/service/usbip/usbhost_darwin.m @@ -49,6 +49,7 @@ struct box_usbhost_device_watcher { IONotificationPortRef port; io_iterator_t matched; io_iterator_t terminated; + void *queue; uintptr_t ref; }; @@ -391,6 +392,7 @@ box_usbhost_device_watcher_t *box_usbhost_device_watcher_create(uintptr_t ref, c return NULL; } dispatch_queue_t queue = dispatch_queue_create("io.nekohasekai.sing-box.usbhost-watch", DISPATCH_QUEUE_SERIAL); + watcher->queue = (void *)CFBridgingRetain(queue); IONotificationPortSetDispatchQueue(watcher->port, queue); CFMutableDictionaryRef matching = box_usbhost_device_matching_dictionary(); @@ -440,6 +442,15 @@ void box_usbhost_device_watcher_destroy(box_usbhost_device_watcher_t *watcher) { IONotificationPortDestroy(watcher->port); watcher->port = NULL; } + if (watcher->queue != NULL) { + // Drain pending IOKit notification callbacks before releasing the + // cgo.Handle: the serial queue guarantees any block already + // enqueued (or executing) finishes before this empty block runs. + dispatch_queue_t queue = (__bridge dispatch_queue_t)watcher->queue; + dispatch_sync(queue, ^{}); + CFRelease(watcher->queue); + watcher->queue = NULL; + } free(watcher); }