@@ -154,6 +154,39 @@ type Server struct {
154
154
// so that we don't embed a Mutex in this struct, which will make the
155
155
// struct non-copyable, which might break some callers.
156
156
state * serverInternalState
157
+
158
+ // Synchronization group used for testing.
159
+ // Outside of tests, this is nil.
160
+ group synctestGroupInterface
161
+ }
162
+
163
+ func (s * Server ) markNewGoroutine () {
164
+ if s .group != nil {
165
+ s .group .Join ()
166
+ }
167
+ }
168
+
169
+ func (s * Server ) now () time.Time {
170
+ if s .group != nil {
171
+ return s .group .Now ()
172
+ }
173
+ return time .Now ()
174
+ }
175
+
176
+ // newTimer creates a new time.Timer, or a synthetic timer in tests.
177
+ func (s * Server ) newTimer (d time.Duration ) timer {
178
+ if s .group != nil {
179
+ return s .group .NewTimer (d )
180
+ }
181
+ return timeTimer {time .NewTimer (d )}
182
+ }
183
+
184
+ // afterFunc creates a new time.AfterFunc timer, or a synthetic timer in tests.
185
+ func (s * Server ) afterFunc (d time.Duration , f func ()) timer {
186
+ if s .group != nil {
187
+ return s .group .AfterFunc (d , f )
188
+ }
189
+ return timeTimer {time .AfterFunc (d , f )}
157
190
}
158
191
159
192
func (s * Server ) initialConnRecvWindowSize () int32 {
@@ -400,6 +433,10 @@ func (o *ServeConnOpts) handler() http.Handler {
400
433
//
401
434
// The opts parameter is optional. If nil, default values are used.
402
435
func (s * Server ) ServeConn (c net.Conn , opts * ServeConnOpts ) {
436
+ s .serveConn (c , opts , nil )
437
+ }
438
+
439
+ func (s * Server ) serveConn (c net.Conn , opts * ServeConnOpts , newf func (* serverConn )) {
403
440
baseCtx , cancel := serverConnBaseContext (c , opts )
404
441
defer cancel ()
405
442
@@ -426,6 +463,9 @@ func (s *Server) ServeConn(c net.Conn, opts *ServeConnOpts) {
426
463
pushEnabled : true ,
427
464
sawClientPreface : opts .SawClientPreface ,
428
465
}
466
+ if newf != nil {
467
+ newf (sc )
468
+ }
429
469
430
470
s .state .registerConn (sc )
431
471
defer s .state .unregisterConn (sc )
@@ -599,8 +639,8 @@ type serverConn struct {
599
639
inFrameScheduleLoop bool // whether we're in the scheduleFrameWrite loop
600
640
needToSendGoAway bool // we need to schedule a GOAWAY frame write
601
641
goAwayCode ErrCode
602
- shutdownTimer * time. Timer // nil until used
603
- idleTimer * time. Timer // nil if unused
642
+ shutdownTimer timer // nil until used
643
+ idleTimer timer // nil if unused
604
644
605
645
// Owned by the writeFrameAsync goroutine:
606
646
headerWriteBuf bytes.Buffer
@@ -649,12 +689,12 @@ type stream struct {
649
689
flow outflow // limits writing from Handler to client
650
690
inflow inflow // what the client is allowed to POST/etc to us
651
691
state streamState
652
- resetQueued bool // RST_STREAM queued for write; set by sc.resetStream
653
- gotTrailerHeader bool // HEADER frame for trailers was seen
654
- wroteHeaders bool // whether we wrote headers (not status 100)
655
- readDeadline * time. Timer // nil if unused
656
- writeDeadline * time. Timer // nil if unused
657
- closeErr error // set before cw is closed
692
+ resetQueued bool // RST_STREAM queued for write; set by sc.resetStream
693
+ gotTrailerHeader bool // HEADER frame for trailers was seen
694
+ wroteHeaders bool // whether we wrote headers (not status 100)
695
+ readDeadline timer // nil if unused
696
+ writeDeadline timer // nil if unused
697
+ closeErr error // set before cw is closed
658
698
659
699
trailer http.Header // accumulated trailers
660
700
reqTrailer http.Header // handler's Request.Trailer
@@ -811,6 +851,7 @@ type readFrameResult struct {
811
851
// consumer is done with the frame.
812
852
// It's run on its own goroutine.
813
853
func (sc * serverConn ) readFrames () {
854
+ sc .srv .markNewGoroutine ()
814
855
gate := make (chan struct {})
815
856
gateDone := func () { gate <- struct {}{} }
816
857
for {
@@ -843,6 +884,7 @@ type frameWriteResult struct {
843
884
// At most one goroutine can be running writeFrameAsync at a time per
844
885
// serverConn.
845
886
func (sc * serverConn ) writeFrameAsync (wr FrameWriteRequest , wd * writeData ) {
887
+ sc .srv .markNewGoroutine ()
846
888
var err error
847
889
if wd == nil {
848
890
err = wr .write .writeFrame (sc )
@@ -922,13 +964,13 @@ func (sc *serverConn) serve() {
922
964
sc .setConnState (http .StateIdle )
923
965
924
966
if sc .srv .IdleTimeout > 0 {
925
- sc .idleTimer = time . AfterFunc (sc .srv .IdleTimeout , sc .onIdleTimer )
967
+ sc .idleTimer = sc . srv . afterFunc (sc .srv .IdleTimeout , sc .onIdleTimer )
926
968
defer sc .idleTimer .Stop ()
927
969
}
928
970
929
971
go sc .readFrames () // closed by defer sc.conn.Close above
930
972
931
- settingsTimer := time . AfterFunc (firstSettingsTimeout , sc .onSettingsTimer )
973
+ settingsTimer := sc . srv . afterFunc (firstSettingsTimeout , sc .onSettingsTimer )
932
974
defer settingsTimer .Stop ()
933
975
934
976
loopNum := 0
@@ -1057,10 +1099,10 @@ func (sc *serverConn) readPreface() error {
1057
1099
errc <- nil
1058
1100
}
1059
1101
}()
1060
- timer := time . NewTimer (prefaceTimeout ) // TODO: configurable on *Server?
1102
+ timer := sc . srv . newTimer (prefaceTimeout ) // TODO: configurable on *Server?
1061
1103
defer timer .Stop ()
1062
1104
select {
1063
- case <- timer .C :
1105
+ case <- timer .C () :
1064
1106
return errPrefaceTimeout
1065
1107
case err := <- errc :
1066
1108
if err == nil {
@@ -1425,7 +1467,7 @@ func (sc *serverConn) goAway(code ErrCode) {
1425
1467
1426
1468
func (sc * serverConn ) shutDownIn (d time.Duration ) {
1427
1469
sc .serveG .check ()
1428
- sc .shutdownTimer = time . AfterFunc (d , sc .onShutdownTimer )
1470
+ sc .shutdownTimer = sc . srv . afterFunc (d , sc .onShutdownTimer )
1429
1471
}
1430
1472
1431
1473
func (sc * serverConn ) resetStream (se StreamError ) {
@@ -2022,7 +2064,7 @@ func (sc *serverConn) processHeaders(f *MetaHeadersFrame) error {
2022
2064
// (in Go 1.8), though. That's a more sane option anyway.
2023
2065
if sc .hs .ReadTimeout > 0 {
2024
2066
sc .conn .SetReadDeadline (time.Time {})
2025
- st .readDeadline = time . AfterFunc (sc .hs .ReadTimeout , st .onReadTimeout )
2067
+ st .readDeadline = sc . srv . afterFunc (sc .hs .ReadTimeout , st .onReadTimeout )
2026
2068
}
2027
2069
2028
2070
return sc .scheduleHandler (id , rw , req , handler )
@@ -2120,7 +2162,7 @@ func (sc *serverConn) newStream(id, pusherID uint32, state streamState) *stream
2120
2162
st .flow .add (sc .initialStreamSendWindowSize )
2121
2163
st .inflow .init (sc .srv .initialStreamRecvWindowSize ())
2122
2164
if sc .hs .WriteTimeout > 0 {
2123
- st .writeDeadline = time . AfterFunc (sc .hs .WriteTimeout , st .onWriteTimeout )
2165
+ st .writeDeadline = sc . srv . afterFunc (sc .hs .WriteTimeout , st .onWriteTimeout )
2124
2166
}
2125
2167
2126
2168
sc .streams [id ] = st
@@ -2344,6 +2386,7 @@ func (sc *serverConn) handlerDone() {
2344
2386
2345
2387
// Run on its own goroutine.
2346
2388
func (sc * serverConn ) runHandler (rw * responseWriter , req * http.Request , handler func (http.ResponseWriter , * http.Request )) {
2389
+ sc .srv .markNewGoroutine ()
2347
2390
defer sc .sendServeMsg (handlerDoneMsg )
2348
2391
didPanic := true
2349
2392
defer func () {
@@ -2640,7 +2683,7 @@ func (rws *responseWriterState) writeChunk(p []byte) (n int, err error) {
2640
2683
var date string
2641
2684
if _ , ok := rws .snapHeader ["Date" ]; ! ok {
2642
2685
// TODO(bradfitz): be faster here, like net/http? measure.
2643
- date = time . Now ().UTC ().Format (http .TimeFormat )
2686
+ date = rws . conn . srv . now ().UTC ().Format (http .TimeFormat )
2644
2687
}
2645
2688
2646
2689
for _ , v := range rws .snapHeader ["Trailer" ] {
@@ -2762,7 +2805,7 @@ func (rws *responseWriterState) promoteUndeclaredTrailers() {
2762
2805
2763
2806
func (w * responseWriter ) SetReadDeadline (deadline time.Time ) error {
2764
2807
st := w .rws .stream
2765
- if ! deadline .IsZero () && deadline .Before (time . Now ()) {
2808
+ if ! deadline .IsZero () && deadline .Before (w . rws . conn . srv . now ()) {
2766
2809
// If we're setting a deadline in the past, reset the stream immediately
2767
2810
// so writes after SetWriteDeadline returns will fail.
2768
2811
st .onReadTimeout ()
@@ -2778,17 +2821,17 @@ func (w *responseWriter) SetReadDeadline(deadline time.Time) error {
2778
2821
if deadline .IsZero () {
2779
2822
st .readDeadline = nil
2780
2823
} else if st .readDeadline == nil {
2781
- st .readDeadline = time . AfterFunc (deadline .Sub (time . Now ()), st .onReadTimeout )
2824
+ st .readDeadline = sc . srv . afterFunc (deadline .Sub (w . rws . conn . srv . now ()), st .onReadTimeout )
2782
2825
} else {
2783
- st .readDeadline .Reset (deadline .Sub (time . Now ()))
2826
+ st .readDeadline .Reset (deadline .Sub (w . rws . conn . srv . now ()))
2784
2827
}
2785
2828
})
2786
2829
return nil
2787
2830
}
2788
2831
2789
2832
func (w * responseWriter ) SetWriteDeadline (deadline time.Time ) error {
2790
2833
st := w .rws .stream
2791
- if ! deadline .IsZero () && deadline .Before (time . Now ()) {
2834
+ if ! deadline .IsZero () && deadline .Before (w . rws . conn . srv . now ()) {
2792
2835
// If we're setting a deadline in the past, reset the stream immediately
2793
2836
// so writes after SetWriteDeadline returns will fail.
2794
2837
st .onWriteTimeout ()
@@ -2804,9 +2847,9 @@ func (w *responseWriter) SetWriteDeadline(deadline time.Time) error {
2804
2847
if deadline .IsZero () {
2805
2848
st .writeDeadline = nil
2806
2849
} else if st .writeDeadline == nil {
2807
- st .writeDeadline = time . AfterFunc (deadline .Sub (time . Now ()), st .onWriteTimeout )
2850
+ st .writeDeadline = sc . srv . afterFunc (deadline .Sub (w . rws . conn . srv . now ()), st .onWriteTimeout )
2808
2851
} else {
2809
- st .writeDeadline .Reset (deadline .Sub (time . Now ()))
2852
+ st .writeDeadline .Reset (deadline .Sub (w . rws . conn . srv . now ()))
2810
2853
}
2811
2854
})
2812
2855
return nil
0 commit comments