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

Fix Darwin USB/IP transfer cleanup

Browse Source
This commit is contained in:
世界
2026-04-24 21:57:52 +08:00
parent d76f9692be
commit 7b73e80df3
6 changed files with 215 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/usbip/client_darwin.go
+8 -1
View File
@@ -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{{
service/usbip/darwin_integration_test.go
+60
View File
@@ -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()
service/usbip/data_protocol.go
+1
View File
@@ -22,6 +22,7 @@ const (
maxUSBIPTransferBufferLength = 16 << 20
maxUSBIPIsoPackets = 4096
nonIsoPacketCount = -1
usbipTransferFlagIsoASAP = 0x0002
usbipStatusECONNRESET = -104
)
service/usbip/server_darwin.go
+44 -3
View File
@@ -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 {
service/usbip/server_darwin_test.go
+91
View File
@@ -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()
service/usbip/usbhost_darwin.go
+11
View File
@@ -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())
}