diff --git a/tcp/conn.go b/tcp/conn.go index 539867b..3aa2b41 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -65,9 +65,14 @@ func (conn *Conn) RemoteAddr() []byte { return conn.remoteAddr } // State returns the TCP state of the socket. func (conn *Conn) State() State { return conn.h.State() } -// BufferedInput returns the number of bytes in the socket's receive/input buffer. +// BufferedInput returns the number of bytes in the socket's receive(input) buffer +// and available to read via a [Conn.Read] call. func (conn *Conn) BufferedInput() int { return conn.h.BufferedInput() } +// AvailableOutput returns amount of bytes available to write to output +// before [Conn.Write] returns an error due to insufficient space to store outgoing data. +func (conn *Conn) AvailableOutput() int { return conn.h.AvailableOutput() } + // OpenActive opens a connection to a remote peer with a known IP address and port combination. // iss is the initial send sequence number which is ideally a random number which is far away from the last sequence number used on a connection to the same host. func (conn *Conn) OpenActive(localPort uint16, remote netip.AddrPort, iss Value) error { diff --git a/tcp/control.go b/tcp/control.go index 2810e2f..e767e41 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -509,6 +509,8 @@ func (tcb *ControlBlock) rstJump() Value { // Abort sets ControlBlock state to Closed and resets all sequence numbers and pending flag. // No more data can be sent nor received after the connection is aborted until opened again. +// An abort call prepares the connection for opening an active connection via a +// SYN packet during Send call in state=StateClosed. func (tcb *ControlBlock) Abort() { tcb.reset() tcb.debug("tcb:abort") diff --git a/tcp/handler.go b/tcp/handler.go index 70e8094..a43e63e 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -86,11 +86,13 @@ func (h *Handler) OpenActive(localPort, remotePort uint16, iss Value) error { if h.bufRx.Size() < minBufferSize || h.bufTx.Size() < minBufferSize { return errBufferTooSmall } - if h.scb.State() != StateClosed { + if h.scb.State() != StateClosed && h.scb.State() != StateTimeWait { return errNeedClosedTCBToOpen } else if remotePort == 0 { return errors.New("zero port on open call") } + // reset/Abort prepares a SCB for active connection by resetting state to closed. + h.scb.reset() h.reset(localPort, remotePort, iss) h.scb.SetRecvWindow(Size(h.bufRx.Size())) return nil @@ -312,7 +314,8 @@ func (h *Handler) Read(b []byte) (int, error) { return h.bufRx.Read(b) } -// BufferedInput returns amount of bytes buffered in receive buffer. +// BufferedInput returns amount of bytes buffered in receive(input) buffer and ready to read +// with a [Handler.Read] call. func (h *Handler) BufferedInput() int { if h.State().IsClosed() { return 0 @@ -320,6 +323,12 @@ func (h *Handler) BufferedInput() int { return h.bufRx.Buffered() } +// AvailableOutput returns amount of bytes available to write to output +// before [Handler.Write] returns an error. +func (h *Handler) AvailableOutput() int { + return h.bufTx.Free() +} + // AwaitingSynResponse returns true if the Handler is an active client opened with [Handler.OpenActive] and has already sent out the first SYN packet to the remote client. func (h *Handler) AwaitingSynResponse() bool { return h.remotePort != 0 && h.scb.State() == StateSynSent diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index 4ebcce2..6c3930a 100644 --- a/x/xnet/xnet_test.go +++ b/x/xnet/xnet_test.go @@ -3,6 +3,7 @@ package xnet import ( "bytes" "errors" + "math/rand" "net/netip" "testing" @@ -17,9 +18,34 @@ const ( finack = tcp.FlagFIN | tcp.FlagACK ) +func TestStackAsyncTCP_multipacket(t *testing.T) { + const seed = 1234 + const MTU = 1500 + const svPort = 8080 + const maxPkt = 30 + client, sv, clconn, svconn := newTCPStacks(t, seed, MTU) + tst := tester{ + t: t, buf: make([]byte, MTU), + } + rng := rand.New(rand.NewSource(seed)) + client2, sv2, clconn2, svconn2 := newTCPStacks(t, seed, MTU) + _, _, _, _ = client2, sv2, clconn2, svconn2 + tst.TestTCPSetupAndEstablish(sv, client, svconn, clconn, svPort, 1337) + tst.TestTCPClose(client, sv, clconn, svconn) + var buf [MTU]byte + for i := 0; i < 30; i++ { + payloadSize := rng.Intn(maxPkt) + 1 + tst.TestTCPSetupAndEstablish(sv, client, svconn, clconn, svPort, 1337) + a, _ := rng.Read(buf[:payloadSize]) + tst.TestTCPEstablishedSingleData(sv, client, svconn, clconn, buf[:a]) + tst.TestTCPClose(client, sv, clconn, svconn) + if t.Failed() { + t.FailNow() + } + } +} + func TestStackAsyncTCP_singlepacket(t *testing.T) { - var err error - _ = err const seed = 1234 const MTU = 1500 const svPort = 80 @@ -112,14 +138,14 @@ func noExchange(source int) tcpExpectExchange { return tcpExpectExchange{SourceIdx: source} } -func (tst *tester) TestTCPSetupAndEstablish(svStack, clStack *StackAsync, svconn, clconn *tcp.Conn, svPort, clPort uint16) { +func (tst *tester) TestTCPSetupAndEstablish(svStack, clStack *StackAsync, svConn, clConn *tcp.Conn, svPort, clPort uint16) { t := tst.t // Attach server and client connections to stacks. - err := svStack.ListenTCP(svconn, svPort) + err := svStack.ListenTCP(svConn, svPort) if err != nil { t.Fatal(err) } - err = clStack.DialTCP(clconn, clPort, netip.AddrPortFrom(svStack.Addr(), svPort)) + err = clStack.DialTCP(clConn, clPort, netip.AddrPortFrom(svStack.Addr(), svPort)) if err != nil { t.Fatal(err) } @@ -151,13 +177,20 @@ func (tst *tester) TestTCPHandshake(stack1, stack2 *StackAsync) { } } -func (tst *tester) TestTCPEstablishedSingleData(stack1, stack2 *StackAsync, conn1, conn2 *tcp.Conn, sendData []byte) { - tst.t.Helper() - _, err := conn1.Write(sendData) - if err != nil { - tst.t.Fatal(err) +func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync, srcConn, dstConn *tcp.Conn, sendData []byte) { + t := tst.t + t.Helper() + avail := srcConn.AvailableOutput() + if avail < len(sendData) { + t.Fatal("insufficient space for write call", avail, len(sendData)) + } else if len(sendData) <= 0 { + panic("empty data!") } - nprev := conn2.BufferedInput() + _, err := srcConn.Write(sendData) + if err != nil { + t.Fatal(err) + } + nprev := dstConn.BufferedInput() tst.exch = append(tst.exch[:0], []tcpExpectExchange{ { SourceIdx: 0, @@ -173,21 +206,21 @@ func (tst *tester) TestTCPEstablishedSingleData(stack1, stack2 *StackAsync, conn noExchange(1), }...) for _, wants := range tst.exch { - tst.TCPExchange(wants, stack1, stack2) + tst.TCPExchange(wants, srcStack, dstStack) } - n, err := conn2.Read(tst.buf) + n, err := dstConn.Read(tst.buf) if err != nil { - tst.t.Errorf("reading back data %q on conn2: %s", sendData, err) + t.Errorf("reading back data %q on conn2: %s", sendData, err) } else if n == len(tst.buf) { - tst.t.Fatalf("buffer topped out in read!") + t.Fatalf("buffer topped out in read!") } nread := n - nprev if nread != len(sendData) { - tst.t.Errorf("expected to read %d bytes, got %d", len(sendData), nread) + t.Errorf("expected to read %d bytes, got %d", len(sendData), nread) } else { got := tst.buf[n-nread : n] if !bytes.Equal(got, sendData) { - tst.t.Errorf("expected to read back %q from conn, got %q", sendData, got) + t.Errorf("expected to read back %q from conn, got %q", sendData, got) } } setzero(tst.buf[:n])