perf: bound idle TCP reuse storage

agent-todo.md

@@ -205,7 +205,15 @@ but host-facing operations call only immediate `tcp.Handler` state/buffer method
 under the namespace lock; `tcp.Conn.Read`, `Write`, and `Flush` remain absent.
 Connect/accept, partial I/O, EOF/reset semantics, half-close, policy, exact
 resource/retained-storage quota, bounded readiness, port reuse, and abort cleanup
-are covered. Accepted-stream close releases resource quota immediately; lneto's
+are covered. Closed listener/outbound bytes are zeroed, and idle reuse retains
+at most one listener pool capped at 256 slots and 1 MiB plus one outbound
+buffer/released-accounting slot capped at 1 MiB; excess high-water storage is
+dropped after quota release. The accounting reuse lowers steady connect/close
+allocation from 1,128 to 936 B/op (17.0%) while `tcp.Conn` remains deliberately
+generation-distinct because lneto may retain its pointer after abort.
+Adapter creation now uses small listener/stream registry hints and no longer
+allocates registries or reuse-index arrays proportional to maximum listener or
+outbound counts. Accepted-stream close releases resource quota immediately; lneto's
 private accepted list is preserved until the next bounded egress service probe,
 which reclaims the pool slot and now reports one charged maintenance operation
 even when it emits no frame. DNS uses adapter-owned immediate IPv4 UDP packets plus lneto DNS codecs,

docs/architecture.md

@@ -174,9 +174,18 @@ It uses only immediate `tcp.Handler` buffer/state primitives and never calls
 pools and
 outbound streams have bounded receive/transmit storage, partial I/O, connect and
 accept progress, half-close, level readiness, endpoint policy, quota ownership,
-port reuse, and deterministic abort cleanup. Adapter creation seeds only a
-small stream-registry capacity hint and grows that registry as streams are
-actually created rather than preallocating for the full theoretical
+port reuse, and deterministic abort cleanup. Closed listener and outbound
+buffers are zeroed before reuse or release. The outbound buffer cache also owns
+one released/reset quota charge, reducing steady wrapper garbage while each
+`tcp.Conn` remains generation-distinct because lneto may retain its registration
+pointer after abort. The adapter retains at most one idle listener pool of no
+more than 256 slots and 1 MiB of storage plus one idle outbound buffer/accounting
+slot of no more than 1 MiB; concurrent high-water buffers and larger
+configurations are dropped after close rather than remaining as uncharged cache.
+Adapter creation seeds only small listener/stream registry capacity hints and no
+longer allocates registries or reuse-index arrays proportional to configured
+listener or outbound limits. It grows the registries as resources are actually
+created rather than preallocating for the full theoretical
 `MaxOutboundStreams + MaxListeners*AcceptBacklog` population. Closing an
 accepted stream releases its resource quota immediately. lneto retains the
 closed pool entry until its listener performs maintenance; the next bounded

internal/backend/lneto/tcp/benchmark_test.go

@@ -38,6 +38,25 @@ func BenchmarkAdapterNew(b *testing.B) {
 	}
 }
 
