From 1437270d71dd9449eda9cb7d9677a4a1a94e4d44 Mon Sep 17 00:00:00 2001 From: Lucas Manning Date: Fri, 9 Feb 2024 12:55:19 -0800 Subject: [PATCH] Add a few small cleanups to tcp timers. First is to have the timer save its callback function rather than calling AfterFunc and Stop immediately. AfterFunc spawns two goroutines under the hood and Stop cancels them, so this avoids a couple scheduler interactions. Second is to unify maybeFailTimerHandler and timerHandler. These two functions do the same thing, the former just handles errors. We can easily modify the functions passed to timerHandler to just return nil errors and only use one function. Last is renaming isZero to isUninitialized, which IMO more accurately describes what the function is checking. PiperOrigin-RevId: 605708542 --- pkg/tcpip/transport/tcp/connect.go | 42 ++++++----------------- pkg/tcpip/transport/tcp/endpoint.go | 2 +- pkg/tcpip/transport/tcp/endpoint_state.go | 6 ++-- pkg/tcpip/transport/tcp/rack.go | 18 +++++----- pkg/tcpip/transport/tcp/snd.go | 4 +-- pkg/tcpip/transport/tcp/timer.go | 36 ++++++++++++------- pkg/tcpip/transport/tcp/timer_test.go | 4 +-- 7 files changed, 50 insertions(+), 62 deletions(-) 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.