From 7a8dc5e4dae6bede121c13feb69ad6a2bef316fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Fri, 15 May 2026 18:39:27 +0800 Subject: [PATCH] usbip: serialize darwin submit scheduling and commit reconcile state --- service/usbip/host.go | 8 +- service/usbip/host_darwin.go | 249 +++++++++++++++++++++++++++-------- service/usbip/host_linux.go | 157 ++++++++++++++++++---- service/usbip/server.go | 5 +- 4 files changed, 329 insertions(+), 90 deletions(-) diff --git a/service/usbip/host.go b/service/usbip/host.go index 3afaf5ac4..786b6f44a 100644 --- a/service/usbip/host.go +++ b/service/usbip/host.go @@ -8,9 +8,11 @@ import ( ) // ExportHost lifecycle: Start → Reconcile* → Close. Reconcile may be -// called many times; FinishImport runs after each data session ends so -// the platform can do post-import cleanup (Linux: write -1 to -// usbip_sockfd; Darwin: release stale-marked captures). +// called many times; it returns the committed post-reconcile export +// state, and callers must apply snapshot/released even when err != nil. +// FinishImport runs after each data session ends so the platform can do +// post-import cleanup (Linux: write -1 to usbip_sockfd; Darwin: +// release stale-marked captures). type ExportHost interface { Start(ctx context.Context) error Close() error diff --git a/service/usbip/host_darwin.go b/service/usbip/host_darwin.go index 67e15aea0..5f83628d9 100644 --- a/service/usbip/host_darwin.go +++ b/service/usbip/host_darwin.go @@ -356,7 +356,8 @@ type darwinServerDataSession struct { device *darwinUSBHostDevice writeAccess sync.Mutex access sync.Mutex - pending map[uint32]darwinServerPendingSubmit + pending map[uint32]darwinServerSubmitState + endpoints map[uint8]*darwinServerEndpointState wg sync.WaitGroup done chan struct{} @@ -370,20 +371,33 @@ type darwinServerDataSession struct { closeErr error } -type darwinServerPendingSubmit struct { +type darwinServerSubmitState struct { + command SubmitCommand endpoint uint8 + started bool unlinked bool drained chan struct{} } +type darwinServerEndpointState struct { + active uint32 + queued []uint32 +} + +type darwinServerNextSubmit struct { + sequence uint32 + command SubmitCommand +} + func newDarwinServerDataSession(ctx context.Context, logger log.ContextLogger, conn net.Conn, device *darwinUSBHostDevice) *darwinServerDataSession { return &darwinServerDataSession{ - ctx: ctx, - logger: logger, - conn: conn, - device: device, - pending: make(map[uint32]darwinServerPendingSubmit), - done: make(chan struct{}), + ctx: ctx, + logger: logger, + conn: conn, + device: device, + pending: make(map[uint32]darwinServerSubmitState), + endpoints: make(map[uint8]*darwinServerEndpointState), + done: make(chan struct{}), } } @@ -458,32 +472,23 @@ func (s *darwinServerDataSession) serve() error { if err != nil { return err } - s.trackSubmit(command.Header.SeqNum, commandEndpoint(command)) - s.wg.Add(1) - go func() { - defer s.wg.Done() - response := s.handleSubmit(command) - if !s.finishSubmit(command.Header.SeqNum) { - return - } - s.writeAccess.Lock() - err := WriteSubmitResponse(s.conn, response) - s.writeAccess.Unlock() - if err != nil { - _ = s.conn.Close() - } - }() + next, shouldStart := s.enqueueSubmit(command) + if shouldStart { + s.startSubmit(next) + } case CmdUnlink: command, err := ReadUnlinkCommandBody(s.conn, header) if err != nil { return err } status := int32(0) - endpoint, drained, found := s.markSubmitUnlinked(command.SeqNum) + endpoint, drained, shouldAbort, found := s.unlinkSubmit(command.SeqNum) if found { - abortErr := s.device.abortEndpoint(endpoint) - if abortErr != nil { - s.logger.Debug("abort endpoint 0x", hex8(endpoint), ": ", abortErr) + if shouldAbort { + abortErr := s.device.abortEndpoint(endpoint) + if abortErr != nil { + s.logger.Debug("abort endpoint 0x", hex8(endpoint), ": ", abortErr) + } } <-drained status = usbipStatusECONNRESET @@ -503,6 +508,57 @@ func (s *darwinServerDataSession) serve() error { } } +func (s *darwinServerDataSession) enqueueSubmit(command SubmitCommand) (darwinServerNextSubmit, bool) { + endpoint := submitScheduleEndpoint(command) + sequence := command.Header.SeqNum + + s.access.Lock() + defer s.access.Unlock() + + state := darwinServerSubmitState{ + command: command, + endpoint: endpoint, + } + endpointState, found := s.endpoints[endpoint] + if !found { + endpointState = &darwinServerEndpointState{} + s.endpoints[endpoint] = endpointState + } + if endpointState.active == 0 { + state.started = true + s.pending[sequence] = state + endpointState.active = sequence + return darwinServerNextSubmit{ + sequence: sequence, + command: command, + }, true + } + s.pending[sequence] = state + endpointState.queued = append(endpointState.queued, sequence) + return darwinServerNextSubmit{}, false +} + +func (s *darwinServerDataSession) startSubmit(next darwinServerNextSubmit) { + s.wg.Add(1) + go func() { + defer s.wg.Done() + + response := s.handleSubmit(next.command) + shouldSend, followUp, hasFollowUp := s.finishSubmit(next.sequence) + if shouldSend { + s.writeAccess.Lock() + err := WriteSubmitResponse(s.conn, response) + s.writeAccess.Unlock() + if err != nil { + _ = s.conn.Close() + } + } + if hasFollowUp { + s.startSubmit(followUp) + } + }() +} + func (s *darwinServerDataSession) handleSubmit(command SubmitCommand) SubmitResponse { response := SubmitResponse{ Header: DataHeader{ @@ -589,64 +645,128 @@ func packIsoInResponseBuffer(buffer []byte, packets []IsoPacketDescriptor) []byt return packed } -func (s *darwinServerDataSession) trackSubmit(seq uint32, endpoint uint8) { - s.access.Lock() - defer s.access.Unlock() - s.pending[seq] = darwinServerPendingSubmit{endpoint: endpoint} -} +func (s *darwinServerDataSession) unlinkSubmit(seq uint32) (uint8, <-chan struct{}, bool, bool) { + var drained chan struct{} -func (s *darwinServerDataSession) markSubmitUnlinked(seq uint32) (uint8, <-chan struct{}, bool) { - s.access.Lock() - defer s.access.Unlock() - pending, found := s.pending[seq] - if !found { - return 0, nil, false - } - pending.unlinked = true - if pending.drained == nil { - pending.drained = make(chan struct{}) - } - s.pending[seq] = pending - return pending.endpoint, pending.drained, true -} - -func (s *darwinServerDataSession) finishSubmit(seq uint32) bool { s.access.Lock() pending, found := s.pending[seq] if !found { s.access.Unlock() - return true + return 0, nil, false, false + } + if pending.drained == nil { + pending.drained = make(chan struct{}) + } + drained = pending.drained + if !pending.started { + endpointState := s.endpoints[pending.endpoint] + if endpointState != nil { + endpointState.queued = removeQueuedSequence(endpointState.queued, seq) + if endpointState.active == 0 && len(endpointState.queued) == 0 { + delete(s.endpoints, pending.endpoint) + } + } + delete(s.pending, seq) + s.access.Unlock() + close(drained) + return pending.endpoint, drained, false, true + } + shouldAbort := !pending.unlinked + pending.unlinked = true + s.pending[seq] = pending + s.access.Unlock() + return pending.endpoint, drained, shouldAbort, true +} + +func (s *darwinServerDataSession) finishSubmit(seq uint32) (bool, darwinServerNextSubmit, bool) { + var drained chan struct{} + var followUp darwinServerNextSubmit + var hasFollowUp bool + + s.access.Lock() + pending, found := s.pending[seq] + if !found { + s.access.Unlock() + return true, darwinServerNextSubmit{}, false + } + endpointState := s.endpoints[pending.endpoint] + if endpointState != nil && endpointState.active == seq { + endpointState.active = 0 } delete(s.pending, seq) - drained := pending.drained + if endpointState != nil { + for len(endpointState.queued) > 0 { + nextSequence := endpointState.queued[0] + endpointState.queued = endpointState.queued[1:] + nextPending, nextFound := s.pending[nextSequence] + if !nextFound { + continue + } + nextPending.started = true + s.pending[nextSequence] = nextPending + endpointState.active = nextSequence + followUp = darwinServerNextSubmit{ + sequence: nextSequence, + command: nextPending.command, + } + hasFollowUp = true + break + } + if endpointState.active == 0 && len(endpointState.queued) == 0 { + delete(s.endpoints, pending.endpoint) + } + } + drained = pending.drained unlinked := pending.unlinked s.access.Unlock() if drained != nil { close(drained) } - return !unlinked + return !unlinked, followUp, hasFollowUp } func (s *darwinServerDataSession) abortPendingSubmits() { + var ( + activeEndpoints []uint8 + drained []chan struct{} + ) + s.access.Lock() seen := make(map[uint8]struct{}) for seq, pending := range s.pending { + if !pending.started { + delete(s.pending, seq) + if pending.drained != nil { + drained = append(drained, pending.drained) + } + continue + } if !pending.unlinked { seen[pending.endpoint] = struct{}{} } pending.unlinked = true s.pending[seq] = pending } - endpoints := make([]uint8, 0, len(seen)) - for endpoint := range seen { - endpoints = append(endpoints, endpoint) + for endpoint := range s.endpoints { + endpointState := s.endpoints[endpoint] + if endpointState != nil { + endpointState.queued = nil + } } - slices.Sort(endpoints) s.access.Unlock() + + for _, drainedChannel := range drained { + close(drainedChannel) + } + activeEndpoints = make([]uint8, 0, len(seen)) + for endpoint := range seen { + activeEndpoints = append(activeEndpoints, endpoint) + } + slices.Sort(activeEndpoints) if s.device == nil { return } - for _, endpoint := range endpoints { + for _, endpoint := range activeEndpoints { err := s.device.abortEndpoint(endpoint) if err != nil { s.logger.Debug("abort endpoint 0x", hex8(endpoint), ": ", err) @@ -654,6 +774,23 @@ func (s *darwinServerDataSession) abortPendingSubmits() { } } +func removeQueuedSequence(queue []uint32, sequence uint32) []uint32 { + for index, current := range queue { + if current != sequence { + continue + } + return append(queue[:index], queue[index+1:]...) + } + return queue +} + +func submitScheduleEndpoint(command SubmitCommand) uint8 { + if command.Header.Endpoint == 0 { + return 0 + } + return commandEndpoint(command) +} + func commandEndpoint(command SubmitCommand) uint8 { endpoint := uint8(command.Header.Endpoint & 0x0f) if command.Header.Direction == USBIPDirIn { diff --git a/service/usbip/host_linux.go b/service/usbip/host_linux.go index 0e0f05949..1fcfedf60 100644 --- a/service/usbip/host_linux.go +++ b/service/usbip/host_linux.go @@ -324,41 +324,79 @@ func (h *linuxExportHost) Reconcile(ctx context.Context, isBusy func(busid strin h.access.Unlock() plan := classifyLinuxReconcile(current, desired, isBusy) - toAdd := make([]*linuxExport, 0, len(plan.toBind)) - for _, exp := range plan.toRelease { - err := h.releaseExport(exp, false) - if err != nil { - h.logger.Warn("release ", exp.busid, ": ", err) + committed := make(map[string]*linuxExport, len(current)+len(plan.toBind)) + maps.Copy(committed, current) + var reconcileErrors []error + + for _, busid := range plan.toStale { + exp, found := committed[busid] + if found { + staleExport := cloneLinuxExport(exp) + staleExport.stale = true + committed[busid] = staleExport } } + for _, exp := range plan.toRelease { + releaseErr := h.releaseExport(exp, false) + if releaseErr != nil { + h.logger.Warn("release ", exp.busid, ": ", releaseErr) + reconcileErrors = append(reconcileErrors, E.Cause(releaseErr, "release ", exp.busid)) + } + var desiredDevice *sysfsDevice + desiredEntry, found := desired[exp.busid] + if found { + desiredDevice = &desiredEntry + } + resolved, resolveErr := h.resolveCommittedRelease(exp, desiredDevice) + if resolveErr != nil { + reconcileErrors = append(reconcileErrors, resolveErr) + } + if resolved == nil { + delete(committed, exp.busid) + continue + } + committed[exp.busid] = resolved + } + for busid, device := range plan.toBind { - exp, bindErr := h.bindOne(&device) - if bindErr != nil { - return h.snapshotSelf(), nil, E.Cause(bindErr, "bind ", busid) + _, found := committed[busid] + if found { + continue } - toAdd = append(toAdd, exp) + previousDriver, probeErr := currentDriver(busid) + if probeErr != nil { + reconcileErrors = append(reconcileErrors, E.Cause(probeErr, "probe driver before bind ", busid)) + } + exp, bindErr := h.bindOne(&device) + if bindErr == nil { + committed[busid] = exp + continue + } + reconcileErrors = append(reconcileErrors, E.Cause(bindErr, "bind ", busid)) + resolved, resolveErr := h.resolveCommittedBind(busid, &device, previousDriver) + if resolveErr != nil { + reconcileErrors = append(reconcileErrors, resolveErr) + } + if resolved != nil { + committed[busid] = resolved + } + } + + released := make([]string, 0, len(plan.released)) + for _, busid := range plan.released { + exp, found := committed[busid] + if found && !exp.stale { + continue + } + released = append(released, busid) } h.access.Lock() - for _, busid := range plan.toStale { - exp, ok := h.exports[busid] - if ok { - exp.stale = true - } - } - for _, exp := range plan.toRelease { - currentExport, ok := h.exports[exp.busid] - if ok && currentExport == exp { - delete(h.exports, exp.busid) - } - } - for _, exp := range toAdd { - h.exports[exp.busid] = exp - } + h.exports = committed h.access.Unlock() - return h.snapshotSelf(), plan.released, nil + return snapshotLinuxExports(committed), released, E.Errors(reconcileErrors...) } func (h *linuxExportHost) FinishImport(ctx context.Context, busid string) (bool, error) { @@ -389,8 +427,12 @@ func (h *linuxExportHost) FinishImport(ctx context.Context, busid string) (bool, func (h *linuxExportHost) snapshotSelf() map[string]Export { h.access.Lock() defer h.access.Unlock() - out := make(map[string]Export, len(h.exports)) - for busid, exp := range h.exports { + return snapshotLinuxExports(h.exports) +} + +func snapshotLinuxExports(exports map[string]*linuxExport) map[string]Export { + out := make(map[string]Export, len(exports)) + for busid, exp := range exports { if exp.stale { continue } @@ -399,6 +441,67 @@ func (h *linuxExportHost) snapshotSelf() map[string]Export { return out } +func cloneLinuxExport(exp *linuxExport) *linuxExport { + if exp == nil { + return nil + } + clone := *exp + clone.descriptor.Interfaces = slices.Clone(exp.descriptor.Interfaces) + clone.identity.Interfaces = slices.Clone(exp.identity.Interfaces) + return &clone +} + +func (h *linuxExportHost) resolveCommittedRelease(exp *linuxExport, desired *sysfsDevice) (*linuxExport, error) { + if desired != nil { + resolved, found, err := h.probeDesiredBoundExport(exp.busid, desired, exp.managed, exp.originalDriver) + if err != nil { + return exp, err + } + if found { + return resolved, nil + } + } + driver, err := currentDriver(exp.busid) + if err != nil { + return exp, E.Cause(err, "probe driver ", exp.busid) + } + if driver != "usbip-host" { + return nil, nil + } + return exp, nil +} + +func (h *linuxExportHost) resolveCommittedBind(busid string, desired *sysfsDevice, originalDriver string) (*linuxExport, error) { + if desired == nil { + return nil, nil + } + resolved, found, err := h.probeDesiredBoundExport(busid, desired, true, originalDriver) + if err != nil { + return nil, err + } + if !found { + return nil, nil + } + return resolved, nil +} + +func (h *linuxExportHost) probeDesiredBoundExport(busid string, desired *sysfsDevice, managed bool, originalDriver string) (*linuxExport, bool, error) { + driver, err := currentDriver(busid) + if err != nil { + return nil, false, E.Cause(err, "probe driver ", busid) + } + if driver != "usbip-host" { + return nil, false, nil + } + descriptor, err := readSysfsDevice(busid, filepath.Join(sysBusUSBDevices, busid)) + if err != nil { + descriptor = *desired + } else if !newLinuxExportIdentity(descriptor).Equal(newLinuxExportIdentity(*desired)) { + return nil, false, nil + } + return h.newExport(descriptor, managed, originalDriver), true, nil +} + func (h *linuxExportHost) bindOne(d *sysfsDevice) (*linuxExport, error) { var ( exp *linuxExport diff --git a/service/usbip/server.go b/service/usbip/server.go index 556f71ef8..79be54c8f 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -168,16 +168,13 @@ func (s *ServerService) reconcileAndBroadcast(notify bool) error { return nil } snapshot, released, err := s.host.Reconcile(s.ctx, s.ledger.IsBusy) - if err != nil { - return err - } s.ledger.ApplyHostSnapshot(snapshot, released) if notify { s.ledger.BroadcastIfChanged(s.ctx) } else { s.ledger.SeedBroadcastState(s.ctx) } - return nil + return err } func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) {