diff --git a/examples/stack/main.go b/examples/stack/main.go index a23ae27..8d84603 100644 --- a/examples/stack/main.go +++ b/examples/stack/main.go @@ -249,7 +249,7 @@ func (stack *Stack) OpenTCPListener(port uint16) (*internet.NodeTCPListener, err func (stack *Stack) OpenPassiveTCP(port uint16, iss tcp.Value) (*tcp.Conn, error) { mtu := stack.ethernet.MTU() conn := new(tcp.Conn) - err := conn.Configure(&tcp.ConnConfig{ + err := conn.Configure(tcp.ConnConfig{ RxBuf: make([]byte, mtu), TxBuf: make([]byte, mtu), TxPacketQueueSize: 3, @@ -296,7 +296,7 @@ func (naiveTCPPool) GetTCP() (*tcp.Conn, tcp.Value) { rand.Read(buf[:]) randVal := binary.LittleEndian.Uint32(buf[:]) var conn tcp.Conn - err := conn.Configure(&tcp.ConnConfig{ + err := conn.Configure(tcp.ConnConfig{ RxBuf: make([]byte, 1024), TxBuf: make([]byte, 1024), TxPacketQueueSize: 3, diff --git a/examples/stackbasic/main.go b/examples/stackbasic/main.go index 98088cb..8e124bb 100644 --- a/examples/stackbasic/main.go +++ b/examples/stackbasic/main.go @@ -182,7 +182,7 @@ func NewEthernetTCPStack(ourMAC, gwMAC [6]byte, ip netip.AddrPort, mtu uint16, s lport: 0, }) var conn tcp.Conn - err = conn.Configure(&tcp.ConnConfig{ + err = conn.Configure(tcp.ConnConfig{ RxBuf: make([]byte, mtu), TxBuf: make([]byte, mtu), TxPacketQueueSize: 3, diff --git a/examples/tap/main.go b/examples/tap/main.go index ca8d056..8831190 100644 --- a/examples/tap/main.go +++ b/examples/tap/main.go @@ -32,6 +32,7 @@ func run() error { flagInterface = "tap0" ) flag.StringVar(&flagInterface, "i", flagInterface, "Interface to select. tap* creates a tap interface. Any other name will create a bridge to the name of the interface i.e: 'enp7s0', 'wlp8s0', 'lo'") + flag.Parse() var ( flagNet = "192.168.10.1/24" flagiface = "tap0" diff --git a/examples/xnet/main.go b/examples/xnet/main.go index 859b704..66d8da9 100644 --- a/examples/xnet/main.go +++ b/examples/xnet/main.go @@ -12,8 +12,10 @@ import ( "time" "github.com/soypat/lneto/dns" + "github.com/soypat/lneto/http/httpraw" "github.com/soypat/lneto/internal" "github.com/soypat/lneto/internal/ltesto" + "github.com/soypat/lneto/internet/pcap" "github.com/soypat/lneto/x/xnet" ) @@ -35,9 +37,11 @@ func run() (err error) { flagHostToResolve = "" flagRequestedIP = "" flagDoNTP = false + flagHTTPGet = false ) + flag.BoolVar(&flagHTTPGet, "httpget", flagHTTPGet, "Do an HTTP GET request ") flag.StringVar(&flagInterface, "i", flagInterface, "Interface to use. Either tap* or the name of an existing interface to bridge to.") - flag.BoolVar(&flagUseHTTP, "http", flagUseHTTP, "Use HTTP tap interface.") + flag.BoolVar(&flagUseHTTP, "ihttp", flagUseHTTP, "Use HTTP tap interface.") flag.StringVar(&flagHostToResolve, "host", flagHostToResolve, "Hostname to resolve via DNS.") flag.StringVar(&flagRequestedIP, "addr", flagRequestedIP, "IP address to request via DHCP.") flag.BoolVar(&flagDoNTP, "ntp", flagDoNTP, "Do NTP round and print result time") @@ -89,6 +93,9 @@ func run() (err error) { RandSeed: softRand, HardwareAddress: brHW, MTU: uint16(mtu), + MaxTCPConns: 1, + TCPBufferSizeTx: 2048, + TCPBufferSizeRx: 2048, }) if err != nil { return err @@ -97,12 +104,17 @@ func run() (err error) { go func() { lastAction := time.Now() buf := make([]byte, mtu) + var cap pcap.PacketBreakdown for { clear(buf) nwrite, err := stack.Encapsulate(buf[:], 0) if err != nil { fmt.Println("ERR:ENCAPSULATE", err) } else if nwrite > 0 { + frames, err := cap.CaptureEthernet(nil, buf[:nwrite], 0) + if len(frames) > 0 { + fmt.Println("OUT", frames) + } n, err := iface.Write(buf[:nwrite]) if err != nil { log.Fatal("groutine encapsulate:", err) @@ -179,6 +191,43 @@ func run() (err error) { return fmt.Errorf("DNS of host %q failed: %w", flagHostToResolve, err) } fmt.Printf("DNS resolution of %q complete and resolved to %v\n", flagHostToResolve, addrs) + if flagHTTPGet { + var hdr httpraw.Header + hdr.SetMethod("GET") + hdr.SetRequestURI("/") + hdr.SetProtocol("HTTP/1.1") + hdr.Set("Host", flagHostToResolve) + hdr.Set("User-Agent", "lneto") + hdr.Set("Accept-Language", "en-US,en;q=0.5") + req, err := hdr.AppendRequest(nil) + if err != nil { + return err + } + target := netip.AddrPortFrom(addrs[0], 80) + conn, err := rstack.DoDialTCP(uint16(softRand&0xefff)+1024, target, internetTimeout, internetRetries) + if err != nil { + return fmt.Errorf("TCP failed: %w", err) + } + conn.SetDeadline(time.Now().Add(internetTimeout)) + _, err = conn.Write(req) + if err != nil { + return err + } + rxbuf := make([]byte, 2048) + var page []byte + for { + var n int + n, err = conn.Read(rxbuf) + page = append(page, rxbuf[:n]...) + if err != nil { + break + } + } + if len(page) == 0 { + return err + } + os.Stdout.Write(page) + } return nil } diff --git a/internet/stack-ports.go b/internet/stack-ports.go index a9087d1..2016e30 100644 --- a/internet/stack-ports.go +++ b/internet/stack-ports.go @@ -93,15 +93,24 @@ func (ps *StackPorts) Demux(b []byte, offset int) (err error) { func (ps *StackPorts) Register(h StackNode) error { port := h.LocalPort() proto := h.Protocol() + if port <= 0 { return errZeroPort } else if proto != uint64(ps.protocol) { return errInvalidProto } + var cid uint64 + cidPtr := h.ConnectionID() + if cidPtr != nil { + cid = *cidPtr + } return registerNode(&ps.handlers, node{ demux: h.Demux, encapsulate: h.Encapsulate, port: port, + currConnID: cid, + connID: cidPtr, + proto: uint16(proto), }) } diff --git a/internet/stackbasic_test.go b/internet/stackbasic_test.go index c431653..39a704a 100644 --- a/internet/stackbasic_test.go +++ b/internet/stackbasic_test.go @@ -94,7 +94,7 @@ func setupClientServer(t *testing.T, rng *rand.Rand, client, server *StackIP, co server.Reset(svip.Addr(), maxNodes) client.Reset(clip.Addr(), maxNodes) - err := connServer.Configure(&tcp.ConnConfig{ + err := connServer.Configure(tcp.ConnConfig{ RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 3, @@ -103,7 +103,7 @@ func setupClientServer(t *testing.T, rng *rand.Rand, client, server *StackIP, co if err != nil { t.Fatal(err) } - err = connClient.Configure(&tcp.ConnConfig{ + err = connClient.Configure(tcp.ConnConfig{ RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 3, diff --git a/tcp/conn.go b/tcp/conn.go index 2c68035..662cdff 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -44,7 +44,7 @@ type ConnConfig struct { Logger *slog.Logger } -func (conn *Conn) Configure(config *ConnConfig) (err error) { +func (conn *Conn) Configure(config ConnConfig) (err error) { err = conn.h.SetBuffers(config.TxBuf, config.RxBuf, config.TxPacketQueueSize) if err != nil { return err @@ -109,9 +109,17 @@ func (conn *Conn) Abort() { h: conn.h, remoteAddr: conn.remoteAddr[:0], logger: conn.logger, + ipID: conn.ipID, } } +// InternalHandler returns the internal [Handler] instance. The Handler contains lower level implementation logic for a TCP connection. +// Typical users should not be using this method unless implementing a stack which manages several TCP connections and thus need +// access to low level internals for careful memory management. +func (conn *Conn) InternalHandler() *Handler { + return &conn.h +} + // Write writes argument data to the TCPConns's output buffer which is queued to be sent. func (conn *Conn) Write(b []byte) (int, error) { err := conn.checkPipeOpen() diff --git a/tcp/handler.go b/tcp/handler.go index 10c31f9..bcd339b 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -53,16 +53,17 @@ func (h *Handler) ConnectionID() *uint64 { func (h *Handler) State() State { return h.scb.State() } // SetBuffers sets the internal buffers used to receive and transmit bytes asynchronously via [Handler.Write] and [Handler.Read] calls. +// If the argument buffer is nil then the respective currently set buffer will be reused. func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error { + if h.bufRx.Buf == nil && (len(rxbuf) < minBufferSize || len(txbuf) < minBufferSize) { + return errors.New("tcp: short buffer") + } 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) < minBufferSize { - return errors.New("short rx buffer") - } h.scb.SetRecvWindow(Size(h.bufRx.Size())) h.bufRx.Reset() return h.bufTx.ResetOrReuse(txbuf, packets, 0) @@ -270,11 +271,16 @@ func (h *Handler) Send(b []byte) (int, error) { return int(offset)*4 + int(segment.DATALEN), nil } -// Free returns the amount of space free in the transmit buffer. A call to [Handler.Write] with a larger buffer will fail. -func (h *Handler) Free() int { +// FreeTx returns the amount of space free in the transmit buffer. A call to [Handler.Write] with a larger buffer will fail. +func (h *Handler) FreeTx() int { return h.bufTx.Free() } +// SizeRx returns the size of the TCP receive ring buffer. +func (h *Handler) SizeRx() int { + return h.bufRx.Size() +} + // Write implements [io.Writer] by copying b to a internal buffer to be sent over the network on the next // [Handler.Send] call that can send data to remote peer. Use [Handler.Free] to know the maximum length the argument slice can be before erroring. func (h *Handler) Write(b []byte) (int, error) { diff --git a/x/xnet/stack-async.go b/x/xnet/stack-async.go index 5aed0a7..aa8a7f7 100644 --- a/x/xnet/stack-async.go +++ b/x/xnet/stack-async.go @@ -2,6 +2,7 @@ package xnet import ( "errors" + "io" "net/netip" "sync" "time" @@ -16,6 +17,10 @@ import ( "github.com/soypat/lneto/tcp" ) +const ( + minTCPBuffer = 256 +) + type StackAsync struct { mu sync.Mutex hostname string @@ -52,6 +57,8 @@ type StackConfig struct { NTPServer netip.Addr Hostname string MaxTCPConns int + TCPBufferSizeTx int // Size of transmit buffer for TCP connections. + TCPBufferSizeRx int // Size of receive buffer for TCP connections. RandSeed int64 HardwareAddress [6]byte MTU uint16 @@ -112,9 +119,35 @@ func (s *StackAsync) Reset(cfg StackConfig) error { // Enable TCP if connections present. if cfg.MaxTCPConns > 0 { + if cfg.TCPBufferSizeRx < minTCPBuffer || cfg.TCPBufferSizeTx < minTCPBuffer { + return io.ErrShortBuffer + } if cap(s.tcpconns) < cfg.MaxTCPConns { s.tcpconns = make([]tcp.Conn, cfg.MaxTCPConns) } + s.tcpconns = s.tcpconns[:cfg.MaxTCPConns] + for i := range s.tcpconns { + c := &s.tcpconns[i] + c.Abort() // Abort forcibly closes connection to allow buffer setting and reduce risk of race conditions. + h := c.InternalHandler() + sizetx := h.FreeTx() + sizerx := h.SizeRx() + // If size is sufficient then buffers will remain nil and memory will be reused. + var rxbuf, txbuf []byte + if sizetx < cfg.TCPBufferSizeTx || sizerx < cfg.TCPBufferSizeRx { + buf := make([]byte, cfg.TCPBufferSizeTx+cfg.TCPBufferSizeRx) + rxbuf = buf[:cfg.TCPBufferSizeRx] + txbuf = buf[cfg.TCPBufferSizeRx : cfg.TCPBufferSizeRx+cfg.TCPBufferSizeTx] + } + err = c.Configure(tcp.ConnConfig{ + RxBuf: rxbuf, + TxBuf: txbuf, + TxPacketQueueSize: 4, + }) + if err != nil { + return err + } + } err = s.tcps.ResetTCP(cfg.MaxTCPConns) if err != nil { return err @@ -213,7 +246,7 @@ func (s *StackAsync) DialTCP(localPort uint16, addrp netip.AddrPort) (conn *tcp. } for i := range s.tcpconns { maybeFreeConn := &s.tcpconns[i] - state := conn.State() + state := maybeFreeConn.State() if state.IsClosed() { conn = maybeFreeConn break // Can be used! diff --git a/x/xnet/stack-blocking.go b/x/xnet/stack-blocking.go index a52a5a7..b31f44b 100644 --- a/x/xnet/stack-blocking.go +++ b/x/xnet/stack-blocking.go @@ -122,21 +122,20 @@ func (s StackBlocking) DoDialTCP(localPort uint16, addrp netip.AddrPort, timeout deadline := time.Now().Add(timeout) for i := 0; i < maxIter; i++ { state := conn.State() - switch state { - case tcp.StateEstablished: - break - case tcp.StateSynSent, tcp.StateSynRcvd: + if state == tcp.StateEstablished { + return conn, nil + } else if state == tcp.StateSynSent || state == tcp.StateSynRcvd || conn.InternalHandler().AwaitingSynSend() { if err = s.checkDeadline(deadline); err != nil { return nil, err } time.Sleep(sleep) - default: + } else { // Unexpected state, abort and terminate connection. conn.Abort() return nil, errTCPFailedToConnect } } - return conn, nil + return conn, errDeadlineExceed } func (s StackBlocking) checkDeadline(deadline time.Time) error {