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 <[email protected]>
LUCI-TryBot-Result: Go LUCI <[email protected]>

Author: neild

internal/quic/conn.go

@@ -20,13 +20,14 @@ import (
 // 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,28 +62,24 @@ type Conn struct {
 	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,
 		config:               config,
 		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 @@ func newConn(now time.Time, side connSide, initialConnID []byte, peerAddr netip.
 	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 @@ func (c *Conn) String() string {
 	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 @@ func (c *Conn) confirmHandshake(now time.Time) {
 	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 @@ func (c *Conn) receiveTransportParameters(p transportParameters) error {
 	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 @@ type (
 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 @@ func firstTime(a, b time.Time) time.Time {
 		return b
 	}
 }
-
-func (c *Conn) newConnIDFunc() newConnIDFunc {
-	if c.testHooks != nil {
-		return c.testHooks.newConnID
-	}
-	return newRandomConnID
-}

internal/quic/conn_id.go

@@ -55,10 +55,10 @@ type connID struct {
 	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,18 +70,20 @@ func (s *connIDState) initClient(newID newConnIDFunc) error {
 
 	// 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
 	}
 	s.remote = append(s.remote, connID{
 		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 @@ func (s *connIDState) initServer(newID newConnIDFunc, dstConnID []byte) error {
 
 	// 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 @@ func (s *connIDState) initServer(newID newConnIDFunc, dstConnID []byte) error {
 		cid: locid,
 	})
 	s.nextLocalSeq = 1
+	const retired = false
+	c.listener.connIDsChanged(c, retired, s.local[:])
 	return nil
 }
 
@@ -125,20 +129,21 @@ func (s *connIDState) dstConnID() (cid []byte, ok bool) {
 
 // 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 @@ func (s *connIDState) issueLocalIDs(newID newConnIDFunc) error {
 		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 @@ func (s *connIDState) handlePacket(side connSide, ptype packetType, srcConnID []
 				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 @@ func (s *connIDState) handlePacket(side connSide, ptype packetType, srcConnID []
 				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 @@ func (s *connIDState) retireRemote(rcid *connID) {
 	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 @@ func cloneBytes(b []byte) []byte {
 	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,