diff --git a/pkg/tcpip/transport/tcp/connect.go b/pkg/tcpip/transport/tcp/connect.go index 127aca135..cc7bfafa9 100644 --- a/pkg/tcpip/transport/tcp/connect.go +++ b/pkg/tcpip/transport/tcp/connect.go @@ -115,15 +115,14 @@ type handshake struct { retransmitTimer *backoffTimer `state:"nosave"` } -// maybeFailTimerHandler takes a handler function for a timer that may fail and -// returns a function that will invoke the provided handler with the endpoint -// mutex held. In addition the returned function will perform any cleanup that -// maybe required if the timer handler returns an error and in case of no errors -// will notify the processor if there are pending segments that need to be -// processed. - +// timerHandler takes a handler function for a timer and returns a function that +// will invoke the provided handler with the endpoint mutex held. In addition +// the returned function will perform any cleanup that may be required if the +// timer handler returns an error. In the case of no errors it will notify the +// processor if there are pending segments that need to be processed. +// // NOTE: e.mu is held for the duration of the call to f(). -func maybeFailTimerHandler(e *endpoint, f func() tcpip.Error) func() { +func timerHandler(e *endpoint, f func() tcpip.Error) func() { return func() { e.mu.Lock() if err := f(); err != nil { @@ -154,27 +153,6 @@ func maybeFailTimerHandler(e *endpoint, f func() tcpip.Error) func() { } } -// timerHandler takes a handler function for a timer that never results in a -// connection being aborted and returns a function that will invoke the provided -// handler with the endpoint mutex held. In addition the returned function will -// notify the processor if there are pending segments that need to be processed -// once the handler function completes. -// -// NOTE: e.mu is held for the duration of the call to f() -func timerHandler(e *endpoint, f func()) func() { - return func() { - e.mu.Lock() - f() - processor := e.protocol.dispatcher.selectProcessor(e.ID) - e.mu.Unlock() - // notify processor if there are pending segments to be - // processed. - if !e.segmentQueue.empty() { - processor.queueEndpoint(e) - } - } -} - // +checklocks:e.mu // +checklocksacquire:h.ep.mu func (e *endpoint) newHandshake() (h *handshake) { @@ -190,7 +168,7 @@ func (e *endpoint) newHandshake() (h *handshake) { e.h = h // By the time handshake is created, e.ID is already initialized. e.TSOffset = e.protocol.tsOffset(e.ID.LocalAddress, e.ID.RemoteAddress) - timer, err := newBackoffTimer(h.ep.stack.Clock(), InitialRTO, MaxRTO, maybeFailTimerHandler(e, h.retransmitHandlerLocked)) + timer, err := newBackoffTimer(h.ep.stack.Clock(), InitialRTO, MaxRTO, timerHandler(e, h.retransmitHandlerLocked)) if err != nil { panic(fmt.Sprintf("newBackOffTimer(_, %s, %s, _) failed: %s", InitialRTO, MaxRTO, err)) } @@ -1315,7 +1293,7 @@ func (e *endpoint) keepaliveTimerExpired() tcpip.Error { userTimeout := e.userTimeout e.keepalive.Lock() - if !e.SocketOptions().GetKeepAlive() || e.keepalive.timer.isZero() || !e.keepalive.timer.checkExpiration() { + if !e.SocketOptions().GetKeepAlive() || e.keepalive.timer.isUninitialized() || !e.keepalive.timer.checkExpiration() { e.keepalive.Unlock() return nil } @@ -1348,7 +1326,7 @@ func (e *endpoint) keepaliveTimerExpired() tcpip.Error { func (e *endpoint) resetKeepaliveTimer(receivedData bool) { e.keepalive.Lock() defer e.keepalive.Unlock() - if e.keepalive.timer.isZero() { + if e.keepalive.timer.isUninitialized() { if state := e.EndpointState(); !state.closed() { panic(fmt.Sprintf("Unexpected state when the keepalive time is cleaned up, got %s, want %s or %s", state, StateClose, StateError)) } diff --git a/pkg/tcpip/transport/tcp/endpoint.go b/pkg/tcpip/transport/tcp/endpoint.go index 7eada75e4..348c1004d 100644 --- a/pkg/tcpip/transport/tcp/endpoint.go +++ b/pkg/tcpip/transport/tcp/endpoint.go @@ -914,7 +914,7 @@ func newEndpoint(s *stack.Stack, protocol *protocol, netProto tcpip.NetworkProto // TODO(https://gvisor.dev/issues/7493): Defer creating the timer until TCP connection becomes // established. - e.keepalive.timer.init(e.stack.Clock(), maybeFailTimerHandler(e, e.keepaliveTimerExpired)) + e.keepalive.timer.init(e.stack.Clock(), timerHandler(e, e.keepaliveTimerExpired)) return e } diff --git a/pkg/tcpip/transport/tcp/endpoint_state.go b/pkg/tcpip/transport/tcp/endpoint_state.go index 8382b35b9..c817b0fb8 100644 --- a/pkg/tcpip/transport/tcp/endpoint_state.go +++ b/pkg/tcpip/transport/tcp/endpoint_state.go @@ -121,10 +121,10 @@ func (e *endpoint) afterLoad() { // Resume implements tcpip.ResumableEndpoint.Resume. func (e *endpoint) Resume(s *stack.Stack) { if !e.EndpointState().closed() { - e.keepalive.timer.init(s.Clock(), maybeFailTimerHandler(e, e.keepaliveTimerExpired)) + e.keepalive.timer.init(s.Clock(), timerHandler(e, e.keepaliveTimerExpired)) } if snd := e.snd; snd != nil { - snd.resendTimer.init(s.Clock(), maybeFailTimerHandler(e, e.snd.retransmitTimerExpired)) + snd.resendTimer.init(s.Clock(), timerHandler(e, e.snd.retransmitTimerExpired)) snd.reorderTimer.init(s.Clock(), timerHandler(e, e.snd.rc.reorderTimerExpired)) snd.probeTimer.init(s.Clock(), timerHandler(e, e.snd.probeTimerExpired)) } @@ -243,7 +243,7 @@ func (e *endpoint) Resume(s *stack.Stack) { panic(fmt.Sprintf("FindRoute failed when restoring endpoint w/ ID: %+v", e.ID)) } e.route = r - timer, err := newBackoffTimer(e.stack.Clock(), InitialRTO, MaxRTO, maybeFailTimerHandler(e, e.h.retransmitHandlerLocked)) + timer, err := newBackoffTimer(e.stack.Clock(), InitialRTO, MaxRTO, timerHandler(e, e.h.retransmitHandlerLocked)) if err != nil { panic(fmt.Sprintf("newBackOffTimer(_, %s, %s, _) failed: %s", InitialRTO, MaxRTO, err)) } diff --git a/pkg/tcpip/transport/tcp/rack.go b/pkg/tcpip/transport/tcp/rack.go index 7dccc9562..66ea6e5b0 100644 --- a/pkg/tcpip/transport/tcp/rack.go +++ b/pkg/tcpip/transport/tcp/rack.go @@ -188,9 +188,9 @@ func (s *sender) schedulePTO() { // https://tools.ietf.org/html/draft-ietf-tcpm-rack-08#section-7.5.2. // // +checklocks:s.ep.mu -func (s *sender) probeTimerExpired() { - if s.probeTimer.isZero() || !s.probeTimer.checkExpiration() { - return +func (s *sender) probeTimerExpired() tcpip.Error { + if s.probeTimer.isUninitialized() || !s.probeTimer.checkExpiration() { + return nil } var dataSent bool @@ -231,7 +231,7 @@ func (s *sender) probeTimerExpired() { // not the probe timer. This ensures that the sender does not send repeated, // back-to-back tail loss probes. s.postXmit(dataSent, false /* shouldScheduleProbe */) - return + return nil } // detectTLPRecovery detects if recovery was accomplished by the loss probes @@ -388,14 +388,14 @@ func (rc *rackControl) detectLoss(rcvTime tcpip.MonotonicTime) int { // before the reorder timer expired. // // +checklocks:rc.snd.ep.mu -func (rc *rackControl) reorderTimerExpired() { - if rc.snd.reorderTimer.isZero() || !rc.snd.reorderTimer.checkExpiration() { - return +func (rc *rackControl) reorderTimerExpired() tcpip.Error { + if rc.snd.reorderTimer.isUninitialized() || !rc.snd.reorderTimer.checkExpiration() { + return nil } numLost := rc.detectLoss(rc.snd.ep.stack.Clock().NowMonotonic()) if numLost == 0 { - return + return nil } fastRetransmit := false @@ -406,7 +406,7 @@ func (rc *rackControl) reorderTimerExpired() { } rc.DoRecovery(nil, fastRetransmit) - return + return nil } // DoRecovery implements lossRecovery.DoRecovery. diff --git a/pkg/tcpip/transport/tcp/snd.go b/pkg/tcpip/transport/tcp/snd.go index 78abc07d4..838e299fe 100644 --- a/pkg/tcpip/transport/tcp/snd.go +++ b/pkg/tcpip/transport/tcp/snd.go @@ -205,7 +205,7 @@ func newSender(ep *endpoint, iss, irs seqnum.Value, sndWnd seqnum.Size, mss uint s.SndWndScale = uint8(sndWndScale) } - s.resendTimer.init(s.ep.stack.Clock(), maybeFailTimerHandler(s.ep, s.retransmitTimerExpired)) + s.resendTimer.init(s.ep.stack.Clock(), timerHandler(s.ep, s.retransmitTimerExpired)) s.reorderTimer.init(s.ep.stack.Clock(), timerHandler(s.ep, s.rc.reorderTimerExpired)) s.probeTimer.init(s.ep.stack.Clock(), timerHandler(s.ep, s.probeTimerExpired)) @@ -437,7 +437,7 @@ func (s *sender) resendSegment() { func (s *sender) retransmitTimerExpired() tcpip.Error { // Check if the timer actually expired or if it's a spurious wake due // to a previously orphaned runtime timer. - if s.resendTimer.isZero() || !s.resendTimer.checkExpiration() { + if s.resendTimer.isUninitialized() || !s.resendTimer.checkExpiration() { return nil } diff --git a/pkg/tcpip/transport/tcp/timer.go b/pkg/tcpip/transport/tcp/timer.go index 208009263..7111789d5 100644 --- a/pkg/tcpip/transport/tcp/timer.go +++ b/pkg/tcpip/transport/tcp/timer.go @@ -15,7 +15,6 @@ package tcp import ( - "math" "time" "gvisor.dev/gvisor/pkg/tcpip" @@ -24,8 +23,10 @@ import ( type timerState int const ( + // The timer has not been initialized yet or has been cleaned up. + timerUninitialized timerState = iota // The timer is disabled. - timerStateDisabled timerState = iota + timerStateDisabled // The timer is enabled, but the clock timer may be set to an earlier // expiration time due to a previous orphaned state. timerStateEnabled @@ -66,6 +67,9 @@ type timer struct { // timer is the clock timer used to wait on. timer tcpip.Timer + + // callback is the function that's called when the timer expires. + callback func() } // init initializes the timer. Once it expires the function callback @@ -73,11 +77,7 @@ type timer struct { func (t *timer) init(clock tcpip.Clock, f func()) { t.state = timerStateDisabled t.clock = clock - - // Initialize a clock timer that will call the callback func, then - // immediately stop it. - t.timer = t.clock.AfterFunc(math.MaxInt64, f) - t.timer.Stop() + t.callback = f } // cleanup frees all resources associated with the timer. @@ -90,15 +90,15 @@ func (t *timer) cleanup() { *t = timer{} } -// isZero returns true if the timer is in the zero state. This is usually -// only true if init() has never been called or if cleanup has been called. -func (t *timer) isZero() bool { - return *t == timer{} +// isUninitialized returns true if the timer is in the uninitialized state. This +// is only true if init() has never been called or if cleanup has been called. +func (t *timer) isUninitialized() bool { + return t.state == timerUninitialized } // checkExpiration checks if the given timer has actually expired, it should be // called whenever the callback function is called, and is used to check if it's -// a supurious timer expiration (due to a previously orphaned timer) or a +// a spurious timer expiration (due to a previously orphaned timer) or a // legitimate one. func (t *timer) checkExpiration() bool { // Transition to fully disabled state if we're just consuming an @@ -143,8 +143,18 @@ func (t *timer) enable(d time.Duration) { // Check if we need to set the runtime timer. if t.state == timerStateDisabled || t.target.Before(t.clockTarget) { t.clockTarget = t.target - t.timer.Reset(d) + t.resetOrStart(d) } t.state = timerStateEnabled } + +// resetOrStart creates the timer if it doesn't already exist or resets it with +// the given duration if it does. +func (t *timer) resetOrStart(d time.Duration) { + if t.timer == nil { + t.timer = t.clock.AfterFunc(d, t.callback) + } else { + t.timer.Reset(d) + } +} diff --git a/pkg/tcpip/transport/tcp/timer_test.go b/pkg/tcpip/transport/tcp/timer_test.go index 30f8c8e6f..a6dc984a3 100644 --- a/pkg/tcpip/transport/tcp/timer_test.go +++ b/pkg/tcpip/transport/tcp/timer_test.go @@ -36,8 +36,8 @@ func TestCleanup(t *testing.T) { tmr.enable(timerDurationSeconds * time.Second) tmr.cleanup() - if want := (timer{}); tmr != want { - t.Errorf("got tmr = %+v, want = %+v", tmr, want) + if !tmr.isUninitialized() { + t.Errorf("got tmr.isUninitialized = false, want = true") } // The waker should not be asserted.