@cryptotaxi247 / kubo / commits / 6bc26f175

writing swarm2 using go-peerstream

omg wow such pass

Juan Batiz-Benet committed Dec 15, 2014 at 14:42 UTC 6bc26f175242c5e84a7a82ef6f0b984418ab3623
8 files changed +686 -66
net/conn/handshake.go
+2 -2
@@ -2,13 +2,13 @@ package conn
2
3 import (
4 "fmt"
5 + "io"
6
7 handshake "github.com/jbenet/go-ipfs/net/handshake"
8 hspb "github.com/jbenet/go-ipfs/net/handshake/pb"
9
10 ggprotoio "code.google.com/p/gogoprotobuf/io"
11 context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
11 - ps "github.com/jbenet/go-peerstream"
12 )
13
14 // Handshake1 exchanges local and remote versions and compares them
@@ -53,7 +53,7 @@ func Handshake1(ctx context.Context, c Conn) error {
53 }
54
55 // Handshake3 exchanges local and remote service information
56 -func Handshake3(ctx context.Context, stream ps.Stream, c Conn) (*handshake.Handshake3Result, error) {
56 +func Handshake3(ctx context.Context, stream io.ReadWriter, c Conn) (*handshake.Handshake3Result, error) {
57 rpeer := c.RemotePeer()
58 lpeer := c.LocalPeer()
59
net/swarm2/simul_test.go new
+80
@@ -0,0 +1,80 @@
1 +package swarm
2 +
3 +import (
4 + "fmt"
5 + "sync"
6 + "testing"
7 +
8 + peer "github.com/jbenet/go-ipfs/peer"
9 + "github.com/jbenet/go-ipfs/util/testutil"
10 +
11 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
12 +)
13 +
14 +func TestSimultOpen(t *testing.T) {
15 + t.Skip("skipping for another test")
16 +
17 + addrs := []string{
18 + "/ip4/127.0.0.1/tcp/1244",
19 + "/ip4/127.0.0.1/tcp/1245",
20 + }
21 +
22 + ctx := context.Background()
23 + swarms, _ := makeSwarms(ctx, t, addrs)
24 +
25 + // connect everyone
26 + {
27 + var wg sync.WaitGroup
28 + connect := func(s *Swarm, dst peer.Peer) {
29 + // copy for other peer
30 + cp := testutil.NewPeerWithID(dst.ID())
31 + cp.AddAddress(dst.Addresses()[0])
32 +
33 + if _, err := s.Dial(cp); err != nil {
34 + t.Fatal("error swarm dialing to peer", err)
35 + }
36 + wg.Done()
37 + }
38 +
39 + log.Info("Connecting swarms simultaneously.")
40 + wg.Add(2)
41 + go connect(swarms[0], swarms[1].local)
42 + go connect(swarms[1], swarms[0].local)
43 + wg.Wait()
44 + }
45 +
46 + for _, s := range swarms {
47 + s.Close()
48 + }
49 +}
50 +
51 +func TestSimultOpenMany(t *testing.T) {
52 + t.Skip("very very slow")
53 +
54 + many := 500
55 + addrs := []string{}
56 + for i := 2200; i < (2200 + many); i++ {
57 + s := fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", i)
58 + addrs = append(addrs, s)
59 + }
60 +
61 + SubtestSwarm(t, addrs, 10)
62 +}
63 +
64 +func TestSimultOpenFewStress(t *testing.T) {
65 + if testing.Short() {
66 + t.SkipNow()
67 + }
68 + t.Skip("skipping for another test")
69 +
70 + num := 10
71 + // num := 100
72 + for i := 0; i < num; i++ {
73 + addrs := []string{
74 + fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", 1900+i),
75 + fmt.Sprintf("/ip4/127.0.0.1/tcp/%d", 2900+i),
76 + }
77 +
78 + SubtestSwarm(t, addrs, 10)
79 + }
80 +}
net/swarm2/swarm.go
+31 -15
@@ -3,7 +3,6 @@
3 package swarm
4
5 import (
6 - conn "github.com/jbenet/go-ipfs/net/conn"
6 peer "github.com/jbenet/go-ipfs/peer"
7 eventlog "github.com/jbenet/go-ipfs/util/eventlog"
8
@@ -51,22 +50,43 @@ func (s *Swarm) teardown() error {
50 return s.swarm.Close()
51 }
52
53 +// Close stops the Swarm. See
54 func (s *Swarm) Close() error {
55 return s.cg.Close()
56 }
57
58 +// StreamSwarm returns the underlying peerstream.Swarm
59 func (s *Swarm) StreamSwarm() *ps.Swarm {
60 return s.swarm
61 }
62
63 +// SetStreamHandler assigns the handler for new streams.
64 +// See peerstream.
65 +func (s *Swarm) SetStreamHandler(handler StreamHandler) {
66 + s.swarm.SetStreamHandler(func(s *ps.Stream) {
67 + handler(wrapStream(s))
68 + })
69 +}
70 +
71 +// NewStreamWithPeer creates a new stream on any available connection to p
72 +func (s *Swarm) NewStreamWithPeer(p peer.Peer) (*Stream, error) {
73 + st, err := s.swarm.NewStreamWithGroup(p)
74 + return wrapStream(st), err
75 +}
76 +
77 +// StreamsWithPeer returns all the live Streams to p
78 +func (s *Swarm) StreamsWithPeer(p peer.Peer) []*Stream {
79 + return wrapStreams(ps.StreamsWithGroup(p, s.swarm.Streams()))
80 +}
81 +
82 +// ConnectionsToPeer returns all the live connections to p
83 +func (s *Swarm) ConnectionsToPeer(p peer.Peer) []*SwarmConn {
84 + return wrapConns(ps.ConnsWithGroup(p, s.swarm.Conns()))
85 +}
86 +
87 // Connections returns a slice of all connections.
63 -func (s *Swarm) Connections() []conn.Conn {
64 - conns1 := s.swarm.Conns()
65 - conns2 := make([]conn.Conn, len(conns1))
66 - for i, c1 := range conns1 {
67 - conns2[i] = UnwrapConn(c1)
68 - }
69 - return conns2
88 +func (s *Swarm) Connections() []*SwarmConn {
89 + return wrapConns(s.swarm.Conns())
90 }
91
92 // CloseConnection removes a given peer from swarm + closes the connection
@@ -80,16 +100,16 @@ func (s *Swarm) CloseConnection(p peer.Peer) error {
100
101 // GetPeerList returns a copy of the set of peers swarm is connected to.
102 func (s *Swarm) GetPeerList() []peer.Peer {
83 - conns := s.swarm.Conns()
103 + conns := s.Connections()
104
105 seen := make(map[peer.Peer]struct{})
106 peers := make([]peer.Peer, 0, len(conns))
107 for _, c := range conns {
88 - c2 := UnwrapConn(c)
89 - p := c2.RemotePeer()
108 + p := c.RemotePeer()
109 if _, found := seen[p]; found {
110 continue
111 }
112 +
113 peers = append(peers, p)
114 }
115 return peers
@@ -99,7 +119,3 @@ func (s *Swarm) GetPeerList() []peer.Peer {
119 func (s *Swarm) LocalPeer() peer.Peer {
120 return s.local
121 }
102 -
103 -func UnwrapConn(c *ps.Conn) conn.Conn {
104 - return c.NetConn().(conn.Conn)
105 -}
net/swarm2/swarm_conn.go new
+141
@@ -0,0 +1,141 @@
1 +package swarm
2 +
3 +import (
4 + "fmt"
5 +
6 + conn "github.com/jbenet/go-ipfs/net/conn"
7 + peer "github.com/jbenet/go-ipfs/peer"
8 +
9 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10 + ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
11 + ps "github.com/jbenet/go-peerstream"
12 +)
13 +
14 +// a SwarmConn is a simple wrapper around a ps.Conn that also exposes
15 +// some of the methods from the underlying conn.Conn.
16 +// There's **five** "layers" to each connection:
17 +// - 0. the net.Conn - underlying net.Conn (TCP/UDP/UTP/etc)
18 +// - 1. the manet.Conn - provides multiaddr friendly Conn
19 +// - 2. the conn.Conn - provides Peer friendly Conn (inc Secure channel)
20 +// - 3. the peerstream.Conn - provides peerstream / spdysptream happiness
21 +// - 4. the SwarmConn - abstracts everyting out, exposing only key parts of underlying layers
22 +// (I know, this is kinda crazy. it's more historical than a good design. though the
23 +// layers do build up pieces of functionality. and they're all just io.RW :) )
24 +type SwarmConn ps.Conn
25 +
26 +func (c *SwarmConn) StreamConn() *ps.Conn {
27 + return (*ps.Conn)(c)
28 +}
29 +
30 +func (c *SwarmConn) RawConn() conn.Conn {
31 + // righly panic if these things aren't true. it is an expected
32 + // invariant that these Conns are all of the typewe expect:
33 + // ps.Conn wrapping a conn.Conn
34 + // if we get something else it is programmer error.
35 + return (*ps.Conn)(c).NetConn().(conn.Conn)
36 +}
37 +
38 +// LocalMultiaddr is the Multiaddr on this side
39 +func (c *SwarmConn) LocalMultiaddr() ma.Multiaddr {
40 + return c.RawConn().LocalMultiaddr()
41 +}
42 +
43 +// LocalPeer is the Peer on our side of the connection
44 +func (c *SwarmConn) LocalPeer() peer.Peer {
45 + return c.RawConn().LocalPeer()
46 +}
47 +
48 +// RemoteMultiaddr is the Multiaddr on the remote side
49 +func (c *SwarmConn) RemoteMultiaddr() ma.Multiaddr {
50 + return c.RawConn().RemoteMultiaddr()
51 +}
52 +
53 +// RemotePeer is the Peer on the remote side
54 +func (c *SwarmConn) RemotePeer() peer.Peer {
55 + return c.RawConn().RemotePeer()
56 +}
57 +
58 +// NewStream returns a new Stream from this connection
59 +func (c *SwarmConn) NewStream() (*Stream, error) {
60 + s, err := c.StreamConn().NewStream()
61 + return wrapStream(s), err
62 +}
63 +
64 +func (c *SwarmConn) Close() error {
65 + return c.StreamConn().Close()
66 +}
67 +
68 +func wrapConn(psc *ps.Conn) (*SwarmConn, error) {
69 + // grab the underlying connection.
70 + if _, ok := psc.NetConn().(conn.Conn); !ok {
71 + // this should never happen. if we see it ocurring it means that we added
72 + // a Listener to the ps.Swarm that is NOT one of our net/conn.Listener.
73 + return nil, fmt.Errorf("swarm connHandler: invalid conn (not a conn.Conn): %s", psc)
74 + }
75 + return (*SwarmConn)(psc), nil
76 +}
77 +
78 +// wrapConns returns a *SwarmConn for all these ps.Conns
79 +func wrapConns(conns1 []*ps.Conn) []*SwarmConn {
80 + conns2 := make([]*SwarmConn, len(conns1))
81 + for i, c1 := range conns1 {
82 + if c2, err := wrapConn(c1); err == nil {
83 + conns2[i] = c2
84 + }
85 + }
86 + return conns2
87 +}
88 +
89 +// newConnSetup does the swarm's "setup" for a connection. returns the underlying
90 +// conn.Conn this method is used by both swarm.Dial and ps.Swarm connHandler
91 +func (s *Swarm) newConnSetup(ctx context.Context, psConn *ps.Conn) (*SwarmConn, error) {
92 +
93 + // wrap with a SwarmConn
94 + sc, err := wrapConn(psConn)
95 + if err != nil {
96 + return nil, err
97 + }
98 +
99 + // removing this for now, as it has to change. we can put this in a different
100 + // sub-protocol anyway.
101 + // // run Handshake3
102 + // if err := runHandshake3(ctx, s, sc); err != nil {
103 + // return nil, err
104 + // }
105 +
106 + // ok great! we can use it. add it to our group.
107 +
108 + // set the RemotePeer as a group on the conn. this lets us group
109 + // connections in the StreamSwarm by peer, and get a streams from
110 + // any available connection in the group (better multiconn):
111 + // swarm.StreamSwarm().NewStreamWithGroup(remotePeer)
112 + psConn.AddGroup(sc.RemotePeer())
113 +
114 + return sc, nil
115 +}
116 +
117 +// func runHandshake3(ctx context.Context, s *Swarm, c *SwarmConn) error {
118 +// log.Event(ctx, "newConnection", c.LocalPeer(), c.RemotePeer())
119 +
120 +// stream, err := c.NewStream()
121 +// if err != nil {
122 +// return err
123 +// }
124 +
125 +// // handshake3 (this whole thing is ugly. maybe lets get rid of it...)
126 +// h3result, err := conn.Handshake3(ctx, stream, c.RawConn())
127 +// if err != nil {
128 +// return fmt.Errorf("Handshake3 failed: %s", err)
129 +// }
130 +
131 +// // check for nats. you know, just in case.
132 +// if h3result.LocalObservedAddress != nil {
133 +// checkNATWarning(s, h3result.LocalObservedAddress, c.LocalMultiaddr())
134 +// } else {
135 +// log.Warningf("Received nil observed address from %s", c.RemotePeer())
136 +// }
137 +
138 +// stream.Close()
139 +// log.Event(ctx, "handshake3Succeeded", c.LocalPeer(), c.RemotePeer())
140 +// return nil
141 +// }
net/swarm2/swarm_dial.go new
+98
@@ -0,0 +1,98 @@
1 +package swarm
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 +
7 + conn "github.com/jbenet/go-ipfs/net/conn"
8 + peer "github.com/jbenet/go-ipfs/peer"
9 + lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
10 +
11 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
12 +)
13 +
14 +// Dial connects to a peer.
15 +//
16 +// The idea is that the client of Swarm does not need to know what network
17 +// the connection will happen over. Swarm can use whichever it choses.
18 +// This allows us to use various transport protocols, do NAT traversal/relay,
19 +// etc. to achive connection.
20 +func (s *Swarm) Dial(p peer.Peer) (*SwarmConn, error) {
21 + ctx := context.TODO()
22 +
23 + if p.ID().Equal(s.local.ID()) {
24 + return nil, errors.New("Attempted connection to self!")
25 + }
26 +
27 + // check if we already have an open connection first
28 + cs := s.ConnectionsToPeer(p)
29 + for _, c := range cs {
30 + if c != nil { // dump out the first one we find
31 + return c, nil
32 + }
33 + }
34 +
35 + // check if we don't have the peer in Peerstore
36 + p, err := s.peers.Add(p)
37 + if err != nil {
38 + return nil, err
39 + }
40 +
41 + // open connection to peer
42 + d := &conn.Dialer{
43 + LocalPeer: s.local,
44 + Peerstore: s.peers,
45 + }
46 +
47 + if len(p.Addresses()) == 0 {
48 + return nil, errors.New("peer has no addresses")
49 + }
50 +
51 + // try to connect to one of the peer's known addresses.
52 + // for simplicity, we do this sequentially.
53 + // A future commit will do this asynchronously.
54 + var connC conn.Conn
55 + for _, addr := range p.Addresses() {
56 + connC, err = d.DialAddr(ctx, addr, p)
57 + if err == nil {
58 + break
59 + }
60 + }
61 + if err != nil {
62 + return nil, err
63 + }
64 +
65 + // ok try to setup the new connection.
66 + swarmC, err := dialConnSetup(ctx, s, connC)
67 + if err != nil {
68 + log.Error("Dial newConnSetup failed. disconnecting.")
69 + log.Event(ctx, "dialFailureDisconnect", lgbl.NetConn(connC), lgbl.Error(err))
70 + swarmC.Close() // close the connection. didn't work out :(
71 + return nil, err
72 + }
73 +
74 + log.Event(ctx, "dial", p)
75 + return swarmC, nil
76 +}
77 +
78 +// dialConnSetup is the setup logic for a connection from the dial side. it
79 +// needs to add the Conn to the StreamSwarm, then run newConnSetup
80 +func dialConnSetup(ctx context.Context, s *Swarm, connC conn.Conn) (*SwarmConn, error) {
81 +
82 + psC, err := s.swarm.AddConn(connC)
83 + if err != nil {
84 + // connC is closed by caller if we fail.
85 + return nil, fmt.Errorf("failed to add conn to ps.Swarm: %s", err)
86 + }
87 +
88 + // ok try to setup the new connection. (newConnSetup will add to group)
89 + swarmC, err := s.newConnSetup(ctx, psC)
90 + if err != nil {
91 + log.Error("Dial newConnSetup failed. disconnecting.")
92 + log.Event(ctx, "dialFailureDisconnect", lgbl.NetConn(connC), lgbl.Error(err))
93 + swarmC.Close() // we need to call this to make sure psC is Closed.
94 + return nil, err
95 + }
96 +
97 + return swarmC, err
98 +}
net/swarm2/swarm_listen.go
+18 -49
@@ -1,9 +1,8 @@
1 package swarm
2
3 import (
4 - "fmt"
5 -
4 conn "github.com/jbenet/go-ipfs/net/conn"
5 + lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
6
7 context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
8 ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
@@ -58,51 +57,21 @@ func (s *Swarm) setupListener(maddr ma.Multiaddr) error {
57 // here we configure it slightly. Note that this is sequential, so if anything
58 // will take a while do it in a goroutine.
59 // See https://godoc.org/github.com/jbenet/go-peerstream for more information
61 -func (s *Swarm) connHandler(c1 *ps.Conn) {
62 -
63 - // grab the underlying connection.
64 - if c2, ok := c1.NetConn().(conn.Conn); ok {
65 -
66 - // set the RemotePeer as a group on the conn. this lets us group
67 - // connections in the StreamSwarm by peer, and get a streams from
68 - // any available connection in the group (better multiconn):
69 - // swarm.StreamSwarm().NewStreamWithGroup(remotePeer)
70 - c1.AddGroup(c2.RemotePeer())
71 -
72 - go func() {
73 - ctx := context.Background()
74 - err := runHandshake3(ctx, s, c1, c2)
75 - if err != nil {
76 - log.Error("Handshake3 failed. disconnecting", err)
77 - log.Event(ctx, "Handshake3FailureDisconnect", c2.LocalPeer(), c2.RemotePeer())
78 - c1.Close() // boom.
79 - }
80 - }()
81 - }
82 -}
83 -
84 -func runHandshake3(ctx context.Context, s *Swarm, sc *ps.Conn, c conn.Conn) error {
85 - log.Event(ctx, "newConnection", c.LocalPeer(), c.RemotePeer())
86 -
87 - stream, err := sc.NewStream()
88 - if err != nil {
89 - return err
90 - }
91 -
92 - // handshake3
93 - h3result, err := conn.Handshake3(ctx, stream, c)
94 - if err != nil {
95 - return fmt.Errorf("Handshake3 failed: %s", err)
96 - }
97 -
98 - // check for nats. you know, just in case.
99 - if h3result.LocalObservedAddress != nil {
100 - checkNATWarning(s, h3result.LocalObservedAddress, c.LocalMultiaddr())
101 - } else {
102 - log.Warningf("Received nil observed address from %s", c.RemotePeer())
103 - }
104 -
105 - stream.Close()
106 - log.Event(ctx, "handshake3Succeeded", c.LocalPeer(), c.RemotePeer())
107 - return nil
60 +func (s *Swarm) connHandler(c *ps.Conn) {
61 + go func() {
62 + ctx := context.Background()
63 + // this context is for running the handshake, which -- when receiveing connections
64 + // -- we have no bound on beyond what the transport protocol bounds it at.
65 + // note that setup + the handshake are bounded by underlying io.
66 + // (i.e. if TCP or UDP disconnects (or the swarm closes), we're done.
67 + // Q: why not have a shorter handshake? think about an HTTP server on really slow conns.
68 + // as long as the conn is live (TCP says its online), it tries its best. we follow suit.)
69 +
70 + if _, err := s.newConnSetup(ctx, c); err != nil {
71 + log.Error(err)
72 + log.Event(ctx, "newConnHandlerDisconnect", lgbl.NetConn(c.NetConn()), lgbl.Error(err))
73 + c.Close() // boom. close it.
74 + return
75 + }
76 + }()
77 }
net/swarm2/swarm_stream.go new
+52
@@ -0,0 +1,52 @@
1 +package swarm
2 +
3 +import (
4 + ps "github.com/jbenet/go-peerstream"
5 +)
6 +
7 +// a Stream is a wrapper around a ps.Stream that exposes a way to get
8 +// our SwarmConn and Swarm (instead of just the ps.Conn and ps.Swarm)
9 +type Stream ps.Stream
10 +
11 +// StreamHandler is called when new streams are opened from remote peers.
12 +// See peerstream.StreamHandler
13 +type StreamHandler func(*Stream)
14 +
15 +// Stream returns the underlying peerstream.Stream
16 +func (s *Stream) Stream() *ps.Stream {
17 + return (*ps.Stream)(s)
18 +}
19 +
20 +// Conn returns the Conn associated with this Stream
21 +func (s *Stream) Conn() *SwarmConn {
22 + return (*SwarmConn)(s.Stream().Conn())
23 +}
24 +
25 +// Write writes bytes to a stream, calling write data for each call.
26 +func (s *Stream) Wait() error {
27 + return s.Stream().Wait()
28 +}
29 +
30 +func (s *Stream) Read(p []byte) (n int, err error) {
31 + return s.Stream().Read(p)
32 +}
33 +
34 +func (s *Stream) Write(p []byte) (n int, err error) {
35 + return s.Stream().Write(p)
36 +}
37 +
38 +func (s *Stream) Close() error {
39 + return s.Stream().Close()
40 +}
41 +
42 +func wrapStream(pss *ps.Stream) *Stream {
43 + return (*Stream)(pss)
44 +}
45 +
46 +func wrapStreams(st []*ps.Stream) []*Stream {
47 + out := make([]*Stream, len(st))
48 + for i, s := range st {
49 + out[i] = wrapStream(s)
50 + }
51 + return out
52 +}
net/swarm2/swarm_test.go new
+264
@@ -0,0 +1,264 @@
1 +package swarm
2 +
3 +import (
4 + "bytes"
5 + "io"
6 + "sync"
7 + "testing"
8 + "time"
9 +
10 + ci "github.com/jbenet/go-ipfs/crypto"
11 + peer "github.com/jbenet/go-ipfs/peer"
12 + u "github.com/jbenet/go-ipfs/util"
13 + errors "github.com/jbenet/go-ipfs/util/debugerror"
14 + testutil "github.com/jbenet/go-ipfs/util/testutil"
15 +
16 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
17 + ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
18 +)
19 +
20 +func EchoStreamHandler(stream *Stream) {
21 + go func() {
22 + defer stream.Close()
23 +
24 + // pull out the ipfs conn
25 + c := stream.Conn().RawConn()
26 + log.Debugf("%s ponging to %s", c.LocalPeer(), c.RemotePeer())
27 +
28 + buf := make([]byte, 4)
29 +
30 + for {
31 + if _, err := stream.Read(buf); err != nil {
32 + if err != io.EOF {
33 + log.Error("ping receive error:", err)
34 + }
35 + return
36 + }
37 +
38 + if !bytes.Equal(buf, []byte("ping")) {
39 + log.Errorf("ping receive error: ping != %s %v", buf, buf)
40 + return
41 + }
42 +
43 + if _, err := stream.Write([]byte("pong")); err != nil {
44 + log.Error("pond send error:", err)
45 + return
46 + }
47 + }
48 + }()
49 +}
50 +
51 +func setupPeer(t *testing.T, addr string) peer.Peer {
52 + tcp, err := ma.NewMultiaddr(addr)
53 + if err != nil {
54 + t.Fatal(err)
55 + }
56 +
57 + sk, pk, err := ci.GenerateKeyPair(ci.RSA, 512)
58 + if err != nil {
59 + t.Fatal(err)
60 + }
61 +
62 + p, err := testutil.NewPeerWithKeyPair(sk, pk)
63 + if err != nil {
64 + t.Fatal(err)
65 + }
66 + p.AddAddress(tcp)
67 + return p
68 +}
69 +
70 +func makeSwarms(ctx context.Context, t *testing.T, addrs []string) ([]*Swarm, []peer.Peer) {
71 + swarms := []*Swarm{}
72 +
73 + for _, addr := range addrs {
74 + local := setupPeer(t, addr)
75 + peerstore := peer.NewPeerstore()
76 + swarm, err := NewSwarm(ctx, local.Addresses(), local, peerstore)
77 + if err != nil {
78 + t.Fatal(err)
79 + }
80 + swarm.SetStreamHandler(EchoStreamHandler)
81 + swarms = append(swarms, swarm)
82 + }
83 +
84 + peers := make([]peer.Peer, len(swarms))
85 + for i, s := range swarms {
86 + peers[i] = s.local
87 + }
88 +
89 + return swarms, peers
90 +}
91 +
92 +func SubtestSwarm(t *testing.T, addrs []string, MsgNum int) {
93 + // t.Skip("skipping for another test")
94 +
95 + ctx := context.Background()
96 + swarms, peers := makeSwarms(ctx, t, addrs)
97 +
98 + // connect everyone
99 + {
100 + var wg sync.WaitGroup
101 + connect := func(s *Swarm, dst peer.Peer) {
102 + // copy for other peer
103 +
104 + cp, err := s.peers.FindOrCreate(dst.ID())
105 + if err != nil {
106 + t.Fatal(err)
107 + }
108 + cp.AddAddress(dst.Addresses()[0])
109 +
110 + log.Infof("SWARM TEST: %s dialing %s", s.local, dst)
111 + if _, err := s.Dial(cp); err != nil {
112 + t.Fatal("error swarm dialing to peer", err)
113 + }
114 + log.Infof("SWARM TEST: %s connected to %s", s.local, dst)
115 + wg.Done()
116 + }
117 +
118 + log.Info("Connecting swarms simultaneously.")
119 + for _, s := range swarms {
120 + for _, p := range peers {
121 + if p != s.local { // don't connect to self.
122 + wg.Add(1)
123 + connect(s, p)
124 + }
125 + }
126 + }
127 + wg.Wait()
128 +
129 + for _, s := range swarms {
130 + log.Infof("%s swarm routing table: %s", s.local, s.GetPeerList())
131 + }
132 + }
133 +
134 + // ping/pong
135 + for _, s1 := range swarms {
136 + log.Debugf("-------------------------------------------------------")
137 + log.Debugf("%s ping pong round", s1.local)
138 + log.Debugf("-------------------------------------------------------")
139 +
140 + _, cancel := context.WithCancel(ctx)
141 + peers, err := s1.peers.All()
142 + if err != nil {
143 + t.Fatal(err)
144 + }
145 +
146 + got := map[u.Key]int{}
147 + errChan := make(chan error, MsgNum*len(*peers))
148 + streamChan := make(chan *Stream, MsgNum)
149 +
150 + // send out "ping" x MsgNum to every peer
151 + go func() {
152 + defer close(streamChan)
153 +
154 + var wg sync.WaitGroup
155 + send := func(p peer.Peer) {
156 + defer wg.Done()
157 +
158 + // first, one stream per peer (nice)
159 + stream, err := s1.NewStreamWithPeer(p)
160 + if err != nil {
161 + errChan <- errors.Wrap(err)
162 + return
163 + }
164 +
165 + // send out ping!
166 + for k := 0; k < MsgNum; k++ { // with k messages
167 + msg := "ping"
168 + log.Debugf("%s %s %s (%d)", s1.local, msg, p, k)
169 + stream.Write([]byte(msg))
170 + }
171 +
172 + // read it later
173 + streamChan <- stream
174 + }
175 +
176 + for _, p := range *peers {
177 + wg.Add(1)
178 + go send(p)
179 + }
180 + wg.Wait()
181 + }()
182 +
183 + // receive "pong" x MsgNum from every peer
184 + go func() {
185 + defer close(errChan)
186 + count := 0
187 + countShouldBe := (MsgNum * len(*peers))
188 + for stream := range streamChan { // one per peer
189 + defer stream.Close()
190 +
191 + // get peer on the other side
192 + p := stream.Conn().RemotePeer()
193 +
194 + // receive pings
195 + msgCount := 0
196 + msg := make([]byte, 4)
197 + for k := 0; k < MsgNum; k++ { // with k messages
198 +
199 + // read from the stream
200 + if _, err := stream.Read(msg); err != nil {
201 + errChan <- errors.Wrap(err)
202 + continue
203 + }
204 +
205 + if string(msg) != "pong" {
206 + errChan <- errors.Errorf("unexpected message: %s", msg)
207 + continue
208 + }
209 +
210 + log.Debugf("%s %s %s (%d)", s1.local, msg, p, k)
211 + msgCount++
212 + }
213 +
214 + got[p.Key()] = msgCount
215 + count += msgCount
216 + }
217 +
218 + if count != countShouldBe {
219 + errChan <- errors.Errorf("count mismatch: %d != %d", count, countShouldBe)
220 + }
221 + }()
222 +
223 + // check any errors (blocks till consumer is done)
224 + for err := range errChan {
225 + if err != nil {
226 + t.Fatal(err.Error())
227 + }
228 + }
229 +
230 + log.Debugf("%s got pongs", s1.local)
231 + if len(*peers) != len(got) {
232 + t.Error("got less messages than sent")
233 + }
234 +
235 + for p, n := range got {
236 + if n != MsgNum {
237 + t.Error("peer did not get all msgs", p, n, "/", MsgNum)
238 + }
239 + }
240 +
241 + cancel()
242 + <-time.After(10 * time.Millisecond)
243 + }
244 +
245 + for _, s := range swarms {
246 + s.Close()
247 + }
248 +}
249 +
250 +func TestSwarm(t *testing.T) {
251 + // t.Skip("skipping for another test")
252 +
253 + addrs := []string{
254 + "/ip4/127.0.0.1/tcp/10234",
255 + "/ip4/127.0.0.1/tcp/10235",
256 + "/ip4/127.0.0.1/tcp/10236",
257 + "/ip4/127.0.0.1/tcp/10237",
258 + "/ip4/127.0.0.1/tcp/10238",
259 + }
260 +
261 + // msgs := 1000
262 + msgs := 100
263 + SubtestSwarm(t, addrs, msgs)
264 +}