Commit 9910e6a7 authored by Jeromy's avatar Jeromy
Browse files

switching to separated conn and interface-conn packages

parent a54a624e
...@@ -7,11 +7,11 @@ import ( ...@@ -7,11 +7,11 @@ import (
peer "github.com/ipfs/go-libp2p-peer" peer "github.com/ipfs/go-libp2p-peer"
ma "github.com/jbenet/go-multiaddr" ma "github.com/jbenet/go-multiaddr"
addrutil "github.com/libp2p/go-addr-util" addrutil "github.com/libp2p/go-addr-util"
conn "github.com/libp2p/go-libp2p-conn" iconn "github.com/libp2p/go-libp2p-interface-conn"
) )
type dialResult struct { type dialResult struct {
Conn conn.Conn Conn iconn.Conn
Err error Err error
} }
...@@ -38,14 +38,14 @@ type dialLimiter struct { ...@@ -38,14 +38,14 @@ type dialLimiter struct {
fdLimit int fdLimit int
waitingOnFd []*dialJob waitingOnFd []*dialJob
dialFunc func(context.Context, peer.ID, ma.Multiaddr) (conn.Conn, error) dialFunc func(context.Context, peer.ID, ma.Multiaddr) (iconn.Conn, error)
activePerPeer map[peer.ID]int activePerPeer map[peer.ID]int
perPeerLimit int perPeerLimit int
waitingOnPeerLimit map[peer.ID][]*dialJob waitingOnPeerLimit map[peer.ID][]*dialJob
} }
type dialfunc func(context.Context, peer.ID, ma.Multiaddr) (conn.Conn, error) type dialfunc func(context.Context, peer.ID, ma.Multiaddr) (iconn.Conn, error)
func newDialLimiter(df dialfunc) *dialLimiter { func newDialLimiter(df dialfunc) *dialLimiter {
return newDialLimiterWithParams(df, concurrentFdDials, defaultPerPeerRateLimit) return newDialLimiterWithParams(df, concurrentFdDials, defaultPerPeerRateLimit)
......
...@@ -10,7 +10,7 @@ import ( ...@@ -10,7 +10,7 @@ import (
peer "github.com/ipfs/go-libp2p-peer" peer "github.com/ipfs/go-libp2p-peer"
ma "github.com/jbenet/go-multiaddr" ma "github.com/jbenet/go-multiaddr"
conn "github.com/libp2p/go-libp2p-conn" iconn "github.com/libp2p/go-libp2p-interface-conn"
mafmt "github.com/whyrusleeping/mafmt" mafmt "github.com/whyrusleeping/mafmt"
) )
...@@ -55,13 +55,13 @@ func tryDialAddrs(ctx context.Context, l *dialLimiter, p peer.ID, addrs []ma.Mul ...@@ -55,13 +55,13 @@ func tryDialAddrs(ctx context.Context, l *dialLimiter, p peer.ID, addrs []ma.Mul
} }
func hangDialFunc(hang chan struct{}) dialfunc { func hangDialFunc(hang chan struct{}) dialfunc {
return func(ctx context.Context, p peer.ID, a ma.Multiaddr) (conn.Conn, error) { return func(ctx context.Context, p peer.ID, a ma.Multiaddr) (iconn.Conn, error) {
if mafmt.UTP.Matches(a) { if mafmt.UTP.Matches(a) {
return conn.Conn(nil), nil return iconn.Conn(nil), nil
} }
if tcpPortOver(a, 10) { if tcpPortOver(a, 10) {
return conn.Conn(nil), nil return iconn.Conn(nil), nil
} }
<-hang <-hang
...@@ -171,9 +171,9 @@ func TestFDLimiting(t *testing.T) { ...@@ -171,9 +171,9 @@ func TestFDLimiting(t *testing.T) {
func TestTokenRedistribution(t *testing.T) { func TestTokenRedistribution(t *testing.T) {
hangchs := make(map[peer.ID]chan struct{}) hangchs := make(map[peer.ID]chan struct{})
df := func(ctx context.Context, p peer.ID, a ma.Multiaddr) (conn.Conn, error) { df := func(ctx context.Context, p peer.ID, a ma.Multiaddr) (iconn.Conn, error) {
if tcpPortOver(a, 10) { if tcpPortOver(a, 10) {
return (conn.Conn)(nil), nil return (iconn.Conn)(nil), nil
} }
<-hangchs[p] <-hangchs[p]
...@@ -260,9 +260,9 @@ func TestTokenRedistribution(t *testing.T) { ...@@ -260,9 +260,9 @@ func TestTokenRedistribution(t *testing.T) {
} }
func TestStressLimiter(t *testing.T) { func TestStressLimiter(t *testing.T) {
df := func(ctx context.Context, p peer.ID, a ma.Multiaddr) (conn.Conn, error) { df := func(ctx context.Context, p peer.ID, a ma.Multiaddr) (iconn.Conn, error) {
if tcpPortOver(a, 1000) { if tcpPortOver(a, 1000) {
return conn.Conn(nil), nil return iconn.Conn(nil), nil
} }
time.Sleep(time.Millisecond * time.Duration(5+rand.Intn(100))) time.Sleep(time.Millisecond * time.Duration(5+rand.Intn(100)))
......
...@@ -3,7 +3,7 @@ package swarm ...@@ -3,7 +3,7 @@ package swarm
import ( import (
ma "github.com/jbenet/go-multiaddr" ma "github.com/jbenet/go-multiaddr"
addrutil "github.com/libp2p/go-addr-util" addrutil "github.com/libp2p/go-addr-util"
conn "github.com/libp2p/go-libp2p-conn" iconn "github.com/libp2p/go-libp2p-interface-conn"
) )
// ListenAddresses returns a list of addresses at which this swarm listens. // ListenAddresses returns a list of addresses at which this swarm listens.
...@@ -11,7 +11,7 @@ func (s *Swarm) ListenAddresses() []ma.Multiaddr { ...@@ -11,7 +11,7 @@ func (s *Swarm) ListenAddresses() []ma.Multiaddr {
listeners := s.swarm.Listeners() listeners := s.swarm.Listeners()
addrs := make([]ma.Multiaddr, 0, len(listeners)) addrs := make([]ma.Multiaddr, 0, len(listeners))
for _, l := range listeners { for _, l := range listeners {
if l2, ok := l.NetListener().(conn.Listener); ok { if l2, ok := l.NetListener().(iconn.Listener); ok {
addrs = append(addrs, l2.Multiaddr()) addrs = append(addrs, l2.Multiaddr())
} }
} }
......
...@@ -4,13 +4,12 @@ import ( ...@@ -4,13 +4,12 @@ import (
"context" "context"
"fmt" "fmt"
inet "github.com/libp2p/go-libp2p-net"
ic "github.com/ipfs/go-libp2p-crypto" ic "github.com/ipfs/go-libp2p-crypto"
peer "github.com/ipfs/go-libp2p-peer" peer "github.com/ipfs/go-libp2p-peer"
ma "github.com/jbenet/go-multiaddr" ma "github.com/jbenet/go-multiaddr"
ps "github.com/jbenet/go-peerstream" ps "github.com/jbenet/go-peerstream"
conn "github.com/libp2p/go-libp2p-conn" iconn "github.com/libp2p/go-libp2p-interface-conn"
inet "github.com/libp2p/go-libp2p-net"
) )
// Conn is a simple wrapper around a ps.Conn that also exposes // Conn is a simple wrapper around a ps.Conn that also exposes
...@@ -33,12 +32,12 @@ func (c *Conn) StreamConn() *ps.Conn { ...@@ -33,12 +32,12 @@ func (c *Conn) StreamConn() *ps.Conn {
return (*ps.Conn)(c) return (*ps.Conn)(c)
} }
func (c *Conn) RawConn() conn.Conn { func (c *Conn) RawConn() iconn.Conn {
// righly panic if these things aren't true. it is an expected // righly panic if these things aren't true. it is an expected
// invariant that these Conns are all of the typewe expect: // invariant that these Conns are all of the typewe expect:
// ps.Conn wrapping a conn.Conn // ps.Conn wrapping a conn.Conn
// if we get something else it is programmer error. // if we get something else it is programmer error.
return (*ps.Conn)(c).NetConn().(conn.Conn) return (*ps.Conn)(c).NetConn().(iconn.Conn)
} }
func (c *Conn) String() string { func (c *Conn) String() string {
...@@ -94,7 +93,7 @@ func (c *Conn) Close() error { ...@@ -94,7 +93,7 @@ func (c *Conn) Close() error {
func wrapConn(psc *ps.Conn) (*Conn, error) { func wrapConn(psc *ps.Conn) (*Conn, error) {
// grab the underlying connection. // grab the underlying connection.
if _, ok := psc.NetConn().(conn.Conn); !ok { if _, ok := psc.NetConn().(iconn.Conn); !ok {
// this should never happen. if we see it ocurring it means that we added // this should never happen. if we see it ocurring it means that we added
// a Listener to the ps.Swarm that is NOT one of our net/conn.Listener. // a Listener to the ps.Swarm that is NOT one of our net/conn.Listener.
return nil, fmt.Errorf("swarm connHandler: invalid conn (not a conn.Conn): %s", psc) return nil, fmt.Errorf("swarm connHandler: invalid conn (not a conn.Conn): %s", psc)
......
...@@ -11,7 +11,7 @@ import ( ...@@ -11,7 +11,7 @@ import (
peer "github.com/ipfs/go-libp2p-peer" peer "github.com/ipfs/go-libp2p-peer"
ma "github.com/jbenet/go-multiaddr" ma "github.com/jbenet/go-multiaddr"
addrutil "github.com/libp2p/go-addr-util" addrutil "github.com/libp2p/go-addr-util"
conn "github.com/libp2p/go-libp2p-conn" iconn "github.com/libp2p/go-libp2p-interface-conn"
) )
// Diagram of dial sync: // Diagram of dial sync:
...@@ -284,7 +284,7 @@ func (s *Swarm) dial(ctx context.Context, p peer.ID) (*Conn, error) { ...@@ -284,7 +284,7 @@ func (s *Swarm) dial(ctx context.Context, p peer.ID) (*Conn, error) {
return swarmC, nil return swarmC, nil
} }
func (s *Swarm) dialAddrs(ctx context.Context, p peer.ID, remoteAddrs <-chan ma.Multiaddr) (conn.Conn, error) { func (s *Swarm) dialAddrs(ctx context.Context, p peer.ID, remoteAddrs <-chan ma.Multiaddr) (iconn.Conn, error) {
log.Debugf("%s swarm dialing %s %s", s.local, p, remoteAddrs) log.Debugf("%s swarm dialing %s %s", s.local, p, remoteAddrs)
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
...@@ -344,7 +344,7 @@ func (s *Swarm) limitedDial(ctx context.Context, p peer.ID, a ma.Multiaddr, resp ...@@ -344,7 +344,7 @@ func (s *Swarm) limitedDial(ctx context.Context, p peer.ID, a ma.Multiaddr, resp
}) })
} }
func (s *Swarm) dialAddr(ctx context.Context, p peer.ID, addr ma.Multiaddr) (conn.Conn, error) { func (s *Swarm) dialAddr(ctx context.Context, p peer.ID, addr ma.Multiaddr) (iconn.Conn, error) {
log.Debugf("%s swarm dialing %s %s", s.local, p, addr) log.Debugf("%s swarm dialing %s %s", s.local, p, addr)
connC, err := s.dialer.Dial(ctx, addr, p) connC, err := s.dialer.Dial(ctx, addr, p)
...@@ -376,7 +376,7 @@ var ConnSetupTimeout = time.Minute * 5 ...@@ -376,7 +376,7 @@ var ConnSetupTimeout = time.Minute * 5
// dialConnSetup is the setup logic for a connection from the dial side. it // dialConnSetup is the setup logic for a connection from the dial side. it
// needs to add the Conn to the StreamSwarm, then run newConnSetup // needs to add the Conn to the StreamSwarm, then run newConnSetup
func dialConnSetup(ctx context.Context, s *Swarm, connC conn.Conn) (*Conn, error) { func dialConnSetup(ctx context.Context, s *Swarm, connC iconn.Conn) (*Conn, error) {
deadline, ok := ctx.Deadline() deadline, ok := ctx.Deadline()
if !ok { if !ok {
......
...@@ -8,6 +8,7 @@ import ( ...@@ -8,6 +8,7 @@ import (
ma "github.com/jbenet/go-multiaddr" ma "github.com/jbenet/go-multiaddr"
ps "github.com/jbenet/go-peerstream" ps "github.com/jbenet/go-peerstream"
conn "github.com/libp2p/go-libp2p-conn" conn "github.com/libp2p/go-libp2p-conn"
iconn "github.com/libp2p/go-libp2p-interface-conn"
mconn "github.com/libp2p/go-libp2p-metrics/conn" mconn "github.com/libp2p/go-libp2p-metrics/conn"
inet "github.com/libp2p/go-libp2p-net" inet "github.com/libp2p/go-libp2p-net"
transport "github.com/libp2p/go-libp2p-transport" transport "github.com/libp2p/go-libp2p-transport"
...@@ -98,7 +99,7 @@ func (s *Swarm) addListener(tptlist transport.Listener) error { ...@@ -98,7 +99,7 @@ func (s *Swarm) addListener(tptlist transport.Listener) error {
return s.addConnListener(list) return s.addConnListener(list)
} }
func (s *Swarm) addConnListener(list conn.Listener) error { func (s *Swarm) addConnListener(list iconn.Listener) error {
// AddListener to the peerstream Listener. this will begin accepting connections // AddListener to the peerstream Listener. this will begin accepting connections
// and streams! // and streams!
sl, err := s.swarm.AddListener(list) sl, err := s.swarm.AddListener(list)
......
...@@ -215,21 +215,27 @@ ...@@ -215,21 +215,27 @@
}, },
{ {
"author": "whyrusleeping", "author": "whyrusleeping",
"hash": "QmTaW4q1AbqMkpfDLUYzW18nW62GsrnFvtVcvR1pnaURm6", "hash": "Qmd4eGc7VMmVZu225mdCzsagdWJBaAVHmmvNFYSvxSUeX5",
"name": "go-libp2p-conn", "name": "go-libp2p-conn",
"version": "1.0.0" "version": "1.1.0"
}, },
{ {
"author": "whyrusleeping", "author": "whyrusleeping",
"hash": "QmSctCnUNkE7c7C2LGRGYrdmU9SDH3MtbAeueN7Jq1NN2q", "hash": "Qmcr6YS5QCPwejhBxZk5iFEvVEHSVzewbLVMyc64zefaUf",
"name": "go-libp2p-net", "name": "go-libp2p-net",
"version": "1.0.0" "version": "1.0.0"
}, },
{ {
"author": "whyrusleeping", "author": "whyrusleeping",
"hash": "QmS38TTyj47fP5NVNznxpw4KpFYSU9zHkCEfmuBSkHzb6d", "hash": "QmbidKjoAxLVE2skY74Rfw1QafReyf4w2FZqA8Sd7MxoVd",
"name": "go-libp2p-metrics", "name": "go-libp2p-metrics",
"version": "1.0.0" "version": "1.1.0"
},
{
"author": "whyrusleeping",
"hash": "QmRxzoGdQaN6HYyqWnT82NnLLHBAZbTUvxPEfTBTjU7KCn",
"name": "go-libp2p-interface-conn",
"version": "0.1.0"
} }
], ],
"gxVersion": "0.4.0", "gxVersion": "0.4.0",
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment