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

usbip: write OP_REP_IMPORT before kernel handoff; surface darwin re-capture

Browse Source 
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.
This commit is contained in:
世界
2026-05-15 13:12:25 +08:00
parent 843032af68
commit 1a17ea0ba9
6 changed files with 84 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/usbip/handoff_linux.go
+33 -14
View File
@@ -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) {
service/usbip/host.go
+7
View File
@@ -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
}
service/usbip/host_darwin.go
+11 -4
View File
@@ -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
}
service/usbip/host_linux.go
+7 -4
View File
@@ -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))
service/usbip/linux_test.go
+6 -6
View File
@@ -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)
service/usbip/server.go
+20 -10
View File
@@ -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())