@cryptotaxi247 / kubo / commits / a2abf108a

peerstream update

peerstream was updated to use pluggable transports, including muxado. The interface was also simplified slightly.

Juan Batiz-Benet committed Jan 1, 2015 at 05:58 UTC a2abf108a0f9ff99c4dce70478be604183756928
19 files changed +826 -97
Godeps/Godeps.json
+1 -1
@@ -137,7 +137,7 @@
137 },
138 {
139 "ImportPath": "github.com/jbenet/go-peerstream",
140 - "Rev": "c3ee65e805acc6a27036b9b892e0a95fc9769c3c"
140 + "Rev": "3f8972989ecf7b99db5d718b7ff2e1bf31011d4f"
141 },
142 {
143 "ImportPath": "github.com/jbenet/go-random",
Godeps/_workspace/src/github.com/jbenet/go-peerstream/conn.go
+19 -28
@@ -3,10 +3,9 @@ package peerstream
3 import (
4 "errors"
5 "net"
6 - "net/http"
6 "sync"
7
9 - ss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/spdystream"
8 + pst "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
9 )
10
11 // ConnHandler is a function which receives a Conn. It allows
@@ -36,7 +35,7 @@ var ErrNoConnections = errors.New("no connections")
35
36 // Conn is a Swarm-associated connection.
37 type Conn struct {
39 - ssConn *ss.Connection
38 + pstConn pst.Conn
39 netConn net.Conn // underlying connection
40
41 swarm *Swarm
@@ -46,10 +45,10 @@ type Conn struct {
45 streamLock sync.RWMutex
46 }
47
49 -func newConn(nconn net.Conn, sconn *ss.Connection, s *Swarm) *Conn {
48 +func newConn(nconn net.Conn, tconn pst.Conn, s *Swarm) *Conn {
49 return &Conn{
50 netConn: nconn,
52 - ssConn: sconn,
51 + pstConn: tconn,
52 swarm: s,
53 groups: groupSet{m: make(map[Group]struct{})},
54 streams: make(map[*Stream]struct{}),
@@ -66,10 +65,10 @@ func (c *Conn) NetConn() net.Conn {
65 return c.netConn
66 }
67
69 -// SPDYConn returns the spdystream.Connection we use
68 +// Conn returns the underlying transport Connection we use
69 // Warning: modifying this object is undefined.
71 -func (c *Conn) SPDYConn() *ss.Connection {
72 - return c.ssConn
70 +func (c *Conn) Conn() pst.Conn {
71 + return c.pstConn
72 }
73
74 // Groups returns the Groups this Conn belongs to
@@ -144,7 +143,7 @@ func ConnInConns(c1 *Conn, conns []*Conn) bool {
143
144 // addConn is the internal version of AddConn. we need the server bool
145 // as spdystream requires it.
147 -func (s *Swarm) addConn(netConn net.Conn, server bool) (*Conn, error) {
146 +func (s *Swarm) addConn(netConn net.Conn, isServer bool) (*Conn, error) {
147 if netConn == nil {
148 return nil, errors.New("nil conn")
149 }
@@ -164,7 +163,7 @@ func (s *Swarm) addConn(netConn net.Conn, server bool) (*Conn, error) {
163 }
164
165 // create a new spdystream connection
167 - ssConn, err := ss.NewConnection(netConn, server)
166 + ssConn, err := s.transport.NewConn(netConn, isServer)
167 if err != nil {
168 return nil, err
169 }
@@ -183,10 +182,9 @@ func (s *Swarm) addConn(netConn net.Conn, server bool) (*Conn, error) {
182 s.ConnHandler()(c)
183
184 // go listen for incoming streams on this connection
186 - go c.ssConn.Serve(func(ssS *ss.Stream) {
185 + go c.pstConn.Serve(func(ss pst.Stream) {
186 // log.Printf("accepted stream %d from %s\n", ssS.Identifier(), netConn.RemoteAddr())
188 - ssS.SendReply(http.Header{}, false)
189 - stream := s.setupSSStream(ssS, c)
187 + stream := s.setupStream(ss, c)
188 s.StreamHandler()(stream) // call our handler
189 })
190
@@ -197,25 +195,23 @@ func (s *Swarm) addConn(netConn net.Conn, server bool) (*Conn, error) {
195 // all validation has happened.
196 func (s *Swarm) createStream(c *Conn) (*Stream, error) {
197
200 - // Create a new ss.Stream
201 - ssStream, err := c.ssConn.CreateStream(http.Header{}, nil, false)
198 + // Create a new pst.Stream
199 + pstStream, err := c.pstConn.OpenStream()
200 if err != nil {
201 return nil, err
202 }
203
206 - // create a new stream
207 - return s.setupSSStream(ssStream, c), nil
204 + return s.setupStream(pstStream, c), nil
205 }
206
207 // newStream is the internal function that creates a new stream. assumes
208 // all validation has happened.
212 -func (s *Swarm) setupSSStream(ssS *ss.Stream, c *Conn) *Stream {
213 - // create a new *Stream
214 - stream := newStream(ssS, c)
209 +func (s *Swarm) setupStream(pstStream pst.Stream, c *Conn) *Stream {
210
216 - // add it to our streams maps
211 + // create a new stream
212 + stream := newStream(pstStream, c)
213
218 - // add it to our map
214 + // add it to our streams maps
215 s.streamLock.Lock()
216 c.streamLock.Lock()
217 s.streams[stream] = struct{}{}
@@ -235,12 +231,7 @@ func (s *Swarm) removeStream(stream *Stream) error {
231 s.streamLock.Unlock()
232 stream.conn.streamLock.Unlock()
233
238 - // Reset is spdystream's full bidirectional close.
239 - // We expose bidirectional close as our `Close`.
240 - // To close only half of the connection, and use other
241 - // spdystream options, just get the stream with:
242 - // stream.SPDYStream()
243 - return stream.ssStream.Reset()
234 + return stream.pstStream.Close()
235 }
236
237 func (s *Swarm) removeConn(conn *Conn) error {
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/.gitignore new
+1
@@ -0,0 +1 @@
1 +example
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/blockhandler/blockhandler.go
+3 -3
@@ -8,6 +8,7 @@ import (
8 "time"
9
10 ps "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
11 + pstss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream"
12 )
13
14 func die(err error) {
@@ -17,7 +18,7 @@ func die(err error) {
18
19 func main() {
20 // create a new Swarm
20 - swarm := ps.NewSwarm()
21 + swarm := ps.NewSwarm(pstss.Transport)
22 defer swarm.Close()
23
24 // tell swarm what to do with a new incoming streams.
@@ -62,13 +63,12 @@ func main() {
63 nSndStream := 0
64 for {
65 <-time.After(200 * time.Millisecond)
65 - s, err := swarm.NewStreamWithConn(c)
66 + _, err := swarm.NewStreamWithConn(c)
67 if err != nil {
68 die(err)
69 }
70 log("sender got new stream %d", nSndStream)
71 nSndStream++
71 - s.Wait()
72 }
73 }
74
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/closer/closer
Binary files /dev/null and b/Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/closer/closer differ
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/closer/closer.go new
+90
@@ -0,0 +1,90 @@
1 +package main
2 +
3 +import (
4 + "fmt"
5 + "net"
6 + "os"
7 + "time"
8 +
9 + ps "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
10 +)
11 +
12 +func die(err error) {
13 + fmt.Fprintf(os.Stderr, "error: %s\n")
14 + os.Exit(1)
15 +}
16 +
17 +func main() {
18 + // create a new Swarm
19 + swarm := ps.NewSwarm()
20 + defer swarm.Close()
21 +
22 + // tell swarm what to do with a new incoming streams.
23 + // EchoHandler just echos back anything they write.
24 + swarm.SetStreamHandler(ps.EchoHandler)
25 +
26 + l, err := net.Listen("tcp", "localhost:8001")
27 + if err != nil {
28 + die(err)
29 + }
30 +
31 + if _, err := swarm.AddListener(l); err != nil {
32 + die(err)
33 + }
34 +
35 + nc, err := net.Dial("tcp", "localhost:8001")
36 + if err != nil {
37 + die(err)
38 + }
39 +
40 + c, err := swarm.AddConn(nc)
41 + if err != nil {
42 + die(err)
43 + }
44 +
45 + hello := []byte("hello")
46 + goodbye := []byte("goodbye")
47 + swarm.SetStreamHandler(func(s *ps.Stream) {
48 + go func() {
49 + log("handler: got new stream.")
50 + // s.Wait()
51 + // log("handler: done waiting on new stream.")
52 + buf := make([]byte, len(hello))
53 + s.Read(buf)
54 + log("handler: read: %s", buf)
55 + s.Write(goodbye)
56 + log("handler: wrote: %s", goodbye)
57 + s.Close()
58 + log("handler: closed.")
59 + }()
60 + })
61 +
62 + for {
63 + s, err := swarm.NewStreamWithConn(c)
64 + if err != nil {
65 + die(err)
66 + }
67 + // s.Wait()
68 + log("sender: got new stream")
69 + for {
70 + <-time.After(500 * time.Millisecond)
71 + log("sender: writing hello...")
72 + if _, err := s.Write(hello); err != nil {
73 + log("sender: write error: %s", err)
74 + break
75 + }
76 + buf := make([]byte, len(goodbye))
77 + if _, err := s.Read(buf); err != nil {
78 + log("sender: read error: %s", err)
79 + break
80 + }
81 + }
82 + if err := s.Close(); err != nil {
83 + log("sender: close error: %s", err)
84 + }
85 + }
86 +}
87 +
88 +func log(s string, ifs ...interface{}) {
89 + fmt.Fprintf(os.Stderr, s+"\n", ifs...)
90 +}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/example
Binary files a/Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/example and b/Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/example differ
Godeps/_workspace/src/github.com/jbenet/go-peerstream/example/example.go
+16 -16
@@ -7,23 +7,28 @@ import (
7 "os"
8
9 ps "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
10 + pstss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream"
11 )
12
13 func main() {
13 - // create a new Swarm
14 - swarm := ps.NewSwarm()
14 +
15 + log("creating a new swarm with spdystream transport") // create a new Swarm
16 + swarm := ps.NewSwarm(pstss.Transport)
17 defer swarm.Close()
18
19 // tell swarm what to do with a new incoming streams.
20 // EchoHandler just echos back anything they write.
21 + log("setup EchoHandler")
22 swarm.SetStreamHandler(ps.EchoHandler)
23
24 // Okay, let's try listening on some transports
25 + log("listening at localhost:8001")
26 l1, err := net.Listen("tcp", "localhost:8001")
27 if err != nil {
28 panic(err)
29 }
30
31 + log("listening at localhost:8002")
32 l2, err := net.Listen("tcp", "localhost:8002")
33 if err != nil {
34 panic(err)
@@ -39,11 +44,13 @@ func main() {
44 }
45
46 // ok, let's try some outgoing connections
47 + log("dialing localhost:8001")
48 nc1, err := net.Dial("tcp", "localhost:8001")
49 if err != nil {
50 panic(err)
51 }
52
53 + log("dialing localhost:8002")
54 nc2, err := net.Dial("tcp", "localhost:8002")
55 if err != nil {
56 panic(err)
@@ -66,6 +73,7 @@ func main() {
73
74 // now let's try opening some streams!
75 // You can specify what connection you want to use
76 + log("opening stream with NewStreamWithConn(c1)")
77 s1, err := swarm.NewStreamWithConn(c1)
78 if err != nil {
79 panic(err)
@@ -73,6 +81,7 @@ func main() {
81
82 // Or, you can specify a SelectConn function that picks between all
83 // (it calls NewStreamWithConn underneath the hood)
84 + log("opening stream with NewStreamSelectConn(.)")
85 s2, err := swarm.NewStreamSelectConn(func(conns []*ps.Conn) *ps.Conn {
86 if len(conns) > 0 {
87 return conns[0]
@@ -92,6 +101,7 @@ func main() {
101 // connection it finds in that group, using a SelectConn you can rebind:
102 // swarm.SetGroupSelectConn(1, SelectConn)
103 // swarm.SetDegaultGroupSelectConn(SelectConn)
104 + log("opening stream with NewStreamWithGroup(1)")
105 s3, err := swarm.NewStreamWithGroup(1)
106 if err != nil {
107 panic(err)
@@ -106,33 +116,23 @@ func main() {
116 // streams from github.com/docker/spdystream, so they work the same
117 // way:
118
119 + log("preparing the streams")
120 for i, stream := range []*ps.Stream{s1, s2, s3} {
110 - stream.Wait()
121 str := "stream %d ready:"
122 fmt.Fprintf(stream, str, i)
123
124 buf := make([]byte, len(str))
125 + log(fmt.Sprintf("reading from stream %d", i))
126 stream.Read(buf)
127 fmt.Println(string(buf))
128 }
129
130 + log("let's test the streams")
131 + log("enter some text below:\n")
132 go io.Copy(os.Stdout, s1)
133 go io.Copy(os.Stdout, s2)
134 go io.Copy(os.Stdout, s3)
135 io.Copy(io.MultiWriter(s1, s2, s3), os.Stdin)
123 -
124 - // r := peerstream.ProtoRouter()
125 - // r.AddRoute("bitswap", BitswapHandler)
126 - // r.AddRoute("dht", DHTHandler)
127 - // r.AddRoute("id", IDHandler)
128 -
129 - // // The router's StreamHandler does this
130 - // swarm.SetStreamHandler(router.StreamHandler())
131 -
132 - // func (r *router) StreamHandler(s Stream) {
133 -
134 - // }
135 -
136 }
137
138 func log(s string) {
Godeps/_workspace/src/github.com/jbenet/go-peerstream/stream.go
+10 -15
@@ -1,7 +1,7 @@
1 package peerstream
2
3 import (
4 - ss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/spdystream"
4 + pst "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
5 )
6
7 // StreamHandler is a function which receives a Stream. It
@@ -15,25 +15,25 @@ type StreamHandler func(s *Stream)
15 // Stream is an io.{Read,Write,Close}r to a remote counterpart.
16 // It wraps a spdystream.Stream, and links it to a Conn and groups
17 type Stream struct {
18 - ssStream *ss.Stream
18 + pstStream pst.Stream
19
20 conn *Conn
21 groups groupSet
22 }
23
24 -func newStream(ssS *ss.Stream, c *Conn) *Stream {
24 +func newStream(ss pst.Stream, c *Conn) *Stream {
25 s := &Stream{
26 - conn: c,
27 - ssStream: ssS,
28 - groups: groupSet{m: make(map[Group]struct{})},
26 + conn: c,
27 + pstStream: ss,
28 + groups: groupSet{m: make(map[Group]struct{})},
29 }
30 s.groups.AddSet(&c.groups) // inherit groups
31 return s
32 }
33
34 // SPDYStream returns the underlying *spdystream.Stream
35 -func (s *Stream) SPDYStream() *ss.Stream {
36 - return s.ssStream
35 +func (s *Stream) Stream() pst.Stream {
36 + return s.pstStream
37 }
38
39 // Conn returns the Conn associated with this Stream
@@ -61,17 +61,12 @@ func (s *Stream) AddGroup(g Group) {
61 s.groups.Add(g)
62 }
63
64 -// Write writes bytes to a stream, calling write data for each call.
65 -func (s *Stream) Wait() error {
66 - return s.ssStream.Wait()
67 -}
68 -
64 func (s *Stream) Read(p []byte) (n int, err error) {
70 - return s.ssStream.Read(p)
65 + return s.pstStream.Read(p)
66 }
67
68 func (s *Stream) Write(p []byte) (n int, err error) {
74 - return s.ssStream.Write(p)
69 + return s.pstStream.Write(p)
70 }
71
72 func (s *Stream) Close() error {
Godeps/_workspace/src/github.com/jbenet/go-peerstream/swarm.go
+7 -1
@@ -4,12 +4,17 @@ import (
4 "errors"
5 "net"
6 "sync"
7 +
8 + pst "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
9 )
10
11 // fd is a (file) descriptor, unix style
12 type fd uint32
13
14 type Swarm struct {
15 + // the transport we'll use.
16 + transport pst.Transport
17 +
18 // active streams.
19 streams map[*Stream]struct{}
20 streamLock sync.RWMutex
@@ -30,8 +35,9 @@ type Swarm struct {
35 selectConn SelectConn // default SelectConn function
36 }
37
33 -func NewSwarm() *Swarm {
38 +func NewSwarm(t pst.Transport) *Swarm {
39 return &Swarm{
40 + transport: t,
41 streams: make(map[*Stream]struct{}),
42 conns: make(map[*Conn]struct{}),
43 listeners: make(map[*Listener]struct{}),
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/muxado/muxado.go new
+80
@@ -0,0 +1,80 @@
1 +package peerstream_muxado
2 +
3 +import (
4 + "net"
5 +
6 + muxado "github.com/inconshreveable/muxado"
7 + pst "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
8 +)
9 +
10 +// stream implements pst.Stream using a ss.Stream
11 +type stream struct {
12 + ms muxado.Stream
13 +}
14 +
15 +func (s *stream) muxadoStream() muxado.Stream {
16 + return s.ms
17 +}
18 +
19 +func (s *stream) Read(buf []byte) (int, error) {
20 + return s.ms.Read(buf)
21 +}
22 +
23 +func (s *stream) Write(buf []byte) (int, error) {
24 + return s.ms.Write(buf)
25 +}
26 +
27 +func (s *stream) Close() error {
28 + return s.ms.Close()
29 +}
30 +
31 +// Conn is a connection to a remote peer.
32 +type conn struct {
33 + ms muxado.Session
34 +}
35 +
36 +func (c *conn) muxadoSession() muxado.Session {
37 + return c.ms
38 +}
39 +
40 +func (c *conn) Close() error {
41 + return c.ms.Close()
42 +}
43 +
44 +// OpenStream creates a new stream.
45 +func (c *conn) OpenStream() (pst.Stream, error) {
46 + s, err := c.ms.Open()
47 + if err != nil {
48 + return nil, err
49 + }
50 +
51 + return &stream{ms: s}, nil
52 +}
53 +
54 +// Serve starts listening for incoming requests and handles them
55 +// using given StreamHandler
56 +func (c *conn) Serve(handler pst.StreamHandler) {
57 + for { // accept loop
58 + s, err := c.ms.Accept()
59 + if err != nil {
60 + return // err always means closed.
61 + }
62 + go handler(&stream{ms: s})
63 + }
64 +}
65 +
66 +type transport struct{}
67 +
68 +// Transport is a go-peerstream transport that constructs
69 +// spdystream-backed connections.
70 +var Transport = transport{}
71 +
72 +func (t transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
73 + var s muxado.Session
74 + if isServer {
75 + s = muxado.Server(nc)
76 + } else {
77 + s = muxado.Client(nc)
78 + }
79 + return &conn{ms: s}, nil
80 +}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/muxado/muxado_test.go new
+11
@@ -0,0 +1,11 @@
1 +package peerstream_muxado
2 +
3 +import (
4 + "testing"
5 +
6 + psttest "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test"
7 +)
8 +
9 +func TestMuxadoTransport(t *testing.T) {
10 + psttest.SubtestAll(t, Transport)
11 +}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream/spdystream.go new
+89
@@ -0,0 +1,89 @@
1 +package peerstream_spdystream
2 +
3 +import (
4 + "net"
5 + "net/http"
6 +
7 + pst "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
8 + ss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/spdystream"
9 +)
10 +
11 +// stream implements pst.Stream using a ss.Stream
12 +type stream ss.Stream
13 +
14 +func (s *stream) spdyStream() *ss.Stream {
15 + return (*ss.Stream)(s)
16 +}
17 +
18 +func (s *stream) Read(buf []byte) (int, error) {
19 + return s.spdyStream().Read(buf)
20 +}
21 +
22 +func (s *stream) Write(buf []byte) (int, error) {
23 + return s.spdyStream().Write(buf)
24 +}
25 +
26 +func (s *stream) Close() error {
27 + // Reset is spdystream's full bidirectional close.
28 + // We expose bidirectional close as our `Close`.
29 + // To close only half of the connection, and use other
30 + // spdystream options, just get the stream with:
31 + // ssStream := (*ss.Stream)(stream)
32 + return s.spdyStream().Reset()
33 +}
34 +
35 +// Conn is a connection to a remote peer.
36 +type conn ss.Connection
37 +
38 +func (c *conn) spdyConn() *ss.Connection {
39 + return (*ss.Connection)(c)
40 +}
41 +
42 +func (c *conn) Close() error {
43 + return c.spdyConn().Close()
44 +}
45 +
46 +// OpenStream creates a new stream.
47 +func (c *conn) OpenStream() (pst.Stream, error) {
48 + s, err := c.spdyConn().CreateStream(http.Header{}, nil, false)
49 + if err != nil {
50 + return nil, err
51 + }
52 +
53 + // wait for a response before writing. for some reason
54 + // spdystream does not make forward progress unless you do this.
55 + s.Wait()
56 + return (*stream)(s), nil
57 +}
58 +
59 +// Serve starts listening for incoming requests and handles them
60 +// using given StreamHandler
61 +func (c *conn) Serve(handler pst.StreamHandler) {
62 + c.spdyConn().Serve(func(s *ss.Stream) {
63 +
64 + // Flow control and backpressure of Opening streams is broken.
65 + // I believe that spdystream has one set of workers that both send
66 + // data AND accept new streams (as it's just more data). there
67 + // is a problem where if the new stream handlers want to throttle,
68 + // they also eliminate the ability to read/write data, which makes
69 + // forward-progress impossible. Thus, throttling this function is
70 + // -- at this moment -- not the solution. Either spdystream must
71 + // change, or we must throttle another way. go-peerstream handles
72 + // every new stream in its own goroutine.
73 + go func() {
74 + s.SendReply(http.Header{}, false)
75 + handler((*stream)(s))
76 + }()
77 + })
78 +}
79 +
80 +type transport struct{}
81 +
82 +// Transport is a go-peerstream transport that constructs
83 +// spdystream-backed connections.
84 +var Transport = transport{}
85 +
86 +func (t transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
87 + c, err := ss.NewConnection(nc, isServer)
88 + return (*conn)(c), err
89 +}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream/spdystream_test.go new
+11
@@ -0,0 +1,11 @@
1 +package peerstream_spdystream
2 +
3 +import (
4 + "testing"
5 +
6 + psttest "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test"
7 +)
8 +
9 +func TestSpdyStreamTransport(t *testing.T) {
10 + psttest.SubtestAll(t, Transport)
11 +}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/test/ttest.go new
+438
@@ -0,0 +1,438 @@
1 +package peerstream_transport_test
2 +
3 +import (
4 + "bytes"
5 + crand "crypto/rand"
6 + "fmt"
7 + "io"
8 + mrand "math/rand"
9 + "net"
10 + "os"
11 + "reflect"
12 + "runtime"
13 + "sync"
14 + "testing"
15 +
16 + ps "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
17 + pst "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
18 +)
19 +
20 +var randomness []byte
21 +var nextPort = 20000
22 +
23 +func init() {
24 + // read 1MB of randomness
25 + randomness = make([]byte, 1<<20)
26 + if _, err := crand.Read(randomness); err != nil {
27 + panic(err)
28 + }
29 +}
30 +
31 +func randBuf(size int) []byte {
32 + n := len(randomness) - size
33 + if size < 1 {
34 + panic(fmt.Errorf("requested too large buffer (%d). max is %d", size, len(randomness)))
35 + }
36 +
37 + start := mrand.Intn(n)
38 + return randomness[start : start+size]
39 +}
40 +
41 +func checkErr(t *testing.T, err error) {
42 + if err != nil {
43 + t.Fatal(err)
44 + }
45 +}
46 +
47 +func getNextPort() int {
48 + nextPort++
49 + return nextPort
50 +}
51 +
52 +func log(s string, v ...interface{}) {
53 + if testing.Verbose() {
54 + fmt.Fprintf(os.Stderr, "> "+s+"\n", v...)
55 + }
56 +}
57 +
58 +type echoSetup struct {
59 + swarm *ps.Swarm
60 + conns []*ps.Conn
61 +}
62 +
63 +func singleConn(t *testing.T, tr pst.Transport) echoSetup {
64 + swarm := ps.NewSwarm(tr)
65 + swarm.SetStreamHandler(func(s *ps.Stream) {
66 + defer s.Close()
67 + log("accepted stream")
68 + io.Copy(s, s) // echo everything
69 + log("closing stream")
70 + })
71 +
72 + port := getNextPort()
73 + addr := fmt.Sprintf("localhost:%d", port)
74 + log("listening at %s", addr)
75 + l, err := net.Listen("tcp", addr)
76 + checkErr(t, err)
77 +
78 + _, err = swarm.AddListener(l)
79 + checkErr(t, err)
80 +
81 + log("dialing to %s", addr)
82 + nc1, err := net.Dial("tcp", addr)
83 + checkErr(t, err)
84 +
85 + c1, err := swarm.AddConn(nc1)
86 + checkErr(t, err)
87 +
88 + return echoSetup{
89 + swarm: swarm,
90 + conns: []*ps.Conn{c1},
91 + }
92 +}
93 +
94 +func makeSwarm(t *testing.T, tr pst.Transport, nListeners int) *ps.Swarm {
95 + swarm := ps.NewSwarm(tr)
96 + swarm.SetStreamHandler(func(s *ps.Stream) {
97 + defer s.Close()
98 + log("accepted stream")
99 + io.Copy(s, s) // echo everything
100 + log("closing stream")
101 + })
102 +
103 + for i := 0; i < nListeners; i++ {
104 + port := getNextPort()
105 + addr := fmt.Sprintf("localhost:%d", port)
106 + log("%p listening at %s", swarm, addr)
107 + l, err := net.Listen("tcp", addr)
108 + checkErr(t, err)
109 + _, err = swarm.AddListener(l)
110 + checkErr(t, err)
111 + }
112 +
113 + return swarm
114 +}
115 +
116 +func makeSwarms(t *testing.T, tr pst.Transport, nSwarms, nListeners int) []*ps.Swarm {
117 + swarms := make([]*ps.Swarm, nSwarms)
118 + for i := 0; i < nSwarms; i++ {
119 + swarms[i] = makeSwarm(t, tr, nListeners)
120 + }
121 + return swarms
122 +}
123 +
124 +func SubtestConstructSwarm(t *testing.T, tr pst.Transport) {
125 + ps.NewSwarm(tr)
126 +}
127 +
128 +func SubtestSimpleWrite(t *testing.T, tr pst.Transport) {
129 + swarm := ps.NewSwarm(tr)
130 + defer swarm.Close()
131 +
132 + piper, pipew := io.Pipe()
133 + swarm.SetStreamHandler(func(s *ps.Stream) {
134 + defer s.Close()
135 + log("accepted stream")
136 + w := io.MultiWriter(s, pipew)
137 + io.Copy(w, s) // echo everything and write it to pipew
138 + log("closing stream")
139 + })
140 +
141 + port := getNextPort()
142 + addr := fmt.Sprintf("localhost:%d", port)
143 + log("listening at %s", addr)
144 + l, err := net.Listen("tcp", addr)
145 + checkErr(t, err)
146 +
147 + _, err = swarm.AddListener(l)
148 + checkErr(t, err)
149 +
150 + log("dialing to %s", addr)
151 + nc1, err := net.Dial("tcp", addr)
152 + checkErr(t, err)
153 +
154 + c1, err := swarm.AddConn(nc1)
155 + checkErr(t, err)
156 + defer c1.Close()
157 +
158 + log("creating stream")
159 + s1, err := c1.NewStream()
160 + checkErr(t, err)
161 + defer s1.Close()
162 +
163 + buf1 := randBuf(4096)
164 + log("writing %d bytes to stream", len(buf1))
165 + _, err = s1.Write(buf1)
166 + checkErr(t, err)
167 +
168 + buf2 := make([]byte, len(buf1))
169 + log("reading %d bytes from stream (echoed)", len(buf2))
170 + _, err = s1.Read(buf2)
171 + checkErr(t, err)
172 + if string(buf2) != string(buf1) {
173 + t.Error("buf1 and buf2 not equal: %s != %s", string(buf1), string(buf2))
174 + }
175 +
176 + buf3 := make([]byte, len(buf1))
177 + log("reading %d bytes from pipe (tee)", len(buf3))
178 + _, err = piper.Read(buf3)
179 + checkErr(t, err)
180 + if string(buf3) != string(buf1) {
181 + t.Error("buf1 and buf3 not equal: %s != %s", string(buf1), string(buf3))
182 + }
183 +}
184 +
185 +func SubtestSimpleWrite100msgs(t *testing.T, tr pst.Transport) {
186 +
187 + msgs := 100
188 + msgsize := 1 << 19
189 + es := singleConn(t, tr)
190 +
191 + log("creating stream")
192 + stream, err := es.conns[0].NewStream()
193 + checkErr(t, err)
194 +
195 + bufs := make(chan []byte, msgs)
196 + errs := make(chan error, msgs*100)
197 + var wg sync.WaitGroup
198 +
199 + wg.Add(1)
200 + go func() {
201 + defer wg.Done()
202 +
203 + for i := 0; i < msgs; i++ {
204 + buf := randBuf(msgsize)
205 + bufs <- buf
206 + log("writing %d bytes (message %d/%d #%x)", len(buf), i, msgs, buf[:3])
207 + if _, err := stream.Write(buf); err != nil {
208 + errs <- err
209 + continue
210 + }
211 + }
212 + close(bufs)
213 + }()
214 +
215 + wg.Add(1)
216 + go func() {
217 + defer wg.Done()
218 +
219 + buf2 := make([]byte, msgsize)
220 + i := 0
221 + for buf1 := range bufs {
222 + log("reading %d bytes (message %d/%d #%x)", len(buf1), i, msgs, buf1[:3])
223 + i++
224 +
225 + if _, err := io.ReadFull(stream, buf2); err != nil {
226 + errs <- err
227 + continue
228 + }
229 + if !bytes.Equal(buf1, buf2) {
230 + errs <- fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3])
231 + }
232 + }
233 + }()
234 +
235 + wg.Wait()
236 + close(errs)
237 + for err := range errs {
238 + t.Error(err)
239 + }
240 +}
241 +
242 +func SubtestStressNSwarmNConnNStreamNMsg(t *testing.T, tr pst.Transport, nSwarm, nConn, nStream, nMsg int) {
243 +
244 + msgsize := 1 << 11
245 + errs := make(chan error, nSwarm*nConn*nStream*nMsg*100) // dont block anything.
246 +
247 + rateLimitN := 5000
248 + rateLimitChan := make(chan struct{}, rateLimitN) // max of 5k funcs.
249 + for i := 0; i < rateLimitN; i++ {
250 + rateLimitChan <- struct{}{}
251 + }
252 +
253 + rateLimit := func(f func()) {
254 + <-rateLimitChan
255 + f()
256 + rateLimitChan <- struct{}{}
257 + }
258 +
259 + writeStream := func(s *ps.Stream, bufs chan<- []byte) {
260 + log("writeStream %p, %d nMsg", s, nMsg)
261 +
262 + for i := 0; i < nMsg; i++ {
263 + buf := randBuf(msgsize)
264 + bufs <- buf
265 + log("%p writing %d bytes (message %d/%d #%x)", s, len(buf), i, nMsg, buf[:3])
266 + if _, err := s.Write(buf); err != nil {
267 + errs <- err
268 + continue
269 + }
270 + }
271 + }
272 +
273 + readStream := func(s *ps.Stream, bufs <-chan []byte) {
274 + log("readStream %p, %d nMsg", s, nMsg)
275 +
276 + buf2 := make([]byte, msgsize)
277 + i := 0
278 + for buf1 := range bufs {
279 + log("%p reading %d bytes (message %d/%d #%x)", s, len(buf1), i, nMsg, buf1[:3])
280 + i++
281 +
282 + if _, err := io.ReadFull(s, buf2); err != nil {
283 + errs <- err
284 + continue
285 + }
286 + if !bytes.Equal(buf1, buf2) {
287 + errs <- fmt.Errorf("buffers not equal (%x != %x)", buf1[:3], buf2[:3])
288 + }
289 + }
290 + }
291 +
292 + openStreamAndRW := func(c *ps.Conn) {
293 + log("openStreamAndRW %p, %d nMsg", c, nMsg)
294 +
295 + s, err := c.NewStream()
296 + if err != nil {
297 + errs <- fmt.Errorf("Failed to create NewStream: %s", err)
298 + return
299 + }
300 +
301 + bufs := make(chan []byte, nMsg)
302 + go func() {
303 + writeStream(s, bufs)
304 + close(bufs)
305 + }()
306 +
307 + readStream(s, bufs)
308 + s.Close()
309 + }
310 +
311 + openConnAndRW := func(a, b *ps.Swarm) {
312 + log("openConnAndRW %p -> %p, %d nStream", a, b, nConn)
313 +
314 + ls := b.Listeners()
315 + l := ls[mrand.Intn(len(ls))]
316 + nl := l.NetListener()
317 + nla := nl.Addr()
318 +
319 + nc, err := net.Dial(nla.Network(), nla.String())
320 + if err != nil {
321 + errs <- err
322 + return
323 + }
324 +
325 + c, err := a.AddConn(nc)
326 + if err != nil {
327 + errs <- err
328 + return
329 + }
330 +
331 + var wg sync.WaitGroup
332 + for i := 0; i < nStream; i++ {
333 + wg.Add(1)
334 + go rateLimit(func() {
335 + defer wg.Done()
336 + openStreamAndRW(c)
337 + })
338 + }
339 + wg.Wait()
340 + c.Close()
341 + }
342 +
343 + openConnsAndRW := func(a, b *ps.Swarm) {
344 + log("openConnsAndRW %p -> %p, %d conns", a, b, nConn)
345 +
346 + var wg sync.WaitGroup
347 + for i := 0; i < nConn; i++ {
348 + wg.Add(1)
349 + go rateLimit(func() {
350 + defer wg.Done()
351 + openConnAndRW(a, b)
352 + })
353 + }
354 + wg.Wait()
355 + }
356 +
357 + connectSwarmsAndRW := func(swarms []*ps.Swarm) {
358 + log("connectSwarmsAndRW %d swarms", len(swarms))
359 +
360 + var wg sync.WaitGroup
361 + for _, a := range swarms {
362 + for _, b := range swarms {
363 + wg.Add(1)
364 + go rateLimit(func() {
365 + defer wg.Done()
366 + openConnsAndRW(a, b)
367 + })
368 + }
369 + }
370 + wg.Wait()
371 + }
372 +
373 + swarms := makeSwarms(t, tr, nSwarm, 3) // 3 listeners per swarm.
374 +
375 + go func() {
376 + connectSwarmsAndRW(swarms)
377 + close(errs) // done
378 + }()
379 +
380 + for err := range errs {
381 + t.Error(err)
382 + }
383 +
384 +}
385 +
386 +func SubtestStress1Swarm1Conn1Stream1Msg(t *testing.T, tr pst.Transport) {
387 + SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 1, 1, 1)
388 +}
389 +
390 +func SubtestStress1Swarm1Conn1Stream100Msg(t *testing.T, tr pst.Transport) {
391 + SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 1, 1, 100)
392 +}
393 +
394 +func SubtestStress1Swarm1Conn100Stream100Msg(t *testing.T, tr pst.Transport) {
395 + SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 1, 100, 100)
396 +}
397 +
398 +func SubtestStress1Swarm10Conn50Stream50Msg(t *testing.T, tr pst.Transport) {
399 + SubtestStressNSwarmNConnNStreamNMsg(t, tr, 1, 10, 50, 50)
400 +}
401 +
402 +func SubtestStress5Swarm2Conn20Stream20Msg(t *testing.T, tr pst.Transport) {
403 + SubtestStressNSwarmNConnNStreamNMsg(t, tr, 5, 2, 20, 20)
404 +}
405 +
406 +func SubtestStress10Swarm2Conn100Stream100Msg(t *testing.T, tr pst.Transport) {
407 + SubtestStressNSwarmNConnNStreamNMsg(t, tr, 10, 2, 100, 100)
408 +}
409 +
410 +func SubtestAll(t *testing.T, tr pst.Transport) {
411 +
412 + tests := []TransportTest{
413 + SubtestConstructSwarm,
414 + SubtestSimpleWrite,
415 + SubtestSimpleWrite100msgs,
416 + SubtestStress1Swarm1Conn1Stream1Msg,
417 + SubtestStress1Swarm1Conn1Stream100Msg,
418 + SubtestStress1Swarm1Conn100Stream100Msg,
419 + SubtestStress1Swarm10Conn50Stream50Msg,
420 + SubtestStress5Swarm2Conn20Stream20Msg,
421 + // SubtestStress10Swarm2Conn100Stream100Msg, <-- this hoses the osx network stack...
422 + }
423 +
424 + for _, f := range tests {
425 + if testing.Verbose() {
426 + fmt.Fprintf(os.Stderr, "==== RUN %s\n", GetFunctionName(f))
427 + }
428 + f(t, tr)
429 + }
430 +}
431 +
432 +type TransportTest func(t *testing.T, tr pst.Transport)
433 +
434 +func TestNoOp(t *testing.T) {}
435 +
436 +func GetFunctionName(i interface{}) string {
437 + return runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
438 +}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/transport.go new
+36
@@ -0,0 +1,36 @@
1 +package peerstream_transport
2 +
3 +import (
4 + "io"
5 + "net"
6 +)
7 +
8 +// Stream is a bidirectional io pipe within a connection
9 +type Stream interface {
10 + io.Reader
11 + io.Writer
12 + io.Closer
13 +}
14 +
15 +// StreamHandler is a function that handles streams
16 +// (usually those opened by the remote side)
17 +type StreamHandler func(Stream)
18 +
19 +// Conn is a stream-multiplexing connection to a remote peer.
20 +type Conn interface {
21 + io.Closer
22 +
23 + // OpenStream creates a new stream.
24 + OpenStream() (Stream, error)
25 +
26 + // Serve starts listening for incoming requests and handles them
27 + // using given StreamHandler
28 + Serve(StreamHandler)
29 +}
30 +
31 +// Transport constructs go-peerstream compatible connections.
32 +type Transport interface {
33 +
34 + // NewConn constructs a new connection
35 + NewConn(c net.Conn, isServer bool) (Conn, error)
36 +}
net/mux.go
+12 -27
@@ -89,34 +89,19 @@ func (m *Mux) SetHandler(p ProtocolID, h StreamHandler) {
89
90 // Handle reads the next name off the Stream, and calls a function
91 func (m *Mux) Handle(s Stream) {
92 + ctx := context.Background()
93
93 - // Flow control and backpressure of Opening streams is broken.
94 - // I believe that spdystream has one set of workers that both send
95 - // data AND accept new streams (as it's just more data). there
96 - // is a problem where if the new stream handlers want to throttle,
97 - // they also eliminate the ability to read/write data, which makes
98 - // forward-progress impossible. Thus, throttling this function is
99 - // -- at this moment -- not the solution. Either spdystream must
100 - // change, or we must throttle another way.
101 - //
102 - // In light of this, we use a goroutine for now (otherwise the
103 - // spdy worker totally blocks, and we can't even read the protocol
104 - // header). The better route in the future is to use a worker pool.
105 - go func() {
106 - ctx := context.Background()
107 -
108 - name, handler, err := m.ReadProtocolHeader(s)
109 - if err != nil {
110 - err = fmt.Errorf("protocol mux error: %s", err)
111 - log.Error(err)
112 - log.Event(ctx, "muxError", lgbl.Error(err))
113 - return
114 - }
115 -
116 - log.Info("muxer handle protocol: %s", name)
117 - log.Event(ctx, "muxHandle", eventlog.Metadata{"protocol": name})
118 - handler(s)
119 - }()
94 + name, handler, err := m.ReadProtocolHeader(s)
95 + if err != nil {
96 + err = fmt.Errorf("protocol mux error: %s", err)
97 + log.Error(err)
98 + log.Event(ctx, "muxError", lgbl.Error(err))
99 + return
100 + }
101 +
102 + log.Info("muxer handle protocol: %s", name)
103 + log.Event(ctx, "muxHandle", eventlog.Metadata{"protocol": name})
104 + handler(s)
105 }
106
107 // ReadLengthPrefix reads the name from Reader with a length-byte-prefix.
net/swarm/swarm.go
+2 -1
@@ -10,6 +10,7 @@ import (
10 ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
11 ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
12 ps "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream"
13 + psss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/spdystream"
14 )
15
16 var log = eventlog.Logger("swarm2")
@@ -34,7 +35,7 @@ func NewSwarm(ctx context.Context, listenAddrs []ma.Multiaddr,
35 local peer.ID, peers peer.Peerstore) (*Swarm, error) {
36
37 s := &Swarm{
37 - swarm: ps.NewSwarm(),
38 + swarm: ps.NewSwarm(psss.Transport),
39 local: local,
40 peers: peers,
41 cg: ctxgroup.WithContext(ctx),
net/swarm/swarm_stream.go
-5
@@ -22,11 +22,6 @@ func (s *Stream) Conn() *Conn {
22 return (*Conn)(s.Stream().Conn())
23 }
24
25 -// Wait waits for the stream to receive a reply.
26 -func (s *Stream) Wait() error {
27 - return s.Stream().Wait()
28 -}
29 -
25 // Read reads bytes from a stream.
26 func (s *Stream) Read(p []byte) (n int, err error) {
27 return s.Stream().Read(p)