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)