Add tcp.Listener (#15)

* begin reviewing TCP listener

* begin adding listener tests

* adding TestExchange implementation

* split out legacy and Step tests

* move listener to tcp package

* fix up tcp tests taking very long

* add xnet.TCPPool and work on TCPPool semantics

* xnet: TCPPool only accepts established connections

* xnet: getting to bottom of panic in xnet.Listener

* xnet: improve listener tests

* fix several data races in testing fixtures

* fix more synchronization things in listener test

* finalize tcplistener test

x/xnet/stack-async.go

@@ -219,6 +219,12 @@ func (s *StackAsync) Addr() netip.Addr {
 	return s.ip.Addr()
 }
 
+func (s *StackAsync) SetSubnet(subnetMask netip.Prefix) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.subnet = subnetMask
+}
+
 func (s *StackAsync) SetHardwareAddress(hw [6]byte) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -286,6 +292,16 @@ func (s *StackAsync) ListenTCP(conn *tcp.Conn, localPort uint16) (err error) {
 	return nil
 }
 
+func (s *StackAsync) RegisterListener(listener *tcp.Listener) (err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	lport := listener.LocalPort()
+	if lport == 0 {
+		return lneto.ErrZeroSource
+	}
+	return s.tcps.Register(listener, nil)
+}
+
 var errNoDNSServer = errors.New("no DNS server- did DHCP complete? You can set a predetermined DNS server in Stack configuration")
 
 func (s *StackAsync) StartLookupIP(host string) error {

x/xnet/tcppool.go

@@ -0,0 +1,152 @@
+package xnet
+
+import (
+	"errors"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/soypat/lneto/tcp"
+)
+
+// TCPPool implements tcp.pool.
+type TCPPool struct {
+	mu             sync.Mutex
+	naqcuired      int
+	conns          []tcp.Conn
+	acquiredAt     []time.Time
+	closingAt      []time.Time
+	abortedAt      []time.Time
+	nextISS        tcp.Value
+	_now           func() time.Time
+	estbTimeout    time.Duration
+	closingTimeout time.Duration
+}
+
+func _() {
+	var l tcp.Listener
+	l.Reset(0, &TCPPool{}) // compile time guarantee of interface implementation.
+}
+
+type TCPPoolConfig struct {
+	PoolSize   int
+	QueueSize  int
+	BufferSize int
+	ConnLogger *slog.Logger
+	Now        func() time.Time
+	// EstablishedTimeout sets the timeout for a TCP connection since it is acquired until it is established.
+	// If the connection does not establish in this time it will be closed by the pool.
+	EstablishedTimeout time.Duration
+	// ClosingTimeout sets the timeout for a TCP connection to close and be returned to Pool.
+	// If the connection is not closed in this time it will be aborted by the pool.
+	ClosingTimeout time.Duration
+}
+
+func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) {
+	if cfg.EstablishedTimeout <= 0 || cfg.ClosingTimeout <= 0 {
+		return nil, errors.New("invalid timeout")
+	}
+	n := cfg.PoolSize
+	bufsize := cfg.BufferSize
+	pool := &TCPPool{
+		acquiredAt:     make([]time.Time, n),
+		closingAt:      make([]time.Time, n),
+		abortedAt:      make([]time.Time, n),
+		conns:          make([]tcp.Conn, n),
+		_now:           cfg.Now,
+		estbTimeout:    cfg.EstablishedTimeout,
+		closingTimeout: cfg.ClosingTimeout,
+	}
+	bufSpace := make([]byte, 2*n*bufsize)
+	for i := range pool.conns {
+		bufoff := 2 * i * bufsize
+		err := pool.conns[i].Configure(tcp.ConnConfig{
+			RxBuf:             bufSpace[bufoff : bufoff+bufsize],
+			TxBuf:             bufSpace[bufoff+bufsize : bufoff+2*bufsize],
+			TxPacketQueueSize: cfg.QueueSize,
+			Logger:            cfg.ConnLogger,
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+	return pool, nil
+}
+
+func (p *TCPPool) GetTCP() (*tcp.Conn, tcp.Value) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	for i := range p.conns {
+		if p.acquiredAt[i].IsZero() {
+			p.acquiredAt[i] = p.now()
+			p.nextISS += 1000
+			p.naqcuired++
+			return &p.conns[i], p.nextISS
+		}
+	}
+	return nil, 0
+}
+
+func (p *TCPPool) PutTCP(conn *tcp.Conn) {
+	for i := range p.conns {
+		if &p.conns[i] == conn {
+			p.mu.Lock()
+			p.conns[i].Abort()
+			p.acquiredAt[i] = time.Time{}
+			p.abortedAt[i] = time.Time{}
+			p.closingAt[i] = time.Time{}
+			p.naqcuired--
+			p.mu.Unlock()
+			return
+		}
+	}
+	panic("conn does not belong to this pool")
+}
+
+func (p *TCPPool) CheckTimeouts() {
+	for i := range p.conns {
+		st := p.conns[i].State()
+		if st == tcp.StateEstablished {
+			continue
+		}
+		p.mu.Lock()
+		acq := p.acquiredAt[i]
+		p.mu.Unlock()
+		if acq.IsZero() {
+			continue
+		} else if st.IsPreestablished() && p.since(acq) > p.estbTimeout {
+			// Was acquired and did not reach establishment state so we close.
+			// This is part of a syn-flood defense mechanism.
+			p.conns[i].Close()
+		} else if st.IsClosed() || st.IsClosing() {
+			p.mu.Lock()
+			if p.closingAt[i].IsZero() {
+				p.closingAt[i] = p.now()
+			} else if p.abortedAt[i].IsZero() && p.since(p.closingAt[i]) > p.closingTimeout {
+				p.abortedAt[i] = p.now()
+				p.conns[i].Abort()
+			} else if p.since(p.abortedAt[i]) > 10*time.Second {
+				println("connection aborted and still not returned to TCPPool")
+			}
+			p.mu.Unlock()
+		}
+	}
+}
+
+func (p *TCPPool) since(t time.Time) time.Duration {
+	if p._now == nil {
+		return time.Since(t)
+	}
+	return p._now().Sub(t)
+}
+
+func (p *TCPPool) now() time.Time {
+	if p._now == nil {
+		return time.Now()
+	}
+	return p._now()
+}
+
+func (p *TCPPool) NumberOfAcquired() int {
+	return p.naqcuired
+}

x/xnet/xnet_listener_test.go

@@ -0,0 +1,225 @@
+package xnet
+
+import (
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/soypat/lneto/tcp"
+)
+
+func TestStackAsyncListener_SingleConnection(t *testing.T) {
+	const seed int64 = 1234
+	const MTU = 1500
+	const svPort = 80
+	const clPort = 1337
+
+	// Create two stacks.
+	client, sv := new(StackAsync), new(StackAsync)
+	err := client.Reset(StackConfig{
+		Hostname:        "Client",
+		RandSeed:        seed,
+		StaticAddress:   netip.AddrFrom4([4]byte{10, 0, 0, 1}),
+		MaxTCPConns:     1,
+		HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 1},
+		MTU:             MTU,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = sv.Reset(StackConfig{
+		Hostname:        "Server",
+		RandSeed:        ^seed,
+		StaticAddress:   netip.AddrFrom4([4]byte{10, 0, 0, 2}),
+		MaxTCPConns:     1, // Note: We use listener, not direct TCP conn registration.
+		HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 2},
+		MTU:             MTU,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	client.SetGateway6(sv.HardwareAddress())
+	sv.SetGateway6(client.HardwareAddress())
+
+	// Create client connection.
+	var clConn tcp.Conn
+	err = clConn.Configure(tcp.ConnConfig{
+		RxBuf:             make([]byte, MTU),
+		TxBuf:             make([]byte, MTU),
+		TxPacketQueueSize: 4,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create pool and listener for server.
+	pool, err := NewTCPPool(TCPPoolConfig{
+		PoolSize:           1,
+		QueueSize:          4,
+		BufferSize:         MTU,
+		EstablishedTimeout: 10e9,
+		ClosingTimeout:     10e9,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var listener tcp.Listener
+	err = listener.Reset(svPort, pool)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = sv.RegisterListener(&listener)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Client dials server.
+	err = client.DialTCP(&clConn, clPort, netip.AddrPortFrom(sv.Addr(), svPort))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tst := testerFrom(t, MTU)
+
+	// Complete TCP handshake.
+	tst.TestTCPHandshake(client, sv)
+
+	// After handshake, TryAccept should work.
+	if listener.NumberOfReadyToAccept() != 1 {
+		t.Fatalf("after handshake: expected 1 ready, got %d", listener.NumberOfReadyToAccept())
+	}
+	svConn, err := listener.TryAccept()
+	if err != nil {
+		t.Fatalf("TryAccept: %v", err)
+	}
+	if listener.NumberOfReadyToAccept() != 0 {
+		t.Fatalf("after accept: expected 0 ready, got %d", listener.NumberOfReadyToAccept())
+	}
+
+	// Verify both connections are established.
+	if clConn.State() != tcp.StateEstablished {
+		t.Fatalf("client: expected StateEstablished, got %s", clConn.State())
+	}
+	if svConn.State() != tcp.StateEstablished {
+		t.Fatalf("server: expected StateEstablished, got %s", svConn.State())
+	}
+
+	// Test data exchange: client -> server.
+	sendData := []byte("hello from client")
+	tst.TestTCPEstablishedSingleData(client, sv, &clConn, svConn, sendData)
+
+	// Test data exchange: server -> client.
+	replyData := []byte("hello from server")
+	tst.TestTCPEstablishedSingleData(sv, client, svConn, &clConn, replyData)
+
+	// Test close (client-initiated).
+	tst.TestTCPClose(client, sv, &clConn, svConn)
+}
+
+func TestStackAsyncListener_MultiSequentialConn(t *testing.T) {
+	const seed int64 = 1234
+	const MTU = 1500
+	const svPort = 80
+	const clPort = 1337
+	const poolsize = 10
+	const bufsize = 128
+	// Create two stacks.
+	sv := new(StackAsync)
+	err := sv.Reset(StackConfig{
+		Hostname:        "Server",
+		RandSeed:        ^seed,
+		StaticAddress:   netip.AddrFrom4([4]byte{10, 0, 0, 2}),
+		MaxTCPConns:     1, // Note: We use listener, not direct TCP conn registration.
+		HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 2},
+		MTU:             MTU,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Create pool and listener for server.
+	pool, err := NewTCPPool(TCPPoolConfig{
+		PoolSize:           poolsize,
+		QueueSize:          4,
+		BufferSize:         bufsize,
+		EstablishedTimeout: 10e9,
+		ClosingTimeout:     10e9,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var listener tcp.Listener
+	err = listener.Reset(svPort, pool)
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = sv.RegisterListener(&listener)
+	if err != nil {
+		t.Fatal(err)
+	}
+	caddr := netip.AddrFrom4([4]byte{10, 0, 0, 1})
+	chw := [6]byte{0xbe, 0xef, 0, 0, 0, 1}
+	sv.SetGateway6(chw)
+	tst := testerFrom(t, MTU)
+	doRequest := func(caddrp netip.AddrPort, sleep time.Duration, data []byte) {
+		var client StackAsync
+		err := client.Reset(StackConfig{
+			Hostname:        "Client",
+			RandSeed:        seed,
+			StaticAddress:   caddrp.Addr(),
+			MaxTCPConns:     1,
+			HardwareAddress: chw,
+			MTU:             MTU,
+		})
+		if err != nil {
+			panic(err)
+		}
+		client.SetGateway6(sv.HardwareAddress())
+		// Create client connection.
+		var clConn tcp.Conn
+		err = clConn.Configure(tcp.ConnConfig{
+			RxBuf:             make([]byte, bufsize),
+			TxBuf:             make([]byte, bufsize),
+			TxPacketQueueSize: 4,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		// Client dials server.
+		err = client.DialTCP(&clConn, caddrp.Port(), netip.AddrPortFrom(sv.Addr(), svPort))
+		if err != nil {
+			t.Fatal(err)
+		}
+		// Complete TCP handshake.
+		tst.TestTCPHandshake(&client, sv)
+		// After handshake, TryAccept should work.
+		if listener.NumberOfReadyToAccept() != 1 {
+			t.Fatalf("after handshake: expected 1 ready, got %d", listener.NumberOfReadyToAccept())
+		}
+		svconn, err := listener.TryAccept()
+		if err != nil {
+			t.Fatal(err)
+		} else if svconn.RemotePort() != clConn.LocalPort() ||
+			[4]byte(svconn.RemoteAddr()) != client.Addr().As4() {
+			t.Fatal("race condition to listener acquisition")
+		}
+		// Verify both connections are established.
+		if clConn.State() != tcp.StateEstablished {
+			t.Fatalf("client: expected StateEstablished, got %s", clConn.State())
+		}
+		if len(data) > 0 {
+			tst.TestTCPEstablishedSingleData(&client, sv, &clConn, svconn, data)
+		}
+		if sleep > 0 {
+			time.Sleep(sleep)
+		}
+		tst.TestTCPClose(&client, sv, &clConn, svconn)
+	}
+
+	for range 1000 {
+		caddr := caddr.Next()
+		doRequest(netip.AddrPortFrom(caddr, uint16(sv.Prand32())), 0, []byte("HTTP 1.0\r\n"))
+	}
+}

x/xnet/xnet_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"math/rand"
 	"net/netip"
+	"sync"
 	"testing"
 
 	"github.com/soypat/lneto/arp"
@@ -14,6 +15,8 @@ import (
 )
 
 const (
+	logExchange = false
+
 	synack = tcp.FlagSYN | tcp.FlagACK
 	pshack = tcp.FlagPSH | tcp.FlagACK
 	finack = tcp.FlagFIN | tcp.FlagACK
@@ -143,12 +146,12 @@ func testerFrom(t *testing.T, mtu int) *tester {
 }
 
 type tester struct {
-	t       *testing.T
-	cap     pcap.PacketBreakdown
-	frmbuf  []pcap.Frame
-	buf     []byte
-	exch    []tcpExpectExchange
-	lastSeg tcp.Segment
+	t *testing.T
+
+	cap    pcap.PacketBreakdown
+	frmbuf []pcap.Frame
+	bufmu  sync.Mutex
+	buf    []byte
 }
 
 type tcpExpectExchange struct {
@@ -177,7 +180,7 @@ func (tst *tester) TestTCPSetupAndEstablish(svStack, clStack *StackAsync, svConn
 
 func (tst *tester) TestTCPHandshake(stack1, stack2 *StackAsync) {
 	tst.t.Helper()
-	tst.exch = append(tst.exch[:0], []tcpExpectExchange{
+	exch := [...]tcpExpectExchange{
 		{
 			SourceIdx: 0,
 			WantFlags: tcp.FlagSYN,
@@ -194,8 +197,8 @@ func (tst *tester) TestTCPHandshake(stack1, stack2 *StackAsync) {
 		},
 		noExchange(0),
 		noExchange(1),
-	}...)
-	for _, wants := range tst.exch {
+	}
+	for _, wants := range exch {
 		tst.TCPExchange(wants, stack1, stack2)
 	}
 }
@@ -217,7 +220,7 @@ func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync,
 		t.Fatal(err)
 	}
 	nprev := dstConn.BufferedInput()
-	tst.exch = append(tst.exch[:0], []tcpExpectExchange{
+	exch := [...]tcpExpectExchange{
 		{
 			SourceIdx: 0,
 			WantFlags: pshack,
@@ -230,10 +233,12 @@ func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync,
 		},
 		noExchange(0),
 		noExchange(1),
-	}...)
-	for _, wants := range tst.exch {
+	}
+	for _, wants := range exch {
 		tst.TCPExchange(wants, srcStack, dstStack)
 	}
+	tst.bufmu.Lock()
+	defer tst.bufmu.Unlock()
 	n, err := dstConn.Read(tst.buf)
 	if err != nil {
 		t.Errorf("reading back data %q on conn2: %s", sendData, err)
@@ -249,6 +254,7 @@ func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync,
 			t.Errorf("expected to read back %q from conn, got %q", sendData, got)
 		}
 	}
+
 	setzero(tst.buf[:n])
 }
 
@@ -263,7 +269,7 @@ func (tst *tester) TestTCPClose(stack1, stack2 *StackAsync, conn1, conn2 *tcp.Co
 	if err != nil {
 		t.Fatal(err)
 	}
-	tst.exch = append(tst.exch[:0], []tcpExpectExchange{
+	exch := [...]tcpExpectExchange{
 		{
 			SourceIdx: 0,
 			WantFlags: finack, // Closer sends FINACK
@@ -284,19 +290,22 @@ func (tst *tester) TestTCPClose(stack1, stack2 *StackAsync, conn1, conn2 *tcp.Co
 		},
 		noExchange(0),
 		noExchange(1),
-	}...)
-	t.Log(conn1.State().String(), conn2.State().String())
-	for i, exch := range tst.exch {
+	}
+	if logExchange {
+		t.Log(conn1.State().String(), conn2.State().String())
+	}
+	for i, exch := range exch {
 		failed := t.Failed()
-		tst.TCPExchange(exch, stack1, stack2)
+		seg := tst.TCPExchange(exch, stack1, stack2)
 		if !failed && t.Failed() {
 			t.Error(i, exch.SourceIdx, "close failure")
 		}
 		if exch.WantFlags == 0 {
 			continue
 		}
-
-		t.Log(i, tcp.StringExchange(tst.lastSeg, conn1.State(), conn2.State(), exch.SourceIdx != 0))
+		if logExchange {
+			t.Log(i, tcp.StringExchange(seg, conn1.State(), conn2.State(), exch.SourceIdx != 0))
+		}
 	}
 
 	state1 := conn1.State()
@@ -315,8 +324,9 @@ func (tst *tester) TestTCPClose(stack1, stack2 *StackAsync, conn1, conn2 *tcp.Co
 	}
 }
 
-func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAsync) {
-	tst.lastSeg = tcp.Segment{}
+func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAsync) tcp.Segment {
+	tst.bufmu.Lock()
+	defer tst.bufmu.Unlock()
 	var src, dst *StackAsync
 	defer func(failed bool) {
 		if !failed && tst.t.Failed() {
@@ -328,7 +338,6 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs
 	t.Helper()
 	buf := tst.buf[:cap(tst.buf)]
 	nodata := expect.WantFlags == 0
-
 	switch expect.SourceIdx {
 	case 0:
 		src, dst = stack1, stack2
@@ -337,17 +346,18 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs
 	default:
 		panic("OOB")
 	}
+
 	n, err := src.Encapsulate(buf[:], -1, 0)
 	if err != nil {
 		t.Fatal(err)
 	} else if n == 0 {
 		if nodata {
-			return // No data sent and no data expected.
+			return tcp.Segment{} // No data sent and no data expected.
 		}
 		t.Error("zero bits sent")
 	} else if nodata && n > 0 {
 		t.Error("expected no data sent and got data")
-		return
+		return tcp.Segment{}
 	}
 	defer setzero(buf[:n])
 
@@ -379,7 +389,6 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs
 
 	payload := tfrm.Payload()
 	seg := tfrm.Segment(len(payload))
-	tst.lastSeg = seg
 	if !bytes.Equal(payload, expect.WantData) {
 		t.Errorf("mismatched data sent, \nwant=%q\ngot=%q\n", expect.WantData, payload)
 	}
@@ -390,11 +399,14 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs
 	if err != nil {
 		t.Fatal(err)
 	}
+	return seg
 }
 
 func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) {
 	t := tst.t
 	t.Helper()
+	tst.bufmu.Lock()
+	defer tst.bufmu.Unlock()
 	buf := tst.buf[:cap(tst.buf)]
 
 	// === PHASE 1: ARP Request from querying stack ===