add support for the resource manager

go.mod

@@ -5,14 +5,14 @@ go 1.16
 require (
 	github.com/ipfs/go-log/v2 v2.4.0
 	github.com/libp2p/go-conn-security-multistream v0.3.0
-	github.com/libp2p/go-libp2p-core v0.13.1-0.20220104083644-a3dd401efe36
+	github.com/libp2p/go-libp2p-core v0.13.1-0.20220107083545-005b84b557bb
 	github.com/libp2p/go-libp2p-peerstore v0.6.0
-	github.com/libp2p/go-libp2p-quic-transport v0.13.0
-	github.com/libp2p/go-libp2p-testing v0.5.0
-	github.com/libp2p/go-libp2p-transport-upgrader v0.6.1-0.20220104084635-5fc0a74b41f0
+	github.com/libp2p/go-libp2p-quic-transport v0.15.1-0.20220108115653-809bf14dd0e9
+	github.com/libp2p/go-libp2p-testing v0.6.1-0.20220107070227-f126df46085e
+	github.com/libp2p/go-libp2p-transport-upgrader v0.6.1-0.20220108163839-722a883f45d4
 	github.com/libp2p/go-libp2p-yamux v0.5.0
 	github.com/libp2p/go-stream-muxer-multistream v0.3.0
-	github.com/libp2p/go-tcp-transport v0.4.1-0.20220104085503-4ad75e6f32a5
+	github.com/libp2p/go-tcp-transport v0.4.1-0.20220108170023-5b0f844d771b
 	github.com/multiformats/go-multiaddr v0.5.0
 	github.com/multiformats/go-multiaddr-fmt v0.1.0
 	github.com/stretchr/testify v1.7.0

go.sum

@@ -276,24 +276,27 @@ github.com/libp2p/go-libp2p-core v0.5.1/go.mod h1:uN7L2D4EvPCvzSH5SrhR72UWbnSGpt
 github.com/libp2p/go-libp2p-core v0.8.0/go.mod h1:FfewUH/YpvWbEB+ZY9AQRQ4TAD8sJBt/G1rVvhz5XT8=
 github.com/libp2p/go-libp2p-core v0.10.0/go.mod h1:ECdxehoYosLYHgDDFa2N4yE8Y7aQRAMf0sX9mf2sbGg=
 github.com/libp2p/go-libp2p-core v0.12.0/go.mod h1:ECdxehoYosLYHgDDFa2N4yE8Y7aQRAMf0sX9mf2sbGg=
-github.com/libp2p/go-libp2p-core v0.13.1-0.20220104083644-a3dd401efe36 h1:b/pMmgc5EV+dqSc+MjkX5xPa1nV6EKiOb0L0XT03Lic=
-github.com/libp2p/go-libp2p-core v0.13.1-0.20220104083644-a3dd401efe36/go.mod h1:KlkHsZ0nKerWsXLZJm3LfFQwusI5k3iN4BgtYTE4IYE=
+github.com/libp2p/go-libp2p-core v0.13.1-0.20220106080354-2192774acd82/go.mod h1:KlkHsZ0nKerWsXLZJm3LfFQwusI5k3iN4BgtYTE4IYE=
+github.com/libp2p/go-libp2p-core v0.13.1-0.20220107083545-005b84b557bb h1:mMWwfcMQgaUY7AN2WKd+L7k0XnTkmvx159KNvHPg2oo=
+github.com/libp2p/go-libp2p-core v0.13.1-0.20220107083545-005b84b557bb/go.mod h1:KlkHsZ0nKerWsXLZJm3LfFQwusI5k3iN4BgtYTE4IYE=
 github.com/libp2p/go-libp2p-mplex v0.4.1 h1:/pyhkP1nLwjG3OM+VuaNJkQT/Pqq73WzB3aDN3Fx1sc=
 github.com/libp2p/go-libp2p-mplex v0.4.1/go.mod h1:cmy+3GfqfM1PceHTLL7zQzAAYaryDu6iPSC+CIb094g=
 github.com/libp2p/go-libp2p-peerstore v0.6.0 h1:HJminhQSGISBIRb93N6WK3t6Fa8OOTnHd/VBjL4mY5A=
 github.com/libp2p/go-libp2p-peerstore v0.6.0/go.mod h1:DGEmKdXrcYpK9Jha3sS7MhqYdInxJy84bIPtSu65bKc=
 github.com/libp2p/go-libp2p-pnet v0.2.0 h1:J6htxttBipJujEjz1y0a5+eYoiPcFHhSYHH6na5f0/k=
 github.com/libp2p/go-libp2p-pnet v0.2.0/go.mod h1:Qqvq6JH/oMZGwqs3N1Fqhv8NVhrdYcO0BW4wssv21LA=
-github.com/libp2p/go-libp2p-quic-transport v0.13.0 h1:MTVojS4AnGD/rng6rF/HXEqwMHL27rHUEf3DaqSdnUw=
-github.com/libp2p/go-libp2p-quic-transport v0.13.0/go.mod h1:39/ZWJ1TW/jx1iFkKzzUg00W6tDJh73FC0xYudjr7Hc=
+github.com/libp2p/go-libp2p-quic-transport v0.15.1-0.20220108115653-809bf14dd0e9 h1:vKncWRP2UbfShrpqEdF7a1pkKlU0mdqPFY0bo3q0LuQ=
+github.com/libp2p/go-libp2p-quic-transport v0.15.1-0.20220108115653-809bf14dd0e9/go.mod h1:QgXb66laBD2IR9Jz0nbiK22cd+ywp+6D5TlPS29mNFs=
 github.com/libp2p/go-libp2p-testing v0.1.2-0.20200422005655-8775583591d8/go.mod h1:Qy8sAncLKpwXtS2dSnDOP8ktexIAHKu+J+pnZOFZLTc=
 github.com/libp2p/go-libp2p-testing v0.4.0/go.mod h1:Q+PFXYoiYFN5CAEG2w3gLPEzotlKsNSbKQ/lImlOWF0=
-github.com/libp2p/go-libp2p-testing v0.5.0 h1:bTjC29TTQ/ODq0ld3+0KLq3irdA5cAH3OMbRi0/QsvE=
 github.com/libp2p/go-libp2p-testing v0.5.0/go.mod h1:QBk8fqIL1XNcno/l3/hhaIEn4aLRijpYOR+zVjjlh+A=
+github.com/libp2p/go-libp2p-testing v0.6.1-0.20220107070227-f126df46085e h1:gy++yXkcKv6pC6sgGPJRi0CkWPdgYFAx6zOWGxNiORg=
+github.com/libp2p/go-libp2p-testing v0.6.1-0.20220107070227-f126df46085e/go.mod h1:sKGFutxeBvU1FajPDMzOdNeRBTujG3IvX1Q1+xb5L38=
 github.com/libp2p/go-libp2p-tls v0.3.0 h1:8BgvUJiOTcj0Gp6XvEicF0rL5aUtRg/UzEdeZDmDlC8=
 github.com/libp2p/go-libp2p-tls v0.3.0/go.mod h1:fwF5X6PWGxm6IDRwF3V8AVCCj/hOd5oFlg+wo2FxJDY=
-github.com/libp2p/go-libp2p-transport-upgrader v0.6.1-0.20220104084635-5fc0a74b41f0 h1:eD/QJCpcImYOUl6MdBuxMByVaEe5VMm463zJG6oUg9o=
-github.com/libp2p/go-libp2p-transport-upgrader v0.6.1-0.20220104084635-5fc0a74b41f0/go.mod h1:ByIyNe8asQhgcyIHetb4f+UgV+hDrA8pQ3L/TgNs+RI=
+github.com/libp2p/go-libp2p-transport-upgrader v0.6.1-0.20220108150807-be66941c1ea9/go.mod h1:FpQXSlueDNp9POKfwEnKqqgtI9mXy3ibpe7Z3ueBros=
+github.com/libp2p/go-libp2p-transport-upgrader v0.6.1-0.20220108163839-722a883f45d4 h1:1jc/z9tiidFBmkufCAjD9YcxD02+5YwYvXmNWzGahOw=
+github.com/libp2p/go-libp2p-transport-upgrader v0.6.1-0.20220108163839-722a883f45d4/go.mod h1:FpQXSlueDNp9POKfwEnKqqgtI9mXy3ibpe7Z3ueBros=
 github.com/libp2p/go-libp2p-yamux v0.5.0 h1:ZzmUhbQE+X7NuYUT2naxN31JyebZfRmpZVhKtRP13ys=
 github.com/libp2p/go-libp2p-yamux v0.5.0/go.mod h1:AyR8k5EzyM2QN9Bbdg6X1SkVVuqLwTGf0L4DFq9g6po=
 github.com/libp2p/go-maddr-filter v0.1.0/go.mod h1:VzZhTXkMucEGGEOSKddrwGiOv0tUhgnKqNEmIAz/bPU=
@@ -317,14 +320,14 @@ github.com/libp2p/go-sockaddr v0.1.0 h1:Y4s3/jNoryVRKEBrkJ576F17CPOaMIzUeCsg7dlT
 github.com/libp2p/go-sockaddr v0.1.0/go.mod h1:syPvOmNs24S3dFVGJA1/mrqdeijPxLV2Le3BRLKd68k=
 github.com/libp2p/go-stream-muxer-multistream v0.3.0 h1:TqnSHPJEIqDEO7h1wZZ0p3DXdvDSiLHQidKKUGZtiOY=
 github.com/libp2p/go-stream-muxer-multistream v0.3.0/go.mod h1:yDh8abSIzmZtqtOt64gFJUXEryejzNb0lisTt+fAMJA=
-github.com/libp2p/go-tcp-transport v0.4.1-0.20220104085503-4ad75e6f32a5 h1:/x3GSszKipn1nlKY0C5at59fBLYyJeObd5gm32DrobM=
-github.com/libp2p/go-tcp-transport v0.4.1-0.20220104085503-4ad75e6f32a5/go.mod h1:YPwlF5gW5BnFikKoQBuJeQkPXAn+z2wTzDpJKamkgjY=
+github.com/libp2p/go-tcp-transport v0.4.1-0.20220108170023-5b0f844d771b h1:KWpzjWlTdMN8KlREoEgWbYLPPhouy7fMh5xXttgYk0M=
+github.com/libp2p/go-tcp-transport v0.4.1-0.20220108170023-5b0f844d771b/go.mod h1:Rhw3Zrx0b5eR9BZ1cHH7NP28PdcHbBwUcMRwJTGacrs=
 github.com/libp2p/go-yamux v1.4.1 h1:P1Fe9vF4th5JOxxgQvfbOHkrGqIZniTLf+ddhZp8YTI=
 github.com/libp2p/go-yamux v1.4.1/go.mod h1:fr7aVgmdNGJK+N1g+b6DW6VxzbRCjCOejR/hkmpooHE=
 github.com/lightstep/lightstep-tracer-common/golang/gogo v0.0.0-20190605223551-bc2310a04743/go.mod h1:qklhhLq1aX+mtWk9cPHPzaBjWImj5ULL6C7HFJtXQMM=
 github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4=
-github.com/lucas-clemente/quic-go v0.23.0 h1:5vFnKtZ6nHDFsc/F3uuiF4T3y/AXaQdxjUqiVw26GZE=
-github.com/lucas-clemente/quic-go v0.23.0/go.mod h1:paZuzjXCE5mj6sikVLMvqXk8lJV2AsqtJ6bDhjEfxx0=
+github.com/lucas-clemente/quic-go v0.24.0 h1:ToR7SIIEdrgOhgVTHvPgdVRJfgVy+N0wQAagH7L4d5g=
+github.com/lucas-clemente/quic-go v0.24.0/go.mod h1:paZuzjXCE5mj6sikVLMvqXk8lJV2AsqtJ6bDhjEfxx0=
 github.com/lunixbochs/vtclean v1.0.0/go.mod h1:pHhQNgMf3btfWnGBVipUOjRYhoOsdGqdm/+2c2E2WMI=
 github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
@@ -740,7 +743,6 @@ golang.org/x/sys v0.0.0-20210317225723-c4fcb01b228e/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20210511113859-b0526f3d8744/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=

swarm.go

@@ -76,6 +76,13 @@ func WithDialTimeoutLocal(t time.Duration) Option {
 	}
 }
 
