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)