Add ICMPv4 and lneto.StackNode and deprecate internet.StackNode (#65)

* begin adding icmp client

* rely on anon structs

* add tests

* tests passing

* icmp fleshed out

* rework icmp to include ip addr

* rename StackAsync.Demux/Encapsulate to RecvEthernet and SendEthernet

* remove legacy unreachable TCP tests

* remove uses of deprecated internet.StackNode in preference of lneto.StackNode

* documentation

* rename methods to signal no I/O happening

x/xnet/stack-async.go

@@ -1,6 +1,7 @@
 package xnet
 
 import (
+	"encoding/binary"
 	"errors"
 	"log/slog"
 	"net/netip"
@@ -14,6 +15,7 @@ import (
 	"github.com/soypat/lneto/ethernet"
 	"github.com/soypat/lneto/internal"
 	"github.com/soypat/lneto/internet"
+	"github.com/soypat/lneto/ipv4/icmpv4"
 	"github.com/soypat/lneto/ntp"
 	"github.com/soypat/lneto/tcp"
 )
@@ -29,6 +31,7 @@ type StackAsync struct {
 	link     internet.StackEthernet
 	ip       internet.StackIP
 	arp      arp.Handler
+	icmp     icmpv4.Client
 	udps     internet.StackPorts
 	tcps     internet.StackPortsMACFiltered
 
@@ -72,41 +75,71 @@ type StackConfig struct {
 	MTU             uint16
 	// Accept multicast ethernet and IP packets. Needed for MDNS.
 	AcceptMulticast bool
+	// ICMPQueueLimit sets maximum number of input/output packets queued for processing.
+	// If set to zero ICMP cannot be enabled on the stack.
+	ICMPQueueLimit int
 }
 
 func (s *StackAsync) Hostname() string {
 	return s.hostname
 }
 
-func (s *StackAsync) Demux(carrierData []byte, etherOff int) error {
+// IngressEthernet receives an Ethernet frame from the network and processes it through the stack. The frame should include the Ethernet header and payload and CRC if enabled.
+func (s *StackAsync) IngressEthernet(ethernetFrame []byte) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.totalrecv += uint64(len(carrierData) - etherOff)
-	return s.link.Demux(carrierData, etherOff)
+	s.totalrecv += uint64(len(ethernetFrame))
+	return s.link.Demux(ethernetFrame, 0)
 }
 
-func (s *StackAsync) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) {
+// EgressEthernet writes the next ethernet frame to send into dstEthernetFrame from the stack.
+// The length of dstEthernetFrame should be at least MTU + Ethernet header (14) + CRC (4 if enabled).
+func (s *StackAsync) EgressEthernet(dstEthernetFrame []byte) (int, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-
-	n, err := s.link.Encapsulate(carrierData, offsetToIP, offsetToFrame)
+	n, err := s.link.Encapsulate(dstEthernetFrame, -1, 0)
 	s.totalsent += uint64(n)
 	return n, err
 }
 
+// IngressIP processes an incoming IP frame through the stack and omits ethernet header processing.
+func (s *StackAsync) IngressIP(ipFrame []byte) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.totalrecv += uint64(len(ipFrame))
+	return s.ip.Demux(ipFrame, 0)
+}
+
+// EgressIP writes the next IP frame to send into dstIPFrame from the stack. The length of dstIPFrame should be at least MTU.
+func (s *StackAsync) EgressIP(dstIPFrame []byte) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(dstIPFrame) < s.link.MTU() {
+		return 0, lneto.ErrShortBuffer
+	}
+	n, err := s.ip.Encapsulate(dstIPFrame, 0, 0)
+	s.totalsent += uint64(n)
+	return n, err
+}
+
+// MTU is the Maximum Transmission Unit of the stack corresponding
+// to the maximum payload size of an ethernet frame that can be sent through the stack.
+// Important to note that the actual ethernet frame size is MTU + Ethernet header (14) + CRC (4 if enabled), this is known as the Maximum Frame Length.
 func (s *StackAsync) MTU() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	return s.link.MTU()
 }
 
 func (s *StackAsync) Reset(cfg StackConfig) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	mac := cfg.HardwareAddress
