feat(tcpreuse): add options for sharing TCP listeners amongst TCP, WS and WSS transports

config/config.go

@@ -38,6 +38,7 @@ import (
 	"github.com/libp2p/go-libp2p/p2p/protocol/holepunch"
 	"github.com/libp2p/go-libp2p/p2p/protocol/identify"
 	"github.com/libp2p/go-libp2p/p2p/transport/quicreuse"
+	"github.com/libp2p/go-libp2p/p2p/transport/tcpreuse"
 	libp2pwebrtc "github.com/libp2p/go-libp2p/p2p/transport/webrtc"
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -145,6 +146,8 @@ type Config struct {
 	CustomIPv6BlackHoleSuccessCounter bool
 
 	UserFxOptions []fx.Option
+
+	ShareTCPListener bool
 }
 
 func (cfg *Config) makeSwarm(eventBus event.Bus, enableMetrics bool) (*swarm.Swarm, error) {
@@ -289,6 +292,12 @@ func (cfg *Config) addTransports() ([]fx.Option, error) {
 		fx.Provide(func() connmgr.ConnectionGater { return cfg.ConnectionGater }),
 		fx.Provide(func() pnet.PSK { return cfg.PSK }),
 		fx.Provide(func() network.ResourceManager { return cfg.ResourceManager }),
+		fx.Provide(func(gater connmgr.ConnectionGater, rcmgr network.ResourceManager) *tcpreuse.ConnMgr {
+			if !cfg.ShareTCPListener {
+				return nil
+			}
+			return tcpreuse.NewConnMgr(tcpreuse.EnvReuseportVal, gater, rcmgr)
+		}),
 		fx.Provide(func(cm *quicreuse.ConnManager, sw *swarm.Swarm) libp2pwebrtc.ListenUDPFn {
 			hasQuicAddrPortFor := func(network string, laddr *net.UDPAddr) bool {
 				quicAddrPorts := map[string]struct{}{}

libp2p_test.go

@@ -59,7 +59,7 @@ func TestTransportConstructor(t *testing.T) {
 		_ connmgr.ConnectionGater,
 		upgrader transport.Upgrader,
 	) transport.Transport {
-		tpt, err := tcp.NewTCPTransport(upgrader, nil)
+		tpt, err := tcp.NewTCPTransport(upgrader, nil, nil)
 		require.NoError(t, err)
 		return tpt
 	}
@@ -751,3 +751,16 @@ func getTLSConf(t *testing.T, ip net.IP, start, end time.Time) *tls.Config {
 		}},
 	}
 }
+
+func TestSharedTCPAddr(t *testing.T) {
+	h, err := New(
+		ShareTCPListener(),
+		Transport(tcp.NewTCPTransport),
+		Transport(websocket.New),
+		ListenAddrStrings("/ip4/0.0.0.0/tcp/8888"),
+		ListenAddrStrings("/ip4/0.0.0.0/tcp/8888/ws"),
+	)
+	require.NoError(t, err)
+	fmt.Println(h.Addrs())
+	h.Close()
+}

options.go

@@ -643,3 +643,11 @@ func WithFxOption(opts ...fx.Option) Option {
 		return nil
 	}
 }
+
+// ShareTCPListener shares the same listen address between TCP and Websocket transports.
+func ShareTCPListener() Option {
+	return func(cfg *Config) error {
+		cfg.ShareTCPListener = true
+		return nil
+	}
+}

p2p/net/swarm/dial_worker_test.go

@@ -84,7 +84,7 @@ func makeSwarmWithNoListenAddrs(t *testing.T, opts ...Option) *Swarm {
 	upgrader := makeUpgrader(t, s)
 	var tcpOpts []tcp.Option
 	tcpOpts = append(tcpOpts, tcp.DisableReuseport())
-	tcpTransport, err := tcp.NewTCPTransport(upgrader, nil, tcpOpts...)
+	tcpTransport, err := tcp.NewTCPTransport(upgrader, nil, nil, tcpOpts...)
 	require.NoError(t, err)
 	if err := s.AddTransport(tcpTransport); err != nil {
 		t.Fatal(err)

p2p/net/swarm/swarm_addr_test.go

@@ -79,7 +79,7 @@ func TestDialAddressSelection(t *testing.T) {
 	s, err := swarm.NewSwarm("local", nil, eventbus.NewBus())
 	require.NoError(t, err)
 
-	tcpTr, err := tcp.NewTCPTransport(nil, nil)
+	tcpTr, err := tcp.NewTCPTransport(nil, nil, nil)
 	require.NoError(t, err)
 	require.NoError(t, s.AddTransport(tcpTr))
 	reuse, err := quicreuse.NewConnManager(quic.StatelessResetKey{}, quic.TokenGeneratorKey{})

p2p/net/swarm/swarm_dial_test.go

@@ -53,7 +53,7 @@ func TestAddrsForDial(t *testing.T) {
 	ps.AddPrivKey(id, priv)
 	t.Cleanup(func() { ps.Close() })
 
-	tpt, err := websocket.New(nil, &network.NullResourceManager{})
+	tpt, err := websocket.New(nil, &network.NullResourceManager{}, nil)
 	require.NoError(t, err)
 	s, err := NewSwarm(id, ps, eventbus.NewBus(), WithMultiaddrResolver(ResolverFromMaDNS{resolver}))
 	require.NoError(t, err)
@@ -100,7 +100,7 @@ func TestDedupAddrsForDial(t *testing.T) {
 	require.NoError(t, err)
 	defer s.Close()
 
-	tpt, err := tcp.NewTCPTransport(nil, &network.NullResourceManager{})
+	tpt, err := tcp.NewTCPTransport(nil, &network.NullResourceManager{}, nil)
 	require.NoError(t, err)
 	err = s.AddTransport(tpt)
 	require.NoError(t, err)
@@ -134,7 +134,7 @@ func newTestSwarmWithResolver(t *testing.T, resolver *madns.Resolver) *Swarm {
 	})
 
 	// Add a tcp transport so that we know we can dial a tcp multiaddr and we don't filter it out.
-	tpt, err := tcp.NewTCPTransport(nil, &network.NullResourceManager{})
+	tpt, err := tcp.NewTCPTransport(nil, &network.NullResourceManager{}, nil)
 	require.NoError(t, err)
 	err = s.AddTransport(tpt)
 	require.NoError(t, err)
@@ -151,7 +151,7 @@ func newTestSwarmWithResolver(t *testing.T, resolver *madns.Resolver) *Swarm {
 	err = s.AddTransport(wtTpt)
 	require.NoError(t, err)
 
-	wsTpt, err := websocket.New(nil, &network.NullResourceManager{})
+	wsTpt, err := websocket.New(nil, &network.NullResourceManager{}, nil)
 	require.NoError(t, err)
 	err = s.AddTransport(wsTpt)
 	require.NoError(t, err)

p2p/net/swarm/testing/testing.go

@@ -164,7 +164,7 @@ func GenSwarm(t testing.TB, opts ...Option) *swarm.Swarm {
 		if cfg.disableReuseport {
 			tcpOpts = append(tcpOpts, tcp.DisableReuseport())
 		}
-		tcpTransport, err := tcp.NewTCPTransport(upgrader, nil, tcpOpts...)
+		tcpTransport, err := tcp.NewTCPTransport(upgrader, nil, nil, tcpOpts...)
 		require.NoError(t, err)
 		if err := s.AddTransport(tcpTransport); err != nil {
 			t.Fatal(err)

p2p/net/upgrader/listener.go

@@ -84,23 +84,33 @@ func (l *listener) handleIncoming() {
 		}
 		catcher.Reset()
 
-		// gate the connection if applicable
-		if l.upgrader.connGater != nil && !l.upgrader.connGater.InterceptAccept(maconn) {
-			log.Debugf("gater blocked incoming connection on local addr %s from %s",
-				maconn.LocalMultiaddr(), maconn.RemoteMultiaddr())
-			if err := maconn.Close(); err != nil {
-				log.Warnf("failed to close incoming connection rejected by gater: %s", err)
-			}
-			continue
+		// Check if we already have a connection scope. See the comment in tcpreuse/listener.go for an explanation.
+		var connScope network.ConnManagementScope
+		if sc, ok := maconn.(interface {
+			Scope() network.ConnManagementScope
+		}); ok {
+			connScope = sc.Scope()
 		}
+		if connScope == nil {
+			// gate the connection if applicable
+			if l.upgrader.connGater != nil && !l.upgrader.connGater.InterceptAccept(maconn) {
+				log.Debugf("gater blocked incoming connection on local addr %s from %s",
+					maconn.LocalMultiaddr(), maconn.RemoteMultiaddr())
+				if err := maconn.Close(); err != nil {
+					log.Warnf("failed to close incoming connection rejected by gater: %s", err)
+				}
+				continue
+			}
 
-		connScope, err := l.rcmgr.OpenConnection(network.DirInbound, true, maconn.RemoteMultiaddr())
-		if err != nil {
-			log.Debugw("resource manager blocked accept of new connection", "error", err)
-			if err := maconn.Close(); err != nil {
-				log.Warnf("failed to incoming connection rejected by resource manager: %s", err)
+			var err error
+			connScope, err = l.rcmgr.OpenConnection(network.DirInbound, true, maconn.RemoteMultiaddr())
+			if err != nil {
+				log.Debugw("resource manager blocked accept of new connection", "error", err)
+				if err := maconn.Close(); err != nil {
+					log.Warnf("failed to incoming connection rejected by resource manager: %s", err)
+				}
+				continue
 			}
-			continue
 		}
 
 		// The go routine below calls Release when the context is

p2p/protocol/circuitv2/relay/relay_test.go

@@ -60,7 +60,7 @@ func getNetHosts(t *testing.T, ctx context.Context, n int) (hosts []host.Host, u
 		upgrader := swarmt.GenUpgrader(t, netw, nil)
 		upgraders = append(upgraders, upgrader)
 
-		tpt, err := tcp.NewTCPTransport(upgrader, nil)
+		tpt, err := tcp.NewTCPTransport(upgrader, nil, nil)
 		if err != nil {
 			t.Fatal(err)
 		}

p2p/test/transport/gating_test.go

@@ -2,6 +2,8 @@ package transport_integration
 
 import (
 	"context"
+	"encoding/binary"
+	"net/netip"
 	"strings"
 	"testing"
 	"time"
@@ -30,6 +32,23 @@ func stripCertHash(addr ma.Multiaddr) ma.Multiaddr {
 	return addr
 }
 
+func addrPort(addr ma.Multiaddr) netip.AddrPort {
+	a := netip.Addr{}
+	p := uint16(0)
+	ma.ForEach(addr, func(c ma.Component) bool {
+		if c.Protocol().Code == ma.P_IP4 || c.Protocol().Code == ma.P_IP6 {
+			a, _ = netip.AddrFromSlice(c.RawValue())
+			return false
+		}
+		if c.Protocol().Code == ma.P_UDP || c.Protocol().Code == ma.P_TCP {
+			p = binary.BigEndian.Uint16(c.RawValue())
+			return true
+		}
+		return false
+	})
+	return netip.AddrPortFrom(a, p)
+}
+
 func TestInterceptPeerDial(t *testing.T) {
 	if race.WithRace() {
 		t.Skip("The upgrader spawns a new Go routine, which leads to race conditions when using GoMock.")
@@ -173,10 +192,14 @@ func TestInterceptAccept(t *testing.T) {
 					// remove the certhash component from WebTransport addresses
 					require.Equal(t, stripCertHash(h2.Addrs()[0]), addrs.LocalMultiaddr())
 				}).AnyTimes()
+			} else if strings.Contains(tc.Name, "WebSocket-Shared") {
+				connGater.EXPECT().InterceptAccept(gomock.Any()).Do(func(addrs network.ConnMultiaddrs) {
+					require.Equal(t, addrPort(h2.Addrs()[0]), addrPort(addrs.LocalMultiaddr()))
+				})
 			} else {
 				connGater.EXPECT().InterceptAccept(gomock.Any()).Do(func(addrs network.ConnMultiaddrs) {
 					// remove the certhash component from WebTransport addresses
-					require.Equal(t, stripCertHash(h2.Addrs()[0]), addrs.LocalMultiaddr())
+					require.Equal(t, stripCertHash(h2.Addrs()[0]), addrs.LocalMultiaddr(), "%s\n%s", h2.Addrs()[0], addrs.LocalMultiaddr())
 				})
 			}
 

p2p/test/transport/transport_test.go

@@ -99,6 +99,38 @@ var transportsToTest = []TransportTestCase{
 			return h
 		},
 	},
+	{
+		Name: "TCP-Shared / TLS / Yamux",
+		HostGenerator: func(t *testing.T, opts TransportTestCaseOpts) host.Host {
+			libp2pOpts := transformOpts(opts)
+			libp2pOpts = append(libp2pOpts, libp2p.ShareTCPListener())
+			libp2pOpts = append(libp2pOpts, libp2p.Security(tls.ID, tls.New))
+			libp2pOpts = append(libp2pOpts, libp2p.Muxer(yamux.ID, yamux.DefaultTransport))
+			if opts.NoListen {
+				libp2pOpts = append(libp2pOpts, libp2p.NoListenAddrs)
+			} else {
+				libp2pOpts = append(libp2pOpts, libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0"))
+			}
+			h, err := libp2p.New(libp2pOpts...)
+			require.NoError(t, err)
+			return h
+		},
+	},
+	{
+		Name: "WebSocket-Shared",
+		HostGenerator: func(t *testing.T, opts TransportTestCaseOpts) host.Host {
+			libp2pOpts := transformOpts(opts)
+			libp2pOpts = append(libp2pOpts, libp2p.ShareTCPListener())
+			if opts.NoListen {
+				libp2pOpts = append(libp2pOpts, libp2p.NoListenAddrs)
+			} else {
+				libp2pOpts = append(libp2pOpts, libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0/ws"))
+			}
+			h, err := libp2p.New(libp2pOpts...)
+			require.NoError(t, err)
+			return h
+		},
+	},
 	{
 		Name: "WebSocket",
 		HostGenerator: func(t *testing.T, opts TransportTestCaseOpts) host.Host {

p2p/transport/tcp/metrics.go

@@ -24,7 +24,7 @@ var (
 
 const collectFrequency = 10 * time.Second
 
-var collector *aggregatingCollector
+var defaultCollector *aggregatingCollector
 
 var initMetricsOnce sync.Once
 
@@ -34,8 +34,8 @@ func initMetrics() {
 	bytesSentDesc = prometheus.NewDesc("tcp_sent_bytes", "TCP bytes sent", nil, nil)
 	bytesRcvdDesc = prometheus.NewDesc("tcp_rcvd_bytes", "TCP bytes received", nil, nil)
 
-	collector = newAggregatingCollector()
-	prometheus.MustRegister(collector)
+	defaultCollector = newAggregatingCollector()
+	prometheus.MustRegister(defaultCollector)
 
 	const direction = "direction"
 
@@ -196,14 +196,16 @@ func (c *aggregatingCollector) Collect(metrics chan<- prometheus.Metric) {
 
 func (c *aggregatingCollector) ClosedConn(conn *tracingConn, direction string) {
 	c.mutex.Lock()
-	collector.removeConn(conn.id)
+	c.removeConn(conn.id)
 	c.mutex.Unlock()
 	closedConns.WithLabelValues(direction).Inc()
 }
 
 type tracingConn struct {
 	id uint64
 
+	collector *aggregatingCollector
+
 	startTime time.Time
 	isClient  bool
 
@@ -213,7 +215,8 @@ type tracingConn struct {
 	closeErr  error
 }
 
-func newTracingConn(c manet.Conn, isClient bool) (*tracingConn, error) {
+// newTracingConn wraps a manet.Conn with a tracingConn. A nil collector will use the default collector.
+func newTracingConn(c manet.Conn, collector *aggregatingCollector, isClient bool) (*tracingConn, error) {
 	initMetricsOnce.Do(func() { initMetrics() })
 	conn, err := tcp.NewConn(c)
 	if err != nil {
@@ -224,8 +227,12 @@ func newTracingConn(c manet.Conn, isClient bool) (*tracingConn, error) {
 		isClient:  isClient,
 		Conn:      c,
 		tcpConn:   conn,
+		collector: collector,
+	}
+	if tc.collector == nil {
+		tc.collector = defaultCollector
 	}
-	tc.id = collector.AddConn(tc)
+	tc.id = tc.collector.AddConn(tc)
 	newConns.WithLabelValues(tc.getDirection()).Inc()
 	return tc, nil
 }
@@ -239,7 +246,7 @@ func (c *tracingConn) getDirection() string {
 
 func (c *tracingConn) Close() error {
 	c.closeOnce.Do(func() {
-		collector.ClosedConn(c, c.getDirection())
+		c.collector.ClosedConn(c, c.getDirection())
 		c.closeErr = c.Conn.Close()
 	})
 	return c.closeErr
@@ -258,16 +265,18 @@ func (c *tracingConn) getTCPInfo() (*tcpinfo.Info, error) {
 
 type tracingListener struct {
 	manet.Listener
+	collector *aggregatingCollector
 }
 
-func newTracingListener(l manet.Listener) *tracingListener {
-	return &tracingListener{Listener: l}
+// newTracingListener wraps a manet.Listener with a tracingListener. A nil collector will use the default collector.
+func newTracingListener(l manet.Listener, collector *aggregatingCollector) *tracingListener {
+	return &tracingListener{Listener: l, collector: collector}
 }
 
 func (l *tracingListener) Accept() (manet.Conn, error) {
 	conn, err := l.Listener.Accept()
 	if err != nil {
 		return nil, err
 	}
-	return newTracingConn(conn, false)
+	return newTracingConn(conn, l.collector, false)
 }

p2p/transport/tcp/metrics_none.go

@@ -6,5 +6,9 @@ package tcp
 
 import manet "github.com/multiformats/go-multiaddr/net"
 
-func newTracingConn(c manet.Conn, _ bool) (manet.Conn, error) { return c, nil }
-func newTracingListener(l manet.Listener) manet.Listener      { return l }
+type aggregatingCollector struct{}
+
+func newTracingConn(c manet.Conn, collector *aggregatingCollector, isClient bool) (manet.Conn, error) {
+	return c, nil
+}
+func newTracingListener(l manet.Listener, collector *aggregatingCollector) manet.Listener { return l }