+func BenchmarkAdapterNewMaximumListenerConfig(b *testing.B) {
+	config := Config{MaxListeners: ^uint16(0), AcceptBacklog: 1, ReceiveBytes: 256, TransmitBytes: 256, TransmitPackets: 4}
+	b.ReportAllocs()
+	for b.Loop() {
+		common := newConfigTestCore(b, config.MaxListeners)
+		adapter, err := New(common, config)
+		if err != nil {
+			common.Close()
+			b.Fatal(err)
+		}
+		if err := common.Close(); err != nil {
+			b.Fatal(err)
+		}
+		if adapter == nil {
+			b.Fatal("nil adapter")
+		}
+	}
+}
+
 func BenchmarkAdapterTryListenClose(b *testing.B) {
 	_, adapter := newTestAdapter(b, 111, 1, 0)
 	local := nscore.Endpoint{Address: netip.MustParseAddr("192.0.2.111"), Port: 4211}

internal/backend/lneto/tcp/tcp.go

@@ -24,6 +24,8 @@ const (
 	ingressOrder                    = 15
 	closeOrder                      = 20
 	maxEagerTCPListenerStorageBytes = 256 << 20
+	maxIdleTCPReuseStorageBytes     = 1 << 20
+	maxIdleTCPListenerSlots         = 256
 	maxTCPStreamCapacityHint        = 16
 )
 
@@ -42,10 +44,10 @@ type Adapter struct {
 	quotas              *quota.Account
 	config              Config
 	listeners           []*tcpListener
-	freeListenerPools   []tcpPool
+	freeListenerPool    tcpPool
 	streams             []*tcpStream
 	outboundStreams     int
-	freeOutboundStorage [][]byte
+	freeOutboundStorage *tcpOutboundStorage
 	portOwner           *lnetocore.TCPPortOwner
 	nextISS             lnetotcp.Value
 }
@@ -95,9 +97,8 @@ func New(common *lnetocore.Namespace, config Config) (*Adapter, error) {
 		common.Unlock()
 		return n, nil
 	}
-	n.listeners = make([]*tcpListener, 0, config.MaxListeners)
+	n.listeners = make([]*tcpListener, 0, listenerCapacityHint(config))
 	n.streams = make([]*tcpStream, 0, streamCapacityHint(config))
-	n.prepareReusePools()
 	common.Unlock()
 	if err := common.Install(lnetocore.Participant{IngressOrder: ingressOrder, Ingress: n.ingressLocked, CloseOrder: closeOrder, Close: n.CloseLocked}); err != nil {
 		return nil, err
@@ -167,40 +168,35 @@ func tcpStreamStorageBytes(config Config) (uint64, bool) {
 	return checked.AddUint64(uint64(config.ReceiveBytes), uint64(config.TransmitBytes))
 }
 
-func streamCapacityHint(config Config) int {
-	hint := uint64(config.MaxListeners) + uint64(config.MaxOutboundStreams)
+func listenerCapacityHint(config Config) int {
+	hint := uint64(config.MaxListeners)
 	if hint > maxTCPStreamCapacityHint {
 		hint = maxTCPStreamCapacityHint
 	}
 	return int(hint)
 }
 
-func (n *Adapter) prepareReusePools() {
-	if n == nil {
-		return
-	}
-	if n.config.MaxListeners > 0 {
-		n.freeListenerPools = make([]tcpPool, 0, n.config.MaxListeners)
-	}
-	if n.config.MaxOutboundStreams > 0 {
-		n.freeOutboundStorage = make([][]byte, 0, n.config.MaxOutboundStreams)
+func streamCapacityHint(config Config) int {
+	hint := uint64(config.MaxListeners) + uint64(config.MaxOutboundStreams)
+	if hint > maxTCPStreamCapacityHint {
+		hint = maxTCPStreamCapacityHint
 	}
+	return int(hint)
 }
 
 func (n *Adapter) acquireListenerLocked(local nscore.Endpoint) (*tcpListener, error) {
 	if n == nil {
 		return nil, lneto.ErrInvalidConfig
 	}
-	var pool tcpPool
-	if len(n.freeListenerPools) == 0 {
+	pool := n.freeListenerPool
+	n.freeListenerPool = tcpPool{}
+	if pool.slots == nil {
 		created, err := newTCPPool(n, n.config.AcceptBacklog, n.config)
 		if err != nil {
 			return nil, err
 		}
 		pool = created
 	} else {
-		pool = n.freeListenerPools[len(n.freeListenerPools)-1]
-		n.freeListenerPools = n.freeListenerPools[:len(n.freeListenerPools)-1]
 		pool.resetLocked(n)
 	}
 	return &tcpListener{owner: n, local: local, pool: pool}, nil
@@ -211,7 +207,11 @@ func (n *Adapter) recycleListenerLocked(listener *tcpListener) {
 		return
 	}
 	listener.pool.releaseLocked()
-	n.freeListenerPools = append(n.freeListenerPools, listener.pool)
+	if n.freeListenerPool.slots == nil && len(listener.pool.slots) <= maxIdleTCPListenerSlots && len(listener.pool.storage) <= maxIdleTCPReuseStorageBytes {
+		n.freeListenerPool = listener.pool
+	} else {
+		listener.pool.destroyLocked()
+	}
 	listener.pool = tcpPool{}
 	listener.listener = lnetotcp.Listener{}
 	listener.local = nscore.Endpoint{}
@@ -224,34 +224,61 @@ func (n *Adapter) prepareOutboundStreamLocked(stream *tcpStream, retained uint64
 	if n == nil || stream == nil || stream.owner != n || stream.portLease.TCPPort() == 0 {
 		return lneto.ErrInvalidConfig
 	}
-	if err := n.quotas.AcquireResourceAndQueuedBytes(&stream.retained, quota.ResourceTCP, 1, retained); err != nil {
+	storage := n.freeOutboundStorage
+	reused := storage != nil
+	n.freeOutboundStorage = nil
+	if storage == nil {
+		storage = new(tcpOutboundStorage)
+	}
+	stream.outboundStorage = storage
+	if err := n.quotas.AcquireResourceAndQueuedBytes(&storage.retained, quota.ResourceTCP, 1, retained); err != nil {
+		storage.retained = quota.Charge{}
+		if reused {
+			n.freeOutboundStorage = storage
+		}
+		stream.outboundStorage = nil
 		return err
 	}
-	stream.allocation = &stream.retained
-	if len(n.freeOutboundStorage) == 0 {
-		stream.storage = make([]byte, int(retained))
-	} else {
-		stream.storage = n.freeOutboundStorage[len(n.freeOutboundStorage)-1]
-		n.freeOutboundStorage = n.freeOutboundStorage[:len(n.freeOutboundStorage)-1]
+	if storage.bytes == nil {
+		storage.bytes = make([]byte, int(retained))
 	}
+	stream.conn = &stream.connValue
+	stream.storage = storage.bytes
+	stream.allocation = &storage.retained
 	return nil
 }
 
+func (n *Adapter) recycleOutboundStorageLocked(storage *tcpOutboundStorage) {
+	if n == nil || storage == nil {
+		return
+	}
+	storage.retained = quota.Charge{}
+	clear(storage.bytes)
+	if n.freeOutboundStorage == nil && len(storage.bytes) <= maxIdleTCPReuseStorageBytes {
+		n.freeOutboundStorage = storage
+	} else {
+		storage.bytes = nil
+	}
+}
+
 func (n *Adapter) recycleOutboundStreamLocked(stream *tcpStream) {
 	if n == nil || stream == nil {
 		return
 	}
-	clear(stream.storage)
-	n.freeOutboundStorage = append(n.freeOutboundStorage, stream.storage)
+	if stream.outboundStorage != nil {
+		n.recycleOutboundStorageLocked(stream.outboundStorage)
+	} else {
+		clear(stream.storage)
+	}
 	stream.conn = nil
 	stream.connValue = lnetotcp.Conn{}
 	stream.storage = nil
+	stream.outboundStorage = nil
 	stream.local = nscore.Endpoint{}
 	stream.remote = nscore.Endpoint{}
 	stream.portLease = lnetocore.TCPPortLease{}
 	stream.slot = nil
 	stream.allocation = nil
-	stream.retained = quota.Charge{}
 	stream.connected = false
 	stream.shutdown = false
 	stream.terminal = false
@@ -270,17 +297,22 @@ type tcpListener struct {
 	closed    bool
 }
 
+type tcpOutboundStorage struct {
+	retained quota.Charge
+	bytes    []byte
+}
+
 type tcpStream struct {
-	owner     *Adapter
-	conn      *lnetotcp.Conn
-	connValue lnetotcp.Conn
-	storage   []byte
-	local     nscore.Endpoint
-	remote    nscore.Endpoint
-	slot      *tcpPoolSlot
+	owner           *Adapter
+	conn            *lnetotcp.Conn
+	connValue       lnetotcp.Conn
+	storage         []byte
+	outboundStorage *tcpOutboundStorage
+	local           nscore.Endpoint
+	remote          nscore.Endpoint
+	slot            *tcpPoolSlot
 
 	allocation *quota.Charge
-	retained   quota.Charge
 	portLease  lnetocore.TCPPortLease
 	connected  bool
 	shutdown   bool
@@ -292,6 +324,7 @@ type tcpStream struct {
 type tcpPool struct {
 	owner   *Adapter
 	slots   []tcpPoolSlot
+	storage []byte
 	nextISS lnetotcp.Value
 }
 
@@ -311,14 +344,14 @@ func newTCPPool(owner *Adapter, count uint16, config Config) (tcpPool, error) {
 		return pool, nil
 	}
 	stride, _ := tcpStreamStorageBytes(config)
-	storage := make([]byte, int(uint64(count)*stride))
+	pool.storage = make([]byte, int(uint64(count)*stride))
 	strideBytes := int(stride)
 	for i := range pool.slots {
 		start := i * strideBytes
 		rxEnd := start + config.ReceiveBytes
 		if err := pool.slots[i].conn.Configure(lnetotcp.ConnConfig{
-			RxBuf:             storage[start:rxEnd],
-			TxBuf:             storage[rxEnd : start+strideBytes],
+			RxBuf:             pool.storage[start:rxEnd],
+			TxBuf:             pool.storage[rxEnd : start+strideBytes],
 			TxPacketQueueSize: config.TransmitPackets,
 			RWBackoff:         immediateBackoff,
 		}); err != nil {
@@ -412,6 +445,7 @@ func (p *tcpPool) releaseLocked() {
 		slot.inUse = false
 		slot.quotaOwned = false
 	}
+	clear(p.storage)
 }
 
 func (p *tcpPool) destroyLocked() {
@@ -420,6 +454,7 @@ func (p *tcpPool) destroyLocked() {
 	}
 	p.releaseLocked()
 	p.slots = nil
+	p.storage = nil
 	p.owner = nil
 }
 
@@ -685,23 +720,23 @@ func (n *Adapter) TryConnectAuthorized(remote nscore.Endpoint, authorize Connect
 		stream.portLease.ReleaseLocked()
 		return nil, 0, lnetocore.MapError(err)
 	}
-	conn := &stream.connValue
+	conn := stream.conn
 	if err := conn.Configure(lnetotcp.ConnConfig{
 		RxBuf:             stream.storage[:n.config.ReceiveBytes],
 		TxBuf:             stream.storage[n.config.ReceiveBytes:],
 		TxPacketQueueSize: n.config.TransmitPackets,
 		RWBackoff:         immediateBackoff,
 	}); err != nil {
 		stream.allocation.Release()
-		stream.retained.ResetReleased()
+		stream.outboundStorage.retained.ResetReleased()
 		stream.portLease.ReleaseLocked()
 		n.recycleOutboundStreamLocked(stream)
 		return nil, 0, lnetocore.MapError(err)
 	}
 	if err := n.stack.DialTCP(conn, localPort, netip.AddrPortFrom(remote.Address, remote.Port)); err != nil {
 		conn.Abort()
 		stream.allocation.Release()
-		stream.retained.ResetReleased()
+		stream.outboundStorage.retained.ResetReleased()
 		stream.portLease.ReleaseLocked()
 		n.recycleOutboundStreamLocked(stream)
 		return nil, 0, lnetocore.MapError(err)
@@ -1045,8 +1080,8 @@ func (s *tcpStream) closeLocked() error {
 	}
 	if s.allocation != nil {
 		s.allocation.Release()
-		if s.allocation == &s.retained {
-			s.retained.ResetReleased()
+		if s.outboundStorage != nil && s.allocation == &s.outboundStorage.retained {
+			s.outboundStorage.retained.ResetReleased()
 		}
 		s.allocation = nil
 	}
@@ -1114,14 +1149,13 @@ func (n *Adapter) CloseLocked() {
 	for len(n.streams) > 0 {
 		n.streams[len(n.streams)-1].closeLocked()
 	}
-	for i := range n.freeListenerPools {
-		n.freeListenerPools[i].destroyLocked()
-	}
-	for i := range n.freeOutboundStorage {
-		clear(n.freeOutboundStorage[i])
+	n.freeListenerPool.destroyLocked()
+	if n.freeOutboundStorage != nil {
+		n.recycleOutboundStorageLocked(n.freeOutboundStorage)
+		n.freeOutboundStorage = nil
 	}
 	n.portOwner = nil
-	n.freeListenerPools = nil
+	n.freeListenerPool = tcpPool{}
 	n.listeners = nil
 	n.freeOutboundStorage = nil
 	n.streams = nil