diff --git a/service/usbip/client_shared.go b/service/usbip/client_shared.go index c3ecba0dc..715c283c4 100644 --- a/service/usbip/client_shared.go +++ b/service/usbip/client_shared.go @@ -6,6 +6,7 @@ import ( "context" "errors" "fmt" + "io" "slices" "time" @@ -15,17 +16,21 @@ import ( ) const ( - clientReconnectDelay = 5 * time.Second - clientShutdownTimeout = 15 * time.Second - controlPingInterval = 10 * time.Second - controlReadTimeout = 30 * time.Second - controlWriteTimeout = 5 * time.Second - controlSessionIdleHint = "control session lost" + clientReconnectDelay = 5 * time.Second + clientShutdownTimeout = 15 * time.Second + controlPingInterval = 10 * time.Second + controlReadTimeout = 30 * time.Second + controlWriteTimeout = 5 * time.Second + controlSessionIdleHint = "control session lost" + controlHandshakeBackoffStart = time.Second + controlHandshakeBackoffMax = 30 * time.Second + controlHandshakeMaxTransient = 3 ) var ( errImmediateReconnect = E.New("usbip control reconnect") errControlUnsupported = E.New("usbip control unsupported") + errControlTransient = E.New("usbip control transient") ) type clientAssignedWorker struct { @@ -57,31 +62,79 @@ func (c *ClientService) initializeWorkers() { func (c *ClientService) run() { defer c.wg.Done() - for immediate := true; immediate || sleepCtx(c.ctx, clientReconnectDelay); { - err := c.runControlSession() - if errors.Is(err, errControlUnsupported) { - c.logger.Info("control channel unsupported by ", c.serverAddr, "; using standard usbip mode") - for { - err = c.syncRemoteStateContext(c.ctx) - if err != nil { - err = E.Cause(err, "devlist sync") - break - } - if !sleepCtx(c.ctx, clientReconnectDelay) { - err = nil - break + defer c.stopAllWorkers() + + var transientStreak int + backoff := controlHandshakeBackoffStart + immediate := true + + for { + if !immediate { + delay := clientReconnectDelay + if transientStreak > 0 { + delay = backoff + backoff *= 2 + if backoff > controlHandshakeBackoffMax { + backoff = controlHandshakeBackoffMax } } + if !sleepCtx(c.ctx, delay) { + return + } } + immediate = false + + err := c.runControlSession() if c.ctx.Err() != nil { - break + return } + + if errors.Is(err, errControlUnsupported) { + c.logger.Info("control channel unsupported by ", c.serverAddr, "; using standard usbip mode") + c.runStandardPollLoop() + if c.ctx.Err() != nil { + return + } + transientStreak = 0 + backoff = controlHandshakeBackoffStart + continue + } + + if errors.Is(err, errControlTransient) { + transientStreak++ + c.logger.Warn("control handshake ", c.serverAddr, ": ", err) + if transientStreak >= controlHandshakeMaxTransient { + c.logger.Info("control handshake failed ", transientStreak, " times against ", c.serverAddr, "; using standard usbip mode") + c.runStandardPollLoop() + if c.ctx.Err() != nil { + return + } + transientStreak = 0 + backoff = controlHandshakeBackoffStart + } + continue + } + if err != nil { c.logger.Error("control ", c.serverAddr, ": ", err) } + transientStreak = 0 + backoff = controlHandshakeBackoffStart immediate = errors.Is(err, errImmediateReconnect) } - c.stopAllWorkers() +} + +func (c *ClientService) runStandardPollLoop() { + for { + err := c.syncRemoteStateContext(c.ctx) + if err != nil { + c.logger.Error("control ", c.serverAddr, ": ", E.Cause(err, "devlist sync")) + return + } + if !sleepCtx(c.ctx, clientReconnectDelay) { + return + } + } } func (c *ClientService) runControlSession() error { @@ -97,7 +150,7 @@ func (c *ClientService) runControlSession() error { _ = conn.SetReadDeadline(time.Now().Add(controlWriteTimeout)) _, err = conn.Write(controlPreface[:]) if err != nil { - return E.Cause(errControlUnsupported, "write control preface: ", err) + return E.Cause(errControlTransient, "write control preface: ", err) } err = writeControlMessage(conn, controlFrame{ Type: controlFrameHello, @@ -105,12 +158,19 @@ func (c *ClientService) runControlSession() error { Capabilities: controlCapabilities, }, nil) if err != nil { - return E.Cause(errControlUnsupported, "write control hello: ", err) + return E.Cause(errControlTransient, "write control hello: ", err) } var cr controlReader ackMessage, err := cr.read(conn) if err != nil { - return E.Cause(errControlUnsupported, "read control ack: ", err) + // A plain usbipd reads our preface as an op-header, finds a bogus + // version, and closes cleanly: the client sees io.EOF. Other I/O + // errors (timeout, RST, partial read) point at a transient + // network problem, not "server lacks CONTROL". + if errors.Is(err, io.EOF) { + return E.Cause(errControlUnsupported, "read control ack: ", err) + } + return E.Cause(errControlTransient, "read control ack: ", err) } if len(ackMessage.Payload) > 0 { return E.Cause(errControlUnsupported, "unexpected control ack payload length ", len(ackMessage.Payload)) diff --git a/service/usbip/control_protocol.go b/service/usbip/control_protocol.go index 22ffff2d9..c72e14ea9 100644 --- a/service/usbip/control_protocol.go +++ b/service/usbip/control_protocol.go @@ -278,7 +278,7 @@ func deviceInfoV2ToEntries(devices []DeviceInfoV2, availableOnly bool) []DeviceE continue } var info DeviceInfoTruncated - encodePathField(&info.Path, device.Path) + encodePathField(&info.Path, device.Path, device.Serial) copy(info.BusID[:], device.BusID) info.Speed = device.Speed info.IDVendor = device.VendorID diff --git a/service/usbip/darwin_integration_test.go b/service/usbip/darwin_integration_test.go index 016b72e5b..b829df41a 100644 --- a/service/usbip/darwin_integration_test.go +++ b/service/usbip/darwin_integration_test.go @@ -484,7 +484,7 @@ func truncateDarwinFakeDescriptor(data []byte, length int) []byte { func darwinFakeDeviceEntry() DeviceEntry { var info DeviceInfoTruncated - encodePathField(&info.Path, "fake-darwin-usbip") + encodePathField(&info.Path, "fake-darwin-usbip", "codex-usbip-fake") copy(info.BusID[:], darwinFakeBusID) info.BusNum = 1 info.DevNum = 1 diff --git a/service/usbip/host_darwin.go b/service/usbip/host_darwin.go index 2f17c92a8..2a6b95770 100644 --- a/service/usbip/host_darwin.go +++ b/service/usbip/host_darwin.go @@ -127,7 +127,7 @@ func (h *darwinExportHost) Reconcile(ctx context.Context, isBusy func(busid stri var ( toAdd []*darwinExport toRemove []*darwinExport - toStale []string + toStale []darwinStaleMark released []string ) for busid, info := range desired { @@ -136,7 +136,7 @@ func (h *darwinExportHost) Reconcile(ctx context.Context, isBusy func(busid stri } if exp, ok := current[busid]; ok { if isBusy(busid) { - toStale = append(toStale, busid) + toStale = append(toStale, darwinStaleMark{busid: busid, pendingRegistryID: info.registryID}) continue } toRemove = append(toRemove, exp) @@ -162,7 +162,7 @@ func (h *darwinExportHost) Reconcile(ctx context.Context, isBusy func(busid stri continue } if isBusy(busid) { - toStale = append(toStale, busid) + toStale = append(toStale, darwinStaleMark{busid: busid}) continue } toRemove = append(toRemove, exp) @@ -171,12 +171,15 @@ func (h *darwinExportHost) Reconcile(ctx context.Context, isBusy func(busid stri } h.access.Lock() - for _, busid := range toStale { - exp, ok := h.exports[busid] - if !ok || exp.stale { + for _, mark := range toStale { + exp, ok := h.exports[mark.busid] + if !ok { continue } exp.stale = true + if mark.pendingRegistryID != 0 { + exp.pendingRegistryID = mark.pendingRegistryID + } } for _, exp := range toRemove { delete(h.exports, exp.busid) @@ -208,12 +211,44 @@ func (h *darwinExportHost) FinishImport(ctx context.Context, busid string) (bool h.access.Unlock() return false, nil } - delete(h.exports, busid) + pending := exp.pendingRegistryID + if pending == 0 { + delete(h.exports, busid) + h.access.Unlock() + if exp.device != nil { + exp.device.Close() + } + return true, nil + } + h.access.Unlock() + + device, err := darwinOpenUSBHostDevice(pending, true) + if err != nil { + h.logger.Warn("re-capture ", busid, " (registry ", pending, "): ", err) + h.access.Lock() + delete(h.exports, busid) + h.access.Unlock() + if exp.device != nil { + exp.device.Close() + } + return true, nil + } + info := device.info + replacement := &darwinExport{ + busid: info.key.BusID, + registryID: info.registryID, + device: device, + entry: info.entry, + logger: h.logger, + } + h.access.Lock() + h.exports[busid] = replacement h.access.Unlock() if exp.device != nil { exp.device.Close() } - return true, nil + h.logger.Info("re-exported ", busid, " through IOUSBHost re-capture (registry ", pending, ")") + return false, nil } func (h *darwinExportHost) snapshotSelf() map[string]Export { @@ -230,12 +265,18 @@ func (h *darwinExportHost) snapshotSelf() map[string]Export { } type darwinExport struct { - busid string - registryID uint64 - device *darwinUSBHostDevice - entry DeviceEntry - logger log.ContextLogger - stale bool + busid string + registryID uint64 + pendingRegistryID uint64 + device *darwinUSBHostDevice + entry DeviceEntry + logger log.ContextLogger + stale bool +} + +type darwinStaleMark struct { + busid string + pendingRegistryID uint64 } func (e *darwinExport) BusID() string { diff --git a/service/usbip/protocol.go b/service/usbip/protocol.go index 3284a8f6c..ce63ce697 100644 --- a/service/usbip/protocol.go +++ b/service/usbip/protocol.go @@ -222,6 +222,7 @@ func ReadOpRepDevListBody(r io.Reader) ([]DeviceEntry, error) { if err != nil { return nil, err } + entries[i].Serial = entries[i].Info.SerialString() bodyBytes += deviceInfoWireSize if bodyBytes > maxOpRepDevListBodyBytes { return nil, E.New("OP_REP_DEVLIST body too large") @@ -261,8 +262,18 @@ func (d *DeviceInfoTruncated) DevID() uint32 { return (d.BusNum << 16) | (d.DevNum & 0xffff) } -func encodePathField(dst *[256]byte, path string) { - copy(dst[:], path) +func encodePathField(dst *[256]byte, path string, serial string) { + *dst = [256]byte{} + pathLen := copy(dst[:len(dst)-1], path) + if serial == "" { + return + } + trailer := "serial=" + serial + trailerStart := pathLen + 1 + if trailerStart+len(trailer)+1 > len(dst) { + return + } + copy(dst[trailerStart:], trailer) } func cstring(b []byte) string { diff --git a/service/usbip/server.go b/service/usbip/server.go index 88309aebf..ec87a3410 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -289,6 +289,12 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended <-session.Done() released, _ := s.host.FinishImport(s.ctx, busid) s.ledger.ReleaseImport(s.ctx, busid, released) + if released { + reconcileErr := s.reconcileAndBroadcast(true) + if reconcileErr != nil { + s.logger.Debug("reconcile after ", busid, ": ", reconcileErr) + } + } return false } s.logger.Info("attached ", busid, " to remote ", conn.RemoteAddr()) @@ -299,6 +305,12 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended s.logger.Debug("finish import ", busid, ": ", err) } s.ledger.ReleaseImport(s.ctx, busid, released) + if released { + err = s.reconcileAndBroadcast(true) + if err != nil { + s.logger.Debug("reconcile after ", busid, ": ", err) + } + } }() return true } diff --git a/service/usbip/sysfs_linux.go b/service/usbip/sysfs_linux.go index bea277583..42fad5d64 100644 --- a/service/usbip/sysfs_linux.go +++ b/service/usbip/sysfs_linux.go @@ -46,7 +46,7 @@ type sysfsDevice struct { func (d *sysfsDevice) toProtocol() DeviceInfoTruncated { var info DeviceInfoTruncated - encodePathField(&info.Path, d.Path) + encodePathField(&info.Path, d.Path, d.Serial) copy(info.BusID[:], d.BusID) info.BusNum = d.BusNum info.DevNum = d.DevNum diff --git a/service/usbip/usbhost_darwin.go b/service/usbip/usbhost_darwin.go index 72d6ddbf2..0a4fbd2c3 100644 --- a/service/usbip/usbhost_darwin.go +++ b/service/usbip/usbhost_darwin.go @@ -484,7 +484,7 @@ func darwinDeviceInfoFromC(info *C.box_usbhost_device_info_t) darwinUSBHostDevic Serial: serial, } copy(entry.Info.BusID[:], busid) - encodePathField(&entry.Info.Path, path) + encodePathField(&entry.Info.Path, path, serial) interfaceCount := int(info.interface_count) if interfaceCount > C.BOX_USBHOST_MAX_INTERFACES { interfaceCount = C.BOX_USBHOST_MAX_INTERFACES