+func WithResourceManager(m network.ResourceManager) Option {
+	return func(s *Swarm) error {
+		s.rcmgr = m
+		return nil
+	}
+}
+
 // Swarm is a connection muxer, allowing connections to other peers to
 // be opened and closed, while still using the same Chan for all
 // communication. The Chan sends/receives Messages, which note the
@@ -88,6 +95,8 @@ type Swarm struct {
 	// down before continuing.
 	refs sync.WaitGroup
 
+	rcmgr network.ResourceManager
+
 	local peer.ID
 	peers peerstore.Peerstore
 
@@ -156,6 +165,9 @@ func NewSwarm(local peer.ID, peers peerstore.Peerstore, opts ...Option) (*Swarm,
 			return nil, err
 		}
 	}
+	if s.rcmgr == nil {
+		s.rcmgr = network.NullResourceManager
+	}
 
 	s.dsync = newDialSync(s.dialWorkerLoop)
 	s.limiter = newDialLimiter(s.dialAddr)
@@ -586,6 +598,10 @@ func (s *Swarm) String() string {
 	return fmt.Sprintf("<Swarm %s>", s.LocalPeer())
 }
 
+func (s *Swarm) ResourceManager() network.ResourceManager {
+	return s.rcmgr
+}
+
 // Swarm is a Network.
 var _ network.Network = (*Swarm)(nil)
 var _ transport.TransportNetwork = (*Swarm)(nil)

