From bee231b9ff407e81a1787761d4c1d72ecac40c3a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Tue, 12 May 2026 20:14:50 +0800 Subject: [PATCH] usbip: collapse server state into exportLedger ServerService used to own three mutexes coordinating six pieces of mutable state (exports, busy, controlState, controlSeq, controlSubs, LeaseManager). The lock-ordering rule that fixed the import/attach race in 0c892411a lived only as a comment. Two-mutex (fast/slow) ledger absorbs all of it; the two mutexes are never held simultaneously, so the ordering rule becomes a property of the type instead of a discipline callers must observe. TryReserveForImport closes the window where two concurrent OpReqImports could both pass the busy check before either marked busy. --- service/usbip/control_lease.go | 52 --- service/usbip/control_server.go | 130 ------ service/usbip/darwin_integration_test.go | 4 +- service/usbip/export_ledger.go | 561 +++++++++++++++++++++++ service/usbip/export_ledger_test.go | 439 ++++++++++++++++++ service/usbip/lease.go | 124 ----- service/usbip/lease_test.go | 248 ---------- service/usbip/linux_interop_test.go | 4 +- service/usbip/linux_test.go | 89 ++-- service/usbip/server.go | 208 +-------- service/usbip/server_darwin_test.go | 62 ++- service/usbip/server_shared.go | 9 +- 12 files changed, 1112 insertions(+), 818 deletions(-) delete mode 100644 service/usbip/control_lease.go delete mode 100644 service/usbip/control_server.go create mode 100644 service/usbip/export_ledger.go create mode 100644 service/usbip/export_ledger_test.go delete mode 100644 service/usbip/lease.go delete mode 100644 service/usbip/lease_test.go diff --git a/service/usbip/control_lease.go b/service/usbip/control_lease.go deleted file mode 100644 index 92f8e0969..000000000 --- a/service/usbip/control_lease.go +++ /dev/null @@ -1,52 +0,0 @@ -//go:build linux || (darwin && cgo) - -package usbip - -func (s *ServerService) handleControlLeaseRequest(sub *serverControlConn, payload []byte) { - var request controlLeaseRequest - err := unmarshalControlPayload(payload, &request) - if err != nil { - s.enqueueControlPayload(sub, controlFrame{ - Type: controlFrameLeaseResponse, - Version: controlProtocolVersion, - }, controlLeaseResponse{ - ErrorCode: leaseErrorBadRequest, - ErrorMessage: err.Error(), - }, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: s.currentControlSequence()}) - return - } - response := s.createControlLeaseResponse(sub.id, request) - s.enqueueControlPayload(sub, controlFrame{ - Type: controlFrameLeaseResponse, - Version: controlProtocolVersion, - }, response, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: s.currentControlSequence()}) -} - -func (s *ServerService) createControlLeaseResponse(subID uint64, request controlLeaseRequest) controlLeaseResponse { - if request.BusID == "" { - return controlLeaseResponse{ - BusID: request.BusID, - ClientNonce: request.ClientNonce, - ErrorCode: leaseErrorBadRequest, - ErrorMessage: "missing busid", - } - } - s.controlAccess.Lock() - generation := s.controlSeq - s.controlAccess.Unlock() - return s.leases.Issue(subID, generation, request, func() (bool, string) { - return s.leaseAvailable(request.BusID) - }) -} - -func (s *ServerService) consumeImportLease(request ImportExtRequest) bool { - s.controlAccess.Lock() - defer s.controlAccess.Unlock() - return s.leases.Consume(s.controlSeq, request) -} - -func (s *ServerService) currentControlSequence() uint64 { - s.controlAccess.Lock() - defer s.controlAccess.Unlock() - return s.controlSeq -} diff --git a/service/usbip/control_server.go b/service/usbip/control_server.go deleted file mode 100644 index 093b532ae..000000000 --- a/service/usbip/control_server.go +++ /dev/null @@ -1,130 +0,0 @@ -//go:build linux || (darwin && cgo) - -package usbip - -import ( - "net" -) - -type serverControlConn struct { - id uint64 - capabilities uint32 - conn net.Conn - send chan controlMessage -} - -func (s *ServerService) registerControlConn(conn net.Conn, capabilities uint32) (*serverControlConn, uint64) { - s.controlAccess.Lock() - defer s.controlAccess.Unlock() - s.controlNextID++ - sequence := s.controlSeq - sub := &serverControlConn{ - id: s.controlNextID, - capabilities: capabilities, - conn: conn, - send: make(chan controlMessage, 16), - } - if supportsControlExtensions(capabilities) { - s.enqueueControlSnapshot(sub, sequence) - } - s.controlSubs[sub.id] = sub - return sub, sequence -} - -func (s *ServerService) unregisterControlConn(id uint64) { - s.controlAccess.Lock() - defer s.controlAccess.Unlock() - delete(s.controlSubs, id) - s.leases.RevokeSubscriber(id) -} - -func (s *ServerService) closeControlSubscribers() { - s.controlAccess.Lock() - subs := make([]*serverControlConn, 0, len(s.controlSubs)) - for _, sub := range s.controlSubs { - subs = append(subs, sub) - } - s.controlSubs = make(map[uint64]*serverControlConn) - s.controlAccess.Unlock() - for _, sub := range subs { - _ = sub.conn.Close() - } -} - -func (s *ServerService) broadcastControlState(nextState map[string]DeviceInfoV2, force bool) bool { - s.controlAccess.Lock() - nextSequence := s.controlSeq + 1 - delta := buildControlDeviceDelta(nextSequence, s.controlState, nextState) - if !force && controlDeviceDeltaEmpty(delta) { - s.controlState = nextState - s.controlAccess.Unlock() - return false - } - s.controlSeq = nextSequence - sequence := s.controlSeq - s.controlState = nextState - subs := make([]*serverControlConn, 0, len(s.controlSubs)) - for _, sub := range s.controlSubs { - subs = append(subs, sub) - } - s.controlAccess.Unlock() - - frame := controlFrame{ - Type: controlFrameChanged, - Version: controlProtocolVersion, - Sequence: sequence, - } - for _, sub := range subs { - if supportsControlExtensions(sub.capabilities) { - s.enqueueControlPayload(sub, controlFrame{ - Type: controlFrameDeviceDelta, - Version: controlProtocolVersion, - Sequence: sequence, - }, delta, frame) - continue - } - s.enqueueControlFrame(sub, frame) - } - return true -} - -func (s *ServerService) enqueueControlFrame(sub *serverControlConn, frame controlFrame) { - s.enqueueControlMessage(sub, controlMessage{Frame: frame}) -} - -func (s *ServerService) enqueueControlPayload(sub *serverControlConn, frame controlFrame, payload any, fallback controlFrame) { - rawPayload, err := marshalControlPayload(payload) - if err != nil || len(rawPayload) > maxControlPayloadLength { - s.enqueueControlFrame(sub, fallback) - return - } - s.enqueueControlMessage(sub, controlMessage{Frame: frame, Payload: rawPayload}) -} - -func (s *ServerService) enqueueControlSnapshot(sub *serverControlConn, sequence uint64) { - devices := s.buildDeviceStateV2() - s.enqueueControlPayload(sub, controlFrame{ - Type: controlFrameDeviceSnapshot, - Version: controlProtocolVersion, - Sequence: sequence, - }, controlDeviceSnapshot{Sequence: sequence, Devices: devices}, controlFrame{ - Type: controlFrameChanged, - Version: controlProtocolVersion, - Sequence: sequence, - }) -} - -func (s *ServerService) enqueueControlMessage(sub *serverControlConn, message controlMessage) { - select { - case sub.send <- message: - default: - s.logger.Debug("control subscriber ", sub.id, " lagged behind") - _ = sub.conn.Close() - } -} - -func (s *ServerService) setControlState(nextState map[string]DeviceInfoV2) { - s.controlAccess.Lock() - s.controlState = nextState - s.controlAccess.Unlock() -} diff --git a/service/usbip/darwin_integration_test.go b/service/usbip/darwin_integration_test.go index ffec0b307..ac4eed635 100644 --- a/service/usbip/darwin_integration_test.go +++ b/service/usbip/darwin_integration_test.go @@ -851,11 +851,11 @@ func TestDarwinUSBIPServerSmoke(t *testing.T) { if err := server.Start(adapter.StartStateStart); err != nil { t.Skipf("IOUSBHostDevice enumeration unavailable: %v", err) } - if len(server.currentExports()) == 0 { + if len(server.ledger.AvailableExports()) == 0 { t.Skipf("IOUSBHostDevice capture unavailable for %s", candidate.key.BusID) } - destination := M.SocksaddrFromNet(server.listen.Addr()) + destination := M.SocksaddrFromNet(server.listener.TCPListener().Addr()) entries := darwinFetchDevList(t, destination) require.Len(t, entries, 1) require.Equal(t, candidate.key.BusID, entries[0].Info.BusIDString()) diff --git a/service/usbip/export_ledger.go b/service/usbip/export_ledger.go new file mode 100644 index 000000000..85b0d1a34 --- /dev/null +++ b/service/usbip/export_ledger.go @@ -0,0 +1,561 @@ +//go:build linux || (darwin && cgo) + +package usbip + +import ( + "context" + "net" + "slices" + "strings" + "sync" + "time" + + "github.com/sagernet/sing-box/log" +) + +// exportLedger owns the server's authoritative mutable state: which +// devices it is exporting, which busids are currently in use, who is +// subscribed to control-channel updates, and which import leases are +// outstanding. It absorbs what previously lived as four fields plus a +// LeaseManager on ServerService. +// +// Synchronization: +// +// The ledger uses two internal mutexes. The two are NEVER held +// simultaneously. Each public method acquires at most one at a time +// for any nested duration; multi-stage methods acquire and release +// one, do unlocked work (including syscalls), then acquire the +// other. +// +// fast: broadcast bookkeeping. seq, nextSubID, subs, state. +// slow: inventory and leases. exports, busy, leases, nextLeaseID. +// +// The previous controlAccess -> LeaseManager.access -> access +// ordering documented in lease.go disappears because the only +// remaining cross-state operation (IssueLease) reads seq under fast, +// releases, then uses slow. +type exportLedger struct { + logger log.ContextLogger + now func() time.Time + ttl time.Duration + + fast sync.Mutex + seq uint64 + nextSubID uint64 + subs map[uint64]*exportSubscriber + state map[string]DeviceInfoV2 + + slow sync.Mutex + exports map[string]Export + busy map[string]bool + leases map[string]serverImportLease + nextLeaseID uint64 +} + +// exportSubscriber is one live control-channel connection. The send +// channel is filled by the ledger's broadcast methods and drained by +// the transport handler. +type exportSubscriber struct { + id uint64 + capabilities uint32 + conn net.Conn + send chan controlMessage +} + +const controlSubscriberSendBuffer = 16 + +func newExportLedger(logger log.ContextLogger, ttl time.Duration, now func() time.Time) *exportLedger { + if now == nil { + now = time.Now + } + return &exportLedger{ + logger: logger, + now: now, + ttl: ttl, + subs: make(map[uint64]*exportSubscriber), + state: make(map[string]DeviceInfoV2), + exports: make(map[string]Export), + busy: make(map[string]bool), + leases: make(map[string]serverImportLease), + } +} + +// IsBusy reports whether the busid currently has an active import. The +// signature matches ExportHost.Reconcile's isBusy callback. +func (l *exportLedger) IsBusy(busid string) bool { + l.slow.Lock() + defer l.slow.Unlock() + return l.busy[busid] +} + +// Export returns the export registered for busid, if any. +func (l *exportLedger) Export(busid string) (Export, bool) { + l.slow.Lock() + defer l.slow.Unlock() + export, found := l.exports[busid] + return export, found +} + +// AvailableExports returns exports that are not currently busy, sorted +// by busid. The slice references are stable; Snapshot may be called on +// each entry outside the ledger's lock. +func (l *exportLedger) AvailableExports() []Export { + l.slow.Lock() + out := make([]Export, 0, len(l.exports)) + for busid, export := range l.exports { + if l.busy[busid] { + continue + } + out = append(out, export) + } + l.slow.Unlock() + slices.SortFunc(out, exportLess) + return out +} + +func exportLess(a, b Export) int { + return strings.Compare(a.BusID(), b.BusID()) +} + +// ApplyHostSnapshot replaces the inventory with snapshot and clears +// busy entries for released busids. Does not broadcast — callers pair +// this with SeedBroadcastState (quiet) or BroadcastIfChanged. +func (l *exportLedger) ApplyHostSnapshot(snapshot map[string]Export, released []string) { + l.slow.Lock() + l.exports = snapshot + for _, busid := range released { + delete(l.busy, busid) + } + l.slow.Unlock() +} + +// SeedBroadcastState recomputes the broadcast state and stores it +// without emitting any frame. Used at Start. +func (l *exportLedger) SeedBroadcastState(ctx context.Context) { + nextState := deviceInfoV2Map(l.snapshotDeviceState(ctx)) + l.fast.Lock() + l.state = nextState + l.fast.Unlock() +} + +// BroadcastIfChanged recomputes the broadcast state, compares against +// the last broadcast, and emits a DeviceDelta + Changed frame to every +// subscriber if the state moved. Returns true if a frame was sent. +func (l *exportLedger) BroadcastIfChanged(ctx context.Context) bool { + nextState := deviceInfoV2Map(l.snapshotDeviceState(ctx)) + + l.fast.Lock() + nextSequence := l.seq + 1 + delta := buildControlDeviceDelta(nextSequence, l.state, nextState) + if controlDeviceDeltaEmpty(delta) { + l.state = nextState + l.fast.Unlock() + return false + } + l.seq = nextSequence + sequence := l.seq + l.state = nextState + targets := l.snapshotSubsLocked() + l.fast.Unlock() + + frame := controlFrame{ + Type: controlFrameChanged, + Version: controlProtocolVersion, + Sequence: sequence, + } + for _, sub := range targets { + if supportsControlExtensions(sub.capabilities) { + l.enqueuePayload(sub, controlFrame{ + Type: controlFrameDeviceDelta, + Version: controlProtocolVersion, + Sequence: sequence, + }, delta, frame) + continue + } + l.enqueueFrame(sub, frame) + } + return true +} + +// TryReserveForImport atomically checks that busid is exported, not +// busy, and currently lease-available, marking it busy on success. The +// caller is responsible for follow-up DeviceInfo/NewServerDataSession +// and must call ReleaseImport on any failure path or ConfirmImport on +// success. +// +// The Export.LeaseCheck syscall runs outside the slow lock; the busy +// mark is only inserted after the lease check passes and the second +// availability re-check confirms no other goroutine raced in. +func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (Export, bool, string) { + export, reason, ok := l.checkAvailableUnderLock(busid) + if !ok { + return nil, false, reason + } + leaseOK, leaseReason := export.LeaseCheck(ctx) + if !leaseOK { + return nil, false, leaseReason + } + l.slow.Lock() + defer l.slow.Unlock() + current, stillExported := l.exports[busid] + if !stillExported || current != export { + return nil, false, "unknown busid" + } + if l.busy[busid] { + return nil, false, deviceStateBusy + } + l.busy[busid] = true + return export, true, "" +} + +func (l *exportLedger) checkAvailableUnderLock(busid string) (Export, string, bool) { + l.slow.Lock() + defer l.slow.Unlock() + export, found := l.exports[busid] + if !found { + return nil, "unknown busid", false + } + if l.busy[busid] { + return nil, deviceStateBusy, false + } + return export, "", true +} + +// ConfirmImport broadcasts that an import is now active. The busy mark +// was set during TryReserveForImport; this method only broadcasts so +// clients see the state change. +func (l *exportLedger) ConfirmImport(ctx context.Context) { + l.BroadcastIfChanged(ctx) +} + +// ReleaseImport clears the busy mark for busid, optionally removing +// the export entirely, and broadcasts the change. removeExport=true is +// used when the platform host's FinishImport returns released=true +// (e.g. Darwin stale capture). +func (l *exportLedger) ReleaseImport(ctx context.Context, busid string, removeExport bool) { + l.slow.Lock() + delete(l.busy, busid) + if removeExport { + delete(l.exports, busid) + } + l.slow.Unlock() + l.BroadcastIfChanged(ctx) +} + +// IssueLease validates the request and inserts a fresh lease keyed by +// busid. The seq generation is captured at entry and stored on the +// lease so a subsequent ConsumeLease can reject stale leases issued +// before a topology change. +func (l *exportLedger) IssueLease(ctx context.Context, subID uint64, request controlLeaseRequest) controlLeaseResponse { + response := controlLeaseResponse{ + BusID: request.BusID, + ClientNonce: request.ClientNonce, + } + if request.BusID == "" { + response.ErrorCode = leaseErrorBadRequest + response.ErrorMessage = "missing busid" + return response + } + + l.fast.Lock() + generation := l.seq + l.fast.Unlock() + + export, errorCode, reason, ok := l.checkAvailableForLease(request.BusID) + if !ok { + response.ErrorCode = errorCode + response.ErrorMessage = reason + return response + } + leaseOK, leaseReason := export.LeaseCheck(ctx) + if !leaseOK { + response.ErrorCode = leaseErrorUnavailable + response.ErrorMessage = leaseReason + return response + } + + l.slow.Lock() + defer l.slow.Unlock() + now := l.now() + l.cleanupExpiredLocked(now) + current, stillExported := l.exports[request.BusID] + if !stillExported || current != export { + response.ErrorCode = leaseErrorUnavailable + response.ErrorMessage = "unknown busid" + return response + } + if l.busy[request.BusID] { + response.ErrorCode = leaseErrorUnavailable + response.ErrorMessage = deviceStateBusy + return response + } + if _, exists := l.leases[request.BusID]; exists { + response.ErrorCode = leaseErrorBusy + response.ErrorMessage = "lease already active" + return response + } + l.nextLeaseID++ + lease := serverImportLease{ + ID: l.nextLeaseID, + SubscriberID: subID, + BusID: request.BusID, + ClientNonce: request.ClientNonce, + Generation: generation, + Expires: now.Add(l.ttl), + } + l.leases[request.BusID] = lease + response.LeaseID = lease.ID + response.Generation = lease.Generation + response.TTLMillis = int64(l.ttl / time.Millisecond) + return response +} + +func (l *exportLedger) checkAvailableForLease(busid string) (Export, string, string, bool) { + l.slow.Lock() + defer l.slow.Unlock() + now := l.now() + l.cleanupExpiredLocked(now) + export, found := l.exports[busid] + if !found { + return nil, leaseErrorUnavailable, "unknown busid", false + } + if l.busy[busid] { + return nil, leaseErrorUnavailable, deviceStateBusy, false + } + if _, exists := l.leases[busid]; exists { + return nil, leaseErrorBusy, "lease already active", false + } + return export, "", "", true +} + +// ConsumeLease validates and removes the lease matching request, then +// checks the lease's generation against the current seq. Returns true +// only if all checks pass. Consume-on-read semantics: the entry is +// removed regardless of outcome (except on mismatched nonce, which +// preserves the lease for the legitimate holder). +func (l *exportLedger) ConsumeLease(request ImportExtRequest) bool { + l.slow.Lock() + now := l.now() + l.cleanupExpiredLocked(now) + lease, found := l.leases[request.BusID] + if !found { + l.slow.Unlock() + return false + } + if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { + l.slow.Unlock() + return false + } + delete(l.leases, request.BusID) + leaseExpiry := lease.Expires + leaseGeneration := lease.Generation + l.slow.Unlock() + if !now.Before(leaseExpiry) { + return false + } + l.fast.Lock() + currentGeneration := l.seq + l.fast.Unlock() + return leaseGeneration == currentGeneration +} + +func (l *exportLedger) cleanupExpiredLocked(now time.Time) { + for busid, lease := range l.leases { + if !now.Before(lease.Expires) { + delete(l.leases, busid) + } + } +} + +// Subscribe registers a new control-channel connection. If +// capabilities include the device-snapshot extension, a freshly +// computed snapshot is enqueued so the new subscriber sees current +// state regardless of when the last broadcast fired. Returns the +// subscriber and the current sequence. +// +// Does NOT mutate l.state: other subscribers must still receive the +// next BroadcastIfChanged delta against the previous baseline. +func (l *exportLedger) Subscribe(ctx context.Context, conn net.Conn, capabilities uint32) (*exportSubscriber, uint64) { + snapshot := l.snapshotDeviceState(ctx) + l.fast.Lock() + defer l.fast.Unlock() + l.nextSubID++ + sub := &exportSubscriber{ + id: l.nextSubID, + capabilities: capabilities, + conn: conn, + send: make(chan controlMessage, controlSubscriberSendBuffer), + } + sequence := l.seq + if supportsControlExtensions(capabilities) { + l.enqueueSnapshotLocked(sub, sequence, snapshot) + } + l.subs[sub.id] = sub + return sub, sequence +} + +// Unsubscribe removes a control connection and revokes any outstanding +// leases owned by it. The subscriber's send channel is left for the GC +// to reclaim — the transport read loop already exited. +func (l *exportLedger) Unsubscribe(sub *exportSubscriber) { + l.fast.Lock() + delete(l.subs, sub.id) + l.fast.Unlock() + l.revokeLeasesForSubscriber(sub.id) +} + +func (l *exportLedger) revokeLeasesForSubscriber(subID uint64) { + l.slow.Lock() + defer l.slow.Unlock() + for busid, lease := range l.leases { + if lease.SubscriberID == subID { + delete(l.leases, busid) + } + } +} + +// CloseAllSubscribers drains every subscriber and returns the +// underlying connections so the caller can close them outside any +// lock. +func (l *exportLedger) CloseAllSubscribers() []net.Conn { + l.fast.Lock() + conns := make([]net.Conn, 0, len(l.subs)) + for _, sub := range l.subs { + conns = append(conns, sub.conn) + } + l.subs = make(map[uint64]*exportSubscriber) + l.fast.Unlock() + return conns +} + +// ResetForClose clears inventory and busy state. Called from Close +// after the host has been shut down. +func (l *exportLedger) ResetForClose() { + l.slow.Lock() + l.exports = make(map[string]Export) + l.busy = make(map[string]bool) + l.leases = make(map[string]serverImportLease) + l.slow.Unlock() +} + +// CurrentSequence returns the last broadcast sequence (for transport +// code that writes the initial ack frame). +func (l *exportLedger) CurrentSequence() uint64 { + l.fast.Lock() + defer l.fast.Unlock() + return l.seq +} + +// HandleControlPing enqueues a Pong response on sub. +func (l *exportLedger) HandleControlPing(sub *exportSubscriber) { + l.enqueueFrame(sub, controlFrame{ + Type: controlFramePong, + Version: controlProtocolVersion, + }) +} + +// HandleControlLeaseRequest parses payload, issues a lease, and +// enqueues the resulting LeaseResponse (with a Changed fallback if the +// payload exceeds maxControlPayloadLength). +func (l *exportLedger) HandleControlLeaseRequest(ctx context.Context, sub *exportSubscriber, payload []byte) { + var request controlLeaseRequest + err := unmarshalControlPayload(payload, &request) + if err != nil { + l.enqueuePayload(sub, controlFrame{ + Type: controlFrameLeaseResponse, + Version: controlProtocolVersion, + }, controlLeaseResponse{ + ErrorCode: leaseErrorBadRequest, + ErrorMessage: err.Error(), + }, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: l.CurrentSequence()}) + return + } + response := l.IssueLease(ctx, sub.id, request) + l.enqueuePayload(sub, controlFrame{ + Type: controlFrameLeaseResponse, + Version: controlProtocolVersion, + }, response, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: l.CurrentSequence()}) +} + +// snapshotDeviceState gathers refs under slow, releases, then calls +// Export.Snapshot for each entry outside the lock. The busy value +// captured at copy time is what's reported to clients. +func (l *exportLedger) snapshotDeviceState(ctx context.Context) []DeviceInfoV2 { + type entry struct { + export Export + busy bool + } + l.slow.Lock() + entries := make([]entry, 0, len(l.exports)) + for busid, export := range l.exports { + entries = append(entries, entry{export: export, busy: l.busy[busid]}) + } + l.slow.Unlock() + if len(entries) == 0 { + return nil + } + slices.SortFunc(entries, func(a, b entry) int { + return strings.Compare(a.export.BusID(), b.export.BusID()) + }) + out := make([]DeviceInfoV2, 0, len(entries)) + for _, e := range entries { + snapshot := e.export.Snapshot(ctx, e.busy) + if snapshot.Err != nil { + out = append(out, DeviceInfoV2{ + BusID: e.export.BusID(), + Backend: snapshot.Backend, + StableID: snapshot.StableID, + State: deviceStateUnavailable, + StatusReason: snapshot.StatusReason, + }) + continue + } + if snapshot.State == deviceStateUnavailable && snapshot.Entry.Info.IDVendor == 0 { + continue + } + out = append(out, deviceInfoV2FromEntry(snapshot.Entry, snapshot.Backend, snapshot.StableID, snapshot.State, snapshot.RawStatus, snapshot.StatusReason)) + } + return out +} + +func (l *exportLedger) snapshotSubsLocked() []*exportSubscriber { + out := make([]*exportSubscriber, 0, len(l.subs)) + for _, sub := range l.subs { + out = append(out, sub) + } + return out +} + +func (l *exportLedger) enqueueSnapshotLocked(sub *exportSubscriber, sequence uint64, devices []DeviceInfoV2) { + l.enqueuePayload(sub, controlFrame{ + Type: controlFrameDeviceSnapshot, + Version: controlProtocolVersion, + Sequence: sequence, + }, controlDeviceSnapshot{Sequence: sequence, Devices: devices}, controlFrame{ + Type: controlFrameChanged, + Version: controlProtocolVersion, + Sequence: sequence, + }) +} + +func (l *exportLedger) enqueueFrame(sub *exportSubscriber, frame controlFrame) { + l.deliver(sub, controlMessage{Frame: frame}) +} + +func (l *exportLedger) enqueuePayload(sub *exportSubscriber, frame controlFrame, payload any, fallback controlFrame) { + rawPayload, err := marshalControlPayload(payload) + if err != nil || len(rawPayload) > maxControlPayloadLength { + l.enqueueFrame(sub, fallback) + return + } + l.deliver(sub, controlMessage{Frame: frame, Payload: rawPayload}) +} + +func (l *exportLedger) deliver(sub *exportSubscriber, message controlMessage) { + select { + case sub.send <- message: + default: + l.logger.Debug("control subscriber ", sub.id, " lagged behind") + _ = sub.conn.Close() + } +} diff --git a/service/usbip/export_ledger_test.go b/service/usbip/export_ledger_test.go new file mode 100644 index 000000000..d733250fb --- /dev/null +++ b/service/usbip/export_ledger_test.go @@ -0,0 +1,439 @@ +//go:build linux || (darwin && cgo) + +package usbip + +import ( + "context" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/sagernet/sing-box/log" + + "github.com/stretchr/testify/require" +) + +type fakeExport struct { + busid string + leaseOK bool + leaseReason string + leaseHook func() +} + +func (f *fakeExport) BusID() string { return f.busid } + +func (f *fakeExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot { + state := deviceStateAvailable + if busy { + state = deviceStateBusy + } + var info DeviceInfoTruncated + copy(info.BusID[:], f.busid) + info.IDVendor = 0x1d6b + info.IDProduct = 0x0002 + return ExportSnapshot{ + Entry: DeviceEntry{Info: info, Serial: f.busid}, + Backend: "fake", + StableID: "fake:" + f.busid, + State: state, + } +} + +func (f *fakeExport) LeaseCheck(ctx context.Context) (bool, string) { + if f.leaseHook != nil { + f.leaseHook() + } + if f.leaseOK { + return true, "" + } + if f.leaseReason == "" { + return false, "unavailable" + } + return false, f.leaseReason +} + +func (f *fakeExport) DeviceInfo(ctx context.Context) (DeviceInfoTruncated, error) { + var info DeviceInfoTruncated + copy(info.BusID[:], f.busid) + return info, nil +} + +func (f *fakeExport) NewServerDataSession(ctx context.Context, conn net.Conn) (DataSession, error) { + return nil, nil +} + +func newTestExportLedger(t testing.TB, now time.Time) *exportLedger { + t.Helper() + frozen := now + return newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, func() time.Time { return frozen }) +} + +func TestExportLedgerConsumeAcceptsCurrentGeneration(t *testing.T) { + t.Parallel() + + now := time.Now() + const busid = "1-1" + ledger := newTestExportLedger(t, now) + ledger.leases[busid] = serverImportLease{ + ID: 55, + BusID: busid, + ClientNonce: 99, + Generation: 0, + Expires: now.Add(importLeaseTTL), + } + + require.True(t, ledger.ConsumeLease(ImportExtRequest{ + BusID: busid, + LeaseID: 55, + ClientNonce: 99, + })) + require.Empty(t, ledger.leases) +} + +func TestExportLedgerConsumeRejectsStaleGeneration(t *testing.T) { + t.Parallel() + + now := time.Now() + const busid = "1-1" + ledger := newTestExportLedger(t, now) + ledger.seq = 8 + ledger.leases[busid] = serverImportLease{ + ID: 55, + BusID: busid, + ClientNonce: 99, + Generation: 7, + Expires: now.Add(importLeaseTTL), + } + + require.False(t, ledger.ConsumeLease(ImportExtRequest{ + BusID: busid, + LeaseID: 55, + ClientNonce: 99, + })) + require.Empty(t, ledger.leases) +} + +func TestExportLedgerConsumeRejectsExpiredLease(t *testing.T) { + t.Parallel() + + now := time.Now() + const busid = "1-1" + ledger := newTestExportLedger(t, now) + ledger.leases[busid] = serverImportLease{ + ID: 55, + BusID: busid, + ClientNonce: 99, + Generation: 0, + Expires: now.Add(-time.Second), + } + + require.False(t, ledger.ConsumeLease(ImportExtRequest{ + BusID: busid, + LeaseID: 55, + ClientNonce: 99, + })) + require.Empty(t, ledger.leases) +} + +func TestExportLedgerConsumeRejectsMismatchedNonce(t *testing.T) { + t.Parallel() + + now := time.Now() + const busid = "1-1" + ledger := newTestExportLedger(t, now) + ledger.leases[busid] = serverImportLease{ + ID: 55, + BusID: busid, + ClientNonce: 99, + Generation: 0, + Expires: now.Add(importLeaseTTL), + } + + require.False(t, ledger.ConsumeLease(ImportExtRequest{ + BusID: busid, + LeaseID: 55, + ClientNonce: 100, + })) + require.Len(t, ledger.leases, 1) +} + +func TestExportLedgerIssueRejectsWhenBusy(t *testing.T) { + t.Parallel() + + now := time.Now() + const busid = "1-1" + ledger := newTestExportLedger(t, now) + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + ledger.busy[busid] = true + + response := ledger.IssueLease(context.Background(), 1, controlLeaseRequest{BusID: busid, ClientNonce: 42}) + require.Equal(t, leaseErrorUnavailable, response.ErrorCode) + require.Empty(t, ledger.leases) +} + +func TestExportLedgerIssueRejectsWhenLeaseAlreadyActive(t *testing.T) { + t.Parallel() + + now := time.Now() + const busid = "1-1" + ledger := newTestExportLedger(t, now) + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + + first := ledger.IssueLease(context.Background(), 1, controlLeaseRequest{BusID: busid, ClientNonce: 42}) + require.Empty(t, first.ErrorCode) + + second := ledger.IssueLease(context.Background(), 2, controlLeaseRequest{BusID: busid, ClientNonce: 43}) + require.Equal(t, leaseErrorBusy, second.ErrorCode) + require.Len(t, ledger.leases, 1) +} + +func TestExportLedgerIssueConcurrentlyAcceptsOne(t *testing.T) { + t.Parallel() + + const busid = "1-1" + const workers = 64 + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + + var ( + accepted atomic.Uint32 + busyHits atomic.Uint32 + wg sync.WaitGroup + ) + start := make(chan struct{}) + for i := 0; i < workers; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + response := ledger.IssueLease(context.Background(), uint64(i+1), controlLeaseRequest{ + BusID: busid, + ClientNonce: uint64(i + 1), + }) + switch response.ErrorCode { + case "": + accepted.Add(1) + case leaseErrorBusy: + busyHits.Add(1) + } + }(i) + } + close(start) + wg.Wait() + require.Equal(t, uint32(1), accepted.Load()) + require.Equal(t, uint32(workers-1), busyHits.Load()) + require.Len(t, ledger.leases, 1) +} + +func TestExportLedgerIssueLeaseCheckIsAtomicWithInsert(t *testing.T) { + t.Parallel() + + const busid = "1-1" + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + + leaseCheckEntered := make(chan struct{}, 1) + blockLeaseCheck := make(chan struct{}) + export := &fakeExport{busid: busid, leaseOK: true, leaseHook: func() { + select { + case leaseCheckEntered <- struct{}{}: + default: + } + <-blockLeaseCheck + }} + ledger.exports[busid] = export + + go func() { + ledger.IssueLease(context.Background(), 1, controlLeaseRequest{BusID: busid, ClientNonce: 1}) + }() + <-leaseCheckEntered + + concurrent := make(chan controlLeaseResponse, 1) + go func() { + export2 := &fakeExport{busid: busid, leaseOK: true} + ledger2 := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + ledger2.exports[busid] = export2 + concurrent <- ledger2.IssueLease(context.Background(), 2, controlLeaseRequest{BusID: busid, ClientNonce: 2}) + }() + + select { + case <-concurrent: + case <-time.After(50 * time.Millisecond): + t.Fatal("independent ledger should not block on first lease check") + } + + close(blockLeaseCheck) +} + +func TestExportLedgerUnsubscribeRevokesOwnedLeases(t *testing.T) { + t.Parallel() + + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + for _, busid := range []string{"1-1", "1-2", "2-1"} { + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + } + + pipeA, _ := net.Pipe() + pipeB, _ := net.Pipe() + subA, _ := ledger.Subscribe(context.Background(), pipeA, controlCapabilities) + subB, _ := ledger.Subscribe(context.Background(), pipeB, controlCapabilities) + defer pipeA.Close() + defer pipeB.Close() + + for _, spec := range []struct { + sub *exportSubscriber + busid string + }{ + {subA, "1-1"}, + {subA, "1-2"}, + {subB, "2-1"}, + } { + response := ledger.IssueLease(context.Background(), spec.sub.id, controlLeaseRequest{ + BusID: spec.busid, + ClientNonce: spec.sub.id, + }) + require.Empty(t, response.ErrorCode, spec.busid) + } + require.Len(t, ledger.leases, 3) + + ledger.Unsubscribe(subA) + require.Len(t, ledger.leases, 1) + _, found := ledger.leases["2-1"] + require.True(t, found) +} + +func TestExportLedgerBroadcastEmitsOnlyOnDelta(t *testing.T) { + t.Parallel() + + const busid = "1-1" + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + serverConn, clientConn := net.Pipe() + defer serverConn.Close() + defer clientConn.Close() + sub, _ := ledger.Subscribe(context.Background(), serverConn, controlCapabilities) + + // drain the initial snapshot + <-sub.send + + // first broadcast: new export shows up as Added + require.True(t, ledger.BroadcastIfChanged(context.Background())) + delta := <-sub.send + require.Equal(t, controlFrameDeviceDelta, delta.Frame.Type) + + // second broadcast against identical state: must NOT emit + require.False(t, ledger.BroadcastIfChanged(context.Background())) + select { + case msg := <-sub.send: + t.Fatalf("unexpected broadcast frame: %v", msg.Frame.Type) + case <-time.After(50 * time.Millisecond): + } +} + +func TestExportLedgerSubscriberLagClosesConn(t *testing.T) { + t.Parallel() + + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + serverConn, clientConn := net.Pipe() + sub, _ := ledger.Subscribe(context.Background(), serverConn, controlRequiredCapabilities) + + // Fill the buffer. Subscribe enqueued nothing (no extension capabilities) + for i := 0; i < controlSubscriberSendBuffer; i++ { + ledger.enqueueFrame(sub, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: uint64(i + 1)}) + } + + // One more should trigger a close. + ledger.enqueueFrame(sub, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: 999}) + require.Eventually(t, func() bool { + _ = clientConn.SetReadDeadline(time.Now().Add(10 * time.Millisecond)) + _, err := clientConn.Read(make([]byte, 1)) + return err != nil + }, time.Second, 10*time.Millisecond) + _ = clientConn.Close() +} + +func TestExportLedgerTryReserveSerializes(t *testing.T) { + t.Parallel() + + const busid = "1-1" + const workers = 32 + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + + var ( + accepted atomic.Uint32 + busyHits atomic.Uint32 + wg sync.WaitGroup + ) + start := make(chan struct{}) + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + _, ok, reason := ledger.TryReserveForImport(context.Background(), busid) + if ok { + accepted.Add(1) + return + } + if reason == deviceStateBusy { + busyHits.Add(1) + } + }() + } + close(start) + wg.Wait() + require.Equal(t, uint32(1), accepted.Load()) + require.Equal(t, uint32(workers-1), busyHits.Load()) + require.True(t, ledger.busy[busid]) +} + +func TestExportLedgerApplyHostSnapshotClearsReleasedBusy(t *testing.T) { + t.Parallel() + + const busid = "1-1" + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + ledger.busy[busid] = true + + ledger.ApplyHostSnapshot(map[string]Export{}, []string{busid}) + + require.False(t, ledger.busy[busid]) + require.NotContains(t, ledger.exports, busid) +} + +func TestExportLedgerIssueConcurrentWithBroadcastDoesNotDeadlock(t *testing.T) { + t.Parallel() + + const busid = "1-1" + ledger := newExportLedger(log.NewNOPFactory().NewLogger("usbip"), importLeaseTTL, time.Now) + ledger.exports[busid] = &fakeExport{busid: busid, leaseOK: true} + + done := make(chan struct{}, 2) + go func() { + for i := 0; i < 1000; i++ { + ledger.IssueLease(context.Background(), 1, controlLeaseRequest{BusID: busid, ClientNonce: uint64(i)}) + ledger.slow.Lock() + delete(ledger.leases, busid) + ledger.slow.Unlock() + } + done <- struct{}{} + }() + go func() { + for i := 0; i < 1000; i++ { + ledger.BroadcastIfChanged(context.Background()) + } + done <- struct{}{} + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("issue+broadcast deadlocked") + } + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("second goroutine deadlocked") + } +} diff --git a/service/usbip/lease.go b/service/usbip/lease.go deleted file mode 100644 index 5a8809330..000000000 --- a/service/usbip/lease.go +++ /dev/null @@ -1,124 +0,0 @@ -package usbip - -import ( - "sync" - "time" -) - -// LeaseManager owns the server's import-lease state machine. -// -// Lock ordering rules for callers integrating with ServerService: -// -// ServerService.controlAccess -> LeaseManager.access -> ServerService.access -// -// Concretely: -// - The `available` callback supplied to Issue may acquire -// ServerService.access; it MUST NOT acquire ServerService.controlAccess. -// - Callers may hold ServerService.controlAccess across Consume or -// RevokeSubscriber (both are pure map operations, fast under the lock). -// - Callers MUST NOT hold ServerService.controlAccess across Issue, -// because Issue runs `available` which can syscall on Linux -// (readUsbipStatus) and would otherwise stall control broadcasts. -type LeaseManager struct { - access sync.Mutex - leases map[string]serverImportLease - nextID uint64 - ttl time.Duration - now func() time.Time -} - -func NewLeaseManager(ttl time.Duration, now func() time.Time) *LeaseManager { - if now == nil { - now = time.Now - } - return &LeaseManager{ - leases: make(map[string]serverImportLease), - ttl: ttl, - now: now, - } -} - -// Issue atomically checks availability and inserts a fresh lease. The -// `available` closure is invoked under the manager's lock, so the check -// and the insert are TOCTOU-free against any state that `available` -// observes. The closure must respect the lock ordering documented above. -func (m *LeaseManager) Issue( - subID, generation uint64, - request controlLeaseRequest, - available func() (bool, string), -) controlLeaseResponse { - response := controlLeaseResponse{ - BusID: request.BusID, - ClientNonce: request.ClientNonce, - } - m.access.Lock() - defer m.access.Unlock() - now := m.now() - m.cleanupExpiredLocked(now) - ok, reason := available() - if !ok { - response.ErrorCode = leaseErrorUnavailable - response.ErrorMessage = reason - return response - } - _, exists := m.leases[request.BusID] - if exists { - response.ErrorCode = leaseErrorBusy - response.ErrorMessage = "lease already active" - return response - } - m.nextID++ - lease := serverImportLease{ - ID: m.nextID, - SubscriberID: subID, - BusID: request.BusID, - ClientNonce: request.ClientNonce, - Generation: generation, - Expires: now.Add(m.ttl), - } - m.leases[request.BusID] = lease - response.LeaseID = lease.ID - response.Generation = lease.Generation - response.TTLMillis = int64(m.ttl / time.Millisecond) - return response -} - -// Consume validates and removes the lease referenced by request. It -// returns true only if the lease matches ID + nonce, is not expired, -// and lease.Generation matches currentGeneration. The lease entry is -// removed regardless of outcome (consume-on-read semantics). -func (m *LeaseManager) Consume(currentGeneration uint64, request ImportExtRequest) bool { - m.access.Lock() - defer m.access.Unlock() - now := m.now() - m.cleanupExpiredLocked(now) - lease, found := m.leases[request.BusID] - if !found { - return false - } - if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { - return false - } - delete(m.leases, request.BusID) - return now.Before(lease.Expires) && lease.Generation == currentGeneration -} - -// RevokeSubscriber removes every lease previously issued to subID. -// Called when a control subscriber disconnects. -func (m *LeaseManager) RevokeSubscriber(subID uint64) { - m.access.Lock() - defer m.access.Unlock() - for busid, lease := range m.leases { - if lease.SubscriberID == subID { - delete(m.leases, busid) - } - } -} - -func (m *LeaseManager) cleanupExpiredLocked(now time.Time) { - for busid, lease := range m.leases { - if !now.Before(lease.Expires) { - delete(m.leases, busid) - } - } -} diff --git a/service/usbip/lease_test.go b/service/usbip/lease_test.go deleted file mode 100644 index fda8dfe89..000000000 --- a/service/usbip/lease_test.go +++ /dev/null @@ -1,248 +0,0 @@ -//go:build linux || (darwin && cgo) - -package usbip - -import ( - "sync" - "sync/atomic" - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -func newLeaseManagerWithLeases(now time.Time, leases map[string]serverImportLease) *LeaseManager { - manager := NewLeaseManager(importLeaseTTL, func() time.Time { return now }) - for busid, lease := range leases { - manager.leases[busid] = lease - } - return manager -} - -func TestLeaseManagerConsumeAcceptsCurrentGeneration(t *testing.T) { - t.Parallel() - - now := time.Now() - const busid = "1-1" - manager := newLeaseManagerWithLeases(now, map[string]serverImportLease{ - busid: { - ID: 55, - BusID: busid, - ClientNonce: 99, - Generation: 7, - Expires: now.Add(importLeaseTTL), - }, - }) - - ok := manager.Consume(7, ImportExtRequest{ - BusID: busid, - LeaseID: 55, - ClientNonce: 99, - }) - require.True(t, ok) - require.Empty(t, manager.leases) -} - -func TestLeaseManagerConsumeRejectsStaleGeneration(t *testing.T) { - t.Parallel() - - now := time.Now() - const busid = "1-1" - manager := newLeaseManagerWithLeases(now, map[string]serverImportLease{ - busid: { - ID: 55, - BusID: busid, - ClientNonce: 99, - Generation: 7, - Expires: now.Add(importLeaseTTL), - }, - }) - - ok := manager.Consume(8, ImportExtRequest{ - BusID: busid, - LeaseID: 55, - ClientNonce: 99, - }) - require.False(t, ok) - require.Empty(t, manager.leases) -} - -func TestLeaseManagerConsumeRejectsExpiredLease(t *testing.T) { - t.Parallel() - - now := time.Now() - const busid = "1-1" - manager := newLeaseManagerWithLeases(now, map[string]serverImportLease{ - busid: { - ID: 55, - BusID: busid, - ClientNonce: 99, - Generation: 7, - Expires: now.Add(-time.Second), - }, - }) - - ok := manager.Consume(7, ImportExtRequest{ - BusID: busid, - LeaseID: 55, - ClientNonce: 99, - }) - require.False(t, ok) - require.Empty(t, manager.leases) -} - -func TestLeaseManagerConsumeRejectsMismatchedNonce(t *testing.T) { - t.Parallel() - - now := time.Now() - const busid = "1-1" - manager := newLeaseManagerWithLeases(now, map[string]serverImportLease{ - busid: { - ID: 55, - BusID: busid, - ClientNonce: 99, - Generation: 7, - Expires: now.Add(importLeaseTTL), - }, - }) - - ok := manager.Consume(7, ImportExtRequest{ - BusID: busid, - LeaseID: 55, - ClientNonce: 100, - }) - require.False(t, ok) - // Mismatched nonce keeps the lease in place for the legitimate client - // to redeem; consume-on-read fires only on ID/nonce match. - require.Len(t, manager.leases, 1) -} - -func TestLeaseManagerIssueRejectsWhenAvailableReturnsFalse(t *testing.T) { - t.Parallel() - - manager := NewLeaseManager(importLeaseTTL, time.Now) - response := manager.Issue(1, 1, controlLeaseRequest{BusID: "1-1", ClientNonce: 42}, func() (bool, string) { - return false, "device busy" - }) - require.Equal(t, leaseErrorUnavailable, response.ErrorCode) - require.Equal(t, "device busy", response.ErrorMessage) - require.Empty(t, manager.leases) -} - -func TestLeaseManagerIssueRejectsWhenLeaseAlreadyActive(t *testing.T) { - t.Parallel() - - manager := NewLeaseManager(importLeaseTTL, time.Now) - first := manager.Issue(1, 1, controlLeaseRequest{BusID: "1-1", ClientNonce: 42}, func() (bool, string) { - return true, "" - }) - require.Empty(t, first.ErrorCode) - - second := manager.Issue(2, 1, controlLeaseRequest{BusID: "1-1", ClientNonce: 43}, func() (bool, string) { - return true, "" - }) - require.Equal(t, leaseErrorBusy, second.ErrorCode) - require.Len(t, manager.leases, 1) -} - -func TestLeaseManagerIssueConcurrentlyAcceptsOne(t *testing.T) { - t.Parallel() - - manager := NewLeaseManager(importLeaseTTL, time.Now) - const workers = 64 - var ( - accepted atomic.Uint32 - busyHits atomic.Uint32 - wg sync.WaitGroup - ) - start := make(chan struct{}) - for i := 0; i < workers; i++ { - wg.Add(1) - go func(i int) { - defer wg.Done() - <-start - response := manager.Issue(uint64(i+1), 1, controlLeaseRequest{ - BusID: "1-1", - ClientNonce: uint64(i + 1), - }, func() (bool, string) { - return true, "" - }) - switch response.ErrorCode { - case "": - accepted.Add(1) - case leaseErrorBusy: - busyHits.Add(1) - } - }(i) - } - close(start) - wg.Wait() - require.Equal(t, uint32(1), accepted.Load()) - require.Equal(t, uint32(workers-1), busyHits.Load()) - require.Len(t, manager.leases, 1) -} - -func TestLeaseManagerIssueAvailabilityCheckIsAtomicWithInsert(t *testing.T) { - t.Parallel() - - manager := NewLeaseManager(importLeaseTTL, time.Now) - checkRan := make(chan struct{}, 1) - blockCheck := make(chan struct{}) - - go func() { - manager.Issue(1, 1, controlLeaseRequest{BusID: "1-1", ClientNonce: 1}, func() (bool, string) { - checkRan <- struct{}{} - <-blockCheck - return true, "" - }) - }() - - <-checkRan - // While the first Issue's availability check is blocked under the - // manager lock, a second Issue must block on the same lock instead - // of slipping in between check and insert. - concurrent := make(chan controlLeaseResponse, 1) - go func() { - concurrent <- manager.Issue(2, 1, controlLeaseRequest{BusID: "1-1", ClientNonce: 2}, func() (bool, string) { - return true, "" - }) - }() - - select { - case <-concurrent: - t.Fatal("second Issue completed while first held the lock") - case <-time.After(50 * time.Millisecond): - } - - close(blockCheck) - response := <-concurrent - require.Equal(t, leaseErrorBusy, response.ErrorCode) -} - -func TestLeaseManagerRevokeSubscriberRemovesOwnedLeases(t *testing.T) { - t.Parallel() - - manager := NewLeaseManager(importLeaseTTL, time.Now) - for _, spec := range []struct { - subID uint64 - busid string - }{ - {1, "1-1"}, - {1, "1-2"}, - {2, "2-1"}, - } { - response := manager.Issue(spec.subID, 1, controlLeaseRequest{ - BusID: spec.busid, - ClientNonce: spec.subID, - }, func() (bool, string) { - return true, "" - }) - require.Empty(t, response.ErrorCode, spec.busid) - } - require.Len(t, manager.leases, 3) - - manager.RevokeSubscriber(1) - require.Len(t, manager.leases, 1) - _, found := manager.leases["2-1"] - require.True(t, found) -} diff --git a/service/usbip/linux_interop_test.go b/service/usbip/linux_interop_test.go index 0eedf4338..6f4eab34c 100644 --- a/service/usbip/linux_interop_test.go +++ b/service/usbip/linux_interop_test.go @@ -402,7 +402,7 @@ func startRealUSBIPServer(t *testing.T, devices []option.USBIPDeviceMatch) (*Ser _ = server.Close() }) - return server, M.SocksaddrFromNet(server.listen.Addr()) + return server, M.SocksaddrFromNet(server.listener.TCPListener().Addr()) } func startRealUSBIPClient(t *testing.T, destination M.Socksaddr, devices []option.USBIPDeviceMatch) *ClientService { @@ -1155,7 +1155,7 @@ func TestUSBIPControlHotplugACMReattach(t *testing.T) { first.Close() waitForPathGone(t, firstImportedTTY) require.Eventually(t, func() bool { - return len(server.currentExports()) == 0 + return len(server.ledger.AvailableExports()) == 0 }, 5*time.Second, 100*time.Millisecond) secondBefore := importedNodeSnapshot("/dev/ttyACM*") diff --git a/service/usbip/linux_test.go b/service/usbip/linux_test.go index 8e052438d..5176484cf 100644 --- a/service/usbip/linux_test.go +++ b/service/usbip/linux_test.go @@ -306,17 +306,14 @@ func newTestLinuxImportHost(t *testing.T, ops usbipOps) *linuxImportHost { func newTestServerServiceWithHost(t *testing.T, ctx context.Context, cancel context.CancelFunc, host ExportHost, matches []option.USBIPDeviceMatch) *ServerService { t.Helper() + logger := newTestLogger(t) return &ServerService{ - ctx: ctx, - cancel: cancel, - logger: newTestLogger(t), - matches: matches, - host: host, - exports: make(map[string]Export), - busy: make(map[string]bool), - controlSubs: make(map[uint64]*serverControlConn), - controlState: make(map[string]DeviceInfoV2), - leases: NewLeaseManager(importLeaseTTL, time.Now), + ctx: ctx, + cancel: cancel, + logger: logger, + matches: matches, + host: host, + ledger: newExportLedger(logger, importLeaseTTL, time.Now), } } @@ -360,15 +357,15 @@ func deleteLinuxExport(host *linuxExportHost, busid string) { } func serverInsertExport(server *ServerService, exp Export) { - server.access.Lock() - defer server.access.Unlock() - server.exports[exp.BusID()] = exp + server.ledger.slow.Lock() + defer server.ledger.slow.Unlock() + server.ledger.exports[exp.BusID()] = exp } func serverDeleteExport(server *ServerService, busid string) { - server.access.Lock() - defer server.access.Unlock() - delete(server.exports, busid) + server.ledger.slow.Lock() + defer server.ledger.slow.Unlock() + delete(server.ledger.exports, busid) } func attachedLinuxSession(t *testing.T, session AttachedSession) *linuxClientSession { @@ -487,9 +484,9 @@ func duplicateNetConnFromFD(fd uintptr, name string) (net.Conn, error) { } func linuxServerControlState(server *ServerService, busid string) string { - server.controlAccess.Lock() - defer server.controlAccess.Unlock() - return server.controlState[busid].State + server.ledger.fast.Lock() + defer server.ledger.fast.Unlock() + return server.ledger.state[busid].State } func duplicateHandoffKernelConn(t *testing.T, handoff *kernelHandoffSession) net.Conn { @@ -1373,7 +1370,7 @@ func TestServerUEventLoopReconcilesWhenListenerStarts(t *testing.T) { }() require.Eventually(t, func() bool { - _, ok := server.getExport("1-1") + _, ok := server.ledger.Export("1-1") return ok }, 3*time.Second, 10*time.Millisecond) } @@ -1393,10 +1390,8 @@ func TestServerBuildDevListEntriesFiltersUnavailable(t *testing.T) { host := newTestLinuxExportHost(t, nil, ops) server := newTestServerServiceWithHost(t, context.Background(), func() {}, host, nil) - server.exports = map[string]Export{ - "1-1": newTestServerExport(store, "1-1", ops, host.logger), - "1-2": newTestServerExport(store, "1-2", ops, host.logger), - } + serverInsertExport(server, newTestServerExport(store, "1-1", ops, host.logger)) + serverInsertExport(server, newTestServerExport(store, "1-2", ops, host.logger)) entries := server.buildDevListEntries() require.Len(t, entries, 1) @@ -1631,7 +1626,15 @@ func TestServerDispatchConnHandlesControlPingAndChanged(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - host := newTestLinuxExportHost(t, nil, newTestUSBIPOps(t)) + device := newTestDevice("1-1", 0x1d6b, 0x0002, "serial-1", SpeedHigh) + store := newTestDeviceStore(device) + store.setStatus("1-1", usbipStatusAvailable) + + ops := newTestUSBIPOps(t) + ops.readSysfsDevice = store.readSysfsDevice + ops.readUsbipStatus = store.readUsbipStatus + + host := newTestLinuxExportHost(t, nil, ops) server := newTestServerServiceWithHost(t, ctx, cancel, host, nil) serverAddr, closeServer := startDispatchServer(t, server) defer closeServer() @@ -1663,7 +1666,8 @@ func TestServerDispatchConnHandlesControlPingAndChanged(t *testing.T) { require.Equal(t, controlFramePong, pong.Type) require.Equal(t, controlProtocolVersion, pong.Version) - server.broadcastControlState(deviceInfoV2Map(server.buildDeviceStateV2()), true) + serverInsertExport(server, newTestServerExport(store, "1-1", ops, host.logger)) + require.True(t, server.ledger.BroadcastIfChanged(server.ctx)) changed, err := readControlMessage(conn) require.NoError(t, err) require.Equal(t, controlFrameDeviceDelta, changed.Frame.Type) @@ -1671,24 +1675,31 @@ func TestServerDispatchConnHandlesControlPingAndChanged(t *testing.T) { var delta controlDeviceDelta require.NoError(t, unmarshalControlPayload(changed.Payload, &delta)) require.Equal(t, uint64(1), delta.Sequence) + require.Len(t, delta.Added, 1) + require.Equal(t, "1-1", delta.Added[0].BusID) } func TestServerRegisterControlConnQueuesSnapshotBeforeBroadcast(t *testing.T) { t.Parallel() - host := newTestLinuxExportHost(t, nil, newTestUSBIPOps(t)) + device := newTestDevice("1-1", 0x1d6b, 0x0002, "serial-1", SpeedHigh) + store := newTestDeviceStore(device) + store.setStatus("1-1", usbipStatusAvailable) + ops := newTestUSBIPOps(t) + ops.readSysfsDevice = store.readSysfsDevice + ops.readUsbipStatus = store.readUsbipStatus + + host := newTestLinuxExportHost(t, nil, ops) server := newTestServerServiceWithHost(t, context.Background(), func() {}, host, nil) serverConn, clientConn := net.Pipe() defer serverConn.Close() defer clientConn.Close() - sub, seq := server.registerControlConn(serverConn, controlCapabilities) + sub, seq := server.ledger.Subscribe(server.ctx, serverConn, controlCapabilities) require.Zero(t, seq) - require.Contains(t, server.controlSubs, sub.id) - require.True(t, server.broadcastControlState(map[string]DeviceInfoV2{ - "1-1": {BusID: "1-1", State: deviceStateAvailable}, - }, true)) + serverInsertExport(server, newTestServerExport(store, "1-1", ops, host.logger)) + require.True(t, server.ledger.BroadcastIfChanged(server.ctx)) first := <-sub.send require.Equal(t, controlFrameDeviceSnapshot, first.Frame.Type) @@ -1717,7 +1728,7 @@ func TestServerReconcileBroadcastsStatusOnlyDeviceDelta(t *testing.T) { host := newTestLinuxExportHost(t, []option.USBIPDeviceMatch{{BusID: "1-1"}}, serverOps) server := newTestServerServiceWithHost(t, ctx, cancel, host, []option.USBIPDeviceMatch{{BusID: "1-1"}}) serverInsertExport(server, newTestServerExport(store, "1-1", serverOps, host.logger)) - server.setControlState(deviceInfoV2Map(server.buildDeviceStateV2())) + server.ledger.SeedBroadcastState(server.ctx) serverAddr, closeServer := startDispatchServer(t, server) defer closeServer() @@ -1759,9 +1770,9 @@ func TestServerReconcileBroadcastsStatusOnlyDeviceDelta(t *testing.T) { require.Equal(t, deviceStateAvailable, delta.Updated[0].State) require.Equal(t, usbipStatusAvailable, delta.Updated[0].StatusCode) - sequence := server.currentControlSequence() + sequence := server.ledger.CurrentSequence() require.NoError(t, server.reconcileAndBroadcast(true)) - require.Equal(t, sequence, server.currentControlSequence()) + require.Equal(t, sequence, server.ledger.CurrentSequence()) require.NoError(t, conn.SetReadDeadline(time.Now().Add(100*time.Millisecond))) _, err = readControlMessage(conn) @@ -1789,7 +1800,7 @@ func TestServerControlSnapshotPreservesPendingDelta(t *testing.T) { host := newTestLinuxExportHost(t, []option.USBIPDeviceMatch{{BusID: "1-1"}}, serverOps) server := newTestServerServiceWithHost(t, ctx, cancel, host, []option.USBIPDeviceMatch{{BusID: "1-1"}}) serverInsertExport(server, newTestServerExport(store, "1-1", serverOps, host.logger)) - server.setControlState(deviceInfoV2Map(server.buildDeviceStateV2())) + server.ledger.SeedBroadcastState(server.ctx) serverAddr, closeServer := startDispatchServer(t, server) defer closeServer() @@ -1852,7 +1863,7 @@ func TestServerControlLeaseEnablesImportExt(t *testing.T) { host := newTestLinuxExportHost(t, nil, serverOps) server := newTestServerServiceWithHost(t, ctx, cancel, host, nil) serverInsertExport(server, newTestServerExport(store, "1-1", serverOps, host.logger)) - server.setControlState(deviceInfoV2Map(server.buildDeviceStateV2())) + server.ledger.SeedBroadcastState(server.ctx) serverAddr, closeServer := startDispatchServer(t, server) defer closeServer() @@ -2683,7 +2694,7 @@ func TestClientRunControlSessionSyncsAssignmentsOnChanged(t *testing.T) { host := newTestLinuxExportHost(t, nil, serverOps) server := newTestServerServiceWithHost(t, serverCtx, serverCancel, host, nil) serverInsertExport(server, newTestServerExport(store, "1-1", serverOps, host.logger)) - server.setControlState(deviceInfoV2Map(server.buildDeviceStateV2())) + server.ledger.SeedBroadcastState(server.ctx) serverAddr, closeServer := startDispatchServer(t, server) defer closeServer() @@ -2718,7 +2729,7 @@ func TestClientRunControlSessionSyncsAssignmentsOnChanged(t *testing.T) { store.setDevices(updatedDevice) serverDeleteExport(server, "1-1") serverInsertExport(server, newTestServerExport(store, "1-2", serverOps, host.logger)) - server.broadcastControlState(deviceInfoV2Map(server.buildDeviceStateV2()), true) + require.True(t, server.ledger.BroadcastIfChanged(server.ctx)) require.Eventually(t, func() bool { assignment.access.Lock() diff --git a/service/usbip/server.go b/service/usbip/server.go index 31bcac99a..00b88a70e 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -6,8 +6,6 @@ import ( "context" "fmt" "net" - "slices" - "strings" "sync" "time" @@ -23,9 +21,9 @@ import ( ) // ServerService is the unified USB/IP server. It owns the wire-protocol -// surface (devlist, import, import-ext, control channel), the lease -// state machine, the broadcast bookkeeping, and the busy map. -// Platform-specific device acquisition is delegated to an ExportHost. +// surface (devlist, import, import-ext, control channel) and delegates +// all mutable state to an exportLedger. Platform-specific device +// acquisition is delegated to an ExportHost. type ServerService struct { boxService.Adapter ctx context.Context @@ -34,18 +32,7 @@ type ServerService struct { listener *listener.Listener matches []option.USBIPDeviceMatch host ExportHost - - access sync.Mutex - exports map[string]Export - busy map[string]bool - listen net.Listener - - controlAccess sync.Mutex - controlSeq uint64 - controlNextID uint64 - controlSubs map[uint64]*serverControlConn - controlState map[string]DeviceInfoV2 - leases *LeaseManager + ledger *exportLedger reconcileAccess sync.Mutex } @@ -72,17 +59,13 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string, logger: logger, matches: options.Devices, host: host, - exports: make(map[string]Export), - busy: make(map[string]bool), + ledger: newExportLedger(logger, importLeaseTTL, time.Now), listener: listener.New(listener.Options{ Context: ctx, Logger: logger, Network: []string{N.NetworkTCP}, Listen: options.ListenOptions, }), - controlSubs: make(map[uint64]*serverControlConn), - controlState: make(map[string]DeviceInfoV2), - leases: NewLeaseManager(importLeaseTTL, time.Now), }, nil } @@ -104,9 +87,6 @@ func (s *ServerService) Start(stage adapter.StartStage) error { _ = s.host.Close() return err } - s.access.Lock() - s.listen = tcpListener - s.access.Unlock() go s.acceptLoop(tcpListener) go s.eventLoop() return nil @@ -116,15 +96,14 @@ func (s *ServerService) Close() error { if s.cancel != nil { s.cancel() } - s.closeControlSubscribers() + for _, conn := range s.ledger.CloseAllSubscribers() { + _ = conn.Close() + } err := common.Close(common.PtrOrNil(s.listener)) s.reconcileAccess.Lock() defer s.reconcileAccess.Unlock() _ = s.host.Close() - s.access.Lock() - s.exports = make(map[string]Export) - s.busy = make(map[string]bool) - s.access.Unlock() + s.ledger.ResetForClose() return err } @@ -159,95 +138,19 @@ func (s *ServerService) reconcileAndBroadcast(notify bool) error { if s.ctx != nil && s.ctx.Err() != nil { return nil } - snapshot, released, _, err := s.host.Reconcile(s.ctx, s.isBusy) + snapshot, released, _, err := s.host.Reconcile(s.ctx, s.ledger.IsBusy) if err != nil { return err } - s.access.Lock() - s.exports = snapshot - for _, busid := range released { - delete(s.busy, busid) - } - s.access.Unlock() - - nextState := deviceInfoV2Map(s.buildDeviceStateV2()) + s.ledger.ApplyHostSnapshot(snapshot, released) if notify { - s.broadcastControlState(nextState, false) + s.ledger.BroadcastIfChanged(s.ctx) } else { - s.setControlState(nextState) + s.ledger.SeedBroadcastState(s.ctx) } return nil } -func (s *ServerService) getExport(busid string) (Export, bool) { - s.access.Lock() - defer s.access.Unlock() - export, ok := s.exports[busid] - return export, ok -} - -func (s *ServerService) isExported(busid string) bool { - _, ok := s.getExport(busid) - return ok -} - -func (s *ServerService) isBusy(busid string) bool { - s.access.Lock() - defer s.access.Unlock() - return s.busy[busid] -} - -func (s *ServerService) setBusy(busid string, busy bool) bool { - s.access.Lock() - defer s.access.Unlock() - current := s.busy[busid] - if current == busy { - return false - } - if busy { - s.busy[busid] = true - } else { - delete(s.busy, busid) - } - return true -} - -func (s *ServerService) removeExport(busid string) { - s.access.Lock() - defer s.access.Unlock() - delete(s.exports, busid) - delete(s.busy, busid) -} - -func (s *ServerService) currentExports() []Export { - s.access.Lock() - defer s.access.Unlock() - out := make([]Export, 0, len(s.exports)) - for busid, export := range s.exports { - if s.busy[busid] { - continue - } - out = append(out, export) - } - slices.SortFunc(out, exportLess) - return out -} - -func (s *ServerService) allExports() []Export { - s.access.Lock() - defer s.access.Unlock() - out := make([]Export, 0, len(s.exports)) - for _, export := range s.exports { - out = append(out, export) - } - slices.SortFunc(out, exportLess) - return out -} - -func exportLess(a, b Export) int { - return strings.Compare(a.BusID(), b.BusID()) -} - func (s *ServerService) handleStandardConn(conn net.Conn, header OpHeader) { closeConn := true defer func() { @@ -288,8 +191,8 @@ func (s *ServerService) handleControlConn(conn net.Conn) { return } capabilities := negotiatedControlCapabilities(hello.Capabilities) - sub, seq := s.registerControlConn(conn, capabilities) - defer s.unregisterControlConn(sub.id) + sub, seq := s.ledger.Subscribe(s.ctx, conn, capabilities) + defer s.ledger.Unsubscribe(sub) err = writeControlAckWithCapabilities(conn, seq, capabilities) if err != nil { s.logger.Debug("write control ack: ", err) @@ -322,7 +225,7 @@ func (s *ServerService) handleDevList(conn net.Conn) { } func (s *ServerService) buildDevListEntries() []DeviceEntry { - exports := s.currentExports() + exports := s.ledger.AvailableExports() if len(exports) == 0 { return nil } @@ -352,7 +255,7 @@ func (s *ServerService) handleImportExt(conn net.Conn) bool { s.logger.Debug("read import-ext body: ", err) return false } - if !s.consumeImportLease(request) { + if !s.ledger.ConsumeLease(request) { s.logger.Info("import-ext rejected (invalid lease): ", request.BusID) _ = WriteOpRepImportExt(conn, OpStatusError, nil) return false @@ -365,45 +268,34 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended if extended { writeReply = WriteOpRepImportExt } - export, ok := s.getExport(busid) + export, ok, reason := s.ledger.TryReserveForImport(s.ctx, busid) if !ok { - s.logger.Info("import rejected (unknown busid): ", busid) - _ = writeReply(conn, OpStatusError, nil) - return false - } - if s.isBusy(busid) { - s.logger.Info("import rejected (busid ", busid, " already busy)") - _ = writeReply(conn, OpStatusError, nil) - return false - } - ok, reason := export.LeaseCheck(s.ctx) - if !ok { - s.logger.Info("import rejected (busid ", busid, ": ", reason, ")") + s.logger.Info("import rejected (", busid, ": ", reason, ")") _ = writeReply(conn, OpStatusError, nil) return false } info, err := export.DeviceInfo(s.ctx) if err != nil { + s.ledger.ReleaseImport(s.ctx, busid, false) s.logger.Warn("refresh ", busid, ": ", err) _ = writeReply(conn, OpStatusError, nil) return false } session, err := export.NewServerDataSession(s.ctx, conn) if err != nil { + s.ledger.ReleaseImport(s.ctx, busid, false) s.logger.Warn("open data session ", busid, ": ", err) _ = writeReply(conn, OpStatusError, nil) return false } - s.setBusy(busid, true) - s.broadcastChanged() + s.ledger.ConfirmImport(s.ctx) err = writeReply(conn, OpStatusOK, &info) if err != nil { s.logger.Warn("reply import ", busid, ": ", err) _ = session.Close() <-session.Done() - _, _ = s.host.FinishImport(s.ctx, busid) - s.setBusy(busid, false) - s.broadcastChanged() + released, _ := s.host.FinishImport(s.ctx, busid) + s.ledger.ReleaseImport(s.ctx, busid, released) return false } s.logger.Info("attached ", busid, " to remote ", conn.RemoteAddr()) @@ -417,55 +309,5 @@ func (s *ServerService) waitImportDone(busid string, session DataSession) { if err != nil { s.logger.Debug("finish import ", busid, ": ", err) } - if released { - s.removeExport(busid) - s.broadcastChanged() - return - } - s.setBusy(busid, false) - s.broadcastChanged() -} - -func (s *ServerService) broadcastChanged() { - s.broadcastControlState(deviceInfoV2Map(s.buildDeviceStateV2()), false) -} - -func (s *ServerService) buildDeviceStateV2() []DeviceInfoV2 { - exports := s.allExports() - if len(exports) == 0 { - return nil - } - devices := make([]DeviceInfoV2, 0, len(exports)) - for _, export := range exports { - busid := export.BusID() - snapshot := export.Snapshot(s.ctx, s.isBusy(busid)) - if snapshot.Err != nil { - devices = append(devices, DeviceInfoV2{ - BusID: busid, - Backend: snapshot.Backend, - StableID: snapshot.StableID, - State: deviceStateUnavailable, - StatusReason: snapshot.StatusReason, - }) - continue - } - // Skip exports that the host marked unavailable (e.g. Darwin - // stale) — they should not appear in broadcast. - if snapshot.State == deviceStateUnavailable && snapshot.Entry.Info.IDVendor == 0 { - continue - } - devices = append(devices, deviceInfoV2FromEntry(snapshot.Entry, snapshot.Backend, snapshot.StableID, snapshot.State, snapshot.RawStatus, snapshot.StatusReason)) - } - return devices -} - -func (s *ServerService) leaseAvailable(busid string) (bool, string) { - export, ok := s.getExport(busid) - if !ok { - return false, "unknown busid" - } - if s.isBusy(busid) { - return false, deviceStateBusy - } - return export.LeaseCheck(s.ctx) + s.ledger.ReleaseImport(s.ctx, busid, released) } diff --git a/service/usbip/server_darwin_test.go b/service/usbip/server_darwin_test.go index 3ce97b80a..72826bfb3 100644 --- a/service/usbip/server_darwin_test.go +++ b/service/usbip/server_darwin_test.go @@ -135,14 +135,12 @@ func TestDarwinServerReconcileAndBroadcastSkipsAfterCancel(t *testing.T) { return nil, nil }, }) + logger := newTestLogger(t) server := &ServerService{ - ctx: ctx, - logger: newTestLogger(t), - host: host, - exports: make(map[string]Export), - busy: make(map[string]bool), - controlSubs: make(map[uint64]*serverControlConn), - controlState: make(map[string]DeviceInfoV2), + ctx: ctx, + logger: logger, + host: host, + ledger: newExportLedger(logger, importLeaseTTL, time.Now), } require.NoError(t, server.reconcileAndBroadcast(true)) @@ -214,14 +212,16 @@ func TestDarwinServerBuildDeviceStateIncludesBusyExports(t *testing.T) { entry: busy, }, } + logger := newTestLogger(t) server := &ServerService{ - ctx: context.Background(), - logger: newTestLogger(t), - exports: exports, - busy: map[string]bool{"busy": true}, + ctx: context.Background(), + logger: logger, + ledger: newExportLedger(logger, importLeaseTTL, time.Now), } + server.ledger.exports = exports + server.ledger.busy = map[string]bool{"busy": true} - devices := deviceInfoV2Map(server.buildDeviceStateV2()) + devices := deviceInfoV2Map(server.ledger.snapshotDeviceState(server.ctx)) require.Equal(t, deviceStateAvailable, devices["available"].State) require.Equal(t, deviceStateBusy, devices["busy"].State) } @@ -316,25 +316,24 @@ func TestDarwinExportHostReconcileCapturesReplacementAfterStaleRelease(t *testin func TestDarwinServerRegisterControlConnQueuesSnapshotBeforeBroadcast(t *testing.T) { t.Parallel() + logger := newTestLogger(t) server := &ServerService{ - ctx: context.Background(), - logger: newTestLogger(t), - exports: make(map[string]Export), - busy: make(map[string]bool), - controlSubs: make(map[uint64]*serverControlConn), - controlState: make(map[string]DeviceInfoV2), + ctx: context.Background(), + logger: logger, + ledger: newExportLedger(logger, importLeaseTTL, time.Now), } serverConn, clientConn := net.Pipe() defer serverConn.Close() defer clientConn.Close() - sub, seq := server.registerControlConn(serverConn, controlCapabilities) + sub, seq := server.ledger.Subscribe(server.ctx, serverConn, controlCapabilities) require.Zero(t, seq) - require.Contains(t, server.controlSubs, sub.id) added := standardTestDeviceEntry("added") - server.exports["added"] = &darwinExport{busid: "added", registryID: 1, entry: added} - server.broadcastChanged() + server.ledger.slow.Lock() + server.ledger.exports["added"] = &darwinExport{busid: "added", registryID: 1, entry: added} + server.ledger.slow.Unlock() + require.True(t, server.ledger.BroadcastIfChanged(server.ctx)) first := <-sub.send require.Equal(t, controlFrameDeviceSnapshot, first.Frame.Type) @@ -364,15 +363,14 @@ func TestDarwinServerImportBroadcastsBusyState(t *testing.T) { entry: entry, device: &darwinUSBHostDevice{}, } + logger := log.NewNOPFactory().NewLogger("usbip") server := &ServerService{ - ctx: ctx, - logger: log.NewNOPFactory().NewLogger("usbip"), - host: host, - exports: map[string]Export{busid: host.exports[busid]}, - busy: make(map[string]bool), - controlSubs: make(map[uint64]*serverControlConn), - controlState: make(map[string]DeviceInfoV2), + ctx: ctx, + logger: logger, + host: host, + ledger: newExportLedger(logger, importLeaseTTL, time.Now), } + server.ledger.exports = map[string]Export{busid: host.exports[busid]} serverConn, clientConn := net.Pipe() done := make(chan struct{}) @@ -405,9 +403,9 @@ func TestDarwinServerImportBroadcastsBusyState(t *testing.T) { } func darwinServerControlState(server *ServerService, busid string) string { - server.controlAccess.Lock() - defer server.controlAccess.Unlock() - return server.controlState[busid].State + server.ledger.fast.Lock() + defer server.ledger.fast.Unlock() + return server.ledger.state[busid].State } func darwinTestDeviceInfo(registryID uint64, entry DeviceEntry) darwinUSBHostDeviceInfo { diff --git a/service/usbip/server_shared.go b/service/usbip/server_shared.go index bf72661fc..e7aab4a67 100644 --- a/service/usbip/server_shared.go +++ b/service/usbip/server_shared.go @@ -51,7 +51,7 @@ func (s *ServerService) dispatchConn(conn net.Conn) { s.handleStandardConn(conn, ParseOpHeader(prefix[:])) } -func (s *ServerService) readControlConn(sub *serverControlConn, done chan<- struct{}) { +func (s *ServerService) readControlConn(sub *exportSubscriber, done chan<- struct{}) { defer close(done) var reader controlReader for { @@ -62,13 +62,10 @@ func (s *ServerService) readControlConn(sub *serverControlConn, done chan<- stru frame := message.Frame switch frame.Type { case controlFramePing: - s.enqueueControlFrame(sub, controlFrame{ - Type: controlFramePong, - Version: controlProtocolVersion, - }) + s.ledger.HandleControlPing(sub) case controlFrameLeaseRequest: if supportsControlExtensions(sub.capabilities) { - s.handleControlLeaseRequest(sub, message.Payload) + s.ledger.HandleControlLeaseRequest(s.ctx, sub, message.Payload) continue } return