Skip to content

Commit

Permalink
http2: use synthetic time in server tests
Browse files Browse the repository at this point in the history
Change newServerTester to return a server using fake time
and a fake net.Conn.

Change-Id: I9d5db0cbe75696aed6d99ff1cd2369c2dea426c3
Reviewed-on: https://go-review.googlesource.com/c/net/+/586247
LUCI-TryBot-Result: Go LUCI <[email protected]>
Reviewed-by: Jonathan Amsterdam <[email protected]>
  • Loading branch information
neild committed May 28, 2024
1 parent 022530c commit 03c24c2
Show file tree
Hide file tree
Showing 7 changed files with 318 additions and 156 deletions.
13 changes: 13 additions & 0 deletions http2/http2.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package http2 // import "golang.org/x/net/http2"

import (
"bufio"
"context"
"crypto/tls"
"fmt"
"io"
Expand All @@ -26,6 +27,7 @@ import (
"strconv"
"strings"
"sync"
"time"

"golang.org/x/net/http/httpguts"
)
Expand Down Expand Up @@ -377,3 +379,14 @@ func validPseudoPath(v string) bool {
// makes that struct also non-comparable, and generally doesn't add
// any size (as long as it's first).
type incomparable [0]func()

// synctestGroupInterface is the methods of synctestGroup used by Server and Transport.
// It's defined as an interface here to let us keep synctestGroup entirely test-only
// and not a part of non-test builds.
type synctestGroupInterface interface {
Join()
Now() time.Time
NewTimer(d time.Duration) timer
AfterFunc(d time.Duration, f func()) timer
ContextWithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc)
}
87 changes: 65 additions & 22 deletions http2/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,39 @@ type Server struct {
// so that we don't embed a Mutex in this struct, which will make the
// struct non-copyable, which might break some callers.
state *serverInternalState

// Synchronization group used for testing.
// Outside of tests, this is nil.
group synctestGroupInterface
}

func (s *Server) markNewGoroutine() {
if s.group != nil {
s.group.Join()
}
}

func (s *Server) now() time.Time {
if s.group != nil {
return s.group.Now()
}
return time.Now()
}

// newTimer creates a new time.Timer, or a synthetic timer in tests.
func (s *Server) newTimer(d time.Duration) timer {
if s.group != nil {
return s.group.NewTimer(d)
}
return timeTimer{time.NewTimer(d)}
}

// afterFunc creates a new time.AfterFunc timer, or a synthetic timer in tests.
func (s *Server) afterFunc(d time.Duration, f func()) timer {
if s.group != nil {
return s.group.AfterFunc(d, f)
}
return timeTimer{time.AfterFunc(d, f)}
}

func (s *Server) initialConnRecvWindowSize() int32 {
Expand Down Expand Up @@ -400,6 +433,10 @@ func (o *ServeConnOpts) handler() http.Handler {
//
// The opts parameter is optional. If nil, default values are used.
func (s *Server) ServeConn(c net.Conn, opts *ServeConnOpts) {
s.serveConn(c, opts, nil)
}

func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverConn)) {
baseCtx, cancel := serverConnBaseContext(c, opts)
defer cancel()

Expand All @@ -426,6 +463,9 @@ func (s *Server) ServeConn(c net.Conn, opts *ServeConnOpts) {
pushEnabled: true,
sawClientPreface: opts.SawClientPreface,
}
if newf != nil {
newf(sc)
}

s.state.registerConn(sc)
defer s.state.unregisterConn(sc)
Expand Down Expand Up @@ -599,8 +639,8 @@ type serverConn struct {
inFrameScheduleLoop bool // whether we're in the scheduleFrameWrite loop
needToSendGoAway bool // we need to schedule a GOAWAY frame write
goAwayCode ErrCode
shutdownTimer *time.Timer // nil until used
idleTimer *time.Timer // nil if unused
shutdownTimer timer // nil until used
idleTimer timer // nil if unused

// Owned by the writeFrameAsync goroutine:
headerWriteBuf bytes.Buffer
Expand Down Expand Up @@ -649,12 +689,12 @@ type stream struct {
flow outflow // limits writing from Handler to client
inflow inflow // what the client is allowed to POST/etc to us
state streamState
resetQueued bool // RST_STREAM queued for write; set by sc.resetStream
gotTrailerHeader bool // HEADER frame for trailers was seen
wroteHeaders bool // whether we wrote headers (not status 100)
readDeadline *time.Timer // nil if unused
writeDeadline *time.Timer // nil if unused
closeErr error // set before cw is closed
resetQueued bool // RST_STREAM queued for write; set by sc.resetStream
gotTrailerHeader bool // HEADER frame for trailers was seen
wroteHeaders bool // whether we wrote headers (not status 100)
readDeadline timer // nil if unused
writeDeadline timer // nil if unused
closeErr error // set before cw is closed

trailer http.Header // accumulated trailers
reqTrailer http.Header // handler's Request.Trailer
Expand Down Expand Up @@ -811,6 +851,7 @@ type readFrameResult struct {
// consumer is done with the frame.
// It's run on its own goroutine.
func (sc *serverConn) readFrames() {
sc.srv.markNewGoroutine()
gate := make(chan struct{})
gateDone := func() { gate <- struct{}{} }
for {
Expand Down Expand Up @@ -843,6 +884,7 @@ type frameWriteResult struct {
// At most one goroutine can be running writeFrameAsync at a time per
// serverConn.
func (sc *serverConn) writeFrameAsync(wr FrameWriteRequest, wd *writeData) {
sc.srv.markNewGoroutine()
var err error
if wd == nil {
err = wr.write.writeFrame(sc)
Expand Down Expand Up @@ -922,13 +964,13 @@ func (sc *serverConn) serve() {
sc.setConnState(http.StateIdle)

if sc.srv.IdleTimeout > 0 {
sc.idleTimer = time.AfterFunc(sc.srv.IdleTimeout, sc.onIdleTimer)
sc.idleTimer = sc.srv.afterFunc(sc.srv.IdleTimeout, sc.onIdleTimer)
defer sc.idleTimer.Stop()
}

go sc.readFrames() // closed by defer sc.conn.Close above

settingsTimer := time.AfterFunc(firstSettingsTimeout, sc.onSettingsTimer)
settingsTimer := sc.srv.afterFunc(firstSettingsTimeout, sc.onSettingsTimer)
defer settingsTimer.Stop()

loopNum := 0
Expand Down Expand Up @@ -1057,10 +1099,10 @@ func (sc *serverConn) readPreface() error {
errc <- nil
}
}()
timer := time.NewTimer(prefaceTimeout) // TODO: configurable on *Server?
timer := sc.srv.newTimer(prefaceTimeout) // TODO: configurable on *Server?
defer timer.Stop()
select {
case <-timer.C:
case <-timer.C():
return errPrefaceTimeout
case err := <-errc:
if err == nil {
Expand Down Expand Up @@ -1425,7 +1467,7 @@ func (sc *serverConn) goAway(code ErrCode) {

func (sc *serverConn) shutDownIn(d time.Duration) {
sc.serveG.check()
sc.shutdownTimer = time.AfterFunc(d, sc.onShutdownTimer)
sc.shutdownTimer = sc.srv.afterFunc(d, sc.onShutdownTimer)
}

func (sc *serverConn) resetStream(se StreamError) {
Expand Down Expand Up @@ -2022,7 +2064,7 @@ func (sc *serverConn) processHeaders(f *MetaHeadersFrame) error {
// (in Go 1.8), though. That's a more sane option anyway.
if sc.hs.ReadTimeout > 0 {
sc.conn.SetReadDeadline(time.Time{})
st.readDeadline = time.AfterFunc(sc.hs.ReadTimeout, st.onReadTimeout)
st.readDeadline = sc.srv.afterFunc(sc.hs.ReadTimeout, st.onReadTimeout)
}

return sc.scheduleHandler(id, rw, req, handler)
Expand Down Expand Up @@ -2120,7 +2162,7 @@ func (sc *serverConn) newStream(id, pusherID uint32, state streamState) *stream
st.flow.add(sc.initialStreamSendWindowSize)
st.inflow.init(sc.srv.initialStreamRecvWindowSize())
if sc.hs.WriteTimeout > 0 {
st.writeDeadline = time.AfterFunc(sc.hs.WriteTimeout, st.onWriteTimeout)
st.writeDeadline = sc.srv.afterFunc(sc.hs.WriteTimeout, st.onWriteTimeout)
}

sc.streams[id] = st
Expand Down Expand Up @@ -2344,6 +2386,7 @@ func (sc *serverConn) handlerDone() {

// Run on its own goroutine.
func (sc *serverConn) runHandler(rw *responseWriter, req *http.Request, handler func(http.ResponseWriter, *http.Request)) {
sc.srv.markNewGoroutine()
defer sc.sendServeMsg(handlerDoneMsg)
didPanic := true
defer func() {
Expand Down Expand Up @@ -2640,7 +2683,7 @@ func (rws *responseWriterState) writeChunk(p []byte) (n int, err error) {
var date string
if _, ok := rws.snapHeader["Date"]; !ok {
// TODO(bradfitz): be faster here, like net/http? measure.
date = time.Now().UTC().Format(http.TimeFormat)
date = rws.conn.srv.now().UTC().Format(http.TimeFormat)
}

for _, v := range rws.snapHeader["Trailer"] {
Expand Down Expand Up @@ -2762,7 +2805,7 @@ func (rws *responseWriterState) promoteUndeclaredTrailers() {

func (w *responseWriter) SetReadDeadline(deadline time.Time) error {
st := w.rws.stream
if !deadline.IsZero() && deadline.Before(time.Now()) {
if !deadline.IsZero() && deadline.Before(w.rws.conn.srv.now()) {
// If we're setting a deadline in the past, reset the stream immediately
// so writes after SetWriteDeadline returns will fail.
st.onReadTimeout()
Expand All @@ -2778,17 +2821,17 @@ func (w *responseWriter) SetReadDeadline(deadline time.Time) error {
if deadline.IsZero() {
st.readDeadline = nil
} else if st.readDeadline == nil {
st.readDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onReadTimeout)
st.readDeadline = sc.srv.afterFunc(deadline.Sub(w.rws.conn.srv.now()), st.onReadTimeout)
} else {
st.readDeadline.Reset(deadline.Sub(time.Now()))
st.readDeadline.Reset(deadline.Sub(w.rws.conn.srv.now()))
}
})
return nil
}

func (w *responseWriter) SetWriteDeadline(deadline time.Time) error {
st := w.rws.stream
if !deadline.IsZero() && deadline.Before(time.Now()) {
if !deadline.IsZero() && deadline.Before(w.rws.conn.srv.now()) {
// If we're setting a deadline in the past, reset the stream immediately
// so writes after SetWriteDeadline returns will fail.
st.onWriteTimeout()
Expand All @@ -2804,9 +2847,9 @@ func (w *responseWriter) SetWriteDeadline(deadline time.Time) error {
if deadline.IsZero() {
st.writeDeadline = nil
} else if st.writeDeadline == nil {
st.writeDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onWriteTimeout)
st.writeDeadline = sc.srv.afterFunc(deadline.Sub(w.rws.conn.srv.now()), st.onWriteTimeout)
} else {
st.writeDeadline.Reset(deadline.Sub(time.Now()))
st.writeDeadline.Reset(deadline.Sub(w.rws.conn.srv.now()))
}
})
return nil
Expand Down
6 changes: 3 additions & 3 deletions http2/server_push_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ func TestServer_Push_Success(t *testing.T) {
errc <- fmt.Errorf("unknown RequestURL %q", r.URL.RequestURI())
}
})
stURL = st.ts.URL
stURL = "https://" + st.authority()

// Send one request, which should push two responses.
st.greet()
Expand Down Expand Up @@ -169,7 +169,7 @@ func TestServer_Push_Success(t *testing.T) {
return checkPushPromise(f, 2, [][2]string{
{":method", "GET"},
{":scheme", "https"},
{":authority", st.ts.Listener.Addr().String()},
{":authority", st.authority()},
{":path", "/pushed?get"},
{"user-agent", userAgent},
})
Expand All @@ -178,7 +178,7 @@ func TestServer_Push_Success(t *testing.T) {
return checkPushPromise(f, 4, [][2]string{
{":method", "HEAD"},
{":scheme", "https"},
{":authority", st.ts.Listener.Addr().String()},
{":authority", st.authority()},
{":path", "/pushed?head"},
{"cookie", cookie},
{"user-agent", userAgent},
Expand Down
Loading

0 comments on commit 03c24c2

Please sign in to comment.