diff --git a/internal/server/usb/pending_test.go b/internal/server/usb/pending_test.go new file mode 100644 index 00000000..ac43a1f6 --- /dev/null +++ b/internal/server/usb/pending_test.go @@ -0,0 +1,97 @@ +package usb + +import ( + "context" + "sync" + "testing" +) + +// Exactly one party may reply for a given seqnum. USBIP_CMD_UNLINK and the async IN +// completion goroutines both race to claim it, and claimPending is what decides. + +func TestClaimPendingGrantsOwnershipToExactlyOneCaller(t *testing.T) { + var mu sync.Mutex + pending := map[uint32]context.CancelFunc{} + _, cancel := context.WithCancel(context.Background()) + defer cancel() + pending[7] = cancel + + if _, owned := claimPending(&mu, pending, 7); !owned { + t.Fatal("first claim must win") + } + if _, owned := claimPending(&mu, pending, 7); owned { + t.Fatal("second claim must lose: two replies for one seqnum is the bug") + } + if _, present := pending[7]; present { + t.Fatal("a claimed seqnum must be removed from the table") + } +} + +func TestClaimPendingReturnsTheCancelFuncToTheWinner(t *testing.T) { + // UNLINK needs the cancel func, and only when it actually won the claim - + // cancelling a request somebody else already completed is not harmless. + var mu sync.Mutex + pending := map[uint32]context.CancelFunc{} + cancelled := false + pending[3] = func() { cancelled = true } + + cancel, owned := claimPending(&mu, pending, 3) + if !owned || cancel == nil { + t.Fatal("winner must receive the cancel func") + } + cancel() + if !cancelled { + t.Fatal("the returned cancel func must be the stored one") + } + + if cancel, owned := claimPending(&mu, pending, 3); owned || cancel != nil { + t.Fatal("loser must receive neither ownership nor a cancel func") + } +} + +func TestClaimPendingOnUnknownSeqnum(t *testing.T) { + // A completion for a seqnum that was never registered, or an UNLINK for one that + // already completed, must both report "not mine" rather than panicking. + var mu sync.Mutex + pending := map[uint32]context.CancelFunc{} + if cancel, owned := claimPending(&mu, pending, 99); owned || cancel != nil { + t.Fatal("unknown seqnum must not be claimable") + } +} + +func TestClaimPendingIsRaceFree(t *testing.T) { + // The real interleaving: many goroutines contend for the same seqnum, as UNLINK + // and a completion do. Exactly one may come away with the right to reply. + // Meaningful under -race, which the repo's CI runs. + const contenders = 64 + for round := 0; round < 200; round++ { + var mu sync.Mutex + pending := map[uint32]context.CancelFunc{} + _, cancel := context.WithCancel(context.Background()) + pending[1] = cancel + + var wins int64 + var winsMu sync.Mutex + var start, done sync.WaitGroup + start.Add(1) + done.Add(contenders) + for i := 0; i < contenders; i++ { + go func() { + defer done.Done() + start.Wait() + if _, owned := claimPending(&mu, pending, 1); owned { + winsMu.Lock() + wins++ + winsMu.Unlock() + } + }() + } + start.Done() + done.Wait() + cancel() + + if wins != 1 { + t.Fatalf("round %d: %d winners, want exactly 1", round, wins) + } + } +} diff --git a/internal/server/usb/server.go b/internal/server/usb/server.go index deb6199c..a03e330d 100644 --- a/internal/server/usb/server.go +++ b/internal/server/usb/server.go @@ -193,6 +193,24 @@ const ( errConnReset = -104 // -ECONNRESET ) +// claimPending removes seq from the in-flight table and reports whether this caller +// was the one that removed it. +// +// Exactly one party may reply for a given seqnum. USBIP_CMD_UNLINK and the async IN +// completion goroutines race for that right, and the winner is whoever removes the +// entry: the loser must stay silent. Without this, an UNLINK arriving just before a +// completion takes the lock produces both a RET_UNLINK(-ECONNRESET) and a RET_SUBMIT +// for one request, leaving the client holding a completion for a URB it already +// unlinked. +func claimPending(mu *sync.Mutex, pending map[uint32]context.CancelFunc, + seq uint32) (context.CancelFunc, bool) { + mu.Lock() + defer mu.Unlock() + cancel, owned := pending[seq] + delete(pending, seq) + return cancel, owned +} + type Server struct { config *ServerConfig logger *slog.Logger @@ -805,12 +823,7 @@ func (s *Server) handleUrbStream(conn net.Conn, dev usb.Device) error { if cmd == usbip.CmdUnlinkCode { unlinkSeq := binary.BigEndian.Uint32(hdr[urbHdrOffsetUnlink : urbHdrOffsetUnlink+4]) s.logger.Debug("USBIP_CMD_UNLINK", "seq", seq, "unlink", unlinkSeq) - pendingMu.Lock() - cancel, found := pending[unlinkSeq] - if found { - delete(pending, unlinkSeq) - } - pendingMu.Unlock() + cancel, found := claimPending(&pendingMu, pending, unlinkSeq) // -ECONNRESET signals the URB was unlinked before completion; // status 0 means it already completed normally. status := int32(0) @@ -964,9 +977,10 @@ func (s *Server) handleUrbStream(conn net.Conn, dev usb.Device) error { // not stall or bunch the microphone sampling clock. signalNext(isoServiceEnd) - pendingMu.Lock() - delete(pending, seq) - pendingMu.Unlock() + if _, owned := claimPending(&pendingMu, pending, seq); !owned { + s.logger.Debug("URB ISO-IN completion suppressed; already unlinked", "seq", seq) + return + } if err := writeRet(seq, uint32(len(respData)), respData, completedPackets, iso, true); err != nil { if isClientDisconnect(err) { @@ -1010,9 +1024,10 @@ func (s *Server) handleUrbStream(conn net.Conn, dev usb.Device) error { break } - pendingMu.Lock() - delete(pending, seq) - pendingMu.Unlock() + if _, owned := claimPending(&pendingMu, pending, seq); !owned { + s.logger.Debug("URB completion suppressed; already unlinked", "seq", seq) + return + } completedPackets = completeIsoPackets(submitted, uint32(len(respData))) if err := writeRet(seq, uint32(len(respData)), respData, completedPackets, iso, true); err != nil {