From 38007832320a3631cdb12efb4b44c65ea0be7189 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Wed, 20 May 2026 16:32:14 +0800 Subject: [PATCH] usbip: drop control-extension delta frames in favor of full snapshots The control extension carried device-state updates in two wire forms: a full snapshot on subscribe and Added/Updated/Removed deltas thereafter. At realistic USB device counts the delta path saves no measurable bandwidth and matches the snapshot's hotplug latency on the same socket, so the bookkeeping (sequence field on every frame, lastSeq+1 jump check, Subscribe sequence-stability retry loop, applyControlDelta) was pure ceremony. Every broadcast now emits controlFrameDeviceSnapshot with the full device list; clients overwrite their remote map from each snapshot. controlFrame loses its Sequence field, shrinking the wire header from 12 to 4 bytes. BroadcastIfChanged uses maps.EqualFunc with the existing deviceInfoV2Equal to skip no-op broadcasts. --- service/usbip/client_control.go | 28 ------ service/usbip/client_shared.go | 16 ---- service/usbip/control_protocol.go | 41 +-------- service/usbip/export_ledger.go | 61 ++++--------- service/usbip/export_ledger_darwin_test.go | 17 ++-- service/usbip/export_ledger_test.go | 101 --------------------- service/usbip/server.go | 7 +- 7 files changed, 33 insertions(+), 238 deletions(-) delete mode 100644 service/usbip/client_control.go delete mode 100644 service/usbip/export_ledger_test.go diff --git a/service/usbip/client_control.go b/service/usbip/client_control.go deleted file mode 100644 index b52bb3a6d..000000000 --- a/service/usbip/client_control.go +++ /dev/null @@ -1,28 +0,0 @@ -//go:build linux || (darwin && cgo) || windows - -package usbip - -func (c *ClientService) applyControlDelta(delta controlDeviceDelta) { - c.remoteAccess.Lock() - if c.remoteDevicesV2 == nil { - c.remoteDevicesV2 = make(map[string]DeviceInfoV2) - } - for _, busid := range delta.Removed { - delete(c.remoteDevicesV2, busid) - } - for _, device := range delta.Added { - if device.BusID == "" { - continue - } - c.remoteDevicesV2[device.BusID] = device - } - for _, device := range delta.Updated { - if device.BusID == "" { - continue - } - c.remoteDevicesV2[device.BusID] = device - } - values := sortedDeviceInfoV2Values(c.remoteDevicesV2) - c.remoteAccess.Unlock() - c.applyRemoteDeviceState(values) -} diff --git a/service/usbip/client_shared.go b/service/usbip/client_shared.go index bc1e5404b..80604bbec 100644 --- a/service/usbip/client_shared.go +++ b/service/usbip/client_shared.go @@ -180,7 +180,6 @@ func (c *ClientService) runControlSession() error { go c.controlPingLoop(conn, pingDone) defer close(pingDone) - lastSeq := ack.Sequence var reader controlReader for { err = conn.SetReadDeadline(time.Now().Add(controlReadTimeout)) @@ -200,27 +199,12 @@ func (c *ClientService) runControlSession() error { if err != nil { return E.Cause(errImmediateReconnect, "read device snapshot: ", err) } - lastSeq = frame.Sequence devices := deviceInfoV2Map(snapshot.Devices) values := sortedDeviceInfoV2Values(devices) c.remoteAccess.Lock() c.remoteDevicesV2 = devices c.remoteAccess.Unlock() c.applyRemoteDeviceState(values) - case controlFrameDeviceDelta: - if frame.Sequence != lastSeq+1 { - c.remoteAccess.Lock() - c.remoteDevicesV2 = nil - c.remoteAccess.Unlock() - return E.Cause(errImmediateReconnect, "control sequence jumped from ", lastSeq, " to ", frame.Sequence) - } - var delta controlDeviceDelta - err = unmarshalControlPayload(message.Payload, &delta) - if err != nil { - return E.Cause(errImmediateReconnect, "read device delta: ", err) - } - lastSeq = frame.Sequence - c.applyControlDelta(delta) case controlFramePong: default: return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type) diff --git a/service/usbip/control_protocol.go b/service/usbip/control_protocol.go index 8a7035d4e..4e798ba4a 100644 --- a/service/usbip/control_protocol.go +++ b/service/usbip/control_protocol.go @@ -7,7 +7,6 @@ import ( "encoding/json" "io" "slices" - "strings" E "github.com/sagernet/sing/common/exceptions" ) @@ -20,10 +19,9 @@ const ( controlFramePing uint8 = 4 controlFramePong uint8 = 5 controlFrameDeviceSnapshot uint8 = 6 - controlFrameDeviceDelta uint8 = 7 controlPrefaceSize = 8 - controlFrameSize = 12 + controlFrameSize = 4 maxControlPayloadLength = 64<<10 - 1 deviceStateAvailable = "available" @@ -41,7 +39,6 @@ type controlFrame struct { Type uint8 Version uint8 PayloadLength uint16 - Sequence uint64 } type controlMessage struct { @@ -78,15 +75,7 @@ type DeviceInfoV2 struct { } type controlDeviceSnapshot struct { - Sequence uint64 `json:"sequence"` - Devices []DeviceInfoV2 `json:"devices"` -} - -type controlDeviceDelta struct { - Sequence uint64 `json:"sequence"` - Added []DeviceInfoV2 `json:"added,omitempty"` - Updated []DeviceInfoV2 `json:"updated,omitempty"` - Removed []string `json:"removed,omitempty"` + Devices []DeviceInfoV2 `json:"devices"` } // controlReader reuses its payload scratch across successive reads on a @@ -105,7 +94,6 @@ func (cr *controlReader) read(r io.Reader) (controlMessage, error) { Type: raw[0], Version: raw[1], PayloadLength: binary.BigEndian.Uint16(raw[2:4]), - Sequence: binary.BigEndian.Uint64(raw[4:12]), } var payload []byte if frame.PayloadLength > 0 { @@ -134,7 +122,6 @@ func writeControlMessage(w io.Writer, frame controlFrame, payload any) error { raw[0] = frame.Type raw[1] = frame.Version binary.BigEndian.PutUint16(raw[2:4], frame.PayloadLength) - binary.BigEndian.PutUint64(raw[4:12], frame.Sequence) _, err = w.Write(raw[:]) if err != nil { return err @@ -259,30 +246,6 @@ func deviceInfoV2ToEntries(devices []DeviceInfoV2, availableOnly bool) []DeviceE return entries } -func buildControlDeviceDelta(sequence uint64, previous map[string]DeviceInfoV2, current map[string]DeviceInfoV2) controlDeviceDelta { - delta := controlDeviceDelta{Sequence: sequence} - for busid, device := range current { - prev, ok := previous[busid] - if !ok { - delta.Added = append(delta.Added, device) - continue - } - if !deviceInfoV2Equal(prev, device) { - delta.Updated = append(delta.Updated, device) - } - } - for busid := range previous { - _, ok := current[busid] - if !ok { - delta.Removed = append(delta.Removed, busid) - } - } - slices.SortFunc(delta.Added, func(a, b DeviceInfoV2) int { return strings.Compare(a.BusID, b.BusID) }) - slices.SortFunc(delta.Updated, func(a, b DeviceInfoV2) int { return strings.Compare(a.BusID, b.BusID) }) - slices.Sort(delta.Removed) - return delta -} - func deviceInfoV2Equal(a, b DeviceInfoV2) bool { if a.BusID != b.BusID || a.StableID != b.StableID || diff --git a/service/usbip/export_ledger.go b/service/usbip/export_ledger.go index 4ab2fdc26..665e0bb54 100644 --- a/service/usbip/export_ledger.go +++ b/service/usbip/export_ledger.go @@ -3,6 +3,7 @@ package usbip import ( + "maps" "net" "slices" "strings" @@ -12,15 +13,15 @@ import ( "github.com/sagernet/sing-box/log" ) -// exportLedger holds two mutexes that are never acquired together. The -// inventory lock must be released before BroadcastIfChanged re-takes it -// through snapshotDeviceState. +// exportLedger holds two mutexes that are never acquired together. +// BroadcastIfChanged computes the next state under inventoryAccess via +// snapshotDeviceState, then takes broadcastAccess to compare against +// l.state, swap it in, and snapshot the subscriber list. type exportLedger struct { logger log.ContextLogger now func() time.Time broadcastAccess sync.Mutex - seq uint64 nextSubID uint64 subs map[uint64]*exportSubscriber state map[string]DeviceInfoV2 @@ -124,28 +125,22 @@ func (l *exportLedger) BroadcastIfChanged() bool { nextState := deviceInfoV2Map(l.snapshotDeviceState()) l.broadcastAccess.Lock() - nextSequence := l.seq + 1 - delta := buildControlDeviceDelta(nextSequence, l.state, nextState) - if len(delta.Added) == 0 && len(delta.Updated) == 0 && len(delta.Removed) == 0 { - l.state = nextState + if maps.EqualFunc(l.state, nextState, deviceInfoV2Equal) { l.broadcastAccess.Unlock() return false } - l.seq = nextSequence - sequence := l.seq l.state = nextState + devices := sortedDeviceInfoV2Values(nextState) targets := make([]*exportSubscriber, 0, len(l.subs)) for _, sub := range l.subs { targets = append(targets, sub) } l.broadcastAccess.Unlock() + frame := controlFrame{Type: controlFrameDeviceSnapshot, Version: controlProtocolVersion} + payload := controlDeviceSnapshot{Devices: devices} for _, sub := range targets { - l.enqueuePayload(sub, controlFrame{ - Type: controlFrameDeviceDelta, - Version: controlProtocolVersion, - Sequence: sequence, - }, delta) + l.enqueuePayload(sub, frame, payload) } return true } @@ -190,27 +185,12 @@ func (l *exportLedger) ReleaseImport(busid string, removeExport bool) { }) } -// Subscribe enqueues a freshly computed snapshot so the new subscriber -// sees current state regardless of when the last broadcast fired. Does -// NOT mutate l.state: other subscribers must still receive the next -// BroadcastIfChanged delta against the previous baseline. -func (l *exportLedger) Subscribe(conn net.Conn) (*exportSubscriber, uint64) { - var snapshot []DeviceInfoV2 - var sequence uint64 - // Keep the snapshot and sequence from the same stable generation. - for { - l.broadcastAccess.Lock() - sequence = l.seq - l.broadcastAccess.Unlock() - - snapshot = l.snapshotDeviceState() - - l.broadcastAccess.Lock() - if sequence == l.seq { - break - } - l.broadcastAccess.Unlock() - } +// Subscribe enqueues the current broadcast state so the new subscriber +// sees the same device list every existing subscriber has received. +// l.state is maintained by SeedBroadcastState and BroadcastIfChanged +// under broadcastAccess, so reading it here is race-free. +func (l *exportLedger) Subscribe(conn net.Conn) *exportSubscriber { + l.broadcastAccess.Lock() defer l.broadcastAccess.Unlock() l.nextSubID++ sub := &exportSubscriber{ @@ -219,12 +199,11 @@ func (l *exportLedger) Subscribe(conn net.Conn) (*exportSubscriber, uint64) { send: make(chan controlMessage, controlSubscriberSendBuffer), } l.enqueuePayload(sub, controlFrame{ - Type: controlFrameDeviceSnapshot, - Version: controlProtocolVersion, - Sequence: sequence, - }, controlDeviceSnapshot{Sequence: sequence, Devices: snapshot}) + Type: controlFrameDeviceSnapshot, + Version: controlProtocolVersion, + }, controlDeviceSnapshot{Devices: sortedDeviceInfoV2Values(l.state)}) l.subs[sub.id] = sub - return sub, sequence + return sub } // Unsubscribe leaves the subscriber's send channel for the GC to diff --git a/service/usbip/export_ledger_darwin_test.go b/service/usbip/export_ledger_darwin_test.go index 5cea350e9..d986cadd0 100644 --- a/service/usbip/export_ledger_darwin_test.go +++ b/service/usbip/export_ledger_darwin_test.go @@ -23,7 +23,7 @@ func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) { ledger.ApplyHostSnapshot(map[string]Export{export.busid: export}, nil) ledger.SeedBroadcastState() - sub, _ := ledger.Subscribe(nil) + sub := ledger.Subscribe(nil) select { case <-sub.send: case <-time.After(time.Second): @@ -39,15 +39,14 @@ func TestDarwinStaleExportBroadcastsUnavailableUpdate(t *testing.T) { select { case message := <-sub.send: - require.Equal(t, controlFrameDeviceDelta, message.Frame.Type) + require.Equal(t, controlFrameDeviceSnapshot, message.Frame.Type) - var delta controlDeviceDelta - require.NoError(t, unmarshalControlPayload(message.Payload, &delta)) - require.Empty(t, delta.Removed) - require.Len(t, delta.Updated, 1) - require.Equal(t, export.busid, delta.Updated[0].BusID) - require.Equal(t, deviceStateUnavailable, delta.Updated[0].State) - require.Equal(t, "device replaced", delta.Updated[0].StatusReason) + var snapshot controlDeviceSnapshot + require.NoError(t, unmarshalControlPayload(message.Payload, &snapshot)) + require.Len(t, snapshot.Devices, 1) + require.Equal(t, export.busid, snapshot.Devices[0].BusID) + require.Equal(t, deviceStateUnavailable, snapshot.Devices[0].State) + require.Equal(t, "device replaced", snapshot.Devices[0].StatusReason) case <-time.After(time.Second): t.Fatal("timed out waiting for unavailable update") } diff --git a/service/usbip/export_ledger_test.go b/service/usbip/export_ledger_test.go deleted file mode 100644 index 18379922d..000000000 --- a/service/usbip/export_ledger_test.go +++ /dev/null @@ -1,101 +0,0 @@ -//go:build linux || (darwin && cgo) || windows - -package usbip - -import ( - "context" - "net" - "testing" - "time" -) - -func TestSubscribeRetriesSnapshotWhenSequenceAdvances(t *testing.T) { - ledger := newExportLedger(nil, 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() - - oldExport.onSnapshot = func() { - ledger.ApplyHostSnapshot(map[string]Export{newExport.busid: newExport}, nil) - if !ledger.BroadcastIfChanged() { - t.Fatal("expected broadcast after replacing export") - } - } - - sub, sequence := ledger.Subscribe(nil) - if sequence != 1 { - t.Fatalf("expected subscription sequence 1, got %d", sequence) - } - - select { - case message := <-sub.send: - if message.Frame.Type != controlFrameDeviceSnapshot { - t.Fatalf("expected device snapshot frame, got %d", message.Frame.Type) - } - if message.Frame.Sequence != sequence { - t.Fatalf("expected frame sequence %d, got %d", sequence, message.Frame.Sequence) - } - var snapshot controlDeviceSnapshot - err := unmarshalControlPayload(message.Payload, &snapshot) - if err != nil { - t.Fatal(err) - } - if snapshot.Sequence != sequence { - t.Fatalf("expected payload sequence %d, got %d", sequence, snapshot.Sequence) - } - if len(snapshot.Devices) != 1 || snapshot.Devices[0].BusID != newExport.busid { - t.Fatalf("expected fresh snapshot for %s, got %#v", newExport.busid, snapshot.Devices) - } - default: - t.Fatal("expected queued device snapshot") - } -} - -type testExport struct { - busid string - vendorID uint16 - productID uint16 - - onSnapshot func() -} - -func (e *testExport) BusID() string { - return e.busid -} - -func (e *testExport) Snapshot(busy bool) ExportSnapshot { - onSnapshot := e.onSnapshot - e.onSnapshot = nil - if onSnapshot != nil { - onSnapshot() - } - state := deviceStateAvailable - if busy { - state = deviceStateBusy - } - return ExportSnapshot{ - Entry: DeviceEntry{ - Info: e.deviceInfo(), - }, - State: state, - } -} - -func (e *testExport) DeviceInfo() (DeviceInfoTruncated, error) { - return e.deviceInfo(), nil -} - -func (e *testExport) NewServerDataSession(ctx context.Context, conn net.Conn) (DataSession, error) { - return nil, nil -} - -func (e *testExport) deviceInfo() DeviceInfoTruncated { - var info DeviceInfoTruncated - copy(info.BusID[:], e.busid) - info.IDVendor = e.vendorID - info.IDProduct = e.productID - info.Speed = 2 - return info -} diff --git a/service/usbip/server.go b/service/usbip/server.go index 6492549d1..c67f73529 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -222,12 +222,11 @@ func (s *ServerService) handleControlConn(conn net.Conn) { s.logger.Debug("unsupported control version ", hello.Version) return } - sub, seq := s.ledger.Subscribe(conn) + sub := s.ledger.Subscribe(conn) defer s.ledger.Unsubscribe(sub) err = writeControlMessage(conn, controlFrame{ - Type: controlFrameAck, - Version: controlProtocolVersion, - Sequence: seq, + Type: controlFrameAck, + Version: controlProtocolVersion, }, nil) if err != nil { s.logger.Debug("write control ack: ", err)