@cryptotaxi247 / kubo / commits / dadb8b775

host interface + services

The separation of work in the p2p pkg is as follows: - net implements the Swarm and connectivity - protocol has muxer and header protocols - host implements protocol muxing + services - identify took over handshake completely! yay. - p2p package works as a whole

Juan Batiz-Benet committed Jan 1, 2015 at 10:42 UTC dadb8b775b1fb9ce387e78620f497b214d73dd28
13 files changed +1586
p2p/host/basic/basic_host.go new
+149
@@ -0,0 +1,149 @@
1 +package basichost
2 +
3 +import (
4 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5 +
6 + eventlog "github.com/jbenet/go-ipfs/util/eventlog"
7 +
8 + inet "github.com/jbenet/go-ipfs/p2p/net2"
9 + peer "github.com/jbenet/go-ipfs/p2p/peer"
10 + protocol "github.com/jbenet/go-ipfs/p2p/protocol"
11 + identify "github.com/jbenet/go-ipfs/p2p/protocol/identify"
12 + relay "github.com/jbenet/go-ipfs/p2p/protocol/relay"
13 +)
14 +
15 +var log = eventlog.Logger("p2p/host/basic")
16 +
17 +type BasicHost struct {
18 + network inet.Network
19 + mux protocol.Mux
20 + ids *identify.IDService
21 + relay *relay.RelayService
22 +}
23 +
24 +// New constructs and sets up a new *BasicHost with given Network
25 +func New(net inet.Network) *BasicHost {
26 + h := &BasicHost{
27 + network: net,
28 + mux: protocol.Mux{Handlers: protocol.StreamHandlerMap{}},
29 + }
30 +
31 + // setup host services
32 + h.ids = identify.NewIDService(h)
33 + h.relay = relay.NewRelayService(h, h.Mux().HandleSync)
34 +
35 + net.SetConnHandler(h.newConnHandler)
36 + net.SetStreamHandler(h.newStreamHandler)
37 +
38 + return h
39 +}
40 +
41 +// newConnHandler is the remote-opened conn handler for inet.Network
42 +func (h *BasicHost) newConnHandler(c inet.Conn) {
43 + h.ids.IdentifyConn(c)
44 +}
45 +
46 +// newStreamHandler is the remote-opened stream handler for inet.Network
47 +func (h *BasicHost) newStreamHandler(s inet.Stream) {
48 + h.Mux().Handle(s)
49 +}
50 +
51 +// ID returns the (local) peer.ID associated with this Host
52 +func (h *BasicHost) ID() peer.ID {
53 + return h.Network().LocalPeer()
54 +}
55 +
56 +// Peerstore returns the Host's repository of Peer Addresses and Keys.
57 +func (h *BasicHost) Peerstore() peer.Peerstore {
58 + return h.Network().Peerstore()
59 +}
60 +
61 +// Networks returns the Network interface of the Host
62 +func (h *BasicHost) Network() inet.Network {
63 + return h.network
64 +}
65 +
66 +// Mux returns the Mux multiplexing incoming streams to protocol handlers
67 +func (h *BasicHost) Mux() *protocol.Mux {
68 + return &h.mux
69 +}
70 +
71 +func (h *BasicHost) IDService() *identify.IDService {
72 + return h.ids
73 +}
74 +
75 +// SetStreamHandler sets the protocol handler on the Host's Mux.
76 +// This is equivalent to:
77 +// host.Mux().SetHandler(proto, handler)
78 +// (Threadsafe)
79 +func (h *BasicHost) SetStreamHandler(pid protocol.ID, handler inet.StreamHandler) {
80 + h.Mux().SetHandler(pid, handler)
81 +}
82 +
83 +// NewStream opens a new stream to given peer p, and writes a p2p/protocol
84 +// header with given protocol.ID. If there is no connection to p, attempts
85 +// to create one. If ProtocolID is "", writes no header.
86 +// (Threadsafe)
87 +func (h *BasicHost) NewStream(pid protocol.ID, p peer.ID) (inet.Stream, error) {
88 + s, err := h.Network().NewStream(p)
89 + if err != nil {
90 + return nil, err
91 + }
92 +
93 + if err := protocol.WriteHeader(s, pid); err != nil {
94 + s.Close()
95 + return nil, err
96 + }
97 +
98 + return s, nil
99 +}
100 +
101 +// Connect ensures there is a connection between this host and the peer with
102 +// given peer.ID. Connect will absorb the addresses in pi into its internal
103 +// peerstore. If there is not an active connection, Connect will issue a
104 +// h.Network.Dial, and block until a connection is open, or an error is
105 +// returned. // TODO: Relay + NAT.
106 +func (h *BasicHost) Connect(ctx context.Context, pi peer.PeerInfo) error {
107 +
108 + // absorb addresses into peerstore
109 + h.Peerstore().AddPeerInfo(pi)
110 +
111 + cs := h.Network().ConnsToPeer(pi.ID)
112 + if len(cs) > 0 {
113 + return nil
114 + }
115 +
116 + return h.dialPeer(ctx, pi.ID)
117 +}
118 +
119 +// dialPeer opens a connection to peer, and makes sure to identify
120 +// the connection once it has been opened.
121 +func (h *BasicHost) dialPeer(ctx context.Context, p peer.ID) error {
122 + log.Debugf("host %s dialing %s", h.ID, p)
123 + c, err := h.Network().DialPeer(ctx, p)
124 + if err != nil {
125 + return err
126 + }
127 +
128 + // identify the connection before returning.
129 + done := make(chan struct{})
130 + go func() {
131 + h.ids.IdentifyConn(c)
132 + close(done)
133 + }()
134 +
135 + // respect don contexteone
136 + select {
137 + case <-done:
138 + case <-ctx.Done():
139 + return ctx.Err()
140 + }
141 +
142 + log.Debugf("host %s finished dialing %s", h.ID, p)
143 + return nil
144 +}
145 +
146 +// Close shuts down the Host's services (network, etc).
147 +func (h *BasicHost) Close() error {
148 + return h.Network().Close()
149 +}
p2p/host/basic/basic_host_test.go new
+63
@@ -0,0 +1,63 @@
1 +package basichost_test
2 +
3 +import (
4 + "bytes"
5 + "io"
6 + "testing"
7 +
8 + inet "github.com/jbenet/go-ipfs/p2p/net2"
9 + protocol "github.com/jbenet/go-ipfs/p2p/protocol"
10 + testutil "github.com/jbenet/go-ipfs/p2p/test/util"
11 +
12 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
13 +)
14 +
15 +func TestHostSimple(t *testing.T) {
16 +
17 + ctx := context.Background()
18 + h1 := testutil.GenHostSwarm(t, ctx)
19 + h2 := testutil.GenHostSwarm(t, ctx)
20 + defer h1.Close()
21 + defer h2.Close()
22 +
23 + h2pi := h2.Peerstore().PeerInfo(h2.ID())
24 + if err := h1.Connect(ctx, h2pi); err != nil {
25 + t.Fatal(err)
26 + }
27 +
28 + piper, pipew := io.Pipe()
29 + h2.SetStreamHandler(protocol.TestingID, func(s inet.Stream) {
30 + defer s.Close()
31 + w := io.MultiWriter(s, pipew)
32 + io.Copy(w, s) // mirror everything
33 + })
34 +
35 + s, err := h1.NewStream(protocol.TestingID, h2pi.ID)
36 + if err != nil {
37 + t.Fatal(err)
38 + }
39 +
40 + // write to the stream
41 + buf1 := []byte("abcdefghijkl")
42 + if _, err := s.Write(buf1); err != nil {
43 + t.Fatal(err)
44 + }
45 +
46 + // get it from the stream (echoed)
47 + buf2 := make([]byte, len(buf1))
48 + if _, err := io.ReadFull(s, buf2); err != nil {
49 + t.Fatal(err)
50 + }
51 + if !bytes.Equal(buf1, buf2) {
52 + t.Fatal("buf1 != buf2 -- %x != %x", buf1, buf2)
53 + }
54 +
55 + // get it from the pipe (tee)
56 + buf3 := make([]byte, len(buf1))
57 + if _, err := io.ReadFull(piper, buf3); err != nil {
58 + t.Fatal(err)
59 + }
60 + if !bytes.Equal(buf1, buf3) {
61 + t.Fatal("buf1 != buf3 -- %x != %x", buf1, buf3)
62 + }
63 +}
p2p/host/host.go new
+54
@@ -0,0 +1,54 @@
1 +package host
2 +
3 +import (
4 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5 +
6 + eventlog "github.com/jbenet/go-ipfs/util/eventlog"
7 +
8 + inet "github.com/jbenet/go-ipfs/p2p/net2"
9 + peer "github.com/jbenet/go-ipfs/p2p/peer"
10 + protocol "github.com/jbenet/go-ipfs/p2p/protocol"
11 +)
12 +
13 +var log = eventlog.Logger("p2p/host")
14 +
15 +// Host is an object participating in a p2p network, which
16 +// implements protocols or provides services. It handles
17 +// requests like a Server, and issues requests like a Client.
18 +// It is called Host because it is both Server and Client (and Peer
19 +// may be confusing).
20 +type Host interface {
21 + // ID returns the (local) peer.ID associated with this Host
22 + ID() peer.ID
23 +
24 + // Peerstore returns the Host's repository of Peer Addresses and Keys.
25 + Peerstore() peer.Peerstore
26 +
27 + // Networks returns the Network interface of the Host
28 + Network() inet.Network
29 +
30 + // Mux returns the Mux multiplexing incoming streams to protocol handlers
31 + Mux() *protocol.Mux
32 +
33 + // Connect ensures there is a connection between this host and the peer with
34 + // given peer.ID. Connect will absorb the addresses in pi into its internal
35 + // peerstore. If there is not an active connection, Connect will issue a
36 + // h.Network.Dial, and block until a connection is open, or an error is
37 + // returned. // TODO: Relay + NAT.
38 + Connect(ctx context.Context, pi peer.PeerInfo) error
39 +
40 + // SetStreamHandler sets the protocol handler on the Host's Mux.
41 + // This is equivalent to:
42 + // host.Mux().SetHandler(proto, handler)
43 + // (Threadsafe)
44 + SetStreamHandler(pid protocol.ID, handler inet.StreamHandler)
45 +
46 + // NewStream opens a new stream to given peer p, and writes a p2p/protocol
47 + // header with given protocol.ID. If there is no connection to p, attempts
48 + // to create one. If ProtocolID is "", writes no header.
49 + // (Threadsafe)
50 + NewStream(pid protocol.ID, p peer.ID) (inet.Stream, error)
51 +
52 + // Close shuts down the host, its Network, and services.
53 + Close() error
54 +}
p2p/protocol/identify/id.go new
+212
@@ -0,0 +1,212 @@
1 +package identify
2 +
3 +import (
4 + "fmt"
5 + "sync"
6 +
7 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
8 + ggio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
9 + semver "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/coreos/go-semver/semver"
10 + ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
11 +
12 + config "github.com/jbenet/go-ipfs/config"
13 + eventlog "github.com/jbenet/go-ipfs/util/eventlog"
14 +
15 + host "github.com/jbenet/go-ipfs/p2p/host"
16 + inet "github.com/jbenet/go-ipfs/p2p/net2"
17 + protocol "github.com/jbenet/go-ipfs/p2p/protocol"
18 +
19 + pb "github.com/jbenet/go-ipfs/p2p/protocol/identify/pb"
20 +)
21 +
22 +var log = eventlog.Logger("net/identify")
23 +
24 +// ID is the protocol.ID of the Identify Service.
25 +const ID protocol.ID = "/ipfs/identify"
26 +
27 +// IpfsVersion holds the current protocol version for a client running this code
28 +var IpfsVersion *semver.Version
29 +var ClientVersion = "go-ipfs/" + config.CurrentVersionNumber
30 +
31 +func init() {
32 + var err error
33 + IpfsVersion, err = semver.NewVersion("0.0.1")
34 + if err != nil {
35 + panic(fmt.Errorf("invalid protocol version: %v", err))
36 + }
37 +}
38 +
39 +// IDService is a structure that implements ProtocolIdentify.
40 +// It is a trivial service that gives the other peer some
41 +// useful information about the local peer. A sort of hello.
42 +//
43 +// The IDService sends:
44 +// * Our IPFS Protocol Version
45 +// * Our IPFS Agent Version
46 +// * Our public Listen Addresses
47 +type IDService struct {
48 + Host host.Host
49 +
50 + // connections undergoing identification
51 + // for wait purposes
52 + currid map[inet.Conn]chan struct{}
53 + currmu sync.RWMutex
54 +}
55 +
56 +func NewIDService(h host.Host) *IDService {
57 + s := &IDService{
58 + Host: h,
59 + currid: make(map[inet.Conn]chan struct{}),
60 + }
61 + h.SetStreamHandler(ID, s.RequestHandler)
62 + return s
63 +}
64 +
65 +func (ids *IDService) IdentifyConn(c inet.Conn) {
66 + ids.currmu.Lock()
67 + if wait, found := ids.currid[c]; found {
68 + ids.currmu.Unlock()
69 + log.Debugf("IdentifyConn called twice on: %s", c)
70 + <-wait // already identifying it. wait for it.
71 + return
72 + }
73 + ids.currid[c] = make(chan struct{})
74 + ids.currmu.Unlock()
75 +
76 + s, err := c.NewStream()
77 + if err != nil {
78 + log.Error("error opening initial stream for %s", ID)
79 + log.Event(context.TODO(), "IdentifyOpenFailed", c.RemotePeer())
80 + } else {
81 +
82 + // ok give the response to our handler.
83 + if err := protocol.WriteHeader(s, ID); err != nil {
84 + log.Error("error writing stream header for %s", ID)
85 + log.Event(context.TODO(), "IdentifyOpenFailed", c.RemotePeer())
86 + }
87 + ids.ResponseHandler(s)
88 + }
89 +
90 + ids.currmu.Lock()
91 + ch, found := ids.currid[c]
92 + delete(ids.currid, c)
93 + ids.currmu.Unlock()
94 +
95 + if !found {
96 + log.Errorf("IdentifyConn failed to find channel (programmer error) for %s", c)
97 + return
98 + }
99 +
100 + close(ch) // release everyone waiting.
101 +}
102 +
103 +func (ids *IDService) RequestHandler(s inet.Stream) {
104 + defer s.Close()
105 + c := s.Conn()
106 +
107 + w := ggio.NewDelimitedWriter(s)
108 + mes := pb.Identify{}
109 + ids.populateMessage(&mes, s.Conn())
110 + w.WriteMsg(&mes)
111 +
112 + log.Debugf("%s sent message to %s %s", ID,
113 + c.RemotePeer(), c.RemoteMultiaddr())
114 +}
115 +
116 +func (ids *IDService) ResponseHandler(s inet.Stream) {
117 + defer s.Close()
118 + c := s.Conn()
119 +
120 + r := ggio.NewDelimitedReader(s, 2048)
121 + mes := pb.Identify{}
122 + if err := r.ReadMsg(&mes); err != nil {
123 + log.Errorf("%s error receiving message from %s %s", ID,
124 + c.RemotePeer(), c.RemoteMultiaddr())
125 + return
126 + }
127 + ids.consumeMessage(&mes, c)
128 +
129 + log.Debugf("%s received message from %s %s", ID,
130 + c.RemotePeer(), c.RemoteMultiaddr())
131 +}
132 +
133 +func (ids *IDService) populateMessage(mes *pb.Identify, c inet.Conn) {
134 +
135 + // set protocols this node is currently handling
136 + protos := ids.Host.Mux().Protocols()
137 + mes.Protocols = make([]string, len(protos))
138 + for i, p := range protos {
139 + mes.Protocols[i] = string(p)
140 + }
141 +
142 + // observed address so other side is informed of their
143 + // "public" address, at least in relation to us.
144 + mes.ObservedAddr = c.RemoteMultiaddr().Bytes()
145 +
146 + // set listen addrs
147 + laddrs, err := ids.Host.Network().InterfaceListenAddresses()
148 + if err != nil {
149 + log.Error(err)
150 + } else {
151 + mes.ListenAddrs = make([][]byte, len(laddrs))
152 + for i, addr := range laddrs {
153 + mes.ListenAddrs[i] = addr.Bytes()
154 + }
155 + log.Debugf("%s sent listen addrs to %s: %s", c.LocalPeer(), c.RemotePeer(), laddrs)
156 + }
157 +
158 + // set protocol versions
159 + s := IpfsVersion.String()
160 + mes.ProtocolVersion = &s
161 + mes.AgentVersion = &ClientVersion
162 +}
163 +
164 +func (ids *IDService) consumeMessage(mes *pb.Identify, c inet.Conn) {
165 + p := c.RemotePeer()
166 +
167 + // mes.Protocols
168 + // mes.ObservedAddr
169 +
170 + // mes.ListenAddrs
171 + laddrs := mes.GetListenAddrs()
172 + lmaddrs := make([]ma.Multiaddr, 0, len(laddrs))
173 + for _, addr := range laddrs {
174 + maddr, err := ma.NewMultiaddrBytes(addr)
175 + if err != nil {
176 + log.Errorf("%s failed to parse multiaddr from %s %s", ID,
177 + p, c.RemoteMultiaddr())
178 + continue
179 + }
180 + lmaddrs = append(lmaddrs, maddr)
181 + }
182 +
183 + // update our peerstore with the addresses.
184 + ids.Host.Peerstore().AddAddresses(p, lmaddrs)
185 + log.Debugf("%s received listen addrs for %s: %s", c.LocalPeer(), c.RemotePeer(), lmaddrs)
186 +
187 + // get protocol versions
188 + pv := *mes.ProtocolVersion
189 + av := *mes.AgentVersion
190 + ids.Host.Peerstore().Put(p, "ProtocolVersion", pv)
191 + ids.Host.Peerstore().Put(p, "AgentVersion", av)
192 +}
193 +
194 +// IdentifyWait returns a channel which will be closed once
195 +// "ProtocolIdentify" (handshake3) finishes on given conn.
196 +// This happens async so the connection can start to be used
197 +// even if handshake3 knowledge is not necesary.
198 +// Users **MUST** call IdentifyWait _after_ IdentifyConn
199 +func (ids *IDService) IdentifyWait(c inet.Conn) <-chan struct{} {
200 + ids.currmu.Lock()
201 + ch, found := ids.currid[c]
202 + ids.currmu.Unlock()
203 + if found {
204 + return ch
205 + }
206 +
207 + // if not found, it means we are already done identifying it, or
208 + // haven't even started. either way, return a new channel closed.
209 + ch = make(chan struct{})
210 + close(ch)
211 + return ch
212 +}
p2p/protocol/identify/id_test.go new
+106
@@ -0,0 +1,106 @@
1 +package identify_test
2 +
3 +import (
4 + "testing"
5 + "time"
6 +
7 + host "github.com/jbenet/go-ipfs/p2p/host"
8 + peer "github.com/jbenet/go-ipfs/p2p/peer"
9 + identify "github.com/jbenet/go-ipfs/p2p/protocol/identify"
10 + testutil "github.com/jbenet/go-ipfs/p2p/test/util"
11 +
12 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
13 + ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
14 +)
15 +
16 +func subtestIDService(t *testing.T, postDialWait time.Duration) {
17 +
18 + // the generated networks should have the id service wired in.
19 + ctx := context.Background()
20 + h1 := testutil.GenHostSwarm(t, ctx)
21 + h2 := testutil.GenHostSwarm(t, ctx)
22 +
23 + h1p := h1.ID()
24 + h2p := h2.ID()
25 +
26 + testKnowsAddrs(t, h1, h2p, []ma.Multiaddr{}) // nothing
27 + testKnowsAddrs(t, h2, h1p, []ma.Multiaddr{}) // nothing
28 +
29 + h2pi := h2.Peerstore().PeerInfo(h2p)
30 + if err := h1.Connect(ctx, h2pi); err != nil {
31 + t.Fatal(err)
32 + }
33 +
34 + // we need to wait here if Dial returns before ID service is finished.
35 + if postDialWait > 0 {
36 + <-time.After(postDialWait)
37 + }
38 +
39 + // the IDService should be opened automatically, by the network.
40 + // what we should see now is that both peers know about each others listen addresses.
41 + testKnowsAddrs(t, h1, h2p, h2.Peerstore().Addresses(h2p)) // has them
42 + testHasProtocolVersions(t, h1, h2p)
43 +
44 + // now, this wait we do have to do. it's the wait for the Listening side
45 + // to be done identifying the connection.
46 + c := h2.Network().ConnsToPeer(h1.ID())
47 + if len(c) < 1 {
48 + t.Fatal("should have connection by now at least.")
49 + }
50 + <-h2.IDService().IdentifyWait(c[0])
51 +
52 + // and the protocol versions.
53 + testKnowsAddrs(t, h2, h1p, h1.Peerstore().Addresses(h1p)) // has them
54 + testHasProtocolVersions(t, h2, h1p)
55 +}
56 +
57 +func testKnowsAddrs(t *testing.T, h host.Host, p peer.ID, expected []ma.Multiaddr) {
58 + actual := h.Peerstore().Addresses(p)
59 +
60 + if len(actual) != len(expected) {
61 + t.Error("dont have the same addresses")
62 + }
63 +
64 + have := map[string]struct{}{}
65 + for _, addr := range actual {
66 + have[addr.String()] = struct{}{}
67 + }
68 + for _, addr := range expected {
69 + if _, found := have[addr.String()]; !found {
70 + t.Errorf("%s did not have addr for %s: %s", h.ID(), p, addr)
71 + // panic("ahhhhhhh")
72 + }
73 + }
74 +}
75 +
76 +func testHasProtocolVersions(t *testing.T, h host.Host, p peer.ID) {
77 + v, err := h.Peerstore().Get(p, "ProtocolVersion")
78 + if v == nil {
79 + t.Error("no protocol version")
80 + return
81 + }
82 + if v.(string) != identify.IpfsVersion.String() {
83 + t.Error("protocol mismatch", err)
84 + }
85 + v, err = h.Peerstore().Get(p, "AgentVersion")
86 + if v.(string) != identify.ClientVersion {
87 + t.Error("agent version mismatch", err)
88 + }
89 +}
90 +
91 +// TestIDServiceWait gives the ID service 100ms to finish after dialing
92 +// this is becasue it used to be concurrent. Now, Dial wait till the
93 +// id service is done.
94 +func TestIDServiceWait(t *testing.T) {
95 + N := 3
96 + for i := 0; i < N; i++ {
97 + subtestIDService(t, 100*time.Millisecond)
98 + }
99 +}
100 +
101 +func TestIDServiceNoWait(t *testing.T) {
102 + N := 3
103 + for i := 0; i < N; i++ {
104 + subtestIDService(t, 0)
105 + }
106 +}
p2p/protocol/identify/pb/Makefile new
+11
@@ -0,0 +1,11 @@
1 +
2 +PB = $(wildcard *.proto)
3 +GO = $(PB:.proto=.pb.go)
4 +
5 +all: $(GO)
6 +
7 +%.pb.go: %.proto
8 + protoc --gogo_out=. --proto_path=../../../../../../:/usr/local/opt/protobuf/include:. $<
9 +
10 +clean:
11 + rm *.pb.go
p2p/protocol/identify/pb/identify.pb.go new
+93
@@ -0,0 +1,93 @@
1 +// Code generated by protoc-gen-gogo.
2 +// source: identify.proto
3 +// DO NOT EDIT!
4 +
5 +/*
6 +Package identify_pb is a generated protocol buffer package.
7 +
8 +It is generated from these files:
9 + identify.proto
10 +
11 +It has these top-level messages:
12 + Identify
13 +*/
14 +package identify_pb
15 +
16 +import proto "code.google.com/p/gogoprotobuf/proto"
17 +import json "encoding/json"
18 +import math "math"
19 +
20 +// Reference proto, json, and math imports to suppress error if they are not otherwise used.
21 +var _ = proto.Marshal
22 +var _ = &json.SyntaxError{}
23 +var _ = math.Inf
24 +
25 +type Identify struct {
26 + // protocolVersion determines compatibility between peers
27 + ProtocolVersion *string `protobuf:"bytes,5,opt,name=protocolVersion" json:"protocolVersion,omitempty"`
28 + // agentVersion is like a UserAgent string in browsers, or client version in bittorrent
29 + // includes the client name and client.
30 + AgentVersion *string `protobuf:"bytes,6,opt,name=agentVersion" json:"agentVersion,omitempty"`
31 + // publicKey is this node's public key (which also gives its node.ID)
32 + // - may not need to be sent, as secure channel implies it has been sent.
33 + // - then again, if we change / disable secure channel, may still want it.
34 + PublicKey []byte `protobuf:"bytes,1,opt,name=publicKey" json:"publicKey,omitempty"`
35 + // listenAddrs are the multiaddrs the sender node listens for open connections on
36 + ListenAddrs [][]byte `protobuf:"bytes,2,rep,name=listenAddrs" json:"listenAddrs,omitempty"`
37 + // oservedAddr is the multiaddr of the remote endpoint that the sender node perceives
38 + // this is useful information to convey to the other side, as it helps the remote endpoint
39 + // determine whether its connection to the local peer goes through NAT.
40 + ObservedAddr []byte `protobuf:"bytes,4,opt,name=observedAddr" json:"observedAddr,omitempty"`
41 + // protocols are the services this node is running
42 + Protocols []string `protobuf:"bytes,3,rep,name=protocols" json:"protocols,omitempty"`
43 + XXX_unrecognized []byte `json:"-"`
44 +}
45 +
46 +func (m *Identify) Reset() { *m = Identify{} }
47 +func (m *Identify) String() string { return proto.CompactTextString(m) }
48 +func (*Identify) ProtoMessage() {}
49 +
50 +func (m *Identify) GetProtocolVersion() string {
51 + if m != nil && m.ProtocolVersion != nil {
52 + return *m.ProtocolVersion
53 + }
54 + return ""
55 +}
56 +
57 +func (m *Identify) GetAgentVersion() string {
58 + if m != nil && m.AgentVersion != nil {
59 + return *m.AgentVersion
60 + }
61 + return ""
62 +}
63 +
64 +func (m *Identify) GetPublicKey() []byte {
65 + if m != nil {
66 + return m.PublicKey
67 + }
68 + return nil
69 +}
70 +
71 +func (m *Identify) GetListenAddrs() [][]byte {
72 + if m != nil {
73 + return m.ListenAddrs
74 + }
75 + return nil
76 +}
77 +
78 +func (m *Identify) GetObservedAddr() []byte {
79 + if m != nil {
80 + return m.ObservedAddr
81 + }
82 + return nil
83 +}
84 +
85 +func (m *Identify) GetProtocols() []string {
86 + if m != nil {
87 + return m.Protocols
88 + }
89 + return nil
90 +}
91 +
92 +func init() {
93 +}
p2p/protocol/identify/pb/identify.proto new
+27
@@ -0,0 +1,27 @@
1 +package identify.pb;
2 +
3 +message Identify {
4 +
5 + // protocolVersion determines compatibility between peers
6 + optional string protocolVersion = 5; // e.g. ipfs/1.0.0
7 +
8 + // agentVersion is like a UserAgent string in browsers, or client version in bittorrent
9 + // includes the client name and client.
10 + optional string agentVersion = 6; // e.g. go-ipfs/0.1.0
11 +
12 + // publicKey is this node's public key (which also gives its node.ID)
13 + // - may not need to be sent, as secure channel implies it has been sent.
14 + // - then again, if we change / disable secure channel, may still want it.
15 + optional bytes publicKey = 1;
16 +
17 + // listenAddrs are the multiaddrs the sender node listens for open connections on
18 + repeated bytes listenAddrs = 2;
19 +
20 + // oservedAddr is the multiaddr of the remote endpoint that the sender node perceives
21 + // this is useful information to convey to the other side, as it helps the remote endpoint
22 + // determine whether its connection to the local peer goes through NAT.
23 + optional bytes observedAddr = 4;
24 +
25 + // protocols are the services this node is running
26 + repeated string protocols = 3;
27 +}
p2p/protocol/relay/relay.go new
+156
@@ -0,0 +1,156 @@
1 +package relay
2 +
3 +import (
4 + "fmt"
5 + "io"
6 +
7 + mh "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multihash"
8 +
9 + host "github.com/jbenet/go-ipfs/p2p/host"
10 + inet "github.com/jbenet/go-ipfs/p2p/net2"
11 + peer "github.com/jbenet/go-ipfs/p2p/peer"
12 + protocol "github.com/jbenet/go-ipfs/p2p/protocol"
13 + eventlog "github.com/jbenet/go-ipfs/util/eventlog"
14 +)
15 +
16 +var log = eventlog.Logger("p2p/protocol/relay")
17 +
18 +// ID is the protocol.ID of the Relay Service.
19 +const ID protocol.ID = "/ipfs/relay"
20 +
21 +// Relay is a structure that implements ProtocolRelay.
22 +// It is a simple relay service which forwards traffic
23 +// between two directly connected peers.
24 +//
25 +// the protocol is very simple:
26 +//
27 +// /ipfs/relay\n
28 +// <multihash src id>
29 +// <multihash dst id>
30 +// <data stream>
31 +//
32 +type RelayService struct {
33 + host host.Host
34 + handler inet.StreamHandler // for streams sent to us locally.
35 +}
36 +
37 +func NewRelayService(h host.Host, sh inet.StreamHandler) *RelayService {
38 + s := &RelayService{
39 + host: h,
40 + handler: sh,
41 + }
42 + h.SetStreamHandler(ID, s.requestHandler)
43 + return s
44 +}
45 +
46 +// requestHandler is the function called by clients
47 +func (rs *RelayService) requestHandler(s inet.Stream) {
48 + if err := rs.handleStream(s); err != nil {
49 + log.Error("RelayService error:", err)
50 + }
51 +}
52 +
53 +// handleStream is our own handler, which returns an error for simplicity.
54 +func (rs *RelayService) handleStream(s inet.Stream) error {
55 + defer s.Close()
56 +
57 + // read the header (src and dst peer.IDs)
58 + src, dst, err := ReadHeader(s)
59 + if err != nil {
60 + return fmt.Errorf("stream with bad header: %s", err)
61 + }
62 +
63 + local := rs.host.ID()
64 +
65 + switch {
66 + case src == local:
67 + return fmt.Errorf("relaying from self")
68 + case dst == local: // it's for us! yaaay.
69 + log.Debugf("%s consuming stream from %s", local, src)
70 + return rs.consumeStream(s)
71 + default: // src and dst are not local. relay it.
72 + log.Debugf("%s relaying stream %s <--> %s", local, src, dst)
73 + return rs.pipeStream(src, dst, s)
74 + }
75 +}
76 +
77 +// consumeStream connects streams directed to the local peer
78 +// to our handler, with the header now stripped (read).
79 +func (rs *RelayService) consumeStream(s inet.Stream) error {
80 + rs.handler(s) // boom.
81 + return nil
82 +}
83 +
84 +// pipeStream relays over a stream to a remote peer. It's like `cat`
85 +func (rs *RelayService) pipeStream(src, dst peer.ID, s inet.Stream) error {
86 + s2, err := rs.openStreamToPeer(dst)
87 + if err != nil {
88 + return fmt.Errorf("failed to open stream to peer: %s -- %s", dst, err)
89 + }
90 +
91 + if err := WriteHeader(s2, src, dst); err != nil {
92 + return err
93 + }
94 +
95 + // connect the series of tubes.
96 + done := make(chan retio, 2)
97 + go func() {
98 + n, err := io.Copy(s2, s)
99 + done <- retio{n, err}
100 + }()
101 + go func() {
102 + n, err := io.Copy(s, s2)
103 + done <- retio{n, err}
104 + }()
105 +
106 + r1 := <-done
107 + r2 := <-done
108 + log.Infof("%s relayed %d/%d bytes between %s and %s", rs.host.ID(), r1.n, r2.n, src, dst)
109 +
110 + if r1.err != nil {
111 + return r1.err
112 + }
113 + return r2.err
114 +}
115 +
116 +// openStreamToPeer opens a pipe to a remote endpoint
117 +// for now, can only open streams to directly connected peers.
118 +// maybe we can do some routing later on.
119 +func (rs *RelayService) openStreamToPeer(p peer.ID) (inet.Stream, error) {
120 + return rs.host.NewStream(ID, p)
121 +}
122 +
123 +func ReadHeader(r io.Reader) (src, dst peer.ID, err error) {
124 +
125 + mhr := mh.NewReader(r)
126 +
127 + s, err := mhr.ReadMultihash()
128 + if err != nil {
129 + return "", "", err
130 + }
131 +
132 + d, err := mhr.ReadMultihash()
133 + if err != nil {
134 + return "", "", err
135 + }
136 +
137 + return peer.ID(s), peer.ID(d), nil
138 +}
139 +
140 +func WriteHeader(w io.Writer, src, dst peer.ID) error {
141 + // write header to w.
142 + mhw := mh.NewWriter(w)
143 + if err := mhw.WriteMultihash(mh.Multihash(src)); err != nil {
144 + return fmt.Errorf("failed to write relay header: %s -- %s", dst, err)
145 + }
146 + if err := mhw.WriteMultihash(mh.Multihash(dst)); err != nil {
147 + return fmt.Errorf("failed to write relay header: %s -- %s", dst, err)
148 + }
149 +
150 + return nil
151 +}
152 +
153 +type retio struct {
154 + n int64
155 + err error
156 +}
p2p/protocol/relay/relay_test.go new
+303
@@ -0,0 +1,303 @@
1 +package relay_test
2 +
3 +import (
4 + "io"
5 + "testing"
6 +
7 + inet "github.com/jbenet/go-ipfs/p2p/net2"
8 + protocol "github.com/jbenet/go-ipfs/p2p/protocol"
9 + relay "github.com/jbenet/go-ipfs/p2p/protocol/relay"
10 + testutil "github.com/jbenet/go-ipfs/p2p/test/util"
11 + eventlog "github.com/jbenet/go-ipfs/util/eventlog"
12 +
13 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
14 +)
15 +
16 +var log = eventlog.Logger("relay_test")
17 +
18 +func TestRelaySimple(t *testing.T) {
19 +
20 + ctx := context.Background()
21 +
22 + // these networks have the relay service wired in already.
23 + n1 := testutil.GenHostSwarm(t, ctx)
24 + n2 := testutil.GenHostSwarm(t, ctx)
25 + n3 := testutil.GenHostSwarm(t, ctx)
26 +
27 + n1p := n1.ID()
28 + n2p := n2.ID()
29 + n3p := n3.ID()
30 +
31 + n2pi := n2.Peerstore().PeerInfo(n2p)
32 + if err := n1.Connect(ctx, n2pi); err != nil {
33 + t.Fatal("Failed to connect:", err)
34 + }
35 + if err := n3.Connect(ctx, n2pi); err != nil {
36 + t.Fatal("Failed to connect:", err)
37 + }
38 +
39 + // setup handler on n3 to copy everything over to the pipe.
40 + piper, pipew := io.Pipe()
41 + n3.SetStreamHandler(protocol.TestingID, func(s inet.Stream) {
42 + log.Debug("relay stream opened to n3!")
43 + log.Debug("piping and echoing everything")
44 + w := io.MultiWriter(s, pipew)
45 + io.Copy(w, s)
46 + log.Debug("closing stream")
47 + s.Close()
48 + })
49 +
50 + // ok, now we can try to relay n1--->n2--->n3.
51 + log.Debug("open relay stream")
52 + s, err := n1.NewStream(relay.ID, n2p)
53 + if err != nil {
54 + t.Fatal(err)
55 + }
56 +
57 + // ok first thing we write the relay header n1->n3
58 + log.Debug("write relay header")
59 + if err := relay.WriteHeader(s, n1p, n3p); err != nil {
60 + t.Fatal(err)
61 + }
62 +
63 + // ok now the header's there, we can write the next protocol header.
64 + log.Debug("write testing header")
65 + if err := protocol.WriteHeader(s, protocol.TestingID); err != nil {
66 + t.Fatal(err)
67 + }
68 +
69 + // okay, now we should be able to write text, and read it out.
70 + buf1 := []byte("abcdefghij")
71 + buf2 := make([]byte, 10)
72 + buf3 := make([]byte, 10)
73 + log.Debug("write in some text.")
74 + if _, err := s.Write(buf1); err != nil {
75 + t.Fatal(err)
76 + }
77 +
78 + // read it out from the pipe.
79 + log.Debug("read it out from the pipe.")
80 + if _, err := io.ReadFull(piper, buf2); err != nil {
81 + t.Fatal(err)
82 + }
83 + if string(buf1) != string(buf2) {
84 + t.Fatal("should've gotten that text out of the pipe")
85 + }
86 +
87 + // read it out from the stream (echoed)
88 + log.Debug("read it out from the stream (echoed).")
89 + if _, err := io.ReadFull(s, buf3); err != nil {
90 + t.Fatal(err)
91 + }
92 + if string(buf1) != string(buf3) {
93 + t.Fatal("should've gotten that text out of the stream")
94 + }
95 +
96 + // sweet. relay works.
97 + log.Debug("sweet, relay works.")
98 + s.Close()
99 +}
100 +
101 +func TestRelayAcrossFour(t *testing.T) {
102 +
103 + ctx := context.Background()
104 +
105 + // these networks have the relay service wired in already.
106 + n1 := testutil.GenHostSwarm(t, ctx)
107 + n2 := testutil.GenHostSwarm(t, ctx)
108 + n3 := testutil.GenHostSwarm(t, ctx)
109 + n4 := testutil.GenHostSwarm(t, ctx)
110 + n5 := testutil.GenHostSwarm(t, ctx)
111 +
112 + n1p := n1.ID()
113 + n2p := n2.ID()
114 + n3p := n3.ID()
115 + n4p := n4.ID()
116 + n5p := n5.ID()
117 +
118 + n2pi := n2.Peerstore().PeerInfo(n2p)
119 + n4pi := n4.Peerstore().PeerInfo(n4p)
120 +
121 + if err := n1.Connect(ctx, n2pi); err != nil {
122 + t.Fatalf("Failed to dial:", err)
123 + }
124 + if err := n3.Connect(ctx, n2pi); err != nil {
125 + t.Fatalf("Failed to dial:", err)
126 + }
127 + if err := n3.Connect(ctx, n4pi); err != nil {
128 + t.Fatalf("Failed to dial:", err)
129 + }
130 + if err := n5.Connect(ctx, n4pi); err != nil {
131 + t.Fatalf("Failed to dial:", err)
132 + }
133 +
134 + // setup handler on n5 to copy everything over to the pipe.
135 + piper, pipew := io.Pipe()
136 + n5.SetStreamHandler(protocol.TestingID, func(s inet.Stream) {
137 + log.Debug("relay stream opened to n5!")
138 + log.Debug("piping and echoing everything")
139 + w := io.MultiWriter(s, pipew)
140 + io.Copy(w, s)
141 + log.Debug("closing stream")
142 + s.Close()
143 + })
144 +
145 + // ok, now we can try to relay n1--->n2--->n3--->n4--->n5
146 + log.Debug("open relay stream")
147 + s, err := n1.NewStream(relay.ID, n2p)
148 + if err != nil {
149 + t.Fatal(err)
150 + }
151 +
152 + log.Debugf("write relay header n1->n3 (%s -> %s)", n1p, n3p)
153 + if err := relay.WriteHeader(s, n1p, n3p); err != nil {
154 + t.Fatal(err)
155 + }
156 +
157 + log.Debugf("write relay header n1->n4 (%s -> %s)", n1p, n4p)
158 + if err := protocol.WriteHeader(s, relay.ID); err != nil {
159 + t.Fatal(err)
160 + }
161 + if err := relay.WriteHeader(s, n1p, n4p); err != nil {
162 + t.Fatal(err)
163 + }
164 +
165 + log.Debugf("write relay header n1->n5 (%s -> %s)", n1p, n5p)
166 + if err := protocol.WriteHeader(s, relay.ID); err != nil {
167 + t.Fatal(err)
168 + }
169 + if err := relay.WriteHeader(s, n1p, n5p); err != nil {
170 + t.Fatal(err)
171 + }
172 +
173 + // ok now the header's there, we can write the next protocol header.
174 + log.Debug("write testing header")
175 + if err := protocol.WriteHeader(s, protocol.TestingID); err != nil {
176 + t.Fatal(err)
177 + }
178 +
179 + // okay, now we should be able to write text, and read it out.
180 + buf1 := []byte("abcdefghij")
181 + buf2 := make([]byte, 10)
182 + buf3 := make([]byte, 10)
183 + log.Debug("write in some text.")
184 + if _, err := s.Write(buf1); err != nil {
185 + t.Fatal(err)
186 + }
187 +
188 + // read it out from the pipe.
189 + log.Debug("read it out from the pipe.")
190 + if _, err := io.ReadFull(piper, buf2); err != nil {
191 + t.Fatal(err)
192 + }
193 + if string(buf1) != string(buf2) {
194 + t.Fatal("should've gotten that text out of the pipe")
195 + }
196 +
197 + // read it out from the stream (echoed)
198 + log.Debug("read it out from the stream (echoed).")
199 + if _, err := io.ReadFull(s, buf3); err != nil {
200 + t.Fatal(err)
201 + }
202 + if string(buf1) != string(buf3) {
203 + t.Fatal("should've gotten that text out of the stream")
204 + }
205 +
206 + // sweet. relay works.
207 + log.Debug("sweet, relaying across 4 works.")
208 + s.Close()
209 +}
210 +
211 +func TestRelayStress(t *testing.T) {
212 + buflen := 1 << 18
213 + iterations := 10
214 +
215 + ctx := context.Background()
216 +
217 + // these networks have the relay service wired in already.
218 + n1 := testutil.GenHostSwarm(t, ctx)
219 + n2 := testutil.GenHostSwarm(t, ctx)
220 + n3 := testutil.GenHostSwarm(t, ctx)
221 +
222 + n1p := n1.ID()
223 + n2p := n2.ID()
224 + n3p := n3.ID()
225 +
226 + n2pi := n2.Peerstore().PeerInfo(n2p)
227 + if err := n1.Connect(ctx, n2pi); err != nil {
228 + t.Fatalf("Failed to dial:", err)
229 + }
230 + if err := n3.Connect(ctx, n2pi); err != nil {
231 + t.Fatalf("Failed to dial:", err)
232 + }
233 +
234 + // setup handler on n3 to copy everything over to the pipe.
235 + piper, pipew := io.Pipe()
236 + n3.SetStreamHandler(protocol.TestingID, func(s inet.Stream) {
237 + log.Debug("relay stream opened to n3!")
238 + log.Debug("piping and echoing everything")
239 + w := io.MultiWriter(s, pipew)
240 + io.Copy(w, s)
241 + log.Debug("closing stream")
242 + s.Close()
243 + })
244 +
245 + // ok, now we can try to relay n1--->n2--->n3.
246 + log.Debug("open relay stream")
247 + s, err := n1.NewStream(relay.ID, n2p)
248 + if err != nil {
249 + t.Fatal(err)
250 + }
251 +
252 + // ok first thing we write the relay header n1->n3
253 + log.Debug("write relay header")
254 + if err := relay.WriteHeader(s, n1p, n3p); err != nil {
255 + t.Fatal(err)
256 + }
257 +
258 + // ok now the header's there, we can write the next protocol header.
259 + log.Debug("write testing header")
260 + if err := protocol.WriteHeader(s, protocol.TestingID); err != nil {
261 + t.Fatal(err)
262 + }
263 +
264 + // okay, now write lots of text and read it back out from both
265 + // the pipe and the stream.
266 + buf1 := make([]byte, buflen)
267 + buf2 := make([]byte, len(buf1))
268 + buf3 := make([]byte, len(buf1))
269 +
270 + fillbuf := func(buf []byte, b byte) {
271 + for i := range buf {
272 + buf[i] = b
273 + }
274 + }
275 +
276 + for i := 0; i < iterations; i++ {
277 + fillbuf(buf1, byte(int('a')+i))
278 + log.Debugf("writing %d bytes (%d/%d)", len(buf1), i, iterations)
279 + if _, err := s.Write(buf1); err != nil {
280 + t.Fatal(err)
281 + }
282 +
283 + log.Debug("read it out from the pipe.")
284 + if _, err := io.ReadFull(piper, buf2); err != nil {
285 + t.Fatal(err)
286 + }
287 + if string(buf1) != string(buf2) {
288 + t.Fatal("should've gotten that text out of the pipe")
289 + }
290 +
291 + // read it out from the stream (echoed)
292 + log.Debug("read it out from the stream (echoed).")
293 + if _, err := io.ReadFull(s, buf3); err != nil {
294 + t.Fatal(err)
295 + }
296 + if string(buf1) != string(buf3) {
297 + t.Fatal("should've gotten that text out of the stream")
298 + }
299 + }
300 +
301 + log.Debug("sweet, relay works under stress.")
302 + s.Close()
303 +}
p2p/test/backpressure/backpressure.go new
+1
@@ -0,0 +1 @@
1 +package backpressure_tests
p2p/test/backpressure/backpressure_test.go new
+374
@@ -0,0 +1,374 @@
1 +package backpressure_tests
2 +
3 +import (
4 + crand "crypto/rand"
5 + "io"
6 + "math/rand"
7 + "testing"
8 + "time"
9 +
10 + host "github.com/jbenet/go-ipfs/p2p/host"
11 + inet "github.com/jbenet/go-ipfs/p2p/net2"
12 + peer "github.com/jbenet/go-ipfs/p2p/peer"
13 + protocol "github.com/jbenet/go-ipfs/p2p/protocol"
14 + testutil "github.com/jbenet/go-ipfs/p2p/test/util"
15 + eventlog "github.com/jbenet/go-ipfs/util/eventlog"
16 +
17 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
18 +)
19 +
20 +var log = eventlog.Logger("backpressure")
21 +
22 +// TestBackpressureStreamHandler tests whether mux handler
23 +// ratelimiting works. Meaning, since the handler is sequential
24 +// it should block senders.
25 +//
26 +// Important note: spdystream (which peerstream uses) has a set
27 +// of n workers (n=spdsystream.FRAME_WORKERS) which handle new
28 +// frames, including those starting new streams. So all of them
29 +// can be in the handler at one time. Also, the sending side
30 +// does not rate limit unless we call stream.Wait()
31 +//
32 +//
33 +// Note: right now, this happens muxer-wide. the muxer should
34 +// learn to flow control, so handlers cant block each other.
35 +func TestBackpressureStreamHandler(t *testing.T) {
36 + t.Skip(`Sadly, as cool as this test is, it doesn't work
37 +Because spdystream doesnt handle stream open backpressure
38 +well IMO. I'll see about rewriting that part when it becomes
39 +a problem.
40 +`)
41 +
42 + // a number of concurrent request handlers
43 + limit := 10
44 +
45 + // our way to signal that we're done with 1 request
46 + requestHandled := make(chan struct{})
47 +
48 + // handler rate limiting
49 + receiverRatelimit := make(chan struct{}, limit)
50 + for i := 0; i < limit; i++ {
51 + receiverRatelimit <- struct{}{}
52 + }
53 +
54 + // sender counter of successfully opened streams
55 + senderOpened := make(chan struct{}, limit*100)
56 +
57 + // sender signals it's done (errored out)
58 + senderDone := make(chan struct{})
59 +
60 + // the receiver handles requests with some rate limiting
61 + receiver := func(s inet.Stream) {
62 + log.Debug("receiver received a stream")
63 +
64 + <-receiverRatelimit // acquire
65 + go func() {
66 + // our request handler. can do stuff here. we
67 + // simulate something taking time by waiting
68 + // on requestHandled
69 + log.Error("request worker handling...")
70 + <-requestHandled
71 + log.Error("request worker done!")
72 + receiverRatelimit <- struct{}{} // release
73 + }()
74 + }
75 +
76 + // the sender opens streams as fast as possible
77 + sender := func(host host.Host, remote peer.ID) {
78 + var s inet.Stream
79 + var err error
80 + defer func() {
81 + t.Error(err)
82 + log.Debug("sender error. exiting.")
83 + senderDone <- struct{}{}
84 + }()
85 +
86 + for {
87 + s, err = host.NewStream(protocol.TestingID, remote)
88 + if err != nil {
89 + return
90 + }
91 +
92 + _ = s
93 + // if err = s.SwarmStream().Stream().Wait(); err != nil {
94 + // return
95 + // }
96 +
97 + // "count" another successfully opened stream
98 + // (large buffer so shouldn't block in normal operation)
99 + log.Debug("sender opened another stream!")
100 + senderOpened <- struct{}{}
101 + }
102 + }
103 +
104 + // count our senderOpened events
105 + countStreamsOpenedBySender := func(min int) int {
106 + opened := 0
107 + for opened < min {
108 + log.Debugf("countStreamsOpenedBySender got %d (min %d)", opened, min)
109 + select {
110 + case <-senderOpened:
111 + opened++
112 + case <-time.After(10 * time.Millisecond):
113 + }
114 + }
115 + return opened
116 + }
117 +
118 + // count our received events
119 + // waitForNReceivedStreams := func(n int) {
120 + // for n > 0 {
121 + // log.Debugf("waiting for %d received streams...", n)
122 + // select {
123 + // case <-receiverRatelimit:
124 + // n--
125 + // }
126 + // }
127 + // }
128 +
129 + testStreamsOpened := func(expected int) {
130 + log.Debugf("testing rate limited to %d streams", expected)
131 + if n := countStreamsOpenedBySender(expected); n != expected {
132 + t.Fatalf("rate limiting did not work :( -- %d != %d", expected, n)
133 + }
134 + }
135 +
136 + // ok that's enough setup. let's do it!
137 +
138 + ctx := context.Background()
139 + h1 := testutil.GenHostSwarm(t, ctx)
140 + h2 := testutil.GenHostSwarm(t, ctx)
141 +
142 + // setup receiver handler
143 + h1.SetStreamHandler(protocol.TestingID, receiver)
144 +
145 + h2pi := h2.Peerstore().PeerInfo(h2.ID())
146 + log.Debugf("dialing %s", h2pi.Addrs)
147 + if err := h1.Connect(ctx, h2pi); err != nil {
148 + t.Fatalf("Failed to connect:", err)
149 + }
150 +
151 + // launch sender!
152 + go sender(h2, h1.ID())
153 +
154 + // ok, what do we expect to happen? the receiver should
155 + // receive 10 requests and stop receiving, blocking the sender.
156 + // we can test this by counting 10x senderOpened requests
157 +
158 + <-senderOpened // wait for the sender to successfully open some.
159 + testStreamsOpened(limit - 1)
160 +
161 + // let's "handle" 3 requests.
162 + <-requestHandled
163 + <-requestHandled
164 + <-requestHandled
165 + // the sender should've now been able to open exactly 3 more.
166 +
167 + testStreamsOpened(3)
168 +
169 + // shouldn't have opened anything more
170 + testStreamsOpened(0)
171 +
172 + // let's "handle" 100 requests in batches of 5
173 + for i := 0; i < 20; i++ {
174 + <-requestHandled
175 + <-requestHandled
176 + <-requestHandled
177 + <-requestHandled
178 + <-requestHandled
179 + testStreamsOpened(5)
180 + }
181 +
182 + // success!
183 +
184 + // now for the sugar on top: let's tear down the receiver. it should
185 + // exit the sender.
186 + h1.Close()
187 +
188 + // shouldn't have opened anything more
189 + testStreamsOpened(0)
190 +
191 + select {
192 + case <-time.After(100 * time.Millisecond):
193 + t.Error("receiver shutdown failed to exit sender")
194 + case <-senderDone:
195 + log.Info("handler backpressure works!")
196 + }
197 +}
198 +
199 +// TestStBackpressureStreamWrite tests whether streams see proper
200 +// backpressure when writing data over the network streams.
201 +func TestStBackpressureStreamWrite(t *testing.T) {
202 +
203 + // senderWrote signals that the sender wrote bytes to remote.
204 + // the value is the count of bytes written.
205 + senderWrote := make(chan int, 10000)
206 +
207 + // sender signals it's done (errored out)
208 + senderDone := make(chan struct{})
209 +
210 + // writeStats lets us listen to all the writes and return
211 + // how many happened and how much was written
212 + writeStats := func() (int, int) {
213 + writes := 0
214 + bytes := 0
215 + for {
216 + select {
217 + case n := <-senderWrote:
218 + writes++
219 + bytes = bytes + n
220 + default:
221 + log.Debugf("stats: sender wrote %d bytes, %d writes", bytes, writes)
222 + return bytes, writes
223 + }
224 + }
225 + }
226 +
227 + // sender attempts to write as fast as possible, signaling on the
228 + // completion of every write. This makes it possible to see how
229 + // fast it's actually writing. We pair this with a receiver
230 + // that waits for a signal to read.
231 + sender := func(s inet.Stream) {
232 + defer func() {
233 + s.Close()
234 + senderDone <- struct{}{}
235 + }()
236 +
237 + // ready a buffer of random data
238 + buf := make([]byte, 65536)
239 + crand.Read(buf)
240 +
241 + for {
242 + // send a randomly sized subchunk
243 + from := rand.Intn(len(buf) / 2)
244 + to := rand.Intn(len(buf) / 2)
245 + sendbuf := buf[from : from+to]
246 +
247 + n, err := s.Write(sendbuf)
248 + if err != nil {
249 + log.Debug("sender error. exiting:", err)
250 + return
251 + }
252 +
253 + log.Debugf("sender wrote %d bytes", n)
254 + senderWrote <- n
255 + }
256 + }
257 +
258 + // receive a number of bytes from a stream.
259 + // returns the number of bytes written.
260 + receive := func(s inet.Stream, expect int) {
261 + log.Debugf("receiver to read %d bytes", expect)
262 + rbuf := make([]byte, expect)
263 + n, err := io.ReadFull(s, rbuf)
264 + if err != nil {
265 + t.Error("read failed:", err)
266 + }
267 + if expect != n {
268 + t.Error("read len differs: %d != %d", expect, n)
269 + }
270 + }
271 +
272 + // ok let's do it!
273 +
274 + // setup the networks
275 + ctx := context.Background()
276 + h1 := testutil.GenHostSwarm(t, ctx)
277 + h2 := testutil.GenHostSwarm(t, ctx)
278 +
279 + // setup sender handler on 1
280 + h1.SetStreamHandler(protocol.TestingID, sender)
281 +
282 + h2pi := h2.Peerstore().PeerInfo(h2.ID())
283 + log.Debugf("dialing %s", h2pi.Addrs)
284 + if err := h1.Connect(ctx, h2pi); err != nil {
285 + t.Fatalf("Failed to connect:", err)
286 + }
287 +
288 + // open a stream, from 2->1, this is our reader
289 + s, err := h2.NewStream(protocol.TestingID, h1.ID())
290 + if err != nil {
291 + t.Fatal(err)
292 + }
293 +
294 + // let's make sure r/w works.
295 + testSenderWrote := func(bytesE int) {
296 + bytesA, writesA := writeStats()
297 + if bytesA != bytesE {
298 + t.Errorf("numbers failed: %d =?= %d bytes, via %d writes", bytesA, bytesE, writesA)
299 + }
300 + }
301 +
302 + // 500ms rounds of lockstep write + drain
303 + roundsStart := time.Now()
304 + roundsTotal := 0
305 + for roundsTotal < (2 << 20) {
306 + // let the sender fill its buffers, it will stop sending.
307 + <-time.After(300 * time.Millisecond)
308 + b, _ := writeStats()
309 + testSenderWrote(0)
310 + testSenderWrote(0)
311 +
312 + // drain it all, wait again
313 + receive(s, b)
314 + roundsTotal = roundsTotal + b
315 + }
316 + roundsTime := time.Now().Sub(roundsStart)
317 +
318 + // now read continously, while we measure stats.
319 + stop := make(chan struct{})
320 + contStart := time.Now()
321 +
322 + go func() {
323 + for {
324 + select {
325 + case <-stop:
326 + return
327 + default:
328 + receive(s, 2<<15)
329 + }
330 + }
331 + }()
332 +
333 + contTotal := 0
334 + for contTotal < (2 << 20) {
335 + n := <-senderWrote
336 + contTotal += n
337 + }
338 + stop <- struct{}{}
339 + contTime := time.Now().Sub(contStart)
340 +
341 + // now compare! continuous should've been faster AND larger
342 + if roundsTime < contTime {
343 + t.Error("continuous should have been faster")
344 + }
345 +
346 + if roundsTotal < contTotal {
347 + t.Error("continuous should have been larger, too!")
348 + }
349 +
350 + // and a couple rounds more for good measure ;)
351 + for i := 0; i < 3; i++ {
352 + // let the sender fill its buffers, it will stop sending.
353 + <-time.After(300 * time.Millisecond)
354 + b, _ := writeStats()
355 + testSenderWrote(0)
356 + testSenderWrote(0)
357 +
358 + // drain it all, wait again
359 + receive(s, b)
360 + }
361 +
362 + // this doesn't work :(:
363 + // // now for the sugar on top: let's tear down the receiver. it should
364 + // // exit the sender.
365 + // n1.Close()
366 + // testSenderWrote(0)
367 + // testSenderWrote(0)
368 + // select {
369 + // case <-time.After(2 * time.Second):
370 + // t.Error("receiver shutdown failed to exit sender")
371 + // case <-senderDone:
372 + // log.Info("handler backpressure works!")
373 + // }
374 +}
p2p/test/util/util.go new
+37
@@ -0,0 +1,37 @@
1 +package testutil
2 +
3 +import (
4 + "testing"
5 +
6 + bhost "github.com/jbenet/go-ipfs/p2p/host/basic"
7 + inet "github.com/jbenet/go-ipfs/p2p/net2"
8 + swarm "github.com/jbenet/go-ipfs/p2p/net2/swarm"
9 + peer "github.com/jbenet/go-ipfs/p2p/peer"
10 + tu "github.com/jbenet/go-ipfs/util/testutil"
11 +
12 + context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
13 +)
14 +
15 +func GenSwarmNetwork(t *testing.T, ctx context.Context) *swarm.Network {
16 + p := tu.RandPeerNetParamsOrFatal(t)
17 + ps := peer.NewPeerstore()
18 + ps.AddAddress(p.ID, p.Addr)
19 + ps.AddPubKey(p.ID, p.PubKey)
20 + ps.AddPrivKey(p.ID, p.PrivKey)
21 + n, err := swarm.NewNetwork(ctx, ps.Addresses(p.ID), p.ID, ps)
22 + if err != nil {
23 + t.Fatal(err)
24 + }
25 + return n
26 +}
27 +
28 +func DivulgeAddresses(a, b inet.Network) {
29 + id := a.LocalPeer()
30 + addrs := a.Peerstore().Addresses(id)
31 + b.Peerstore().AddAddresses(id, addrs)
32 +}
33 +
34 +func GenHostSwarm(t *testing.T, ctx context.Context) *bhost.BasicHost {
35 + n := GenSwarmNetwork(t, ctx)
36 + return bhost.New(n)
37 +}