wangjia/sing-box
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

usbip: drop ctx ceremony from leaf APIs, move ExportHost lifecycle ctx into constructor

Browse Source 
Export.{Snapshot,LeaseCheck,DeviceInfo}, ExportHost.{Reconcile,FinishImport},
ImportHost.Start, and all 11 exportLedger methods carried ctx params that
implementations never consumed (linux FinishImport now reaches into h.runCtx
internally). UrbTransaction.{Wait,Cancel} did consume ctx, but every call site
passed context.Background(), forcing the reverse pattern in endpoint_darwin
where e.ctx was already cancelled. Cancel becomes synchronous, Wait reads
under the close(t.done) happens-before. ExportHost's runCtx now derives in
newPlatformExportHost so Start() is just precondition-check (linux
ensureKernelPath / darwin no-op), and the three pre-Start nil defences in
Close/Events fall out.
This commit is contained in:
世界
2026-05-17 01:02:36 +08:00
parent 1615c56504
commit b9cf48a827
14 changed files with 145 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/usbip/client.go
+1 -1
View File
@@ -79,7 +79,7 @@ func (c *ClientService) Start(stage adapter.StartStage) error {
if stage != adapter.StartStateStart {
return nil
}
err := c.host.Start(c.ctx)
err := c.host.Start()
if err != nil {
return err
}
service/usbip/endpoint_darwin.go
+2 -2
View File
@@ -162,7 +162,7 @@ func (e *darwinEndpoint) handleCommand(message darwinCIMessage) bool {
}
func (e *darwinEndpoint) abortPending(pending *pendingTransfer) {
_ = pending.transaction.Cancel(context.Background())
_ = pending.transaction.Cancel()
select {
case <-pending.transaction.Done():
case <-e.peer.Done():
@@ -171,7 +171,7 @@ func (e *darwinEndpoint) abortPending(pending *pendingTransfer) {
}
func (e *darwinEndpoint) finalizePending(pending *pendingTransfer) {
response, err := pending.transaction.Wait(context.Background())
response, err := pending.transaction.Wait()
var status int32
var length int
switch {
service/usbip/export_ledger.go
+27 -28
View File
@@ -3,7 +3,6 @@
package usbip
import (
"context"
"net"
"slices"
"strings"
@@ -74,12 +73,12 @@ func newExportLedger(logger log.ContextLogger, ttl time.Duration, now func() tim
// withInventoryWrite broadcasts iff body returns true. body must not
// acquire the broadcast lock.
func (l *exportLedger) withInventoryWrite(ctx context.Context, body func() bool) {
func (l *exportLedger) withInventoryWrite(body func() bool) {
l.inventoryAccess.Lock()
changed := body()
l.inventoryAccess.Unlock()
if changed {
l.BroadcastIfChanged(ctx)
l.BroadcastIfChanged()
}
}
@@ -145,15 +144,15 @@ func (l *exportLedger) ApplyHostSnapshot(snapshot map[string]Export, released []
})
}
func (l *exportLedger) SeedBroadcastState(ctx context.Context) {
nextState := deviceInfoV2Map(l.snapshotDeviceState(ctx))
func (l *exportLedger) SeedBroadcastState() {
nextState := deviceInfoV2Map(l.snapshotDeviceState())
l.broadcastAccess.Lock()
l.state = nextState
l.broadcastAccess.Unlock()
}
func (l *exportLedger) BroadcastIfChanged(ctx context.Context) bool {
nextState := deviceInfoV2Map(l.snapshotDeviceState(ctx))
func (l *exportLedger) BroadcastIfChanged() bool {
nextState := deviceInfoV2Map(l.snapshotDeviceState())
l.broadcastAccess.Lock()
nextSequence := l.seq + 1
@@ -194,7 +193,7 @@ func (l *exportLedger) BroadcastIfChanged(ctx context.Context) bool {
// TryReserveForImport runs LeaseCheck outside the lock and re-checks
// availability before marking busy. Caller must pair success with
// ReleaseImport and broadcast once the session is wired up.
func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (Export, bool, string) {
func (l *exportLedger) TryReserveForImport(busid string) (Export, bool, string) {
var (
export Export
found bool
@@ -211,7 +210,7 @@ func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (E
if reserved {
return nil, false, deviceStateBusy
}
leaseOK, leaseReason := export.LeaseCheck(ctx)
leaseOK, leaseReason := export.LeaseCheck()
if !leaseOK {
return nil, false, leaseReason
}
@@ -238,8 +237,8 @@ func (l *exportLedger) TryReserveForImport(ctx context.Context, busid string) (E
return export, true, ""
}
func (l *exportLedger) ReleaseImport(ctx context.Context, busid string, removeExport bool) {
l.withInventoryWrite(ctx, func() bool {
func (l *exportLedger) ReleaseImport(busid string, removeExport bool) {
l.withInventoryWrite(func() bool {
delete(l.busy, busid)
if removeExport {
delete(l.exports, busid)
@@ -250,7 +249,7 @@ func (l *exportLedger) ReleaseImport(ctx context.Context, busid string, removeEx
// IssueLease pins lease correctness to the export identity; the
// broadcast sequence on the response is opaque metadata for clients.
func (l *exportLedger) IssueLease(ctx context.Context, subID uint64, request controlLeaseRequest) controlLeaseResponse {
func (l *exportLedger) IssueLease(subID uint64, request controlLeaseRequest) controlLeaseResponse {
response := controlLeaseResponse{
BusID: request.BusID,
ClientNonce: request.ClientNonce,
@@ -270,7 +269,7 @@ func (l *exportLedger) IssueLease(ctx context.Context, subID uint64, request con
identity ExportLeaseIdentity
preCheckOK bool
)
l.withInventoryWrite(ctx, func() bool {
l.withInventoryWrite(func() bool {
now := l.now()
changed := l.cleanupExpiredLocked(now)
currentExport, found := l.exports[request.BusID]
@@ -298,14 +297,14 @@ func (l *exportLedger) IssueLease(ctx context.Context, subID uint64, request con
return response
}
leaseOK, leaseReason := export.LeaseCheck(ctx)
leaseOK, leaseReason := export.LeaseCheck()
if !leaseOK {
response.ErrorCode = leaseErrorUnavailable
response.ErrorMessage = leaseReason
return response
}
l.withInventoryWrite(ctx, func() bool {
l.withInventoryWrite(func() bool {
now := l.now()
changed := l.cleanupExpiredLocked(now)
current, stillExported := l.exports[request.BusID]
@@ -346,14 +345,14 @@ func (l *exportLedger) IssueLease(ctx context.Context, subID uint64, request con
// ConsumeLeaseAndReserve consumes the lease on every outcome except an
// ID/nonce mismatch (the latter preserves the lease for the legitimate
// holder). Caller must pair success with ReleaseImport.
func (l *exportLedger) ConsumeLeaseAndReserve(ctx context.Context, request ImportExtRequest) (Export, bool, string) {
func (l *exportLedger) ConsumeLeaseAndReserve(request ImportExtRequest) (Export, bool, string) {
var (
export Export
identity ExportLeaseIdentity
phase1OK bool
reason string
)
l.withInventoryWrite(ctx, func() bool {
l.withInventoryWrite(func() bool {
now := l.now()
changed := l.cleanupExpiredLocked(now)
@@ -396,9 +395,9 @@ func (l *exportLedger) ConsumeLeaseAndReserve(ctx context.Context, request Impor
return nil, false, reason
}
leaseOK, leaseReason := export.LeaseCheck(ctx)
leaseOK, leaseReason := export.LeaseCheck()
if !leaseOK {
l.withInventoryWrite(ctx, func() bool {
l.withInventoryWrite(func() bool {
currentLease, exists := l.leases[request.BusID]
if !exists {
return false
@@ -416,7 +415,7 @@ func (l *exportLedger) ConsumeLeaseAndReserve(ctx context.Context, request Impor
finalExport Export
finalOK bool
)
l.withInventoryWrite(ctx, func() bool {
l.withInventoryWrite(func() bool {
now := l.now()
changed := l.cleanupExpiredLocked(now)
@@ -478,7 +477,7 @@ func (l *exportLedger) cleanupExpiredLocked(now time.Time) bool {
// broadcast fired. 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) {
func (l *exportLedger) Subscribe(conn net.Conn, capabilities uint32) (*exportSubscriber, uint64) {
extended := supportsControlExtensions(capabilities)
var snapshot []DeviceInfoV2
var sequence uint64
@@ -489,7 +488,7 @@ func (l *exportLedger) Subscribe(ctx context.Context, conn net.Conn, capabilitie
sequence = l.seq
l.broadcastAccess.Unlock()
snapshot = l.snapshotDeviceState(ctx)
snapshot = l.snapshotDeviceState()
l.broadcastAccess.Lock()
if sequence == l.seq {
@@ -529,11 +528,11 @@ func (l *exportLedger) Subscribe(ctx context.Context, conn net.Conn, capabilitie
// 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) {
func (l *exportLedger) Unsubscribe(sub *exportSubscriber) {
l.broadcastAccess.Lock()
delete(l.subs, sub.id)
l.broadcastAccess.Unlock()
l.withInventoryWrite(ctx, func() bool {
l.withInventoryWrite(func() bool {
released := false
for busid, lease := range l.leases {
if lease.SubscriberID == sub.id {
@@ -566,7 +565,7 @@ func (l *exportLedger) ResetForClose() {
})
}
func (l *exportLedger) HandleControlLeaseRequest(ctx context.Context, sub *exportSubscriber, payload []byte) {
func (l *exportLedger) HandleControlLeaseRequest(sub *exportSubscriber, payload []byte) {
var request controlLeaseRequest
err := unmarshalControlPayload(payload, &request)
if err != nil {
@@ -582,7 +581,7 @@ func (l *exportLedger) HandleControlLeaseRequest(ctx context.Context, sub *expor
}, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: sequence})
return
}
response := l.IssueLease(ctx, sub.id, request)
response := l.IssueLease(sub.id, request)
l.broadcastAccess.Lock()
sequence := l.seq
l.broadcastAccess.Unlock()
@@ -592,7 +591,7 @@ func (l *exportLedger) HandleControlLeaseRequest(ctx context.Context, sub *expor
}, response, controlFrame{Type: controlFrameChanged, Version: controlProtocolVersion, Sequence: sequence})
}
func (l *exportLedger) snapshotDeviceState(ctx context.Context) []DeviceInfoV2 {
func (l *exportLedger) snapshotDeviceState() []DeviceInfoV2 {
type entry struct {
export Export
busy bool
@@ -612,7 +611,7 @@ func (l *exportLedger) snapshotDeviceState(ctx context.Context) []DeviceInfoV2 {
})
out := make([]DeviceInfoV2, 0, len(entries))
for _, e := range entries {
snapshot := e.export.Snapshot(ctx, e.busy)
snapshot := e.export.Snapshot(e.busy)
if snapshot.State == deviceStateUnavailable && snapshot.Entry.Info.BusIDString() == "" {
continue
}
service/usbip/export_ledger_darwin_test.go
+3 -5
View File
@@ -3,7 +3,6 @@
package usbip
import (
"context"
"testing"
"time"
@@ -13,7 +12,6 @@ import (
func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) {
t.Parallel()
ctx := context.Background()
ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) })
entry := darwinFakeDeviceEntry()
export := &darwinExport{
@@ -23,9 +21,9 @@ func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) {
}
ledger.ApplyHostSnapshot(map[string]Export{export.busid: export}, nil)
ledger.SeedBroadcastState(ctx)
ledger.SeedBroadcastState()
sub, _ := ledger.Subscribe(ctx, nil, controlCapabilities)
sub, _ := ledger.Subscribe(nil, controlCapabilities)
select {
case <-sub.send:
case <-time.After(time.Second):
@@ -35,7 +33,7 @@ func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) {
export.stale = true
export.pendingRegistryID = 0x5678
if !ledger.BroadcastIfChanged(ctx) {
if !ledger.BroadcastIfChanged() {
t.Fatal("expected stale darwin export to broadcast an update")
}
service/usbip/export_ledger_test.go
+22 -27
View File
@@ -10,22 +10,21 @@ import (
)
func TestSubscribeRetriesSnapshotWhenSequenceAdvances(t *testing.T) {
ctx := context.Background()
ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) })
oldExport := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001}
newExport := &testExport{busid: "2-1", vendorID: 0x2222, productID: 0x0002}
ledger.ApplyHostSnapshot(map[string]Export{oldExport.busid: oldExport}, nil)
ledger.SeedBroadcastState(ctx)
ledger.SeedBroadcastState()
oldExport.onSnapshot = func() {
ledger.ApplyHostSnapshot(map[string]Export{newExport.busid: newExport}, nil)
if !ledger.BroadcastIfChanged(ctx) {
if !ledger.BroadcastIfChanged() {
t.Fatal("expected broadcast after replacing export")
}
}
sub, sequence := ledger.Subscribe(ctx, nil, controlCapabilities)
sub, sequence := ledger.Subscribe(nil, controlCapabilities)
if sequence != 1 {
t.Fatalf("expected subscription sequence 1, got %d", sequence)
}
@@ -55,16 +54,15 @@ func TestSubscribeRetriesSnapshotWhenSequenceAdvances(t *testing.T) {
}
func TestConsumeLeaseAndReserveRejectsIdentityReplacement(t *testing.T) {
ctx := context.Background()
ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) })
original := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001, identity: "linux:original"}
replacement := &testExport{busid: "1-1", vendorID: 0x1111, productID: 0x0001, identity: "linux:replacement"}
ledger.ApplyHostSnapshot(map[string]Export{original.busid: original}, nil)
lease := ledger.IssueLease(ctx, 1, controlLeaseRequest{BusID: original.busid, ClientNonce: 7})
lease := ledger.IssueLease(1, controlLeaseRequest{BusID: original.busid, ClientNonce: 7})
ledger.ApplyHostSnapshot(map[string]Export{replacement.busid: replacement}, nil)
_, ok, reason := ledger.ConsumeLeaseAndReserve(ctx, ImportExtRequest{
_, ok, reason := ledger.ConsumeLeaseAndReserve(ImportExtRequest{
BusID: original.busid,
LeaseID: lease.LeaseID,
ClientNonce: lease.ClientNonce,
@@ -78,14 +76,13 @@ func TestConsumeLeaseAndReserveRejectsIdentityReplacement(t *testing.T) {
}
func TestConsumeLeaseAndReserveRejectsUnavailableExport(t *testing.T) {
ctx := context.Background()
ledger := newExportLedger(nil, time.Second, func() time.Time { return time.Unix(0, 0) })
available := true
exp := &testExport{
busid: "1-1",
vendorID: 0x1111,
productID: 0x0001,
leaseCheck: func(context.Context) (bool, string) {
leaseCheck: func() (bool, string) {
if available {
return true, ""
}
@@ -94,10 +91,10 @@ func TestConsumeLeaseAndReserveRejectsUnavailableExport(t *testing.T) {
}
ledger.ApplyHostSnapshot(map[string]Export{exp.busid: exp}, nil)
lease := ledger.IssueLease(ctx, 1, controlLeaseRequest{BusID: exp.busid, ClientNonce: 9})
lease := ledger.IssueLease(1, controlLeaseRequest{BusID: exp.busid, ClientNonce: 9})
available = false
_, ok, reason := ledger.ConsumeLeaseAndReserve(ctx, ImportExtRequest{
_, ok, reason := ledger.ConsumeLeaseAndReserve(ImportExtRequest{
BusID: exp.busid,
LeaseID: lease.LeaseID,
ClientNonce: lease.ClientNonce,
@@ -109,7 +106,7 @@ func TestConsumeLeaseAndReserveRejectsUnavailableExport(t *testing.T) {
t.Fatalf("expected capture released, got %q", reason)
}
_, ok, reason = ledger.ConsumeLeaseAndReserve(ctx, ImportExtRequest{
_, ok, reason = ledger.ConsumeLeaseAndReserve(ImportExtRequest{
BusID: exp.busid,
LeaseID: lease.LeaseID,
ClientNonce: lease.ClientNonce,
@@ -123,17 +120,16 @@ 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)
ledger.SeedBroadcastState()
holder, _ := ledger.Subscribe(ctx, nil, controlCapabilities)
holder, _ := ledger.Subscribe(nil, controlCapabilities)
drainSubscriber(holder)
response := ledger.IssueLease(ctx, holder.id, controlLeaseRequest{BusID: exp.busid, ClientNonce: 7})
response := ledger.IssueLease(holder.id, controlLeaseRequest{BusID: exp.busid, ClientNonce: 7})
if response.ErrorCode != "" {
t.Fatalf("IssueLease failed: %s", response.ErrorMessage)
}
@@ -141,13 +137,13 @@ func TestUnsubscribeBroadcastsLeaseRelease(t *testing.T) {
// 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)
ledger.BroadcastIfChanged()
drainSubscriber(holder)
observer, _ := ledger.Subscribe(ctx, nil, controlCapabilities)
observer, _ := ledger.Subscribe(nil, controlCapabilities)
drainSubscriber(observer)
ledger.Unsubscribe(ctx, holder)
ledger.Unsubscribe(holder)
select {
case msg := <-observer.send:
@@ -178,14 +174,13 @@ func drainSubscriber(sub *exportSubscriber) {
}
func TestConsumeLeaseAndReserveMarksBusyOnSuccess(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)
lease := ledger.IssueLease(ctx, 1, controlLeaseRequest{BusID: exp.busid, ClientNonce: 11})
lease := ledger.IssueLease(1, controlLeaseRequest{BusID: exp.busid, ClientNonce: 11})
reserved, ok, reason := ledger.ConsumeLeaseAndReserve(ctx, ImportExtRequest{
reserved, ok, reason := ledger.ConsumeLeaseAndReserve(ImportExtRequest{
BusID: exp.busid,
LeaseID: lease.LeaseID,
ClientNonce: lease.ClientNonce,
@@ -207,7 +202,7 @@ type testExport struct {
productID uint16
identity ExportLeaseIdentity
leaseCheck func(context.Context) (bool, string)
leaseCheck func() (bool, string)
onSnapshot func()
}
@@ -222,7 +217,7 @@ func (e *testExport) LeaseIdentity() ExportLeaseIdentity {
return ExportLeaseIdentity(e.busid)
}
func (e *testExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
func (e *testExport) Snapshot(busy bool) ExportSnapshot {
onSnapshot := e.onSnapshot
e.onSnapshot = nil
if onSnapshot != nil {
@@ -240,14 +235,14 @@ func (e *testExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
}
}
func (e *testExport) LeaseCheck(ctx context.Context) (bool, string) {
func (e *testExport) LeaseCheck() (bool, string) {
if e.leaseCheck != nil {
return e.leaseCheck(ctx)
return e.leaseCheck()
}
return true, ""
}
func (e *testExport) DeviceInfo(ctx context.Context) (DeviceInfoTruncated, error) {
func (e *testExport) DeviceInfo() (DeviceInfoTruncated, error) {
return e.deviceInfo(), nil
}
service/usbip/host.go
+7 -7
View File
@@ -10,17 +10,17 @@ import (
// ExportHost lifecycle: Start → Reconcile* → Close. Callers must apply
// the Reconcile snapshot and released list even when err != nil.
type ExportHost interface {
Start(ctx context.Context) error
Start() error
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)
Reconcile(isReserved func(busid string) bool) (snapshot map[string]Export, released []string, err error)
FinishImport(busid string) (released bool, err error)
// Events MUST be called after Start succeeds. The channel closes when
// Close() runs.
Events() (<-chan struct{}, error)
}
type ImportHost interface {
Start(ctx context.Context) error
Start() error
Close() error
// Attach takes ownership of conn for the lifetime of the import.
Attach(ctx context.Context, info DeviceInfoTruncated, conn net.Conn) (AttachedSession, error)
@@ -34,10 +34,10 @@ type ExportLeaseIdentity string
// swap it into their committed map under the host's own lock.
type Export interface {
BusID() string
Snapshot(ctx context.Context, busy bool) ExportSnapshot
Snapshot(busy bool) ExportSnapshot
LeaseIdentity() ExportLeaseIdentity
LeaseCheck(ctx context.Context) (ok bool, reason string)
DeviceInfo(ctx context.Context) (DeviceInfoTruncated, error)
LeaseCheck() (ok bool, reason string)
DeviceInfo() (DeviceInfoTruncated, error)
NewServerDataSession(ctx context.Context, conn net.Conn) (DataSession, error)
}
service/usbip/host_darwin.go
+17 -22
View File
@@ -20,8 +20,8 @@ import (
"golang.org/x/sys/unix"
)
func newPlatformExportHost(logger log.ContextLogger, matches []option.USBIPDeviceMatch) (ExportHost, error) {
return newDarwinExportHost(logger, matches), nil
func newPlatformExportHost(ctx context.Context, logger log.ContextLogger, matches []option.USBIPDeviceMatch) (ExportHost, error) {
return newDarwinExportHost(ctx, logger, matches), nil
}
func newPlatformImportHost(logger log.ContextLogger) (ImportHost, error) {
@@ -42,23 +42,21 @@ type darwinExportHost struct {
watcher *darwinUSBHostDeviceWatcher
}
func newDarwinExportHost(logger log.ContextLogger, matches []option.USBIPDeviceMatch) *darwinExportHost {
func newDarwinExportHost(ctx context.Context, logger log.ContextLogger, matches []option.USBIPDeviceMatch) *darwinExportHost {
runCtx, runCancel := context.WithCancel(ctx)
return &darwinExportHost{
logger: logger,
matches: matches,
exports: make(map[string]*darwinExport),
runCtx: runCtx,
runCancel: runCancel,
logger: logger,
matches: matches,
exports: make(map[string]*darwinExport),
}
}
func (h *darwinExportHost) Start(ctx context.Context) error {
h.runCtx, h.runCancel = context.WithCancel(ctx)
return nil
}
func (h *darwinExportHost) Start() error { return nil }
func (h *darwinExportHost) Close() error {
if h.runCancel != nil {
h.runCancel()
}
h.runCancel()
h.access.Lock()
watcher := h.watcher
h.watcher = nil
@@ -77,9 +75,6 @@ func (h *darwinExportHost) Close() error {
}
func (h *darwinExportHost) Events() (<-chan struct{}, error) {
if h.runCtx == nil {
return nil, E.New("usbip host: Events called before Start")
}
ch := make(chan struct{}, 1)
signal := func() {
select {
@@ -109,7 +104,7 @@ func (h *darwinExportHost) Events() (<-chan struct{}, error) {
return ch, nil
}
func (h *darwinExportHost) Reconcile(ctx context.Context, isReserved func(busid string) bool) (map[string]Export, []string, error) {
func (h *darwinExportHost) Reconcile(isReserved func(busid string) bool) (map[string]Export, []string, error) {
devices, err := darwinCopyUSBHostDevices()
if err != nil {
return h.snapshotSelf(), nil, E.Cause(err, "enumerate IOUSBHost devices")
@@ -211,7 +206,7 @@ func (h *darwinExportHost) Reconcile(ctx context.Context, isReserved func(busid
return snapshotDarwinExports(committed), released, nil
}
func (h *darwinExportHost) FinishImport(ctx context.Context, busid string) (bool, error) {
func (h *darwinExportHost) FinishImport(busid string) (bool, error) {
h.access.Lock()
exp, ok := h.exports[busid]
if !ok || !exp.stale {
@@ -315,7 +310,7 @@ func (e *darwinExport) staleReason() string {
return "capture released"
}
func (e *darwinExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
func (e *darwinExport) Snapshot(busy bool) ExportSnapshot {
stableID := fmt.Sprintf("darwin-registry:%016x", e.registryID)
if e.stale {
return ExportSnapshot{
@@ -338,14 +333,14 @@ func (e *darwinExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
}
}
func (e *darwinExport) LeaseCheck(ctx context.Context) (bool, string) {
func (e *darwinExport) LeaseCheck() (bool, string) {
if e.stale {
return false, e.staleReason()
}
return true, ""
}
func (e *darwinExport) DeviceInfo(ctx context.Context) (DeviceInfoTruncated, error) {
func (e *darwinExport) DeviceInfo() (DeviceInfoTruncated, error) {
return e.entry.Info, nil
}
@@ -360,7 +355,7 @@ type darwinImportHost struct {
logger log.ContextLogger
}
func (h *darwinImportHost) Start(ctx context.Context) error {
func (h *darwinImportHost) Start() error {
return nil
}
service/usbip/host_linux.go
+19 -26
View File
@@ -23,8 +23,8 @@ import (
"golang.org/x/sys/unix"
)
func newPlatformExportHost(logger log.ContextLogger, matches []option.USBIPDeviceMatch) (ExportHost, error) {
return newLinuxExportHost(logger, matches), nil
func newPlatformExportHost(ctx context.Context, logger log.ContextLogger, matches []option.USBIPDeviceMatch) (ExportHost, error) {
return newLinuxExportHost(ctx, logger, matches), nil
}
func newPlatformImportHost(logger log.ContextLogger) (ImportHost, error) {
@@ -153,27 +153,23 @@ type linuxReconcilePlan struct {
released []string
}
func newLinuxExportHost(logger log.ContextLogger, matches []option.USBIPDeviceMatch) *linuxExportHost {
func newLinuxExportHost(ctx context.Context, logger log.ContextLogger, matches []option.USBIPDeviceMatch) *linuxExportHost {
runCtx, runCancel := context.WithCancel(ctx)
return &linuxExportHost{
logger: logger,
matches: matches,
exports: make(map[string]*linuxExport),
runCtx: runCtx,
runCancel: runCancel,
logger: logger,
matches: matches,
exports: make(map[string]*linuxExport),
}
}
func (h *linuxExportHost) Start(ctx context.Context) error {
err := ensureKernelPath(sysUsbipHostDriver, "usbip-host", "usbip-host driver")
if err != nil {
return err
}
h.runCtx, h.runCancel = context.WithCancel(ctx)
return nil
func (h *linuxExportHost) Start() error {
return ensureKernelPath(sysUsbipHostDriver, "usbip-host", "usbip-host driver")
}
func (h *linuxExportHost) Close() error {
if h.runCancel != nil {
h.runCancel()
}
h.runCancel()
h.access.Lock()
exports := h.exports
h.exports = make(map[string]*linuxExport)
@@ -188,9 +184,6 @@ func (h *linuxExportHost) Close() error {
}
func (h *linuxExportHost) Events() (<-chan struct{}, error) {
if h.runCtx == nil {
return nil, E.New("usbip host: Events called before Start")
}
ch := make(chan struct{}, 1)
go h.ueventLoop(h.runCtx, ch)
return ch, nil
@@ -295,7 +288,7 @@ func classifyLinuxReconcile(current map[string]*linuxExport, desired map[string]
return plan
}
func (h *linuxExportHost) Reconcile(ctx context.Context, isReserved func(busid string) bool) (map[string]Export, []string, error) {
func (h *linuxExportHost) Reconcile(isReserved func(busid string) bool) (map[string]Export, []string, error) {
devices, err := listUSBDevices()
if err != nil {
return h.snapshotSelf(), nil, E.Cause(err, "enumerate usb devices")
@@ -412,12 +405,12 @@ func (h *linuxExportHost) Reconcile(ctx context.Context, isReserved func(busid s
return snapshotLinuxExports(committed), released, E.Errors(reconcileErrors...)
}
func (h *linuxExportHost) FinishImport(ctx context.Context, busid string) (bool, error) {
func (h *linuxExportHost) FinishImport(busid string) (bool, error) {
err := writeSysfs(filepath.Join(sysBusUSBDevices, busid, "usbip_sockfd"), "-1")
if err != nil && !os.IsNotExist(err) && !isMissingUSBDeviceError(err) {
h.logger.Debug("release ", busid, " from usbip-host: ", err)
}
waitForUsbipStatusCleared(ctx, busid)
waitForUsbipStatusCleared(h.runCtx, busid)
h.access.Lock()
exp, ok := h.exports[busid]
h.access.Unlock()
@@ -669,7 +662,7 @@ func (e *linuxExport) LeaseIdentity() ExportLeaseIdentity {
return e.identity.LeaseIdentity()
}
func (e *linuxExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
func (e *linuxExport) Snapshot(busy bool) ExportSnapshot {
stableID := "linux-busid:" + e.descriptor.BusID
if e.descriptor.Serial != "" {
stableID = fmt.Sprintf("usb:%04x:%04x:%s", e.descriptor.VendorID, e.descriptor.ProductID, e.descriptor.Serial)
@@ -721,7 +714,7 @@ func (e *linuxExport) Snapshot(ctx context.Context, busy bool) ExportSnapshot {
}
}
func (e *linuxExport) LeaseCheck(ctx context.Context) (bool, string) {
func (e *linuxExport) LeaseCheck() (bool, string) {
if e.stale {
return false, "device replaced"
}
@@ -735,7 +728,7 @@ func (e *linuxExport) LeaseCheck(ctx context.Context) (bool, string) {
return true, ""
}
func (e *linuxExport) DeviceInfo(ctx context.Context) (DeviceInfoTruncated, error) {
func (e *linuxExport) DeviceInfo() (DeviceInfoTruncated, error) {
return e.descriptor.toProtocol(), nil
}
@@ -771,7 +764,7 @@ type linuxImportHost struct {
ports map[int]struct{}
}
func (h *linuxImportHost) Start(ctx context.Context) error {
func (h *linuxImportHost) Start() error {
return ensureKernelPath(sysVHCIControllerV0, "vhci-hcd", "vhci_hcd.0")
}
service/usbip/linux_test.go
+5 -5
View File
@@ -378,7 +378,7 @@ func TestUSBIPLinuxSmoke(t *testing.T) {
require.NoError(t, err)
require.Equal(t, gadget.busid, device.BusID)
host := newLinuxExportHost(newTestLogger(t), nil)
host := newLinuxExportHost(context.Background(), newTestLogger(t), nil)
exp, err := host.bindOne(&device)
require.NoError(t, err)
setLinuxExport(host, exp)
@@ -413,10 +413,10 @@ func TestUSBIPLinuxReconcileReleaseRestoresOriginalDriver(t *testing.T) {
requireVHCI(t)
gadget := newTestUSBGadget(t)
host := newLinuxExportHost(newTestLogger(t), []option.USBIPDeviceMatch{{BusID: gadget.busid}})
require.NoError(t, host.Start(context.Background()))
host := newLinuxExportHost(context.Background(), newTestLogger(t), []option.USBIPDeviceMatch{{BusID: gadget.busid}})
require.NoError(t, host.Start())
snapshot, released, err := host.Reconcile(context.Background(), func(string) bool { return false })
snapshot, released, err := host.Reconcile(func(string) bool { return false })
require.NoError(t, err)
require.Empty(t, released)
_, exported := snapshot[gadget.busid]
@@ -427,7 +427,7 @@ func TestUSBIPLinuxReconcileReleaseRestoresOriginalDriver(t *testing.T) {
require.Equal(t, "usbip-host", driver)
host.matches = nil
snapshot, released, err = host.Reconcile(context.Background(), func(string) bool { return false })
snapshot, released, err = host.Reconcile(func(string) bool { return false })
require.NoError(t, err)
require.Equal(t, []string{gadget.busid}, released)
require.Empty(t, snapshot)
service/usbip/server.go
+21 -20
View File
@@ -50,11 +50,12 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string,
if options.ListenPort == 0 {
options.ListenPort = DefaultPort
}
host, err := newPlatformExportHost(logger, options.Devices)
ctx, cancel := context.WithCancel(ctx)
host, err := newPlatformExportHost(ctx, logger, options.Devices)
if err != nil {
cancel()
return nil, err
}
ctx, cancel := context.WithCancel(ctx)
return &ServerService{
Adapter: boxService.NewAdapter(C.TypeUSBIPServer, tag),
ctx: ctx,
@@ -83,7 +84,7 @@ func (s *ServerService) Start(stage adapter.StartStage) (err error) {
_ = s.host.Close()
}
}()
err = s.host.Start(s.ctx)
err = s.host.Start()
if err != nil {
return err
}
@@ -154,8 +155,8 @@ func (s *ServerService) eventLoop(events <-chan struct{}) {
func (s *ServerService) tearDownPreparedSession(busid string, session DataSession) {
_ = session.Close()
<-session.Done()
released, _ := s.host.FinishImport(s.ctx, busid)
s.ledger.ReleaseImport(s.ctx, busid, released)
released, _ := s.host.FinishImport(busid)
s.ledger.ReleaseImport(busid, released)
if released {
err := s.reconcileAndBroadcast(true)
if err != nil {
@@ -167,15 +168,15 @@ func (s *ServerService) tearDownPreparedSession(busid string, session DataSessio
func (s *ServerService) reconcileAndBroadcast(notify bool) error {
s.reconcileAccess.Lock()
defer s.reconcileAccess.Unlock()
if s.ctx != nil && s.ctx.Err() != nil {
if s.ctx.Err() != nil {
return nil
}
snapshot, released, err := s.host.Reconcile(s.ctx, s.ledger.IsReserved)
snapshot, released, err := s.host.Reconcile(s.ledger.IsReserved)
s.ledger.ApplyHostSnapshot(snapshot, released)
if notify {
s.ledger.BroadcastIfChanged(s.ctx)
s.ledger.BroadcastIfChanged()
} else {
s.ledger.SeedBroadcastState(s.ctx)
s.ledger.SeedBroadcastState()
}
return err
}
@@ -230,8 +231,8 @@ func (s *ServerService) handleControlConn(conn net.Conn) {
return
}
capabilities := hello.Capabilities & controlCapabilities
sub, seq := s.ledger.Subscribe(s.ctx, conn, capabilities)
defer s.ledger.Unsubscribe(s.ctx, sub)
sub, seq := s.ledger.Subscribe(conn, capabilities)
defer s.ledger.Unsubscribe(sub)
err = writeControlMessage(conn, controlFrame{
Type: controlFrameAck,
Version: controlProtocolVersion,
@@ -267,7 +268,7 @@ func (s *ServerService) buildDevListEntries() []DeviceEntry {
}
entries := make([]DeviceEntry, 0, len(exports))
for _, export := range exports {
snapshot := export.Snapshot(s.ctx, false)
snapshot := export.Snapshot(false)
if snapshot.State != deviceStateAvailable {
continue
}
@@ -282,7 +283,7 @@ func (s *ServerService) handleImportExt(conn net.Conn) bool {
s.logger.Debug("read import-ext body: ", err)
return false
}
export, ok, reason := s.ledger.ConsumeLeaseAndReserve(s.ctx, request)
export, ok, reason := s.ledger.ConsumeLeaseAndReserve(request)
if !ok {
s.logger.Info("import-ext rejected (", request.BusID, ": ", reason, ")")
_ = WriteOpRepImport(conn, OpRepImportExt, OpStatusError, nil)
@@ -292,7 +293,7 @@ func (s *ServerService) handleImportExt(conn net.Conn) bool {
}
func (s *ServerService) handleImportBusID(conn net.Conn, busid string) bool {
export, ok, reason := s.ledger.TryReserveForImport(s.ctx, busid)
export, ok, reason := s.ledger.TryReserveForImport(busid)
if !ok {
s.logger.Info("import rejected (", busid, ": ", reason, ")")
_ = WriteOpRepImport(conn, OpRepImport, OpStatusError, nil)
@@ -306,21 +307,21 @@ func (s *ServerService) handleImportReserved(conn net.Conn, busid string, export
if extended {
opCode = OpRepImportExt
}
info, err := export.DeviceInfo(s.ctx)
info, err := export.DeviceInfo()
if err != nil {
s.ledger.ReleaseImport(s.ctx, busid, false)
s.ledger.ReleaseImport(busid, false)
s.logger.Warn("refresh ", busid, ": ", err)
_ = WriteOpRepImport(conn, opCode, OpStatusError, nil)
return false
}
session, err := export.NewServerDataSession(s.ctx, conn)
if err != nil {
s.ledger.ReleaseImport(s.ctx, busid, false)
s.ledger.ReleaseImport(busid, false)
s.logger.Warn("open data session ", busid, ": ", err)
_ = WriteOpRepImport(conn, opCode, OpStatusError, nil)
return false
}
s.ledger.BroadcastIfChanged(s.ctx)
s.ledger.BroadcastIfChanged()
err = WriteOpRepImport(conn, opCode, OpStatusOK, &info)
if err != nil {
s.logger.Warn("reply import ", busid, ": ", err)
@@ -357,11 +358,11 @@ func (s *ServerService) handleImportReserved(conn net.Conn, busid string, export
s.sessionsAccess.Lock()
delete(s.sessions, session)
s.sessionsAccess.Unlock()
released, err := s.host.FinishImport(s.ctx, busid)
released, err := s.host.FinishImport(busid)
if err != nil {
s.logger.Debug("finish import ", busid, ": ", err)
}
s.ledger.ReleaseImport(s.ctx, busid, released)
s.ledger.ReleaseImport(busid, released)
if released {
err = s.reconcileAndBroadcast(true)
if err != nil {
service/usbip/server_shared.go
+1 -1
View File
@@ -70,7 +70,7 @@ func (s *ServerService) readControlConn(sub *exportSubscriber, done chan<- struc
})
case controlFrameLeaseRequest:
if supportsControlExtensions(sub.capabilities) {
s.ledger.HandleControlLeaseRequest(s.ctx, sub, message.Payload)
s.ledger.HandleControlLeaseRequest(sub, message.Payload)
continue
}
return
service/usbip/urb_transaction.go
+7 -25
View File
@@ -3,7 +3,6 @@
package usbip
import (
"context"
"sync"
)
@@ -32,18 +31,12 @@ func (t *UrbTransaction) Done() <-chan struct{} {
return t.done
}
func (t *UrbTransaction) Wait(ctx context.Context) (SubmitResponse, error) {
select {
case <-t.done:
case <-ctx.Done():
return SubmitResponse{}, ctx.Err()
}
t.access.Lock()
defer t.access.Unlock()
func (t *UrbTransaction) Wait() (SubmitResponse, error) {
<-t.done
return t.response, t.err
}
func (t *UrbTransaction) Cancel(ctx context.Context) error {
func (t *UrbTransaction) Cancel() error {
t.access.Lock()
if t.terminal || t.canceling {
t.access.Unlock()
@@ -51,22 +44,11 @@ func (t *UrbTransaction) Cancel(ctx context.Context) error {
}
t.canceling = true
t.access.Unlock()
type writeResult struct{ err error }
resultCh := make(chan writeResult, 1)
go func() {
resultCh <- writeResult{err: t.peer.cancel(t.seqnum)}
}()
select {
case result := <-resultCh:
if result.err != nil {
t.finalize(SubmitResponse{}, ErrCanceled)
}
return result.err
case <-ctx.Done():
return ctx.Err()
err := t.peer.cancel(t.seqnum)
if err != nil {
t.finalize(SubmitResponse{}, ErrCanceled)
}
return err
}
func (t *UrbTransaction) finalize(response SubmitResponse, err error) {
service/usbip/urb_transaction_test.go
+8 -35
View File
@@ -3,40 +3,13 @@
package usbip
import (
"context"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestUrbTransactionWaitContextCancel(t *testing.T) {
peer, server, _ := newPeerPair(t)
go func() {
_ = server.readSubmit(t)
// Never reply.
}()
transaction, err := peer.Submit(SubmitCommand{
Header: DataHeader{
Command: CmdSubmit,
DevID: 1,
Direction: USBIPDirIn,
Endpoint: 1,
},
TransferBufferLength: 8,
})
require.NoError(t, err)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
defer cancel()
_, err = transaction.Wait(ctx)
require.ErrorIs(t, err, context.DeadlineExceeded)
}
// TestUrbTransactionCancelIdempotent exercises the canonical Linux
// cancel wire: after CMD_UNLINK the server replies only with
// RET_UNLINK (status ECONNRESET), never a parallel RET_SUBMIT. The
@@ -76,13 +49,13 @@ func TestUrbTransactionCancelIdempotent(t *testing.T) {
for range callers {
go func() {
defer wg.Done()
err := transaction.Cancel(context.Background())
err := transaction.Cancel()
require.NoError(t, err)
}()
}
wg.Wait()
_, err = transaction.Wait(context.Background())
_, err = transaction.Wait()
require.ErrorIs(t, err, ErrCanceled)
serverDone.Wait()
@@ -123,9 +96,9 @@ func TestUrbTransactionCancelAfterUrbCompleted(t *testing.T) {
})
require.NoError(t, err)
require.NoError(t, transaction.Cancel(context.Background()))
require.NoError(t, transaction.Cancel())
_, err = transaction.Wait(context.Background())
_, err = transaction.Wait()
require.ErrorIs(t, err, ErrCanceled)
serverDone.Wait()
@@ -166,9 +139,9 @@ func TestUrbTransactionCancelWireCarriesDevID(t *testing.T) {
})
require.NoError(t, err)
require.NoError(t, transaction.Cancel(context.Background()))
require.NoError(t, transaction.Cancel())
_, err = transaction.Wait(context.Background())
_, err = transaction.Wait()
require.ErrorIs(t, err, ErrCanceled)
serverDone.Wait()
@@ -200,10 +173,10 @@ func TestUrbTransactionCancelAfterTerminalNoWire(t *testing.T) {
})
require.NoError(t, err)
_, err = transaction.Wait(context.Background())
_, err = transaction.Wait()
require.NoError(t, err)
require.NoError(t, transaction.Cancel(context.Background()))
require.NoError(t, transaction.Cancel())
require.NoError(t, peer.Close())
serverDone.Wait()
service/usbip/usbip_peer_test.go
+5 -42
View File
@@ -109,7 +109,7 @@ func TestUsbIpPeerSubmitRoundTrip(t *testing.T) {
})
require.NoError(t, err)
response, err := transaction.Wait(context.Background())
response, err := transaction.Wait()
require.NoError(t, err)
require.Equal(t, int32(0), response.Status)
require.Equal(t, int32(len(payload)), response.ActualLength)
@@ -142,7 +142,7 @@ func TestUsbIpPeerSessionCloseFailsPending(t *testing.T) {
_ = peer.Close()
}()
_, err = transaction.Wait(context.Background())
_, err = transaction.Wait()
require.ErrorIs(t, err, ErrPeerClosed)
}
@@ -179,51 +179,14 @@ func TestUsbIpPeerCancelMidFlight(t *testing.T) {
go func() {
time.Sleep(20 * time.Millisecond)
require.NoError(t, transaction.Cancel(context.Background()))
require.NoError(t, transaction.Cancel())
}()
_, err = transaction.Wait(context.Background())
_, err = transaction.Wait()
require.ErrorIs(t, err, ErrCanceled)
serverDone.Wait()
}
func TestUsbIpPeerCancelAfterCompletion(t *testing.T) {
peer, server, _ := newPeerPair(t)
var serverDone sync.WaitGroup
serverDone.Add(1)
go func() {
defer serverDone.Done()
submit := server.readSubmit(t)
server.writeSubmitResponse(t, USBIPDirOut, submit.Header.SeqNum, 0, nil, nil)
}()
transaction, err := peer.Submit(SubmitCommand{
Header: DataHeader{
Command: CmdSubmit,
DevID: 1,
Direction: USBIPDirOut,
Endpoint: 1,
},
TransferBufferLength: 4,
Buffer: []byte{9, 9, 9, 9},
})
require.NoError(t, err)
_, err = transaction.Wait(context.Background())
require.NoError(t, err)
serverDone.Wait()
// Cancel after completion must be a no-op; if it tried to write CMD_UNLINK
// the test server would have already exited and the wire would either block
// (net.Pipe drops writes when there's no reader) or fail. The call must not
// hang.
cancelCtx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
err = transaction.Cancel(cancelCtx)
require.NoError(t, err)
}
func TestUsbIpPeerUnknownSeqnumClosesPeer(t *testing.T) {
peer, server, _ := newPeerPair(t)
@@ -308,7 +271,7 @@ func TestUsbIpPeerConcurrentSubmits(t *testing.T) {
TransferBufferLength: 8,
})
require.NoError(t, err)
response, err := transaction.Wait(context.Background())
response, err := transaction.Wait()
require.NoError(t, err)
require.Equal(t, int32(8), response.ActualLength)
require.Equal(t, byte(transaction.SeqNum()), response.Buffer[0])