From 0cb43c47a4cd6577da90ab6b6713355da4189705 Mon Sep 17 00:00:00 2001 From: soypat Date: Mon, 10 Feb 2025 20:26:02 -0300 Subject: [PATCH] tcpHandler looks rounded off, add to example --- definitions.go | 2 +- examples/stack/main.go | 80 ++++++++----------------------- lneto2/validation.go | 28 +++++++++-- tcp/control.go | 64 ++++++++++++------------- tcp/control_rcvhandlers.go | 18 +++---- tcp/debug.go | 4 +- tcp/frame.go | 14 ++++++ tcp/handler.go | 96 +++++++++++++++++++++++++++++++------- tcp/internal_test.go | 8 ++-- tcp/txqueue_test.go | 7 --- 10 files changed, 184 insertions(+), 137 deletions(-) diff --git a/definitions.go b/definitions.go index 4d542a8..44ef685 100644 --- a/definitions.go +++ b/definitions.go @@ -122,7 +122,7 @@ const ( IPProtoGGP IPProto = 3 // Gateway-to-Gateway [RFC823] IPProtoIPv4 IPProto = 4 // IPv4 encapsulation [RFC2003] IPProtoST IPProto = 5 // Stream [RFC1190, RFC1819] - IPProtoTCP IPProto = 6 // Transmission Control [RFC793] + IPProtoTCP IPProto = 6 // Transmission Control [RFC9293] IPProtoCBT IPProto = 7 // CBT [Ballardie] IPProtoEGP IPProto = 8 // Exterior Gateway Protocol [RFC888] IPProtoIGP IPProto = 9 // any private interior gateway (used by Cisco for their IGRP) diff --git a/examples/stack/main.go b/examples/stack/main.go index 63b2a49..9c3e774 100644 --- a/examples/stack/main.go +++ b/examples/stack/main.go @@ -14,6 +14,7 @@ import ( ) func main() { + const mtu = 1500 rng := rand.New(rand.NewSource(1)) var gen ltesto.PacketGen gen.RandomizeAddrs(rng) @@ -21,7 +22,7 @@ func main() { lStack := LinkStack{ logger: slogger, mac: gen.DstMAC, - mtu: 1500, + mtu: mtu, } iStack := &IPv4Stack{ ip: gen.DstIPv4, @@ -31,12 +32,20 @@ func main() { logger: slogger, } pStack := &TCPPort{ - lport: gen.DstTCP, - rport: gen.SrcTCP, - tcb: tcp.ControlBlock{}, + handler: tcp.Handler{}, } - - err := iStack.Register(tStack, &gen.SrcIPv4) + iss := tcp.Value(100) + txbuf := make([]byte, mtu) + rxbuf := make([]byte, mtu) + err := pStack.handler.SetBuffers(txbuf, rxbuf, 3) + if err != nil { + log.Fatal(err) + } + err = pStack.handler.Open(tcp.StateListen, gen.DstTCP, gen.SrcTCP, iss) + if err != nil { + log.Fatal(err) + } + err = iStack.Register(tStack, &gen.SrcIPv4) if err != nil { log.Fatal(err) } @@ -44,12 +53,7 @@ func main() { if err != nil { log.Fatal(err) } - err = tStack.Register(pStack, pStack.lport) - if err != nil { - log.Fatal(err) - } - iss := tcp.Value(100) - err = pStack.tcb.Open(iss, 256, tcp.StateListen) + err = tStack.Register(pStack, pStack.handler.LocalPort()) if err != nil { log.Fatal(err) } @@ -364,10 +368,7 @@ func (ts *TCPStack) Handle(ipFrame []byte, tcpOff int) (n int, err error) { } type TCPPort struct { - tcb tcp.ControlBlock - validator lneto.Validator - lport uint16 - rport uint16 + handler tcp.Handler } func (tp *TCPPort) Protocol() uint32 { return uint32(lneto.IPProtoTCP) } @@ -376,57 +377,14 @@ func (tp *TCPPort) Recv(tcpFrame []byte, off int) error { if off != 0 { return errors.New("TCP API expected 0 offset") } - tfrm, err := lneto.NewTCPFrame(tcpFrame) - if err != nil { - return err - } - tp.validator.ResetErr() - tfrm.ValidateExceptCRC(&tp.validator) - if err = tp.validator.Err(); err != nil { - return err - } - if tfrm.DestinationPort() != tp.lport { - return errors.New("port mismatch") - } - seg := tfrm.Segment(len(tfrm.Payload())) - err = tp.tcb.Recv(seg) - if err != nil { - return err - } - return nil + return tp.handler.Recv(tcpFrame) } func (tp *TCPPort) Handle(tcpFrame []byte, off int) (n int, err error) { if off != 0 { return 0, errors.New("TCP API expected 0 offset") - } else if tp.tcb.State().IsClosed() { - return 0, io.EOF } - tfrm, err := lneto.NewTCPFrame(tcpFrame) - if err != nil { - return 0, err - } - if !tp.tcb.HasPending() { - return 0, nil - } - - seg, ok := tp.tcb.PendingSegment(0) - if !ok { - return 0, nil - } - err = tp.tcb.Send(seg) - if err != nil { - return 0, err - } - tfrm.SetSourcePort(tp.lport) - tfrm.SetDestinationPort(tp.rport) - tfrm.SetOffsetAndFlags(5, seg.Flags) - tfrm.SetSeq(seg.SEQ) - tfrm.SetAck(seg.ACK) - tfrm.SetUrgentPtr(0) - tfrm.SetWindowSize(uint16(seg.WND)) - - return 20, nil + return tp.handler.Send(tcpFrame) } type logger struct { diff --git a/lneto2/validation.go b/lneto2/validation.go index 34baa98..4dfba8f 100644 --- a/lneto2/validation.go +++ b/lneto2/validation.go @@ -5,11 +5,26 @@ import ( "fmt" ) +type ValidateFlags uint64 + +const ( + validateReserved ValidateFlags = 1 << iota + validateEvilBit + validateAllowMultiErrors +) + +func (vf ValidateFlags) has(v ValidateFlags) bool { + return vf&v == v +} + type Validator struct { - checkEvil bool - allowMultiErrs bool - accum []error - accumBitpos []BitPosErr + accum []error + accumBitpos []BitPosErr + flags ValidateFlags +} + +func (v *Validator) Flags() ValidateFlags { + return v.flags } func (v *Validator) ResetErr() { @@ -18,6 +33,9 @@ func (v *Validator) ResetErr() { } func (v *Validator) HasError() bool { + if v.flags.has(validateReserved) { + panic("reserved bit set") + } return len(v.accum) != 0 } @@ -37,7 +55,7 @@ func (v *Validator) gotErr(err error) { func (v *Validator) AddError(err error) { if err == nil { panic("error argument to AddError cannot be nil") - } else if len(v.accum) != 0 && !v.allowMultiErrs { + } else if len(v.accum) != 0 && !v.flags.has(validateAllowMultiErrors) { return } v.accum = append(v.accum, err) diff --git a/tcp/control.go b/tcp/control.go index 9c45b2f..9627257 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -51,13 +51,13 @@ type ControlBlock struct { // On a call to Send the queue is advanced and flags set in the segment are unset. // The second position of the queue is used for FIN segments. pending [2]Flags - state State + _state State // leading underscore so field not suggested on top of exported State method when developing. challengeAck bool logger } // State returns the current state of the TCP connection. -func (tcb *ControlBlock) State() State { return tcb.state } +func (tcb *ControlBlock) State() State { return tcb._state } // RecvNext returns the next sequence number expected to be received from remote. // This implementation will reject segments that are not the next expected sequence. @@ -73,7 +73,7 @@ func (tcb *ControlBlock) ISS() Value { return tcb.snd.ISS } // MaxInFlightData returns the maximum size of a segment that can be sent by taking into account // the send window size and the unacked data. Returns 0 before StateSynRcvd. func (tcb *ControlBlock) MaxInFlightData() Size { - if !tcb.state.hasIRS() { + if !tcb._state.hasIRS() { return 0 // SYN not yet received. } unacked := Sizeof(tcb.snd.UNA, tcb.snd.NXT) @@ -142,7 +142,7 @@ type recvSpace struct { // state must be StateListen or StateSynSent. func (tcb *ControlBlock) Open(iss Value, wnd Size, state State) (err error) { switch { - case tcb.state != StateClosed && tcb.state != StateListen: + case tcb._state != StateClosed && tcb._state != StateListen: err = errTCBNotClosed case state != StateListen && state != StateSynSent: err = errInvalidState @@ -153,14 +153,14 @@ func (tcb *ControlBlock) Open(iss Value, wnd Size, state State) (err error) { tcb.logerr("tcb:open", slog.String("err", err.Error())) return err } - tcb.state = state + tcb._state = state tcb.resetRcv(wnd, 0) tcb.resetSnd(iss, 1) tcb.pending = [2]Flags{} if state == StateSynSent { tcb.pending[0] = FlagSYN } - tcb.trace("tcb:open", slog.String("state", tcb.state.String())) + tcb.trace("tcb:open", slog.String("state", tcb._state.String())) return nil } @@ -175,8 +175,8 @@ func (tcb *ControlBlock) PendingSegment(payloadLen int) (_ Segment, ok bool) { return Segment{SEQ: tcb.snd.NXT, ACK: tcb.rcv.NXT, Flags: FlagACK, WND: tcb.rcv.WND}, true } pending := tcb.pending[0] - established := tcb.state == StateEstablished - if !established && tcb.state != StateCloseWait { + established := tcb._state == StateEstablished + if !established && tcb._state != StateCloseWait { payloadLen = 0 // Can't send data if not established. } if pending == 0 && payloadLen == 0 { @@ -238,7 +238,7 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) { prevNxt := tcb.snd.NXT var pending Flags - switch tcb.state { + switch tcb._state { case StateListen: pending, err = tcb.rcvListen(seg) case StateSynSent: @@ -259,10 +259,10 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) { case StateClosing: // Thanks to @knieriem for finding and reporting this bug. if seg.Flags.HasAny(FlagACK) { - tcb.state = StateTimeWait + tcb._state = StateTimeWait } default: - panic("unexpected recv state:" + tcb.state.String()) + panic("unexpected recv state:" + tcb._state.String()) } if err != nil { return err @@ -270,7 +270,7 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) { tcb.pending[0] |= pending if prevNxt != 0 && tcb.snd.NXT != prevNxt && tcb.logenabled(slog.LevelDebug) { - tcb.debug("tcb:snd.nxt-change", slog.String("state", tcb.state.String()), + tcb.debug("tcb:snd.nxt-change", slog.String("state", tcb._state.String()), slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT)), slog.Uint64("prevnxt", uint64(prevNxt)), slog.Uint64("seg.seq", uint64(seg.SEQ))) } @@ -304,22 +304,22 @@ func (tcb *ControlBlock) Send(seg Segment) error { hasFIN := seg.Flags.HasAny(FlagFIN) hasACK := seg.Flags.HasAny(FlagACK) var newPending Flags - switch tcb.state { + switch tcb._state { case StateSynRcvd: if hasFIN { - tcb.state = StateFinWait1 // RFC 9293: 3.10.4 CLOSE call. + tcb._state = StateFinWait1 // RFC 9293: 3.10.4 CLOSE call. } case StateClosing: if hasACK { - tcb.state = StateTimeWait + tcb._state = StateTimeWait } case StateEstablished: if hasFIN { - tcb.state = StateFinWait1 + tcb._state = StateFinWait1 } case StateCloseWait: if hasFIN { - tcb.state = StateLastAck + tcb._state = StateLastAck } else if hasACK { newPending = finack // Queue finack. } @@ -355,7 +355,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { outOfWindow := checkSeq && !seg.SEQ.InWindow(tcb.snd.NXT, tcb.snd.WND) && !zeroWindowOK switch { - case tcb.state == StateClosed: + case tcb._state == StateClosed: err = io.ErrClosedPipe case seg.WND > math.MaxUint16: err = errWindowTooLarge @@ -369,7 +369,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { err = errSeqNotInWindow } - case seg.DATALEN > 0 && (tcb.state == StateFinWait1 || tcb.state == StateFinWait2): + case seg.DATALEN > 0 && (tcb._state == StateFinWait1 || tcb._state == StateFinWait2): err = errConnectionClosing // Case 1: No further SENDs from the user will be accepted by the TCP implementation. case checkSeq && tcb.snd.WND == 0 && seg.DATALEN > 0 && seg.SEQ == tcb.snd.NXT: @@ -386,8 +386,8 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { hasAck := flags.HasAll(FlagACK) // Short circuit SEQ checks if SYN present since the incoming segment initialize1s connection. checkSEQ := !flags.HasAny(FlagSYN) - established := tcb.state == StateEstablished - preestablished := tcb.state.IsPreestablished() + established := tcb._state == StateEstablished + preestablished := tcb._state.IsPreestablished() acksOld := hasAck && !tcb.snd.UNA.LessThan(seg.ACK) acksUnsentData := hasAck && !seg.ACK.LessThanEq(tcb.snd.NXT) ctlOrDataSegment := established && (seg.DATALEN > 0 || flags.HasAny(FlagFIN|FlagRST)) @@ -396,7 +396,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { switch { case seg.WND > math.MaxUint16: err = errWindowOverflow - case tcb.state == StateClosed: + case tcb._state == StateClosed: err = io.ErrClosedPipe case checkSEQ && tcb.rcv.WND == 0 && seg.DATALEN > 0 && seg.SEQ == tcb.rcv.NXT: @@ -429,7 +429,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { err = errDropSegment tcb.pending[0] &= FlagFIN // Completely ignore duplicate ACKs but do not erase fin bit. if isDebug { - tcb.debug("rcv:ACK-dup", slog.String("state", tcb.state.String()), + tcb.debug("rcv:ACK-dup", slog.String("state", tcb._state.String()), slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.una", uint64(tcb.snd.UNA))) } @@ -437,7 +437,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { err = errDropSegment tcb.pending[0] = FlagACK // Send ACK for unsent data. if isDebug { - tcb.debug("rcv:ACK-unsent", slog.String("state", tcb.state.String()), + tcb.debug("rcv:ACK-unsent", slog.String("state", tcb._state.String()), slog.Uint64("seg.ack", uint64(seg.ACK)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT))) } @@ -447,7 +447,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { tcb.rstPtr = seg.ACK tcb.resetSnd(tcb.snd.ISS, seg.WND) if isDebug { - tcb.debug("rcv:RST-old", slog.String("state", tcb.state.String()), slog.Uint64("ack", uint64(seg.ACK))) + tcb.debug("rcv:RST-old", slog.String("state", tcb._state.String()), slog.Uint64("ack", uint64(seg.ACK))) } } return err @@ -472,16 +472,16 @@ func (tcb *ControlBlock) resetRcv(localWND Size, remoteISS Value) { } func (tcb *ControlBlock) handleRST(seq Value) error { - tcb.debug("rcv:RST", slog.String("state", tcb.state.String())) + tcb.debug("rcv:RST", slog.String("state", tcb._state.String())) if seq != tcb.rcv.NXT { // See RFC9293: If the RST bit is set and the sequence number does not exactly match the next expected sequence value, yet is within the current receive window, TCP endpoints MUST send an acknowledgment (challenge ACK). tcb.challengeAck = true tcb.pending[0] |= FlagACK return errDropSegment } - if tcb.state.IsPreestablished() { + if tcb._state.IsPreestablished() { tcb.pending[0] = 0 - tcb.state = StateListen + tcb._state = StateListen tcb.resetSnd(tcb.snd.ISS+tcb.rstJump(), tcb.snd.WND) tcb.resetRcv(tcb.rcv.WND, 3_14159_2653^tcb.rcv.IRS) } else { @@ -497,7 +497,7 @@ func (tcb *ControlBlock) rstJump() Value { // close sets ControlBlock state to closed and resets all sequence numbers and pending flag. func (tcb *ControlBlock) close() { - tcb.state = StateClosed + tcb._state = StateClosed tcb.pending = [2]Flags{} tcb.resetRcv(0, 0) tcb.resetSnd(0, 0) @@ -510,11 +510,11 @@ func (tcb *ControlBlock) close() { // Close returns an error if the connection is already closed or closing. func (tcb *ControlBlock) Close() (err error) { // See RFC 9293: 3.10.4 CLOSE call. - switch tcb.state { + switch tcb._state { case StateClosed: err = errConnNotexist case StateCloseWait: - tcb.state = StateLastAck + tcb._state = StateLastAck tcb.pending = [2]Flags{FlagFIN, FlagACK} case StateListen, StateSynSent: tcb.close() @@ -528,7 +528,7 @@ func (tcb *ControlBlock) Close() (err error) { err = errInvalidState } if err == nil { - tcb.trace("tcb:close", slog.String("state", tcb.state.String())) + tcb.trace("tcb:close", slog.String("state", tcb._state.String())) } else { tcb.logerr("tcb:close", slog.String("err", err.Error())) } diff --git a/tcp/control_rcvhandlers.go b/tcp/control_rcvhandlers.go index 4d23e07..8bc412a 100644 --- a/tcp/control_rcvhandlers.go +++ b/tcp/control_rcvhandlers.go @@ -14,7 +14,7 @@ func (tcb *ControlBlock) rcvListen(seg Segment) (pending Flags, err error) { // We must respond with SYN|ACK frame after receiving SYN in listen state (three way handshake). tcb.pending[0] = synack - tcb.state = StateSynRcvd + tcb._state = StateSynRcvd return synack, nil } @@ -33,13 +33,13 @@ func (tcb *ControlBlock) rcvSynSent(seg Segment) (pending Flags, err error) { } if hasAck { - tcb.state = StateEstablished + tcb._state = StateEstablished pending = FlagACK tcb.resetRcv(tcb.rcv.WND, seg.SEQ) } else { // Simultaneous connection sync edge case. pending = synack - tcb.state = StateSynRcvd + tcb._state = StateSynRcvd tcb.resetSnd(tcb.snd.ISS, seg.WND) tcb.resetRcv(tcb.rcv.WND, seg.SEQ) } @@ -56,7 +56,7 @@ func (tcb *ControlBlock) rcvSynRcvd(seg Segment) (pending Flags, err error) { if err != nil { return 0, err } - tcb.state = StateEstablished + tcb._state = StateEstablished return 0, nil } @@ -69,7 +69,7 @@ func (tcb *ControlBlock) rcvEstablished(seg Segment) (pending Flags, err error) pending = FlagACK if hasFin { // See Figure 5: TCP Connection State Diagram of RFC 9293. - tcb.state = StateCloseWait + tcb._state = StateCloseWait tcb.pending[1] = FlagFIN // Queue FIN for after the CloseWait ACK. } } @@ -85,12 +85,12 @@ func (tcb *ControlBlock) rcvFinWait1(seg Segment) (pending Flags, err error) { case hasFin && hasAck && seg.ACK == tcb.snd.NXT: // Special case: Server sent a FINACK response to our FIN so we enter TimeWait directly. // We have to check ACK against send NXT to avoid simultaneous close sequence edge case. - tcb.state = StateTimeWait + tcb._state = StateTimeWait case hasFin: - tcb.state = StateClosing + tcb._state = StateClosing case hasAck: // TODO(soypat): Check if this branch does NOT need ACK queued. Online flowcharts say not needed. - tcb.state = StateFinWait2 + tcb._state = StateFinWait2 default: return 0, errFinwaitExpectedACK } @@ -102,6 +102,6 @@ func (tcb *ControlBlock) rcvFinWait2(seg Segment) (pending Flags, err error) { if !seg.Flags.HasAll(finack) { return pending, errFinwaitExpectedFinack } - tcb.state = StateTimeWait + tcb._state = StateTimeWait return FlagACK, nil } diff --git a/tcp/debug.go b/tcp/debug.go index 5bb109b..962b4c0 100644 --- a/tcp/debug.go +++ b/tcp/debug.go @@ -33,7 +33,7 @@ func (l logger) logenabled(lvl slog.Level) bool { func (tcb *ControlBlock) traceSnd(msg string) { tcb.trace(msg, - slog.String("state", tcb.state.String()), + slog.String("state", tcb._state.String()), slog.Uint64("pend", uint64(tcb.pending[0])), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT)), slog.Uint64("snd.una", uint64(tcb.snd.UNA)), @@ -43,7 +43,7 @@ func (tcb *ControlBlock) traceSnd(msg string) { func (tcb *ControlBlock) traceRcv(msg string) { tcb.trace(msg, - slog.String("state", tcb.state.String()), + slog.String("state", tcb._state.String()), slog.Uint64("rcv.nxt", uint64(tcb.rcv.NXT)), slog.Uint64("rcv.wnd", uint64(tcb.rcv.WND)), slog.Bool("challenge", tcb.challengeAck), diff --git a/tcp/frame.go b/tcp/frame.go index 78f3d59..2b6f6fa 100644 --- a/tcp/frame.go +++ b/tcp/frame.go @@ -148,6 +148,20 @@ func (tfrm Frame) Segment(payloadSize int) Segment { } } +// SetSegment sets the sequence, acknowledgment, offset, window and flag fields of the TCP header from the the [Segment]. +// Offset, like in [Frame.SetOffset], is expressed in words with minimum being 5. +func (tfrm Frame) SetSegment(seg Segment, offset uint8) { + if offset >= 1<<4 { + panic("tcp offset too large") + } else if seg.WND > math.MaxUint16 { + panic("tcp window overflow") + } + tfrm.SetSeq(seg.SEQ) + tfrm.SetAck(seg.ACK) + tfrm.SetOffsetAndFlags(offset, seg.Flags) + tfrm.SetWindowSize(uint16(seg.WND)) +} + // Options returns the TCP option buffer portion of the frame. The returned slice may be zero length. // Be sure to call [Frame.ValidateSize] beforehand to avoid panic. func (tfrm Frame) Options() []byte { diff --git a/tcp/handler.go b/tcp/handler.go index 90f363f..0b623d9 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -33,14 +33,43 @@ type Handler struct { logger } -func (h *Handler) Reset() error { - *h = Handler{ - connid: h.connid + 1, - bufTx: h.bufTx, - bufRx: h.bufRx, +func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error { + if !h.scb.State().IsClosed() { + return errors.New("tcp.Handler must be closed before setting buffers") + } + if rxbuf != nil { + h.bufRx.Buf = rxbuf + } + if len(h.bufRx.Buf) < 1 { + return errors.New("short rx buffer") } h.bufRx.Reset() - h.bufTx.ResetOrReuse(nil, 0, 0) + return h.bufTx.ResetOrReuse(txbuf, packets, 0) +} + +func (h *Handler) LocalPort() uint16 { + return h.localPort +} + +func (h *Handler) Open(state State, localPort, remotePort uint16, iss Value) error { + // Open will fail unless SCB in closed state. + err := h.scb.Open(iss, Size(h.bufRx.Size()), state) + if err != nil { + return err + } + *h = Handler{ + scb: h.scb, + bufTx: h.bufTx, + bufRx: h.bufRx, + connid: h.connid + 1, + localPort: localPort, + remotePort: remotePort, + validator: h.validator, + logger: h.logger, + closing: false, + } + h.bufTx.ResetOrReuse(nil, 0, iss) + h.bufRx.Reset() return nil } @@ -100,26 +129,54 @@ func (h *Handler) Recv(b []byte) error { return nil } -func (h *Handler) Handle(b []byte) (int, error) { +func (h *Handler) Send(b []byte) (int, error) { h.trace("tcp.Handler:start", slog.Uint64("port", uint64(h.localPort))) if h.isClosed() { return 0, net.ErrClosed - } else if h.AwaitingSyn() { - return h.sendInitSyn(b) } tfrm, err := NewFrame(b) if err != nil { return 0, err } - + var segment Segment + if h.AwaitingSyn() { + // Handling init syn segment. + segment = Segment{ + SEQ: h.scb.ISS(), + ACK: 0, + Flags: FlagSYN, + WND: h.scb.RecvWindow(), + DATALEN: 0, + } + } else { + var ok bool + available := min(h.bufTx.Buffered(), len(b)-sizeHeaderTCP) + segment, ok = h.scb.PendingSegment(available) + if !ok { + // No pending control segment or data to send. Yield. + return 0, nil + } + n, seq, err := h.bufTx.MakePacket(b[sizeHeaderTCP : sizeHeaderTCP+segment.DATALEN]) + if err != nil { + return 0, err + } else if seq != segment.SEQ { + panic("mismatching sequence numbers") + } else if n != int(segment.DATALEN) { + panic("expected n == available") + } + } + prevState := h.scb.State() + err = h.scb.Send(segment) + if err != nil { + return 0, err + } else if prevState != h.scb.State() && h.logenabled(slog.LevelInfo) { + h.info("tcp.Handler:tx-statechange", slog.Uint64("port", uint64(h.localPort)), slog.String("oldState", prevState.String()), slog.String("newState", h.scb.State().String()), slog.String("txflags", segment.Flags.String())) + } tfrm.SetSourcePort(h.localPort) tfrm.SetDestinationPort(h.remotePort) - - return 0, nil -} - -func (h *Handler) sendInitSyn(b []byte) (int, error) { - return 0, nil + tfrm.SetSegment(segment, 5) // No TCP options. + tfrm.SetUrgentPtr(0) + return sizeHeaderTCP + int(segment.DATALEN), nil } // AwaitingSyn checks if the Handler is waiting for a Syn to arrive. @@ -130,3 +187,10 @@ func (h *Handler) AwaitingSyn() bool { func (h *Handler) isClosed() bool { return h.closing || h.scb.State().IsClosed() } + +func min(a, b int) int { + if a < b { + return a + } + return b +} diff --git a/tcp/internal_test.go b/tcp/internal_test.go index ea998ff..a2cb593 100644 --- a/tcp/internal_test.go +++ b/tcp/internal_test.go @@ -26,12 +26,12 @@ func (tcb *ControlBlock) HelperExchange(t *testing.T, exchange []Exchange) { if t.Failed() { t.Errorf("exchange failed:\nwant: %s\ngot: %s", ex.RFC9293String(ex.WantState, ex.WantPeerState), - ex.RFC9293String(tcb.state, ex.WantPeerState), + ex.RFC9293String(tcb._state, ex.WantPeerState), ) } }() const pfx = "exchange" - t.Log(tcb.state, "Exchange start") + t.Log(tcb._state, "Exchange start") for i, ex = range exchange { if ex.Outgoing != nil && ex.Incoming != nil { t.Fatalf(pfx+"[%d] cannot send and receive in the same exchange, please split into two exchanges.", i) @@ -60,7 +60,7 @@ func (tcb *ControlBlock) HelperExchange(t *testing.T, exchange []Exchange) { } } - t.Log(ex.RFC9293String(tcb.state, ex.WantPeerState)) + t.Log(ex.RFC9293String(tcb._state, ex.WantPeerState)) state := tcb.State() if state != ex.WantState { @@ -78,7 +78,7 @@ func (tcb *ControlBlock) HelperExchange(t *testing.T, exchange []Exchange) { } func (tcb *ControlBlock) HelperInitState(state State, localISS, localNXT Value, localWindow Size) { - tcb.state = state + tcb._state = state tcb.snd = sendSpace{ ISS: localISS, UNA: localISS, diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 3be716b..279352d 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -322,10 +322,3 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, arg } testQueueSanity(t, rtx) } - -func min(a, b int) int { - if a < b { - return a - } - return b -}