From 74a1820cebc1005d95f929a9204c1767e605edd8 Mon Sep 17 00:00:00 2001 From: Bhasker Hariharan Date: Mon, 18 Apr 2022 17:26:26 -0700 Subject: [PATCH] Remove TCP endpoint goroutines. This change removes all endpoint goroutines and all TCP processing is now done inline in the TCP processor loop. TCP timers directly invoke handlers as required rather than assert a waker. UnlockUser is also simplified to just queue the endpoint to the processor instead of trying to process segments inline. This allows us to centralize logic for TCP state handling in the processor. This potentially could involve an extra wakeup but now that endpoint goroutines do not exist this is not such a big concern as in case of busy servers the processor goroutines will already be running anyway. This change also allows us to clean up S/R as now restoring a TCP endpoint does not require restarting a goroutine and moving it to the right logical point but only requires that we restart any timers that may have been running when the save was done and restore any port bindings as required. Endpoint.Release is now removed in favor of Endpoint.Abort by using Abort in places where we use Endpoint.Release. Updates #231 PiperOrigin-RevId: 442673015 --- pkg/refsvfs2/refs_map.go | 22 +- pkg/sentry/inet/inet.go | 3 + pkg/sentry/inet/test_stack.go | 3 + pkg/sentry/kernel/kernel.go | 9 + pkg/sentry/socket/hostinet/stack.go | 3 + pkg/sentry/socket/netstack/stack.go | 5 + pkg/tcpip/header/tcp.go | 2 + pkg/tcpip/socketops.go | 16 +- pkg/tcpip/stack/registration.go | 7 + pkg/tcpip/stack/stack.go | 11 + pkg/tcpip/stack/transport_demuxer.go | 9 +- pkg/tcpip/stack/transport_test.go | 6 + pkg/tcpip/tcpip.go | 5 +- pkg/tcpip/transport/icmp/endpoint.go | 3 - pkg/tcpip/transport/icmp/protocol.go | 6 + pkg/tcpip/transport/packet/endpoint.go | 3 - pkg/tcpip/transport/raw/endpoint.go | 3 - pkg/tcpip/transport/tcp/accept.go | 200 +----- pkg/tcpip/transport/tcp/connect.go | 655 +++++------------- pkg/tcpip/transport/tcp/dispatcher.go | 380 ++++++++-- pkg/tcpip/transport/tcp/endpoint.go | 530 +++++++------- pkg/tcpip/transport/tcp/endpoint_state.go | 98 ++- pkg/tcpip/transport/tcp/forwarder.go | 4 - pkg/tcpip/transport/tcp/protocol.go | 10 + pkg/tcpip/transport/tcp/rack.go | 23 +- pkg/tcpip/transport/tcp/rcv.go | 17 +- pkg/tcpip/transport/tcp/snd.go | 36 +- pkg/tcpip/transport/tcp/test/e2e/tcp_test.go | 35 +- .../transport/tcp/testing/context/context.go | 2 +- pkg/tcpip/transport/tcp/timer.go | 25 +- pkg/tcpip/transport/tcp/timer_test.go | 2 +- pkg/tcpip/transport/udp/endpoint.go | 2 - pkg/tcpip/transport/udp/protocol.go | 6 + 33 files changed, 984 insertions(+), 1157 deletions(-) diff --git a/pkg/refsvfs2/refs_map.go b/pkg/refsvfs2/refs_map.go index 8bc3ef377..186f4fc7d 100644 --- a/pkg/refsvfs2/refs_map.go +++ b/pkg/refsvfs2/refs_map.go @@ -47,9 +47,9 @@ func init() { liveObjects = make(map[CheckedObject]struct{}) } -// leakCheckEnabled returns whether leak checking is enabled. The following +// LeakCheckEnabled returns whether leak checking is enabled. The following // functions should only be called if it returns true. -func leakCheckEnabled() bool { +func LeakCheckEnabled() bool { return refs_vfs1.GetLeakMode() != refs_vfs1.NoLeakChecking } @@ -61,14 +61,14 @@ func leakCheckPanicEnabled() bool { // Register adds obj to the live object map. func Register(obj CheckedObject) { - if leakCheckEnabled() { + if LeakCheckEnabled() { liveObjectsMu.Lock() if _, ok := liveObjects[obj]; ok { panic(fmt.Sprintf("Unexpected entry in leak checking map: reference %p already added", obj)) } liveObjects[obj] = struct{}{} liveObjectsMu.Unlock() - if leakCheckEnabled() && obj.LogRefs() { + if LeakCheckEnabled() && obj.LogRefs() { logEvent(obj, "registered") } } @@ -76,14 +76,14 @@ func Register(obj CheckedObject) { // Unregister removes obj from the live object map. func Unregister(obj CheckedObject) { - if leakCheckEnabled() { + if LeakCheckEnabled() { liveObjectsMu.Lock() defer liveObjectsMu.Unlock() if _, ok := liveObjects[obj]; !ok { panic(fmt.Sprintf("Expected to find entry in leak checking map for reference %p", obj)) } delete(liveObjects, obj) - if leakCheckEnabled() && obj.LogRefs() { + if LeakCheckEnabled() && obj.LogRefs() { logEvent(obj, "unregistered") } } @@ -91,21 +91,21 @@ func Unregister(obj CheckedObject) { // LogIncRef logs a reference increment. func LogIncRef(obj CheckedObject, refs int64) { - if leakCheckEnabled() && obj.LogRefs() { + if LeakCheckEnabled() && obj.LogRefs() { logEvent(obj, fmt.Sprintf("IncRef to %d", refs)) } } // LogTryIncRef logs a successful TryIncRef call. func LogTryIncRef(obj CheckedObject, refs int64) { - if leakCheckEnabled() && obj.LogRefs() { + if LeakCheckEnabled() && obj.LogRefs() { logEvent(obj, fmt.Sprintf("TryIncRef to %d", refs)) } } // LogDecRef logs a reference decrement. func LogDecRef(obj CheckedObject, refs int64) { - if leakCheckEnabled() && obj.LogRefs() { + if LeakCheckEnabled() && obj.LogRefs() { logEvent(obj, fmt.Sprintf("DecRef to %d", refs)) } } @@ -128,7 +128,7 @@ var checkOnce sync.Once // anymore, at which point anything left in the map is considered a leak. On // multiple calls, only the first call will perform the leak check. func DoLeakCheck() { - if leakCheckEnabled() { + if LeakCheckEnabled() { checkOnce.Do(doLeakCheck) } } @@ -136,7 +136,7 @@ func DoLeakCheck() { // DoRepeatedLeakCheck is the same as DoLeakCheck except that it can be called // multiple times by the caller to incrementally perform leak checking. func DoRepeatedLeakCheck() { - if leakCheckEnabled() { + if LeakCheckEnabled() { doLeakCheck() } } diff --git a/pkg/sentry/inet/inet.go b/pkg/sentry/inet/inet.go index a76ab1087..b80e07679 100644 --- a/pkg/sentry/inet/inet.go +++ b/pkg/sentry/inet/inet.go @@ -79,6 +79,9 @@ type Stack interface { // RouteTable returns the network stack's route table. RouteTable() []Route + // Pause pauses the network stack before save. + Pause() + // Resume restarts the network stack after restore. Resume() diff --git a/pkg/sentry/inet/test_stack.go b/pkg/sentry/inet/test_stack.go index 621f47e1f..fef7391b9 100644 --- a/pkg/sentry/inet/test_stack.go +++ b/pkg/sentry/inet/test_stack.go @@ -144,6 +144,9 @@ func (s *TestStack) RouteTable() []Route { return s.RouteList } +// Pause implements Stack. +func (s *TestStack) Pause() {} + // Resume implements Stack. func (s *TestStack) Resume() {} diff --git a/pkg/sentry/kernel/kernel.go b/pkg/sentry/kernel/kernel.go index 2dcc5be95..82926c5d2 100644 --- a/pkg/sentry/kernel/kernel.go +++ b/pkg/sentry/kernel/kernel.go @@ -552,6 +552,15 @@ func (k *Kernel) SaveTo(ctx context.Context, w wire.Writer) error { // Save the timekeeper's state. + if rootNS := k.rootNetworkNamespace; rootNS != nil && rootNS.Stack() != nil { + // Pause the network stack. + netstackPauseStart := time.Now() + log.Infof("Pausing root network namespace") + k.rootNetworkNamespace.Stack().Pause() + defer k.rootNetworkNamespace.Stack().Resume() + log.Infof("Pausing root network namespace took [%s].", time.Since(netstackPauseStart)) + } + // Save the kernel state. kernelStart := time.Now() stats, err := state.Save(ctx, w, k) diff --git a/pkg/sentry/socket/hostinet/stack.go b/pkg/sentry/socket/hostinet/stack.go index e4598be82..3e176d80c 100644 --- a/pkg/sentry/socket/hostinet/stack.go +++ b/pkg/sentry/socket/hostinet/stack.go @@ -481,6 +481,9 @@ func (s *Stack) RouteTable() []inet.Route { return append([]inet.Route(nil), s.routes...) } +// Pause implements inet.Stack.Pause. +func (*Stack) Pause() {} + // Resume implements inet.Stack.Resume. func (*Stack) Resume() {} diff --git a/pkg/sentry/socket/netstack/stack.go b/pkg/sentry/socket/netstack/stack.go index ea199f223..36f377edf 100644 --- a/pkg/sentry/socket/netstack/stack.go +++ b/pkg/sentry/socket/netstack/stack.go @@ -444,6 +444,11 @@ func (s *Stack) IPTables() (*stack.IPTables, error) { return s.Stack.IPTables(), nil } +// Pause implements inet.Stack.Pause. +func (s *Stack) Pause() { + s.Stack.Pause() +} + // Resume implements inet.Stack.Resume. func (s *Stack) Resume() { s.Stack.Resume() diff --git a/pkg/tcpip/header/tcp.go b/pkg/tcpip/header/tcp.go index 568ee5d7e..391f6d5ae 100644 --- a/pkg/tcpip/header/tcp.go +++ b/pkg/tcpip/header/tcp.go @@ -134,6 +134,8 @@ type TCPFields struct { // TCPSynOptions is used to return the parsed TCP Options in a syn // segment. +// +// +stateify savable type TCPSynOptions struct { // MSS is the maximum segment size provided by the peer in the SYN. MSS uint16 diff --git a/pkg/tcpip/socketops.go b/pkg/tcpip/socketops.go index 9223afdd7..00622c408 100644 --- a/pkg/tcpip/socketops.go +++ b/pkg/tcpip/socketops.go @@ -55,8 +55,10 @@ type SocketOptionsHandler interface { // buffer size. It also returns the newly set value. OnSetSendBufferSize(v int64) (newSz int64) - // OnSetReceiveBufferSize is invoked by SO_RCVBUF and SO_RCVBUFFORCE. - OnSetReceiveBufferSize(v, oldSz int64) (newSz int64) + // OnSetReceiveBufferSize is invoked by SO_RCVBUF and SO_RCVBUFFORCE. The + // handler can optionally return a callback which will be called after + // the buffer size is updated to newSz. + OnSetReceiveBufferSize(v, oldSz int64) (newSz int64, postSet func()) // WakeupWriters is invoked when the send buffer size for an endpoint is // changed. The handler notifies the writers if the send buffer size is @@ -107,8 +109,8 @@ func (*DefaultSocketOptionsHandler) OnSetSendBufferSize(v int64) (newSz int64) { func (*DefaultSocketOptionsHandler) WakeupWriters() {} // OnSetReceiveBufferSize implements SocketOptionsHandler.OnSetReceiveBufferSize. -func (*DefaultSocketOptionsHandler) OnSetReceiveBufferSize(v, oldSz int64) (newSz int64) { - return v +func (*DefaultSocketOptionsHandler) OnSetReceiveBufferSize(v, oldSz int64) (newSz int64, postSet func()) { + return v, nil } // StackHandler holds methods to access the stack options. These must be @@ -700,11 +702,15 @@ func (so *SocketOptions) ReceiveBufferLimits() (min, max int64) { // SetReceiveBufferSize sets the value of the SO_RCVBUF option, optionally // notifying the owning endpoint. func (so *SocketOptions) SetReceiveBufferSize(receiveBufferSize int64, notify bool) { + var postSet func() if notify { oldSz := so.receiveBufferSize.Load() - receiveBufferSize = so.handler.OnSetReceiveBufferSize(receiveBufferSize, oldSz) + receiveBufferSize, postSet = so.handler.OnSetReceiveBufferSize(receiveBufferSize, oldSz) } so.receiveBufferSize.Store(receiveBufferSize) + if postSet != nil { + postSet() + } } // GetRcvlowat gets value for SO_RCVLOWAT option. diff --git a/pkg/tcpip/stack/registration.go b/pkg/tcpip/stack/registration.go index 72ed62621..ac6c82300 100644 --- a/pkg/tcpip/stack/registration.go +++ b/pkg/tcpip/stack/registration.go @@ -225,6 +225,13 @@ type TransportProtocol interface { // Wait waits for any worker goroutines owned by the protocol to stop. Wait() + // Pause requests that any protocol level background workers pause. + Pause() + + // Resume resumes any protocol level background workers that were + // previously paused by Pause. + Resume() + // Parse sets pkt.TransportHeader and trims pkt.Data appropriately. It does // neither and returns false if pkt.Data is too small, i.e. pkt.Data.Size() < // MinimumPacketSize() diff --git a/pkg/tcpip/stack/stack.go b/pkg/tcpip/stack/stack.go index e4a3aeb06..43e7c2a48 100644 --- a/pkg/tcpip/stack/stack.go +++ b/pkg/tcpip/stack/stack.go @@ -1606,6 +1606,13 @@ func (s *Stack) Wait() { } } +// Pause pauses any protocol level background workers. +func (s *Stack) Pause() { + for _, p := range s.transportProtocols { + p.proto.Pause() + } +} + // Resume restarts the stack after a restore. This must be called after the // entire system has been restored. func (s *Stack) Resume() { @@ -1618,6 +1625,10 @@ func (s *Stack) Resume() { for _, e := range eps { e.Resume(s) } + // Now resume any protocol level background workers. + for _, p := range s.transportProtocols { + p.proto.Resume() + } } // RegisterPacketEndpoint registers ep with the stack, causing it to receive diff --git a/pkg/tcpip/stack/transport_demuxer.go b/pkg/tcpip/stack/transport_demuxer.go index 088913b83..8996c25eb 100644 --- a/pkg/tcpip/stack/transport_demuxer.go +++ b/pkg/tcpip/stack/transport_demuxer.go @@ -181,22 +181,22 @@ func (epsByNIC *endpointsByNIC) handlePacket(id TransportEndpointID, pkt *Packet epsByNIC.mu.RUnlock() return true } + epsByNIC.mu.RUnlock() transEP.HandlePacket(id, pkt) - epsByNIC.mu.RUnlock() // Don't use defer for performance reasons. return true } // handleError delivers an error to the transport endpoint identified by id. func (epsByNIC *endpointsByNIC) handleError(n *nic, id TransportEndpointID, transErr TransportError, pkt *PacketBuffer) { epsByNIC.mu.RLock() - defer epsByNIC.mu.RUnlock() mpep, ok := epsByNIC.endpoints[n.ID()] if !ok { mpep, ok = epsByNIC.endpoints[0] } if !ok { + epsByNIC.mu.RUnlock() return } @@ -204,7 +204,10 @@ func (epsByNIC *endpointsByNIC) handleError(n *nic, id TransportEndpointID, tran // broadcast like we are doing with handlePacket above? // multiPortEndpoints are guaranteed to have at least one element. - mpep.selectEndpoint(id, epsByNIC.seed).HandleError(transErr, pkt) + transEP := mpep.selectEndpoint(id, epsByNIC.seed) + epsByNIC.mu.RUnlock() + + transEP.HandleError(transErr, pkt) } // registerEndpoint returns true if it succeeds. It fails and returns diff --git a/pkg/tcpip/stack/transport_test.go b/pkg/tcpip/stack/transport_test.go index c817e9de0..13c2afb77 100644 --- a/pkg/tcpip/stack/transport_test.go +++ b/pkg/tcpip/stack/transport_test.go @@ -331,6 +331,12 @@ func (*fakeTransportProtocol) Close() {} // Wait implements TransportProtocol.Wait. func (*fakeTransportProtocol) Wait() {} +// Pause implements TransportProtocol.Pause. +func (*fakeTransportProtocol) Pause() {} + +// Resume implements TransportProtocol.Resume. +func (*fakeTransportProtocol) Resume() {} + // Parse implements TransportProtocol.Parse. func (*fakeTransportProtocol) Parse(pkt *stack.PacketBuffer) bool { if _, ok := pkt.TransportHeader().Consume(fakeTransHeaderLen); ok { diff --git a/pkg/tcpip/tcpip.go b/pkg/tcpip/tcpip.go index 63096ce0a..be97aab46 100644 --- a/pkg/tcpip/tcpip.go +++ b/pkg/tcpip/tcpip.go @@ -674,9 +674,6 @@ type Endpoint interface { // SocketOptions returns the structure which contains all the socket // level options. SocketOptions() *SocketOptions - - // Release releases all reference counted objects held by the endpoint. - Release() } // LinkPacketInfo holds Link layer information for a received packet. @@ -2499,7 +2496,7 @@ func ReleaseDanglingEndpoints() { // Calling Release on a dangling endpoint that has been deleted is a noop. eps := GetDanglingEndpoints() for _, ep := range eps { - ep.Release() + ep.Abort() } } diff --git a/pkg/tcpip/transport/icmp/endpoint.go b/pkg/tcpip/transport/icmp/endpoint.go index f86fcc672..1fcc4d0ac 100644 --- a/pkg/tcpip/transport/icmp/endpoint.go +++ b/pkg/tcpip/transport/icmp/endpoint.go @@ -767,9 +767,6 @@ func (e *endpoint) HandlePacket(id stack.TransportEndpointID, pkt *stack.PacketB // HandleError implements stack.TransportEndpoint. func (*endpoint) HandleError(stack.TransportError, *stack.PacketBuffer) {} -// Release implements stack.TransportEndpoint. -func (*endpoint) Release() {} - // State implements tcpip.Endpoint.State. The ICMP endpoint currently doesn't // expose internal socket state. func (e *endpoint) State() uint32 { diff --git a/pkg/tcpip/transport/icmp/protocol.go b/pkg/tcpip/transport/icmp/protocol.go index fa82affc1..5d4bbb10c 100644 --- a/pkg/tcpip/transport/icmp/protocol.go +++ b/pkg/tcpip/transport/icmp/protocol.go @@ -121,6 +121,12 @@ func (*protocol) Close() {} // Wait implements stack.TransportProtocol.Wait. func (*protocol) Wait() {} +// Pause implements stack.TransportProtocol.Pause. +func (*protocol) Pause() {} + +// Resume implements stack.TransportProtocol.Resume. +func (*protocol) Resume() {} + // Parse implements stack.TransportProtocol.Parse. func (*protocol) Parse(pkt *stack.PacketBuffer) bool { // Right now, the Parse() method is tied to enabled protocols passed into diff --git a/pkg/tcpip/transport/packet/endpoint.go b/pkg/tcpip/transport/packet/endpoint.go index 7fe039a46..7afc88c1a 100644 --- a/pkg/tcpip/transport/packet/endpoint.go +++ b/pkg/tcpip/transport/packet/endpoint.go @@ -492,9 +492,6 @@ func (ep *endpoint) Stats() tcpip.EndpointStats { // SetOwner implements tcpip.Endpoint.SetOwner. func (*endpoint) SetOwner(tcpip.PacketOwner) {} -// Release implements tcpip.Release. -func (*endpoint) Release() {} - // SocketOptions implements tcpip.Endpoint.SocketOptions. func (ep *endpoint) SocketOptions() *tcpip.SocketOptions { return &ep.ops diff --git a/pkg/tcpip/transport/raw/endpoint.go b/pkg/tcpip/transport/raw/endpoint.go index 15eb6a32e..c13919955 100644 --- a/pkg/tcpip/transport/raw/endpoint.go +++ b/pkg/tcpip/transport/raw/endpoint.go @@ -735,9 +735,6 @@ func (*endpoint) LastError() tcpip.Error { return nil } -// Release implements stack.TransportEndpoint.Release. -func (*endpoint) Release() {} - // SocketOptions implements tcpip.Endpoint.SocketOptions. func (e *endpoint) SocketOptions() *tcpip.SocketOptions { return &e.ops diff --git a/pkg/tcpip/transport/tcp/accept.go b/pkg/tcpip/transport/tcp/accept.go index ab61f0ed4..52085ab18 100644 --- a/pkg/tcpip/transport/tcp/accept.go +++ b/pkg/tcpip/transport/tcp/accept.go @@ -23,7 +23,6 @@ import ( "io" "time" - "gvisor.dev/gvisor/pkg/sleep" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" @@ -225,10 +224,12 @@ func (l *listenContext) createConnectingEndpoint(s *segment, rcvdSynOpts header. // handshake in progress, which includes the new endpoint in the SYN-RCVD // state. // -// On success, a handshake h is returned with h.ep.mu held. +// On success, a handshake h is returned. +// +// NOTE: h.ep.mu is not held and must be acquired if any state needs to be +// modified. // // Precondition: if l.listenEP != nil, l.listenEP.mu must be locked. -// +checklocksacquire:h.ep.mu func (l *listenContext) startHandshake(s *segment, opts header.TCPSynOptions, queue *waiter.Queue, owner tcpip.PacketOwner) (h *handshake, _ tcpip.Error) { // Create new endpoint. irs := s.sequenceNumber @@ -290,57 +291,56 @@ func (l *listenContext) startHandshake(s *segment, opts header.TCPSynOptions, qu h = ep.newPassiveHandshake(isn, irs, opts, deferAccept) h.listenEP = l.listenEP h.start() + h.ep.mu.Unlock() return h, nil } // performHandshake performs a TCP 3-way handshake. On success, the new -// established endpoint is returned with e.mu held. +// established endpoint is returned. // // Precondition: if l.listenEP != nil, l.listenEP.mu must be locked. func (l *listenContext) performHandshake(s *segment, opts header.TCPSynOptions, queue *waiter.Queue, owner tcpip.PacketOwner) (*endpoint, tcpip.Error) { + waitEntry, notifyCh := waiter.NewChannelEntry(waiter.WritableEvents) + queue.EventRegister(&waitEntry) + defer queue.EventUnregister(&waitEntry) + h, err := l.startHandshake(s, opts, queue, owner) if err != nil { return nil, err } - ep := h.ep - // N.B. the endpoint is generated above by startHandshake, and will be - // returned locked. This first call is forced. - if err := h.complete(); err != nil { // +checklocksforce + // performHandshake is used by the Forwarder which will block till the + // handshake either succeeds or fails. We do this by registering for + // events above and block on the notification channel. + <-notifyCh + + ep := h.ep + ep.mu.Lock() + if !ep.EndpointState().connected() { ep.stack.Stats().TCP.FailedConnectionAttempts.Increment() ep.stats.FailedConnectionAttempts.Increment() - l.cleanupFailedHandshake(h) + ep.h = nil + ep.mu.Unlock() + ep.Close() + ep.notifyAborted() + ep.drainClosingSegmentQueue() return nil, err } - l.cleanupCompletedHandshake(h) - return ep, nil -} -// +checklocks:h.ep.mu -func (l *listenContext) cleanupFailedHandshake(h *handshake) { - e := h.ep - e.mu.Unlock() - e.Close() - e.notifyAborted() - e.drainClosingSegmentQueue() - e.h = nil -} - -// cleanupCompletedHandshake transfers any state from the completed handshake to -// the new endpoint. -// -// +checklocks:h.ep.mu -func (l *listenContext) cleanupCompletedHandshake(h *handshake) { - e := h.ep - e.isConnectNotified = true + ep.isConnectNotified = true + // Transfer any state from the completed handshake to the endpoint. + // // Update the receive window scaling. We can't do it before the // handshake because it's possible that the peer doesn't support window // scaling. - e.rcv.RcvWndScale = e.h.effectiveRcvWndScale() + ep.rcv.RcvWndScale = ep.h.effectiveRcvWndScale() - // Clean up handshake state stored in the endpoint so that it can be GCed. - e.h = nil + // Clean up handshake state stored in the endpoint so that it can be + // GCed. + ep.h = nil + ep.mu.Unlock() + return ep, nil } // propagateInheritableOptionsLocked propagates any options set on the listening @@ -418,7 +418,7 @@ type acceptQueue struct { } func (a *acceptQueue) isFull() bool { - return a.endpoints.Len() == a.capacity + return a.endpoints.Len() >= a.capacity } // handleListenSegment is called when a listening endpoint receives a segment @@ -478,59 +478,7 @@ func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Err e.stats.FailedConnectionAttempts.Increment() return false, err } - e.acceptQueue.pendingEndpoints[h.ep] = struct{}{} - e.pendingAccepted.Add(1) - - go func() { - defer func() { - e.pendingAccepted.Done() - - e.acceptMu.Lock() - defer e.acceptMu.Unlock() - delete(e.acceptQueue.pendingEndpoints, h.ep) - }() - - // Note that startHandshake returns a locked endpoint. The force call - // here just makes it so. - if err := h.complete(); err != nil { // +checklocksforce - e.stack.Stats().TCP.FailedConnectionAttempts.Increment() - e.stats.FailedConnectionAttempts.Increment() - ctx.cleanupFailedHandshake(h) - return - } - ctx.cleanupCompletedHandshake(h) - h.ep.startAcceptedLoop() - e.stack.Stats().TCP.PassiveConnectionOpenings.Increment() - - // Deliver the endpoint to the accept queue. - // - // Drop the lock before notifying to avoid deadlock in user-specified - // callbacks. - delivered := func() bool { - e.acceptMu.Lock() - defer e.acceptMu.Unlock() - for { - // The listener is transitioning out of the Listen state; bail. - if e.acceptQueue.capacity == 0 { - return false - } - if e.acceptQueue.isFull() { - e.acceptCond.Wait() - continue - } - - e.acceptQueue.endpoints.PushBack(h.ep) - return true - } - }() - - if delivered { - e.waiterQueue.Notify(waiter.ReadableEvents) - } else { - h.ep.notifyProtocolGoroutine(notifyReset) - } - }() return false, nil }() @@ -711,15 +659,14 @@ func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Err } h.ep.AssertLockHeld(n) h.transitionToStateEstablishedLocked(s) + n.mu.Unlock() // Requeue the segment if the ACK completing the handshake has more info // to be procesed by the newly established endpoint. if (s.flags.Contains(header.TCPFlagFin) || s.data.Size() > 0) && n.enqueueSegment(s) { - n.newSegmentWaker.Assert() + n.notifyProcessor() } - // Start the protocol goroutine. - n.startAcceptedLoop() e.stack.Stats().TCP.PassiveConnectionOpenings.Increment() // Deliver the endpoint to the accept queue. @@ -734,80 +681,3 @@ func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Err return nil } } - -// protocolListenLoop is the main loop of a listening TCP endpoint. It runs in -// its own goroutine and is responsible for handling connection requests. -func (e *endpoint) protocolListenLoop(rcvWnd seqnum.Size) { - e.mu.Lock() - v6Only := e.ops.GetV6Only() - ctx := newListenContext(e.stack, e.protocol, e, rcvWnd, v6Only, e.NetProto) - - defer func() { - e.setEndpointState(StateClose) - - // Do cleanup if needed. - e.completeWorkerLocked() - - if e.drainDone != nil { - close(e.drainDone) - } - e.mu.Unlock() - - e.drainClosingSegmentQueue() - - // Notify waiters that the endpoint is shutdown. - e.waiterQueue.Notify(waiter.ReadableEvents | waiter.WritableEvents | waiter.EventHUp | waiter.EventErr) - }() - - var s sleep.Sleeper - s.AddWaker(&e.notificationWaker) - s.AddWaker(&e.newSegmentWaker) - defer s.Done() - for { - e.mu.Unlock() - w := s.Fetch(true) - e.mu.Lock() - switch w { - case &e.notificationWaker: - n := e.fetchNotifications() - if n¬ifyClose != 0 { - return - } - if n¬ifyDrain != 0 { - for !e.segmentQueue.empty() { - s := e.segmentQueue.dequeue() - // TODO(gvisor.dev/issue/4690): Better handle errors instead of - // silently dropping. - _ = e.handleListenSegment(ctx, s) - s.DecRef() - } - close(e.drainDone) - e.mu.Unlock() - <-e.undrain - e.mu.Lock() - } - - case &e.newSegmentWaker: - // Process at most maxSegmentsPerWake segments. - mayRequeue := true - for i := 0; i < maxSegmentsPerWake; i++ { - s := e.segmentQueue.dequeue() - if s == nil { - mayRequeue = false - break - } - - // TODO(gvisor.dev/issue/4690): Better handle errors instead of - // silently dropping. - _ = e.handleListenSegment(ctx, s) - s.DecRef() - } - - // If the queue is not empty, make sure we'll wake up - // in the next iteration. - if mayRequeue && !e.segmentQueue.empty() { - e.newSegmentWaker.Assert() - } - } - } -} diff --git a/pkg/tcpip/transport/tcp/connect.go b/pkg/tcpip/transport/tcp/connect.go index dbc3ea599..e143b1ee7 100644 --- a/pkg/tcpip/transport/tcp/connect.go +++ b/pkg/tcpip/transport/tcp/connect.go @@ -16,10 +16,10 @@ package tcp import ( "encoding/binary" + "fmt" "math" "time" - "gvisor.dev/gvisor/pkg/sleep" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -60,6 +60,8 @@ const ( // // NOTE: handshake.ep.mu is held during handshake processing. It is released if // we are going to block and reacquired when we start processing an event. +// +// +stateify savable type handshake struct { ep *endpoint listenEP *endpoint @@ -107,6 +109,65 @@ type handshake struct { // tell; then RTT can only be sampled when the incoming segment has timestamp // options enabled. sampleRTTWithTSOnly bool + + // retransmitTimer is used to retransmit SYN/SYN-ACK with exponential backoff + // till handshake is either completed or timesout. + 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. + +// NOTE: e.mu is held for the duration of the call to f(). +func maybeFailTimerHandler(e *endpoint, f func() tcpip.Error) func() { + return func() { + e.mu.Lock() + if err := f(); err != nil { + e.lastErrorMu.Lock() + e.lastError = err + e.lastErrorMu.Unlock() + e.hardError = err + e.stack.Stats().TCP.CurrentConnected.Decrement() + e.cleanupLocked() + e.setEndpointState(StateError) + e.mu.Unlock() + e.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) + return + } + 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) + } + } +} + +// 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 @@ -124,6 +185,11 @@ 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)) + if err != nil { + panic(fmt.Sprintf("newBackOffTimer(_, %s, %s, _) failed: %s", InitialRTO, MaxRTO, err)) + } + h.retransmitTimer = timer return h } @@ -242,7 +308,6 @@ func (h *handshake) synSentState(s *segment) tcpip.Error { // RFC 793, page 67, states that "If the RST bit is set [and] If the ACK // was acceptable then signal the user "error: connection reset", drop // the segment, enter CLOSED state, delete TCB, and return." - h.ep.workerCleanup = true // Although the RFC above calls out ECONNRESET, Linux actually returns // ECONNREFUSED here so we do as well. return &tcpip.ErrConnectionRefused{} @@ -417,13 +482,13 @@ func (h *handshake) synRcvdState(s *segment) tcpip.Error { } h.state = handshakeCompleted - h.transitionToStateEstablishedLocked(s) // Requeue the segment if the ACK completing the handshake has more info // to be procesed by the newly established endpoint. if (s.flags.Contains(header.TCPFlagFin) || s.data.Size() > 0) && h.ep.enqueueSegment(s) { - h.ep.newSegmentWaker.Assert() + h.ep.protocol.dispatcher.selectProcessor(h.ep.ID).queueEndpoint(h.ep) + } return nil } @@ -471,12 +536,6 @@ func (h *handshake) processSegments() tcpip.Error { } } - // If the queue is not empty, make sure we'll wake up in the next - // iteration. - if !h.ep.segmentQueue.empty() { - h.ep.newSegmentWaker.Assert() - } - return nil } @@ -525,99 +584,45 @@ func (h *handshake) start() { }, synOpts) } -// complete completes the TCP 3-way handshake initiated by h.start(). +// retransmitHandler handles retransmissions of un-acked SYNs. // +checklocks:h.ep.mu -func (h *handshake) complete() tcpip.Error { - // Set up the wakers. - var s sleep.Sleeper - resendWaker := sleep.Waker{} - s.AddWaker(&resendWaker) - s.AddWaker(&h.ep.notificationWaker) - s.AddWaker(&h.ep.newSegmentWaker) - defer s.Done() +func (h *handshake) retransmitHandlerLocked() tcpip.Error { + e := h.ep + // If the endpoint has already transition out of a connecting state due + // to say an error (e.g) peer send RST or an ICMP error. Then just + // return. Any required cleanup should have been done when the RST/error + // was handled. + if !e.EndpointState().connecting() { + return nil + } - // Initialize the resend timer. - timer, err := newBackoffTimer(h.ep.stack.Clock(), InitialRTO, MaxRTO, resendWaker.Assert) - if err != nil { + if err := h.retransmitTimer.reset(); err != nil { return err } - defer timer.stop() - for h.state != handshakeCompleted { - // Unlock before blocking, and reacquire again afterwards (h.ep.mu is held - // throughout handshake processing). - h.ep.mu.Unlock() - w := s.Fetch(true /* block */) - h.ep.mu.Lock() - switch w { - case &resendWaker: - if err := timer.reset(); err != nil { - return err - } - // Resend the SYN/SYN-ACK only if the following conditions hold. - // - It's an active handshake (deferAccept does not apply) - // - It's a passive handshake and we have not yet got the final-ACK. - // - It's a passive handshake and we got an ACK but deferAccept is - // enabled and we are now past the deferAccept duration. - // The last is required to provide a way for the peer to complete - // the connection with another ACK or data (as ACKs are never - // retransmitted on their own). - if h.active || !h.acked || h.deferAccept != 0 && h.ep.stack.Clock().NowMonotonic().Sub(h.startTime) > h.deferAccept { - h.ep.sendSynTCP(h.ep.route, tcpFields{ - id: h.ep.TransportEndpointInfo.ID, - ttl: calculateTTL(h.ep.route, h.ep.ipv4TTL, h.ep.ipv6HopLimit), - tos: h.ep.sendTOS, - flags: h.flags, - seq: h.iss, - ack: h.ackNum, - rcvWnd: h.rcvWnd, - }, h.sendSYNOpts) - // If we have ever retransmitted the SYN-ACK or - // SYN segment, we should only measure RTT if - // TS option is present. - h.sampleRTTWithTSOnly = true - } - case &h.ep.notificationWaker: - n := h.ep.fetchNotifications() - if (n¬ifyClose)|(n¬ifyAbort) != 0 { - return &tcpip.ErrAborted{} - } - if n¬ifyShutdown != 0 { - return &tcpip.ErrConnectionReset{} - } - if n¬ifyDrain != 0 { - for !h.ep.segmentQueue.empty() { - s := h.ep.segmentQueue.dequeue() - err := h.handleSegment(s) - s.DecRef() - if err != nil { - return err - } - if h.state == handshakeCompleted { - return nil - } - } - close(h.ep.drainDone) - h.ep.mu.Unlock() - <-h.ep.undrain - h.ep.mu.Lock() - } - // Check for any ICMP errors notified to us. - if n¬ifyError != 0 { - if err := h.ep.lastErrorLocked(); err != nil { - return err - } - // Flag the handshake failure as aborted if the lastError is - // cleared because of a socket layer call. - return &tcpip.ErrConnectionAborted{} - } - case &h.ep.newSegmentWaker: - if err := h.processSegments(); err != nil { - return err - } - } + // Resend the SYN/SYN-ACK only if the following conditions hold. + // - It's an active handshake (deferAccept does not apply) + // - It's a passive handshake and we have not yet got the final-ACK. + // - It's a passive handshake and we got an ACK but deferAccept is + // enabled and we are now past the deferAccept duration. + // The last is required to provide a way for the peer to complete + // the connection with another ACK or data (as ACKs are never + // retransmitted on their own). + if h.active || !h.acked || h.deferAccept != 0 && e.stack.Clock().NowMonotonic().Sub(h.startTime) > h.deferAccept { + e.sendSynTCP(e.route, tcpFields{ + id: e.TransportEndpointInfo.ID, + ttl: calculateTTL(e.route, e.ipv4TTL, e.ipv6HopLimit), + tos: e.sendTOS, + flags: h.flags, + seq: h.iss, + ack: h.ackNum, + rcvWnd: h.rcvWnd, + }, h.sendSYNOpts) + // If we have ever retransmitted the SYN-ACK or + // SYN segment, we should only measure RTT if + // TS option is present. + h.sampleRTTWithTSOnly = true } - return nil } @@ -626,6 +631,11 @@ func (h *handshake) complete() tcpip.Error { // initializes sender/receiver. // +checklocks:h.ep.mu func (h *handshake) transitionToStateEstablishedLocked(s *segment) { + // Stop the SYN retransmissions now that handshake is complete. + if h.retransmitTimer != nil { + h.retransmitTimer.stop() + } + // Transfer handshake state to TCP connection. We disable // receive window scaling if the peer doesn't support it // (indicated by a negative send window scale). @@ -654,6 +664,14 @@ func (h *handshake) transitionToStateEstablishedLocked(s *segment) { h.ep.rcvQueueInfo.rcvQueueMu.Unlock() h.ep.setEndpointState(StateEstablished) + + // Completing the 3-way handshake is an indication that the route is valid + // and the remote is reachable as the only way we can complete a handshake + // is if our SYN reached the remote and their ACK reached us. + h.ep.route.ConfirmReachable() + + // Tell waiters that the endpoint is connected and writable. + h.ep.waiterQueue.Notify(waiter.WritableEvents) } type backoffTimer struct { @@ -983,7 +1001,6 @@ func (e *endpoint) sendData(next *segment) { func (e *endpoint) resetConnectionLocked(err tcpip.Error) { // Only send a reset if the connection is being aborted for a reason // other than receiving a reset. - e.setEndpointState(StateError) e.hardError = err switch err.(type) { case *tcpip.ErrConnectionReset, *tcpip.ErrTimeout: @@ -1006,20 +1023,8 @@ func (e *endpoint) resetConnectionLocked(err tcpip.Error) { // to be read. e.purgeWriteQueue() e.purgePendingRcvQueue() -} - -// completeWorkerLocked is called by the worker goroutine when it's about to -// exit. -// +checklocks:e.mu -func (e *endpoint) completeWorkerLocked() { - // Worker is terminating(either due to moving to - // CLOSED or ERROR state, ensure we release all - // registrations port reservations even if the socket - // itself is not yet closed by the application. - e.workerRunning = false - if e.workerCleanup { - e.cleanupLocked() - } + e.cleanupLocked() + e.setEndpointState(StateError) } // transitionToStateCloseLocked ensures that the endpoint is @@ -1039,8 +1044,8 @@ func (e *endpoint) transitionToStateCloseLocked() { e.stack.Stats().TCP.EstablishedClosed.Increment() } - // Mark the endpoint as fully closed for reads/writes. e.cleanupLocked() + // Mark the endpoint as fully closed for reads/writes. e.setEndpointState(StateClose) } @@ -1060,16 +1065,18 @@ func (e *endpoint) tryDeliverSegmentFromClosedEndpoint(s *segment) { ) } if ep == nil { - replyWithReset(e.stack, s, stack.DefaultTOS, tcpip.UseDefaultIPv4TTL, tcpip.UseDefaultIPv6HopLimit) + if !s.flags.Contains(header.TCPFlagRst) { + replyWithReset(e.stack, s, stack.DefaultTOS, tcpip.UseDefaultIPv4TTL, tcpip.UseDefaultIPv6HopLimit) + } return } if e == ep { - panic("current endpoint not removed from demuxer, enqueing segments to itself") + panic(fmt.Sprintf("current endpoint not removed from demuxer, enqueing segments to itself, endpoint in state %v", e.EndpointState())) } if ep := ep.(*endpoint); ep.enqueueSegment(s) { - ep.newSegmentWaker.Assert() + ep.notifyProcessor() } } @@ -1118,7 +1125,6 @@ func (e *endpoint) handleReset(s *segment) (ok bool, err tcpip.Error) { case StateCloseWait: e.transitionToStateCloseLocked() e.hardError = &tcpip.ErrAborted{} - e.notifyProtocolGoroutine(notifyTickleWorker) return false, nil default: // RFC 793, page 37 states that "in all states @@ -1129,7 +1135,6 @@ func (e *endpoint) handleReset(s *segment) (ok bool, err tcpip.Error) { // Notify protocol goroutine. This is required when // handleSegment is invoked from the processor goroutine // rather than the worker goroutine. - e.notifyProtocolGoroutine(notifyResetByPeer) return false, &tcpip.ErrConnectionReset{} } } @@ -1140,19 +1145,16 @@ func (e *endpoint) handleReset(s *segment) (ok bool, err tcpip.Error) { // // +checklocks:e.mu // +checklocksalias:e.snd.ep.mu=e.mu -func (e *endpoint) handleSegmentsLocked(fastPath bool) tcpip.Error { - checkRequeue := true +func (e *endpoint) handleSegmentsLocked() tcpip.Error { sndUna := e.snd.SndUna for i := 0; i < maxSegmentsPerWake; i++ { - if state := e.EndpointState(); state.closed() || state == StateTimeWait { + if state := e.EndpointState(); state.closed() || state == StateTimeWait || state == StateError { return nil } s := e.segmentQueue.dequeue() if s == nil { - checkRequeue = false break } - cont, err := e.handleSegmentLocked(s) s.DecRef() if err != nil { @@ -1173,12 +1175,6 @@ func (e *endpoint) handleSegmentsLocked(fastPath bool) tcpip.Error { if sndUna.LessThan(e.snd.SndUna) { e.route.ConfirmReachable() } - // When fastPath is true we don't want to wake up the worker - // goroutine. If the endpoint has more segments to process the - // dispatcher will call handleSegments again anyway. - if !fastPath && checkRequeue && !e.segmentQueue.empty() { - e.newSegmentWaker.Assert() - } // Send an ACK for all processed packets if needed. if e.rcv.RcvNxt != e.snd.MaxSentAck { @@ -1282,7 +1278,7 @@ func (e *endpoint) keepaliveTimerExpired() tcpip.Error { userTimeout := e.userTimeout e.keepalive.Lock() - if !e.SocketOptions().GetKeepAlive() || !e.keepalive.timer.checkExpiration() { + if !e.SocketOptions().GetKeepAlive() || e.keepalive.timer.isZero() || !e.keepalive.timer.checkExpiration() { e.keepalive.Unlock() return nil } @@ -1339,305 +1335,31 @@ func (e *endpoint) disableKeepaliveTimer() { e.keepalive.Unlock() } -// protocolMainLoopDone is called at the end of protocolMainLoop. -// +checklocksrelease:e.mu -func (e *endpoint) protocolMainLoopDone(closeTimer tcpip.Timer) { - if e.snd != nil { - e.snd.resendTimer.cleanup() - e.snd.probeTimer.cleanup() - e.snd.reorderTimer.cleanup() - } - - if closeTimer != nil { - closeTimer.Stop() - } - - e.completeWorkerLocked() - - if e.drainDone != nil { - close(e.drainDone) - } - +// finWait2TimerExpired is called when the FIN-WAIT-2 timeout is hit +// and the peer hasn't sent us a FIN. +func (e *endpoint) finWait2TimerExpired() { + e.mu.Lock() + e.transitionToStateCloseLocked() e.mu.Unlock() - e.drainClosingSegmentQueue() - - // When the protocol loop exits we should wake up our waiters. e.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) } -// handleWakeup handles a wakeup event while connected. -// // +checklocks:e.mu -// +checklocksalias:e.snd.ep.mu=e.mu -// +checklocksalias:e.rcv.ep.mu=e.mu -// +checklocksalias:e.snd.rc.snd.ep.mu=e.mu -func (e *endpoint) handleWakeup(w, closeWaker *sleep.Waker, closeTimer *tcpip.Timer) tcpip.Error { - switch w { - case &e.sndQueueInfo.sndWaker: - e.sendData(nil /* next */) - case &e.newSegmentWaker: - return e.handleSegmentsLocked(false /* fastPath */) - case &e.snd.resendWaker: - if !e.snd.retransmitTimerExpired() { - e.stack.Stats().TCP.EstablishedTimedout.Increment() - return &tcpip.ErrTimeout{} - } - case closeWaker: - // This means the socket is being closed due to the - // TCP-FIN-WAIT2 timeout was hit. Just mark the socket as - // closed. - e.transitionToStateCloseLocked() - e.workerCleanup = true - case &e.snd.probeWaker: - return e.snd.probeTimerExpired() - case &e.keepalive.waker: - return e.keepaliveTimerExpired() - case &e.notificationWaker: - n := e.fetchNotifications() - if n¬ifyNonZeroReceiveWindow != 0 { - e.rcv.nonZeroWindow() - } - - if n¬ifyMTUChanged != 0 { - e.sndQueueInfo.sndQueueMu.Lock() - count := e.sndQueueInfo.PacketTooBigCount - e.sndQueueInfo.PacketTooBigCount = 0 - mtu := e.sndQueueInfo.SndMTU - e.sndQueueInfo.sndQueueMu.Unlock() - - e.snd.updateMaxPayloadSize(mtu, count) - } - - if n¬ifyReset != 0 || n¬ifyAbort != 0 { - return &tcpip.ErrConnectionAborted{} - } - - if n¬ifyResetByPeer != 0 { - return &tcpip.ErrConnectionReset{} - } - - if n¬ifyClose != 0 && e.closed { - switch e.EndpointState() { - case StateEstablished: - // Perform full shutdown if the endpoint is - // still established. This can occur when - // notifyClose was asserted just before - // becoming established. - e.shutdownLocked(tcpip.ShutdownWrite | tcpip.ShutdownRead) - case StateFinWait2: - // The socket has been closed and we are in - // FIN_WAIT2 so start the FIN_WAIT2 timer. - if *closeTimer == nil { - *closeTimer = e.stack.Clock().AfterFunc(e.tcpLingerTimeout, closeWaker.Assert) - } - } - } - - if n¬ifyKeepaliveChanged != 0 { - // The timer could fire in background when the endpoint - // is drained. That's OK. See above. - e.resetKeepaliveTimer(true) - } - - if n¬ifyDrain != 0 { - for !e.segmentQueue.empty() { - if err := e.handleSegmentsLocked(false /* fastPath */); err != nil { - return err - } - } - if !e.EndpointState().closed() { - // Only block the worker if the endpoint - // is not in closed state or error state. - close(e.drainDone) - e.mu.Unlock() - <-e.undrain - e.mu.Lock() - } - } - - // N.B. notifyTickleWorker may be set, but there is no action - // to take in this case. - case &e.snd.reorderWaker: - return e.snd.rc.reorderTimerExpired() - default: - panic("unknown waker") // Shouldn't happen. - } - return nil -} - -// protocolMainLoop is the main loop of the TCP protocol. It runs in its own -// goroutine and is responsible for sending segments and handling received -// segments. -func (e *endpoint) protocolMainLoop(handshake bool, wakerInitDone chan<- struct{}) { - var ( - closeTimer tcpip.Timer - closeWaker sleep.Waker - ) - - e.mu.Lock() - if handshake { - if err := e.h.complete(); err != nil { // +checklocksforce - e.lastErrorMu.Lock() - e.lastError = err - e.lastErrorMu.Unlock() - - e.setEndpointState(StateError) - e.hardError = err - - e.workerCleanup = true - e.protocolMainLoopDone(closeTimer) - return - } - } - - // Reaching this point means that we successfully completed the 3-way - // handshake with our peer. The current endpoint state could be any state - // post ESTABLISHED, including CLOSED or ERROR if the endpoint processes a - // RST from the peer via the dispatcher fast path, before the loop is - // started. - if s := e.EndpointState(); !s.connected() { - switch s { - case StateClose, StateError: - // If the endpoint is in CLOSED/ERROR state, sender state has to be - // initialized if the endpoint was previously established. - if e.snd != nil { - break - } - fallthrough - default: - panic("endpoint was not established, current state " + s.String()) - } - } - - // Completing the 3-way handshake is an indication that the route is valid - // and the remote is reachable as the only way we can complete a handshake - // is if our SYN reached the remote and their ACK reached us. - e.route.ConfirmReachable() - - drained := e.drainDone != nil - if drained { - close(e.drainDone) - e.mu.Unlock() - <-e.undrain - e.mu.Lock() - } - - // Add all wakers. - var s sleep.Sleeper - s.AddWaker(&e.sndQueueInfo.sndWaker) - s.AddWaker(&e.newSegmentWaker) - s.AddWaker(&e.snd.resendWaker) - s.AddWaker(&e.snd.probeWaker) - s.AddWaker(&closeWaker) - s.AddWaker(&e.keepalive.waker) - s.AddWaker(&e.notificationWaker) - s.AddWaker(&e.snd.reorderWaker) - - // Notify the caller that the waker initialization is complete and the - // endpoint is ready. - if wakerInitDone != nil { - close(wakerInitDone) - } - - // Tell waiters that the endpoint is connected and writable. - e.waiterQueue.Notify(waiter.WritableEvents) - - // The following assertions and notifications are needed for restored - // endpoints. Fresh newly created endpoints have empty states and should - // not invoke any. - if !e.segmentQueue.empty() { - e.newSegmentWaker.Assert() - } - - e.rcvQueueInfo.rcvQueueMu.Lock() - if !e.rcvQueueInfo.rcvQueue.Empty() { - e.waiterQueue.Notify(waiter.ReadableEvents) - } - e.rcvQueueInfo.rcvQueueMu.Unlock() - - if e.workerCleanup { - e.notifyProtocolGoroutine(notifyClose) - } - - // Main loop. Handle segments until both send and receive ends of the - // connection have completed. - cleanupOnError := func(err tcpip.Error) { - e.stack.Stats().TCP.CurrentConnected.Decrement() - e.workerCleanup = true - if err != nil { - e.resetConnectionLocked(err) - e.releaseLocked() - } - } - -loop: - for { - switch e.EndpointState() { - case StateTimeWait, StateClose, StateError: - break loop - } - - e.mu.Unlock() - w := s.Fetch(true /* block */) - e.mu.Lock() - - // We need to double check here because the notification may be - // stale by the time we got around to processing it. - switch e.EndpointState() { - case StateError: - // If the endpoint has already transitioned to an ERROR - // state just pass nil here as any reset that may need - // to be sent etc should already have been done and we - // just want to terminate the loop and cleanup the - // endpoint. - cleanupOnError(nil) - e.protocolMainLoopDone(closeTimer) - return - case StateTimeWait: - fallthrough - case StateClose: - break loop - default: - if err := e.handleWakeup(w, &closeWaker, &closeTimer); err != nil { - cleanupOnError(err) - e.protocolMainLoopDone(closeTimer) - return - } - } - } - - var reuseTW func() - if e.EndpointState() == StateTimeWait { - // Disable close timer as we now entering real TIME_WAIT. - if closeTimer != nil { - closeTimer.Stop() - } - // Mark the current sleeper done so as to free all associated - // wakers. - s.Done() - // Wake up any waiters before we enter TIME_WAIT. - e.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) - e.workerCleanup = true - reuseTW = e.doTimeWait() - } - - // Handle any StateError transition from StateTimeWait. - if e.EndpointState() == StateError { - cleanupOnError(nil) - e.protocolMainLoopDone(closeTimer) - return - } - - e.transitionToStateCloseLocked() - - e.protocolMainLoopDone(closeTimer) - - // A new SYN was received during TIME_WAIT and we need to abort - // the timewait and redirect the segment to the listener queue - if reuseTW != nil { - reuseTW() +func (e *endpoint) handshakeFailed(err tcpip.Error) { + e.lastErrorMu.Lock() + e.lastError = err + e.lastErrorMu.Unlock() + // handshakeFailed is also called from startHandshake when a listener + // transitions out of Listen state by the time the SYN is processed. In + // such cases the handshake is never initialized and the newly created + // endpoint is closed right away. + if e.h != nil && e.h.retransmitTimer != nil { + e.h.retransmitTimer.stop() } + e.hardError = err + e.cleanupLocked() + e.setEndpointState(StateError) } // handleTimeWaitSegments processes segments received during TIME_WAIT @@ -1645,11 +1367,9 @@ loop: // +checklocks:e.mu // +checklocksalias:e.rcv.ep.mu=e.mu func (e *endpoint) handleTimeWaitSegments() (extendTimeWait bool, reuseTW func()) { - checkRequeue := true for i := 0; i < maxSegmentsPerWake; i++ { s := e.segmentQueue.dequeue() if s == nil { - checkRequeue = false break } extTW, newSyn := e.rcv.handleTimeWaitSegment(s) @@ -1673,7 +1393,7 @@ func (e *endpoint) handleTimeWaitSegments() (extendTimeWait bool, reuseTW func() if !tcpEP.enqueueSegment(s) { return } - tcpEP.newSegmentWaker.Assert() + tcpEP.notifyProcessor() s.DecRef() } // We explicitly do not DecRef the segment as it's still valid and @@ -1688,22 +1408,11 @@ func (e *endpoint) handleTimeWaitSegments() (extendTimeWait bool, reuseTW func() } s.DecRef() } - if checkRequeue && !e.segmentQueue.empty() { - e.newSegmentWaker.Assert() - } return extendTimeWait, nil } -// doTimeWait is responsible for handling the TCP behaviour once a socket -// enters the TIME_WAIT state. Optionally it can return a closure that -// should be executed after releasing the endpoint registrations. This is -// done in cases where a new SYN is received during TIME_WAIT that carries -// a sequence number larger than one see on the connection. // +checklocks:e.mu -func (e *endpoint) doTimeWait() (twReuse func()) { - // Trigger a 2 * MSL time wait state. During this period - // we will drop all incoming segments. - // NOTE: On Linux this is not configurable and is fixed at 60 seconds. +func (e *endpoint) getTimeWaitDuration() time.Duration { timeWaitDuration := DefaultTCPTimeWaitTimeout // Get the stack wide configuration. @@ -1711,50 +1420,36 @@ func (e *endpoint) doTimeWait() (twReuse func()) { if err := e.stack.TransportProtocolOption(ProtocolNumber, &tcpTW); err == nil { timeWaitDuration = time.Duration(tcpTW) } + return timeWaitDuration +} - var s sleep.Sleeper - defer s.Done() - s.AddWaker(&e.newSegmentWaker) - s.AddWaker(&e.notificationWaker) +// timeWaitTimerExpired is called when an endpoint completes the required time +// (typically 2 * MSL unless configured to something else at a stack level) in +// TIME-WAIT state. +func (e *endpoint) timeWaitTimerExpired() { + e.mu.Lock() + if e.EndpointState() != StateTimeWait { + e.mu.Unlock() + return + } + e.transitionToStateCloseLocked() + e.mu.Unlock() + e.drainClosingSegmentQueue() + e.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) +} - var timeWaitWaker sleep.Waker - s.AddWaker(&timeWaitWaker) - timeWaitTimer := e.stack.Clock().AfterFunc(timeWaitDuration, timeWaitWaker.Assert) - defer timeWaitTimer.Stop() - - for { - e.mu.Unlock() - w := s.Fetch(true /* block */) - e.mu.Lock() - switch w { - case &e.newSegmentWaker: - extendTimeWait, reuseTW := e.handleTimeWaitSegments() - if reuseTW != nil { - return reuseTW - } - if extendTimeWait { - timeWaitTimer.Reset(timeWaitDuration) - } - case &e.notificationWaker: - n := e.fetchNotifications() - if n¬ifyAbort != 0 { - return nil - } - if n¬ifyDrain != 0 { - for !e.segmentQueue.empty() { - // Ignore extending TIME_WAIT during a - // save. For sockets in TIME_WAIT we just - // terminate the TIME_WAIT early. - e.handleTimeWaitSegments() - } - close(e.drainDone) - e.mu.Unlock() - <-e.undrain - e.mu.Lock() - return nil - } - case &timeWaitWaker: - return nil - } +// notifyProcessor queues this endpoint for processing to its TCP processor. +func (e *endpoint) notifyProcessor() { + // We use TryLock here to avoid deadlocks in cases where a listening endpoint that is being + // closed tries to abort half completed connections which in turn try to queue any segments + // queued to that endpoint back to the same listening endpoint (because it may have got + // segments that matched its id but were either a RST or a new SYN which must be handled + // by a listening endpoint). In such cases the Close() on the listening endpoint will handle + // any queued segments after it releases the lock. + if !e.mu.TryLock() { + return } + processor := e.protocol.dispatcher.selectProcessor(e.ID) + e.mu.Unlock() + processor.queueEndpoint(e) } diff --git a/pkg/tcpip/transport/tcp/dispatcher.go b/pkg/tcpip/transport/tcp/dispatcher.go index a69e0e104..c066b29a6 100644 --- a/pkg/tcpip/transport/tcp/dispatcher.go +++ b/pkg/tcpip/transport/tcp/dispatcher.go @@ -16,6 +16,7 @@ package tcp import ( "encoding/binary" + "fmt" "math/rand" "gvisor.dev/gvisor/pkg/sleep" @@ -24,6 +25,7 @@ import ( "gvisor.dev/gvisor/pkg/tcpip/hash/jenkins" "gvisor.dev/gvisor/pkg/tcpip/header" "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/waiter" ) // epQueue is a queue of endpoints. @@ -35,13 +37,15 @@ type epQueue struct { // enqueue adds e to the queue if the endpoint is not already on the queue. func (q *epQueue) enqueue(e *endpoint) { q.mu.Lock() + defer q.mu.Unlock() + e.pendingProcessingMu.Lock() + defer e.pendingProcessingMu.Unlock() + if e.pendingProcessing { - q.mu.Unlock() return } q.list.PushBack(e) e.pendingProcessing = true - q.mu.Unlock() } // dequeue removes and returns the first element from the queue if available, @@ -50,7 +54,9 @@ func (q *epQueue) dequeue() *endpoint { q.mu.Lock() if e := q.list.Front(); e != nil { q.list.Remove(e) + e.pendingProcessingMu.Lock() e.pendingProcessing = false + e.pendingProcessingMu.Unlock() q.mu.Unlock() return e } @@ -72,6 +78,9 @@ type processor struct { sleeper sleep.Sleeper newEndpointWaker sleep.Waker closeWaker sleep.Waker + pauseWaker sleep.Waker + pauseChan chan struct{} + resumeChan chan struct{} } func (p *processor) close() { @@ -84,58 +93,258 @@ func (p *processor) queueEndpoint(ep *endpoint) { p.newEndpointWaker.Assert() } -const ( - newEndpointWaker = 1 - closeWaker = 2 -) +// deliverAccepted delivers a passively connected endpoint to the accept queue +// of its associated listening endpoint. +// +// +checklocks:ep.mu +func deliverAccepted(ep *endpoint) bool { + lEP := ep.h.listenEP + lEP.acceptMu.Lock() + // Remove endpoint from list of pendingEndpoints as the handshake is now + // complete. + delete(lEP.acceptQueue.pendingEndpoints, ep) + // Deliver this endpoint to the listening socket's accept queue. + if lEP.acceptQueue.capacity == 0 { + lEP.acceptMu.Unlock() + return false + } + + // NOTE: We always queue the endpoint and on purpose do not check if + // accept queue is full at this point. This is similar to linux because + // two racing incoming ACK's can both pass the acceptQueue.isFull check + // and proceed to ESTABLISHED state. In such a case its better to + // deliver both even if it temporarily exceeds the queue limit rather + // than drop a connection that is fully connected. + // + // For reference see: + // https://github.com/torvalds/linux/blob/169e77764adc041b1dacba84ea90516a895d43b2/net/ipv4/tcp_minisocks.c#L764 + // https://github.com/torvalds/linux/blob/169e77764adc041b1dacba84ea90516a895d43b2/net/ipv4/tcp_ipv4.c#L1500 + lEP.acceptQueue.endpoints.PushBack(ep) + lEP.acceptMu.Unlock() + ep.h.listenEP.waiterQueue.Notify(waiter.ReadableEvents) + + return true +} + +// handleConnecting is responsible for TCP processing for an endpoint in one of +// the connecting states. +func (p *processor) handleConnecting(ep *endpoint) { + if !ep.TryLock() { + return + } + cleanup := func() { + ep.mu.Unlock() + ep.drainClosingSegmentQueue() + ep.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) + } + if !ep.EndpointState().connecting() { + // If the endpoint has already transitioned out of a connecting + // stage then just return (only possible if it was closed or + // timed out by the time we got around to processing the wakeup. + ep.mu.Unlock() + return + } + if err := ep.h.processSegments(); err != nil { // +checklocksforce:ep.h.ep.mu + // handshake failed. clean up the tcp endpoint and handshake + // state. + ep.handshakeFailed(err) + cleanup() + return + } + + if ep.EndpointState() == StateEstablished && ep.h.listenEP != nil { + ep.isConnectNotified = true + ep.stack.Stats().TCP.PassiveConnectionOpenings.Increment() + if !deliverAccepted(ep) { + ep.resetConnectionLocked(&tcpip.ErrConnectionAborted{}) + cleanup() + return + } + } + ep.mu.Unlock() +} + +// handleConnected is responsible for TCP processing for an endpoint in one of +// the connected states(StateEstablished, StateFinWait1 etc.) +func (p *processor) handleConnected(ep *endpoint) { + if !ep.TryLock() { + return + } + + if !ep.EndpointState().connected() { + // If the endpoint has already transitioned out of a connected + // state then just return (only possible if it was closed or + // timed out by the time we got around to processing the wakeup. + ep.mu.Unlock() + return + } + + // NOTE: We read this outside of e.mu lock which means that by the time + // we get to handleSegments the endpoint may not be in ESTABLISHED. But + // this should be fine as all normal shutdown states are handled by + // handleSegmentsLocked. + switch err := ep.handleSegmentsLocked(); { + case err != nil: + // Send any active resets if required. + ep.resetConnectionLocked(err) + fallthrough + case ep.EndpointState() == StateClose: + ep.mu.Unlock() + ep.stack.Stats().TCP.CurrentConnected.Decrement() + ep.drainClosingSegmentQueue() + ep.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) + return + case ep.EndpointState() == StateTimeWait: + p.startTimeWait(ep) + } + ep.mu.Unlock() +} + +// startTimeWait starts a new goroutine to handle TIME-WAIT. +// +// +checklocks:ep.mu +func (p *processor) startTimeWait(ep *endpoint) { + // Disable close timer as we are now entering real TIME_WAIT. + if ep.finWait2Timer != nil { + ep.finWait2Timer.Stop() + } + // Wake up any waiters before we start TIME-WAIT. + ep.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) + timeWaitDuration := ep.getTimeWaitDuration() + ep.timeWaitTimer = ep.stack.Clock().AfterFunc(timeWaitDuration, ep.timeWaitTimerExpired) +} + +// handleTimeWait is responsible for TCP processing for an endpoint in TIME-WAIT +// state. +func (p *processor) handleTimeWait(ep *endpoint) { + if !ep.TryLock() { + return + } + + if ep.EndpointState() != StateTimeWait { + // If the endpoint has already transitioned out of a TIME-WAIT + // state then just return (only possible if it was closed or + // timed out by the time we got around to processing the wakeup. + ep.mu.Unlock() + return + } + + extendTimeWait, reuseTW := ep.handleTimeWaitSegments() + if reuseTW != nil { + ep.transitionToStateCloseLocked() + ep.mu.Unlock() + ep.drainClosingSegmentQueue() + ep.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) + reuseTW() + return + } + if extendTimeWait { + ep.timeWaitTimer.Reset(ep.getTimeWaitDuration()) + } + ep.mu.Unlock() +} + +// handleListen is responsible for TCP processing for an endpoint in LISTEN +// state. +func (p *processor) handleListen(ep *endpoint) { + if !ep.TryLock() { + return + } + defer ep.mu.Unlock() + + if ep.EndpointState() != StateListen { + // If the endpoint has already transitioned out of a LISTEN + // state then just return (only possible if it was closed or + // shutdown). + return + } + + for i := 0; i < maxSegmentsPerWake; i++ { + s := ep.segmentQueue.dequeue() + if s == nil { + break + } + + // TODO(gvisor.dev/issue/4690): Better handle errors instead of + // silently dropping. + _ = ep.handleListenSegment(ep.listenCtx, s) + s.DecRef() + } +} + +// start runs the main loop for a processor which is responsible for all TCP +// processing for TCP endpoints. func (p *processor) start(wg *sync.WaitGroup) { defer wg.Done() defer p.sleeper.Done() for { - if w := p.sleeper.Fetch(true); w == &p.closeWaker { - break - } - // If not the closeWaker, it must be &p.newEndpointWaker. - for { - ep := p.epQ.dequeue() - if ep == nil { - break - } - if ep.segmentQueue.empty() { + switch w := p.sleeper.Fetch(true); { + case w == &p.closeWaker: + return + case w == &p.pauseWaker: + if !p.epQ.empty() { + p.newEndpointWaker.Assert() + p.pauseWaker.Assert() continue + } else { + p.pauseChan <- struct{}{} + <-p.resumeChan } - - // If socket has transitioned out of connected state then just let the - // worker handle the packet. - // - // NOTE: We read this outside of e.mu lock which means that by the time - // we get to handleSegments the endpoint may not be in ESTABLISHED. But - // this should be fine as all normal shutdown states are handled by - // handleSegments and if the endpoint moves to a CLOSED/ERROR state - // then handleSegments is a noop. - if ep.EndpointState() == StateEstablished && ep.TryLock() { - // If the endpoint is in a connected state then we do direct delivery - // to ensure low latency and avoid scheduler interactions. - switch err := ep.handleSegmentsLocked(true /* fastPath */); { - case err != nil: - // Send any active resets if required. - ep.resetConnectionLocked(err) - fallthrough - case ep.EndpointState() == StateClose: - ep.notifyProtocolGoroutine(notifyTickleWorker) - case !ep.segmentQueue.empty(): + case w == &p.newEndpointWaker: + for { + ep := p.epQ.dequeue() + if ep == nil { + break + } + if ep.segmentQueue.empty() { + continue + } + switch state := ep.EndpointState(); { + case state.connecting(): + p.handleConnecting(ep) + case state.connected() && state != StateTimeWait: + p.handleConnected(ep) + case state == StateTimeWait: + p.handleTimeWait(ep) + case state == StateListen: + p.handleListen(ep) + case state == StateError || state == StateClose: + // Try to redeliver any still queued + // packets to another endpoint or send a + // RST if it can't be delivered. + ep.mu.Lock() + if st := ep.EndpointState(); st == StateError || st == StateClose { + ep.drainClosingSegmentQueue() + } + ep.mu.Unlock() + default: + panic(fmt.Sprintf("unexpected tcp state in processor: %v", state)) + } + // If there are more segments to process then + // requeue this endpoint for processing. + if !ep.segmentQueue.empty() { p.epQ.enqueue(ep) } - ep.mu.Unlock() - } else { - ep.newSegmentWaker.Assert() } } } } +// pause pauses the processor loop. +func (p *processor) pause() chan struct{} { + p.pauseWaker.Assert() + return p.pauseChan +} + +// resume resumes a previously paused loop. +// +// Precondition: Pause must have been called previously. +func (p *processor) resume() { + p.resumeChan <- struct{}{} +} + // dispatcher manages a pool of TCP endpoint processors which are responsible // for the processing of inbound segments. This fixed pool of processor // goroutines do full tcp processing. The processor is selected based on the @@ -143,20 +352,33 @@ func (p *processor) start(wg *sync.WaitGroup) { // in-order. type dispatcher struct { processors []processor - // seed is a random secret for a jenkins hash. - seed uint32 - wg sync.WaitGroup + wg sync.WaitGroup + hasher jenkinsHasher + mu sync.Mutex + // +checklocks:mu + paused bool + // +checklocks:mu + closed bool } +// init initializes a dispatcher and starts the main loop for all the processors +// owned by this dispatcher. func (d *dispatcher) init(rng *rand.Rand, nProcessors int) { d.close() d.wait() + + d.mu.Lock() + defer d.mu.Unlock() + d.closed = false d.processors = make([]processor, nProcessors) - d.seed = rng.Uint32() + d.hasher = jenkinsHasher{seed: rng.Uint32()} for i := range d.processors { p := &d.processors[i] p.sleeper.AddWaker(&p.newEndpointWaker) p.sleeper.AddWaker(&p.closeWaker) + p.sleeper.AddWaker(&p.pauseWaker) + p.pauseChan = make(chan struct{}) + p.resumeChan = make(chan struct{}) d.wg.Add(1) // NB: sleeper-waker registration must happen synchronously to avoid races // with `close`. It's possible to pull all this logic into `start`, but @@ -165,17 +387,32 @@ func (d *dispatcher) init(rng *rand.Rand, nProcessors int) { } } +// close closes a dispatcher and its processors. func (d *dispatcher) close() { + d.mu.Lock() + d.closed = true + d.mu.Unlock() for i := range d.processors { d.processors[i].close() } } +// wait waits for all processor goroutines to end. func (d *dispatcher) wait() { d.wg.Wait() } +// queuePacket queues an incoming packet to the matching tcp endpoint and +// also queues the endpoint to a processor queue for processing. func (d *dispatcher) queuePacket(stackEP stack.TransportEndpoint, id stack.TransportEndpointID, clock tcpip.Clock, pkt *stack.PacketBuffer) { + d.mu.Lock() + closed := d.closed + d.mu.Unlock() + + if closed { + return + } + ep := stackEP.(*endpoint) s := newIncomingSegment(id, clock, pkt) @@ -202,25 +439,62 @@ func (d *dispatcher) queuePacket(stackEP stack.TransportEndpoint, id stack.Trans return } - // For sockets not in established state let the worker goroutine - // handle the packets. - if ep.EndpointState() != StateEstablished { - ep.newSegmentWaker.Assert() - return - } - d.selectProcessor(id).queueEndpoint(ep) } +// selectProcessor uses a hash of the transport endpoint ID to queue the +// endpoint to a specific processor. This is required to main TCP ordering as +// queueing the same endpoint to multiple processors can *potentially* result in +// out of order processing of incoming segments. It also ensures that a dispatcher +// evenly loads the processor goroutines. func (d *dispatcher) selectProcessor(id stack.TransportEndpointID) *processor { + return &d.processors[d.hasher.hash(id)%uint32(len(d.processors))] +} + +// pause pauses a dispatcher and all its processor goroutines. +func (d *dispatcher) pause() { + d.mu.Lock() + d.paused = true + d.mu.Unlock() + for i := range d.processors { + <-d.processors[i].pause() + } +} + +// resume resumes a previously paused dispatcher and its processor goroutines. +// Calling resume on a dispatcher that was never paused is a no-op. +func (d *dispatcher) resume() { + d.mu.Lock() + + if !d.paused { + // If this was a restore run the stack is a new instance and + // it was never paused, so just return as there is nothing to + // resume. + d.mu.Unlock() + return + } + d.paused = false + d.mu.Unlock() + for i := range d.processors { + d.processors[i].resume() + } +} + +// jenkinsHasher contains state needed to for a jenkins hash. +type jenkinsHasher struct { + seed uint32 +} + +// hash hashes the provided TransportEndpointID using the jenkins hash +// algorithm. +func (j jenkinsHasher) hash(id stack.TransportEndpointID) uint32 { var payload [4]byte binary.LittleEndian.PutUint16(payload[0:], id.LocalPort) binary.LittleEndian.PutUint16(payload[2:], id.RemotePort) - h := jenkins.Sum32(d.seed) + h := jenkins.Sum32(j.seed) h.Write(payload[:]) h.Write([]byte(id.LocalAddress)) h.Write([]byte(id.RemoteAddress)) - - return &d.processors[h.Sum32()%uint32(len(d.processors))] + return h.Sum32() } diff --git a/pkg/tcpip/transport/tcp/endpoint.go b/pkg/tcpip/transport/tcp/endpoint.go index 55b97df29..79ec2b651 100644 --- a/pkg/tcpip/transport/tcp/endpoint.go +++ b/pkg/tcpip/transport/tcp/endpoint.go @@ -169,28 +169,6 @@ func (s EndpointState) String() string { } } -// Reasons for notifying the protocol goroutine. -const ( - notifyNonZeroReceiveWindow = 1 << iota - notifyClose - notifyMTUChanged - notifyDrain - notifyReset - notifyResetByPeer - // notifyAbort is a request for an expedited teardown. - notifyAbort - notifyKeepaliveChanged - notifyMSSChanged - // notifyTickleWorker is used to tickle the protocol main loop during a - // restore after we update the endpoint state to the correct one. This - // ensures the loop terminates if the final state of the endpoint is - // say TIME_WAIT. - notifyTickleWorker - notifyError - // notifyShutdown means that a connecting socket was shutdown. - notifyShutdown -) - // SACKInfo holds TCP SACK related information for a given endpoint. // // +stateify savable @@ -367,10 +345,12 @@ type endpoint struct { // Precondition: epQueue.mu must be held to read/write this field.. endpointEntry `state:"nosave"` + // pendingProcessingMu protects pendingProcessing. + pendingProcessingMu sync.Mutex `state:"nosave"` + // pendingProcessing is true if this endpoint is queued for processing // to a TCP processor. - // - // Precondition: epQueue.mu must be held to read/write this field.. + // +checklocks:pendingProcessingMu pendingProcessing bool `state:"nosave"` // The following fields are initialized at creation time and do not @@ -444,7 +424,8 @@ type endpoint struct { // h stores a reference to the current handshake state if the endpoint is in // the SYN-SENT or SYN-RECV states, in which case endpoint == endpoint.h.ep. // nil otherwise. - h *handshake `state:"nosave"` + // +checklocks:mu + h *handshake // portFlags stores the current values of port related flags. portFlags ports.Flags @@ -463,14 +444,6 @@ type endpoint struct { // address). effectiveNetProtos []tcpip.NetworkProtocolNumber - // workerRunning specifies if a worker goroutine is running. - workerRunning bool - - // workerCleanup specifies if the worker goroutine must perform cleanup - // before exiting. This can only be set to true when workerRunning is - // also true, and they're both protected by the mutex. - workerCleanup bool - // recentTSTime is the unix time when we last updated // TCPEndpointStateInner.RecentTS. recentTSTime tcpip.MonotonicTime @@ -519,18 +492,6 @@ type endpoint struct { // this endpoint. cc tcpip.CongestionControlOption - // newSegmentWaker is used to indicate to the protocol goroutine that - // it needs to wake up and handle new segments queued to it. - newSegmentWaker sleep.Waker `state:"manual"` - - // notificationWaker is used to indicate to the protocol goroutine that - // it needs to wake up and check for notifications. - notificationWaker sleep.Waker `state:"manual"` - - // notifyFlags is a bitmask of flags used to indicate to the protocol - // goroutine what it was notified; this is only accessed atomically. - notifyFlags uint32 `state:"nosave"` - // keepalive manages TCP keepalive state. When the connection is idle // (no data sent or received) for keepaliveIdle, we start sending // keepalives every keepalive.interval. If we send keepalive.count @@ -550,25 +511,13 @@ type endpoint struct { // listener. deferAccept time.Duration - // pendingAccepted tracks connections queued to be accepted. It is used to - // ensure such queued connections are terminated before the accepted queue is - // marked closed (by setting its capacity to zero). - pendingAccepted sync.WaitGroup `state:"nosave"` - - // acceptMu protects accepted. + // acceptMu protects accepQueue acceptMu sync.Mutex `state:"nosave"` - // acceptCond is a condition variable that can be used to block on when - // accepted is full and an endpoint is ready to be delivered. + // acceptQueue is used by a listening endpoint to send newly accepted + // connections to the endpoint so that they can be read by Accept() + // calls. // - // We use this condition variable to block/unblock goroutines which - // tried to deliver an endpoint but couldn't because accept backlog was - // full ( See: endpoint.deliverAccepted ). - acceptCond *sync.Cond `state:"nosave"` - - // accepted is used by a listening endpoint protocol goroutine to - // send newly accepted connections to the endpoint so that they can be - // read by Accept() calls. // +checklocks:acceptMu acceptQueue acceptQueue @@ -628,6 +577,20 @@ type endpoint struct { // lastOutOfWindowAckTime is the time at which the an ACK was sent in response // to an out of window segment being received by this endpoint. lastOutOfWindowAckTime tcpip.MonotonicTime + + // finWait2Timer is used to reap orphaned sockets in FIN-WAIT-2 where the peer + // is yet to send a FIN but on our end the socket is fully closed i.e. endpoint.Close() + // has been called on the socket. This timer is not started for sockets that + // are waiting for a peer FIN but are not closed. + finWait2Timer tcpip.Timer `state:"nosave"` + + // timeWaitTimer is used to reap a socket once a socket has been in TIME-WAIT state + // for tcp.DefaultTCPTimeWaitTimeout seconds. + timeWaitTimer tcpip.Timer `state:"nosave"` + + // listenCtx is used by listening endpoints to store state used while listening for + // connections. Nil otherwise. + listenCtx *listenContext `state:"nosave"` } // UniqueID implements stack.TransportEndpoint.UniqueID. @@ -685,16 +648,9 @@ func (e *endpoint) LockUser() { } // UnlockUser will check if there are any segments already queued for processing -// and process any such segments before unlocking e.mu. This is required because -// we when packets arrive and endpoint lock is already held then such packets -// are queued up to be processed. If the lock is held by the endpoint goroutine -// then it will process these packets but if the lock is instead held by the -// syscall goroutine then we can have the syscall goroutine process the backlog -// before unlocking. -// -// This avoids an unnecessary wakeup of the endpoint protocol goroutine for the -// endpoint. It's also required eventually when we get rid of the endpoint -// protocol goroutine altogether. +// and wake up a processor goroutine to process them before unlocking e.mu. +// This is required because we when packets arrive and endpoint lock is already +// held then such packets are queued up to be processed. // // Precondition: e.LockUser() must have been called before calling e.UnlockUser() // +checklocksrelease:e.mu @@ -702,34 +658,31 @@ func (e *endpoint) UnlockUser() { // Lock segment queue before checking so that we avoid a race where // segments can be queued between the time we check if queue is empty // and actually unlock the endpoint mutex. - for { - e.segmentQueue.mu.Lock() - if e.segmentQueue.emptyLocked() { - if atomic.SwapUint32(&e.ownedByUser, 0) != 1 { - panic("e.UnlockUser() called without calling e.LockUser()") - } - e.mu.Unlock() - e.segmentQueue.mu.Unlock() - return + e.segmentQueue.mu.Lock() + if e.segmentQueue.emptyLocked() { + if atomic.SwapUint32(&e.ownedByUser, 0) != 1 { + panic("e.UnlockUser() called without calling e.LockUser()") } + e.mu.Unlock() e.segmentQueue.mu.Unlock() - - switch e.EndpointState() { - case StateEstablished: - if err := e.handleSegmentsLocked(true /* fastPath */); err != nil { - e.notifyProtocolGoroutine(notifyTickleWorker) - } - default: - // Since we are waking the endpoint goroutine here just unlock - // and let it process the queued segments. - e.newSegmentWaker.Assert() - if atomic.SwapUint32(&e.ownedByUser, 0) != 1 { - panic("e.UnlockUser() called without calling e.LockUser()") - } - e.mu.Unlock() - return - } + return } + e.segmentQueue.mu.Unlock() + + // Since we are waking the processor goroutine here just unlock + // and let it process the queued segments. + if atomic.SwapUint32(&e.ownedByUser, 0) != 1 { + panic("e.UnlockUser() called without calling e.LockUser()") + } + processor := e.protocol.dispatcher.selectProcessor(e.ID) + e.mu.Unlock() + + // Wake up the processor for this endpoint to process any queued + // segments after releasing the lock to avoid the case where if the + // processor goroutine starts running before we release the lock here + // then it will fail to process as TryLock() will fail. + processor.queueEndpoint(e) + return } // StopWork halts packet processing. Only to be used in tests. @@ -918,8 +871,7 @@ func newEndpoint(s *stack.Stack, protocol *protocol, netProto tcpip.NetworkProto e.segmentQueue.ep = e - e.acceptCond = sync.NewCond(&e.acceptMu) - e.keepalive.timer.init(e.stack.Clock(), &e.keepalive.waker) + e.keepalive.timer.init(e.stack.Clock(), maybeFailTimerHandler(e, e.keepaliveTimerExpired)) return e } @@ -976,51 +928,6 @@ func (e *endpoint) Readiness(mask waiter.EventMask) waiter.EventMask { return result } -func (e *endpoint) fetchNotifications() uint32 { - return atomic.SwapUint32(&e.notifyFlags, 0) -} - -func (e *endpoint) notifyProtocolGoroutine(n uint32) { - for { - v := atomic.LoadUint32(&e.notifyFlags) - if v&n == n { - // The flags are already set. - return - } - - if atomic.CompareAndSwapUint32(&e.notifyFlags, v, v|n) { - if v == 0 { - // We are causing a transition from no flags to - // at least one flag set, so we must cause the - // protocol goroutine to wake up. - e.notificationWaker.Assert() - } - return - } - } -} - -func (e *endpoint) Release() { - e.LockUser() - defer e.UnlockUser() - e.transitionToStateCloseLocked() - e.notifyProtocolGoroutine(notifyTickleWorker) - e.releaseLocked() -} - -// +checklocks:e.mu -func (e *endpoint) releaseLocked() { - e.purgeReadQueue() - e.purgeWriteQueue() - for { - s := e.segmentQueue.dequeue() - if s == nil { - break - } - s.DecRef() - } -} - // Purging pending rcv segments is only necessary on RST. func (e *endpoint) purgePendingRcvQueue() { if e.rcv != nil { @@ -1069,34 +976,18 @@ func (e *endpoint) purgeWriteQueue() { // Abort implements stack.TransportEndpoint.Abort. func (e *endpoint) Abort() { - // The abort notification is not processed synchronously, so no - // synchronization is needed. - // - // If the endpoint becomes connected after this check, we still close - // the endpoint. This worst case results in a slower abort. - // - // If the endpoint disconnected after the check, nothing needs to be - // done, so sending a notification which will potentially be ignored is - // fine. - // - // If the endpoint connecting finishes after the check, the endpoint - // is either in a connected state (where we would notifyAbort anyway), - // SYN-RECV (where we would also notifyAbort anyway), or in an error - // state where nothing is required and the notification can be safely - // ignored. - // - // Endpoints where a Close during connecting or SYN-RECV state would be - // problematic are set to state connecting before being registered (and - // thus possible to be Aborted). They are never available in initial - // state. - // - // Endpoints transitioning from initial to connecting state may be - // safely either closed or sent notifyAbort. - if s := e.EndpointState(); s == StateConnecting || s == StateSynRecv || s.connected() { - e.notifyProtocolGoroutine(notifyAbort) + defer e.drainClosingSegmentQueue() + e.LockUser() + defer e.UnlockUser() + defer e.purgeReadQueue() + // Reset all connected endpoints. + switch state := e.EndpointState(); { + case state.connected(): + e.stack.Stats().TCP.CurrentConnected.Decrement() + e.resetConnectionLocked(&tcpip.ErrAborted{}) return } - e.Close() + e.closeLocked() } // Close puts the endpoint in a closed state and frees all resources associated @@ -1104,14 +995,29 @@ func (e *endpoint) Abort() { // the endpoint. func (e *endpoint) Close() { e.LockUser() - defer e.UnlockUser() if e.closed { + e.UnlockUser() return } // We always want to purge the read queue, but do so after the checks in // shutdownLocked. - defer e.purgeReadQueue() + e.closeLocked() + e.purgeReadQueue() + if e.EndpointState() == StateClose || e.EndpointState() == StateError { + // It should be safe to purge the read queue now as the endpoint + // is now closed or in an error state and further reads are not + // permitted. + e.UnlockUser() + e.drainClosingSegmentQueue() + e.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) + return + } + e.UnlockUser() +} + +// +checklocks:e.mu +func (e *endpoint) closeLocked() { linger := e.SocketOptions().GetLinger() if linger.Enabled && linger.Timeout == 0 { s := e.EndpointState() @@ -1120,15 +1026,6 @@ func (e *endpoint) Close() { // Close the endpoint without doing full shutdown and // send a RST. e.resetConnectionLocked(&tcpip.ErrConnectionAborted{}) - e.closeNoShutdownLocked() - - // Wake up worker to close the endpoint. - switch s { - case StateSynRecv: - e.notifyProtocolGoroutine(notifyClose) - default: - e.notifyProtocolGoroutine(notifyTickleWorker) - } return } } @@ -1137,10 +1034,6 @@ func (e *endpoint) Close() { // if we're connected, or stop accepting if we're listening. e.shutdownLocked(tcpip.ShutdownWrite | tcpip.ShutdownRead) e.closeNoShutdownLocked() - switch e.EndpointState() { - case StateClose, StateError: - e.releaseLocked() - } } // closeNoShutdown closes the endpoint without doing a full shutdown. @@ -1174,29 +1067,32 @@ func (e *endpoint) closeNoShutdownLocked() { // Mark endpoint as closed. e.closed = true - - switch e.EndpointState() { - case StateClose, StateError: - return - } + tcpip.AddDanglingEndpoint(e) eventMask := waiter.ReadableEvents | waiter.WritableEvents - // Either perform the local cleanup or kick the worker to make sure it - // knows it needs to cleanup. - if e.workerRunning { - e.workerCleanup = true - tcpip.AddDanglingEndpoint(e) - // Worker will remove the dangling endpoint when the endpoint - // goroutine terminates. - e.notifyProtocolGoroutine(notifyClose) - } else { - e.transitionToStateCloseLocked() + + switch e.EndpointState() { + case StateInitial, StateBound, StateListen: + e.setEndpointState(StateClose) + fallthrough + case StateClose, StateError: + eventMask |= waiter.EventHUp + e.cleanupLocked() + case StateConnecting, StateSynSent, StateSynRecv: + // Abort the handshake and set the error. // Notify that the endpoint is closed. eventMask |= waiter.EventHUp + e.handshakeFailed(&tcpip.ErrAborted{}) + // Notify that the endpoint is closed. + eventMask |= waiter.EventHUp + case StateFinWait2: + // The socket has been closed and we are in FIN-WAIT-2 so start + // the FIN-WAIT-2 timer. + if e.finWait2Timer == nil { + e.finWait2Timer = e.stack.Clock().AfterFunc(e.tcpLingerTimeout, e.finWait2TimerExpired) + } } - // The TCP closing state-machine would eventually notify EventHUp, but we - // notify EventIn|EventOut immediately to unblock any blocked waiters. e.waiterQueue.Notify(eventMask) } @@ -1204,37 +1100,51 @@ func (e *endpoint) closeNoShutdownLocked() { // handshake but not yet been delivered to the application. func (e *endpoint) closePendingAcceptableConnectionsLocked() { e.acceptMu.Lock() - // Close any endpoints in SYN-RCVD state. - for n := range e.acceptQueue.pendingEndpoints { - n.notifyProtocolGoroutine(notifyClose) - } + + pendingEndpoints := e.acceptQueue.pendingEndpoints e.acceptQueue.pendingEndpoints = nil - // Reset all connections that are waiting to be accepted. + + completedEndpoints := make([]*endpoint, 0, e.acceptQueue.endpoints.Len()) for n := e.acceptQueue.endpoints.Front(); n != nil; n = n.Next() { - n.Value.(*endpoint).notifyProtocolGoroutine(notifyReset) + completedEndpoints = append(completedEndpoints, n.Value.(*endpoint)) } e.acceptQueue.endpoints.Init() + e.acceptQueue.capacity = 0 e.acceptMu.Unlock() - e.acceptCond.Broadcast() + // Close any endpoints in SYN-RCVD state. + for n := range pendingEndpoints { + n.Abort() + } - // Wait for reset of all endpoints that are still waiting to be delivered to - // the now closed accepted. - e.pendingAccepted.Wait() + // Reset all connections that are waiting to be accepted. + for _, n := range completedEndpoints { + n.Abort() + } } -// cleanupLocked frees all resources associated with the endpoint. It is called -// after Close() is called and the worker goroutine (if any) is done with its -// work. +// cleanupLocked frees all resources associated with the endpoint. // +checklocks:e.mu func (e *endpoint) cleanupLocked() { + if e.snd != nil { + e.snd.resendTimer.cleanup() + e.snd.probeTimer.cleanup() + e.snd.reorderTimer.cleanup() + } + + if e.finWait2Timer != nil { + e.finWait2Timer.Stop() + } + + if e.timeWaitTimer != nil { + e.timeWaitTimer.Stop() + } + // Close all endpoints that might have been accepted by TCP but not by // the client. e.closePendingAcceptableConnectionsLocked() e.keepalive.timer.cleanup() - e.workerCleanup = false - if e.isRegistered { e.stack.StartTransportEndpointCleanup(e.effectiveNetProtos, ProtocolNumber, e.TransportEndpointInfo.ID, e, e.boundPortFlags, e.boundBindToDevice) e.isRegistered = false @@ -1262,8 +1172,12 @@ func (e *endpoint) cleanupLocked() { e.route = nil } - // It's not safe to purge the read queues yet, there could be unread data. e.purgeWriteQueue() + // Only purge the read queue here if the socket is fully closed by the + // user. + if e.closed { + e.purgeReadQueue() + } e.stack.CompleteTransportEndpointCleanup(e) tcpip.DeleteDanglingEndpoint(e) } @@ -1309,6 +1223,8 @@ func (e *endpoint) ModerateRecvBuf(copied int) { e.LockUser() defer e.UnlockUser() + sendNonZeroWindowUpdate := false + e.rcvQueueInfo.rcvQueueMu.Lock() if e.rcvQueueInfo.RcvAutoParams.Disabled { e.rcvQueueInfo.rcvQueueMu.Unlock() @@ -1359,7 +1275,7 @@ func (e *endpoint) ModerateRecvBuf(copied int) { e.ops.SetReceiveBufferSize(int64(rcvWnd), false /* notify */) availAfter := wndFromSpace(e.receiveBufferAvailableLocked(rcvWnd)) if crossed, above := e.windowCrossedACKThresholdLocked(availAfter-availBefore, rcvBufSize); crossed && above { - e.notifyProtocolGoroutine(notifyNonZeroReceiveWindow) + sendNonZeroWindowUpdate = true } } @@ -1372,6 +1288,12 @@ func (e *endpoint) ModerateRecvBuf(copied int) { e.rcvQueueInfo.RcvAutoParams.MeasureTime = now e.rcvQueueInfo.RcvAutoParams.CopiedBytes = 0 e.rcvQueueInfo.rcvQueueMu.Unlock() + + // Send the update after unlocking rcvQueueInfo as sending a segment acquires + // e.rcvQueueInfo.rcvQueueMu to calculate the window to be sent. + if e.EndpointState().connected() && sendNonZeroWindowUpdate { + e.rcv.nonZeroWindow() // +checklocksforce:e.rcv.ep.mu + } } // SetOwner implements tcpip.Endpoint.SetOwner. @@ -1537,9 +1459,9 @@ func (e *endpoint) startRead() (first, last *segment, err tcpip.Error) { func (e *endpoint) commitRead(done int) *segment { e.LockUser() defer e.UnlockUser() - e.rcvQueueInfo.rcvQueueMu.Lock() - defer e.rcvQueueInfo.rcvQueueMu.Unlock() + sendNonZeroWindowUpdate := false + e.rcvQueueInfo.rcvQueueMu.Lock() memDelta := 0 s := e.rcvQueueInfo.rcvQueue.Front() for s != nil && s.data.Size() == 0 { @@ -1564,11 +1486,16 @@ func (e *endpoint) commitRead(done int) *segment { // (whichever smaller), then notify the protocol goroutine to send a // window update. if crossed, above := e.windowCrossedACKThresholdLocked(memDelta, int(e.ops.GetReceiveBufferSize())); crossed && above { - e.notifyProtocolGoroutine(notifyNonZeroReceiveWindow) + sendNonZeroWindowUpdate = true } } + nextSeg := e.rcvQueueInfo.rcvQueue.Front() + e.rcvQueueInfo.rcvQueueMu.Unlock() - return e.rcvQueueInfo.rcvQueue.Front() + if e.EndpointState().connected() && sendNonZeroWindowUpdate { + e.rcv.nonZeroWindow() // +checklocksforce:e.rcv.ep.mu + } + return nextSeg } // isEndpointWritableLocked checks if a given endpoint is writable @@ -1774,6 +1701,7 @@ func (e *endpoint) windowCrossedACKThresholdLocked(deltaBefore int, rcvBufSize i if wndThreshold := wndFromSpace(rcvBufSize / rcvBufFraction); threshold > wndThreshold { threshold = wndThreshold } + switch { case oldAvail < threshold && newAvail >= threshold: return true, true @@ -1799,22 +1727,32 @@ func (e *endpoint) OnReusePortSet(v bool) { // OnKeepAliveSet implements tcpip.SocketOptionsHandler.OnKeepAliveSet. func (e *endpoint) OnKeepAliveSet(bool) { - e.notifyProtocolGoroutine(notifyKeepaliveChanged) + e.LockUser() + e.resetKeepaliveTimer(true /* receivedData */) + e.UnlockUser() } // OnDelayOptionSet implements tcpip.SocketOptionsHandler.OnDelayOptionSet. func (e *endpoint) OnDelayOptionSet(v bool) { if !v { + e.LockUser() + defer e.UnlockUser() // Handle delayed data. - e.sndQueueInfo.sndWaker.Assert() + if e.EndpointState().connected() { + e.sendData(nil /* next */) + } } } // OnCorkOptionSet implements tcpip.SocketOptionsHandler.OnCorkOptionSet. func (e *endpoint) OnCorkOptionSet(v bool) { if !v { + e.LockUser() + defer e.UnlockUser() // Handle the corked data. - e.sndQueueInfo.sndWaker.Assert() + if e.EndpointState().connected() { + e.sendData(nil /* next */) + } } } @@ -1823,8 +1761,10 @@ func (e *endpoint) getSendBufferSize() int { } // OnSetReceiveBufferSize implements tcpip.SocketOptionsHandler.OnSetReceiveBufferSize. -func (e *endpoint) OnSetReceiveBufferSize(rcvBufSz, oldSz int64) (newSz int64) { +func (e *endpoint) OnSetReceiveBufferSize(rcvBufSz, oldSz int64) (newSz int64, postSet func()) { e.LockUser() + + sendNonZeroWindowUpdate := false e.rcvQueueInfo.rcvQueueMu.Lock() // Make sure the receive buffer size allows us to send a @@ -1845,12 +1785,21 @@ func (e *endpoint) OnSetReceiveBufferSize(rcvBufSz, oldSz int64) (newSz int64) { // syndrome prevetion, when our available space grows above aMSS // or half receive buffer, whichever smaller. if crossed, above := e.windowCrossedACKThresholdLocked(availAfter-availBefore, int(rcvBufSz)); crossed && above { - e.notifyProtocolGoroutine(notifyNonZeroReceiveWindow) + sendNonZeroWindowUpdate = true } e.rcvQueueInfo.rcvQueueMu.Unlock() + + postSet = func() { + e.LockUser() + defer e.UnlockUser() + if e.EndpointState().connected() && sendNonZeroWindowUpdate { + e.rcv.nonZeroWindow() // +checklocksforce:e.rcv.ep.mu + } + + } e.UnlockUser() - return rcvBufSz + return rcvBufSz, postSet } // OnSetSendBufferSize implements tcpip.SocketOptionsHandler.OnSetSendBufferSize. @@ -1881,10 +1830,12 @@ func (e *endpoint) SetSockOptInt(opt tcpip.SockOptInt, v int) tcpip.Error { switch opt { case tcpip.KeepaliveCountOption: + e.LockUser() e.keepalive.Lock() e.keepalive.count = v e.keepalive.Unlock() - e.notifyProtocolGoroutine(notifyKeepaliveChanged) + e.resetKeepaliveTimer(true /* receivedData */) + e.UnlockUser() case tcpip.IPv4TOSOption: e.LockUser() @@ -1908,7 +1859,6 @@ func (e *endpoint) SetSockOptInt(opt tcpip.SockOptInt, v int) tcpip.Error { e.LockUser() e.userMSS = uint16(userMSS) e.UnlockUser() - e.notifyProtocolGoroutine(notifyMSSChanged) case tcpip.MTUDiscoverOption: // Return not supported if attempting to set this option to @@ -1969,16 +1919,20 @@ func (e *endpoint) HasNIC(id int32) bool { func (e *endpoint) SetSockOpt(opt tcpip.SettableSocketOption) tcpip.Error { switch v := opt.(type) { case *tcpip.KeepaliveIdleOption: + e.LockUser() e.keepalive.Lock() e.keepalive.idle = time.Duration(*v) e.keepalive.Unlock() - e.notifyProtocolGoroutine(notifyKeepaliveChanged) + e.resetKeepaliveTimer(true /* receivedData */) + e.UnlockUser() case *tcpip.KeepaliveIntervalOption: + e.LockUser() e.keepalive.Lock() e.keepalive.interval = time.Duration(*v) e.keepalive.Unlock() - e.notifyProtocolGoroutine(notifyKeepaliveChanged) + e.resetKeepaliveTimer(true /* receivedData */) + e.UnlockUser() case *tcpip.TCPUserTimeoutOption: e.LockUser() @@ -2240,7 +2194,9 @@ func (*endpoint) Disconnect() tcpip.Error { // Connect connects the endpoint to its peer. func (e *endpoint) Connect(addr tcpip.FullAddress) tcpip.Error { - err := e.connect(addr, true, true) + e.LockUser() + defer e.UnlockUser() + err := e.connect(addr, true) if err != nil { if !err.IgnoreStats() { // Connect failed. Let's wake up any waiters. @@ -2252,16 +2208,9 @@ func (e *endpoint) Connect(addr tcpip.FullAddress) tcpip.Error { return err } -// connect connects the endpoint to its peer. In the normal non-S/R case, the -// new connection is expected to run the main goroutine and perform handshake. -// In restore of previously connected endpoints, both ends will be passively -// created (so no new handshaking is done); for stack-accepted connections not -// yet accepted by the app, they are restored without running the main goroutine -// here. -func (e *endpoint) connect(addr tcpip.FullAddress, handshake bool, run bool) tcpip.Error { - e.LockUser() - defer e.UnlockUser() - +// connect connects the endpoint to its peer. +// +checklocks:e.mu +func (e *endpoint) connect(addr tcpip.FullAddress, handshake bool) tcpip.Error { connectingAddr := addr.Addr addr, netProto, err := e.checkV4MappedLocked(addr) @@ -2423,9 +2372,9 @@ func (e *endpoint) connect(addr tcpip.FullAddress, handshake bool, run bool) tcp // Since the endpoint is in TIME-WAIT it should be safe to acquire its // Lock while holding the lock for this endpoint as endpoints in // TIME-WAIT do not acquire locks on other endpoints. - tcpEP.workerCleanup = false - tcpEP.cleanupLocked() - tcpEP.notifyProtocolGoroutine(notifyAbort) + tcpEP.transitionToStateCloseLocked() + tcpEP.drainClosingSegmentQueue() + tcpEP.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) tcpEP.UnlockUser() // Now try and Reserve again if it fails then we skip. portRes := ports.Reservation{ @@ -2502,18 +2451,14 @@ func (e *endpoint) connect(addr tcpip.FullAddress, handshake bool, run bool) tcp // Set the new auto tuned send buffer size after entering // established state. e.ops.SetSendBufferSize(e.computeTCPSendBufferSize(), false /* notify */) + return &tcpip.ErrConnectStarted{} } - if run { - if handshake { - h := e.newHandshake() - e.setEndpointState(StateSynSent) - h.start() - } - e.stack.Stats().TCP.ActiveConnectionOpenings.Increment() - e.workerRunning = true - go e.protocolMainLoop(handshake, nil) // S/R-SAFE: will be drained before save. - } + // Start a new handshake. + h := e.newHandshake() + e.setEndpointState(StateSynSent) + h.start() + e.stack.Stats().TCP.ActiveConnectionOpenings.Increment() return &tcpip.ErrConnectStarted{} } @@ -2534,8 +2479,8 @@ func (e *endpoint) Shutdown(flags tcpip.ShutdownFlags) tcpip.Error { // enter the error state. But this logic cannot belong to the shutdownLocked // method because that method is called during a close(2) (and closing a // connecting socket is not an error). - e.resetConnectionLocked(&tcpip.ErrConnectionReset{}) - e.notifyProtocolGoroutine(notifyShutdown) + e.handshakeFailed(&tcpip.ErrConnectionReset{}) + e.cleanupLocked() e.waiterQueue.Notify(waiter.WritableEvents | waiter.EventHUp | waiter.EventErr) return nil } @@ -2555,13 +2500,10 @@ func (e *endpoint) shutdownLocked(flags tcpip.ShutdownFlags) tcpip.Error { e.rcvQueueInfo.RcvClosed = true rcvBufUsed := e.rcvQueueInfo.RcvBufUsed e.rcvQueueInfo.rcvQueueMu.Unlock() - // If we're fully closed and we have unread data we need to abort // the connection with a RST. if e.shutdownFlags&tcpip.ShutdownWrite != 0 && rcvBufUsed > 0 { e.resetConnectionLocked(&tcpip.ErrConnectionAborted{}) - // Wake up worker to terminate loop. - e.notifyProtocolGoroutine(notifyTickleWorker) return nil } // Wake up any readers that maybe waiting for the stream to become @@ -2658,10 +2600,6 @@ func (e *endpoint) listen(backlog int) tcpip.Error { e.rcvQueueInfo.RcvClosed = false e.rcvQueueInfo.rcvQueueMu.Unlock() - // Notify any blocked goroutines that they can attempt to - // deliver endpoints again. - e.acceptCond.Broadcast() - return nil } @@ -2700,21 +2638,11 @@ func (e *endpoint) listen(backlog int) tcpip.Error { } e.acceptMu.Unlock() - e.workerRunning = true - go e.protocolListenLoop( // S/R-SAFE: drained on save. - seqnum.Size(e.receiveBufferAvailable())) - return nil -} + // Initialize the listening context. + rcvWnd := seqnum.Size(e.receiveBufferAvailable()) + e.listenCtx = newListenContext(e.stack, e.protocol, e, rcvWnd, e.ops.GetV6Only(), e.NetProto) -// startAcceptedLoop sets up required state and starts a goroutine with the -// main loop for accepted connections. -// +checklocksrelease:e.mu -func (e *endpoint) startAcceptedLoop() { - e.workerRunning = true - e.mu.Unlock() - wakerInitDone := make(chan struct{}) - go e.protocolMainLoop(false, wakerInitDone) // S/R-SAFE: drained on save. - <-wakerInitDone + return nil } // Accept returns a new endpoint if a peer has established a connection @@ -2743,7 +2671,6 @@ func (e *endpoint) Accept(peerAddr *tcpip.FullAddress) (tcpip.Endpoint, *waiter. if n == nil { return nil, nil, &tcpip.ErrWouldBlock{} } - e.acceptCond.Signal() if peerAddr != nil { *peerAddr = n.getRemoteAddress() } @@ -2920,20 +2847,43 @@ func (e *endpoint) onICMPError(err tcpip.Error, transErr stack.TransportError, p }) } - // Notify of the error. - e.notifyProtocolGoroutine(notifyError) + if e.EndpointState().connecting() { + e.mu.Lock() + if lEP := e.h.listenEP; lEP != nil { + // Remove from listening endpoints pending list. + lEP.acceptMu.Lock() + delete(lEP.acceptQueue.pendingEndpoints, e) + lEP.acceptMu.Unlock() + lEP.stats.FailedConnectionAttempts.Increment() + } + e.stack.Stats().TCP.FailedConnectionAttempts.Increment() + e.cleanupLocked() + e.hardError = err + e.setEndpointState(StateError) + e.mu.Unlock() + e.drainClosingSegmentQueue() + e.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) + } } // HandleError implements stack.TransportEndpoint. func (e *endpoint) HandleError(transErr stack.TransportError, pkt *stack.PacketBuffer) { handlePacketTooBig := func(mtu uint32) { e.sndQueueInfo.sndQueueMu.Lock() - e.sndQueueInfo.PacketTooBigCount++ + update := false if v := int(mtu); v < e.sndQueueInfo.SndMTU { e.sndQueueInfo.SndMTU = v + update = true } + newMTU := e.sndQueueInfo.SndMTU e.sndQueueInfo.sndQueueMu.Unlock() - e.notifyProtocolGoroutine(notifyMTUChanged) + if update { + e.mu.Lock() + defer e.mu.Unlock() + if e.snd != nil { + e.snd.updateMaxPayloadSize(newMTU, 1 /* count */) // +checklocksforce:e.snd.ep.mu + } + } } // TODO(gvisor.dev/issues/5270): Handle all transport errors. @@ -3213,15 +3163,11 @@ func (e *endpoint) Wait() { waitEntry, notifyCh := waiter.NewChannelEntry(waiter.EventHUp) e.waiterQueue.EventRegister(&waitEntry) defer e.waiterQueue.EventUnregister(&waitEntry) - for { - e.LockUser() - running := e.workerRunning - e.UnlockUser() - if !running { - break - } - <-notifyCh + switch e.EndpointState() { + case StateClose, StateError: + return } + <-notifyCh } // SocketOptions implements tcpip.Endpoint.SocketOptions. diff --git a/pkg/tcpip/transport/tcp/endpoint_state.go b/pkg/tcpip/transport/tcp/endpoint_state.go index 69e074569..527624760 100644 --- a/pkg/tcpip/transport/tcp/endpoint_state.go +++ b/pkg/tcpip/transport/tcp/endpoint_state.go @@ -17,7 +17,6 @@ package tcp import ( "fmt" "sync/atomic" - "time" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" @@ -26,23 +25,6 @@ import ( "gvisor.dev/gvisor/pkg/tcpip/stack" ) -// +checklocks:e.mu -func (e *endpoint) drainSegmentLocked() { - // Drain only up to once. - if e.drainDone != nil { - return - } - - e.drainDone = make(chan struct{}) - e.undrain = make(chan struct{}) - e.mu.Unlock() - - e.notifyProtocolGoroutine(notifyDrain) - <-e.drainDone - - e.mu.Lock() -} - // beforeSave is invoked by stateify. func (e *endpoint) beforeSave() { // Stop incoming packets. @@ -66,30 +48,11 @@ func (e *endpoint) beforeSave() { e.Close() e.mu.Lock() } - if !e.workerRunning { - // The endpoint must be in the accepted queue or has been just - // disconnected and closed. - break - } fallthrough - case epState == StateListen || epState == StateConnecting: - e.drainSegmentLocked() - // Refresh epState, since drainSegmentLocked may have changed it. - epState = e.EndpointState() - if !epState.closed() { - if !e.workerRunning { - panic("endpoint has no worker running in listen, connecting, or connected state") - } - } + case epState == StateListen: + // Nothing to do. case epState.closed(): - for e.workerRunning { - e.mu.Unlock() - time.Sleep(100 * time.Millisecond) - e.mu.Lock() - } - if e.workerRunning { - panic(fmt.Sprintf("endpoint: %+v still has worker running in closed or error state", e.TransportEndpointInfo.ID)) - } + // Nothing to do. default: panic(fmt.Sprintf("endpoint in unknown state %v", e.EndpointState())) } @@ -155,19 +118,16 @@ func (e *endpoint) afterLoad() { // Restore the endpoint to InitialState as it will be moved to // its origEndpointState during Resume. e.state = uint32(StateInitial) - // Condition variables and mutexs are not S/R'ed so reinitialize - // acceptCond with e.acceptMu. - e.acceptCond = sync.NewCond(&e.acceptMu) stack.StackFromEnv.RegisterRestoredEndpoint(e) } // Resume implements tcpip.ResumableEndpoint.Resume. func (e *endpoint) Resume(s *stack.Stack) { - e.keepalive.timer.init(s.Clock(), &e.keepalive.waker) if snd := e.snd; snd != nil { - snd.resendTimer.init(s.Clock(), &snd.resendWaker) - snd.reorderTimer.init(s.Clock(), &snd.reorderWaker) - snd.probeTimer.init(s.Clock(), &snd.probeWaker) + e.keepalive.timer.init(s.Clock(), maybeFailTimerHandler(e, e.keepaliveTimerExpired)) + snd.resendTimer.init(s.Clock(), maybeFailTimerHandler(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)) } e.stack = s e.protocol = protocolFromStack(s) @@ -233,20 +193,22 @@ func (e *endpoint) Resume(s *stack.Stack) { // Reset the scoreboard to reinitialize the sack information as // we do not restore SACK information. e.scoreboard.Reset() - err := e.connect(tcpip.FullAddress{NIC: e.boundNICID, Addr: e.connectingAddress, Port: e.TransportEndpointInfo.ID.RemotePort}, false, e.workerRunning) + e.mu.Lock() + err := e.connect(tcpip.FullAddress{NIC: e.boundNICID, Addr: e.connectingAddress, Port: e.TransportEndpointInfo.ID.RemotePort}, false /* handshake */) if _, ok := err.(*tcpip.ErrConnectStarted); !ok { panic("endpoint connecting failed: " + err.String()) } - e.mu.Lock() e.state = e.origEndpointState - closed := e.closed - e.mu.Unlock() - e.notifyProtocolGoroutine(notifyTickleWorker) - if epState == StateFinWait2 && closed { - // If the endpoint has been closed then make sure we notify so - // that the FIN_WAIT2 timer is started after a restore. - e.notifyProtocolGoroutine(notifyClose) + // For FIN-WAIT-2 and TIME-WAIT we need to start the appropriate timers so + // that the socket is closed correctly. + switch epState { + case StateFinWait2: + e.finWait2Timer = e.stack.Clock().AfterFunc(e.tcpLingerTimeout, e.finWait2TimerExpired) + case StateTimeWait: + e.timeWaitTimer = e.stack.Clock().AfterFunc(e.getTimeWaitDuration(), e.timeWaitTimerExpired) } + + e.mu.Unlock() connectedLoading.Done() case epState == StateListen: tcpip.AsyncLoading.Add(1) @@ -267,7 +229,8 @@ func (e *endpoint) Resume(s *stack.Stack) { listenLoading.Done() tcpip.AsyncLoading.Done() }() - case epState.connecting(): + case epState == StateConnecting: + // Initial SYN hasn't been sent yet so initiate a connect. tcpip.AsyncLoading.Add(1) go func() { connectedLoading.Wait() @@ -280,6 +243,27 @@ func (e *endpoint) Resume(s *stack.Stack) { connectingLoading.Done() tcpip.AsyncLoading.Done() }() + case epState == StateSynSent || epState == StateSynRecv: + connectedLoading.Wait() + listenLoading.Wait() + // Initial SYN has been sent/received so we should bind the + // ports start the retransmit timer for the SYNs and let it + // naturally complete the connection. + bind() + e.mu.Lock() + defer e.mu.Unlock() + e.setEndpointState(epState) + r, err := e.stack.FindRoute(e.boundNICID, e.TransportEndpointInfo.ID.LocalAddress, e.TransportEndpointInfo.ID.RemoteAddress, e.effectiveNetProtos[0], false /* multicastLoop */) + if err != nil { + 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)) + if err != nil { + panic(fmt.Sprintf("newBackOffTimer(_, %s, %s, _) failed: %s", InitialRTO, MaxRTO, err)) + } + e.h.retransmitTimer = timer + connectingLoading.Done() case epState == StateBound: tcpip.AsyncLoading.Add(1) go func() { diff --git a/pkg/tcpip/transport/tcp/forwarder.go b/pkg/tcpip/transport/tcp/forwarder.go index cbe2d3859..fcc1f9dc7 100644 --- a/pkg/tcpip/transport/tcp/forwarder.go +++ b/pkg/tcpip/transport/tcp/forwarder.go @@ -164,9 +164,5 @@ func (r *ForwarderRequest) CreateEndpoint(queue *waiter.Queue) (tcpip.Endpoint, return nil, err } - // Start the protocol goroutine. Note that the endpoint is returned - // from performHandshake locked. - ep.startAcceptedLoop() // +checklocksforce - return ep, nil } diff --git a/pkg/tcpip/transport/tcp/protocol.go b/pkg/tcpip/transport/tcp/protocol.go index af9edb12e..34ab1d9b2 100644 --- a/pkg/tcpip/transport/tcp/protocol.go +++ b/pkg/tcpip/transport/tcp/protocol.go @@ -486,6 +486,16 @@ func (p *protocol) Wait() { p.dispatcher.wait() } +// Pause implements stack.TransportProtocol.Pause. +func (p *protocol) Pause() { + p.dispatcher.pause() +} + +// Resume implements stack.TransportProtocol.Resume. +func (p *protocol) Resume() { + p.dispatcher.resume() +} + // Parse implements stack.TransportProtocol.Parse. func (*protocol) Parse(pkt *stack.PacketBuffer) bool { return parse.TCP(pkt) diff --git a/pkg/tcpip/transport/tcp/rack.go b/pkg/tcpip/transport/tcp/rack.go index fe0a47e13..b8d0bb653 100644 --- a/pkg/tcpip/transport/tcp/rack.go +++ b/pkg/tcpip/transport/tcp/rack.go @@ -186,10 +186,11 @@ func (s *sender) schedulePTO() { // probeTimerExpired is the same as TLP_send_probe() as defined in // https://tools.ietf.org/html/draft-ietf-tcpm-rack-08#section-7.5.2. +// // +checklocks:s.ep.mu -func (s *sender) probeTimerExpired() tcpip.Error { - if !s.probeTimer.checkExpiration() { - return nil +func (s *sender) probeTimerExpired() { + if s.probeTimer.isZero() || !s.probeTimer.checkExpiration() { + return } var dataSent bool @@ -230,7 +231,7 @@ func (s *sender) probeTimerExpired() tcpip.Error { // 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 nil + return } // detectTLPRecovery detects if recovery was accomplished by the loss probes @@ -385,17 +386,16 @@ func (rc *rackControl) detectLoss(rcvTime tcpip.MonotonicTime) int { // reorderTimerExpired will retransmit the segments which have not been acked // before the reorder timer expired. +// // +checklocks:rc.snd.ep.mu -func (rc *rackControl) reorderTimerExpired() tcpip.Error { - // Check if the timer actually expired or if it's a spurious wake due - // to a previously orphaned runtime timer. - if !rc.snd.reorderTimer.checkExpiration() { - return nil +func (rc *rackControl) reorderTimerExpired() { + if rc.snd.reorderTimer.isZero() || !rc.snd.reorderTimer.checkExpiration() { + return } numLost := rc.detectLoss(rc.snd.ep.stack.Clock().NowMonotonic()) if numLost == 0 { - return nil + return } fastRetransmit := false @@ -406,10 +406,11 @@ func (rc *rackControl) reorderTimerExpired() tcpip.Error { } rc.DoRecovery(nil, fastRetransmit) - return nil + return } // DoRecovery implements lossRecovery.DoRecovery. +// // +checklocks:rc.snd.ep.mu func (rc *rackControl) DoRecovery(_ *segment, fastRetransmit bool) { snd := rc.snd diff --git a/pkg/tcpip/transport/tcp/rcv.go b/pkg/tcpip/transport/tcp/rcv.go index 0cccb9d8a..88251d576 100644 --- a/pkg/tcpip/transport/tcp/rcv.go +++ b/pkg/tcpip/transport/tcp/rcv.go @@ -287,9 +287,9 @@ func (r *receiver) consumeSegment(s *segment, segSeq seqnum.Value, segLen seqnum for i := first; i < len(r.pendingRcvdSegments); i++ { r.PendingBufUsed -= r.pendingRcvdSegments[i].segMemSize() r.pendingRcvdSegments[i].DecRef() - // Note that slice truncation does not allow garbage collection of - // truncated items, thus truncated items must be set to nil to avoid - // memory leaks. + // Note that slice truncation does not allow garbage + // collection of truncated items, thus truncated items + // must be set to nil to avoid memory leaks. r.pendingRcvdSegments[i] = nil } r.pendingRcvdSegments = r.pendingRcvdSegments[:first] @@ -303,11 +303,12 @@ func (r *receiver) consumeSegment(s *segment, segSeq seqnum.Value, segLen seqnum switch r.ep.EndpointState() { case StateFinWait1: r.ep.setEndpointState(StateFinWait2) - // Notify protocol goroutine that we have received an - // ACK to our FIN so that it can start the FIN_WAIT2 - // timer to abort connection if the other side does - // not close within 2MSL. - r.ep.notifyProtocolGoroutine(notifyClose) + if e := r.ep; e.closed { + // The socket has been closed and we are in + // FIN-WAIT-2 so start the FIN-WAIT-2 timer. + e.finWait2Timer = e.stack.Clock().AfterFunc(e.tcpLingerTimeout, e.finWait2TimerExpired) + } + case StateClosing: r.ep.setEndpointState(StateTimeWait) case StateLastAck: diff --git a/pkg/tcpip/transport/tcp/snd.go b/pkg/tcpip/transport/tcp/snd.go index e518799a3..20267e4a0 100644 --- a/pkg/tcpip/transport/tcp/snd.go +++ b/pkg/tcpip/transport/tcp/snd.go @@ -20,7 +20,6 @@ import ( "sort" "time" - "gvisor.dev/gvisor/pkg/sleep" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/buffer" @@ -106,8 +105,7 @@ type sender struct { writeNext *segment writeList segmentList - resendTimer timer `state:"nosave"` - resendWaker sleep.Waker `state:"nosave"` + resendTimer timer `state:"nosave"` // rtt.TCPRTTState.SRTT and rtt.TCPRTTState.RTTVar are the "smoothed // round-trip time", and "round-trip time variation", as defined in @@ -138,12 +136,10 @@ type sender struct { // reorderTimer is the timer used to retransmit the segments after RACK // detects them as lost. - reorderTimer timer `state:"nosave"` - reorderWaker sleep.Waker `state:"nosave"` + reorderTimer timer `state:"nosave"` - // probeTimer and probeWaker are used to schedule PTO for RACK TLP algorithm. - probeTimer timer `state:"nosave"` - probeWaker sleep.Waker `state:"nosave"` + // probeTimer is used to schedule PTO for RACK TLP algorithm. + probeTimer timer `state:"nosave"` // spuriousRecovery indicates whether the sender entered recovery // spuriously as described in RFC3522 Section 3.2. @@ -207,9 +203,9 @@ func newSender(ep *endpoint, iss, irs seqnum.Value, sndWnd seqnum.Size, mss uint s.SndWndScale = uint8(sndWndScale) } - s.resendTimer.init(s.ep.stack.Clock(), &s.resendWaker) - s.reorderTimer.init(s.ep.stack.Clock(), &s.reorderWaker) - s.probeTimer.init(s.ep.stack.Clock(), &s.probeWaker) + s.resendTimer.init(s.ep.stack.Clock(), maybeFailTimerHandler(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)) s.ep.AssertLockHeld(ep) s.updateMaxPayloadSize(int(ep.route.MTU()), 0) @@ -432,11 +428,11 @@ func (s *sender) resendSegment() { // Returns true if the connection is still usable, or false if the connection // is deemed lost. // +checklocks:s.ep.mu -func (s *sender) retransmitTimerExpired() bool { +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.checkExpiration() { - return true + if s.resendTimer.isZero() || !s.resendTimer.checkExpiration() { + return nil } // Initialize the variables used to detect spurious recovery after @@ -450,7 +446,7 @@ func (s *sender) retransmitTimerExpired() bool { // when writeList is empty. Remove this once we have a proper fix for this // issue. if s.writeList.Front() == nil { - return true + return nil } s.ep.stack.Stats().TCP.Timeouts.Increment() @@ -485,7 +481,8 @@ func (s *sender) retransmitTimerExpired() bool { // window probes were acknowledged. // net/ipv4/tcp_timer.c::tcp_probe_timer() if remaining <= 0 || s.unackZeroWindowProbes >= s.maxRetries { - return false + s.ep.stack.Stats().TCP.EstablishedTimedout.Increment() + return &tcpip.ErrTimeout{} } // Set new timeout. The timer will be restarted by the call to sendData @@ -556,19 +553,20 @@ func (s *sender) retransmitTimerExpired() bool { // indefinitely. As long as the receiving TCP continues to send // acknowledgments in response to the probe segments, the sending TCP // MUST allow the connection to stay open. - return true + return nil } seg := s.writeNext // RFC 1122 4.2.3.5: Close the connection when the number of // retransmissions for this segment is beyond a limit. if seg != nil && seg.xmitCount > s.maxRetries { - return false + s.ep.stack.Stats().TCP.EstablishedTimedout.Increment() + return &tcpip.ErrTimeout{} } s.sendData() - return true + return nil } // pCount returns the number of packets in the segment. Due to GSO, a segment diff --git a/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go b/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go index dece59835..d1d8ffd37 100644 --- a/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go +++ b/pkg/tcpip/transport/tcp/test/e2e/tcp_test.go @@ -179,16 +179,9 @@ func TestConnectICMPError(t *testing.T) { checker.IPv4(t, syn, checker.TCP(checker.TCPFlags(header.TCPFlagSyn))) wep := ep.(interface { - StopWork() - ResumeWork() LastErrorLocked() tcpip.Error }) - // Stop the protocol loop, ensure that the ICMP error is processed and - // the last ICMP error is read before the loop is resumed. This sanity - // tests the handshake completion logic on ICMP errors. - wep.StopWork() - c.SendICMPPacket(header.ICMPv4DstUnreachable, header.ICMPv4HostUnreachable, nil, syn, e2e.DefaultMTU) for { @@ -201,8 +194,6 @@ func TestConnectICMPError(t *testing.T) { time.Sleep(time.Millisecond) } - wep.ResumeWork() - <-notifyCh // The stack would have unregistered the endpoint because of the ICMP error. @@ -373,9 +364,8 @@ func TestTCPResetsSentIncrement(t *testing.T) { } } -// TestTCPResetsSentNoICMP confirms that we don't get an ICMP -// DstUnreachable packet when we try send a packet which is not part -// of an active session. +// TestTCPResetsSentNoICMP confirms that we don't get an ICMP DstUnreachable +// packet when we try send a packet which is not part of an active session. func TestTCPResetsSentNoICMP(t *testing.T) { c := context.New(t, e2e.DefaultMTU) defer c.Cleanup() @@ -8654,13 +8644,6 @@ func TestReadAfterCloseWithBufferedData(t *testing.T) { } } -func TestReleaseAfterClose(t *testing.T) { - c := context.New(t, e2e.DefaultMTU) - c.CreateConnectedWithOptionsNoDelay(header.TCPSynOptions{}) - c.CloseNoWait() - c.EP.Release() -} - func TestReleaseDanglingEndpoints(t *testing.T) { c := context.New(t, e2e.DefaultMTU) defer c.Cleanup() @@ -8683,8 +8666,18 @@ func TestReleaseDanglingEndpoints(t *testing.T) { ) tcpip.ReleaseDanglingEndpoints() - // Now send an ACK and it should trigger a RST as Release should Close the - // endpoint. + // ReleaseDanglingEndpoints should abort the half-closed endpoint causing + // a RST to be sent. + checker.IPv4(t, c.GetPacket(), + checker.TCP( + checker.DstPort(context.TestPort), + checker.TCPSeqNum(uint32(c.IRS)+2), + checker.TCPAckNum(uint32(iss)), + checker.TCPFlags(header.TCPFlagRst|header.TCPFlagAck), + ), + ) + + // Now send an ACK and it should trigger a RST as the endpoint is aborted. c.SendPacket(nil, &context.Headers{ SrcPort: context.TestPort, DstPort: c.Port, diff --git a/pkg/tcpip/transport/tcp/testing/context/context.go b/pkg/tcpip/transport/tcp/testing/context/context.go index 0f0b49d6d..549590526 100644 --- a/pkg/tcpip/transport/tcp/testing/context/context.go +++ b/pkg/tcpip/transport/tcp/testing/context/context.go @@ -280,8 +280,8 @@ func NewWithOpts(t *testing.T, opts Options) *Context { func (c *Context) Cleanup() { if c.EP != nil { c.EP.Close() - c.EP.Release() } + tcpip.ReleaseDanglingEndpoints() c.Stack().Close() c.Stack().Wait() c.linkEP.Close() diff --git a/pkg/tcpip/transport/tcp/timer.go b/pkg/tcpip/transport/tcp/timer.go index 5645c772e..208009263 100644 --- a/pkg/tcpip/transport/tcp/timer.go +++ b/pkg/tcpip/transport/tcp/timer.go @@ -18,7 +18,6 @@ import ( "math" "time" - "gvisor.dev/gvisor/pkg/sleep" "gvisor.dev/gvisor/pkg/tcpip" ) @@ -69,17 +68,15 @@ type timer struct { timer tcpip.Timer } -// init initializes the timer. Once it expires, it the given waker will be -// asserted. -func (t *timer) init(clock tcpip.Clock, w *sleep.Waker) { +// init initializes the timer. Once it expires the function callback +// passed will be called. +func (t *timer) init(clock tcpip.Clock, f func()) { t.state = timerStateDisabled t.clock = clock - // Initialize a clock timer that will assert the waker, then + // Initialize a clock timer that will call the callback func, then // immediately stop it. - t.timer = t.clock.AfterFunc(math.MaxInt64, func() { - w.Assert() - }) + t.timer = t.clock.AfterFunc(math.MaxInt64, f) t.timer.Stop() } @@ -93,10 +90,16 @@ 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{} +} + // checkExpiration checks if the given timer has actually expired, it should be -// called whenever a sleeper wakes up due to the waker being asserted, and is -// used to check if it's a supurious wake (due to a previously orphaned timer) -// or a legitimate one. +// 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 +// legitimate one. func (t *timer) checkExpiration() bool { // Transition to fully disabled state if we're just consuming an // orphaned timer. diff --git a/pkg/tcpip/transport/tcp/timer_test.go b/pkg/tcpip/transport/tcp/timer_test.go index 479752de7..30f8c8e6f 100644 --- a/pkg/tcpip/transport/tcp/timer_test.go +++ b/pkg/tcpip/transport/tcp/timer_test.go @@ -32,7 +32,7 @@ func TestCleanup(t *testing.T) { tmr := timer{} w := sleep.Waker{} - tmr.init(clock, &w) + tmr.init(clock, w.Assert) tmr.enable(timerDurationSeconds * time.Second) tmr.cleanup() diff --git a/pkg/tcpip/transport/udp/endpoint.go b/pkg/tcpip/transport/udp/endpoint.go index 0e86c2541..17fd766cd 100644 --- a/pkg/tcpip/transport/udp/endpoint.go +++ b/pkg/tcpip/transport/udp/endpoint.go @@ -157,8 +157,6 @@ func (e *endpoint) Abort() { e.Close() } -func (*endpoint) Release() {} - // Close puts the endpoint in a closed state and frees all resources // associated with it. func (e *endpoint) Close() { diff --git a/pkg/tcpip/transport/udp/protocol.go b/pkg/tcpip/transport/udp/protocol.go index 1171aeb79..069d71e91 100644 --- a/pkg/tcpip/transport/udp/protocol.go +++ b/pkg/tcpip/transport/udp/protocol.go @@ -109,6 +109,12 @@ func (*protocol) Close() {} // Wait implements stack.TransportProtocol.Wait. func (*protocol) Wait() {} +// Pause implements stack.TransportProtocol.Pause. +func (*protocol) Pause() {} + +// Resume implements stack.TransportProtocol.Resume. +func (*protocol) Resume() {} + // Parse implements stack.TransportProtocol.Parse. func (*protocol) Parse(pkt *stack.PacketBuffer) bool { return parse.UDP(pkt)