quic: send and receive UDP datagrams Add the Listener type, which manages a UDP socket. For golang/go#58547 Change-Id: Ia23a8b726ef46f8f84c9e052aa4dfc10eab034d6 Reviewed-on: https://go-review.googlesource.com/c/net/+/527758 Reviewed-by: Jonathan Amsterdam <jba@google.com> LUCI-TryBot-Result: Go LUCI <golang-scoped@luci-project-accounts.iam.gserviceaccount.com>
diff --git a/internal/quic/conn.go b/internal/quic/conn.go index c24e790..0063965 100644 --- a/internal/quic/conn.go +++ b/internal/quic/conn.go
@@ -20,13 +20,14 @@ // Multiple goroutines may invoke methods on a Conn simultaneously. type Conn struct { side connSide - listener connListener + listener *Listener config *Config testHooks connTestHooks peerAddr netip.AddrPort msgc chan any donec chan struct{} // closed when conn loop exits + readyc chan struct{} // closed when TLS handshake completes exited bool // set to make the conn loop exit immediately w packetWriter @@ -61,21 +62,16 @@ testSendPing sentVal } -// The connListener is the Conn's Listener. -// Defined as an interface so we can swap it out in tests. -type connListener interface { - sendDatagram(p []byte, addr netip.AddrPort) error -} - // connTestHooks override conn behavior in tests. type connTestHooks interface { nextMessage(msgc chan any, nextTimeout time.Time) (now time.Time, message any) handleTLSEvent(tls.QUICEvent) newConnID(seq int64) ([]byte, error) waitUntil(ctx context.Context, until func() bool) error + timeNow() time.Time } -func newConn(now time.Time, side connSide, initialConnID []byte, peerAddr netip.AddrPort, config *Config, l connListener, hooks connTestHooks) (*Conn, error) { +func newConn(now time.Time, side connSide, initialConnID []byte, peerAddr netip.AddrPort, config *Config, l *Listener, hooks connTestHooks) (*Conn, error) { c := &Conn{ side: side, listener: l, @@ -83,6 +79,7 @@ peerAddr: peerAddr, msgc: make(chan any, 1), donec: make(chan struct{}), + readyc: make(chan struct{}), testHooks: hooks, maxIdleTimeout: defaultMaxIdleTimeout, idleTimeout: now.Add(defaultMaxIdleTimeout), @@ -94,12 +91,12 @@ c.msgc = make(chan any, 1) if c.side == clientSide { - if err := c.connIDState.initClient(c.newConnIDFunc()); err != nil { + if err := c.connIDState.initClient(c); err != nil { return nil, err } initialConnID, _ = c.connIDState.dstConnID() } else { - if err := c.connIDState.initServer(c.newConnIDFunc(), initialConnID); err != nil { + if err := c.connIDState.initServer(c, initialConnID); err != nil { return nil, err } } @@ -134,6 +131,14 @@ return fmt.Sprintf("quic.Conn(%v,->%v)", c.side, c.peerAddr) } +func (c *Conn) Close() error { + // TODO: Implement shutdown for real. + c.runOnLoop(func(now time.Time, c *Conn) { + c.exited = true + }) + return nil +} + // confirmHandshake is called when the handshake is confirmed. // https://www.rfc-editor.org/rfc/rfc9001#section-4.1.2 func (c *Conn) confirmHandshake(now time.Time) { @@ -147,6 +152,7 @@ if c.side == serverSide { // When the server confirms the handshake, it sends a HANDSHAKE_DONE. c.handshakeConfirmed.setUnsent() + c.listener.serverConnEstablished(c) } else { // The client never sends a HANDSHAKE_DONE, so we set handshakeConfirmed // to the received state, indicating that the handshake is confirmed and we @@ -177,7 +183,7 @@ c.streams.peerInitialMaxStreamDataRemote[uniStream] = p.initialMaxStreamDataUni c.peerAckDelayExponent = p.ackDelayExponent c.loss.setMaxAckDelay(p.maxAckDelay) - if err := c.connIDState.setPeerActiveConnIDLimit(p.activeConnIDLimit, c.newConnIDFunc()); err != nil { + if err := c.connIDState.setPeerActiveConnIDLimit(c, p.activeConnIDLimit); err != nil { return err } if p.preferredAddrConnID != nil { @@ -211,6 +217,7 @@ func (c *Conn) loop(now time.Time) { defer close(c.donec) defer c.tls.Close() + defer c.listener.connDrained(c) // The connection timer sends a message to the connection loop on expiry. // We need to give it an expiry when creating it, so set the initial timeout to @@ -371,10 +378,3 @@ return b } } - -func (c *Conn) newConnIDFunc() newConnIDFunc { - if c.testHooks != nil { - return c.testHooks.newConnID - } - return newRandomConnID -}
diff --git a/internal/quic/conn_id.go b/internal/quic/conn_id.go index 561dea2..eb2f3ec 100644 --- a/internal/quic/conn_id.go +++ b/internal/quic/conn_id.go
@@ -55,10 +55,10 @@ send sentVal } -func (s *connIDState) initClient(newID newConnIDFunc) error { +func (s *connIDState) initClient(c *Conn) error { // Client chooses its initial connection ID, and sends it // in the Source Connection ID field of the first Initial packet. - locid, err := newID(0) + locid, err := c.newConnID(0) if err != nil { return err } @@ -70,7 +70,7 @@ // Client chooses an initial, transient connection ID for the server, // and sends it in the Destination Connection ID field of the first Initial packet. - remid, err := newID(-1) + remid, err := c.newConnID(-1) if err != nil { return err } @@ -78,10 +78,12 @@ seq: -1, cid: remid, }) + const retired = false + c.listener.connIDsChanged(c, retired, s.local[:]) return nil } -func (s *connIDState) initServer(newID newConnIDFunc, dstConnID []byte) error { +func (s *connIDState) initServer(c *Conn, dstConnID []byte) error { // Client-chosen, transient connection ID received in the first Initial packet. // The server will not use this as the Source Connection ID of packets it sends, // but remembers it because it may receive packets sent to this destination. @@ -92,7 +94,7 @@ // Server chooses a connection ID, and sends it in the Source Connection ID of // the response to the clent. - locid, err := newID(0) + locid, err := c.newConnID(0) if err != nil { return err } @@ -101,6 +103,8 @@ cid: locid, }) s.nextLocalSeq = 1 + const retired = false + c.listener.connIDsChanged(c, retired, s.local[:]) return nil } @@ -125,20 +129,21 @@ // setPeerActiveConnIDLimit sets the active_connection_id_limit // transport parameter received from the peer. -func (s *connIDState) setPeerActiveConnIDLimit(lim int64, newID newConnIDFunc) error { +func (s *connIDState) setPeerActiveConnIDLimit(c *Conn, lim int64) error { s.peerActiveConnIDLimit = lim - return s.issueLocalIDs(newID) + return s.issueLocalIDs(c) } -func (s *connIDState) issueLocalIDs(newID newConnIDFunc) error { +func (s *connIDState) issueLocalIDs(c *Conn) error { toIssue := min(int(s.peerActiveConnIDLimit), maxPeerActiveConnIDLimit) for i := range s.local { if s.local[i].seq != -1 && !s.local[i].retired { toIssue-- } } + prev := len(s.local) for toIssue > 0 { - cid, err := newID(s.nextLocalSeq) + cid, err := c.newConnID(s.nextLocalSeq) if err != nil { return err } @@ -151,14 +156,16 @@ s.needSend = true toIssue-- } + const retired = false + c.listener.connIDsChanged(c, retired, s.local[prev:]) return nil } // handlePacket updates the connection ID state during the handshake // (Initial and Handshake packets). -func (s *connIDState) handlePacket(side connSide, ptype packetType, srcConnID []byte) { +func (s *connIDState) handlePacket(c *Conn, ptype packetType, srcConnID []byte) { switch { - case ptype == packetTypeInitial && side == clientSide: + case ptype == packetTypeInitial && c.side == clientSide: if len(s.remote) == 1 && s.remote[0].seq == -1 { // We're a client connection processing the first Initial packet // from the server. Replace the transient remote connection ID @@ -168,7 +175,7 @@ cid: cloneBytes(srcConnID), } } - case ptype == packetTypeInitial && side == serverSide: + case ptype == packetTypeInitial && c.side == serverSide: if len(s.remote) == 0 { // We're a server connection processing the first Initial packet // from the client. Set the client's connection ID. @@ -177,11 +184,13 @@ cid: cloneBytes(srcConnID), }) } - case ptype == packetTypeHandshake && side == serverSide: + case ptype == packetTypeHandshake && c.side == serverSide: if len(s.local) > 0 && s.local[0].seq == -1 { // We're a server connection processing the first Handshake packet from // the client. Discard the transient, client-chosen connection ID used // for Initial packets; the client will never send it again. + const retired = true + c.listener.connIDsChanged(c, retired, s.local[0:1]) s.local = append(s.local[:0], s.local[1:]...) } } @@ -263,17 +272,19 @@ s.needSend = true } -func (s *connIDState) handleRetireConnID(seq int64, newID newConnIDFunc) error { +func (s *connIDState) handleRetireConnID(c *Conn, seq int64) error { if seq >= s.nextLocalSeq { return localTransportError(errProtocolViolation) } for i := range s.local { if s.local[i].seq == seq { + const retired = true + c.listener.connIDsChanged(c, retired, s.local[i:i+1]) s.local = append(s.local[:i], s.local[i+1:]...) break } } - s.issueLocalIDs(newID) + s.issueLocalIDs(c) return nil } @@ -355,7 +366,12 @@ return n } -type newConnIDFunc func(seq int64) ([]byte, error) +func (c *Conn) newConnID(seq int64) ([]byte, error) { + if c.testHooks != nil { + return c.testHooks.newConnID(seq) + } + return newRandomConnID(seq) +} func newRandomConnID(_ int64) ([]byte, error) { // It is not necessary for connection IDs to be cryptographically secure,
diff --git a/internal/quic/conn_id_test.go b/internal/quic/conn_id_test.go index d479cd4..c528958 100644 --- a/internal/quic/conn_id_test.go +++ b/internal/quic/conn_id_test.go
@@ -11,100 +11,135 @@ "crypto/tls" "fmt" "net/netip" - "reflect" + "strings" "testing" ) func TestConnIDClientHandshake(t *testing.T) { + tc := newTestConn(t, clientSide) // On initialization, the client chooses local and remote IDs. // // The order in which we allocate the two isn't actually important, // but test is a lot simpler if we assume. - var s connIDState - s.initClient(newConnIDSequence()) - if got, want := string(s.srcConnID()), "local-1"; got != want { - t.Errorf("after initClient: srcConnID = %q, want %q", got, want) + if got, want := tc.conn.connIDState.srcConnID(), testLocalConnID(0); !bytes.Equal(got, want) { + t.Errorf("after initialization: srcConnID = %x, want %x", got, want) } - dstConnID, _ := s.dstConnID() - if got, want := string(dstConnID), "local-2"; got != want { - t.Errorf("after initClient: dstConnID = %q, want %q", got, want) + dstConnID, _ := tc.conn.connIDState.dstConnID() + if got, want := dstConnID, testLocalConnID(-1); !bytes.Equal(got, want) { + t.Errorf("after initialization: dstConnID = %x, want %x", got, want) } // The server's first Initial packet provides the client with a // non-transient remote connection ID. - s.handlePacket(clientSide, packetTypeInitial, []byte("remote-1")) - dstConnID, _ = s.dstConnID() - if got, want := string(dstConnID), "remote-1"; got != want { - t.Errorf("after receiving Initial: dstConnID = %q, want %q", got, want) + tc.writeFrames(packetTypeInitial, + debugFrameCrypto{ + data: tc.cryptoDataIn[tls.QUICEncryptionLevelInitial], + }) + dstConnID, _ = tc.conn.connIDState.dstConnID() + if got, want := dstConnID, testPeerConnID(0); !bytes.Equal(got, want) { + t.Errorf("after receiving Initial: dstConnID = %x, want %x", got, want) } wantLocal := []connID{{ - cid: []byte("local-1"), + cid: testLocalConnID(0), seq: 0, }} - if !reflect.DeepEqual(s.local, wantLocal) { - t.Errorf("local ids: %v, want %v", s.local, wantLocal) + if got := tc.conn.connIDState.local; !connIDListEqual(got, wantLocal) { + t.Errorf("local ids: %v, want %v", fmtConnIDList(got), fmtConnIDList(wantLocal)) } wantRemote := []connID{{ - cid: []byte("remote-1"), + cid: testPeerConnID(0), seq: 0, }} - if !reflect.DeepEqual(s.remote, wantRemote) { - t.Errorf("remote ids: %v, want %v", s.remote, wantRemote) + if got := tc.conn.connIDState.remote; !connIDListEqual(got, wantRemote) { + t.Errorf("remote ids: %v, want %v", fmtConnIDList(got), fmtConnIDList(wantRemote)) } } func TestConnIDServerHandshake(t *testing.T) { + tc := newTestConn(t, serverSide) // On initialization, the server is provided with the client-chosen // transient connection ID, and allocates an ID of its own. // The Initial packet sets the remote connection ID. - var s connIDState - s.initServer(newConnIDSequence(), []byte("transient")) - s.handlePacket(serverSide, packetTypeInitial, []byte("remote-1")) - if got, want := string(s.srcConnID()), "local-1"; got != want { + tc.writeFrames(packetTypeInitial, + debugFrameCrypto{ + data: tc.cryptoDataIn[tls.QUICEncryptionLevelInitial][:1], + }) + if got, want := tc.conn.connIDState.srcConnID(), testLocalConnID(0); !bytes.Equal(got, want) { t.Errorf("after initClient: srcConnID = %q, want %q", got, want) } - dstConnID, _ := s.dstConnID() - if got, want := string(dstConnID), "remote-1"; got != want { + dstConnID, _ := tc.conn.connIDState.dstConnID() + if got, want := dstConnID, testPeerConnID(0); !bytes.Equal(got, want) { t.Errorf("after initClient: dstConnID = %q, want %q", got, want) } + // The Initial flight of CRYPTO data includes transport parameters, + // which cause us to allocate another local connection ID. + tc.writeFrames(packetTypeInitial, + debugFrameCrypto{ + off: 1, + data: tc.cryptoDataIn[tls.QUICEncryptionLevelInitial][1:], + }) wantLocal := []connID{{ - cid: []byte("transient"), + cid: testPeerConnID(-1), seq: -1, }, { - cid: []byte("local-1"), + cid: testLocalConnID(0), seq: 0, + }, { + cid: testLocalConnID(1), + seq: 1, }} - if !reflect.DeepEqual(s.local, wantLocal) { - t.Errorf("local ids: %v, want %v", s.local, wantLocal) + if got := tc.conn.connIDState.local; !connIDListEqual(got, wantLocal) { + t.Errorf("local ids: %v, want %v", fmtConnIDList(got), fmtConnIDList(wantLocal)) } wantRemote := []connID{{ - cid: []byte("remote-1"), + cid: testPeerConnID(0), seq: 0, }} - if !reflect.DeepEqual(s.remote, wantRemote) { - t.Errorf("remote ids: %v, want %v", s.remote, wantRemote) + if got := tc.conn.connIDState.remote; !connIDListEqual(got, wantRemote) { + t.Errorf("remote ids: %v, want %v", fmtConnIDList(got), fmtConnIDList(wantRemote)) } // The client's first Handshake packet permits the server to discard the // transient connection ID. - s.handlePacket(serverSide, packetTypeHandshake, []byte("remote-1")) + tc.writeFrames(packetTypeHandshake, + debugFrameCrypto{ + data: tc.cryptoDataIn[tls.QUICEncryptionLevelHandshake], + }) wantLocal = []connID{{ - cid: []byte("local-1"), + cid: testLocalConnID(0), seq: 0, + }, { + cid: testLocalConnID(1), + seq: 1, }} - if !reflect.DeepEqual(s.local, wantLocal) { - t.Errorf("after handshake local ids: %v, want %v", s.local, wantLocal) + if got := tc.conn.connIDState.local; !connIDListEqual(got, wantLocal) { + t.Errorf("local ids: %v, want %v", fmtConnIDList(got), fmtConnIDList(wantLocal)) } } -func newConnIDSequence() newConnIDFunc { - var n uint64 - return func(_ int64) ([]byte, error) { - n++ - return []byte(fmt.Sprintf("local-%v", n)), nil +func connIDListEqual(a, b []connID) bool { + if len(a) != len(b) { + return false } + for i := range a { + if a[i].seq != b[i].seq { + return false + } + if !bytes.Equal(a[i].cid, b[i].cid) { + return false + } + } + return true +} + +func fmtConnIDList(s []connID) string { + var strs []string + for _, cid := range s { + strs = append(strs, fmt.Sprintf("[seq:%v cid:{%x}]", cid.seq, cid.cid)) + } + return "{" + strings.Join(strs, " ") + "}" } func TestNewRandomConnID(t *testing.T) {
diff --git a/internal/quic/conn_recv.go b/internal/quic/conn_recv.go index 07f17e3..b866d8a 100644 --- a/internal/quic/conn_recv.go +++ b/internal/quic/conn_recv.go
@@ -63,7 +63,7 @@ if logPackets { logInboundLongPacket(c, p) } - c.connIDState.handlePacket(c.side, p.ptype, p.srcConnID) + c.connIDState.handlePacket(c, p.ptype, p.srcConnID) ackEliciting := c.handleFrames(now, ptype, space, p.payload) c.acks[space].receive(now, space, p.num, ackEliciting) if p.ptype == packetTypeHandshake && c.side == serverSide { @@ -377,7 +377,7 @@ if n < 0 { return -1 } - if err := c.connIDState.handleRetireConnID(seq, c.newConnIDFunc()); err != nil { + if err := c.connIDState.handleRetireConnID(c, seq); err != nil { c.abort(now, err) } return n
diff --git a/internal/quic/conn_test.go b/internal/quic/conn_test.go index ea720d5..cdbd466 100644 --- a/internal/quic/conn_test.go +++ b/internal/quic/conn_test.go
@@ -13,7 +13,9 @@ "errors" "flag" "fmt" + "io" "math" + "net" "net/netip" "reflect" "strings" @@ -105,6 +107,7 @@ type testConn struct { t *testing.T conn *Conn + listener *Listener now time.Time timer time.Time timerLastFired time.Time @@ -142,6 +145,8 @@ sentFrames []debugFrame lastPacket *testPacket + recvDatagram chan *datagram + // Transport parameters sent by the conn. sentTransportParameters *transportParameters @@ -173,6 +178,7 @@ }, cryptoDataOut: make(map[tls.QUICEncryptionLevel][]byte), cryptoDataIn: make(map[tls.QUICEncryptionLevel][]byte), + recvDatagram: make(chan *datagram), } t.Cleanup(tc.cleanup) @@ -196,12 +202,7 @@ var initialConnID []byte if side == serverSide { // The initial connection ID for the server is chosen by the client. - // When creating a server-side connection, pick a random connection ID here. - var err error - initialConnID, err = newRandomConnID(0) - if err != nil { - tc.t.Fatal(err) - } + initialConnID = testPeerConnID(-1) } peerQUICConfig := &tls.QUICConfig{TLSConfig: newTestTLSConfig(side.peer())} @@ -213,14 +214,12 @@ tc.peerTLSConn.SetTransportParameters(marshalTransportParameters(peerProvidedParams)) tc.peerTLSConn.Start(context.Background()) - conn, err := newConn( + tc.listener = newListener((*testConnUDPConn)(tc), config, (*testConnHooks)(tc)) + conn, err := tc.listener.newConn( tc.now, side, initialConnID, - netip.MustParseAddrPort("127.0.0.1:443"), - config, - (*testConnListener)(tc), - (*testConnHooks)(tc)) + netip.MustParseAddrPort("127.0.0.1:443")) if err != nil { tc.t.Fatal(err) } @@ -316,6 +315,7 @@ return } tc.conn.exit() + tc.listener.Close(context.Background()) } func (tc *testConn) logDatagram(text string, d *testDatagram) { @@ -844,6 +844,10 @@ return testLocalConnID(seq), nil } +func (tc *testConnHooks) timeNow() time.Time { + return tc.now +} + // testLocalConnID returns the connection ID with a given sequence number // used by a Conn under test. func testLocalConnID(seq int64) []byte { @@ -861,14 +865,31 @@ return []byte{0xbe, 0xee, 0xff, byte(seq)} } -// testConnListener implements connListener. -type testConnListener testConn +// testConnUDPConn implements UDPConn. +type testConnUDPConn testConn -func (tc *testConnListener) sendDatagram(p []byte, addr netip.AddrPort) error { - tc.sentDatagrams = append(tc.sentDatagrams, append([]byte(nil), p...)) +func (tc *testConnUDPConn) Close() error { + close(tc.recvDatagram) return nil } +func (tc *testConnUDPConn) LocalAddr() net.Addr { + return net.UDPAddrFromAddrPort(netip.MustParseAddrPort("127.0.0.1:443")) +} + +func (tc *testConnUDPConn) ReadMsgUDPAddrPort(b, control []byte) (n, controln, flags int, _ netip.AddrPort, _ error) { + for d := range tc.recvDatagram { + n = copy(b, d.b) + return n, 0, 0, d.addr, nil + } + return 0, 0, 0, netip.AddrPort{}, io.EOF +} + +func (tc *testConnUDPConn) WriteToUDPAddrPort(b []byte, addr netip.AddrPort) (int, error) { + tc.sentDatagrams = append(tc.sentDatagrams, append([]byte(nil), b...)) + return len(b), nil +} + // canceledContext returns a canceled Context. // // Functions which take a context preference progress over cancelation.
diff --git a/internal/quic/listener.go b/internal/quic/listener.go new file mode 100644 index 0000000..9869f6e --- /dev/null +++ b/internal/quic/listener.go
@@ -0,0 +1,280 @@ +// Copyright 2023 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +//go:build go1.21 + +package quic + +import ( + "context" + "errors" + "net" + "net/netip" + "sync" + "sync/atomic" + "time" +) + +// A Listener listens for QUIC traffic on a network address. +// It can accept inbound connections or create outbound ones. +// +// Multiple goroutines may invoke methods on a Listener simultaneously. +type Listener struct { + config *Config + udpConn udpConn + testHooks connTestHooks + + acceptQueue queue[*Conn] // new inbound connections + + connsMu sync.Mutex + conns map[*Conn]struct{} + closing bool // set when Close is called + closec chan struct{} // closed when the listen loop exits + + // The datagram receive loop keeps a mapping of connection IDs to conns. + // When a conn's connection IDs change, we add it to connIDUpdates and set + // connIDUpdateNeeded, and the receive loop updates its map. + connIDUpdateMu sync.Mutex + connIDUpdateNeeded atomic.Bool + connIDUpdates []connIDUpdate +} + +// A udpConn is a UDP connection. +// It is implemented by net.UDPConn. +type udpConn interface { + Close() error + LocalAddr() net.Addr + ReadMsgUDPAddrPort(b, control []byte) (n, controln, flags int, _ netip.AddrPort, _ error) + WriteToUDPAddrPort(b []byte, addr netip.AddrPort) (int, error) +} + +type connIDUpdate struct { + conn *Conn + retired bool + cid []byte +} + +// Listen listens on a local network address. +// The configuration config must be non-nil. +func Listen(network, address string, config *Config) (*Listener, error) { + if config.TLSConfig == nil { + return nil, errors.New("TLSConfig is not set") + } + a, err := net.ResolveUDPAddr(network, address) + if err != nil { + return nil, err + } + udpConn, err := net.ListenUDP(network, a) + if err != nil { + return nil, err + } + return newListener(udpConn, config, nil), nil +} + +func newListener(udpConn udpConn, config *Config, hooks connTestHooks) *Listener { + l := &Listener{ + config: config, + udpConn: udpConn, + testHooks: hooks, + conns: make(map[*Conn]struct{}), + acceptQueue: newQueue[*Conn](), + closec: make(chan struct{}), + } + go l.listen() + return l +} + +// LocalAddr returns the local network address. +func (l *Listener) LocalAddr() netip.AddrPort { + a, _ := l.udpConn.LocalAddr().(*net.UDPAddr) + return a.AddrPort() +} + +// Close closes the listener. +// Any blocked operations on the Listener or associated Conns and Stream will be unblocked +// and return errors. +// +// Close aborts every open connection. +// Data in stream read and write buffers is discarded. +// It waits for the peers of any open connection to acknowledge the connection has been closed. +func (l *Listener) Close(ctx context.Context) error { + l.acceptQueue.close(errors.New("listener closed")) + l.connsMu.Lock() + if !l.closing { + l.closing = true + for c := range l.conns { + c.Close() + } + if len(l.conns) == 0 { + l.udpConn.Close() + } + } + l.connsMu.Unlock() + select { + case <-l.closec: + case <-ctx.Done(): + l.connsMu.Lock() + for c := range l.conns { + c.exit() + } + l.connsMu.Unlock() + return ctx.Err() + } + return nil +} + +// Accept waits for and returns the next connection to the listener. +func (l *Listener) Accept(ctx context.Context) (*Conn, error) { + return l.acceptQueue.get(ctx, nil) +} + +// Dial creates and returns a connection to a network address. +func (l *Listener) Dial(ctx context.Context, network, address string) (*Conn, error) { + u, err := net.ResolveUDPAddr(network, address) + if err != nil { + return nil, err + } + addr := u.AddrPort() + addr = netip.AddrPortFrom(addr.Addr().Unmap(), addr.Port()) + c, err := l.newConn(time.Now(), clientSide, nil, addr) + if err != nil { + return nil, err + } + select { + case <-c.readyc: + case <-ctx.Done(): + c.Close() + return nil, ctx.Err() + } + return c, nil +} + +func (l *Listener) newConn(now time.Time, side connSide, initialConnID []byte, peerAddr netip.AddrPort) (*Conn, error) { + l.connsMu.Lock() + defer l.connsMu.Unlock() + if l.closing { + return nil, errors.New("listener closed") + } + c, err := newConn(now, side, initialConnID, peerAddr, l.config, l, l.testHooks) + if err != nil { + return nil, err + } + l.conns[c] = struct{}{} + return c, nil +} + +// serverConnEstablished is called by a conn when the handshake completes +// for an inbound (serverSide) connection. +func (l *Listener) serverConnEstablished(c *Conn) { + l.acceptQueue.put(c) +} + +// connDrained is called by a conn when it leaves the draining state, +// either when the peer acknowledges connection closure or the drain timeout expires. +func (l *Listener) connDrained(c *Conn) { + l.connsMu.Lock() + defer l.connsMu.Unlock() + delete(l.conns, c) + if l.closing && len(l.conns) == 0 { + l.udpConn.Close() + } +} + +// connIDsChanged is called by a conn when its connection IDs change. +func (l *Listener) connIDsChanged(c *Conn, retired bool, cids []connID) { + l.connIDUpdateMu.Lock() + defer l.connIDUpdateMu.Unlock() + for _, cid := range cids { + l.connIDUpdates = append(l.connIDUpdates, connIDUpdate{ + conn: c, + retired: retired, + cid: cid.cid, + }) + } + l.connIDUpdateNeeded.Store(true) +} + +// updateConnIDs is called by the datagram receive loop to update its connection ID map. +func (l *Listener) updateConnIDs(conns map[string]*Conn) { + l.connIDUpdateMu.Lock() + defer l.connIDUpdateMu.Unlock() + for i, u := range l.connIDUpdates { + if u.retired { + delete(conns, string(u.cid)) + } else { + conns[string(u.cid)] = u.conn + } + l.connIDUpdates[i] = connIDUpdate{} // drop refs + } + l.connIDUpdates = l.connIDUpdates[:0] + l.connIDUpdateNeeded.Store(false) +} + +func (l *Listener) listen() { + defer close(l.closec) + conns := map[string]*Conn{} + for { + m := newDatagram() + // TODO: Read and process the ECN (explicit congestion notification) field. + // https://tools.ietf.org/html/draft-ietf-quic-transport-32#section-13.4 + n, _, _, addr, err := l.udpConn.ReadMsgUDPAddrPort(m.b, nil) + if err != nil { + // The user has probably closed the listener. + // We currently don't surface errors from other causes; + // we could check to see if the listener has been closed and + // record the unexpected error if it has not. + return + } + if n == 0 { + continue + } + if l.connIDUpdateNeeded.Load() { + l.updateConnIDs(conns) + } + m.addr = addr + m.b = m.b[:n] + l.handleDatagram(m, conns) + } +} + +func (l *Listener) handleDatagram(m *datagram, conns map[string]*Conn) { + dstConnID, ok := dstConnIDForDatagram(m.b) + if !ok { + return + } + c := conns[string(dstConnID)] + if c == nil { + if getPacketType(m.b) != packetTypeInitial { + // This packet isn't trying to create a new connection. + // It might be associated with some connection we've lost state for. + // TODO: Send a stateless reset when appropriate. + // https://www.rfc-editor.org/rfc/rfc9000.html#section-10.3 + return + } + var now time.Time + if l.testHooks != nil { + now = l.testHooks.timeNow() + } else { + now = time.Now() + } + var err error + c, err = l.newConn(now, serverSide, dstConnID, m.addr) + if err != nil { + // The accept queue is probably full. + // We could send a CONNECTION_CLOSE to the peer to reject the connection. + // Currently, we just drop the datagram. + // https://www.rfc-editor.org/rfc/rfc9000.html#section-5.2.2-5 + return + } + } + + // TODO: This can block the listener while waiting for the conn to accept the dgram. + // Think about buffering between the receive loop and the conn. + c.sendMsg(m) +} + +func (l *Listener) sendDatagram(p []byte, addr netip.AddrPort) error { + _, err := l.udpConn.WriteToUDPAddrPort(p, addr) + return err +}
diff --git a/internal/quic/listener_test.go b/internal/quic/listener_test.go new file mode 100644 index 0000000..a6e0b34 --- /dev/null +++ b/internal/quic/listener_test.go
@@ -0,0 +1,88 @@ +// Copyright 2023 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +//go:build go1.21 + +package quic + +import ( + "bytes" + "context" + "io" + "testing" +) + +func TestConnect(t *testing.T) { + newLocalConnPair(t, &Config{}, &Config{}) +} + +func TestStreamTransfer(t *testing.T) { + ctx := context.Background() + cli, srv := newLocalConnPair(t, &Config{}, &Config{}) + data := makeTestData(1 << 20) + + srvdone := make(chan struct{}) + go func() { + defer close(srvdone) + s, err := srv.AcceptStream(ctx) + if err != nil { + t.Errorf("AcceptStream: %v", err) + return + } + b, err := io.ReadAll(s) + if err != nil { + t.Errorf("io.ReadAll(s): %v", err) + return + } + if !bytes.Equal(b, data) { + t.Errorf("read data mismatch (got %v bytes, want %v", len(b), len(data)) + } + if err := s.Close(); err != nil { + t.Errorf("s.Close() = %v", err) + } + }() + + s, err := cli.NewStream(ctx) + if err != nil { + t.Fatalf("NewStream: %v", err) + } + n, err := io.Copy(s, bytes.NewBuffer(data)) + if n != int64(len(data)) || err != nil { + t.Fatalf("io.Copy(s, data) = %v, %v; want %v, nil", n, err, len(data)) + } + if err := s.Close(); err != nil { + t.Fatalf("s.Close() = %v", err) + } +} + +func newLocalConnPair(t *testing.T, conf1, conf2 *Config) (clientConn, serverConn *Conn) { + t.Helper() + ctx := context.Background() + l1 := newLocalListener(t, serverSide, conf1) + l2 := newLocalListener(t, clientSide, conf2) + c2, err := l2.Dial(ctx, "udp", l1.LocalAddr().String()) + if err != nil { + t.Fatal(err) + } + c1, err := l1.Accept(ctx) + if err != nil { + t.Fatal(err) + } + return c2, c1 +} + +func newLocalListener(t *testing.T, side connSide, conf *Config) *Listener { + t.Helper() + if conf.TLSConfig == nil { + conf.TLSConfig = newTestTLSConfig(side) + } + l, err := Listen("udp", "127.0.0.1:0", conf) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + l.Close(context.Background()) + }) + return l +}
diff --git a/internal/quic/tls.go b/internal/quic/tls.go index 584316f..1d07f17 100644 --- a/internal/quic/tls.go +++ b/internal/quic/tls.go
@@ -73,6 +73,7 @@ // https://www.rfc-editor.org/rfc/rfc9001#section-4.1.2-1 c.confirmHandshake(now) } + close(c.readyc) case tls.QUICTransportParameters: params, err := unmarshalTransportParams(e.Data) if err != nil {