diff --git a/service/usbip/client_control.go b/service/usbip/client_control.go index 508553487..b52bb3a6d 100644 --- a/service/usbip/client_control.go +++ b/service/usbip/client_control.go @@ -2,10 +2,6 @@ package usbip -import ( - "context" -) - func (c *ClientService) applyControlDelta(delta controlDeviceDelta) { c.remoteAccess.Lock() if c.remoteDevicesV2 == nil { @@ -30,27 +26,3 @@ func (c *ClientService) applyControlDelta(delta controlDeviceDelta) { c.remoteAccess.Unlock() c.applyRemoteDeviceState(values) } - -func (c *ClientService) syncRemoteStateAndResetControlState(ctx context.Context) error { - entries, err := c.fetchDevList(ctx) - if err != nil { - return err - } - devices := make(map[string]DeviceInfoV2, len(entries)) - for _, entry := range entries { - device := deviceInfoV2FromEntry(entry, "", "", deviceStateAvailable, 0, deviceStateAvailable) - if device.BusID == "" { - continue - } - devices[device.BusID] = device - } - c.remoteAccess.Lock() - c.remoteDevicesV2 = devices - c.remoteAccess.Unlock() - if !c.assignment.Matched() { - c.applyRemoteExports(entries) - return nil - } - c.applyMatchedExportsWithRetained(entries, nil) - return nil -} diff --git a/service/usbip/client_shared.go b/service/usbip/client_shared.go index b4dd074bc..bc1e5404b 100644 --- a/service/usbip/client_shared.go +++ b/service/usbip/client_shared.go @@ -144,9 +144,8 @@ func (c *ClientService) runControlSession() error { return E.Cause(errControlTransient, "write control preface: ", err) } err = writeControlMessage(conn, controlFrame{ - Type: controlFrameHello, - Version: controlProtocolVersion, - Capabilities: controlCapabilities, + Type: controlFrameHello, + Version: controlProtocolVersion, }, nil) if err != nil { return E.Cause(errControlTransient, "write control hello: ", err) @@ -174,20 +173,9 @@ func (c *ClientService) runControlSession() error { if ack.Version != controlProtocolVersion { return E.Cause(errControlUnsupported, "unsupported control version ", ack.Version) } - if ack.Capabilities&controlRequiredCapabilities != controlRequiredCapabilities { - return E.Cause(errControlUnsupported, "missing control capabilities 0x", ack.Capabilities) - } _ = conn.SetWriteDeadline(time.Time{}) _ = conn.SetReadDeadline(time.Time{}) - extended := supportsControlExtensions(ack.Capabilities) - if !extended { - err = c.syncRemoteStateContext(c.ctx) - if err != nil { - return E.Cause(err, "initial devlist sync") - } - } - pingDone := make(chan struct{}) go c.controlPingLoop(conn, pingDone) defer close(pingDone) @@ -206,23 +194,7 @@ func (c *ClientService) runControlSession() error { } frame := message.Frame switch frame.Type { - case controlFrameChanged: - if frame.Sequence != lastSeq && frame.Sequence != lastSeq+1 { - return E.Cause(errImmediateReconnect, "control sequence jumped from ", lastSeq, " to ", frame.Sequence) - } - lastSeq = frame.Sequence - if extended { - err = c.syncRemoteStateAndResetControlState(c.ctx) - } else { - err = c.syncRemoteStateContext(c.ctx) - } - if err != nil { - return E.Cause(errImmediateReconnect, "devlist sync after change ", frame.Sequence, ": ", err) - } case controlFrameDeviceSnapshot: - if !extended { - return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type) - } var snapshot controlDeviceSnapshot err = unmarshalControlPayload(message.Payload, &snapshot) if err != nil { @@ -236,16 +208,11 @@ func (c *ClientService) runControlSession() error { c.remoteAccess.Unlock() c.applyRemoteDeviceState(values) case controlFrameDeviceDelta: - if !extended { - return E.Cause(errImmediateReconnect, "unexpected control frame ", frame.Type) - } if frame.Sequence != lastSeq+1 { - err = c.syncRemoteStateAndResetControlState(c.ctx) - if err != nil { - return E.Cause(errImmediateReconnect, "devlist sync after sequence jump ", frame.Sequence, ": ", err) - } - lastSeq = frame.Sequence - continue + 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) diff --git a/service/usbip/control_protocol.go b/service/usbip/control_protocol.go index 43dd19bcb..8a7035d4e 100644 --- a/service/usbip/control_protocol.go +++ b/service/usbip/control_protocol.go @@ -17,22 +17,13 @@ const ( controlFrameHello uint8 = 1 controlFrameAck uint8 = 2 - controlFrameChanged uint8 = 3 controlFramePing uint8 = 4 controlFramePong uint8 = 5 controlFrameDeviceSnapshot uint8 = 6 controlFrameDeviceDelta uint8 = 7 - controlCapabilityChanged uint32 = 1 << 0 - controlCapabilityPingPong uint32 = 1 << 1 - controlCapabilityPayloadFrames uint32 = 1 << 2 - controlCapabilityDeviceStateV2 uint32 = 1 << 3 - controlRequiredCapabilities = controlCapabilityChanged | controlCapabilityPingPong - controlExtensionCapabilities = controlCapabilityPayloadFrames | controlCapabilityDeviceStateV2 - controlCapabilities = controlRequiredCapabilities | controlExtensionCapabilities - controlPrefaceSize = 8 - controlFrameSize = 16 + controlFrameSize = 12 maxControlPayloadLength = 64<<10 - 1 deviceStateAvailable = "available" @@ -50,7 +41,6 @@ type controlFrame struct { Type uint8 Version uint8 PayloadLength uint16 - Capabilities uint32 Sequence uint64 } @@ -115,8 +105,7 @@ func (cr *controlReader) read(r io.Reader) (controlMessage, error) { Type: raw[0], Version: raw[1], PayloadLength: binary.BigEndian.Uint16(raw[2:4]), - Capabilities: binary.BigEndian.Uint32(raw[4:8]), - Sequence: binary.BigEndian.Uint64(raw[8:16]), + Sequence: binary.BigEndian.Uint64(raw[4:12]), } var payload []byte if frame.PayloadLength > 0 { @@ -145,8 +134,7 @@ 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.PutUint32(raw[4:8], frame.Capabilities) - binary.BigEndian.PutUint64(raw[8:16], frame.Sequence) + binary.BigEndian.PutUint64(raw[4:12], frame.Sequence) _, err = w.Write(raw[:]) if err != nil { return err @@ -176,10 +164,6 @@ func unmarshalControlPayload(payload []byte, value any) error { return json.Unmarshal(payload, value) } -func supportsControlExtensions(capabilities uint32) bool { - return capabilities&controlExtensionCapabilities == controlExtensionCapabilities -} - func deviceInfoV2FromEntry(entry DeviceEntry, backend string, stableID string, state string, statusCode int, statusReason string) DeviceInfoV2 { interfaces := make([]DeviceInterfaceV2, len(entry.Interfaces)) for i := range entry.Interfaces { diff --git a/service/usbip/darwin_integration_test.go b/service/usbip/darwin_integration_test.go index 2480ed389..9f038a00b 100644 --- a/service/usbip/darwin_integration_test.go +++ b/service/usbip/darwin_integration_test.go @@ -268,23 +268,19 @@ func (s *darwinFakeUSBIPServer) handleControlConn(conn net.Conn) { if hello.Type != controlFrameHello || hello.Version != controlProtocolVersion { return } - capabilities := hello.Capabilities & controlCapabilities err = writeControlMessage(conn, controlFrame{ - Type: controlFrameAck, - Version: controlProtocolVersion, - Capabilities: capabilities, + Type: controlFrameAck, + Version: controlProtocolVersion, }, nil) if err != nil { return } - if supportsControlExtensions(capabilities) { - _ = writeControlMessage(conn, controlFrame{ - Type: controlFrameDeviceSnapshot, - Version: controlProtocolVersion, - }, controlDeviceSnapshot{ - Devices: []DeviceInfoV2{deviceInfoV2FromEntry(s.entry, "darwin-fake", "darwin-fake:"+s.entry.Info.BusIDString(), deviceStateAvailable, 0, "available")}, - }) - } + _ = writeControlMessage(conn, controlFrame{ + Type: controlFrameDeviceSnapshot, + Version: controlProtocolVersion, + }, controlDeviceSnapshot{ + Devices: []DeviceInfoV2{deviceInfoV2FromEntry(s.entry, "darwin-fake", "darwin-fake:"+s.entry.Info.BusIDString(), deviceStateAvailable, 0, "available")}, + }) for { message, err := cr.read(conn) if err != nil { diff --git a/service/usbip/export_ledger.go b/service/usbip/export_ledger.go index 0712e7b69..4ab2fdc26 100644 --- a/service/usbip/export_ledger.go +++ b/service/usbip/export_ledger.go @@ -31,10 +31,9 @@ type exportLedger struct { } type exportSubscriber struct { - id uint64 - capabilities uint32 - conn net.Conn - send chan controlMessage + id uint64 + conn net.Conn + send chan controlMessage } const controlSubscriberSendBuffer = 16 @@ -141,21 +140,12 @@ func (l *exportLedger) BroadcastIfChanged() bool { } l.broadcastAccess.Unlock() - frame := controlFrame{ - Type: controlFrameChanged, - Version: controlProtocolVersion, - Sequence: sequence, - } for _, sub := range targets { - if supportsControlExtensions(sub.capabilities) { - l.enqueuePayload(sub, controlFrame{ - Type: controlFrameDeviceDelta, - Version: controlProtocolVersion, - Sequence: sequence, - }, delta, frame) - continue - } - l.enqueueFrame(sub, frame) + l.enqueuePayload(sub, controlFrame{ + Type: controlFrameDeviceDelta, + Version: controlProtocolVersion, + Sequence: sequence, + }, delta) } return true } @@ -200,53 +190,39 @@ func (l *exportLedger) ReleaseImport(busid string, removeExport bool) { }) } -// Subscribe enqueues a freshly computed snapshot to extension-capable -// subscribers so they see 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, capabilities uint32) (*exportSubscriber, uint64) { - extended := supportsControlExtensions(capabilities) +// 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 - if extended { - // 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() - } - } else { + // 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() } defer l.broadcastAccess.Unlock() l.nextSubID++ sub := &exportSubscriber{ - id: l.nextSubID, - capabilities: capabilities, - conn: conn, - send: make(chan controlMessage, controlSubscriberSendBuffer), - } - if extended { - l.enqueuePayload(sub, controlFrame{ - Type: controlFrameDeviceSnapshot, - Version: controlProtocolVersion, - Sequence: sequence, - }, controlDeviceSnapshot{Sequence: sequence, Devices: snapshot}, controlFrame{ - Type: controlFrameChanged, - Version: controlProtocolVersion, - Sequence: sequence, - }) + id: l.nextSubID, + conn: conn, + send: make(chan controlMessage, controlSubscriberSendBuffer), } + l.enqueuePayload(sub, controlFrame{ + Type: controlFrameDeviceSnapshot, + Version: controlProtocolVersion, + Sequence: sequence, + }, controlDeviceSnapshot{Sequence: sequence, Devices: snapshot}) l.subs[sub.id] = sub return sub, sequence } @@ -317,10 +293,11 @@ func (l *exportLedger) enqueueFrame(sub *exportSubscriber, frame controlFrame) { } } -func (l *exportLedger) enqueuePayload(sub *exportSubscriber, frame controlFrame, payload any, fallback controlFrame) { +func (l *exportLedger) enqueuePayload(sub *exportSubscriber, frame controlFrame, payload any) { rawPayload, err := marshalControlPayload(payload) if err != nil || len(rawPayload) > maxControlPayloadLength { - l.enqueueFrame(sub, fallback) + l.logger.Debug("control subscriber ", sub.id, " payload encode failed; closing") + _ = sub.conn.Close() return } select { diff --git a/service/usbip/export_ledger_darwin_test.go b/service/usbip/export_ledger_darwin_test.go index 423a37e02..5cea350e9 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, controlCapabilities) + sub, _ := ledger.Subscribe(nil) select { case <-sub.send: case <-time.After(time.Second): diff --git a/service/usbip/export_ledger_test.go b/service/usbip/export_ledger_test.go index a210885aa..18379922d 100644 --- a/service/usbip/export_ledger_test.go +++ b/service/usbip/export_ledger_test.go @@ -24,7 +24,7 @@ func TestSubscribeRetriesSnapshotWhenSequenceAdvances(t *testing.T) { } } - sub, sequence := ledger.Subscribe(nil, controlCapabilities) + sub, sequence := ledger.Subscribe(nil) if sequence != 1 { t.Fatalf("expected subscription sequence 1, got %d", sequence) } diff --git a/service/usbip/server.go b/service/usbip/server.go index 241656795..6492549d1 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -222,18 +222,12 @@ func (s *ServerService) handleControlConn(conn net.Conn) { s.logger.Debug("unsupported control version ", hello.Version) return } - if hello.Capabilities&controlRequiredCapabilities != controlRequiredCapabilities { - s.logger.Debug("missing control capabilities 0x", hello.Capabilities) - return - } - capabilities := hello.Capabilities & controlCapabilities - sub, seq := s.ledger.Subscribe(conn, capabilities) + sub, seq := s.ledger.Subscribe(conn) defer s.ledger.Unsubscribe(sub) err = writeControlMessage(conn, controlFrame{ - Type: controlFrameAck, - Version: controlProtocolVersion, - Capabilities: capabilities, - Sequence: seq, + Type: controlFrameAck, + Version: controlProtocolVersion, + Sequence: seq, }, nil) if err != nil { s.logger.Debug("write control ack: ", err)