From badd50f511a57f0817a599a3afc56546a1df1b02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Wed, 10 Jun 2026 09:25:33 +0800 Subject: [PATCH] usbip: make all-devices worker bookkeeping atomic The desired set (assignment lock) and the worker map (workerAccess) were updated in separate critical sections, so a concurrent resync could observe one but not the other: the same busid ended up with two workers (leaking the first cancel), and stop decisions ran against half-applied state. allWorkers is now the single source of truth, diffed, started, and stopped in one workerAccess section; workers remove themselves on exit with an identity check and restart in place if their busid became desired again meanwhile. The per-busid active flag also became a count: during matched-mode handoff two workers briefly reference one busid, and a failing worker's deactivation must not erase the mark of the genuinely attached one (which made ApplyAll detach live devices, oscillating). --- service/usbip/client.go | 12 ++++- service/usbip/client_assignment.go | 62 ++++++++-------------- service/usbip/client_shared.go | 84 +++++++++++++++++++++--------- 3 files changed, 91 insertions(+), 67 deletions(-) diff --git a/service/usbip/client.go b/service/usbip/client.go index 50a8ebdb6..b63d2c00f 100644 --- a/service/usbip/client.go +++ b/service/usbip/client.go @@ -27,15 +27,23 @@ type ClientService struct { serverAddr M.Socksaddr host ImportHost + // workerAccess guards assignedWorkers and allWorkers. All-devices + // worker bookkeeping (diffing the desired set, starting, stopping, + // self-removal) happens in single workerAccess critical sections; + // assignment.access may be taken nested inside, never the reverse. assignment *clientAssignment workerAccess sync.Mutex assignedWorkers []*clientAssignedWorker - allWorkers map[string]context.CancelFunc + allWorkers map[string]*clientRemoteWorker remoteAccess sync.Mutex remoteDevices map[string]ControlDeviceInfo } +type clientRemoteWorker struct { + cancel context.CancelFunc +} + func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPClientServiceOptions) (adapter.Service, error) { for i, m := range options.Devices { if m.IsZero() { @@ -66,7 +74,7 @@ func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, serverAddr: options.ServerOptions.Build(), host: host, assignment: newClientAssignment(options.Devices), - allWorkers: make(map[string]context.CancelFunc), + allWorkers: make(map[string]*clientRemoteWorker), }, nil } diff --git a/service/usbip/client_assignment.go b/service/usbip/client_assignment.go index dfd6f9f30..5aa70cace 100644 --- a/service/usbip/client_assignment.go +++ b/service/usbip/client_assignment.go @@ -17,9 +17,12 @@ type clientAssignment struct { matchedKnownKeys map[string]DeviceKey allDesired map[string]struct{} - registered map[string]struct{} - activeBusIDs map[string]struct{} + // activeBusIDs counts attach attempts per busid. A count, not a + // flag: during matched-mode handoff two workers briefly reference + // the same busid, and a failing worker's deactivation must not + // erase the mark of the worker that is genuinely attached. + activeBusIDs map[string]int } func newClientAssignment(matches []option.USBIPDeviceMatch) *clientAssignment { @@ -42,8 +45,7 @@ func newClientAssignment(matches []option.USBIPDeviceMatch) *clientAssignment { return &clientAssignment{ targets: targets, allDesired: make(map[string]struct{}), - registered: make(map[string]struct{}), - activeBusIDs: make(map[string]struct{}), + activeBusIDs: make(map[string]int), } } @@ -66,10 +68,21 @@ func (a *clientAssignment) SetActive(busid string, active bool) { a.access.Lock() defer a.access.Unlock() if active { - a.activeBusIDs[busid] = struct{}{} - } else { - delete(a.activeBusIDs, busid) + a.activeBusIDs[busid]++ + return } + count := a.activeBusIDs[busid] + if count <= 1 { + delete(a.activeBusIDs, busid) + } else { + a.activeBusIDs[busid] = count - 1 + } +} + +func (a *clientAssignment) IsActive(busid string) bool { + a.access.Lock() + defer a.access.Unlock() + return a.activeBusIDs[busid] > 0 } func (a *clientAssignment) ApplyMatched(entries []DeviceEntry, knownKeys map[string]DeviceKey) (next []string, previous []string) { @@ -90,44 +103,15 @@ func (a *clientAssignment) ApplyMatched(entries []DeviceEntry, knownKeys map[str return nextAssigned, prev } -func (a *clientAssignment) ApplyAll(entries []DeviceEntry) (start []string, stop []string) { - desired := make(map[string]struct{}, len(entries)) - for i := range entries { - busid := entries[i].Info.BusIDString() - if busid == "" { - continue - } - desired[busid] = struct{}{} - } +func (a *clientAssignment) SetAllDesired(desired map[string]struct{}) { a.access.Lock() - defer a.access.Unlock() a.allDesired = desired - for busid := range a.registered { - if _, ok := desired[busid]; ok { - continue - } - if _, active := a.activeBusIDs[busid]; active { - continue - } - stop = append(stop, busid) - delete(a.registered, busid) - } - for busid := range desired { - if _, ok := a.registered[busid]; ok { - continue - } - start = append(start, busid) - a.registered[busid] = struct{}{} - } - return start, stop + a.access.Unlock() } func (a *clientAssignment) IsRetryDesired(busid string) bool { a.access.Lock() defer a.access.Unlock() - if _, registered := a.registered[busid]; !registered { - return false - } _, desired := a.allDesired[busid] return desired } @@ -276,7 +260,7 @@ func (a *clientAssignment) activeCurrentAssignmentsLocked(current []string, know if _, ok := knownKeys[busid]; !ok { continue } - if _, active := a.activeBusIDs[busid]; !active { + if a.activeBusIDs[busid] == 0 { continue } if activeCurrent == nil { diff --git a/service/usbip/client_shared.go b/service/usbip/client_shared.go index 85f354f06..272c08e67 100644 --- a/service/usbip/client_shared.go +++ b/service/usbip/client_shared.go @@ -292,28 +292,52 @@ func (c *ClientService) applyRemoteDeviceState(devices []ControlDeviceInfo) { c.applyMatchedExportsWithRetained(availableEntries, knownKeys) } +// applyRemoteExports reconciles the all-devices worker set against a +// devlist/snapshot in one workerAccess critical section. allWorkers is +// the single source of truth for running workers; diffing and mutating +// it under separate locks let a concurrent sync interleave and start a +// second worker for the same busid. func (c *ClientService) applyRemoteExports(entries []DeviceEntry) { - start, stop := c.assignment.ApplyAll(entries) - - c.workerAccess.Lock() - stopCancels := make([]context.CancelFunc, 0, len(stop)) - for _, busid := range stop { - cancel, ok := c.allWorkers[busid] - if !ok { + desired := make(map[string]struct{}, len(entries)) + for i := range entries { + busid := entries[i].Info.BusIDString() + if busid == "" { continue } - stopCancels = append(stopCancels, cancel) + desired[busid] = struct{}{} + } + + c.workerAccess.Lock() + c.assignment.SetAllDesired(desired) + var stopCancels []context.CancelFunc + for busid, worker := range c.allWorkers { + if _, wanted := desired[busid]; wanted { + continue + } + // A busy device is omitted from devlists; never stop the worker + // that is the reason it is busy. + if c.assignment.IsActive(busid) { + continue + } + stopCancels = append(stopCancels, worker.cancel) delete(c.allWorkers, busid) } + var start []string + for busid := range desired { + if _, exists := c.allWorkers[busid]; exists { + continue + } + start = append(start, busid) + } + slices.Sort(start) + for _, busid := range start { + c.startRemoteBusIDWorkerLocked(busid) + } c.workerAccess.Unlock() for _, cancel := range stopCancels { cancel() } - slices.Sort(start) - for _, busid := range start { - c.startRemoteBusIDWorker(busid, busid) - } } func (c *ClientService) applyMatchedExportsWithRetained(entries []DeviceEntry, knownKeys map[string]DeviceKey) { @@ -391,29 +415,37 @@ func (w *clientAssignedWorker) setDesiredBusID(busid string) { w.updates <- busid } -func (c *ClientService) startRemoteBusIDWorker(busid, description string) { +// startRemoteBusIDWorkerLocked must run under workerAccess. The worker +// removes itself from allWorkers when it exits on its own (export +// disappeared); the identity check keeps a stop-and-restart for the +// same busid from deleting its successor. If the busid became desired +// again while the worker was deciding to exit, restart it immediately — +// the next snapshot may be far away. +func (c *ClientService) startRemoteBusIDWorkerLocked(busid string) { runCtx, cancel := context.WithCancel(c.ctx) - - c.workerAccess.Lock() - c.allWorkers[busid] = cancel - c.workerAccess.Unlock() - + worker := &clientRemoteWorker{cancel: cancel} + c.allWorkers[busid] = worker go func() { - c.runBusIDLoop(runCtx, busid, description) + c.runBusIDLoop(runCtx, busid, busid) + cancel() + c.workerAccess.Lock() + if c.allWorkers[busid] == worker { + delete(c.allWorkers, busid) + if c.ctx.Err() == nil && c.assignment.IsRetryDesired(busid) { + c.startRemoteBusIDWorkerLocked(busid) + } + } + c.workerAccess.Unlock() }() } func (c *ClientService) stopAllWorkers() { - c.assignment.access.Lock() - c.assignment.registered = make(map[string]struct{}) - c.assignment.access.Unlock() - c.workerAccess.Lock() cancels := make([]context.CancelFunc, 0, len(c.allWorkers)) - for _, cancel := range c.allWorkers { - cancels = append(cancels, cancel) + for _, worker := range c.allWorkers { + cancels = append(cancels, worker.cancel) } - c.allWorkers = make(map[string]context.CancelFunc) + c.allWorkers = make(map[string]*clientRemoteWorker) c.workerAccess.Unlock() for _, cancel := range cancels {