From 365f85680d19309afb99607b3f0f62a32a9b4009 Mon Sep 17 00:00:00 2001 From: Kevin Krakauer Date: Mon, 18 Mar 2024 15:24:33 -0700 Subject: [PATCH] netstack: make TCP's endpoint type public This is in preparation for a child CL. PiperOrigin-RevId: 616963627 --- pkg/tcpip/transport/tcp/BUILD | 4 +- pkg/tcpip/transport/tcp/accept.go | 24 +-- pkg/tcpip/transport/tcp/connect.go | 60 +++--- pkg/tcpip/transport/tcp/dispatcher.go | 20 +- pkg/tcpip/transport/tcp/endpoint.go | 233 +++++++++++----------- pkg/tcpip/transport/tcp/endpoint_state.go | 20 +- pkg/tcpip/transport/tcp/rcv.go | 4 +- pkg/tcpip/transport/tcp/segment.go | 4 +- pkg/tcpip/transport/tcp/segment_queue.go | 2 +- pkg/tcpip/transport/tcp/snd.go | 4 +- 10 files changed, 189 insertions(+), 186 deletions(-) diff --git a/pkg/tcpip/transport/tcp/BUILD b/pkg/tcpip/transport/tcp/BUILD index 10aabc402..e7d508ff2 100644 --- a/pkg/tcpip/transport/tcp/BUILD +++ b/pkg/tcpip/transport/tcp/BUILD @@ -36,8 +36,8 @@ go_template_instance( prefix = "endpoint", template = "//pkg/ilist:generic_list", types = { - "Element": "*endpoint", - "Linker": "*endpoint", + "Element": "*Endpoint", + "Linker": "*Endpoint", }, ) diff --git a/pkg/tcpip/transport/tcp/accept.go b/pkg/tcpip/transport/tcp/accept.go index 7e91b026d..adcfdcfd5 100644 --- a/pkg/tcpip/transport/tcp/accept.go +++ b/pkg/tcpip/transport/tcp/accept.go @@ -85,7 +85,7 @@ type listenContext struct { // listenEP is a reference to the listening endpoint associated with // this context. Can be nil if the context is created by the forwarder. - listenEP *endpoint + listenEP *Endpoint // hasherMu protects hasher. hasherMu sync.Mutex @@ -107,7 +107,7 @@ func timeStamp(clock tcpip.Clock) uint32 { } // newListenContext creates a new listen context. -func newListenContext(stk *stack.Stack, protocol *protocol, listenEP *endpoint, rcvWnd seqnum.Size, v6Only bool, netProto tcpip.NetworkProtocolNumber) *listenContext { +func newListenContext(stk *stack.Stack, protocol *protocol, listenEP *Endpoint, rcvWnd seqnum.Size, v6Only bool, netProto tcpip.NetworkProtocolNumber) *listenContext { l := &listenContext{ stack: stk, protocol: protocol, @@ -183,7 +183,7 @@ func (l *listenContext) isCookieValid(id stack.TransportEndpointID, cookie seqnu // the connection parameters given by the arguments. The newly created endpoint // will be locked. // +checklocksacquire:n.mu -func (l *listenContext) createConnectingEndpoint(s *segment, rcvdSynOpts header.TCPSynOptions, queue *waiter.Queue) (n *endpoint, _ tcpip.Error) { +func (l *listenContext) createConnectingEndpoint(s *segment, rcvdSynOpts header.TCPSynOptions, queue *waiter.Queue) (n *Endpoint, _ tcpip.Error) { // Create a new endpoint. netProto := l.netProto if netProto == 0 { @@ -302,7 +302,7 @@ func (l *listenContext) startHandshake(s *segment, opts header.TCPSynOptions, qu // 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) { +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) @@ -357,7 +357,7 @@ func (l *listenContext) performHandshake(s *segment, opts header.TCPSynOptions, // // +checklocks:e.mu // +checklocks:n.mu -func (e *endpoint) propagateInheritableOptionsLocked(n *endpoint) { +func (e *Endpoint) propagateInheritableOptionsLocked(n *Endpoint) { n.userTimeout = e.userTimeout n.portFlags = e.portFlags n.boundBindToDevice = e.boundBindToDevice @@ -370,7 +370,7 @@ func (e *endpoint) propagateInheritableOptionsLocked(n *endpoint) { // Precondition: e.propagateInheritableOptionsLocked has been called. // // +checklocks:e.mu -func (e *endpoint) reserveTupleLocked() bool { +func (e *Endpoint) reserveTupleLocked() bool { dest := tcpip.FullAddress{ Addr: e.TransportEndpointInfo.ID.RemoteAddress, Port: e.TransportEndpointInfo.ID.RemotePort, @@ -400,11 +400,11 @@ func (e *endpoint) reserveTupleLocked() bool { // This is strictly not required normally as a socket that was never accepted // can't really have any registered waiters except when stack.Wait() is called // which waits for all registered endpoints to stop and expects an EventHUp. -func (e *endpoint) notifyAborted() { +func (e *Endpoint) notifyAborted() { e.waiterQueue.Notify(waiter.EventHUp | waiter.EventErr | waiter.ReadableEvents | waiter.WritableEvents) } -func (e *endpoint) acceptQueueIsFull() bool { +func (e *Endpoint) acceptQueueIsFull() bool { e.acceptMu.Lock() full := e.acceptQueue.isFull() e.acceptMu.Unlock() @@ -416,11 +416,11 @@ type acceptQueue struct { // NB: this could be an endpointList, but ilist only permits endpoints to // belong to one list at a time, and endpoints are already stored in the // dispatcher's list. - endpoints list.List `state:".([]*endpoint)"` + endpoints list.List `state:".([]*Endpoint)"` // pendingEndpoints is a set of all endpoints for which a handshake is // in progress. - pendingEndpoints map[*endpoint]struct{} + pendingEndpoints map[*Endpoint]struct{} // capacity is the maximum number of endpoints that can be in endpoints. capacity int @@ -434,7 +434,7 @@ func (a *acceptQueue) isFull() bool { // and needs to handle it. // // +checklocks:e.mu -func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Error { +func (e *Endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Error { e.rcvQueueMu.Lock() rcvClosed := e.RcvClosed e.rcvQueueMu.Unlock() @@ -561,7 +561,7 @@ func (e *endpoint) handleListenSegment(ctx *listenContext, s *segment) tcpip.Err } for _, netProto := range netProtos { if newEP := e.stack.FindTransportEndpoint(netProto, ProtocolNumber, s.id, s.pkt.NICID); newEP != nil && newEP != e { - tcpEP := newEP.(*endpoint) + tcpEP := newEP.(*Endpoint) if !tcpEP.EndpointState().connected() { continue } diff --git a/pkg/tcpip/transport/tcp/connect.go b/pkg/tcpip/transport/tcp/connect.go index dd3fa76a3..4a7297734 100644 --- a/pkg/tcpip/transport/tcp/connect.go +++ b/pkg/tcpip/transport/tcp/connect.go @@ -63,8 +63,8 @@ const ( // // +stateify savable type handshake struct { - ep *endpoint - listenEP *endpoint + ep *Endpoint + listenEP *Endpoint state handshakeState active bool flags header.TCPFlags @@ -122,7 +122,7 @@ type handshake struct { // 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 timerHandler(e *endpoint, f func() tcpip.Error) func() { +func timerHandler(e *Endpoint, f func() tcpip.Error) func() { return func() { e.mu.Lock() if err := f(); err != nil { @@ -155,7 +155,7 @@ func timerHandler(e *endpoint, f func() tcpip.Error) func() { // +checklocks:e.mu // +checklocksacquire:h.ep.mu -func (e *endpoint) newHandshake() (h *handshake) { +func (e *Endpoint) newHandshake() (h *handshake) { h = &handshake{ ep: e, active: true, @@ -178,7 +178,7 @@ func (e *endpoint) newHandshake() (h *handshake) { // +checklocks:e.mu // +checklocksacquire:h.ep.mu -func (e *endpoint) newPassiveHandshake(isn, irs seqnum.Value, opts header.TCPSynOptions, deferAccept time.Duration) (h *handshake) { +func (e *Endpoint) newPassiveHandshake(isn, irs seqnum.Value, opts header.TCPSynOptions, deferAccept time.Duration) (h *handshake) { h = e.newHandshake() h.resetToSynRcvd(isn, irs, opts, deferAccept) return h @@ -796,7 +796,7 @@ type tcpFields struct { txHash uint32 } -func (e *endpoint) sendSynTCP(r *stack.Route, tf tcpFields, opts header.TCPSynOptions) tcpip.Error { +func (e *Endpoint) sendSynTCP(r *stack.Route, tf tcpFields, opts header.TCPSynOptions) tcpip.Error { tf.opts = makeSynOptions(opts) // We ignore SYN send errors and let the callers re-attempt send. p := stack.NewPacketBuffer(stack.PacketBufferOptions{ReserveHeaderBytes: header.TCPMinimumSize + int(r.MaxHeaderLength()) + len(tf.opts)}) @@ -809,7 +809,7 @@ func (e *endpoint) sendSynTCP(r *stack.Route, tf tcpFields, opts header.TCPSynOp } // This method takes ownership of pkt. -func (e *endpoint) sendTCP(r *stack.Route, tf tcpFields, pkt *stack.PacketBuffer, gso stack.GSO) tcpip.Error { +func (e *Endpoint) sendTCP(r *stack.Route, tf tcpFields, pkt *stack.PacketBuffer, gso stack.GSO) tcpip.Error { tf.txHash = e.txHash if err := sendTCP(r, tf, pkt, gso, e.owner); err != nil { e.stats.SendErrors.SegmentSendToNetworkFailed.Increment() @@ -925,7 +925,7 @@ func sendTCP(r *stack.Route, tf tcpFields, pkt *stack.PacketBuffer, gso stack.GS } // makeOptions makes an options slice. -func (e *endpoint) makeOptions(sackBlocks []header.SACKBlock) []byte { +func (e *Endpoint) makeOptions(sackBlocks []header.SACKBlock) []byte { options := getOptions() offset := 0 @@ -964,7 +964,7 @@ func (e *endpoint) makeOptions(sackBlocks []header.SACKBlock) []byte { } // sendEmptyRaw sends a TCP segment with no payload to the endpoint's peer. -func (e *endpoint) sendEmptyRaw(flags header.TCPFlags, seq, ack seqnum.Value, rcvWnd seqnum.Size) tcpip.Error { +func (e *Endpoint) sendEmptyRaw(flags header.TCPFlags, seq, ack seqnum.Value, rcvWnd seqnum.Size) tcpip.Error { pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{}) defer pkt.DecRef() return e.sendRaw(pkt, flags, seq, ack, rcvWnd) @@ -972,7 +972,7 @@ func (e *endpoint) sendEmptyRaw(flags header.TCPFlags, seq, ack seqnum.Value, rc // sendRaw sends a TCP segment to the endpoint's peer. This method takes // ownership of pkt. pkt must not have any headers set. -func (e *endpoint) sendRaw(pkt *stack.PacketBuffer, flags header.TCPFlags, seq, ack seqnum.Value, rcvWnd seqnum.Size) tcpip.Error { +func (e *Endpoint) sendRaw(pkt *stack.PacketBuffer, flags header.TCPFlags, seq, ack seqnum.Value, rcvWnd seqnum.Size) tcpip.Error { var sackBlocks []header.SACKBlock if e.EndpointState() == StateEstablished && e.rcv.pendingRcvdSegments.Len() > 0 && (flags&header.TCPFlagAck != 0) { sackBlocks = e.sack.Blocks[:e.sack.NumBlocks] @@ -994,7 +994,7 @@ func (e *endpoint) sendRaw(pkt *stack.PacketBuffer, flags header.TCPFlags, seq, // +checklocks:e.mu // +checklocksalias:e.snd.ep.mu=e.mu -func (e *endpoint) sendData(next *segment) { +func (e *Endpoint) sendData(next *segment) { // Initialize the next segment to write if it's currently nil. if e.snd.writeNext == nil { if next == nil { @@ -1012,7 +1012,7 @@ func (e *endpoint) sendData(next *segment) { // indicating that the connection is being reset due to receiving a RST. This // method must only be called from the protocol goroutine. // +checklocks:e.mu -func (e *endpoint) resetConnectionLocked(err tcpip.Error) { +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.hardError = err @@ -1047,7 +1047,7 @@ func (e *endpoint) resetConnectionLocked(err tcpip.Error) { // delivered to this endpoint from the demuxer when the endpoint // is transitioned to StateClose. // +checklocks:e.mu -func (e *endpoint) transitionToStateCloseLocked() { +func (e *Endpoint) transitionToStateCloseLocked() { s := e.EndpointState() if s == StateClose { return @@ -1066,7 +1066,7 @@ func (e *endpoint) transitionToStateCloseLocked() { // segment to any other endpoint other than the current one. This is called // only when the endpoint is in StateClose and we want to deliver the segment // to any other listening endpoint. We reply with RST if we cannot find one. -func (e *endpoint) tryDeliverSegmentFromClosedEndpoint(s *segment) { +func (e *Endpoint) tryDeliverSegmentFromClosedEndpoint(s *segment) { ep := e.stack.FindTransportEndpoint(e.NetProto, e.TransProto, e.TransportEndpointInfo.ID, s.pkt.NICID) if ep == nil && e.NetProto == header.IPv6ProtocolNumber && e.TransportEndpointInfo.ID.LocalAddress.To4() != (tcpip.Address{}) { // Dual-stack socket, try IPv4. @@ -1088,7 +1088,7 @@ func (e *endpoint) tryDeliverSegmentFromClosedEndpoint(s *segment) { panic(fmt.Sprintf("current endpoint not removed from demuxer, enqueuing segments to itself, endpoint in state %v", e.EndpointState())) } - if ep := ep.(*endpoint); ep.enqueueSegment(s) { + if ep := ep.(*Endpoint); ep.enqueueSegment(s) { ep.notifyProcessor() } } @@ -1096,7 +1096,7 @@ func (e *endpoint) tryDeliverSegmentFromClosedEndpoint(s *segment) { // Drain segment queue from the endpoint and try to re-match the segment to a // different endpoint. This is used when the current endpoint is transitioned to // StateClose and has been unregistered from the transport demuxer. -func (e *endpoint) drainClosingSegmentQueue() { +func (e *Endpoint) drainClosingSegmentQueue() { for { s := e.segmentQueue.dequeue() if s == nil { @@ -1109,7 +1109,7 @@ func (e *endpoint) drainClosingSegmentQueue() { } // +checklocks:e.mu -func (e *endpoint) handleReset(s *segment) (ok bool, err tcpip.Error) { +func (e *Endpoint) handleReset(s *segment) (ok bool, err tcpip.Error) { if e.rcv.acceptable(s.sequenceNumber, 0) { // RFC 793, page 37 states that "in all states // except SYN-SENT, all reset (RST) segments are @@ -1158,7 +1158,7 @@ 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() tcpip.Error { +func (e *Endpoint) handleSegmentsLocked() tcpip.Error { sndUna := e.snd.SndUna for i := 0; i < maxSegmentsPerWake; i++ { if state := e.EndpointState(); state.closed() || state == StateTimeWait || state == StateError { @@ -1200,7 +1200,7 @@ func (e *endpoint) handleSegmentsLocked() tcpip.Error { } // +checklocks:e.mu -func (e *endpoint) probeSegmentLocked() { +func (e *Endpoint) probeSegmentLocked() { if fn := e.probe; fn != nil { var state stack.TCPEndpointState e.completeStateLocked(&state) @@ -1214,7 +1214,7 @@ func (e *endpoint) probeSegmentLocked() { // +checklocks:e.mu // +checklocksalias:e.rcv.ep.mu=e.mu // +checklocksalias:e.snd.ep.mu=e.mu -func (e *endpoint) handleSegmentLocked(s *segment) (cont bool, err tcpip.Error) { +func (e *Endpoint) handleSegmentLocked(s *segment) (cont bool, err tcpip.Error) { // Invoke the tcp probe if installed. The tcp probe function will update // the TCPEndpointState after the segment is processed. defer e.probeSegmentLocked() @@ -1289,7 +1289,7 @@ func (e *endpoint) handleSegmentLocked(s *segment) (cont bool, err tcpip.Error) // from the other side after a number of tries, we terminate the connection. // +checklocks:e.mu // +checklocksalias:e.snd.ep.mu=e.mu -func (e *endpoint) keepaliveTimerExpired() tcpip.Error { +func (e *Endpoint) keepaliveTimerExpired() tcpip.Error { userTimeout := e.userTimeout e.keepalive.Lock() @@ -1323,7 +1323,7 @@ func (e *endpoint) keepaliveTimerExpired() tcpip.Error { // resetKeepaliveTimer restarts or stops the keepalive timer, depending on // whether it is enabled for this endpoint. -func (e *endpoint) resetKeepaliveTimer(receivedData bool) { +func (e *Endpoint) resetKeepaliveTimer(receivedData bool) { e.keepalive.Lock() defer e.keepalive.Unlock() if e.keepalive.timer.isUninitialized() { @@ -1349,7 +1349,7 @@ func (e *endpoint) resetKeepaliveTimer(receivedData bool) { } // disableKeepaliveTimer stops the keepalive timer. -func (e *endpoint) disableKeepaliveTimer() { +func (e *Endpoint) disableKeepaliveTimer() { e.keepalive.Lock() e.keepalive.timer.disable() e.keepalive.Unlock() @@ -1357,7 +1357,7 @@ func (e *endpoint) disableKeepaliveTimer() { // finWait2TimerExpired is called when the FIN-WAIT-2 timeout is hit // and the peer hasn't sent us a FIN. -func (e *endpoint) finWait2TimerExpired() { +func (e *Endpoint) finWait2TimerExpired() { e.mu.Lock() e.transitionToStateCloseLocked() e.mu.Unlock() @@ -1366,7 +1366,7 @@ func (e *endpoint) finWait2TimerExpired() { } // +checklocks:e.mu -func (e *endpoint) handshakeFailed(err tcpip.Error) { +func (e *Endpoint) handshakeFailed(err tcpip.Error) { e.lastErrorMu.Lock() e.lastError = err e.lastErrorMu.Unlock() @@ -1386,7 +1386,7 @@ func (e *endpoint) handshakeFailed(err tcpip.Error) { // state. // +checklocks:e.mu // +checklocksalias:e.rcv.ep.mu=e.mu -func (e *endpoint) handleTimeWaitSegments() (extendTimeWait bool, reuseTW func()) { +func (e *Endpoint) handleTimeWaitSegments() (extendTimeWait bool, reuseTW func()) { for i := 0; i < maxSegmentsPerWake; i++ { s := e.segmentQueue.dequeue() if s == nil { @@ -1407,7 +1407,7 @@ func (e *endpoint) handleTimeWaitSegments() (extendTimeWait bool, reuseTW func() } for _, netProto := range netProtos { if listenEP := e.stack.FindTransportEndpoint(netProto, info.TransProto, newID, s.pkt.NICID); listenEP != nil { - tcpEP := listenEP.(*endpoint) + tcpEP := listenEP.(*Endpoint) if EndpointState(tcpEP.State()) == StateListen { reuseTW = func() { if !tcpEP.enqueueSegment(s) { @@ -1432,7 +1432,7 @@ func (e *endpoint) handleTimeWaitSegments() (extendTimeWait bool, reuseTW func() } // +checklocks:e.mu -func (e *endpoint) getTimeWaitDuration() time.Duration { +func (e *Endpoint) getTimeWaitDuration() time.Duration { timeWaitDuration := DefaultTCPTimeWaitTimeout // Get the stack wide configuration. @@ -1446,7 +1446,7 @@ func (e *endpoint) getTimeWaitDuration() time.Duration { // 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() { +func (e *Endpoint) timeWaitTimerExpired() { e.mu.Lock() if e.EndpointState() != StateTimeWait { e.mu.Unlock() @@ -1459,7 +1459,7 @@ func (e *endpoint) timeWaitTimerExpired() { } // notifyProcessor queues this endpoint for processing to its TCP processor. -func (e *endpoint) notifyProcessor() { +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 diff --git a/pkg/tcpip/transport/tcp/dispatcher.go b/pkg/tcpip/transport/tcp/dispatcher.go index a2d4adbb1..043b84109 100644 --- a/pkg/tcpip/transport/tcp/dispatcher.go +++ b/pkg/tcpip/transport/tcp/dispatcher.go @@ -35,7 +35,7 @@ type epQueue struct { } // enqueue adds e to the queue if the endpoint is not already on the queue. -func (q *epQueue) enqueue(e *endpoint) { +func (q *epQueue) enqueue(e *Endpoint) { q.mu.Lock() defer q.mu.Unlock() e.pendingProcessingMu.Lock() @@ -50,7 +50,7 @@ func (q *epQueue) enqueue(e *endpoint) { // dequeue removes and returns the first element from the queue if available, // returns nil otherwise. -func (q *epQueue) dequeue() *endpoint { +func (q *epQueue) dequeue() *Endpoint { q.mu.Lock() if e := q.list.Front(); e != nil { q.list.Remove(e) @@ -87,7 +87,7 @@ func (p *processor) close() { p.closeWaker.Assert() } -func (p *processor) queueEndpoint(ep *endpoint) { +func (p *processor) queueEndpoint(ep *Endpoint) { // Queue an endpoint for processing by the processor goroutine. p.epQ.enqueue(ep) p.newEndpointWaker.Assert() @@ -97,7 +97,7 @@ func (p *processor) queueEndpoint(ep *endpoint) { // of its associated listening endpoint. // // +checklocks:ep.mu -func deliverAccepted(ep *endpoint) bool { +func deliverAccepted(ep *Endpoint) bool { lEP := ep.h.listenEP lEP.acceptMu.Lock() @@ -129,7 +129,7 @@ func deliverAccepted(ep *endpoint) bool { // handleConnecting is responsible for TCP processing for an endpoint in one of // the connecting states. -func handleConnecting(ep *endpoint) { +func handleConnecting(ep *Endpoint) { if !ep.TryLock() { return } @@ -172,7 +172,7 @@ func handleConnecting(ep *endpoint) { // handleConnected is responsible for TCP processing for an endpoint in one of // the connected states(StateEstablished, StateFinWait1 etc.) -func handleConnected(ep *endpoint) { +func handleConnected(ep *Endpoint) { if !ep.TryLock() { return } @@ -208,7 +208,7 @@ func handleConnected(ep *endpoint) { // startTimeWait starts a new goroutine to handle TIME-WAIT. // // +checklocks:ep.mu -func startTimeWait(ep *endpoint) { +func startTimeWait(ep *Endpoint) { // Disable close timer as we are now entering real TIME_WAIT. if ep.finWait2Timer != nil { ep.finWait2Timer.Stop() @@ -221,7 +221,7 @@ func startTimeWait(ep *endpoint) { // handleTimeWait is responsible for TCP processing for an endpoint in TIME-WAIT // state. -func handleTimeWait(ep *endpoint) { +func handleTimeWait(ep *Endpoint) { if !ep.TryLock() { return } @@ -251,7 +251,7 @@ func handleTimeWait(ep *endpoint) { // handleListen is responsible for TCP processing for an endpoint in LISTEN // state. -func handleListen(ep *endpoint) { +func handleListen(ep *Endpoint) { if !ep.TryLock() { return } @@ -418,7 +418,7 @@ func (d *dispatcher) queuePacket(stackEP stack.TransportEndpoint, id stack.Trans return } - ep := stackEP.(*endpoint) + ep := stackEP.(*Endpoint) s, err := newIncomingSegment(id, clock, pkt) if err != nil { diff --git a/pkg/tcpip/transport/tcp/endpoint.go b/pkg/tcpip/transport/tcp/endpoint.go index b59238110..4300cb57d 100644 --- a/pkg/tcpip/transport/tcp/endpoint.go +++ b/pkg/tcpip/transport/tcp/endpoint.go @@ -303,7 +303,7 @@ func (sq *sndQueueInfo) CloneState(other *stack.TCPSndBufState) { other.AutoTuneSndBufDisabled = atomicbitops.FromUint32(sq.AutoTuneSndBufDisabled.RacyLoad()) } -// endpoint represents a TCP endpoint. This struct serves as the interface +// Endpoint represents a TCP endpoint. This struct serves as the interface // between users of the endpoint and the protocol implementation; it is legal to // have concurrent goroutines make calls into the endpoint, they are properly // synchronized. The protocol implementation, however, runs in a single @@ -343,7 +343,7 @@ func (sq *sndQueueInfo) CloneState(other *stack.TCPSndBufState) { // e.LockUser/e.UnlockUser methods. // // +stateify savable -type endpoint struct { +type Endpoint struct { stack.TCPEndpointStateInner stack.TransportEndpointInfo tcpip.DefaultSocketOptionsHandler @@ -598,7 +598,7 @@ type endpoint struct { } // UniqueID implements stack.TransportEndpoint.UniqueID. -func (e *endpoint) UniqueID() uint64 { +func (e *Endpoint) UniqueID() uint64 { return e.uniqueID } @@ -620,7 +620,7 @@ func calculateAdvertisedMSS(userMSS uint16, r *stack.Route) uint16 { // isOwnedByUser() returns true if the endpoint lock is currently // held by a user(syscall) goroutine. -func (e *endpoint) isOwnedByUser() bool { +func (e *Endpoint) isOwnedByUser() bool { return e.ownedByUser.Load() == 1 } @@ -634,7 +634,7 @@ func (e *endpoint) isOwnedByUser() bool { // should not be holding the lock for long and spinning reduces latency as we // avoid an expensive sleep/wakeup of the syscall goroutine). // +checklocksacquire:e.mu -func (e *endpoint) LockUser() { +func (e *Endpoint) LockUser() { const iterations = 5 for i := 0; i < iterations; i++ { // Try first if the sock is locked then check if it's owned @@ -689,7 +689,7 @@ func (e *endpoint) LockUser() { // // Precondition: e.LockUser() must have been called before calling e.UnlockUser() // +checklocksrelease:e.mu -func (e *endpoint) UnlockUser() { +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. @@ -722,13 +722,13 @@ func (e *endpoint) UnlockUser() { // StopWork halts packet processing. Only to be used in tests. // +checklocksacquire:e.mu -func (e *endpoint) StopWork() { +func (e *Endpoint) StopWork() { e.mu.Lock() } // ResumeWork resumes packet processing. Only to be used in tests. // +checklocksrelease:e.mu -func (e *endpoint) ResumeWork() { +func (e *Endpoint) ResumeWork() { e.mu.Unlock() } @@ -741,7 +741,7 @@ func (e *endpoint) ResumeWork() { // variable locks. // +checklocks:locked.mu // +checklocksacquire:e.mu -func (e *endpoint) AssertLockHeld(locked *endpoint) { +func (e *Endpoint) AssertLockHeld(locked *Endpoint) { if e != locked { panic("AssertLockHeld failed: locked endpoint != asserting endpoint") } @@ -751,7 +751,7 @@ func (e *endpoint) AssertLockHeld(locked *endpoint) { // adds the necessary checklocks annotations. // TODO(b/226403629): Remove this once checklocks understands TryLock. // +checklocksacquire:e.mu -func (e *endpoint) TryLock() bool { +func (e *Endpoint) TryLock() bool { if e.mu.TryLock() { return true // +checklocksforce } @@ -763,7 +763,7 @@ func (e *endpoint) TryLock() bool { // package but we allow the state to be read freely without holding e.mu. // // +checklocks:e.mu -func (e *endpoint) setEndpointState(state EndpointState) { +func (e *Endpoint) setEndpointState(state EndpointState) { oldstate := EndpointState(e.state.Swap(uint32(state))) switch state { case StateEstablished: @@ -787,18 +787,18 @@ func (e *endpoint) setEndpointState(state EndpointState) { } // EndpointState returns the current state of the endpoint. -func (e *endpoint) EndpointState() EndpointState { +func (e *Endpoint) EndpointState() EndpointState { return EndpointState(e.state.Load()) } // setRecentTimestamp sets the recentTS field to the provided value. -func (e *endpoint) setRecentTimestamp(recentTS uint32) { +func (e *Endpoint) setRecentTimestamp(recentTS uint32) { e.RecentTS = recentTS e.recentTSTime = e.stack.Clock().NowMonotonic() } // recentTimestamp returns the value of the recentTS field. -func (e *endpoint) recentTimestamp() uint32 { +func (e *Endpoint) recentTimestamp() uint32 { return e.RecentTS } @@ -836,8 +836,8 @@ type keepalive struct { waker sleep.Waker `state:"nosave"` } -func newEndpoint(s *stack.Stack, protocol *protocol, netProto tcpip.NetworkProtocolNumber, waiterQueue *waiter.Queue) *endpoint { - e := &endpoint{ +func newEndpoint(s *stack.Stack, protocol *protocol, netProto tcpip.NetworkProtocolNumber, waiterQueue *waiter.Queue) *Endpoint { + e := &Endpoint{ stack: s, protocol: protocol, TransportEndpointInfo: stack.TransportEndpointInfo{ @@ -921,7 +921,7 @@ func newEndpoint(s *stack.Stack, protocol *protocol, netProto tcpip.NetworkProto // Readiness returns the current readiness of the endpoint. For example, if // waiter.EventIn is set, the endpoint is immediately readable. -func (e *endpoint) Readiness(mask waiter.EventMask) waiter.EventMask { +func (e *Endpoint) Readiness(mask waiter.EventMask) waiter.EventMask { result := waiter.EventMask(0) switch e.EndpointState() { @@ -983,7 +983,7 @@ func (e *endpoint) Readiness(mask waiter.EventMask) waiter.EventMask { } // Purging pending rcv segments is only necessary on RST. -func (e *endpoint) purgePendingRcvQueue() { +func (e *Endpoint) purgePendingRcvQueue() { if e.rcv != nil { for e.rcv.pendingRcvdSegments.Len() > 0 { s := heap.Pop(&e.rcv.pendingRcvdSegments).(*segment) @@ -993,7 +993,7 @@ func (e *endpoint) purgePendingRcvQueue() { } // +checklocks:e.mu -func (e *endpoint) purgeReadQueue() { +func (e *Endpoint) purgeReadQueue() { if e.rcv != nil { e.rcvQueueMu.Lock() defer e.rcvQueueMu.Unlock() @@ -1010,7 +1010,7 @@ func (e *endpoint) purgeReadQueue() { } // +checklocks:e.mu -func (e *endpoint) purgeWriteQueue() { +func (e *Endpoint) purgeWriteQueue() { if e.snd != nil { e.sndQueueInfo.sndQueueMu.Lock() defer e.sndQueueInfo.sndQueueMu.Unlock() @@ -1029,7 +1029,7 @@ func (e *endpoint) purgeWriteQueue() { } // Abort implements stack.TransportEndpoint.Abort. -func (e *endpoint) Abort() { +func (e *Endpoint) Abort() { defer e.drainClosingSegmentQueue() e.LockUser() defer e.UnlockUser() @@ -1047,7 +1047,7 @@ func (e *endpoint) Abort() { // Close puts the endpoint in a closed state and frees all resources associated // with it. It must be called only once and with no other concurrent calls to // the endpoint. -func (e *endpoint) Close() { +func (e *Endpoint) Close() { e.LockUser() if e.closed { e.UnlockUser() @@ -1071,7 +1071,7 @@ func (e *endpoint) Close() { } // +checklocks:e.mu -func (e *endpoint) closeLocked() { +func (e *Endpoint) closeLocked() { linger := e.SocketOptions().GetLinger() if linger.Enabled && linger.Timeout == 0 { s := e.EndpointState() @@ -1092,7 +1092,7 @@ func (e *endpoint) closeLocked() { // closeNoShutdown closes the endpoint without doing a full shutdown. // +checklocks:e.mu -func (e *endpoint) closeNoShutdownLocked() { +func (e *Endpoint) closeNoShutdownLocked() { // For listening sockets, we always release ports inline so that they // are immediately available for reuse after Close() is called. If also // registered, we unregister as well otherwise the next user would fail @@ -1152,15 +1152,15 @@ func (e *endpoint) closeNoShutdownLocked() { // closePendingAcceptableConnections closes all connections that have completed // handshake but not yet been delivered to the application. -func (e *endpoint) closePendingAcceptableConnectionsLocked() { +func (e *Endpoint) closePendingAcceptableConnectionsLocked() { e.acceptMu.Lock() pendingEndpoints := e.acceptQueue.pendingEndpoints e.acceptQueue.pendingEndpoints = nil - completedEndpoints := make([]*endpoint, 0, e.acceptQueue.endpoints.Len()) + completedEndpoints := make([]*Endpoint, 0, e.acceptQueue.endpoints.Len()) for n := e.acceptQueue.endpoints.Front(); n != nil; n = n.Next() { - completedEndpoints = append(completedEndpoints, n.Value.(*endpoint)) + completedEndpoints = append(completedEndpoints, n.Value.(*Endpoint)) } e.acceptQueue.endpoints.Init() e.acceptQueue.capacity = 0 @@ -1179,7 +1179,7 @@ func (e *endpoint) closePendingAcceptableConnectionsLocked() { // cleanupLocked frees all resources associated with the endpoint. // +checklocks:e.mu -func (e *endpoint) cleanupLocked() { +func (e *Endpoint) cleanupLocked() { if e.snd != nil { e.snd.resendTimer.cleanup() e.snd.probeTimer.cleanup() @@ -1245,7 +1245,7 @@ func wndFromSpace(space int) int { // initialReceiveWindow returns the initial receive window to advertise in the // SYN/SYN-ACK. -func (e *endpoint) initialReceiveWindow() int { +func (e *Endpoint) initialReceiveWindow() int { rcvWnd := wndFromSpace(e.receiveBufferAvailable()) if rcvWnd > math.MaxUint16 { rcvWnd = math.MaxUint16 @@ -1274,7 +1274,7 @@ func (e *endpoint) initialReceiveWindow() int { // ModerateRecvBuf adjusts the receive buffer and the advertised window // based on the number of bytes copied to userspace. -func (e *endpoint) ModerateRecvBuf(copied int) { +func (e *Endpoint) ModerateRecvBuf(copied int) { e.LockUser() defer e.UnlockUser() @@ -1352,19 +1352,19 @@ func (e *endpoint) ModerateRecvBuf(copied int) { } // SetOwner implements tcpip.Endpoint.SetOwner. -func (e *endpoint) SetOwner(owner tcpip.PacketOwner) { +func (e *Endpoint) SetOwner(owner tcpip.PacketOwner) { e.owner = owner } // +checklocks:e.mu -func (e *endpoint) hardErrorLocked() tcpip.Error { +func (e *Endpoint) hardErrorLocked() tcpip.Error { err := e.hardError e.hardError = nil return err } // +checklocks:e.mu -func (e *endpoint) lastErrorLocked() tcpip.Error { +func (e *Endpoint) lastErrorLocked() tcpip.Error { e.lastErrorMu.Lock() defer e.lastErrorMu.Unlock() err := e.lastError @@ -1373,7 +1373,7 @@ func (e *endpoint) lastErrorLocked() tcpip.Error { } // LastError implements tcpip.Endpoint.LastError. -func (e *endpoint) LastError() tcpip.Error { +func (e *Endpoint) LastError() tcpip.Error { e.LockUser() defer e.UnlockUser() if err := e.hardErrorLocked(); err != nil { @@ -1385,12 +1385,12 @@ func (e *endpoint) LastError() tcpip.Error { // LastErrorLocked reads and clears lastError. // Only to be used in tests. // +checklocks:e.mu -func (e *endpoint) LastErrorLocked() tcpip.Error { +func (e *Endpoint) LastErrorLocked() tcpip.Error { return e.lastErrorLocked() } // UpdateLastError implements tcpip.SocketOptionsHandler.UpdateLastError. -func (e *endpoint) UpdateLastError(err tcpip.Error) { +func (e *Endpoint) UpdateLastError(err tcpip.Error) { e.LockUser() e.lastErrorMu.Lock() e.lastError = err @@ -1399,7 +1399,7 @@ func (e *endpoint) UpdateLastError(err tcpip.Error) { } // Read implements tcpip.Endpoint.Read. -func (e *endpoint) Read(dst io.Writer, opts tcpip.ReadOptions) (tcpip.ReadResult, tcpip.Error) { +func (e *Endpoint) Read(dst io.Writer, opts tcpip.ReadOptions) (tcpip.ReadResult, tcpip.Error) { e.LockUser() defer e.UnlockUser() @@ -1477,7 +1477,7 @@ func (e *endpoint) Read(dst io.Writer, opts tcpip.ReadOptions) (tcpip.ReadResult // checkRead checks that endpoint is in a readable state. // // +checklocks:e.mu -func (e *endpoint) checkReadLocked() tcpip.Error { +func (e *Endpoint) checkReadLocked() tcpip.Error { e.rcvQueueMu.Lock() defer e.rcvQueueMu.Unlock() // When in SYN-SENT state, let the caller block on the receive. @@ -1520,7 +1520,7 @@ func (e *endpoint) checkReadLocked() tcpip.Error { // indicating the reason why it's not writable. // +checklocks:e.mu // +checklocks:e.sndQueueInfo.sndQueueMu -func (e *endpoint) isEndpointWritableLocked() (int, tcpip.Error) { +func (e *Endpoint) isEndpointWritableLocked() (int, tcpip.Error) { // The endpoint cannot be written to if it's not connected. switch s := e.EndpointState(); { case s == StateError: @@ -1554,7 +1554,7 @@ func (e *endpoint) isEndpointWritableLocked() (int, tcpip.Error) { // readFromPayloader reads a slice from the Payloader. // +checklocks:e.mu // +checklocks:e.sndQueueInfo.sndQueueMu -func (e *endpoint) readFromPayloader(p tcpip.Payloader, opts tcpip.WriteOptions, avail int) (buffer.Buffer, tcpip.Error) { +func (e *Endpoint) readFromPayloader(p tcpip.Payloader, opts tcpip.WriteOptions, avail int) (buffer.Buffer, tcpip.Error) { // We can release locks while copying data. // // This is not possible if atomic is set, because we can't allow the @@ -1585,7 +1585,7 @@ func (e *endpoint) readFromPayloader(p tcpip.Payloader, opts tcpip.WriteOptions, // queueSegment reads data from the payloader and returns a segment to be sent. // +checklocks:e.mu -func (e *endpoint) queueSegment(p tcpip.Payloader, opts tcpip.WriteOptions) (*segment, int, tcpip.Error) { +func (e *Endpoint) queueSegment(p tcpip.Payloader, opts tcpip.WriteOptions) (*segment, int, tcpip.Error) { e.sndQueueInfo.sndQueueMu.Lock() defer e.sndQueueInfo.sndQueueMu.Unlock() @@ -1633,7 +1633,7 @@ func (e *endpoint) queueSegment(p tcpip.Payloader, opts tcpip.WriteOptions) (*se } // Write writes data to the endpoint's peer. -func (e *endpoint) Write(p tcpip.Payloader, opts tcpip.WriteOptions) (int64, tcpip.Error) { +func (e *Endpoint) Write(p tcpip.Payloader, opts tcpip.WriteOptions) (int64, tcpip.Error) { // Linux completely ignores any address passed to sendto(2) for TCP sockets // (without the MSG_FASTOPEN flag). Corking is unimplemented, so opts.More // and opts.EndOfRecord are also ignored. @@ -1656,7 +1656,7 @@ func (e *endpoint) Write(p tcpip.Payloader, opts tcpip.WriteOptions) (int64, tcp // applied. // +checklocks:e.mu // +checklocks:e.rcvQueueMu -func (e *endpoint) selectWindowLocked(rcvBufSize int) (wnd seqnum.Size) { +func (e *Endpoint) selectWindowLocked(rcvBufSize int) (wnd seqnum.Size) { wndFromAvailable := wndFromSpace(e.receiveBufferAvailableLocked(rcvBufSize)) maxWindow := wndFromSpace(rcvBufSize) wndFromUsedBytes := maxWindow - e.RcvBufUsed @@ -1679,7 +1679,7 @@ func (e *endpoint) selectWindowLocked(rcvBufSize int) (wnd seqnum.Size) { // selectWindow invokes selectWindowLocked after acquiring e.rcvQueueMu. // +checklocks:e.mu -func (e *endpoint) selectWindow() (wnd seqnum.Size) { +func (e *Endpoint) selectWindow() (wnd seqnum.Size) { e.rcvQueueMu.Lock() wnd = e.selectWindowLocked(int(e.ops.GetReceiveBufferSize())) e.rcvQueueMu.Unlock() @@ -1702,7 +1702,7 @@ func (e *endpoint) selectWindow() (wnd seqnum.Size) { // // +checklocks:e.mu // +checklocks:e.rcvQueueMu -func (e *endpoint) windowCrossedACKThresholdLocked(deltaBefore int, rcvBufSize int) (crossed bool, above bool) { +func (e *Endpoint) windowCrossedACKThresholdLocked(deltaBefore int, rcvBufSize int) (crossed bool, above bool) { newAvail := int(e.selectWindowLocked(rcvBufSize)) oldAvail := newAvail - deltaBefore if oldAvail < 0 { @@ -1726,28 +1726,28 @@ func (e *endpoint) windowCrossedACKThresholdLocked(deltaBefore int, rcvBufSize i } // OnReuseAddressSet implements tcpip.SocketOptionsHandler.OnReuseAddressSet. -func (e *endpoint) OnReuseAddressSet(v bool) { +func (e *Endpoint) OnReuseAddressSet(v bool) { e.LockUser() e.portFlags.TupleOnly = v e.UnlockUser() } // OnReusePortSet implements tcpip.SocketOptionsHandler.OnReusePortSet. -func (e *endpoint) OnReusePortSet(v bool) { +func (e *Endpoint) OnReusePortSet(v bool) { e.LockUser() e.portFlags.LoadBalanced = v e.UnlockUser() } // OnKeepAliveSet implements tcpip.SocketOptionsHandler.OnKeepAliveSet. -func (e *endpoint) OnKeepAliveSet(bool) { +func (e *Endpoint) OnKeepAliveSet(bool) { e.LockUser() e.resetKeepaliveTimer(true /* receivedData */) e.UnlockUser() } // OnDelayOptionSet implements tcpip.SocketOptionsHandler.OnDelayOptionSet. -func (e *endpoint) OnDelayOptionSet(v bool) { +func (e *Endpoint) OnDelayOptionSet(v bool) { if !v { e.LockUser() defer e.UnlockUser() @@ -1759,7 +1759,7 @@ func (e *endpoint) OnDelayOptionSet(v bool) { } // OnCorkOptionSet implements tcpip.SocketOptionsHandler.OnCorkOptionSet. -func (e *endpoint) OnCorkOptionSet(v bool) { +func (e *Endpoint) OnCorkOptionSet(v bool) { if !v { e.LockUser() defer e.UnlockUser() @@ -1773,12 +1773,12 @@ func (e *endpoint) OnCorkOptionSet(v bool) { } } -func (e *endpoint) getSendBufferSize() int { +func (e *Endpoint) getSendBufferSize() int { return int(e.ops.GetSendBufferSize()) } // OnSetReceiveBufferSize implements tcpip.SocketOptionsHandler.OnSetReceiveBufferSize. -func (e *endpoint) OnSetReceiveBufferSize(rcvBufSz, oldSz int64) (newSz int64, postSet func()) { +func (e *Endpoint) OnSetReceiveBufferSize(rcvBufSz, oldSz int64) (newSz int64, postSet func()) { e.LockUser() sendNonZeroWindowUpdate := false @@ -1820,13 +1820,13 @@ func (e *endpoint) OnSetReceiveBufferSize(rcvBufSz, oldSz int64) (newSz int64, p } // OnSetSendBufferSize implements tcpip.SocketOptionsHandler.OnSetSendBufferSize. -func (e *endpoint) OnSetSendBufferSize(sz int64) int64 { +func (e *Endpoint) OnSetSendBufferSize(sz int64) int64 { e.sndQueueInfo.TCPSndBufState.AutoTuneSndBufDisabled.Store(1) return sz } // WakeupWriters implements tcpip.SocketOptionsHandler.WakeupWriters. -func (e *endpoint) WakeupWriters() { +func (e *Endpoint) WakeupWriters() { e.LockUser() defer e.UnlockUser() @@ -1841,7 +1841,7 @@ func (e *endpoint) WakeupWriters() { } // SetSockOptInt sets a socket option. -func (e *endpoint) SetSockOptInt(opt tcpip.SockOptInt, v int) tcpip.Error { +func (e *Endpoint) SetSockOptInt(opt tcpip.SockOptInt, v int) tcpip.Error { // Lower 2 bits represents ECN bits. RFC 3168, section 23.1 const inetECNMask = 3 @@ -1928,12 +1928,13 @@ func (e *endpoint) SetSockOptInt(opt tcpip.SockOptInt, v int) tcpip.Error { return nil } -func (e *endpoint) HasNIC(id int32) bool { +// HasNIC returns true if the NICID is defined in the stack or id is 0. +func (e *Endpoint) HasNIC(id int32) bool { return id == 0 || e.stack.HasNIC(tcpip.NICID(id)) } // SetSockOpt sets a socket option. -func (e *endpoint) SetSockOpt(opt tcpip.SettableSocketOption) tcpip.Error { +func (e *Endpoint) SetSockOpt(opt tcpip.SettableSocketOption) tcpip.Error { switch v := opt.(type) { case *tcpip.KeepaliveIdleOption: e.LockUser() @@ -2026,7 +2027,7 @@ func (e *endpoint) SetSockOpt(opt tcpip.SettableSocketOption) tcpip.Error { } // readyReceiveSize returns the number of bytes ready to be received. -func (e *endpoint) readyReceiveSize() (int, tcpip.Error) { +func (e *Endpoint) readyReceiveSize() (int, tcpip.Error) { e.LockUser() defer e.UnlockUser() @@ -2042,7 +2043,7 @@ func (e *endpoint) readyReceiveSize() (int, tcpip.Error) { } // GetSockOptInt implements tcpip.Endpoint.GetSockOptInt. -func (e *endpoint) GetSockOptInt(opt tcpip.SockOptInt) (int, tcpip.Error) { +func (e *Endpoint) GetSockOptInt(opt tcpip.SockOptInt) (int, tcpip.Error) { switch opt { case tcpip.KeepaliveCountOption: e.keepalive.Lock() @@ -2115,7 +2116,7 @@ func (e *endpoint) GetSockOptInt(opt tcpip.SockOptInt) (int, tcpip.Error) { } } -func (e *endpoint) getTCPInfo() tcpip.TCPInfoOption { +func (e *Endpoint) getTCPInfo() tcpip.TCPInfoOption { info := tcpip.TCPInfoOption{} e.LockUser() if state := e.EndpointState(); state.internal() { @@ -2144,7 +2145,7 @@ func (e *endpoint) getTCPInfo() tcpip.TCPInfoOption { } // GetSockOpt implements tcpip.Endpoint.GetSockOpt. -func (e *endpoint) GetSockOpt(opt tcpip.GettableSocketOption) tcpip.Error { +func (e *Endpoint) GetSockOpt(opt tcpip.GettableSocketOption) tcpip.Error { switch o := opt.(type) { case *tcpip.TCPInfoOption: *o = e.getTCPInfo() @@ -2201,7 +2202,7 @@ func (e *endpoint) GetSockOpt(opt tcpip.GettableSocketOption) tcpip.Error { // checkV4MappedLocked determines the effective network protocol and converts // addr to its canonical form. // +checklocks:e.mu -func (e *endpoint) checkV4MappedLocked(addr tcpip.FullAddress) (tcpip.FullAddress, tcpip.NetworkProtocolNumber, tcpip.Error) { +func (e *Endpoint) checkV4MappedLocked(addr tcpip.FullAddress) (tcpip.FullAddress, tcpip.NetworkProtocolNumber, tcpip.Error) { unwrapped, netProto, err := e.TransportEndpointInfo.AddrNetProtoLocked(addr, e.ops.GetV6Only()) if err != nil { return tcpip.FullAddress{}, 0, err @@ -2210,12 +2211,12 @@ func (e *endpoint) checkV4MappedLocked(addr tcpip.FullAddress) (tcpip.FullAddres } // Disconnect implements tcpip.Endpoint.Disconnect. -func (*endpoint) Disconnect() tcpip.Error { +func (*Endpoint) Disconnect() tcpip.Error { return &tcpip.ErrNotSupported{} } // Connect connects the endpoint to its peer. -func (e *endpoint) Connect(addr tcpip.FullAddress) tcpip.Error { +func (e *Endpoint) Connect(addr tcpip.FullAddress) tcpip.Error { e.LockUser() defer e.UnlockUser() err := e.connect(addr, true) @@ -2233,7 +2234,7 @@ func (e *endpoint) Connect(addr tcpip.FullAddress) tcpip.Error { // registerEndpoint registers the endpoint with the provided address. // // +checklocks:e.mu -func (e *endpoint) registerEndpoint(addr tcpip.FullAddress, netProto tcpip.NetworkProtocolNumber, nicID tcpip.NICID) tcpip.Error { +func (e *Endpoint) registerEndpoint(addr tcpip.FullAddress, netProto tcpip.NetworkProtocolNumber, nicID tcpip.NICID) tcpip.Error { netProtos := []tcpip.NetworkProtocolNumber{netProto} if e.TransportEndpointInfo.ID.LocalPort != 0 { // The endpoint is bound to a port, attempt to register it. @@ -2298,7 +2299,7 @@ func (e *endpoint) registerEndpoint(addr tcpip.FullAddress, netProto tcpip.Netwo return false, nil } - tcpEP := transEP.(*endpoint) + tcpEP := transEP.(*Endpoint) tcpEP.LockUser() // If the endpoint is not in TIME-WAIT or if it is in TIME-WAIT but // less than 1 second has elapsed since its recentTS was updated then @@ -2366,7 +2367,7 @@ func (e *endpoint) registerEndpoint(addr tcpip.FullAddress, netProto tcpip.Netwo // connect connects the endpoint to its peer. // +checklocks:e.mu -func (e *endpoint) connect(addr tcpip.FullAddress, handshake bool) tcpip.Error { +func (e *Endpoint) connect(addr tcpip.FullAddress, handshake bool) tcpip.Error { connectingAddr := addr.Addr addr, netProto, err := e.checkV4MappedLocked(addr) @@ -2479,13 +2480,13 @@ func (e *endpoint) connect(addr tcpip.FullAddress, handshake bool) tcpip.Error { } // ConnectEndpoint is not supported. -func (*endpoint) ConnectEndpoint(tcpip.Endpoint) tcpip.Error { +func (*Endpoint) ConnectEndpoint(tcpip.Endpoint) tcpip.Error { return &tcpip.ErrInvalidEndpointState{} } // Shutdown closes the read and/or write end of the endpoint connection to its // peer. -func (e *endpoint) Shutdown(flags tcpip.ShutdownFlags) tcpip.Error { +func (e *Endpoint) Shutdown(flags tcpip.ShutdownFlags) tcpip.Error { e.LockUser() defer e.UnlockUser() @@ -2503,7 +2504,7 @@ func (e *endpoint) Shutdown(flags tcpip.ShutdownFlags) tcpip.Error { } // +checklocks:e.mu -func (e *endpoint) shutdownLocked(flags tcpip.ShutdownFlags) tcpip.Error { +func (e *Endpoint) shutdownLocked(flags tcpip.ShutdownFlags) tcpip.Error { e.shutdownFlags |= flags switch { case e.EndpointState().connected(): @@ -2585,7 +2586,7 @@ func (e *endpoint) shutdownLocked(flags tcpip.ShutdownFlags) tcpip.Error { // Listen puts the endpoint in "listen" mode, which allows it to accept // new connections. -func (e *endpoint) Listen(backlog int) tcpip.Error { +func (e *Endpoint) Listen(backlog int) tcpip.Error { if err := e.listen(backlog); err != nil { if !err.IgnoreStats() { e.stack.Stats().TCP.FailedConnectionAttempts.Increment() @@ -2596,7 +2597,7 @@ func (e *endpoint) Listen(backlog int) tcpip.Error { return nil } -func (e *endpoint) listen(backlog int) tcpip.Error { +func (e *Endpoint) listen(backlog int) tcpip.Error { e.LockUser() defer e.UnlockUser() @@ -2612,7 +2613,7 @@ func (e *endpoint) listen(backlog int) tcpip.Error { e.acceptQueue.capacity = backlog if e.acceptQueue.pendingEndpoints == nil { - e.acceptQueue.pendingEndpoints = make(map[*endpoint]struct{}) + e.acceptQueue.pendingEndpoints = make(map[*Endpoint]struct{}) } e.shutdownFlags = 0 @@ -2657,7 +2658,7 @@ func (e *endpoint) listen(backlog int) tcpip.Error { // endpoints. e.acceptMu.Lock() if e.acceptQueue.pendingEndpoints == nil { - e.acceptQueue.pendingEndpoints = make(map[*endpoint]struct{}) + e.acceptQueue.pendingEndpoints = make(map[*Endpoint]struct{}) } if e.acceptQueue.capacity == 0 { e.acceptQueue.capacity = backlog @@ -2675,7 +2676,7 @@ func (e *endpoint) listen(backlog int) tcpip.Error { // to an endpoint previously set to listen mode. // // addr if not-nil will contain the peer address of the returned endpoint. -func (e *endpoint) Accept(peerAddr *tcpip.FullAddress) (tcpip.Endpoint, *waiter.Queue, tcpip.Error) { +func (e *Endpoint) Accept(peerAddr *tcpip.FullAddress) (tcpip.Endpoint, *waiter.Queue, tcpip.Error) { e.LockUser() defer e.UnlockUser() @@ -2688,10 +2689,10 @@ func (e *endpoint) Accept(peerAddr *tcpip.FullAddress) (tcpip.Endpoint, *waiter. } // Get the new accepted endpoint. - var n *endpoint + var n *Endpoint e.acceptMu.Lock() if element := e.acceptQueue.endpoints.Front(); element != nil { - n = e.acceptQueue.endpoints.Remove(element).(*endpoint) + n = e.acceptQueue.endpoints.Remove(element).(*Endpoint) } e.acceptMu.Unlock() if n == nil { @@ -2704,7 +2705,7 @@ func (e *endpoint) Accept(peerAddr *tcpip.FullAddress) (tcpip.Endpoint, *waiter. } // Bind binds the endpoint to a specific local port and optionally address. -func (e *endpoint) Bind(addr tcpip.FullAddress) (err tcpip.Error) { +func (e *Endpoint) Bind(addr tcpip.FullAddress) (err tcpip.Error) { e.LockUser() defer e.UnlockUser() @@ -2712,7 +2713,7 @@ func (e *endpoint) Bind(addr tcpip.FullAddress) (err tcpip.Error) { } // +checklocks:e.mu -func (e *endpoint) bindLocked(addr tcpip.FullAddress) (err tcpip.Error) { +func (e *Endpoint) bindLocked(addr tcpip.FullAddress) (err tcpip.Error) { // Don't allow binding once endpoint is not in the initial state // anymore. This is because once the endpoint goes into a connected or // listen state, it is already bound. @@ -2796,7 +2797,7 @@ func (e *endpoint) bindLocked(addr tcpip.FullAddress) (err tcpip.Error) { } // GetLocalAddress returns the address to which the endpoint is bound. -func (e *endpoint) GetLocalAddress() (tcpip.FullAddress, tcpip.Error) { +func (e *Endpoint) GetLocalAddress() (tcpip.FullAddress, tcpip.Error) { e.LockUser() defer e.UnlockUser() @@ -2808,7 +2809,7 @@ func (e *endpoint) GetLocalAddress() (tcpip.FullAddress, tcpip.Error) { } // GetRemoteAddress returns the address to which the endpoint is connected. -func (e *endpoint) GetRemoteAddress() (tcpip.FullAddress, tcpip.Error) { +func (e *Endpoint) GetRemoteAddress() (tcpip.FullAddress, tcpip.Error) { e.LockUser() defer e.UnlockUser() @@ -2819,7 +2820,7 @@ func (e *endpoint) GetRemoteAddress() (tcpip.FullAddress, tcpip.Error) { return e.getRemoteAddress(), nil } -func (e *endpoint) getRemoteAddress() tcpip.FullAddress { +func (e *Endpoint) getRemoteAddress() tcpip.FullAddress { return tcpip.FullAddress{ Addr: e.TransportEndpointInfo.ID.RemoteAddress, Port: e.TransportEndpointInfo.ID.RemotePort, @@ -2827,14 +2828,15 @@ func (e *endpoint) getRemoteAddress() tcpip.FullAddress { } } -func (*endpoint) HandlePacket(stack.TransportEndpointID, *stack.PacketBuffer) { +// HandlePacket implements stack.TransportEndpoint.HandlePacket. +func (*Endpoint) HandlePacket(stack.TransportEndpointID, *stack.PacketBuffer) { // TCP HandlePacket is not required anymore as inbound packets first // land at the Dispatcher which then can either deliver using the // worker go routine or directly do the invoke the tcp processing inline // based on the state of the endpoint. } -func (e *endpoint) enqueueSegment(s *segment) bool { +func (e *Endpoint) enqueueSegment(s *segment) bool { // Send packet to worker goroutine. if !e.segmentQueue.enqueue(s) { // The queue is full, so we drop the segment. @@ -2845,7 +2847,7 @@ func (e *endpoint) enqueueSegment(s *segment) bool { return true } -func (e *endpoint) onICMPError(err tcpip.Error, transErr stack.TransportError, pkt *stack.PacketBuffer) { +func (e *Endpoint) onICMPError(err tcpip.Error, transErr stack.TransportError, pkt *stack.PacketBuffer) { // Update last error first. e.lastErrorMu.Lock() e.lastError = err @@ -2902,7 +2904,7 @@ func (e *endpoint) onICMPError(err tcpip.Error, transErr stack.TransportError, p } // HandleError implements stack.TransportEndpoint. -func (e *endpoint) HandleError(transErr stack.TransportError, pkt *stack.PacketBuffer) { +func (e *Endpoint) HandleError(transErr stack.TransportError, pkt *stack.PacketBuffer) { handlePacketTooBig := func(mtu uint32) { e.sndQueueInfo.sndQueueMu.Lock() update := false @@ -2944,7 +2946,7 @@ func (e *endpoint) HandleError(transErr stack.TransportError, pkt *stack.PacketB // updateSndBufferUsage is called by the protocol goroutine when room opens up // in the send buffer. The number of newly available bytes is v. -func (e *endpoint) updateSndBufferUsage(v int) { +func (e *Endpoint) updateSndBufferUsage(v int) { sendBufferSize := e.getSendBufferSize() e.sndQueueInfo.sndQueueMu.Lock() notify := e.sndQueueInfo.SndBufUsed >= sendBufferSize>>1 @@ -2972,7 +2974,7 @@ func (e *endpoint) updateSndBufferUsage(v int) { // s will be nil). // // +checklocks:e.mu -func (e *endpoint) readyToRead(s *segment) { +func (e *Endpoint) readyToRead(s *segment) { e.rcvQueueMu.Lock() if s != nil { e.RcvBufUsed += s.payloadSize() @@ -2988,7 +2990,7 @@ func (e *endpoint) readyToRead(s *segment) { // receiveBufferAvailableLocked calculates how many bytes are still available // in the receive buffer. // +checklocks:e.rcvQueueMu -func (e *endpoint) receiveBufferAvailableLocked(rcvBufSize int) int { +func (e *Endpoint) receiveBufferAvailableLocked(rcvBufSize int) int { // We may use more bytes than the buffer size when the receive buffer // shrinks. memUsed := e.receiveMemUsed() @@ -3002,7 +3004,7 @@ func (e *endpoint) receiveBufferAvailableLocked(rcvBufSize int) int { // receiveBufferAvailable calculates how many bytes are still available in the // receive buffer based on the actual memory used by all segments held in // receive buffer/pending and segment queue. -func (e *endpoint) receiveBufferAvailable() int { +func (e *Endpoint) receiveBufferAvailable() int { e.rcvQueueMu.Lock() available := e.receiveBufferAvailableLocked(int(e.ops.GetReceiveBufferSize())) e.rcvQueueMu.Unlock() @@ -3010,7 +3012,7 @@ func (e *endpoint) receiveBufferAvailable() int { } // receiveBufferUsed returns the amount of in-use receive buffer. -func (e *endpoint) receiveBufferUsed() int { +func (e *Endpoint) receiveBufferUsed() int { e.rcvQueueMu.Lock() used := e.RcvBufUsed e.rcvQueueMu.Unlock() @@ -3019,18 +3021,18 @@ func (e *endpoint) receiveBufferUsed() int { // receiveMemUsed returns the total memory in use by segments held by this // endpoint. -func (e *endpoint) receiveMemUsed() int { +func (e *Endpoint) receiveMemUsed() int { return int(e.rcvMemUsed.Load()) } // updateReceiveMemUsed adds the provided delta to e.rcvMemUsed. -func (e *endpoint) updateReceiveMemUsed(delta int) { +func (e *Endpoint) updateReceiveMemUsed(delta int) { e.rcvMemUsed.Add(int32(delta)) } // maxReceiveBufferSize returns the stack wide maximum receive buffer size for // an endpoint. -func (e *endpoint) maxReceiveBufferSize() int { +func (e *Endpoint) maxReceiveBufferSize() int { var rs tcpip.TCPReceiveBufferSizeRangeOption if err := e.stack.TransportProtocolOption(ProtocolNumber, &rs); err != nil { // As a fallback return the hardcoded max buffer size. @@ -3040,12 +3042,12 @@ func (e *endpoint) maxReceiveBufferSize() int { } // directionState returns the close state of send and receive part of the endpoint -func (e *endpoint) connDirectionState() connDirectionState { +func (e *Endpoint) connDirectionState() connDirectionState { return connDirectionState(e.connectionDirectionState.Load()) } // updateDirectionState updates the close state of send and receive part of the endpoint -func (e *endpoint) updateConnDirectionState(state connDirectionState) connDirectionState { +func (e *Endpoint) updateConnDirectionState(state connDirectionState) connDirectionState { return connDirectionState(e.connectionDirectionState.Swap(uint32(e.connDirectionState() | state))) } @@ -3054,7 +3056,7 @@ func (e *endpoint) updateConnDirectionState(state connDirectionState) connDirect // disabled then the window scaling factor is based on the size of the // receiveBuffer otherwise we use the max permissible receive buffer size to // compute the scale. -func (e *endpoint) rcvWndScaleForHandshake() int { +func (e *Endpoint) rcvWndScaleForHandshake() int { bufSizeForScale := e.ops.GetReceiveBufferSize() e.rcvQueueMu.Lock() @@ -3069,7 +3071,7 @@ func (e *endpoint) rcvWndScaleForHandshake() int { // updateRecentTimestamp updates the recent timestamp using the algorithm // described in https://tools.ietf.org/html/rfc7323#section-4.3 -func (e *endpoint) updateRecentTimestamp(tsVal uint32, maxSentAck seqnum.Value, segSeq seqnum.Value) { +func (e *Endpoint) updateRecentTimestamp(tsVal uint32, maxSentAck seqnum.Value, segSeq seqnum.Value) { if e.SendTSOk && seqnum.Value(e.recentTimestamp()).LessThan(seqnum.Value(tsVal)) && segSeq.LessThanEq(maxSentAck) { e.setRecentTimestamp(tsVal) } @@ -3078,29 +3080,29 @@ func (e *endpoint) updateRecentTimestamp(tsVal uint32, maxSentAck seqnum.Value, // maybeEnableTimestamp marks the timestamp option enabled for this endpoint if // the SYN options indicate that timestamp option was negotiated. It also // initializes the recentTS with the value provided in synOpts.TSval. -func (e *endpoint) maybeEnableTimestamp(synOpts header.TCPSynOptions) { +func (e *Endpoint) maybeEnableTimestamp(synOpts header.TCPSynOptions) { if synOpts.TS { e.SendTSOk = true e.setRecentTimestamp(synOpts.TSVal) } } -func (e *endpoint) tsVal(now tcpip.MonotonicTime) uint32 { +func (e *Endpoint) tsVal(now tcpip.MonotonicTime) uint32 { return e.TSOffset.TSVal(now) } -func (e *endpoint) tsValNow() uint32 { +func (e *Endpoint) tsValNow() uint32 { return e.tsVal(e.stack.Clock().NowMonotonic()) } -func (e *endpoint) elapsed(now tcpip.MonotonicTime, tsEcr uint32) time.Duration { +func (e *Endpoint) elapsed(now tcpip.MonotonicTime, tsEcr uint32) time.Duration { return e.TSOffset.Elapsed(now, tsEcr) } // maybeEnableSACKPermitted marks the SACKPermitted option enabled for this endpoint // if the SYN options indicate that the SACK option was negotiated and the TCP // stack is configured to enable TCP SACK option. -func (e *endpoint) maybeEnableSACKPermitted(synOpts header.TCPSynOptions) { +func (e *Endpoint) maybeEnableSACKPermitted(synOpts header.TCPSynOptions) { var v tcpip.TCPSACKEnabled if err := e.stack.TransportProtocolOption(ProtocolNumber, &v); err != nil { // Stack doesn't support SACK. So just return. @@ -3113,7 +3115,7 @@ func (e *endpoint) maybeEnableSACKPermitted(synOpts header.TCPSynOptions) { } // maxOptionSize return the maximum size of TCP options. -func (e *endpoint) maxOptionSize() (size int) { +func (e *Endpoint) maxOptionSize() (size int) { var maxSackBlocks [header.TCPMaxSACKBlocks]header.SACKBlock options := e.makeOptions(maxSackBlocks[:]) size = len(options) @@ -3126,7 +3128,7 @@ func (e *endpoint) maxOptionSize() (size int) { // used before invoking the probe. // // +checklocks:e.mu -func (e *endpoint) completeStateLocked(s *stack.TCPEndpointState) { +func (e *Endpoint) completeStateLocked(s *stack.TCPEndpointState) { s.TCPEndpointStateInner = e.TCPEndpointStateInner s.ID = stack.TCPEndpointID(e.TransportEndpointInfo.ID) s.SegTime = e.stack.Clock().NowMonotonic() @@ -3164,7 +3166,7 @@ func (e *endpoint) completeStateLocked(s *stack.TCPEndpointState) { s.Sender.SpuriousRecovery = e.snd.spuriousRecovery } -func (e *endpoint) initHostGSO() { +func (e *Endpoint) initHostGSO() { switch e.route.NetProto() { case header.IPv4ProtocolNumber: e.gso.Type = stack.GSOTCPv4 @@ -3180,7 +3182,7 @@ func (e *endpoint) initHostGSO() { e.gso.MaxSize = e.route.GSOMaxSize() } -func (e *endpoint) initGSO() { +func (e *Endpoint) initGSO() { if e.route.HasHostGSOCapability() { e.initHostGSO() } else if e.route.HasGvisorGSOCapability() { @@ -3194,12 +3196,12 @@ func (e *endpoint) initGSO() { // State implements tcpip.Endpoint.State. It exports the endpoint's protocol // state for diagnostics. -func (e *endpoint) State() uint32 { +func (e *Endpoint) State() uint32 { return uint32(e.EndpointState()) } // Info returns a copy of the endpoint info. -func (e *endpoint) Info() tcpip.EndpointInfo { +func (e *Endpoint) Info() tcpip.EndpointInfo { e.LockUser() // Make a copy of the endpoint info. ret := e.TransportEndpointInfo @@ -3208,12 +3210,12 @@ func (e *endpoint) Info() tcpip.EndpointInfo { } // Stats returns a pointer to the endpoint stats. -func (e *endpoint) Stats() tcpip.EndpointStats { +func (e *Endpoint) Stats() tcpip.EndpointStats { return &e.stats } // Wait implements stack.TransportEndpoint.Wait. -func (e *endpoint) Wait() { +func (e *Endpoint) Wait() { waitEntry, notifyCh := waiter.NewChannelEntry(waiter.EventHUp) e.waiterQueue.EventRegister(&waitEntry) defer e.waiterQueue.EventUnregister(&waitEntry) @@ -3225,7 +3227,7 @@ func (e *endpoint) Wait() { } // SocketOptions implements tcpip.Endpoint.SocketOptions. -func (e *endpoint) SocketOptions() *tcpip.SocketOptions { +func (e *Endpoint) SocketOptions() *tcpip.SocketOptions { return &e.ops } @@ -3242,7 +3244,7 @@ func GetTCPSendBufferLimits(sh tcpip.StackHandler) tcpip.SendBufferSizeOption { } // allowOutOfWindowAck returns true if an out-of-window ACK can be sent now. -func (e *endpoint) allowOutOfWindowAck() bool { +func (e *Endpoint) allowOutOfWindowAck() bool { now := e.stack.Clock().NowMonotonic() if e.lastOutOfWindowAckTime != (tcpip.MonotonicTime{}) { @@ -3275,7 +3277,7 @@ func GetTCPReceiveBufferLimits(s tcpip.StackHandler) tcpip.ReceiveBufferSizeOpti // computeTCPSendBufferSize implements auto tuning of send buffer size and // returns the new send buffer size. -func (e *endpoint) computeTCPSendBufferSize() int64 { +func (e *Endpoint) computeTCPSendBufferSize() int64 { curSndBufSz := int64(e.getSendBufferSize()) // Auto tuning is disabled when the user explicitly sets the send @@ -3306,6 +3308,7 @@ func (e *endpoint) computeTCPSendBufferSize() int64 { return newSndBufSz } -func (e *endpoint) GetAcceptConn() bool { +// GetAcceptConn implements tcpip.SocketOptionsHandler. +func (e *Endpoint) GetAcceptConn() bool { return EndpointState(e.State()) == StateListen } diff --git a/pkg/tcpip/transport/tcp/endpoint_state.go b/pkg/tcpip/transport/tcp/endpoint_state.go index bb0bc8c3e..5a6d2f6a4 100644 --- a/pkg/tcpip/transport/tcp/endpoint_state.go +++ b/pkg/tcpip/transport/tcp/endpoint_state.go @@ -27,7 +27,7 @@ import ( ) // beforeSave is invoked by stateify. -func (e *endpoint) beforeSave() { +func (e *Endpoint) beforeSave() { // Stop incoming packets. e.segmentQueue.freeze() @@ -62,23 +62,23 @@ func (e *endpoint) beforeSave() { } // saveEndpoints is invoked by stateify. -func (a *acceptQueue) saveEndpoints() []*endpoint { - acceptedEndpoints := make([]*endpoint, a.endpoints.Len()) +func (a *acceptQueue) saveEndpoints() []*Endpoint { + acceptedEndpoints := make([]*Endpoint, a.endpoints.Len()) for i, e := 0, a.endpoints.Front(); e != nil; i, e = i+1, e.Next() { - acceptedEndpoints[i] = e.Value.(*endpoint) + acceptedEndpoints[i] = e.Value.(*Endpoint) } return acceptedEndpoints } // loadEndpoints is invoked by stateify. -func (a *acceptQueue) loadEndpoints(_ context.Context, acceptedEndpoints []*endpoint) { +func (a *acceptQueue) loadEndpoints(_ context.Context, acceptedEndpoints []*Endpoint) { for _, ep := range acceptedEndpoints { a.endpoints.PushBack(ep) } } // saveState is invoked by stateify. -func (e *endpoint) saveState() EndpointState { +func (e *Endpoint) saveState() EndpointState { return e.EndpointState() } @@ -92,7 +92,7 @@ var connectingLoading sync.WaitGroup // Bound endpoint loading happens last. // loadState is invoked by stateify. -func (e *endpoint) loadState(_ context.Context, epState EndpointState) { +func (e *Endpoint) loadState(_ context.Context, epState EndpointState) { // This is to ensure that the loading wait groups include all applicable // endpoints before any asynchronous calls to the Wait() methods. // For restore purposes we treat TimeWait like a connected endpoint. @@ -112,7 +112,7 @@ func (e *endpoint) loadState(_ context.Context, epState EndpointState) { } // afterLoad is invoked by stateify. -func (e *endpoint) afterLoad(ctx context.Context) { +func (e *Endpoint) afterLoad(ctx context.Context) { // RacyLoad() can be used because we are initializing e. e.origEndpointState = e.state.RacyLoad() // Restore the endpoint to InitialState as it will be moved to @@ -122,7 +122,7 @@ func (e *endpoint) afterLoad(ctx context.Context) { } // Restore implements tcpip.RestoredEndpoint.Restore. -func (e *endpoint) Restore(s *stack.Stack) { +func (e *Endpoint) Restore(s *stack.Stack) { if !e.EndpointState().closed() { e.keepalive.timer.init(s.Clock(), timerHandler(e, e.keepaliveTimerExpired)) } @@ -280,6 +280,6 @@ func (e *endpoint) Restore(s *stack.Stack) { } // Resume implements tcpip.ResumableEndpoint.Resume. -func (e *endpoint) Resume() { +func (e *Endpoint) Resume() { e.segmentQueue.thaw() } diff --git a/pkg/tcpip/transport/tcp/rcv.go b/pkg/tcpip/transport/tcp/rcv.go index 765f3ab76..349f950f1 100644 --- a/pkg/tcpip/transport/tcp/rcv.go +++ b/pkg/tcpip/transport/tcp/rcv.go @@ -30,7 +30,7 @@ import ( // +stateify savable type receiver struct { stack.TCPReceiverState - ep *endpoint + ep *Endpoint // rcvWnd is the non-scaled receive window last advertised to the peer. rcvWnd seqnum.Size @@ -52,7 +52,7 @@ type receiver struct { lastRcvdAckTime tcpip.MonotonicTime } -func newReceiver(ep *endpoint, irs seqnum.Value, rcvWnd seqnum.Size, rcvWndScale uint8) *receiver { +func newReceiver(ep *Endpoint, irs seqnum.Value, rcvWnd seqnum.Size, rcvWndScale uint8) *receiver { return &receiver{ ep: ep, TCPReceiverState: stack.TCPReceiverState{ diff --git a/pkg/tcpip/transport/tcp/segment.go b/pkg/tcpip/transport/tcp/segment.go index cf18a6f8e..6de583daf 100644 --- a/pkg/tcpip/transport/tcp/segment.go +++ b/pkg/tcpip/transport/tcp/segment.go @@ -55,7 +55,7 @@ type segment struct { segmentEntry segmentRefs - ep *endpoint + ep *Endpoint qFlags queueFlags id stack.TransportEndpointID `state:"manual"` @@ -182,7 +182,7 @@ func (s *segment) merge(oth *segment) { // setOwner sets the owning endpoint for this segment. Its required // to be called to ensure memory accounting for receive/send buffer // queues is done properly. -func (s *segment) setOwner(ep *endpoint, qFlags queueFlags) { +func (s *segment) setOwner(ep *Endpoint, qFlags queueFlags) { switch qFlags { case recvQ: ep.updateReceiveMemUsed(s.segMemSize()) diff --git a/pkg/tcpip/transport/tcp/segment_queue.go b/pkg/tcpip/transport/tcp/segment_queue.go index 53839387e..6f003efc0 100644 --- a/pkg/tcpip/transport/tcp/segment_queue.go +++ b/pkg/tcpip/transport/tcp/segment_queue.go @@ -24,7 +24,7 @@ import ( type segmentQueue struct { mu sync.Mutex `state:"nosave"` list segmentList `state:"wait"` - ep *endpoint + ep *Endpoint frozen bool } diff --git a/pkg/tcpip/transport/tcp/snd.go b/pkg/tcpip/transport/tcp/snd.go index f9f61a309..41b5b534e 100644 --- a/pkg/tcpip/transport/tcp/snd.go +++ b/pkg/tcpip/transport/tcp/snd.go @@ -88,7 +88,7 @@ type lossRecovery interface { // +stateify savable type sender struct { stack.TCPSenderState - ep *endpoint + ep *Endpoint // lr is the loss recovery algorithm used by the sender. lr lossRecovery @@ -171,7 +171,7 @@ type rtt struct { } // +checklocks:ep.mu -func newSender(ep *endpoint, iss, irs seqnum.Value, sndWnd seqnum.Size, mss uint16, sndWndScale int) *sender { +func newSender(ep *Endpoint, iss, irs seqnum.Value, sndWnd seqnum.Size, mss uint16, sndWndScale int) *sender { // The sender MUST reduce the TCP data length to account for any IP or // TCP options that it is including in the packets that it sends. // See: https://tools.ietf.org/html/rfc6691#section-2