From 1a17ea0ba9df5c1bcd7d099d9693caa999011f34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Fri, 15 May 2026 13:12:25 +0800 Subject: [PATCH] usbip: write OP_REP_IMPORT before kernel handoff; surface darwin re-capture On Linux the server was closing the userspace conn inside NewServerDataSession (handoff.Start closes h.conn synchronously in direct-TCP mode), so the subsequent WriteOpRepImport always wrote to a closed socket and every default-mode import failed. Split DataSession into prepare + Start: the server now writes the reply on the live conn and only then calls session.Start() to hand the wire to the kernel. Close also closes h.conn now so the prepare-then-Close error path doesn't leak the fd. On Darwin FinishImport returned false after successfully re-capturing a pending replacement for a stale busy export, so the server skipped reconcileAndBroadcast and the ledger kept the device hidden until another IOKit topology event. Return true to trigger reconcile; the next Reconcile fast-paths the matching registryID back into the snapshot. --- service/usbip/handoff_linux.go | 47 ++++++++++++++++++++++++---------- service/usbip/host.go | 7 +++++ service/usbip/host_darwin.go | 15 ++++++++--- service/usbip/host_linux.go | 11 +++++--- service/usbip/linux_test.go | 12 ++++----- service/usbip/server.go | 30 ++++++++++++++-------- 6 files changed, 84 insertions(+), 38 deletions(-) diff --git a/service/usbip/handoff_linux.go b/service/usbip/handoff_linux.go index 6f2d543b3..f91dbec53 100644 --- a/service/usbip/handoff_linux.go +++ b/service/usbip/handoff_linux.go @@ -21,6 +21,10 @@ import ( var _ DataSession = (*kernelHandoffSession)(nil) type kernelHandoffSession struct { + ctx context.Context + logger log.ContextLogger + side string + busid string conn net.Conn file *os.File monitorFile *os.File @@ -29,11 +33,12 @@ type kernelHandoffSession struct { done chan struct{} doneOnce sync.Once runErr error + startOnce sync.Once closeOnce sync.Once closeErr error } -func newKernelHandoffSession(conn net.Conn) (*kernelHandoffSession, error) { +func newKernelHandoffSession(ctx context.Context, conn net.Conn, logger log.ContextLogger, side string, busid string) (*kernelHandoffSession, error) { if tcpConn, _ := N.UnwrapReader(conn).(*net.TCPConn); tcpConn != nil { file, err := tcpConn.File() if err != nil { @@ -45,6 +50,10 @@ func newKernelHandoffSession(conn net.Conn) (*kernelHandoffSession, error) { return nil, E.Cause(err, "dup TCP socket monitor fd") } return &kernelHandoffSession{ + ctx: ctx, + logger: logger, + side: side, + busid: busid, conn: conn, file: file, monitorFile: monitorFile, @@ -65,6 +74,10 @@ func newKernelHandoffSession(conn net.Conn) (*kernelHandoffSession, error) { return nil, E.Cause(err, "wrap USB/IP relay socket") } return &kernelHandoffSession{ + ctx: ctx, + logger: logger, + side: side, + busid: busid, conn: conn, file: kernelFile, relayConn: relayConn, @@ -95,9 +108,11 @@ func (h *kernelHandoffSession) Close() error { h.closeKernelFD(), common.Close(h.monitorFile), common.Close(h.relayConn), + common.Close(h.conn), ) h.monitorFile = nil h.relayConn = nil + h.conn = nil }) h.markDone(nil) return h.closeErr @@ -110,20 +125,24 @@ func (h *kernelHandoffSession) markDone(err error) { }) } -func (h *kernelHandoffSession) Start(ctx context.Context, logger log.ContextLogger, side string, busid string) { - if h.relayConn == nil { - err := h.conn.Close() - if err != nil && !E.IsClosedOrCanceled(err) { - logger.Debug("close usbip ", side, " userspace socket ", busid, ": ", err) +func (h *kernelHandoffSession) Start() error { + h.startOnce.Do(func() { + if h.relayConn == nil { + err := h.conn.Close() + if err != nil && !E.IsClosedOrCanceled(err) { + h.logger.Debug("close usbip ", h.side, " userspace socket ", h.busid, ": ", err) + } + h.conn = nil + monitorFile := h.monitorFile + h.monitorFile = nil + go h.runDirect(h.ctx, h.logger, h.side, h.busid, monitorFile) + return } - monitorFile := h.monitorFile - h.monitorFile = nil - go h.runDirect(ctx, logger, side, busid, monitorFile) - return - } - relayConn := h.relayConn - h.relayConn = nil - go h.runRelay(ctx, logger, side, busid, relayConn) + relayConn := h.relayConn + h.relayConn = nil + go h.runRelay(h.ctx, h.logger, h.side, h.busid, relayConn) + }) + return nil } func (h *kernelHandoffSession) runDirect(ctx context.Context, logger log.ContextLogger, side string, busid string, file *os.File) { diff --git a/service/usbip/host.go b/service/usbip/host.go index 3c7a72272..ddd0442f9 100644 --- a/service/usbip/host.go +++ b/service/usbip/host.go @@ -52,9 +52,16 @@ type ExportSnapshot struct { // when the session terminates for any reason. Err is only valid after // Done is closed; it returns nil for a clean detach. Close is idempotent // and safe to call from any goroutine. +// +// Start transfers the session from "prepared" to "running". The session +// is returned in the prepared state with the userspace conn still alive +// so the server can send OP_REP_IMPORT before the kernel takes wire +// ownership. Start is idempotent. Close before Start releases all +// resources, including the userspace conn. type DataSession interface { Done() <-chan struct{} Err() error + Start() error Close() error } diff --git a/service/usbip/host_darwin.go b/service/usbip/host_darwin.go index 2a6b95770..c18d19825 100644 --- a/service/usbip/host_darwin.go +++ b/service/usbip/host_darwin.go @@ -248,7 +248,7 @@ func (h *darwinExportHost) FinishImport(ctx context.Context, busid string) (bool exp.device.Close() } h.logger.Info("re-exported ", busid, " through IOUSBHost re-capture (registry ", pending, ")") - return false, nil + return true, nil } func (h *darwinExportHost) snapshotSelf() map[string]Export { @@ -359,6 +359,7 @@ type darwinServerDataSession struct { done chan struct{} doneOnce sync.Once runErr error + startOnce sync.Once closeOnce sync.Once closeErr error } @@ -369,7 +370,7 @@ type darwinServerPendingSubmit struct { } func newDarwinServerDataSession(ctx context.Context, logger log.ContextLogger, conn net.Conn, device *darwinUSBHostDevice) *darwinServerDataSession { - session := &darwinServerDataSession{ + return &darwinServerDataSession{ ctx: ctx, logger: logger, conn: conn, @@ -377,8 +378,6 @@ func newDarwinServerDataSession(ctx context.Context, logger log.ContextLogger, c pending: make(map[uint32]darwinServerPendingSubmit), done: make(chan struct{}), } - go session.run() - return session } func (s *darwinServerDataSession) Done() <-chan struct{} { @@ -389,10 +388,18 @@ func (s *darwinServerDataSession) Err() error { return s.runErr } +func (s *darwinServerDataSession) Start() error { + s.startOnce.Do(func() { + go s.run() + }) + return nil +} + func (s *darwinServerDataSession) Close() error { s.closeOnce.Do(func() { s.closeErr = common.Close(s.conn) }) + s.markDone(nil) <-s.done return s.closeErr } diff --git a/service/usbip/host_linux.go b/service/usbip/host_linux.go index 3e188f739..c4207c786 100644 --- a/service/usbip/host_linux.go +++ b/service/usbip/host_linux.go @@ -437,7 +437,7 @@ func (e *linuxExport) DeviceInfo(ctx context.Context) (DeviceInfoTruncated, erro } func (e *linuxExport) NewServerDataSession(ctx context.Context, conn net.Conn) (DataSession, error) { - handoff, err := newKernelHandoffSession(conn) + handoff, err := newKernelHandoffSession(ctx, conn, e.logger, "server", e.busid) if err != nil { return nil, E.Cause(err, "prepare handoff") } @@ -455,7 +455,6 @@ func (e *linuxExport) NewServerDataSession(ctx context.Context, conn net.Conn) ( if closeErr != nil { e.logger.Debug("close kernel fd ", e.busid, ": ", closeErr) } - handoff.Start(ctx, e.logger, "server", e.busid) return handoff, nil } @@ -475,7 +474,7 @@ func (h *linuxImportHost) Close() error { } func (h *linuxImportHost) Attach(ctx context.Context, info DeviceInfoTruncated, conn net.Conn) (AttachedSession, error) { - handoff, err := newKernelHandoffSession(conn) + handoff, err := newKernelHandoffSession(ctx, conn, h.logger, "client", info.BusIDString()) if err != nil { return nil, E.Cause(err, "prepare handoff") } @@ -489,7 +488,7 @@ func (h *linuxImportHost) Attach(ctx context.Context, info DeviceInfoTruncated, _ = handoff.Close() return nil, attachErr } - handoff.Start(ctx, h.logger, "client", info.BusIDString()) + _ = handoff.Start() return &linuxClientSession{ handoff: handoff, host: h, @@ -562,6 +561,10 @@ func (s *linuxClientSession) Err() error { return s.handoff.Err() } +func (s *linuxClientSession) Start() error { + return s.handoff.Start() +} + func (s *linuxClientSession) Close() error { s.closeOnce.Do(func() { detachErr := writeSysfs(filepath.Join(sysVHCIControllerV0, "detach"), strconv.Itoa(s.port)) diff --git a/service/usbip/linux_test.go b/service/usbip/linux_test.go index b07294b0d..2a4b35f01 100644 --- a/service/usbip/linux_test.go +++ b/service/usbip/linux_test.go @@ -276,13 +276,13 @@ func TestUSBIPConnHandoffDirectTCP(t *testing.T) { acceptedConn := <-accepted defer acceptedConn.Close() - handoff, err := newKernelHandoffSession(conn) + handoff, err := newKernelHandoffSession(context.Background(), conn, newTestLogger(t), "test", "direct") require.NoError(t, err) defer handoff.Close() require.Nil(t, handoff.relayConn) requireStreamSocketFD(t, handoff.file.Fd()) - handoff.Start(context.Background(), newTestLogger(t), "test", "direct") + require.NoError(t, handoff.Start()) _, err = conn.Write([]byte("closed")) require.Error(t, err) @@ -299,7 +299,9 @@ func TestUSBIPConnHandoffRelaySocketpairCopies(t *testing.T) { left, right := net.Pipe() defer right.Close() - handoff, err := newKernelHandoffSession(opaqueConn{Conn: left}) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + handoff, err := newKernelHandoffSession(ctx, opaqueConn{Conn: left}, newTestLogger(t), "test", "relay") require.NoError(t, err) defer handoff.Close() require.NotNil(t, handoff.relayConn) @@ -310,9 +312,7 @@ func TestUSBIPConnHandoffRelaySocketpairCopies(t *testing.T) { setConnDeadline(t, right) setConnDeadline(t, kernelConn) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - handoff.Start(ctx, newTestLogger(t), "test", "relay") + require.NoError(t, handoff.Start()) _, err = right.Write([]byte("ping")) require.NoError(t, err) diff --git a/service/usbip/server.go b/service/usbip/server.go index ec87a3410..80a39044e 100644 --- a/service/usbip/server.go +++ b/service/usbip/server.go @@ -127,6 +127,19 @@ func (s *ServerService) eventLoop() { } } +func (s *ServerService) tearDownPreparedSession(busid string, session DataSession) { + _ = session.Close() + <-session.Done() + released, _ := s.host.FinishImport(s.ctx, busid) + s.ledger.ReleaseImport(s.ctx, busid, released) + if released { + err := s.reconcileAndBroadcast(true) + if err != nil { + s.logger.Debug("reconcile after ", busid, ": ", err) + } + } +} + func (s *ServerService) reconcileAndBroadcast(notify bool) error { s.reconcileAccess.Lock() defer s.reconcileAccess.Unlock() @@ -285,16 +298,13 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended err = WriteOpRepImport(conn, opCode, OpStatusOK, &info) if err != nil { s.logger.Warn("reply import ", busid, ": ", err) - _ = session.Close() - <-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) - } - } + s.tearDownPreparedSession(busid, session) + return false + } + err = session.Start() + if err != nil { + s.logger.Warn("start data session ", busid, ": ", err) + s.tearDownPreparedSession(busid, session) return false } s.logger.Info("attached ", busid, " to remote ", conn.RemoteAddr())