quic: basic connection event loop
Add the Conn type, representing a QUIC connection.
A Conn's behavior is driven by an event loop goroutine.
This goroutine owns most Conn state. External events
(datagrams received, user operations such as writing to streams)
send events to the loop goroutine on a message channel.
The testConn type, used in tests, wraps a Conn and takes
control of its event loop. The testConn permits tests to
interact with a Conn synchronously, sending it events,
observing the result, and controlling the Conn's view
of time passing.
Add a very minimal implementation of connection idle timeouts
(RFC 9000, Section 10.1) to test the implementation of
synthetic time.
For golang/go#58547
Change-Id: Ic517e5e7bb019f4a677f892a807ca0417d6e19b1
Reviewed-on: https://go-review.googlesource.com/c/net/+/506678
TryBot-Result: Gopher Robot <gobot@golang.org>
Run-TryBot: Damien Neil <dneil@google.com>
Reviewed-by: Jonathan Amsterdam <jba@google.com>
diff --git a/internal/quic/conn.go b/internal/quic/conn.go
new file mode 100644
index 0000000..d6dbac1
--- /dev/null
+++ b/internal/quic/conn.go
@@ -0,0 +1,155 @@
+// 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 (
+ "errors"
+ "fmt"
+ "time"
+)
+
+// A Conn is a QUIC connection.
+//
+// Multiple goroutines may invoke methods on a Conn simultaneously.
+type Conn struct {
+ msgc chan any
+ donec chan struct{} // closed when conn loop exits
+ exited bool // set to make the conn loop exit immediately
+
+ testHooks connTestHooks
+
+ // idleTimeout is the time at which the connection will be closed due to inactivity.
+ // https://www.rfc-editor.org/rfc/rfc9000#section-10.1
+ maxIdleTimeout time.Duration
+ idleTimeout time.Time
+}
+
+// connTestHooks override conn behavior in tests.
+type connTestHooks interface {
+ nextMessage(msgc chan any, nextTimeout time.Time) (now time.Time, message any)
+}
+
+func newConn(now time.Time, hooks connTestHooks) (*Conn, error) {
+ c := &Conn{
+ donec: make(chan struct{}),
+ testHooks: hooks,
+ maxIdleTimeout: defaultMaxIdleTimeout,
+ idleTimeout: now.Add(defaultMaxIdleTimeout),
+ }
+
+ // A one-element buffer allows us to wake a Conn's event loop as a
+ // non-blocking operation.
+ c.msgc = make(chan any, 1)
+
+ go c.loop(now)
+ return c, nil
+}
+
+type timerEvent struct{}
+
+// loop is the connection main loop.
+//
+// Except where otherwise noted, all connection state is owned by the loop goroutine.
+//
+// The loop processes messages from c.msgc and timer events.
+// Other goroutines may examine or modify conn state by sending the loop funcs to execute.
+func (c *Conn) loop(now time.Time) {
+ defer close(c.donec)
+
+ // 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
+ // an arbitrary large value. The timer will be reset before this expires (and it
+ // isn't a problem if it does anyway). Skip creating the timer in tests which
+ // take control of the connection message loop.
+ var timer *time.Timer
+ var lastTimeout time.Time
+ hooks := c.testHooks
+ if hooks == nil {
+ timer = time.AfterFunc(1*time.Hour, func() {
+ c.sendMsg(timerEvent{})
+ })
+ defer timer.Stop()
+ }
+
+ for !c.exited {
+ nextTimeout := c.idleTimeout
+
+ var m any
+ if hooks != nil {
+ // Tests only: Wait for the test to tell us to continue.
+ now, m = hooks.nextMessage(c.msgc, nextTimeout)
+ } else if !nextTimeout.IsZero() && nextTimeout.Before(now) {
+ // A connection timer has expired.
+ now = time.Now()
+ m = timerEvent{}
+ } else {
+ // Reschedule the connection timer if necessary
+ // and wait for the next event.
+ if !nextTimeout.Equal(lastTimeout) && !nextTimeout.IsZero() {
+ // Resetting a timer created with time.AfterFunc guarantees
+ // that the timer will run again. We might generate a spurious
+ // timer event under some circumstances, but that's okay.
+ timer.Reset(nextTimeout.Sub(now))
+ lastTimeout = nextTimeout
+ }
+ m = <-c.msgc
+ now = time.Now()
+ }
+ switch m := m.(type) {
+ case timerEvent:
+ // A connection timer has expired.
+ if !now.Before(c.idleTimeout) {
+ // "[...] the connection is silently closed and
+ // its state is discarded [...]"
+ // https://www.rfc-editor.org/rfc/rfc9000#section-10.1-1
+ c.exited = true
+ return
+ }
+ case func(time.Time, *Conn):
+ // Send a func to msgc to run it on the main Conn goroutine
+ m(now, c)
+ default:
+ panic(fmt.Sprintf("quic: unrecognized conn message %T", m))
+ }
+ }
+}
+
+// sendMsg sends a message to the conn's loop.
+// It does not wait for the message to be processed.
+func (c *Conn) sendMsg(m any) error {
+ select {
+ case c.msgc <- m:
+ case <-c.donec:
+ return errors.New("quic: connection closed")
+ }
+ return nil
+}
+
+// runOnLoop executes a function within the conn's loop goroutine.
+func (c *Conn) runOnLoop(f func(now time.Time, c *Conn)) error {
+ donec := make(chan struct{})
+ if err := c.sendMsg(func(now time.Time, c *Conn) {
+ defer close(donec)
+ f(now, c)
+ }); err != nil {
+ return err
+ }
+ select {
+ case <-donec:
+ case <-c.donec:
+ return errors.New("quic: connection closed")
+ }
+ return nil
+}
+
+// exit fully terminates a connection immediately.
+func (c *Conn) exit() {
+ c.runOnLoop(func(now time.Time, c *Conn) {
+ c.exited = true
+ })
+ <-c.donec
+}
diff --git a/internal/quic/conn_test.go b/internal/quic/conn_test.go
new file mode 100644
index 0000000..a170995
--- /dev/null
+++ b/internal/quic/conn_test.go
@@ -0,0 +1,188 @@
+// 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 (
+ "math"
+ "testing"
+ "time"
+)
+
+func TestConnTestConn(t *testing.T) {
+ tc := newTestConn(t, serverSide)
+ if got, want := tc.timeUntilEvent(), defaultMaxIdleTimeout; got != want {
+ t.Errorf("new conn timeout=%v, want %v (max_idle_timeout)", got, want)
+ }
+
+ var ranAt time.Time
+ tc.conn.runOnLoop(func(now time.Time, c *Conn) {
+ ranAt = now
+ })
+ if !ranAt.Equal(tc.now) {
+ t.Errorf("func ran on loop at %v, want %v", ranAt, tc.now)
+ }
+ tc.wait()
+
+ nextTime := tc.now.Add(defaultMaxIdleTimeout / 2)
+ tc.advanceTo(nextTime)
+ tc.conn.runOnLoop(func(now time.Time, c *Conn) {
+ ranAt = now
+ })
+ if !ranAt.Equal(nextTime) {
+ t.Errorf("func ran on loop at %v, want %v", ranAt, nextTime)
+ }
+ tc.wait()
+
+ tc.advanceToTimer()
+ if err := tc.conn.sendMsg(nil); err == nil {
+ t.Errorf("after advancing to idle timeout, sendMsg = nil, want error")
+ }
+ if !tc.conn.exited {
+ t.Errorf("after advancing to idle timeout, exited = false, want true")
+ }
+}
+
+// A testConn is a Conn whose external interactions (sending and receiving packets,
+// setting timers) can be manipulated in tests.
+type testConn struct {
+ t *testing.T
+ conn *Conn
+ now time.Time
+ timer time.Time
+ timerLastFired time.Time
+ idlec chan struct{} // only accessed on the conn's loop
+}
+
+// newTestConn creates a Conn for testing.
+//
+// The Conn's event loop is controlled by the test,
+// allowing test code to access Conn state directly
+// by first ensuring the loop goroutine is idle.
+func newTestConn(t *testing.T, side connSide) *testConn {
+ t.Helper()
+ tc := &testConn{
+ t: t,
+ now: time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC),
+ }
+ t.Cleanup(tc.cleanup)
+
+ conn, err := newConn(tc.now, (*testConnHooks)(tc))
+ if err != nil {
+ tc.t.Fatal(err)
+ }
+ tc.conn = conn
+
+ tc.wait()
+ return tc
+}
+
+// advance causes time to pass.
+func (tc *testConn) advance(d time.Duration) {
+ tc.t.Helper()
+ tc.advanceTo(tc.now.Add(d))
+}
+
+// advanceTo sets the current time.
+func (tc *testConn) advanceTo(now time.Time) {
+ tc.t.Helper()
+ if tc.now.After(now) {
+ tc.t.Fatalf("time moved backwards: %v -> %v", tc.now, now)
+ }
+ tc.now = now
+ if tc.timer.After(tc.now) {
+ return
+ }
+ tc.conn.sendMsg(timerEvent{})
+ tc.wait()
+}
+
+// advanceToTimer sets the current time to the time of the Conn's next timer event.
+func (tc *testConn) advanceToTimer() {
+ if tc.timer.IsZero() {
+ tc.t.Fatalf("advancing to timer, but timer is not set")
+ }
+ tc.advanceTo(tc.timer)
+}
+
+const infiniteDuration = time.Duration(math.MaxInt64)
+
+// timeUntilEvent returns the amount of time until the next connection event.
+func (tc *testConn) timeUntilEvent() time.Duration {
+ if tc.timer.IsZero() {
+ return infiniteDuration
+ }
+ if tc.timer.Before(tc.now) {
+ return 0
+ }
+ return tc.timer.Sub(tc.now)
+}
+
+// wait blocks until the conn becomes idle.
+// The conn is idle when it is blocked waiting for a packet to arrive or a timer to expire.
+// Tests shouldn't need to call wait directly.
+// testConn methods that wake the Conn event loop will call wait for them.
+func (tc *testConn) wait() {
+ tc.t.Helper()
+ idlec := make(chan struct{})
+ fail := false
+ tc.conn.sendMsg(func(now time.Time, c *Conn) {
+ if tc.idlec != nil {
+ tc.t.Errorf("testConn.wait called concurrently")
+ fail = true
+ close(idlec)
+ } else {
+ // nextMessage will close idlec.
+ tc.idlec = idlec
+ }
+ })
+ select {
+ case <-idlec:
+ case <-tc.conn.donec:
+ }
+ if fail {
+ panic(fail)
+ }
+}
+
+func (tc *testConn) cleanup() {
+ if tc.conn == nil {
+ return
+ }
+ tc.conn.exit()
+}
+
+// testConnHooks implements connTestHooks.
+type testConnHooks testConn
+
+// nextMessage is called by the Conn's event loop to request its next event.
+func (tc *testConnHooks) nextMessage(msgc chan any, timer time.Time) (now time.Time, m any) {
+ tc.timer = timer
+ if !timer.IsZero() && !timer.After(tc.now) {
+ if timer.Equal(tc.timerLastFired) {
+ // If the connection timer fires at time T, the Conn should take some
+ // action to advance the timer into the future. If the Conn reschedules
+ // the timer for the same time, it isn't making progress and we have a bug.
+ tc.t.Errorf("connection timer spinning; now=%v timer=%v", tc.now, timer)
+ } else {
+ tc.timerLastFired = timer
+ return tc.now, timerEvent{}
+ }
+ }
+ select {
+ case m := <-msgc:
+ return tc.now, m
+ default:
+ }
+ // If the message queue is empty, then the conn is idle.
+ if tc.idlec != nil {
+ idlec := tc.idlec
+ tc.idlec = nil
+ close(idlec)
+ }
+ m = <-msgc
+ return tc.now, m
+}
diff --git a/internal/quic/quic.go b/internal/quic/quic.go
index 982c675..c69c0b9 100644
--- a/internal/quic/quic.go
+++ b/internal/quic/quic.go
@@ -19,6 +19,8 @@
// Local values of various transport parameters.
// https://www.rfc-editor.org/rfc/rfc9000.html#section-18.2
const (
+ defaultMaxIdleTimeout = 30 * time.Second // max_idle_timeout
+
// The max_udp_payload_size transport parameter is the size of our
// network receive buffer.
//
