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

usbip: drop control-extension delta frames in favor of full snapshots

Browse Source 
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.
This commit is contained in:
世界
2026-05-20 16:32:14 +08:00
parent 3c034f99a0
commit 3800783232
7 changed files with 33 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/usbip/client_control.go
-28
View File
@@ -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)
}
service/usbip/client_shared.go
-16
View File
@@ -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)
service/usbip/control_protocol.go
+2 -39
View File
@@ -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 ||
service/usbip/export_ledger.go
+20 -41
View File
@@ -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
service/usbip/export_ledger_darwin_test.go
+8 -9
View File
@@ -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")
}
service/usbip/export_ledger_test.go
-101
View File
@@ -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
}
service/usbip/server.go
+3 -4
View File
@@ -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)