tcpHandler looks rounded off, add to example

This commit is contained in:
soypat
2025-02-10 20:26:02 -03:00
parent 64a7102064
commit 0cb43c47a4
10 changed files with 184 additions and 137 deletions
+1 -1
View File
@@ -122,7 +122,7 @@ const (
IPProtoGGP IPProto = 3 // Gateway-to-Gateway [RFC823] IPProtoGGP IPProto = 3 // Gateway-to-Gateway [RFC823]
IPProtoIPv4 IPProto = 4 // IPv4 encapsulation [RFC2003] IPProtoIPv4 IPProto = 4 // IPv4 encapsulation [RFC2003]
IPProtoST IPProto = 5 // Stream [RFC1190, RFC1819] IPProtoST IPProto = 5 // Stream [RFC1190, RFC1819]
IPProtoTCP IPProto = 6 // Transmission Control [RFC793] IPProtoTCP IPProto = 6 // Transmission Control [RFC9293]
IPProtoCBT IPProto = 7 // CBT [Ballardie] IPProtoCBT IPProto = 7 // CBT [Ballardie]
IPProtoEGP IPProto = 8 // Exterior Gateway Protocol [RFC888] IPProtoEGP IPProto = 8 // Exterior Gateway Protocol [RFC888]
IPProtoIGP IPProto = 9 // any private interior gateway (used by Cisco for their IGRP) IPProtoIGP IPProto = 9 // any private interior gateway (used by Cisco for their IGRP)
+19 -61
View File
@@ -14,6 +14,7 @@ import (
) )
func main() { func main() {
const mtu = 1500
rng := rand.New(rand.NewSource(1)) rng := rand.New(rand.NewSource(1))
var gen ltesto.PacketGen var gen ltesto.PacketGen
gen.RandomizeAddrs(rng) gen.RandomizeAddrs(rng)
@@ -21,7 +22,7 @@ func main() {
lStack := LinkStack{ lStack := LinkStack{
logger: slogger, logger: slogger,
mac: gen.DstMAC, mac: gen.DstMAC,
mtu: 1500, mtu: mtu,
} }
iStack := &IPv4Stack{ iStack := &IPv4Stack{
ip: gen.DstIPv4, ip: gen.DstIPv4,
@@ -31,12 +32,20 @@ func main() {
logger: slogger, logger: slogger,
} }
pStack := &TCPPort{ pStack := &TCPPort{
lport: gen.DstTCP, handler: tcp.Handler{},
rport: gen.SrcTCP,
tcb: tcp.ControlBlock{},
} }
iss := tcp.Value(100)
err := iStack.Register(tStack, &gen.SrcIPv4) 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 { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
@@ -44,12 +53,7 @@ func main() {
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
err = tStack.Register(pStack, pStack.lport) err = tStack.Register(pStack, pStack.handler.LocalPort())
if err != nil {
log.Fatal(err)
}
iss := tcp.Value(100)
err = pStack.tcb.Open(iss, 256, tcp.StateListen)
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
@@ -364,10 +368,7 @@ func (ts *TCPStack) Handle(ipFrame []byte, tcpOff int) (n int, err error) {
} }
type TCPPort struct { type TCPPort struct {
tcb tcp.ControlBlock handler tcp.Handler
validator lneto.Validator
lport uint16
rport uint16
} }
func (tp *TCPPort) Protocol() uint32 { return uint32(lneto.IPProtoTCP) } 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 { if off != 0 {
return errors.New("TCP API expected 0 offset") return errors.New("TCP API expected 0 offset")
} }
tfrm, err := lneto.NewTCPFrame(tcpFrame) return tp.handler.Recv(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
} }
func (tp *TCPPort) Handle(tcpFrame []byte, off int) (n int, err error) { func (tp *TCPPort) Handle(tcpFrame []byte, off int) (n int, err error) {
if off != 0 { if off != 0 {
return 0, errors.New("TCP API expected 0 offset") return 0, errors.New("TCP API expected 0 offset")
} else if tp.tcb.State().IsClosed() {
return 0, io.EOF
} }
tfrm, err := lneto.NewTCPFrame(tcpFrame) return tp.handler.Send(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
} }
type logger struct { type logger struct {
+23 -5
View File
@@ -5,11 +5,26 @@ import (
"fmt" "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 { type Validator struct {
checkEvil bool accum []error
allowMultiErrs bool accumBitpos []BitPosErr
accum []error flags ValidateFlags
accumBitpos []BitPosErr }
func (v *Validator) Flags() ValidateFlags {
return v.flags
} }
func (v *Validator) ResetErr() { func (v *Validator) ResetErr() {
@@ -18,6 +33,9 @@ func (v *Validator) ResetErr() {
} }
func (v *Validator) HasError() bool { func (v *Validator) HasError() bool {
if v.flags.has(validateReserved) {
panic("reserved bit set")
}
return len(v.accum) != 0 return len(v.accum) != 0
} }
@@ -37,7 +55,7 @@ func (v *Validator) gotErr(err error) {
func (v *Validator) AddError(err error) { func (v *Validator) AddError(err error) {
if err == nil { if err == nil {
panic("error argument to AddError cannot be 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 return
} }
v.accum = append(v.accum, err) v.accum = append(v.accum, err)
+32 -32
View File
@@ -51,13 +51,13 @@ type ControlBlock struct {
// On a call to Send the queue is advanced and flags set in the segment are unset. // 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. // The second position of the queue is used for FIN segments.
pending [2]Flags pending [2]Flags
state State _state State // leading underscore so field not suggested on top of exported State method when developing.
challengeAck bool challengeAck bool
logger logger
} }
// State returns the current state of the TCP connection. // 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. // RecvNext returns the next sequence number expected to be received from remote.
// This implementation will reject segments that are not the next expected sequence. // 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 // 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. // the send window size and the unacked data. Returns 0 before StateSynRcvd.
func (tcb *ControlBlock) MaxInFlightData() Size { func (tcb *ControlBlock) MaxInFlightData() Size {
if !tcb.state.hasIRS() { if !tcb._state.hasIRS() {
return 0 // SYN not yet received. return 0 // SYN not yet received.
} }
unacked := Sizeof(tcb.snd.UNA, tcb.snd.NXT) unacked := Sizeof(tcb.snd.UNA, tcb.snd.NXT)
@@ -142,7 +142,7 @@ type recvSpace struct {
// state must be StateListen or StateSynSent. // state must be StateListen or StateSynSent.
func (tcb *ControlBlock) Open(iss Value, wnd Size, state State) (err error) { func (tcb *ControlBlock) Open(iss Value, wnd Size, state State) (err error) {
switch { switch {
case tcb.state != StateClosed && tcb.state != StateListen: case tcb._state != StateClosed && tcb._state != StateListen:
err = errTCBNotClosed err = errTCBNotClosed
case state != StateListen && state != StateSynSent: case state != StateListen && state != StateSynSent:
err = errInvalidState 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())) tcb.logerr("tcb:open", slog.String("err", err.Error()))
return err return err
} }
tcb.state = state tcb._state = state
tcb.resetRcv(wnd, 0) tcb.resetRcv(wnd, 0)
tcb.resetSnd(iss, 1) tcb.resetSnd(iss, 1)
tcb.pending = [2]Flags{} tcb.pending = [2]Flags{}
if state == StateSynSent { if state == StateSynSent {
tcb.pending[0] = FlagSYN 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 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 return Segment{SEQ: tcb.snd.NXT, ACK: tcb.rcv.NXT, Flags: FlagACK, WND: tcb.rcv.WND}, true
} }
pending := tcb.pending[0] pending := tcb.pending[0]
established := tcb.state == StateEstablished established := tcb._state == StateEstablished
if !established && tcb.state != StateCloseWait { if !established && tcb._state != StateCloseWait {
payloadLen = 0 // Can't send data if not established. payloadLen = 0 // Can't send data if not established.
} }
if pending == 0 && payloadLen == 0 { if pending == 0 && payloadLen == 0 {
@@ -238,7 +238,7 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) {
prevNxt := tcb.snd.NXT prevNxt := tcb.snd.NXT
var pending Flags var pending Flags
switch tcb.state { switch tcb._state {
case StateListen: case StateListen:
pending, err = tcb.rcvListen(seg) pending, err = tcb.rcvListen(seg)
case StateSynSent: case StateSynSent:
@@ -259,10 +259,10 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) {
case StateClosing: case StateClosing:
// Thanks to @knieriem for finding and reporting this bug. // Thanks to @knieriem for finding and reporting this bug.
if seg.Flags.HasAny(FlagACK) { if seg.Flags.HasAny(FlagACK) {
tcb.state = StateTimeWait tcb._state = StateTimeWait
} }
default: default:
panic("unexpected recv state:" + tcb.state.String()) panic("unexpected recv state:" + tcb._state.String())
} }
if err != nil { if err != nil {
return err return err
@@ -270,7 +270,7 @@ func (tcb *ControlBlock) Recv(seg Segment) (err error) {
tcb.pending[0] |= pending tcb.pending[0] |= pending
if prevNxt != 0 && tcb.snd.NXT != prevNxt && tcb.logenabled(slog.LevelDebug) { 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("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))) 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) hasFIN := seg.Flags.HasAny(FlagFIN)
hasACK := seg.Flags.HasAny(FlagACK) hasACK := seg.Flags.HasAny(FlagACK)
var newPending Flags var newPending Flags
switch tcb.state { switch tcb._state {
case StateSynRcvd: case StateSynRcvd:
if hasFIN { if hasFIN {
tcb.state = StateFinWait1 // RFC 9293: 3.10.4 CLOSE call. tcb._state = StateFinWait1 // RFC 9293: 3.10.4 CLOSE call.
} }
case StateClosing: case StateClosing:
if hasACK { if hasACK {
tcb.state = StateTimeWait tcb._state = StateTimeWait
} }
case StateEstablished: case StateEstablished:
if hasFIN { if hasFIN {
tcb.state = StateFinWait1 tcb._state = StateFinWait1
} }
case StateCloseWait: case StateCloseWait:
if hasFIN { if hasFIN {
tcb.state = StateLastAck tcb._state = StateLastAck
} else if hasACK { } else if hasACK {
newPending = finack // Queue finack. 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) && outOfWindow := checkSeq && !seg.SEQ.InWindow(tcb.snd.NXT, tcb.snd.WND) &&
!zeroWindowOK !zeroWindowOK
switch { switch {
case tcb.state == StateClosed: case tcb._state == StateClosed:
err = io.ErrClosedPipe err = io.ErrClosedPipe
case seg.WND > math.MaxUint16: case seg.WND > math.MaxUint16:
err = errWindowTooLarge err = errWindowTooLarge
@@ -369,7 +369,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) {
err = errSeqNotInWindow 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. 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: 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) hasAck := flags.HasAll(FlagACK)
// Short circuit SEQ checks if SYN present since the incoming segment initialize1s connection. // Short circuit SEQ checks if SYN present since the incoming segment initialize1s connection.
checkSEQ := !flags.HasAny(FlagSYN) checkSEQ := !flags.HasAny(FlagSYN)
established := tcb.state == StateEstablished established := tcb._state == StateEstablished
preestablished := tcb.state.IsPreestablished() preestablished := tcb._state.IsPreestablished()
acksOld := hasAck && !tcb.snd.UNA.LessThan(seg.ACK) acksOld := hasAck && !tcb.snd.UNA.LessThan(seg.ACK)
acksUnsentData := hasAck && !seg.ACK.LessThanEq(tcb.snd.NXT) acksUnsentData := hasAck && !seg.ACK.LessThanEq(tcb.snd.NXT)
ctlOrDataSegment := established && (seg.DATALEN > 0 || flags.HasAny(FlagFIN|FlagRST)) ctlOrDataSegment := established && (seg.DATALEN > 0 || flags.HasAny(FlagFIN|FlagRST))
@@ -396,7 +396,7 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) {
switch { switch {
case seg.WND > math.MaxUint16: case seg.WND > math.MaxUint16:
err = errWindowOverflow err = errWindowOverflow
case tcb.state == StateClosed: case tcb._state == StateClosed:
err = io.ErrClosedPipe err = io.ErrClosedPipe
case checkSEQ && tcb.rcv.WND == 0 && seg.DATALEN > 0 && seg.SEQ == tcb.rcv.NXT: 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 err = errDropSegment
tcb.pending[0] &= FlagFIN // Completely ignore duplicate ACKs but do not erase fin bit. tcb.pending[0] &= FlagFIN // Completely ignore duplicate ACKs but do not erase fin bit.
if isDebug { 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))) 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 err = errDropSegment
tcb.pending[0] = FlagACK // Send ACK for unsent data. tcb.pending[0] = FlagACK // Send ACK for unsent data.
if isDebug { 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))) 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.rstPtr = seg.ACK
tcb.resetSnd(tcb.snd.ISS, seg.WND) tcb.resetSnd(tcb.snd.ISS, seg.WND)
if isDebug { 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 return err
@@ -472,16 +472,16 @@ func (tcb *ControlBlock) resetRcv(localWND Size, remoteISS Value) {
} }
func (tcb *ControlBlock) handleRST(seq Value) error { 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 { 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). // 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.challengeAck = true
tcb.pending[0] |= FlagACK tcb.pending[0] |= FlagACK
return errDropSegment return errDropSegment
} }
if tcb.state.IsPreestablished() { if tcb._state.IsPreestablished() {
tcb.pending[0] = 0 tcb.pending[0] = 0
tcb.state = StateListen tcb._state = StateListen
tcb.resetSnd(tcb.snd.ISS+tcb.rstJump(), tcb.snd.WND) tcb.resetSnd(tcb.snd.ISS+tcb.rstJump(), tcb.snd.WND)
tcb.resetRcv(tcb.rcv.WND, 3_14159_2653^tcb.rcv.IRS) tcb.resetRcv(tcb.rcv.WND, 3_14159_2653^tcb.rcv.IRS)
} else { } else {
@@ -497,7 +497,7 @@ func (tcb *ControlBlock) rstJump() Value {
// close sets ControlBlock state to closed and resets all sequence numbers and pending flag. // close sets ControlBlock state to closed and resets all sequence numbers and pending flag.
func (tcb *ControlBlock) close() { func (tcb *ControlBlock) close() {
tcb.state = StateClosed tcb._state = StateClosed
tcb.pending = [2]Flags{} tcb.pending = [2]Flags{}
tcb.resetRcv(0, 0) tcb.resetRcv(0, 0)
tcb.resetSnd(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. // Close returns an error if the connection is already closed or closing.
func (tcb *ControlBlock) Close() (err error) { func (tcb *ControlBlock) Close() (err error) {
// See RFC 9293: 3.10.4 CLOSE call. // See RFC 9293: 3.10.4 CLOSE call.
switch tcb.state { switch tcb._state {
case StateClosed: case StateClosed:
err = errConnNotexist err = errConnNotexist
case StateCloseWait: case StateCloseWait:
tcb.state = StateLastAck tcb._state = StateLastAck
tcb.pending = [2]Flags{FlagFIN, FlagACK} tcb.pending = [2]Flags{FlagFIN, FlagACK}
case StateListen, StateSynSent: case StateListen, StateSynSent:
tcb.close() tcb.close()
@@ -528,7 +528,7 @@ func (tcb *ControlBlock) Close() (err error) {
err = errInvalidState err = errInvalidState
} }
if err == nil { if err == nil {
tcb.trace("tcb:close", slog.String("state", tcb.state.String())) tcb.trace("tcb:close", slog.String("state", tcb._state.String()))
} else { } else {
tcb.logerr("tcb:close", slog.String("err", err.Error())) tcb.logerr("tcb:close", slog.String("err", err.Error()))
} }
+9 -9
View File
@@ -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). // We must respond with SYN|ACK frame after receiving SYN in listen state (three way handshake).
tcb.pending[0] = synack tcb.pending[0] = synack
tcb.state = StateSynRcvd tcb._state = StateSynRcvd
return synack, nil return synack, nil
} }
@@ -33,13 +33,13 @@ func (tcb *ControlBlock) rcvSynSent(seg Segment) (pending Flags, err error) {
} }
if hasAck { if hasAck {
tcb.state = StateEstablished tcb._state = StateEstablished
pending = FlagACK pending = FlagACK
tcb.resetRcv(tcb.rcv.WND, seg.SEQ) tcb.resetRcv(tcb.rcv.WND, seg.SEQ)
} else { } else {
// Simultaneous connection sync edge case. // Simultaneous connection sync edge case.
pending = synack pending = synack
tcb.state = StateSynRcvd tcb._state = StateSynRcvd
tcb.resetSnd(tcb.snd.ISS, seg.WND) tcb.resetSnd(tcb.snd.ISS, seg.WND)
tcb.resetRcv(tcb.rcv.WND, seg.SEQ) tcb.resetRcv(tcb.rcv.WND, seg.SEQ)
} }
@@ -56,7 +56,7 @@ func (tcb *ControlBlock) rcvSynRcvd(seg Segment) (pending Flags, err error) {
if err != nil { if err != nil {
return 0, err return 0, err
} }
tcb.state = StateEstablished tcb._state = StateEstablished
return 0, nil return 0, nil
} }
@@ -69,7 +69,7 @@ func (tcb *ControlBlock) rcvEstablished(seg Segment) (pending Flags, err error)
pending = FlagACK pending = FlagACK
if hasFin { if hasFin {
// See Figure 5: TCP Connection State Diagram of RFC 9293. // 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. 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: case hasFin && hasAck && seg.ACK == tcb.snd.NXT:
// Special case: Server sent a FINACK response to our FIN so we enter TimeWait directly. // 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. // We have to check ACK against send NXT to avoid simultaneous close sequence edge case.
tcb.state = StateTimeWait tcb._state = StateTimeWait
case hasFin: case hasFin:
tcb.state = StateClosing tcb._state = StateClosing
case hasAck: case hasAck:
// TODO(soypat): Check if this branch does NOT need ACK queued. Online flowcharts say not needed. // TODO(soypat): Check if this branch does NOT need ACK queued. Online flowcharts say not needed.
tcb.state = StateFinWait2 tcb._state = StateFinWait2
default: default:
return 0, errFinwaitExpectedACK return 0, errFinwaitExpectedACK
} }
@@ -102,6 +102,6 @@ func (tcb *ControlBlock) rcvFinWait2(seg Segment) (pending Flags, err error) {
if !seg.Flags.HasAll(finack) { if !seg.Flags.HasAll(finack) {
return pending, errFinwaitExpectedFinack return pending, errFinwaitExpectedFinack
} }
tcb.state = StateTimeWait tcb._state = StateTimeWait
return FlagACK, nil return FlagACK, nil
} }
+2 -2
View File
@@ -33,7 +33,7 @@ func (l logger) logenabled(lvl slog.Level) bool {
func (tcb *ControlBlock) traceSnd(msg string) { func (tcb *ControlBlock) traceSnd(msg string) {
tcb.trace(msg, tcb.trace(msg,
slog.String("state", tcb.state.String()), slog.String("state", tcb._state.String()),
slog.Uint64("pend", uint64(tcb.pending[0])), slog.Uint64("pend", uint64(tcb.pending[0])),
slog.Uint64("snd.nxt", uint64(tcb.snd.NXT)), slog.Uint64("snd.nxt", uint64(tcb.snd.NXT)),
slog.Uint64("snd.una", uint64(tcb.snd.UNA)), slog.Uint64("snd.una", uint64(tcb.snd.UNA)),
@@ -43,7 +43,7 @@ func (tcb *ControlBlock) traceSnd(msg string) {
func (tcb *ControlBlock) traceRcv(msg string) { func (tcb *ControlBlock) traceRcv(msg string) {
tcb.trace(msg, 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.nxt", uint64(tcb.rcv.NXT)),
slog.Uint64("rcv.wnd", uint64(tcb.rcv.WND)), slog.Uint64("rcv.wnd", uint64(tcb.rcv.WND)),
slog.Bool("challenge", tcb.challengeAck), slog.Bool("challenge", tcb.challengeAck),
+14
View File
@@ -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. // 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. // Be sure to call [Frame.ValidateSize] beforehand to avoid panic.
func (tfrm Frame) Options() []byte { func (tfrm Frame) Options() []byte {
+80 -16
View File
@@ -33,14 +33,43 @@ type Handler struct {
logger logger
} }
func (h *Handler) Reset() error { func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error {
*h = Handler{ if !h.scb.State().IsClosed() {
connid: h.connid + 1, return errors.New("tcp.Handler must be closed before setting buffers")
bufTx: h.bufTx, }
bufRx: h.bufRx, if rxbuf != nil {
h.bufRx.Buf = rxbuf
}
if len(h.bufRx.Buf) < 1 {
return errors.New("short rx buffer")
} }
h.bufRx.Reset() 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 return nil
} }
@@ -100,26 +129,54 @@ func (h *Handler) Recv(b []byte) error {
return nil 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))) h.trace("tcp.Handler:start", slog.Uint64("port", uint64(h.localPort)))
if h.isClosed() { if h.isClosed() {
return 0, net.ErrClosed return 0, net.ErrClosed
} else if h.AwaitingSyn() {
return h.sendInitSyn(b)
} }
tfrm, err := NewFrame(b) tfrm, err := NewFrame(b)
if err != nil { if err != nil {
return 0, err 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.SetSourcePort(h.localPort)
tfrm.SetDestinationPort(h.remotePort) tfrm.SetDestinationPort(h.remotePort)
tfrm.SetSegment(segment, 5) // No TCP options.
return 0, nil tfrm.SetUrgentPtr(0)
} return sizeHeaderTCP + int(segment.DATALEN), nil
func (h *Handler) sendInitSyn(b []byte) (int, error) {
return 0, nil
} }
// AwaitingSyn checks if the Handler is waiting for a Syn to arrive. // 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 { func (h *Handler) isClosed() bool {
return h.closing || h.scb.State().IsClosed() return h.closing || h.scb.State().IsClosed()
} }
func min(a, b int) int {
if a < b {
return a
}
return b
}
+4 -4
View File
@@ -26,12 +26,12 @@ func (tcb *ControlBlock) HelperExchange(t *testing.T, exchange []Exchange) {
if t.Failed() { if t.Failed() {
t.Errorf("exchange failed:\nwant: %s\ngot: %s", t.Errorf("exchange failed:\nwant: %s\ngot: %s",
ex.RFC9293String(ex.WantState, ex.WantPeerState), ex.RFC9293String(ex.WantState, ex.WantPeerState),
ex.RFC9293String(tcb.state, ex.WantPeerState), ex.RFC9293String(tcb._state, ex.WantPeerState),
) )
} }
}() }()
const pfx = "exchange" const pfx = "exchange"
t.Log(tcb.state, "Exchange start") t.Log(tcb._state, "Exchange start")
for i, ex = range exchange { for i, ex = range exchange {
if ex.Outgoing != nil && ex.Incoming != nil { 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) 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() state := tcb.State()
if state != ex.WantState { 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) { func (tcb *ControlBlock) HelperInitState(state State, localISS, localNXT Value, localWindow Size) {
tcb.state = state tcb._state = state
tcb.snd = sendSpace{ tcb.snd = sendSpace{
ISS: localISS, ISS: localISS,
UNA: localISS, UNA: localISS,
-7
View File
@@ -322,10 +322,3 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, arg
} }
testQueueSanity(t, rtx) testQueueSanity(t, rtx)
} }
func min(a, b int) int {
if a < b {
return a
}
return b
}