From a438f6212b2e0610896b5c409a5fe75016ce98f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Sat, 16 May 2026 11:07:28 +0800 Subject: [PATCH] usbip: fix import-all, lease release broadcast, and Events contract Drop the constructor guard so client import-all (Devices empty/omitted) actually works, route ledger lease mutations through a new mutateAndBroadcast helper so Unsubscribe-triggered release reaches all subscribers, and tighten the Events contract: subscribe at Start so a host failure aborts service start instead of silently disabling hotplug. --- option/usbip.go | 2 + service/usbip/client.go | 5 - service/usbip/export_ledger.go | 217 +++++++++++++++++----------- service/usbip/export_ledger_test.go | 61 +++++++- service/usbip/host.go | 6 +- service/usbip/server.go | 21 ++- 6 files changed, 209 insertions(+), 103 deletions(-) diff --git a/option/usbip.go b/option/usbip.go index 748068582..a84802129 100644 --- a/option/usbip.go +++ b/option/usbip.go @@ -66,5 +66,7 @@ type USBIPServerServiceOptions struct { type USBIPClientServiceOptions struct { ServerOptions DialerOptions + // Devices selects which exported devices to import. Omit or leave + // empty to import every device the server exports. Devices []USBIPDeviceMatch `json:"devices,omitempty"` } diff --git a/service/usbip/client.go b/service/usbip/client.go index bd478c3ab..75a442a31 100644 --- a/service/usbip/client.go +++ b/service/usbip/client.go @@ -26,7 +26,6 @@ type ClientService struct { logger log.ContextLogger dialer N.Dialer serverAddr M.Socksaddr - matches []option.USBIPDeviceMatch host ImportHost assignment *clientAssignment @@ -43,9 +42,6 @@ type ClientService struct { } func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, options option.USBIPClientServiceOptions) (adapter.Service, error) { - if len(options.Devices) == 0 { - return nil, E.New("devices: at least one match is required") - } for i, m := range options.Devices { if m.IsZero() { return nil, E.New("devices[", i, "]: at least one of busid/vendor_id/product_id/serial is required") @@ -73,7 +69,6 @@ func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, logger: logger, dialer: outboundDialer, serverAddr: options.ServerOptions.Build(), - matches: options.Devices, host: host, assignment: newClientAssignment(options.Devices), allWorkers: make(map[string]context.CancelFunc), diff --git a/service/usbip/export_ledger.go b/service/usbip/export_ledger.go index a9caa80c4..2739701e3 100644 --- a/service/usbip/export_ledger.go +++ b/service/usbip/export_ledger.go @@ -192,13 +192,27 @@ func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (E } func (l *exportLedger) ReleaseImport(ctx context.Context, busid string, removeExport bool) { + l.mutateAndBroadcast(ctx, func() bool { + delete(l.busy, busid) + if removeExport { + delete(l.exports, busid) + } + return true + }) +} + +// mutateAndBroadcast holds slow for the body, then broadcasts iff body +// reported a change. This centralizes the "any reservedLocked-affecting +// mutation must broadcast" invariant so future mutation sites cannot +// silently skip it. body must NOT acquire fast (lock-ordering rule at +// the top of this file). +func (l *exportLedger) mutateAndBroadcast(ctx context.Context, body func() bool) { l.slow.Lock() - delete(l.busy, busid) - if removeExport { - delete(l.exports, busid) - } + changed := body() l.slow.Unlock() - l.BroadcastIfChanged(ctx) + if changed { + l.BroadcastIfChanged(ctx) + } } // IssueLease captures the current broadcast sequence as opaque metadata @@ -301,96 +315,127 @@ func (l *exportLedger) ConsumeLeaseAndReserve(ctx context.Context, request Impor var ( export Export identity ExportLeaseIdentity + phase1OK bool + reason string ) - l.slow.Lock() - now := l.now() - l.cleanupExpiredLocked(now) + l.mutateAndBroadcast(ctx, func() bool { + now := l.now() + changed := l.cleanupExpiredLocked(now) - lease, found := l.leases[request.BusID] - if !found { - l.slow.Unlock() - return nil, false, "lease not found" + lease, found := l.leases[request.BusID] + if !found { + reason = "lease not found" + return changed + } + if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { + reason = "lease mismatch" + return changed + } + if !now.Before(lease.Expires) { + delete(l.leases, request.BusID) + reason = "lease expired" + return true + } + current, stillExported := l.exports[request.BusID] + if !stillExported { + delete(l.leases, request.BusID) + reason = "unknown busid" + return true + } + identity = current.LeaseIdentity() + if identity != lease.Identity { + delete(l.leases, request.BusID) + reason = "lease stale" + return true + } + if l.busy[request.BusID] { + delete(l.leases, request.BusID) + reason = deviceStateBusy + return true + } + export = current + phase1OK = true + return changed + }) + if !phase1OK { + return nil, false, reason } - if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { - l.slow.Unlock() - return nil, false, "lease mismatch" - } - if !now.Before(lease.Expires) { - delete(l.leases, request.BusID) - l.slow.Unlock() - return nil, false, "lease expired" - } - export, stillExported := l.exports[request.BusID] - if !stillExported { - delete(l.leases, request.BusID) - l.slow.Unlock() - return nil, false, "unknown busid" - } - identity = export.LeaseIdentity() - if identity != lease.Identity { - delete(l.leases, request.BusID) - l.slow.Unlock() - return nil, false, "lease stale" - } - if l.busy[request.BusID] { - delete(l.leases, request.BusID) - l.slow.Unlock() - return nil, false, deviceStateBusy - } - l.slow.Unlock() leaseOK, leaseReason := export.LeaseCheck(ctx) if !leaseOK { - l.slow.Lock() - currentLease, exists := l.leases[request.BusID] - if exists && currentLease.ID == request.LeaseID && currentLease.ClientNonce == request.ClientNonce { + l.mutateAndBroadcast(ctx, func() bool { + currentLease, exists := l.leases[request.BusID] + if !exists { + return false + } + if currentLease.ID != request.LeaseID || currentLease.ClientNonce != request.ClientNonce { + return false + } delete(l.leases, request.BusID) - } - l.slow.Unlock() + return true + }) return nil, false, leaseReason } - l.slow.Lock() - defer l.slow.Unlock() + var ( + finalExport Export + finalOK bool + ) + l.mutateAndBroadcast(ctx, func() bool { + now := l.now() + changed := l.cleanupExpiredLocked(now) - now = l.now() - l.cleanupExpiredLocked(now) - - lease, found = l.leases[request.BusID] - if !found { - return nil, false, "lease not found" - } - if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { - return nil, false, "lease mismatch" - } - if !now.Before(lease.Expires) { + lease, found := l.leases[request.BusID] + if !found { + reason = "lease not found" + return changed + } + if lease.ID != request.LeaseID || lease.ClientNonce != request.ClientNonce { + reason = "lease mismatch" + return changed + } + if !now.Before(lease.Expires) { + delete(l.leases, request.BusID) + reason = "lease expired" + return true + } + current, stillExported := l.exports[request.BusID] + if !stillExported { + delete(l.leases, request.BusID) + reason = "unknown busid" + return true + } + if lease.Identity != identity || current.LeaseIdentity() != identity { + delete(l.leases, request.BusID) + reason = "lease stale" + return true + } + if l.busy[request.BusID] { + delete(l.leases, request.BusID) + reason = deviceStateBusy + return true + } delete(l.leases, request.BusID) - return nil, false, "lease expired" + l.busy[request.BusID] = true + finalExport = current + finalOK = true + return true + }) + if !finalOK { + return nil, false, reason } - current, stillExported := l.exports[request.BusID] - if !stillExported { - delete(l.leases, request.BusID) - return nil, false, "unknown busid" - } - if lease.Identity != identity || current.LeaseIdentity() != identity { - delete(l.leases, request.BusID) - return nil, false, "lease stale" - } - if l.busy[request.BusID] { - delete(l.leases, request.BusID) - return nil, false, deviceStateBusy - } - delete(l.leases, request.BusID) - l.busy[request.BusID] = true - return current, true, "" + return finalExport, true, "" } -func (l *exportLedger) cleanupExpiredLocked(now time.Time) { +func (l *exportLedger) cleanupExpiredLocked(now time.Time) bool { + changed := false for busid, lease := range l.leases { if !now.Before(lease.Expires) { delete(l.leases, busid) + changed = true } } + return changed } // Subscribe enqueues a freshly computed snapshot to extension-capable @@ -445,18 +490,24 @@ func (l *exportLedger) Subscribe(ctx context.Context, conn net.Conn, capabilitie } // Unsubscribe leaves the subscriber's send channel for the GC to -// reclaim; the transport read loop has already exited. -func (l *exportLedger) Unsubscribe(sub *exportSubscriber) { +// reclaim; the transport read loop has already exited. Any leases the +// subscriber held are released and the resulting state change is +// broadcast so remaining subscribers see the busid become available +// again. +func (l *exportLedger) Unsubscribe(ctx context.Context, sub *exportSubscriber) { l.fast.Lock() delete(l.subs, sub.id) l.fast.Unlock() - l.slow.Lock() - for busid, lease := range l.leases { - if lease.SubscriberID == sub.id { - delete(l.leases, busid) + l.mutateAndBroadcast(ctx, func() bool { + released := false + for busid, lease := range l.leases { + if lease.SubscriberID == sub.id { + delete(l.leases, busid) + released = true + } } - } - l.slow.Unlock() + return released + }) } // CloseAllSubscribers returns the underlying connections so the caller diff --git a/service/usbip/export_ledger_test.go b/service/usbip/export_ledger_test.go index 9e8a8a35d..ca32704de 100644 --- a/service/usbip/export_ledger_test.go +++ b/service/usbip/export_ledger_test.go @@ -121,6 +121,61 @@ func TestConsumeLeaseAndReserveRejectsUnavailableExport(t *testing.T) { } } +func TestUnsubscribeBroadcastsLeaseRelease(t *testing.T) { + ctx := context.Background() + ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) + exp := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001} + + ledger.ApplyHostSnapshot(map[string]Export{exp.busid: exp}, nil) + ledger.SeedBroadcastState(ctx) + + holder, _ := ledger.Subscribe(ctx, nil, controlCapabilities) + drainSubscriber(holder) + + response := ledger.IssueLease(ctx, holder.id, controlLeaseRequest{BusID: exp.busid, ClientNonce: 7}) + if response.ErrorCode != "" { + t.Fatalf("IssueLease failed: %s", response.ErrorMessage) + } + // Pretend a topology event flushed the busy state into the broadcast + // baseline, matching the real-world scenario where hotplug churn keeps + // l.state roughly in sync with reality. Without this the diff machinery + // in BroadcastIfChanged has no busy→available transition to emit. + ledger.BroadcastIfChanged(ctx) + drainSubscriber(holder) + + observer, _ := ledger.Subscribe(ctx, nil, controlCapabilities) + drainSubscriber(observer) + + ledger.Unsubscribe(ctx, holder) + + select { + case msg := <-observer.send: + if msg.Frame.Type != controlFrameDeviceDelta { + t.Fatalf("expected device delta after lease release, got frame type %d", msg.Frame.Type) + } + var delta controlDeviceDelta + err := unmarshalControlPayload(msg.Payload, &delta) + if err != nil { + t.Fatal(err) + } + if len(delta.Updated) != 1 || delta.Updated[0].State != deviceStateAvailable { + t.Fatalf("expected one Updated entry flipped to available, got %#v", delta) + } + default: + t.Fatal("expected observer to receive a delta after holder's lease was released") + } +} + +func drainSubscriber(sub *exportSubscriber) { + for { + select { + case <-sub.send: + default: + return + } + } +} + func TestConsumeLeaseAndReserveMarksBusyOnSuccess(t *testing.T) { ctx := context.Background() ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) }) @@ -172,11 +227,15 @@ func (e *testExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot { if onSnapshot != nil { onSnapshot() } + state := deviceStateAvailable + if busy { + state = deviceStateBusy + } return ExportSnapshot{ Entry: DeviceEntry{ Info: e.deviceInfo(), }, - State: deviceStateAvailable, + State: state, } } diff --git a/service/usbip/host.go b/service/usbip/host.go index 8718784af..5194b10f0 100644 --- a/service/usbip/host.go +++ b/service/usbip/host.go @@ -18,8 +18,10 @@ type ExportHost interface { Close() error Reconcile(ctx context.Context, isReserved func(busid string) bool) (snapshot map[string]Export, released []string, err error) FinishImport(ctx context.Context, busid string) (released bool, err error) - // Events returning (nil, nil) means "no native event source; rely - // on polling". + // Events returns a coalescing channel that signals topology + // changes; the channel is closed when ctx is cancelled. A non-nil + // error means the host could not subscribe and the server must not + // continue. Events(ctx context.Context) (<-chan struct{}, error) } diff --git a/service/usbip/server.go b/service/usbip/server.go index d77ee983d..a993c3d8e 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -81,6 +81,11 @@ func (s *ServerService) Start(stage adapter.StartStage) error { if err != nil { return err } + events, err := s.host.Events(s.ctx) + if err != nil { + _ = s.host.Close() + return E.Cause(err, "subscribe topology events") + } err = s.reconcileAndBroadcast(false) if err != nil { _ = s.host.Close() @@ -92,7 +97,7 @@ func (s *ServerService) Start(stage adapter.StartStage) error { return err } go s.acceptLoop(tcpListener) - go s.eventLoop() + go s.eventLoop(events) return nil } @@ -126,15 +131,7 @@ func (s *ServerService) Close() error { return err } -func (s *ServerService) eventLoop() { - events, err := s.host.Events(s.ctx) - if err != nil { - s.logger.Warn("subscribe topology events: ", err) - return - } - if events == nil { - return - } +func (s *ServerService) eventLoop(events <-chan struct{}) { for { select { case <-s.ctx.Done(): @@ -144,7 +141,7 @@ func (s *ServerService) eventLoop() { return } } - err = s.reconcileAndBroadcast(true) + err := s.reconcileAndBroadcast(true) if err != nil { s.logger.Warn("reconcile exports: ", err) } @@ -231,7 +228,7 @@ func (s *ServerService) handleControlConn(conn net.Conn) { } capabilities := hello.Capabilities & controlCapabilities sub, seq := s.ledger.Subscribe(s.ctx, conn, capabilities) - defer s.ledger.Unsubscribe(sub) + defer s.ledger.Unsubscribe(s.ctx, sub) err = writeControlMessage(conn, controlFrame{ Type: controlFrameAck, Version: controlProtocolVersion,