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

usbip: use event-driven status updates

Browse Source
This commit is contained in:
世界
2026-04-25 03:31:12 +08:00
parent 09491ff209
commit 35347af96d
11 changed files with 510 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
service/usbip/client_linux.go
+27 -84
View File
@@ -19,11 +19,6 @@ import (
N "github.com/sagernet/sing/common/network"
)
const (
clientDetachTimeout = 10 * time.Second
clientDetachPoll = 100 * time.Millisecond
)
type ClientService struct {
boxService.Adapter
ctx context.Context
@@ -129,7 +124,7 @@ func (c *ClientService) runBusIDLoop(ctx context.Context, busid, description str
if ctx.Err() != nil {
return
}
port, err := c.attemptAttach(ctx, busid)
port, done, err := c.attemptAttach(ctx, busid)
if err != nil {
c.logger.Error("attach ", description, " (", busid, "): ", err)
if !sleepCtx(ctx, clientReconnectDelay) {
@@ -139,7 +134,7 @@ func (c *ClientService) runBusIDLoop(ctx context.Context, busid, description str
}
c.logger.Info("attached ", busid, " → vhci port ", port)
c.setBusIDActive(busid, true)
c.watchPort(ctx, port, busid)
c.waitPortSession(ctx, port, busid, done)
c.setBusIDActive(busid, false)
c.trackPort(port, false)
if ctx.Err() != nil {
@@ -156,10 +151,10 @@ func (c *ClientService) runBusIDLoop(ctx context.Context, busid, description str
}
}
func (c *ClientService) attemptAttach(ctx context.Context, busid string) (int, error) {
func (c *ClientService) attemptAttach(ctx context.Context, busid string) (int, <-chan struct{}, error) {
conn, err := c.dialer.DialContext(ctx, N.NetworkTCP, c.serverAddr)
if err != nil {
return -1, E.Cause(err, "dial ", c.serverAddr)
return -1, nil, E.Cause(err, "dial ", c.serverAddr)
}
relayStarted := false
defer func() {
@@ -171,7 +166,7 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (int, e
defer stopCloseOnCancel()
lease, err := c.requestImportLease(ctx, busid)
if err != nil {
return -1, err
return -1, nil, err
}
expectedReply := OpRepImport
if lease.Valid {
@@ -182,34 +177,34 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (int, e
ClientNonce: lease.ClientNonce,
})
if err != nil {
return -1, E.Cause(err, "write OP_REQ_IMPORT_EXT")
return -1, nil, E.Cause(err, "write OP_REQ_IMPORT_EXT")
}
} else {
err = WriteOpReqImport(conn, busid)
if err != nil {
return -1, E.Cause(err, "write OP_REQ_IMPORT")
return -1, nil, E.Cause(err, "write OP_REQ_IMPORT")
}
}
header, err := ReadOpHeader(conn)
if err != nil {
return -1, E.Cause(err, "read OP_REP_IMPORT header")
return -1, nil, E.Cause(err, "read OP_REP_IMPORT header")
}
if header.Version != ProtocolVersion {
return -1, E.New(fmt.Sprintf("unexpected reply version 0x%04x", header.Version))
return -1, nil, E.New(fmt.Sprintf("unexpected reply version 0x%04x", header.Version))
}
if header.Code != expectedReply {
return -1, E.New(fmt.Sprintf("unexpected reply code 0x%04x", header.Code))
return -1, nil, E.New(fmt.Sprintf("unexpected reply code 0x%04x", header.Code))
}
if header.Status != OpStatusOK {
return -1, E.New("remote rejected import (status=", header.Status, ")")
return -1, nil, E.New("remote rejected import (status=", header.Status, ")")
}
info, err := ReadOpRepImportBody(conn)
if err != nil {
return -1, E.Cause(err, "read OP_REP_IMPORT body")
return -1, nil, E.Cause(err, "read OP_REP_IMPORT body")
}
handoff, err := newUSBIPConnHandoff(conn)
if err != nil {
return -1, E.Cause(err, "prepare handoff")
return -1, nil, E.Cause(err, "prepare handoff")
}
defer func() {
if !relayStarted {
@@ -221,86 +216,34 @@ func (c *ClientService) attemptAttach(ctx context.Context, busid string) (int, e
defer c.portAssignAccess.Unlock()
port, err := c.ops.vhciPickFreePort(info.Speed)
if err != nil {
return -1, err
return -1, nil, err
}
if !c.reservePort(port) {
return -1, E.New("vhci port ", port, " already reserved")
return -1, nil, E.New("vhci port ", port, " already reserved")
}
err = c.ops.vhciAttach(port, handoff.kernelFD(), info.DevID(), info.Speed)
if err != nil {
c.trackPort(port, false)
return -1, E.Cause(err, "vhci attach")
return -1, nil, E.Cause(err, "vhci attach")
}
err = handoff.closeKernelFD()
if err != nil {
c.logger.Debug("close kernel fd ", busid, ": ", err)
}
relayStarted = handoff.startRelay(ctx, c.logger, "client", busid)
return port, nil
done := handoff.startRelay(ctx, c.logger, "client", busid)
relayStarted = true
return port, done, nil
}
func (c *ClientService) watchPort(ctx context.Context, port int, busid string) {
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
seenUsed := false
settleDeadline := time.NewTimer(10 * time.Second)
defer settleDeadline.Stop()
for {
select {
case <-ctx.Done():
err := c.ops.vhciDetach(port)
if err != nil {
c.logger.Warn("detach port ", port, " (", busid, "): ", err)
}
c.waitVHCIPortIdle(port, busid)
return
case <-settleDeadline.C:
if !seenUsed {
c.logger.Warn("vhci port ", port, " never reached used state; reattaching ", busid)
err := c.ops.vhciDetach(port)
if err != nil {
c.logger.Warn("detach port ", port, " (", busid, "): ", err)
}
c.waitVHCIPortIdle(port, busid)
return
}
case <-ticker.C:
used, err := c.ops.vhciPortUsed(port)
if err != nil {
c.logger.Debug("poll port ", port, ": ", err)
continue
}
if used {
if !seenUsed {
c.logger.Debug("vhci port ", port, " entered used state for ", busid)
}
seenUsed = true
continue
}
if seenUsed {
c.logger.Debug("vhci port ", port, " left used state for ", busid)
return
}
func (c *ClientService) waitPortSession(ctx context.Context, port int, busid string, done <-chan struct{}) {
select {
case <-ctx.Done():
err := c.ops.vhciDetach(port)
if err != nil {
c.logger.Warn("detach port ", port, " (", busid, "): ", err)
}
}
}
func (c *ClientService) waitVHCIPortIdle(port int, busid string) {
deadline := time.Now().Add(clientDetachTimeout)
for {
used, err := c.ops.vhciPortUsed(port)
if err == nil && !used {
return
}
if time.Now().After(deadline) {
if err != nil {
c.logger.Warn("poll detached vhci port ", port, " (", busid, "): ", err)
} else {
c.logger.Warn("vhci port ", port, " stayed used after detach for ", busid)
}
return
}
time.Sleep(clientDetachPoll)
case <-done:
c.logger.Debug("vhci port ", port, " session ended for ", busid)
}
}
service/usbip/darwin_integration_test.go
+11 -3
View File
@@ -63,7 +63,7 @@ type darwinFakeUSBIPServer struct {
func requireRoot(t *testing.T) {
t.Helper()
if os.Geteuid() != 0 {
t.Skip("root required")
t.Skip("root required; run with go test -exec sudo")
}
}
@@ -455,6 +455,14 @@ func TestDarwinControllerCloseWithNilConn(t *testing.T) {
}
}
func TestDarwinUSBHostDeviceWatcherSmoke(t *testing.T) {
watcher, err := darwinWatchUSBHostDevices(func() {})
if err != nil {
t.Skipf("IOUSBHostDevice watcher unavailable: %v", err)
}
watcher.Close()
}
func startDarwinFakeUSBIPServer(t *testing.T) *darwinFakeUSBIPServer {
t.Helper()
@@ -756,7 +764,7 @@ func darwinFakeDeviceEntry() DeviceEntry {
}
}
func TestDarwinUSBIPClientImportsFakeServer(t *testing.T) {
func TestDarwinUSBIPClientSmoke(t *testing.T) {
requireRoot(t)
requireDarwinUserHCI(t)
@@ -793,7 +801,7 @@ func TestDarwinUSBIPClientImportsFakeServer(t *testing.T) {
}
}
func TestDarwinUSBIPServerSelectedDeviceConfiguresDevice(t *testing.T) {
func TestDarwinUSBIPServerSmoke(t *testing.T) {
requireRoot(t)
candidate, ok := darwinSafeCaptureCandidate(t)
service/usbip/handoff_linux.go
+56 -8
View File
@@ -4,8 +4,10 @@ package usbip
import (
"context"
"errors"
"net"
"os"
"sync"
"github.com/sagernet/sing-box/log"
"github.com/sagernet/sing/common"
@@ -17,9 +19,10 @@ import (
)
type usbipConnHandoff struct {
conn net.Conn
file *os.File
relayConn net.Conn
conn net.Conn
file *os.File
monitorFile *os.File
relayConn net.Conn
}
func newUSBIPConnHandoff(conn net.Conn) (*usbipConnHandoff, error) {
@@ -28,9 +31,15 @@ func newUSBIPConnHandoff(conn net.Conn) (*usbipConnHandoff, error) {
if err != nil {
return nil, E.Cause(err, "dup TCP socket fd")
}
monitorFile, err := tcpConn.File()
if err != nil {
_ = file.Close()
return nil, E.Cause(err, "dup TCP socket monitor fd")
}
return &usbipConnHandoff{
conn: conn,
file: file,
conn: conn,
file: file,
monitorFile: monitorFile,
}, nil
}
@@ -80,21 +89,27 @@ func (h *usbipConnHandoff) closeKernelFD() error {
func (h *usbipConnHandoff) Close() error {
return E.Errors(
h.closeKernelFD(),
common.Close(h.monitorFile),
common.Close(h.relayConn),
)
}
func (h *usbipConnHandoff) startRelay(ctx context.Context, logger log.ContextLogger, side string, busid string) bool {
func (h *usbipConnHandoff) startRelay(ctx context.Context, logger log.ContextLogger, side string, busid string) <-chan struct{} {
done := make(chan struct{})
if !h.relay() {
err := h.conn.Close()
if err != nil && !E.IsClosedOrCanceled(err) {
logger.Debug("close usbip ", side, " userspace socket ", busid, ": ", err)
}
return true
monitorFile := h.monitorFile
h.monitorFile = nil
go monitorDirectHandoff(ctx, logger, side, busid, monitorFile, done)
return done
}
relayConn := h.relayConn
h.relayConn = nil
go func() {
defer close(done)
err := sBufio.CopyConn(ctx, h.conn, relayConn)
if err == nil {
logger.Debug("usbip ", side, " relay ", busid, " closed")
@@ -104,5 +119,38 @@ func (h *usbipConnHandoff) startRelay(ctx context.Context, logger log.ContextLog
logger.Debug("usbip ", side, " relay ", busid, ": ", err)
}
}()
return true
return done
}
func monitorDirectHandoff(ctx context.Context, logger log.ContextLogger, side string, busid string, file *os.File, done chan<- struct{}) {
defer close(done)
if file == nil {
return
}
closeFile := sync.OnceFunc(func() {
_ = file.Close()
})
stopCloseOnCancel := context.AfterFunc(ctx, closeFile)
defer func() {
stopCloseOnCancel()
closeFile()
}()
fd := int32(file.Fd())
for {
events := int16(unix.POLLHUP | unix.POLLERR | unix.POLLRDHUP)
fds := []unix.PollFd{{Fd: fd, Events: events}}
_, err := unix.Poll(fds, -1)
if err == unix.EINTR {
continue
}
if err != nil {
if ctx.Err() == nil && !errors.Is(err, unix.EBADF) {
logger.Debug("usbip ", side, " direct monitor ", busid, ": ", err)
}
return
}
if fds[0].Revents&(events|unix.POLLNVAL) != 0 {
return
}
}
}
service/usbip/linux_test.go
+59 -17
View File
@@ -229,10 +229,6 @@ func newTestUSBIPOps(t *testing.T) usbipOps {
t.Fatalf("unexpected vhciDetach")
return nil
},
vhciPortUsed: func(int) (bool, error) {
t.Fatalf("unexpected vhciPortUsed")
return false, nil
},
}
}
@@ -324,6 +320,12 @@ func duplicateNetConnFromFD(fd uintptr, name string) (net.Conn, error) {
return conn, nil
}
func linuxServerControlState(server *ServerService, busid string) string {
server.controlAccess.Lock()
defer server.controlAccess.Unlock()
return server.controlState[busid].State
}
func duplicateHandoffKernelConn(t *testing.T, handoff *usbipConnHandoff) net.Conn {
t.Helper()
@@ -714,10 +716,16 @@ func TestUSBIPConnHandoffDirectTCP(t *testing.T) {
require.False(t, handoff.relay())
require.Equal(t, "direct", handoff.mode())
requireStreamSocketFD(t, handoff.kernelFD())
require.True(t, handoff.startRelay(context.Background(), newTestLogger(), "test", "direct"))
done := handoff.startRelay(context.Background(), newTestLogger(), "test", "direct")
_, err = conn.Write([]byte("closed"))
require.Error(t, err)
require.NoError(t, acceptedConn.Close())
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("timed out waiting for direct handoff monitor")
}
}
func TestUSBIPConnHandoffRelaySocketpairCopies(t *testing.T) {
@@ -739,7 +747,7 @@ func TestUSBIPConnHandoffRelaySocketpairCopies(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
require.True(t, handoff.startRelay(ctx, newTestLogger(), "test", "relay"))
done := handoff.startRelay(ctx, newTestLogger(), "test", "relay")
_, err = right.Write([]byte("ping"))
require.NoError(t, err)
@@ -748,6 +756,14 @@ func TestUSBIPConnHandoffRelaySocketpairCopies(t *testing.T) {
_, err = kernelConn.Write([]byte("pong"))
require.NoError(t, err)
requireConnRead(t, right, []byte("pong"))
require.NoError(t, right.Close())
require.NoError(t, kernelConn.Close())
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("timed out waiting for relay handoff")
}
}
func TestServerStartRequiresHostDriver(t *testing.T) {
@@ -1229,12 +1245,16 @@ func TestServerHandleImportWithOpaqueConnRelay(t *testing.T) {
ops.readSysfsDevice = store.readSysfsDevice
ops.writeUsbipSockfd = func(busid string, fd int) error {
if fd < 0 {
store.setStatus(busid, usbipStatusAvailable)
store.writeUsbipSockfd(busid, fd)
return nil
}
if busid != "1-1" {
kernelErrCh <- fmt.Errorf("unexpected busid %s", busid)
return nil
}
store.setStatus(busid, usbipStatusUsed)
store.writeUsbipSockfd(busid, fd)
socketType, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_TYPE)
if err != nil {
kernelErrCh <- err
@@ -1254,12 +1274,13 @@ func TestServerHandleImportWithOpaqueConnRelay(t *testing.T) {
}
server := &ServerService{
ctx: ctx,
cancel: cancel,
logger: newTestLogger(),
exports: map[string]serverExport{"1-1": {busid: "1-1"}},
controlSubs: make(map[uint64]*serverControlConn),
ops: ops,
ctx: ctx,
cancel: cancel,
logger: newTestLogger(),
exports: map[string]serverExport{"1-1": {busid: "1-1"}},
controlSubs: make(map[uint64]*serverControlConn),
controlState: make(map[string]DeviceInfoV2),
ops: ops,
}
serverConn, clientConn := net.Pipe()
@@ -1274,6 +1295,9 @@ func TestServerHandleImportWithOpaqueConnRelay(t *testing.T) {
require.Equal(t, OpStatusOK, header.Status)
_, err = ReadOpRepImportBody(clientConn)
require.NoError(t, err)
require.Eventually(t, func() bool {
return linuxServerControlState(server, "1-1") == deviceStateBusy
}, time.Second, 10*time.Millisecond)
var kernelConn net.Conn
select {
@@ -1293,6 +1317,12 @@ func TestServerHandleImportWithOpaqueConnRelay(t *testing.T) {
_, err = kernelConn.Write([]byte("server-out"))
require.NoError(t, err)
requireConnRead(t, clientConn, []byte("server-out"))
require.NoError(t, clientConn.Close())
require.NoError(t, kernelConn.Close())
require.Eventually(t, func() bool {
return store.lastSockfd("1-1") == -1 && linuxServerControlState(server, "1-1") == deviceStateAvailable
}, time.Second, 10*time.Millisecond)
}
func TestServerHandleImportRelayClosesHandoffOnSockfdFailure(t *testing.T) {
@@ -1812,8 +1842,9 @@ func TestClientAttemptAttachUsesImportReplyAndVHCIAttach(t *testing.T) {
ops: clientOps,
}
port, err := client.attemptAttach(ctx, "1-1")
port, done, err := client.attemptAttach(ctx, "1-1")
require.NoError(t, err)
require.NotNil(t, done)
require.Equal(t, 7, port)
require.Equal(t, 7, attachedPort)
info := device.toProtocol()
@@ -1943,8 +1974,9 @@ func TestClientAttemptAttachUsesImportExtLease(t *testing.T) {
client.setControlSession(controlSession)
defer client.clearControlSession(controlSession, errClientControlSessionClosed)
port, err := client.attemptAttach(ctx, "1-1")
port, done, err := client.attemptAttach(ctx, "1-1")
require.NoError(t, err)
require.NotNil(t, done)
require.Equal(t, 4, port)
require.NoError(t, <-controlErrCh)
require.NoError(t, <-deliverErrCh)
@@ -2031,8 +2063,9 @@ func TestClientAttemptAttachWithOpaqueConnRelay(t *testing.T) {
ops: ops,
}
port, err := client.attemptAttach(ctx, "1-1")
port, done, err := client.attemptAttach(ctx, "1-1")
require.NoError(t, err)
require.NotNil(t, done)
require.Equal(t, 4, port)
var serverConn net.Conn
@@ -2060,6 +2093,13 @@ func TestClientAttemptAttachWithOpaqueConnRelay(t *testing.T) {
_, err = kernelConn.Write([]byte("client-out"))
require.NoError(t, err)
requireConnRead(t, serverConn, []byte("client-out"))
require.NoError(t, kernelConn.Close())
select {
case <-done:
case <-time.After(time.Second):
t.Fatal("timed out waiting for client relay handoff")
}
}
func TestClientAttemptAttachRelayClosesHandoffOnVHCIAttachFailure(t *testing.T) {
@@ -2143,8 +2183,9 @@ func TestClientAttemptAttachRelayClosesHandoffOnVHCIAttachFailure(t *testing.T)
ops: ops,
}
port, err := client.attemptAttach(ctx, "1-1")
port, done, err := client.attemptAttach(ctx, "1-1")
require.Equal(t, -1, port)
require.Nil(t, done)
require.ErrorIs(t, err, expectedErr)
var kernelConn net.Conn
@@ -2422,8 +2463,9 @@ func TestClientAttemptAttachRejectsUnexpectedReplyVersion(t *testing.T) {
ops: ops,
}
port, err := client.attemptAttach(ctx, "1-1")
port, done, err := client.attemptAttach(ctx, "1-1")
require.Equal(t, -1, port)
require.Nil(t, done)
require.ErrorContains(t, err, "unexpected reply version")
require.NoError(t, <-serverErr)
}
service/usbip/ops_linux.go
-2
View File
@@ -27,7 +27,6 @@ type usbipOps struct {
vhciPickFreePort func(speed uint32) (int, error)
vhciAttach func(port int, fd uintptr, devid uint32, speed uint32) error
vhciDetach func(port int) error
vhciPortUsed func(port int) (bool, error)
}
var systemUSBIPOps = usbipOps{
@@ -50,5 +49,4 @@ var systemUSBIPOps = usbipOps{
vhciPickFreePort: vhciPickFreePort,
vhciAttach: vhciAttach,
vhciDetach: vhciDetach,
vhciPortUsed: vhciPortUsed,
}
service/usbip/server_darwin.go
+60 -20
View File
@@ -11,7 +11,6 @@ import (
"slices"
"strings"
"sync"
"time"
"github.com/sagernet/sing-box/adapter"
boxService "github.com/sagernet/sing-box/adapter/service"
@@ -34,6 +33,22 @@ type serverExport struct {
busy bool
}
type darwinUSBHostDeviceWatch interface {
Close()
}
type darwinServerOps struct {
copyUSBHostDevices func() ([]darwinUSBHostDeviceInfo, error)
openUSBHostDevice func(registryID uint64, capture bool) (*darwinUSBHostDevice, error)
watchUSBHostDevices func(func()) (darwinUSBHostDeviceWatch, error)
}
var systemDarwinServerOps = darwinServerOps{
copyUSBHostDevices: darwinCopyUSBHostDevices,
openUSBHostDevice: darwinOpenUSBHostDevice,
watchUSBHostDevices: darwinWatchUSBHostDevices,
}
type ServerService struct {
boxService.Adapter
ctx context.Context
@@ -41,10 +56,12 @@ type ServerService struct {
logger log.ContextLogger
listener *listener.Listener
matches []option.USBIPDeviceMatch
ops darwinServerOps
access sync.Mutex
exports map[string]serverExport
listen net.Listener
watcher darwinUSBHostDeviceWatch
controlAccess sync.Mutex
controlSeq uint64
@@ -83,6 +100,7 @@ func NewServerService(ctx context.Context, logger log.ContextLogger, tag string,
controlSubs: make(map[uint64]*serverControlConn),
controlState: make(map[string]DeviceInfoV2),
leasesByBusID: make(map[string]serverImportLease),
ops: systemDarwinServerOps,
}, nil
}
@@ -94,17 +112,23 @@ func (s *ServerService) Start(stage adapter.StartStage) error {
if err != nil {
return err
}
watcher, err := s.newUSBEventWatcher()
if err != nil {
s.rollbackExports()
return err
}
var tcpListener net.Listener
tcpListener, err = s.listener.ListenTCP()
if err != nil {
watcher.Close()
s.rollbackExports()
return err
}
s.access.Lock()
s.listen = tcpListener
s.watcher = watcher
s.access.Unlock()
go s.acceptLoop(tcpListener)
go s.reconcileLoop()
return nil
}
@@ -114,14 +138,46 @@ func (s *ServerService) Close() error {
}
s.closeControlSubscribers()
err := common.Close(common.PtrOrNil(s.listener))
s.access.Lock()
watcher := s.watcher
s.watcher = nil
s.access.Unlock()
if watcher != nil {
watcher.Close()
}
s.reconcileAccess.Lock()
defer s.reconcileAccess.Unlock()
s.rollbackExports()
return err
}
func (s *ServerService) newUSBEventWatcher() (darwinUSBHostDeviceWatch, error) {
ops := s.darwinOps()
return ops.watchUSBHostDevices(func() {
err := s.reconcileAndBroadcast(true)
if err != nil {
s.logger.Warn("reconcile exports: ", err)
}
})
}
func (s *ServerService) darwinOps() darwinServerOps {
ops := s.ops
if ops.copyUSBHostDevices == nil {
ops.copyUSBHostDevices = darwinCopyUSBHostDevices
}
if ops.openUSBHostDevice == nil {
ops.openUSBHostDevice = darwinOpenUSBHostDevice
}
if ops.watchUSBHostDevices == nil {
ops.watchUSBHostDevices = darwinWatchUSBHostDevices
}
return ops
}
func (s *ServerService) reconcileExports() (bool, error) {
devices, err := darwinCopyUSBHostDevices()
ops := s.darwinOps()
devices, err := ops.copyUSBHostDevices()
if err != nil {
return false, E.Cause(err, "enumerate IOUSBHost devices")
}
@@ -153,7 +209,7 @@ func (s *ServerService) reconcileExports() (bool, error) {
export.device.Close()
changed = true
}
device, err := darwinOpenUSBHostDevice(info.registryID, true)
device, err := ops.openUSBHostDevice(info.registryID, true)
if err != nil {
s.logger.Warn("capture ", busid, ": ", err)
continue
@@ -405,22 +461,6 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended
s.broadcastChanged()
}
func (s *ServerService) reconcileLoop() {
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
select {
case <-s.ctx.Done():
return
case <-ticker.C:
}
err := s.reconcileAndBroadcast(true)
if err != nil {
s.logger.Warn("reconcile exports: ", err)
}
}
}
func (s *ServerService) broadcastChanged() {
s.broadcastControlState(deviceInfoV2Map(s.buildDeviceStateV2()), false)
}
service/usbip/server_darwin_test.go
+73
View File
@@ -9,6 +9,7 @@ import (
"time"
"github.com/sagernet/sing-box/log"
"github.com/sagernet/sing-box/option"
"github.com/stretchr/testify/require"
)
@@ -142,6 +143,65 @@ func TestDarwinServerReconcileAndBroadcastSkipsAfterCancel(t *testing.T) {
require.NoError(t, server.reconcileAndBroadcast(true))
}
func TestDarwinServerUSBEventWatcherTriggersReconcile(t *testing.T) {
t.Parallel()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
const busid = "mac-00000001"
entry := standardTestDeviceEntry(busid)
info := darwinUSBHostDeviceInfo{
registryID: 1,
entry: entry,
key: DeviceKey{
BusID: busid,
VendorID: entry.Info.IDVendor,
ProductID: entry.Info.IDProduct,
Serial: entry.Serial,
},
}
var devices []darwinUSBHostDeviceInfo
var fakeWatch *fakeDarwinUSBHostDeviceWatch
server := &ServerService{
ctx: ctx,
logger: newTestLogger(),
matches: []option.USBIPDeviceMatch{{BusID: busid}},
exports: make(map[string]serverExport),
controlSubs: make(map[uint64]*serverControlConn),
controlState: make(map[string]DeviceInfoV2),
ops: darwinServerOps{
copyUSBHostDevices: func() ([]darwinUSBHostDeviceInfo, error) {
return devices, nil
},
openUSBHostDevice: func(registryID uint64, capture bool) (*darwinUSBHostDevice, error) {
require.Equal(t, info.registryID, registryID)
require.True(t, capture)
return &darwinUSBHostDevice{info: info}, nil
},
watchUSBHostDevices: func(callback func()) (darwinUSBHostDeviceWatch, error) {
fakeWatch = &fakeDarwinUSBHostDeviceWatch{callback: callback}
return fakeWatch, nil
},
},
}
watcher, err := server.newUSBEventWatcher()
require.NoError(t, err)
require.NotNil(t, watcher)
require.NotNil(t, fakeWatch)
devices = []darwinUSBHostDeviceInfo{info}
fakeWatch.trigger()
require.Eventually(t, func() bool {
_, ok := server.snapshotExports()[busid]
return ok && darwinServerControlState(server, busid) == deviceStateAvailable
}, time.Second, 10*time.Millisecond)
watcher.Close()
require.True(t, fakeWatch.closed)
}
func TestDarwinServerBuildDeviceStateIncludesBusyExports(t *testing.T) {
t.Parallel()
@@ -249,3 +309,16 @@ func darwinServerControlState(server *ServerService, busid string) string {
defer server.controlAccess.Unlock()
return server.controlState[busid].State
}
type fakeDarwinUSBHostDeviceWatch struct {
callback func()
closed bool
}
func (w *fakeDarwinUSBHostDeviceWatch) Close() {
w.closed = true
}
func (w *fakeDarwinUSBHostDeviceWatch) trigger() {
w.callback()
}
service/usbip/server_linux.go
+86 -67
View File
@@ -31,14 +31,9 @@ type serverExport struct {
busid string
managed bool
originalDriver string
busy bool
}
const (
usbipExportReleaseTimeout = 10 * time.Second
usbipExportReleasePollInterval = 100 * time.Millisecond
serverReconcileBackstop = 30 * time.Second
)
type ServerService struct {
boxService.Adapter
ctx context.Context
@@ -118,7 +113,6 @@ func (s *ServerService) Start(stage adapter.StartStage) error {
s.access.Unlock()
go s.acceptLoop(tcpListener)
go s.ueventLoop()
go s.reconcileLoop()
return nil
}
@@ -267,12 +261,7 @@ func (s *ServerService) releaseExport(export serverExport, restore bool) error {
if err != nil && !os.IsNotExist(err) {
return err
}
if restore {
err = s.waitUSBIPStatusAvailable(export.busid, usbipExportReleaseTimeout)
if err != nil {
return err
}
}
s.setExportBusy(export.busid, false)
}
err := s.ops.hostUnbind(export.busid)
if err != nil && !os.IsNotExist(err) && !(isMissingUSBDeviceError(err) && !restore) {
@@ -301,29 +290,6 @@ func (s *ServerService) releaseExport(export serverExport, restore bool) error {
return nil
}
func (s *ServerService) waitUSBIPStatusAvailable(busid string, timeout time.Duration) error {
deadline := time.Now().Add(timeout)
for {
status, err := s.ops.readUsbipStatus(busid)
if err != nil {
if os.IsNotExist(err) || isMissingUSBDeviceError(err) {
return nil
}
} else if status == usbipStatusAvailable {
return nil
}
if time.Now().After(deadline) {
if err != nil {
return E.Cause(err, "wait for ", busid, " usbip status available")
}
return E.New("timed out waiting for ", busid, " usbip status available")
}
if !sleepCtx(s.ctx, usbipExportReleasePollInterval) {
return s.ctx.Err()
}
}
}
func (s *ServerService) rollbackExports() {
exports := s.snapshotExports()
for _, export := range exports {
@@ -356,14 +322,32 @@ func (s *ServerService) reconcileAndBroadcast(notify bool) error {
return nil
}
func (s *ServerService) currentExports() []string {
func (s *ServerService) currentExports() []serverExport {
s.access.Lock()
defer s.access.Unlock()
out := make([]string, 0, len(s.exports))
for busid := range s.exports {
out = append(out, busid)
out := make([]serverExport, 0, len(s.exports))
for _, export := range s.exports {
if export.busy {
continue
}
out = append(out, export)
}
slices.Sort(out)
slices.SortFunc(out, func(left, right serverExport) int {
return strings.Compare(left.busid, right.busid)
})
return out
}
func (s *ServerService) allExports() []serverExport {
s.access.Lock()
defer s.access.Unlock()
out := make([]serverExport, 0, len(s.exports))
for _, export := range s.exports {
out = append(out, export)
}
slices.SortFunc(out, func(left, right serverExport) int {
return strings.Compare(left.busid, right.busid)
})
return out
}
@@ -383,6 +367,25 @@ func (s *ServerService) setExport(export serverExport) {
s.exports[export.busid] = export
}
func (s *ServerService) getExport(busid string) (serverExport, bool) {
s.access.Lock()
defer s.access.Unlock()
export, ok := s.exports[busid]
return export, ok
}
func (s *ServerService) setExportBusy(busid string, busy bool) bool {
s.access.Lock()
defer s.access.Unlock()
export, ok := s.exports[busid]
if !ok || export.busy == busy {
return false
}
export.busy = busy
s.exports[busid] = export
return true
}
func (s *ServerService) deleteExport(busid string) {
s.access.Lock()
defer s.access.Unlock()
@@ -467,12 +470,13 @@ func (s *ServerService) handleDevList(conn net.Conn) {
}
func (s *ServerService) buildDevListEntries() []DeviceEntry {
busids := s.currentExports()
if len(busids) == 0 {
exports := s.currentExports()
if len(exports) == 0 {
return nil
}
entries := make([]DeviceEntry, 0, len(busids))
for _, busid := range busids {
entries := make([]DeviceEntry, 0, len(exports))
for _, export := range exports {
busid := export.busid
status, err := s.ops.readUsbipStatus(busid)
if err != nil {
s.logger.Debug("status ", busid, ": ", err)
@@ -550,6 +554,8 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended
_ = writeReply(conn, OpStatusError, nil)
return false
}
s.setExportBusy(busid, true)
s.broadcastChanged()
err = handoff.closeKernelFD()
if err != nil {
s.logger.Debug("close kernel fd ", busid, ": ", err)
@@ -559,10 +565,14 @@ func (s *ServerService) handleImportBusID(conn net.Conn, busid string, extended
if err != nil {
s.logger.Warn("reply import ", busid, ": ", err)
_ = s.ops.writeUsbipSockfd(busid, -1)
s.setExportBusy(busid, false)
s.broadcastChanged()
return false
}
s.logger.Info("attached ", busid, " to remote ", conn.RemoteAddr())
return handoff.startRelay(s.ctx, s.logger, "server", busid)
done := handoff.startRelay(s.ctx, s.logger, "server", busid)
go s.waitImportDone(busid, done)
return true
}
func (s *ServerService) isExported(busid string) bool {
@@ -572,6 +582,20 @@ func (s *ServerService) isExported(busid string) bool {
return ok
}
func (s *ServerService) waitImportDone(busid string, done <-chan struct{}) {
select {
case <-s.ctx.Done():
return
case <-done:
}
err := s.ops.writeUsbipSockfd(busid, -1)
if err != nil && !os.IsNotExist(err) && !isMissingUSBDeviceError(err) {
s.logger.Debug("release ", busid, " from usbip-host: ", err)
}
s.setExportBusy(busid, false)
s.broadcastChanged()
}
func (s *ServerService) ueventLoop() {
for {
listener, err := s.ops.newUEventListener()
@@ -615,31 +639,18 @@ func (s *ServerService) ueventLoop() {
}
}
func (s *ServerService) reconcileLoop() {
ticker := time.NewTicker(serverReconcileBackstop)
defer ticker.Stop()
for {
select {
case <-s.ctx.Done():
return
case <-ticker.C:
}
err := s.reconcileAndBroadcast(true)
if err != nil {
s.logger.Warn("reconcile exports: ", err)
}
}
func (s *ServerService) broadcastChanged() {
s.broadcastControlState(deviceInfoV2Map(s.buildDeviceStateV2()), false)
}
func (s *ServerService) buildDeviceStateV2() []DeviceInfoV2 {
busids := s.currentExports()
if len(busids) == 0 {
exports := s.allExports()
if len(exports) == 0 {
return nil
}
devices := make([]DeviceInfoV2, 0, len(busids))
for _, busid := range busids {
devices := make([]DeviceInfoV2, 0, len(exports))
for _, export := range exports {
busid := export.busid
status, statusErr := s.ops.readUsbipStatus(busid)
dev, devErr := s.ops.readSysfsDevice(busid, sysBusDevicePath(busid))
if devErr != nil {
@@ -657,6 +668,10 @@ func (s *ServerService) buildDeviceStateV2() []DeviceInfoV2 {
if statusErr != nil {
state = deviceStateUnavailable
reason = statusErr.Error()
} else if export.busy {
status = usbipStatusUsed
state = deviceStateBusy
reason = linuxUSBIPStatusReason(status)
}
entry := dev.toDeviceEntry()
devices = append(devices, deviceInfoV2FromEntry(entry, backendIDLinuxSysfs, linuxStableID(dev), state, status, reason))
@@ -665,9 +680,13 @@ func (s *ServerService) buildDeviceStateV2() []DeviceInfoV2 {
}
func (s *ServerService) leaseAvailable(busid string) (bool, string) {
if !s.isExported(busid) {
export, ok := s.getExport(busid)
if !ok {
return false, "unknown busid"
}
if export.busy {
return false, linuxUSBIPStatusReason(usbipStatusUsed)
}
status, err := s.ops.readUsbipStatus(busid)
if err != nil {
return false, err.Error()
service/usbip/usbhost_darwin.go
+35
View File
@@ -130,6 +130,31 @@ func darwinOpenUSBHostDevice(registryID uint64, capture bool) (*darwinUSBHostDev
}, nil
}
type darwinUSBHostDeviceWatcher struct {
handle *C.box_usbhost_device_watcher_t
ref cgo.Handle
}
func darwinWatchUSBHostDevices(callback func()) (darwinUSBHostDeviceWatch, error) {
ref := cgo.NewHandle(callback)
var errorPtr *C.char
handle := C.box_usbhost_device_watcher_create(C.uintptr_t(ref), &errorPtr)
if handle == nil {
ref.Delete()
return nil, darwinCError(errorPtr)
}
return &darwinUSBHostDeviceWatcher{handle: handle, ref: ref}, nil
}
func (w *darwinUSBHostDeviceWatcher) Close() {
if w == nil || w.handle == nil {
return
}
C.box_usbhost_device_watcher_destroy(w.handle)
w.handle = nil
w.ref.Delete()
}
func darwinCreateUSBHostController(controller *darwinVirtualController, portCount uint8, speed uint32) (*darwinUSBHostController, error) {
ref := cgo.NewHandle(controller)
var errorPtr *C.char
@@ -357,6 +382,16 @@ func box_usbip_darwin_controller_doorbell(ref C.uintptr_t, doorbell C.uint32_t)
controller.enqueueDoorbell(uint32(doorbell))
}
//export box_usbip_darwin_usb_event
func box_usbip_darwin_usb_event(ref C.uintptr_t) {
handle := cgo.Handle(ref)
callback, ok := handle.Value().(func())
if !ok {
return
}
callback()
}
func (d *darwinUSBHostDevice) Close() {
if d == nil || d.handle == nil {
return
service/usbip/usbhost_darwin.h
+4
View File
@@ -47,12 +47,15 @@ typedef struct box_usbhost_iso_packet {
} box_usbhost_iso_packet_t;
typedef struct box_usbhost_device box_usbhost_device_t;
typedef struct box_usbhost_device_watcher box_usbhost_device_watcher_t;
typedef struct box_usbhost_controller box_usbhost_controller_t;
typedef struct box_usbhost_device_sm box_usbhost_device_sm_t;
typedef struct box_usbhost_endpoint_sm box_usbhost_endpoint_sm_t;
bool box_usbhost_copy_devices(box_usbhost_device_list_t *out, char **error_out);
void box_usbhost_device_list_free(box_usbhost_device_list_t *list);
box_usbhost_device_watcher_t *box_usbhost_device_watcher_create(uintptr_t ref, char **error_out);
void box_usbhost_device_watcher_destroy(box_usbhost_device_watcher_t *watcher);
box_usbhost_device_t *box_usbhost_device_open(uint64_t registry_id, bool capture, box_usbhost_device_info_t *info_out, char **error_out);
void box_usbhost_device_close(box_usbhost_device_t *device);
@@ -97,3 +100,4 @@ void box_usbhost_free_error(char *error);
extern void box_usbip_darwin_controller_command(uintptr_t ref, IOUSBHostCIMessage message);
extern void box_usbip_darwin_controller_doorbell(uintptr_t ref, uint32_t doorbell);
extern void box_usbip_darwin_usb_event(uintptr_t ref);
service/usbip/usbhost_darwin.m
+99 -8
View File
@@ -3,6 +3,7 @@
#import <Foundation/Foundation.h>
#import <IOKit/IOKitLib.h>
#import <IOUSBHost/AppleUSBDescriptorParsing.h>
#import <dispatch/dispatch.h>
#import <mach/mach_time.h>
#import <stdlib.h>
#import <string.h>
@@ -28,6 +29,13 @@ struct box_usbhost_device {
void *object;
};
struct box_usbhost_device_watcher {
IONotificationPortRef port;
io_iterator_t matched;
io_iterator_t terminated;
uintptr_t ref;
};
struct box_usbhost_controller {
void *object;
};
@@ -77,6 +85,29 @@ void box_usbhost_free_error(char *error) {
free(error);
}
static CFMutableDictionaryRef box_usbhost_device_matching_dictionary(void) {
return [IOUSBHostDevice createMatchingDictionaryWithVendorID:nil
productID:nil
bcdDevice:nil
deviceClass:nil
deviceSubclass:nil
deviceProtocol:nil
speed:nil
productIDArray:nil];
}
static void box_usbhost_device_watcher_drain(io_iterator_t iterator) {
io_service_t service = IO_OBJECT_NULL;
while ((service = IOIteratorNext(iterator)) != IO_OBJECT_NULL) {
IOObjectRelease(service);
}
}
static void box_usbhost_device_watcher_callback(void *refcon, io_iterator_t iterator) {
box_usbhost_device_watcher_drain(iterator);
box_usbip_darwin_usb_event((uintptr_t)refcon);
}
static uint32_t box_number_property(io_service_t service, NSString *key) {
CFTypeRef value = IORegistryEntryCreateCFProperty(service, (__bridge CFStringRef)key, kCFAllocatorDefault, 0);
if (value == NULL) {
@@ -274,14 +305,7 @@ bool box_usbhost_copy_devices(box_usbhost_device_list_t *out, char **error_out)
}
memset(out, 0, sizeof(*out));
@autoreleasepool {
CFMutableDictionaryRef matching = [IOUSBHostDevice createMatchingDictionaryWithVendorID:nil
productID:nil
bcdDevice:nil
deviceClass:nil
deviceSubclass:nil
deviceProtocol:nil
speed:nil
productIDArray:nil];
CFMutableDictionaryRef matching = box_usbhost_device_matching_dictionary();
io_iterator_t iterator = IO_OBJECT_NULL;
kern_return_t kr = IOServiceGetMatchingServices(kIOMainPortDefault, matching, &iterator);
if (kr != KERN_SUCCESS) {
@@ -320,6 +344,73 @@ void box_usbhost_device_list_free(box_usbhost_device_list_t *list) {
list->count = 0;
}
box_usbhost_device_watcher_t *box_usbhost_device_watcher_create(uintptr_t ref, char **error_out) {
@autoreleasepool {
box_usbhost_device_watcher_t *watcher = calloc(1, sizeof(*watcher));
if (watcher == NULL) {
box_set_error_string(error_out, @"IOUSBHost watcher: allocate watcher");
return NULL;
}
watcher->ref = ref;
watcher->port = IONotificationPortCreate(kIOMainPortDefault);
if (watcher->port == NULL) {
box_set_error_string(error_out, @"IONotificationPortCreate(IOUSBHostDevice)");
box_usbhost_device_watcher_destroy(watcher);
return NULL;
}
dispatch_queue_t queue = dispatch_queue_create("io.nekohasekai.sing-box.usbhost-watch", DISPATCH_QUEUE_SERIAL);
IONotificationPortSetDispatchQueue(watcher->port, queue);
CFMutableDictionaryRef matching = box_usbhost_device_matching_dictionary();
kern_return_t kr = IOServiceAddMatchingNotification(watcher->port,
kIOFirstMatchNotification,
matching,
box_usbhost_device_watcher_callback,
(void *)ref,
&watcher->matched);
if (kr != KERN_SUCCESS) {
box_set_error_string(error_out, [NSString stringWithFormat:@"IOServiceAddMatchingNotification(first match IOUSBHostDevice): 0x%x", kr]);
box_usbhost_device_watcher_destroy(watcher);
return NULL;
}
box_usbhost_device_watcher_drain(watcher->matched);
matching = box_usbhost_device_matching_dictionary();
kr = IOServiceAddMatchingNotification(watcher->port,
kIOTerminatedNotification,
matching,
box_usbhost_device_watcher_callback,
(void *)ref,
&watcher->terminated);
if (kr != KERN_SUCCESS) {
box_set_error_string(error_out, [NSString stringWithFormat:@"IOServiceAddMatchingNotification(terminated IOUSBHostDevice): 0x%x", kr]);
box_usbhost_device_watcher_destroy(watcher);
return NULL;
}
box_usbhost_device_watcher_drain(watcher->terminated);
return watcher;
}
}
void box_usbhost_device_watcher_destroy(box_usbhost_device_watcher_t *watcher) {
if (watcher == NULL) {
return;
}
if (watcher->matched != IO_OBJECT_NULL) {
IOObjectRelease(watcher->matched);
watcher->matched = IO_OBJECT_NULL;
}
if (watcher->terminated != IO_OBJECT_NULL) {
IOObjectRelease(watcher->terminated);
watcher->terminated = IO_OBJECT_NULL;
}
if (watcher->port != NULL) {
IONotificationPortDestroy(watcher->port);
watcher->port = NULL;
}
free(watcher);
}
box_usbhost_device_t *box_usbhost_device_open(uint64_t registry_id, bool capture, box_usbhost_device_info_t *info_out, char **error_out) {
@autoreleasepool {
CFMutableDictionaryRef matching = IORegistryEntryIDMatching(registry_id);