-	addr := cfg.StaticAddress
-	s.prng = uint32(cfg.RandSeed)
-	if s.prng == 0 {
+	if cfg.RandSeed == 0 {
 		return lneto.ErrInvalidConfig
 	}
+	mac := cfg.HardwareAddress
+	addr := cfg.StaticAddress
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.prng = uint32(cfg.RandSeed)
 	s.hostname = cfg.Hostname
 	if !addr.IsValid() {
 		addr = netip.AddrFrom4([4]byte{}) // If static not set DHCP will be performed and address will be zero.
@@ -133,7 +166,6 @@ func (s *StackAsync) Reset(cfg StackConfig) error {
 		return err
 	}
 	s.ip.SetAcceptMulticast(cfg.AcceptMulticast)
-	//
 	err = s.resetARP()
 	if err != nil {
 		return err
@@ -144,9 +176,6 @@ func (s *StackAsync) Reset(cfg StackConfig) error {
 		return err
 	}
 	internal.SliceReuse(&s.userUDPs, cfg.MaxUDPConns)
-	if err != nil {
-		return err
-	}
 
 	// Enable TCP if connections present.
 	if cfg.MaxTCPConns > 0 {
@@ -170,6 +199,16 @@ func (s *StackAsync) Reset(cfg StackConfig) error {
 	if err != nil {
 		return err
 	}
+	if cfg.ICMPQueueLimit > 0 {
+		err = s.icmp.Configure(icmpv4.ClientConfig{
+			ResponseQueueBuffer: make([]byte, cfg.ICMPQueueLimit*64),
+			ResponseQueueLimit:  cfg.ICMPQueueLimit,
+			HashSeed:            s.Prand32(),
+		})
+		if err != nil {
+			return err
+		}
+	}
 	var timebuf [32]time.Time
 	s.sysprec = ntp.CalculateSystemPrecision(time.Now, timebuf[:])
 	if s.clientID == "" {
@@ -211,6 +250,18 @@ func (s *StackAsync) resetARP() error {
 	return nil
 }
 
+func (s *StackAsync) prandRead(buf []byte) {
+	i := 0
+	for ; i+3 < len(buf); i += 4 {
+		binary.LittleEndian.PutUint32(buf[i:], s.prand32())
+	}
+	v := s.prand32()
+	for i < len(buf) {
+		buf[i] = byte(v >> (8 * (i % 4)))
+		i++
+	}
+}
+
 // Prand32 generates a pseudo random 32-bit unsigned integer from the internal state and advances the seed.
 func (s *StackAsync) Prand32() (randval uint32) {
 	s.mu.Lock()
@@ -221,10 +272,7 @@ func (s *StackAsync) Prand32() (randval uint32) {
 
 func (s *StackAsync) prand32() uint32 {
 	/* Algorithm "xor" from p. 4 of Marsaglia, "Xorshift RNGs" */
-	seed := s.prng
-	seed ^= seed << 13
-	seed ^= seed >> 17
-	seed ^= seed << 5
+	seed := internal.Prand32(s.prng)
 	s.prng = seed
 	return seed
 }
@@ -282,6 +330,22 @@ func (s *StackAsync) Gateway6() [6]byte {
 	return s.link.Gateway6()
 }
 
+// EnableICMP registers an ICMP handler to the stack when enabled is true.
+// If enabled=false the currently registered ICMP handler is unregistered and state reset.
+func (s *StackAsync) EnableICMP(enabled bool) (err error) {
+	if enabled {
+		if s.ip.IsRegistered(lneto.IPProtoICMP) {
+			err = lneto.ErrAlreadyRegistered
+		} else {
+			err = s.ip.Register(&s.icmp)
+		}
+
+	} else {
+		s.icmp.Abort()
+	}
+	return err
+}
+
 func (s *StackAsync) DialTCP(conn *tcp.Conn, localPort uint16, addrp netip.AddrPort) (err error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -337,7 +401,7 @@ func (s *StackAsync) RegisterListener(listener *tcp.Listener) (err error) {
 // RegisterUDP registers a StackNode on a UDP port with the given remote address and port.
 // The StackUDPPort wrapping is handled internally. The number of user-registered UDP ports
 // is limited by [StackConfig.MaxUDPConns].
-func (s *StackAsync) RegisterUDP(node internet.StackNode, remoteAddr []byte, remotePort uint16) error {
+func (s *StackAsync) RegisterUDP(node lneto.StackNode, remoteAddr []byte, remotePort uint16) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	idx := len(s.userUDPs)

x/xnet/stack-blocking.go

@@ -2,9 +2,11 @@ package xnet
 
 import (
 	"errors"
+	"net"
 	"net/netip"
 	"time"
 
+	"github.com/soypat/lneto"
 	"github.com/soypat/lneto/dhcpv4"
 	"github.com/soypat/lneto/tcp"
 )
@@ -58,6 +60,33 @@ func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPRe
 	return s.async.ResultDHCP()
 }
 
+func (s StackBlocking) DoPing(hostAddr netip.Addr, timeout time.Duration) (roundtrip time.Duration, err error) {
+	if !hostAddr.Is4() {
+		return 0, lneto.ErrInvalidAddr
+	}
+	var buf [16]byte
+	s.async.prandRead(buf[:])
+	key, err := s.async.icmp.PingStart(hostAddr.As4(), buf[:], 56) // size=56 so ICMP size is 64, like linux.
+	if err != nil {
+		return 0, err
+	}
+	start := time.Now()
+	sleep := timeout / maxIter
+	for i := 0; i < maxIter; i++ {
+		time.Sleep(sleep)
+		elapsed := time.Since(start)
+		completed, exists := s.async.icmp.PingPop(key)
+		if !exists {
+			return 0, net.ErrClosed // lneto.ErrAborted
+		} else if completed {
+			return elapsed, nil
+		} else if elapsed > timeout {
+			break
+		}
+	}
+	return 0, errDeadlineExceed
+}
+
 func (s StackBlocking) DoNTP(hostAddr netip.Addr, timeout time.Duration) (offset time.Duration, err error) {
 	err = s.async.StartNTP(hostAddr)
 	if err != nil {

x/xnet/xnet_bench_test.go

@@ -46,23 +46,23 @@ func BenchmarkARPExchange(b *testing.B) {
 		if err != nil {
 			b.Fatal(err)
 		}
-		n, err := c1.Encapsulate(buf[:], -1, 0) // Send Request.
+		n, err := c1.EgressEthernet(buf[:]) // Send Request.
 		if err != nil {
 			b.Fatal(err)
 		} else if n == 0 {
 			b.Fatal("expected send of data after first query")
 		}
-		err = c2.Demux(buf[:n], 0) // Receive request.
+		err = c2.IngressEthernet(buf[:n]) // Receive request.
 		if err != nil {
 			b.Fatal(err)
 		}
-		n, err = c2.Encapsulate(buf[:], -1, 0) // Send response.
+		n, err = c2.EgressEthernet(buf[:]) // Send response.
 		if err != nil {
 			b.Fatal(err)
 		} else if n == 0 {
 			b.Fatal("got no response to request")
 		}
-		err = c1.Demux(buf[:n], 0) // Receive response.
+		err = c1.IngressEthernet(buf[:n]) // Receive response.
 		if err != nil {
 			b.Fatal(err)
 		}
@@ -140,31 +140,31 @@ func BenchmarkTCPHandshake(b *testing.B) {
 		}
 
 		// SYN from client.
-		n, err := client.Encapsulate(pktbuf[:], -1, 0)
+		n, err := client.EgressEthernet(pktbuf[:])
 		if err != nil {
 			b.Fatal(err)
 		}
-		err = sv.Demux(pktbuf[:n], 0)
+		err = sv.IngressEthernet(pktbuf[:n])
 		if err != nil {
 			b.Fatal(err)
 		}
 
 		// SYN-ACK from server.
-		n, err = sv.Encapsulate(pktbuf[:], -1, 0)
+		n, err = sv.EgressEthernet(pktbuf[:])
 		if err != nil {
 			b.Fatal(err)
 		}
-		err = client.Demux(pktbuf[:n], 0)
+		err = client.IngressEthernet(pktbuf[:n])
 		if err != nil {
 			b.Fatal(err)
 		}
 
 		// ACK from client.
-		n, err = client.Encapsulate(pktbuf[:], -1, 0)
+		n, err = client.EgressEthernet(pktbuf[:])
 		if err != nil {
 			b.Fatal(err)
 		}
-		err = sv.Demux(pktbuf[:n], 0)
+		err = sv.IngressEthernet(pktbuf[:n])
 		if err != nil {
 			b.Fatal(err)
 		}

x/xnet/xnet_concurrent_test.go

@@ -159,15 +159,15 @@ func kernelLoop(ctx context.Context, server *StackAsync, clients []StackAsync) {
 		}
 
 		// Process server outgoing -> route to appropriate client based on dest IP.
-		if n, _ := server.Encapsulate(buf, -1, 0); n > 0 {
+		if n, _ := server.EgressEthernet(buf); n > 0 {
 			routePacketToClient(buf[:n], clients)
 		}
 
 		// Process each client outgoing in randomized order.
 		rng.Shuffle(len(order), func(i, j int) { order[i], order[j] = order[j], order[i] })
 		for _, idx := range order {
-			if n, _ := clients[idx].Encapsulate(buf, -1, 0); n > 0 {
-				server.Demux(buf[:n], 0) // All clients talk to server.
+			if n, _ := clients[idx].EgressEthernet(buf); n > 0 {
+				server.IngressEthernet(buf[:n]) // All clients talk to server.
 			}
 		}
 
@@ -184,7 +184,7 @@ func routePacketToClient(pkt []byte, clients []StackAsync) {
 
 	for i := range clients {
 		if clients[i].Addr() == dstIP {
-			clients[i].Demux(pkt, 0)
+			clients[i].IngressEthernet(pkt)
 			return
 		}
 	}

x/xnet/xnet_dns_test.go

@@ -49,7 +49,7 @@ func TestDNS_QueryReceivesAnswer(t *testing.T) {
 	// Client sends DNS query.
 	const carrierDataSize = MTU + ethernet.MaxOverheadSize
 	var buf [carrierDataSize]byte
-	n, err := client.Encapsulate(buf[:], -1, 0)
+	n, err := client.EgressEthernet(buf[:])
 	if err != nil {
 		t.Fatal("client Encapsulate failed:", err)
 	}
@@ -70,7 +70,7 @@ func TestDNS_QueryReceivesAnswer(t *testing.T) {
 	}
 
 	// Deliver response to client.
-	err = client.Demux(responsePkt, 0)
+	err = client.IngressEthernet(responsePkt)
 	if err != nil {
 		t.Fatal("client Demux failed:", err)
 	}

x/xnet/xnet_fuzz_test.go

@@ -34,12 +34,12 @@ func FuzzStackAsyncHTTP(f *testing.F) {
 	written := false
 	closed := false
 	for {
-		n1, err := s1.Encapsulate(buf[:], -1, 0)
+		n1, err := s1.EgressEthernet(buf[:])
 		if err != nil {
 			f.Fatal(err)
 		}
 		if n1 > 0 {
-			err = s2.Demux(buf[:n1], 0)
+			err = s2.IngressEthernet(buf[:n1])
 			if err != nil {
 				f.Fatal(err)
 			}
@@ -53,10 +53,10 @@ func FuzzStackAsyncHTTP(f *testing.F) {
 				written = true
 			}
 		}
-		n2, err := s2.Encapsulate(buf[:], -1, 0)
+		n2, err := s2.EgressEthernet(buf[:])
 		if n2 > 0 {
 			pktnum++
-			err = s1.Demux(buf[:n2], 0)
+			err = s1.IngressEthernet(buf[:n2])
 			if err != nil {
 				f.Fatal(err)
 			}
@@ -88,7 +88,7 @@ func FuzzStackAsyncHTTP(f *testing.F) {
 		closed := false
 		const maxpkts = 100
 		for {
-			n1, err := s1.Encapsulate(buf[:], -1, 0)
+			n1, err := s1.EgressEthernet(buf[:])
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -97,21 +97,21 @@ func FuzzStackAsyncHTTP(f *testing.F) {
 					n1 = copy(buf[:], a)
 					fixIPTCPCRCs(buf[:n1])
 				}
-				s2.Demux(buf[:n1], 0)
+				s2.IngressEthernet(buf[:n1])
 				pkt++
 				if !written && c2.State() >= tcp.StateEstablished {
 					c2.Write(data)
 					written = true
 				}
 			}
-			n2, err := s2.Encapsulate(buf[:], -1, 0)
+			n2, err := s2.EgressEthernet(buf[:])
 			if n2 > 0 {
 				if pkt == pktnum {
 					n2 = copy(buf[:], a)
 					fixIPTCPCRCs(buf[:n2])
 				}
 				pkt++
-				s1.Demux(buf[:n2], 0)
+				s1.IngressEthernet(buf[:n2])
 			}
 			if n1 == 0 && n2 == 0 {
 				if !closed {

x/xnet/xnet_mdns_test.go

@@ -114,7 +114,7 @@ func TestMDNS_QueryResponse(t *testing.T) {
 	var buf [carrierDataSize]byte
 
 	// Querier encapsulates query through full stack (Ethernet+IP+UDP+mDNS).
-	n, err := querierStack.Encapsulate(buf[:], -1, 0)
+	n, err := querierStack.EgressEthernet(buf[:])
 	if err != nil || n == 0 {
 		t.Fatal("querier encapsulate:", err, n)
 	}
@@ -135,13 +135,13 @@ func TestMDNS_QueryResponse(t *testing.T) {
 	}
 
 	// Responder demuxes the query (multicast MAC+IP accepted via AcceptMulticast).
-	err = responderStack.Demux(buf[:n], 0)
+	err = responderStack.IngressEthernet(buf[:n])
 	if err != nil {
 		t.Fatal("responder demux:", err)
 	}
 
 	// Responder encapsulates response.
-	n, err = responderStack.Encapsulate(buf[:], -1, 0)
+	n, err = responderStack.EgressEthernet(buf[:])
 	if err != nil || n == 0 {
 		t.Fatal("responder encapsulate:", err, n)
 	}
@@ -165,7 +165,7 @@ func TestMDNS_QueryResponse(t *testing.T) {
 	}
 
 	// Querier demuxes response.
-	err = querierStack.Demux(buf[:n], 0)
+	err = querierStack.IngressEthernet(buf[:n])
 	if err != nil {
 		t.Fatal("querier demux:", err)
 	}
@@ -310,25 +310,25 @@ func mdnsQueryRespond(t *testing.T, querier, responder *StackAsync, buf []byte)
 	t.Helper()
 
 	// Querier encapsulates query.
-	n, err := querier.Encapsulate(buf, -1, 0)
+	n, err := querier.EgressEthernet(buf)
 	if err != nil || n == 0 {
 		t.Fatal("querier encapsulate:", err, n)
 	}
 
 	// Responder demuxes multicast query directly.
-	err = responder.Demux(buf[:n], 0)
+	err = responder.IngressEthernet(buf[:n])
 	if err != nil {
 		t.Fatal("responder demux:", err)
 	}
 
 	// Responder encapsulates response.
-	n, err = responder.Encapsulate(buf, -1, 0)
+	n, err = responder.EgressEthernet(buf)
 	if err != nil || n == 0 {
 		t.Fatal("responder encapsulate:", err, n)
 	}
 
 	// Querier demuxes multicast response.
-	err = querier.Demux(buf[:n], 0)
+	err = querier.IngressEthernet(buf[:n])
 	if err != nil {
 		t.Fatal("querier demux:", err)
 	}
@@ -382,7 +382,7 @@ func TestMDNS_RealWorldQueries(t *testing.T) {
 	var buf [MTU + ethernet.MaxOverheadSize]byte
 	checkNoData := func(msg string) {
 		t.Helper()
-		n, err := responderStack.Encapsulate(buf[:], -1, 0)
+		n, err := responderStack.EgressEthernet(buf[:])
 		if err != nil {
 			t.Fatal(err)
 		} else if n != 0 {
@@ -434,18 +434,18 @@ func TestMDNS_RealWorldQueries(t *testing.T) {
 		ifrm.CRCWriteUDPPseudo(&crc, ufrm.Length())
 		got := crc.PayloadSum16(ifrm.Payload())
 		ufrm.SetCRC(got)
-		err := responderStack.Demux(buf[:14+20+8+msg.Len()], 0)
+		err := responderStack.IngressEthernet(buf[:14+20+8+msg.Len()])
 		if err != nil {
 			t.Fatal(err)
 		}
 	}
-	n, err := responderStack.Encapsulate(buf[:], -1, 0)
+	n, err := responderStack.EgressEthernet(buf[:])
 	if err != nil {
 		t.Fatal(err)
 	} else if n < 14+20+8+dns.SizeHeader {
 		t.Error("expected response", n)
 	}
-	n, err = responderStack.Encapsulate(buf[:], -1, 0)
+	n, err = responderStack.EgressEthernet(buf[:])
 	if err != nil {
 		t.Fatal(err)
 	} else if n != 0 {

x/xnet/xnet_test.go

@@ -68,7 +68,7 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) {
 	// Perform packet exchange to deliver data.
 	tst.bufmu.Lock()
 	buf := tst.buf[:cap(tst.buf)]
-	n, err := client.Encapsulate(buf, -1, 0)
+	n, err := client.EgressEthernet(buf)
 	if err != nil {
 		tst.bufmu.Unlock()
 		t.Fatal(err)
@@ -77,7 +77,7 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) {
 		tst.bufmu.Unlock()
 		t.Fatal("expected data packet from client")
 	}
-	err = sv.Demux(buf[:n], 0)
+	err = sv.IngressEthernet(buf[:n])
 	tst.bufmu.Unlock()
 	if err != nil {
 		t.Fatal(err)
@@ -418,7 +418,7 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs
 		panic("OOB")
 	}
 
-	n, err := src.Encapsulate(buf[:], -1, 0)
+	n, err := src.EgressEthernet(buf[:])
 	if err != nil {
 		t.Fatal(err)
 	} else if n == 0 {
@@ -466,7 +466,7 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs
 	if seg.Flags != expect.WantFlags {
 		t.Errorf("expected flags %s, got %s", expect.WantFlags.String(), seg.Flags.String())
 	}
-	err = dst.Demux(buf[:n], 0)
+	err = dst.IngressEthernet(buf[:n])
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -481,7 +481,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) {
 	buf := tst.buf[:cap(tst.buf)]
 
 	// === PHASE 1: ARP Request from querying stack ===
-	n, err := querying.Encapsulate(buf[:], -1, 0)
+	n, err := querying.EgressEthernet(buf[:])
 	if err != nil {
 		t.Fatal(err)
 	} else if n == 0 {
@@ -526,7 +526,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) {
 	}
 
 	// Deliver request to target
-	err = target.Demux(buf[:n], 0)
+	err = target.IngressEthernet(buf[:n])
 	if err != nil {
 		t.Fatal("target demux request:", err)
 	}
@@ -534,7 +534,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) {
 
 	// === PHASE 2: ARP Reply from target stack ===
 	buf = tst.buf[:cap(tst.buf)]
-	n, err = target.Encapsulate(buf[:], -1, 0)
+	n, err = target.EgressEthernet(buf[:])
 	if err != nil {
 		t.Fatal(err)
 	} else if n == 0 {
@@ -574,7 +574,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) {
 	}
 
 	// Deliver reply to querying stack
-	err = querying.Demux(buf[:n], 0)
+	err = querying.IngressEthernet(buf[:n])
 	if err != nil {
 		t.Fatal("querying demux reply:", err)
 	}
@@ -748,7 +748,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 	// Server sends DATA packet to client.
 	tst.bufmu.Lock()
 	buf := tst.buf[:cap(tst.buf)]
-	n, err := sv.Encapsulate(buf, -1, 0)
+	n, err := sv.EgressEthernet(buf)
 	if err != nil {
 		tst.bufmu.Unlock()
 		t.Fatal("server encapsulate data:", err)
@@ -757,7 +757,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 		tst.bufmu.Unlock()
 		t.Fatal("expected data packet from server")
 	}
-	err = client.Demux(buf[:n], 0)
+	err = client.IngressEthernet(buf[:n])
 	tst.bufmu.Unlock()
 	if err != nil {
 		t.Fatal("client demux data:", err)
@@ -771,13 +771,13 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 	// Client sends ACK for data.
 	tst.bufmu.Lock()
 	buf = tst.buf[:cap(tst.buf)]
-	n, err = client.Encapsulate(buf, -1, 0)
+	n, err = client.EgressEthernet(buf)
 	if err != nil {
 		tst.bufmu.Unlock()
 		t.Fatal("client encapsulate ACK:", err)
 	}
 	if n > 0 {
-		err = sv.Demux(buf[:n], 0)
+		err = sv.IngressEthernet(buf[:n])
 		if err != nil {
 			tst.bufmu.Unlock()
 			t.Fatal("server demux ACK:", err)
@@ -794,7 +794,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 	// Server sends FIN (enters FIN_WAIT_1).
 	tst.bufmu.Lock()
 	buf = tst.buf[:cap(tst.buf)]
-	n, err = sv.Encapsulate(buf, -1, 0)
+	n, err = sv.EgressEthernet(buf)
 	if err != nil {
 		tst.bufmu.Unlock()
 		t.Fatal("server encapsulate FIN:", err)
@@ -803,7 +803,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 		tst.bufmu.Unlock()
 		t.Fatal("expected FIN packet from server")
 	}
-	err = client.Demux(buf[:n], 0)
+	err = client.IngressEthernet(buf[:n])
 	tst.bufmu.Unlock()
 	if err != nil {
 		t.Fatal("client demux FIN:", err)
@@ -819,13 +819,13 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 	// Client sends ACK for FIN.
 	tst.bufmu.Lock()
 	buf = tst.buf[:cap(tst.buf)]
-	n, err = client.Encapsulate(buf, -1, 0)
+	n, err = client.EgressEthernet(buf)
 	if err != nil {
 		tst.bufmu.Unlock()
 		t.Fatal("client encapsulate ACK:", err)
 	}
 	if n > 0 {
-		err = sv.Demux(buf[:n], 0)
+		err = sv.IngressEthernet(buf[:n])
 		if err != nil {
 			tst.bufmu.Unlock()
 			t.Fatal("server demux ACK:", err)
@@ -846,7 +846,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 	// Client sends FIN (enters LAST_ACK).
 	tst.bufmu.Lock()
 	buf = tst.buf[:cap(tst.buf)]
-	n, err = client.Encapsulate(buf, -1, 0)
+	n, err = client.EgressEthernet(buf)
 	if err != nil {
 		tst.bufmu.Unlock()
 		t.Fatal("client encapsulate FIN:", err)
@@ -855,7 +855,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 		tst.bufmu.Unlock()
 		t.Fatal("expected FIN packet from client")
 	}
-	err = sv.Demux(buf[:n], 0)
+	err = sv.IngressEthernet(buf[:n])
 	tst.bufmu.Unlock()
 	if err != nil {
 		t.Fatal("server demux client FIN:", err)
@@ -871,7 +871,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 	// Server sends final ACK.
 	tst.bufmu.Lock()
 	buf = tst.buf[:cap(tst.buf)]
-	n, err = sv.Encapsulate(buf, -1, 0)
+	n, err = sv.EgressEthernet(buf)
 	if err != nil {
 		tst.bufmu.Unlock()
 		t.Fatal("server encapsulate final ACK:", err)
@@ -880,7 +880,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) {
 		tst.bufmu.Unlock()
 		t.Fatal("expected final ACK from server")
 	}
-	err = client.Demux(buf[:n], 0)
+	err = client.IngressEthernet(buf[:n])
 	tst.bufmu.Unlock()
 	if err != nil {
 		t.Fatal("client demux final ACK:", err)
@@ -944,7 +944,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) {
 		SequenceNumber: 1,
 		Payload:        icmpPayload,
 	})
-	err = stack.Demux(pkt, 0)
+	err = stack.IngressEthernet(pkt)
 	if err != nil {
 		t.Fatalf("valid ICMP echo rejected: %v", err)
 	}
@@ -958,7 +958,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) {
 		Payload:        icmpPayload,
 	})
 	pkt = append(pkt, 0xDE, 0xAD, 0xBE, 0xEF) // Simulate Ethernet FCS.
-	err = stack.Demux(pkt, 0)
+	err = stack.IngressEthernet(pkt)
 	if err != nil {
 		t.Fatalf("valid ICMP with trailing FCS rejected: %v", err)
 	}
@@ -970,7 +970,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) {
 		Payload:        icmpPayload,
 	})
 	pkt[len(pkt)-1] ^= 0xFF // Flip bits in last payload byte to corrupt ICMP checksum.
-	err = stack.Demux(pkt, 0)
+	err = stack.IngressEthernet(pkt)
 	if err == nil {
 		t.Fatal("corrupted ICMP accepted, expected CRC error")
 	}
@@ -983,7 +983,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) {
 	})
 	pkt[len(pkt)-1] ^= 0xFF                   // Corrupt ICMP payload.
 	pkt = append(pkt, 0xDE, 0xAD, 0xBE, 0xEF) // Simulate Ethernet FCS.
-	err = stack.Demux(pkt, 0)
+	err = stack.IngressEthernet(pkt)
 	if err == nil {
 		t.Fatal("corrupted ICMP with FCS accepted, expected CRC error")
 	}