diff --git a/service/usbip/client_linux.go b/service/usbip/client_linux.go index cb3c7cc89..6398aa004 100644 --- a/service/usbip/client_linux.go +++ b/service/usbip/client_linux.go @@ -68,6 +68,7 @@ type ClientService struct { assigned []string assignedWorkers []*clientAssignedWorker allWorkers map[string]*clientBusIDWorker + allDesired map[string]struct{} attachMu sync.Mutex // serializes vhci port pick + attach wg sync.WaitGroup @@ -106,6 +107,7 @@ func NewClientService(ctx context.Context, logger log.ContextLogger, tag string, matches: options.Devices, ops: systemUSBIPOps, allWorkers: make(map[string]*clientBusIDWorker), + allDesired: make(map[string]struct{}), ports: make(map[int]struct{}), activeBusIDs: make(map[string]struct{}), }, nil @@ -289,7 +291,11 @@ func (c *ClientService) controlPingLoop(conn net.Conn, done <-chan struct{}) { } func (c *ClientService) syncRemoteState() error { - entries, err := c.fetchDevList(c.ctx) + return c.syncRemoteStateContext(c.ctx) +} + +func (c *ClientService) syncRemoteStateContext(ctx context.Context) error { + entries, err := c.fetchDevList(ctx) if err != nil { return err } @@ -316,6 +322,7 @@ func (c *ClientService) applyRemoteExports(entries []DeviceEntry) { } c.stateMu.Lock() + c.allDesired = desired stopWorkers := make([]*clientBusIDWorker, 0) for busid, worker := range c.allWorkers { if _, ok := desired[busid]; ok { @@ -519,6 +526,10 @@ func (c *ClientService) runBusIDLoop(ctx context.Context, busid, description str if err := ctx.Err(); err != nil { return } + if !c.shouldRetryBusID(ctx, busid) { + c.logger.Info("remote export ", busid, " disappeared; stopping import worker") + return + } c.logger.Info("vhci port ", port, " released; reattaching ", busid) if !sleepCtx(ctx, clientReconnectDelay) { return @@ -595,6 +606,9 @@ func (c *ClientService) watchPort(ctx context.Context, port int, busid string) { case <-settleDeadline.C: if !seenUsed { c.logger.Warn("vhci port ", port, " never reached used state; reattaching ", busid) + if err := c.ops.vhciDetach(port); err != nil { + c.logger.Warn("detach port ", port, " (", busid, "): ", err) + } return } case <-ticker.C: @@ -668,6 +682,29 @@ func (c *ClientService) isBusIDActive(busid string) bool { return exists } +func (c *ClientService) shouldRetryBusID(ctx context.Context, busid string) bool { + if len(c.matches) != 0 { + return true + } + if err := c.syncRemoteStateContext(ctx); err != nil { + c.logger.Warn("refresh remote exports after releasing ", busid, ": ", err) + return true + } + return c.isBusIDRetryDesired(busid) +} + +func (c *ClientService) isBusIDRetryDesired(busid string) bool { + c.stateMu.Lock() + defer c.stateMu.Unlock() + if _, registered := c.allWorkers[busid]; !registered { + return false + } + if _, desired := c.allDesired[busid]; desired { + return true + } + return false +} + func isBusIDOnlyMatch(m option.USBIPDeviceMatch) bool { return m.BusID != "" && m.VendorID == 0 && m.ProductID == 0 && m.Serial == "" } diff --git a/service/usbip/linux_test.go b/service/usbip/linux_test.go index deaf7c9d8..39935ea94 100644 --- a/service/usbip/linux_test.go +++ b/service/usbip/linux_test.go @@ -35,6 +35,18 @@ func (testDialer) ListenPacket(context.Context, M.Socksaddr) (net.PacketConn, er return nil, errors.New("unused") } +type failingDialer struct { + err error +} + +func (d failingDialer) DialContext(context.Context, string, M.Socksaddr) (net.Conn, error) { + return nil, d.err +} + +func (d failingDialer) ListenPacket(context.Context, M.Socksaddr) (net.PacketConn, error) { + return nil, errors.New("unused") +} + type testDeviceStore struct { mu sync.Mutex devices map[string]sysfsDevice @@ -415,6 +427,63 @@ func TestClientApplyRemoteExportsKeepsActiveBusIDWorker(t *testing.T) { require.NotContains(t, client.allWorkers, "1-1") } +func TestClientShouldRetryBusIDRefreshesImportAllState(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + server := &ServerService{ + ctx: ctx, + cancel: cancel, + logger: newTestLogger(), + exports: make(map[string]serverExport), + controlSubs: make(map[uint64]*serverControlConn), + ops: newTestUSBIPOps(t), + } + serverAddr, closeServer := startDispatchServer(t, server) + defer closeServer() + + canceled := false + client := &ClientService{ + ctx: context.Background(), + logger: newTestLogger(), + dialer: testDialer{}, + serverAddr: serverAddr, + allWorkers: map[string]*clientBusIDWorker{"1-1": {cancel: func() { canceled = true }}}, + allDesired: map[string]struct{}{"1-1": {}}, + activeBusIDs: make(map[string]struct{}), + ops: newTestUSBIPOps(t), + } + + require.False(t, client.shouldRetryBusID(context.Background(), "1-1")) + require.True(t, canceled) + require.NotContains(t, client.allWorkers, "1-1") + require.Empty(t, client.allDesired) +} + +func TestClientShouldRetryBusIDKeepsRetryOnRefreshFailure(t *testing.T) { + t.Parallel() + + expectedErr := errors.New("devlist unavailable") + canceled := false + client := &ClientService{ + ctx: context.Background(), + logger: newTestLogger(), + dialer: failingDialer{err: expectedErr}, + serverAddr: M.ParseSocksaddrHostPort("127.0.0.1", 3240), + allWorkers: map[string]*clientBusIDWorker{"1-1": {cancel: func() { canceled = true }}}, + allDesired: map[string]struct{}{"1-1": {}}, + activeBusIDs: make(map[string]struct{}), + ops: newTestUSBIPOps(t), + } + + require.True(t, client.shouldRetryBusID(context.Background(), "1-1")) + require.False(t, canceled) + require.Contains(t, client.allWorkers, "1-1") + require.Contains(t, client.allDesired, "1-1") +} + func TestAssignMatchedBusIDs(t *testing.T) { t.Parallel()