swarm_conn.go

@@ -42,6 +42,8 @@ type Conn struct {
 	stat network.ConnStats
 }
 
+var _ network.Conn = &Conn{}
+
 func (c *Conn) ID() string {
 	// format: <first 10 chars of peer id>-<global conn ordinal>
 	return fmt.Sprintf("%s-%d", c.RemotePeer().Pretty()[0:10], c.id)
@@ -93,6 +95,7 @@ func (c *Conn) removeStream(s *Stream) {
 	c.stat.NumStreams--
 	delete(c.streams.m, s)
 	c.streams.Unlock()
+	s.scope.Done()
 }
 
 // listens for new streams.
@@ -109,9 +112,14 @@ func (c *Conn) start() {
 			if err != nil {
 				return
 			}
+			scope, err := c.swarm.ResourceManager().OpenStream(c.RemotePeer(), network.DirInbound)
+			if err != nil {
+				ts.Reset()
+				continue
+			}
 			c.swarm.refs.Add(1)
 			go func() {
-				s, err := c.addStream(ts, network.DirInbound)
+				s, err := c.addStream(ts, network.DirInbound, scope)
 
 				// Don't defer this. We don't want to block
 				// swarm shutdown on the connection handler.
@@ -186,19 +194,23 @@ func (c *Conn) NewStream(ctx context.Context) (network.Stream, error) {
 		}
 	}
 
+	scope, err := c.swarm.ResourceManager().OpenStream(c.RemotePeer(), network.DirOutbound)
+	if err != nil {
+		return nil, err
+	}
 	ts, err := c.conn.OpenStream(ctx)
-
 	if err != nil {
 		return nil, err
 	}
-	return c.addStream(ts, network.DirOutbound)
+	return c.addStream(ts, network.DirOutbound, scope)
 }
 
-func (c *Conn) addStream(ts mux.MuxedStream, dir network.Direction) (*Stream, error) {
+func (c *Conn) addStream(ts mux.MuxedStream, dir network.Direction, scope network.StreamManagementScope) (*Stream, error) {
 	c.streams.Lock()
 	// Are we still online?
 	if c.streams.m == nil {
 		c.streams.Unlock()
+		scope.Done()
 		ts.Reset()
 		return nil, ErrConnClosed
 	}
@@ -207,6 +219,7 @@ func (c *Conn) addStream(ts mux.MuxedStream, dir network.Direction) (*Stream, er
 	s := &Stream{
 		stream: ts,
 		conn:   c,
+		scope:  scope,
 		stat: network.Stats{
 			Direction: dir,
 			Opened:    time.Now(),
@@ -244,3 +257,7 @@ func (c *Conn) GetStreams() []network.Stream {
 	}
 	return streams
 }
+
+func (c *Conn) Scope() network.ConnScope {
+	return c.conn.Scope()
+}

swarm_stream.go

@@ -21,6 +21,7 @@ type Stream struct {
 
 	stream mux.MuxedStream
 	conn   *Conn
+	scope  network.StreamManagementScope
 
 	closeOnce sync.Once
 
@@ -154,3 +155,7 @@ func (s *Stream) SetWriteDeadline(t time.Time) error {
 func (s *Stream) Stat() network.Stats {
 	return s.stat
 }
+
+func (s *Stream) Scope() network.StreamScope {
+	return s.scope
+}

testing/testing.go

@@ -13,6 +13,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peerstore"
 	"github.com/libp2p/go-libp2p-core/sec/insecure"
 	"github.com/libp2p/go-libp2p-core/transport"
+	"github.com/libp2p/go-tcp-transport"
 
 	csms "github.com/libp2p/go-conn-security-multistream"
 	"github.com/libp2p/go-libp2p-peerstore/pstoremem"
@@ -22,8 +23,6 @@ import (
 	tptu "github.com/libp2p/go-libp2p-transport-upgrader"
 	yamux "github.com/libp2p/go-libp2p-yamux"
 	msmux "github.com/libp2p/go-stream-muxer-multistream"
-	"github.com/libp2p/go-tcp-transport"
-
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/stretchr/testify/require"
 )
@@ -90,7 +89,7 @@ func GenUpgrader(t *testing.T, n *swarm.Swarm, opts ...tptu.Option) transport.Up
 
 	stMuxer := msmux.NewBlankTransport()
 	stMuxer.AddTransport("/yamux/1.0.0", yamux.DefaultTransport)
-	u, err := tptu.New(secMuxer, stMuxer, opts...)
+	u, err := tptu.New(secMuxer, stMuxer, nil, opts...)
 	require.NoError(t, err)
 	return u
 }
@@ -140,7 +139,7 @@ func GenSwarm(t *testing.T, opts ...Option) *swarm.Swarm {
 		if cfg.disableReuseport {
 			tcpOpts = append(tcpOpts, tcp.DisableReuseport())
 		}
-		tcpTransport, err := tcp.NewTCPTransport(upgrader, tcpOpts...)
+		tcpTransport, err := tcp.NewTCPTransport(upgrader, nil, tcpOpts...)
 		require.NoError(t, err)
 		if err := s.AddTransport(tcpTransport); err != nil {
 			t.Fatal(err)
@@ -152,7 +151,7 @@ func GenSwarm(t *testing.T, opts ...Option) *swarm.Swarm {
 		}
 	}
 	if !cfg.disableQUIC {
-		quicTransport, err := quic.NewTransport(p.PrivKey, nil, cfg.connectionGater)
+		quicTransport, err := quic.NewTransport(p.PrivKey, nil, cfg.connectionGater, nil)
 		if err != nil {
 			t.Fatal(err)
 		}