From 7b73e80df37a6a8f15369a4ace82b5b9ae140e30 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Fri, 24 Apr 2026 21:57:52 +0800 Subject: [PATCH] Fix Darwin USB/IP transfer cleanup --- service/usbip/client_darwin.go | 9 ++- service/usbip/darwin_integration_test.go | 60 ++++++++++++++++ service/usbip/data_protocol.go | 1 + service/usbip/server_darwin.go | 47 +++++++++++- service/usbip/server_darwin_test.go | 91 ++++++++++++++++++++++++ service/usbip/usbhost_darwin.go | 11 +++ 6 files changed, 215 insertions(+), 4 deletions(-) diff --git a/service/usbip/client_darwin.go b/service/usbip/client_darwin.go index 84e6292c5..4bf976b37 100644 --- a/service/usbip/client_darwin.go +++ b/service/usbip/client_darwin.go @@ -1165,6 +1165,12 @@ func (c *darwinVirtualController) handleIsoTransfer(key darwinEndpointKey, messa } else { buffer = bytesFromUnsafe(message.bufferPointer(), length) } + startFrame := message.isoFrame() + transferFlags := int32(0) + if message.isoASAP() { + startFrame = 0 + transferFlags = usbipTransferFlagIsoASAP + } response, err := c.sendSubmit(SubmitCommand{ Header: DataHeader{ Command: CmdSubmit, @@ -1172,8 +1178,9 @@ func (c *darwinVirtualController) handleIsoTransfer(key darwinEndpointKey, messa Direction: direction, Endpoint: uint32(key.endpoint & 0x0f), }, + TransferFlags: transferFlags, TransferBufferLength: int32(length), - StartFrame: int32(message.control >> 16 & 0xff), + StartFrame: startFrame, NumberOfPackets: 1, Buffer: buffer, IsoPackets: []IsoPacketDescriptor{{ diff --git a/service/usbip/darwin_integration_test.go b/service/usbip/darwin_integration_test.go index 1120e399c..c441c5e5d 100644 --- a/service/usbip/darwin_integration_test.go +++ b/service/usbip/darwin_integration_test.go @@ -177,6 +177,66 @@ func TestDarwinVirtualControllerReadsCompliantSubmitResponsePayload(t *testing.T } } +func TestDarwinHandleIsoTransferPreservesASAPFlag(t *testing.T) { + t.Parallel() + + clientConn, serverConn := net.Pipe() + defer serverConn.Close() + controller := newDarwinVirtualController(context.Background(), newTestLogger(), clientConn, DeviceInfoTruncated{ + BusNum: 1, + DevNum: 2, + }) + go controller.readLoop() + t.Cleanup(controller.Close) + + type transferResult struct { + status int32 + length int + } + resultCh := make(chan transferResult, 1) + var buffer [8]byte + go func() { + status, length := controller.handleIsoTransfer(darwinEndpointKey{device: 1, endpoint: 0x81}, darwinCIMessage{ + control: ciIsochronousTransferControlASAP | (0x7f << ciIsochronousTransferControlFramePhase), + data0: uint32(len(buffer)), + buffer: unsafe.Pointer(&buffer[0]), + }) + resultCh <- transferResult{status: status, length: length} + }() + + header, err := ReadDataHeader(serverConn) + require.NoError(t, err) + command, err := ReadSubmitCommandBody(serverConn, header) + require.NoError(t, err) + require.Equal(t, CmdSubmit, header.Command) + require.Equal(t, USBIPDirIn, command.Header.Direction) + require.Equal(t, uint32(1), command.Header.Endpoint) + require.Equal(t, int32(usbipTransferFlagIsoASAP), command.TransferFlags) + require.Zero(t, command.StartFrame) + require.Equal(t, int32(1), command.NumberOfPackets) + require.Len(t, command.IsoPackets, 1) + + require.NoError(t, WriteSubmitResponse(serverConn, SubmitResponse{ + Header: DataHeader{ + Command: RetSubmit, + SeqNum: header.SeqNum, + Direction: USBIPDirIn, + }, + Status: 0, + ActualLength: 0, + NumberOfPackets: 1, + IsoPackets: []IsoPacketDescriptor{{}}, + })) + + select { + case result := <-resultCh: + require.Zero(t, result.status) + require.Zero(t, result.length) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for isochronous submit response") + } +} + func TestDarwinSubmitInTransferRejectsOversizedPayload(t *testing.T) { t.Parallel() diff --git a/service/usbip/data_protocol.go b/service/usbip/data_protocol.go index 3cd454c3c..16bc99af9 100644 --- a/service/usbip/data_protocol.go +++ b/service/usbip/data_protocol.go @@ -22,6 +22,7 @@ const ( maxUSBIPTransferBufferLength = 16 << 20 maxUSBIPIsoPackets = 4096 nonIsoPacketCount = -1 + usbipTransferFlagIsoASAP = 0x0002 usbipStatusECONNRESET = -104 ) diff --git a/service/usbip/server_darwin.go b/service/usbip/server_darwin.go index 4dcce5168..0e2e02015 100644 --- a/service/usbip/server_darwin.go +++ b/service/usbip/server_darwin.go @@ -746,7 +746,7 @@ type darwinServerDataSession struct { ctx context.Context logger log.ContextLogger conn net.Conn - device *darwinUSBHostDevice + device darwinServerDataDevice writeMu sync.Mutex mu sync.Mutex pending map[uint32]darwinServerPendingSubmit @@ -758,7 +758,14 @@ type darwinServerPendingSubmit struct { unlinked bool } -func newDarwinServerDataSession(ctx context.Context, logger log.ContextLogger, conn net.Conn, device *darwinUSBHostDevice) *darwinServerDataSession { +type darwinServerDataDevice interface { + control(setup [8]byte, buffer []byte) (int32, int32, []byte, error) + io(endpoint uint8, buffer []byte) (int32, int32, []byte, error) + iso(endpoint uint8, buffer []byte, startFrame int32, packets []IsoPacketDescriptor) (int32, int32, []byte, []IsoPacketDescriptor, error) + abortEndpoint(endpoint uint8) error +} + +func newDarwinServerDataSession(ctx context.Context, logger log.ContextLogger, conn net.Conn, device darwinServerDataDevice) *darwinServerDataSession { return &darwinServerDataSession{ ctx: ctx, logger: logger, @@ -771,7 +778,10 @@ func newDarwinServerDataSession(ctx context.Context, logger log.ContextLogger, c func (s *darwinServerDataSession) serve() error { stopCloseOnCancel := closeConnOnContextDone(s.ctx, s.conn) defer stopCloseOnCancel() - defer s.wg.Wait() + defer func() { + s.abortPendingSubmits() + s.wg.Wait() + }() for { header, err := ReadDataHeader(s.conn) if err != nil { @@ -903,6 +913,37 @@ func (s *darwinServerDataSession) finishSubmit(seq uint32) bool { return !pending.unlinked } +func (s *darwinServerDataSession) abortPendingSubmits() { + endpoints := s.markPendingSubmitsUnlinked() + if s.device == nil { + return + } + for _, endpoint := range endpoints { + if err := s.device.abortEndpoint(endpoint); err != nil { + s.logger.Debug("abort endpoint 0x", hex8(endpoint), ": ", err) + } + } +} + +func (s *darwinServerDataSession) markPendingSubmitsUnlinked() []uint8 { + s.mu.Lock() + defer s.mu.Unlock() + seen := make(map[uint8]struct{}) + for seq, pending := range s.pending { + if !pending.unlinked { + seen[pending.endpoint] = struct{}{} + } + pending.unlinked = true + s.pending[seq] = pending + } + endpoints := make([]uint8, 0, len(seen)) + for endpoint := range seen { + endpoints = append(endpoints, endpoint) + } + slices.Sort(endpoints) + return endpoints +} + func commandEndpoint(command SubmitCommand) uint8 { endpoint := uint8(command.Header.Endpoint & 0x0f) if command.Header.Direction == USBIPDirIn { diff --git a/service/usbip/server_darwin_test.go b/service/usbip/server_darwin_test.go index d0cf04d1d..ca67354a4 100644 --- a/service/usbip/server_darwin_test.go +++ b/service/usbip/server_darwin_test.go @@ -34,6 +34,97 @@ func TestDarwinServerPendingSubmitUnlinkState(t *testing.T) { require.False(t, active) } +func TestDarwinServerAbortPendingSubmitsMarksAndAbortsEndpoints(t *testing.T) { + t.Parallel() + + device := &fakeDarwinServerDataDevice{} + session := &darwinServerDataSession{ + logger: newTestLogger(), + device: device, + pending: make(map[uint32]darwinServerPendingSubmit), + } + + session.trackSubmit(7, 0x81) + session.trackSubmit(8, 0x81) + session.trackSubmit(9, 0x02) + session.abortPendingSubmits() + + require.Equal(t, []uint8{0x02, 0x81}, device.aborted) + require.False(t, session.finishSubmit(7)) + require.False(t, session.finishSubmit(8)) + require.False(t, session.finishSubmit(9)) +} + +func TestDarwinServerServeAbortsPendingSubmitOnClose(t *testing.T) { + t.Parallel() + + serverConn, clientConn := net.Pipe() + device := &fakeDarwinServerDataDevice{ + ioStarted: make(chan struct{}), + abortNotify: make(chan struct{}), + } + session := newDarwinServerDataSession(context.Background(), newTestLogger(), serverConn, device) + done := make(chan error, 1) + go func() { + done <- session.serve() + }() + + require.NoError(t, WriteSubmitCommand(clientConn, SubmitCommand{ + Header: DataHeader{ + Command: CmdSubmit, + SeqNum: 1, + Direction: USBIPDirIn, + Endpoint: 1, + }, + TransferBufferLength: 8, + })) + select { + case <-device.ioStarted: + case <-time.After(time.Second): + t.Fatal("timed out waiting for pending Darwin IO") + } + + require.NoError(t, clientConn.Close()) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timed out waiting for Darwin session shutdown") + } + require.Equal(t, []uint8{0x81}, device.aborted) +} + +type fakeDarwinServerDataDevice struct { + ioStarted chan struct{} + abortNotify chan struct{} + aborted []uint8 +} + +func (d *fakeDarwinServerDataDevice) control(setup [8]byte, buffer []byte) (int32, int32, []byte, error) { + return 0, 0, buffer, nil +} + +func (d *fakeDarwinServerDataDevice) io(endpoint uint8, buffer []byte) (int32, int32, []byte, error) { + if d.ioStarted != nil { + close(d.ioStarted) + } + if d.abortNotify != nil { + <-d.abortNotify + } + return usbipStatusECONNRESET, 0, buffer, nil +} + +func (d *fakeDarwinServerDataDevice) iso(endpoint uint8, buffer []byte, startFrame int32, packets []IsoPacketDescriptor) (int32, int32, []byte, []IsoPacketDescriptor, error) { + return 0, 0, buffer, packets, nil +} + +func (d *fakeDarwinServerDataDevice) abortEndpoint(endpoint uint8) error { + d.aborted = append(d.aborted, endpoint) + if d.abortNotify != nil { + close(d.abortNotify) + } + return nil +} + func TestDarwinServerReconcileAndBroadcastSkipsAfterCancel(t *testing.T) { t.Parallel() diff --git a/service/usbip/usbhost_darwin.go b/service/usbip/usbhost_darwin.go index c54d1c50f..854682215 100644 --- a/service/usbip/usbhost_darwin.go +++ b/service/usbip/usbhost_darwin.go @@ -58,6 +58,9 @@ const ( ciMsgNormalTransfer = 0x39 ciMsgStatusTransfer = 0x3a ciMsgIsochronousTransfer = 0x3b + + ciIsochronousTransferControlFramePhase = 16 + ciIsochronousTransferControlASAP = 1 << 24 ) type darwinCIMessage struct { @@ -318,6 +321,14 @@ func (m darwinCIMessage) bufferPointer() unsafe.Pointer { return m.buffer } +func (m darwinCIMessage) isoASAP() bool { + return m.control&ciIsochronousTransferControlASAP != 0 +} + +func (m darwinCIMessage) isoFrame() int32 { + return int32(m.control >> ciIsochronousTransferControlFramePhase & 0xff) +} + func darwinCIFrameTimestamp() uint64 { return uint64(C.box_usbhost_now()) }