allow multistream to have zero rtt stream opening
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Jul 9, 2015 at 14:42 UTC
2c4eb60961d752e1a02ee12ea94320ab9cfbf7a3
6 files changed
+279
-18
Godeps/Godeps.json
+1
-1
@@ -344,7 +344,7 @@
344
},
345
{
346
"ImportPath": "github.com/whyrusleeping/go-multistream",
347
- "Rev": "c9eea2e3be705b7cfd730351b510cfa12ca038f4"
347
+ "Rev": "30c7a81b6c568654147bf6e106870c5d64ccebc8"
348
},
349
{
350
"ImportPath": "github.com/whyrusleeping/multiaddr-filter",
Godeps/_workspace/src/github.com/whyrusleeping/go-multistream/lazy.go
new
+129
@@ -0,0 +1,129 @@
1
+package multistream
2
+
3
+import (
4
+ "fmt"
5
+ "io"
6
+ "sync"
7
+)
8
+
9
+func NewLazyHandshakeConn(c io.ReadWriteCloser, proto string) io.ReadWriteCloser {
10
+ return &lazyConn{
11
+ proto: proto,
12
+ con: c,
13
+ }
14
+}
15
+
16
+type lazyConn struct {
17
+ rhandshake bool // only accessed by 'Read' should not call read async
18
+
19
+ rhlock sync.Mutex
20
+ rhsync bool //protected by mutex
21
+ rerr error
22
+
23
+ whandshake bool
24
+
25
+ whlock sync.Mutex
26
+ whsync bool
27
+ werr error
28
+
29
+ proto string
30
+ con io.ReadWriteCloser
31
+}
32
+
33
+func (l *lazyConn) Read(b []byte) (int, error) {
34
+ if !l.rhandshake {
35
+ go l.writeHandshake()
36
+ err := l.readHandshake()
37
+ if err != nil {
38
+ return 0, err
39
+ }
40
+
41
+ l.rhandshake = true
42
+ }
43
+
44
+ if len(b) == 0 {
45
+ return 0, nil
46
+ }
47
+
48
+ return l.con.Read(b)
49
+}
50
+
51
+func (l *lazyConn) readHandshake() error {
52
+ l.rhlock.Lock()
53
+ defer l.rhlock.Unlock()
54
+
55
+ // if we've already done this, exit
56
+ if l.rhsync {
57
+ return l.rerr
58
+ }
59
+ l.rhsync = true
60
+
61
+ // read multistream version
62
+ tok, err := ReadNextToken(l.con)
63
+ if err != nil {
64
+ l.rerr = err
65
+ return err
66
+ }
67
+
68
+ if tok != ProtocolID {
69
+ l.rerr = fmt.Errorf("multistream protocol mismatch ( %s != %s )", tok, ProtocolID)
70
+ return l.rerr
71
+ }
72
+
73
+ // read protocol
74
+ tok, err = ReadNextToken(l.con)
75
+ if err != nil {
76
+ l.rerr = err
77
+ return err
78
+ }
79
+
80
+ if tok != l.proto {
81
+ l.rerr = fmt.Errorf("protocol mismatch in lazy handshake ( %s != %s )", tok, l.proto)
82
+ return l.rerr
83
+ }
84
+
85
+ return nil
86
+}
87
+
88
+func (l *lazyConn) writeHandshake() error {
89
+ l.whlock.Lock()
90
+ defer l.whlock.Unlock()
91
+
92
+ if l.whsync {
93
+ return l.werr
94
+ }
95
+
96
+ l.whsync = true
97
+
98
+ err := delimWrite(l.con, []byte(ProtocolID))
99
+ if err != nil {
100
+ l.werr = err
101
+ return err
102
+ }
103
+
104
+ err = delimWrite(l.con, []byte(l.proto))
105
+ if err != nil {
106
+ l.werr = err
107
+ return err
108
+ }
109
+
110
+ return nil
111
+}
112
+
113
+func (l *lazyConn) Write(b []byte) (int, error) {
114
+ if !l.whandshake {
115
+ go l.readHandshake()
116
+ err := l.writeHandshake()
117
+ if err != nil {
118
+ return 0, err
119
+ }
120
+
121
+ l.whandshake = true
122
+ }
123
+
124
+ return l.con.Write(b)
125
+}
126
+
127
+func (l *lazyConn) Close() error {
128
+ return l.con.Close()
129
+}
Godeps/_workspace/src/github.com/whyrusleeping/go-multistream/multistream.go
+19
-11
@@ -100,17 +100,7 @@ loop:
100
101
switch tok {
102
case "ls":
103
- buf := new(bytes.Buffer)
104
- msm.handlerlock.Lock()
105
- for proto, _ := range msm.handlers {
106
- err := delimWrite(buf, []byte(proto))
107
- if err != nil {
108
- msm.handlerlock.Unlock()
109
- return "", nil, err
110
- }
111
- }
112
- msm.handlerlock.Unlock()
113
- err := delimWrite(rwc, buf.Bytes())
103
+ err := msm.Ls(rwc)
104
if err != nil {
105
return "", nil, err
106
}
@@ -138,6 +128,24 @@ loop:
128
129
}
130
131
+func (msm *MultistreamMuxer) Ls(rwc io.Writer) error {
132
+ buf := new(bytes.Buffer)
133
+ msm.handlerlock.Lock()
134
+ for proto, _ := range msm.handlers {
135
+ err := delimWrite(buf, []byte(proto))
136
+ if err != nil {
137
+ msm.handlerlock.Unlock()
138
+ return err
139
+ }
140
+ }
141
+ msm.handlerlock.Unlock()
142
+ err := delimWrite(rwc, buf.Bytes())
143
+ if err != nil {
144
+ return err
145
+ }
146
+ return nil
147
+}
148
+
149
func (msm *MultistreamMuxer) Handle(rwc io.ReadWriteCloser) error {
150
_, h, err := msm.Negotiate(rwc)
151
if err != nil {
Godeps/_workspace/src/github.com/whyrusleeping/go-multistream/multistream_test.go
+106
@@ -118,6 +118,112 @@ func TestSelectOneAndWrite(t *testing.T) {
118
verifyPipe(t, a, b)
119
}
120
121
+func TestLazyConns(t *testing.T) {
122
+ a, b := net.Pipe()
123
+
124
+ mux := NewMultistreamMuxer()
125
+ mux.AddHandler("/a", nil)
126
+ mux.AddHandler("/b", nil)
127
+ mux.AddHandler("/c", nil)
128
+
129
+ la := NewLazyHandshakeConn(a, "/c")
130
+ lb := NewLazyHandshakeConn(b, "/c")
131
+
132
+ verifyPipe(t, la, lb)
133
+}
134
+
135
+func TestLazyAndMux(t *testing.T) {
136
+ a, b := net.Pipe()
137
+
138
+ mux := NewMultistreamMuxer()
139
+ mux.AddHandler("/a", nil)
140
+ mux.AddHandler("/b", nil)
141
+ mux.AddHandler("/c", nil)
142
+
143
+ done := make(chan struct{})
144
+ go func() {
145
+ selected, _, err := mux.Negotiate(a)
146
+ if err != nil {
147
+ t.Fatal(err)
148
+ }
149
+ if selected != "/c" {
150
+ t.Fatal("incorrect protocol selected")
151
+ }
152
+
153
+ msg := make([]byte, 5)
154
+ _, err = a.Read(msg)
155
+ if err != nil {
156
+ t.Fatal(err)
157
+ }
158
+
159
+ close(done)
160
+ }()
161
+
162
+ lb := NewLazyHandshakeConn(b, "/c")
163
+
164
+ // do a write to push the handshake through
165
+ _, err := lb.Write([]byte("hello"))
166
+ if err != nil {
167
+ t.Fatal(err)
168
+ }
169
+
170
+ select {
171
+ case <-time.After(time.Second):
172
+ t.Fatal("failed to complete in time")
173
+ case <-done:
174
+ }
175
+
176
+ verifyPipe(t, a, lb)
177
+}
178
+
179
+func TestLazyAndMuxWrite(t *testing.T) {
180
+ a, b := net.Pipe()
181
+
182
+ mux := NewMultistreamMuxer()
183
+ mux.AddHandler("/a", nil)
184
+ mux.AddHandler("/b", nil)
185
+ mux.AddHandler("/c", nil)
186
+
187
+ done := make(chan struct{})
188
+ go func() {
189
+ selected, _, err := mux.Negotiate(a)
190
+ if err != nil {
191
+ t.Fatal(err)
192
+ }
193
+ if selected != "/c" {
194
+ t.Fatal("incorrect protocol selected")
195
+ }
196
+
197
+ _, err = a.Write([]byte("hello"))
198
+ if err != nil {
199
+ t.Fatal(err)
200
+ }
201
+
202
+ close(done)
203
+ }()
204
+
205
+ lb := NewLazyHandshakeConn(b, "/c")
206
+
207
+ // do a write to push the handshake through
208
+ msg := make([]byte, 5)
209
+ _, err := lb.Read(msg)
210
+ if err != nil {
211
+ t.Fatal(err)
212
+ }
213
+
214
+ if string(msg) != "hello" {
215
+ t.Fatal("wrong!")
216
+ }
217
+
218
+ select {
219
+ case <-time.After(time.Second):
220
+ t.Fatal("failed to complete in time")
221
+ case <-done:
222
+ }
223
+
224
+ verifyPipe(t, a, lb)
225
+}
226
+
227
func verifyPipe(t *testing.T, a, b io.ReadWriter) {
228
mes := make([]byte, 1024)
229
rand.Read(mes)
p2p/host/basic/basic_host.go
+18
-6
@@ -170,12 +170,11 @@ func (h *BasicHost) NewStream(pid protocol.ID, p peer.ID) (inet.Stream, error) {
170
171
logStream := mstream.WrapStream(s, pid, h.bwc)
172
173
- if err := msmux.SelectProtoOrFail(string(pid), logStream); err != nil {
174
- logStream.Close()
175
- return nil, err
176
- }
177
-
178
- return logStream, nil
173
+ lzcon := msmux.NewLazyHandshakeConn(logStream, string(pid))
174
+ return &streamWrapper{
175
+ Stream: logStream,
176
+ rw: lzcon,
177
+ }, nil
178
}
179
180
// Connect ensures there is a connection between this host and the peer with
@@ -254,3 +253,16 @@ func (h *BasicHost) Close() error {
253
func (h *BasicHost) GetBandwidthReporter() metrics.Reporter {
254
return h.bwc
255
}
256
+
257
+type streamWrapper struct {
258
+ inet.Stream
259
+ rw io.ReadWriter
260
+}
261
+
262
+func (s *streamWrapper) Read(b []byte) (int, error) {
263
+ return s.rw.Read(b)
264
+}
265
+
266
+func (s *streamWrapper) Write(b []byte) (int, error) {
267
+ return s.rw.Write(b)
268
+}
p2p/test/backpressure/backpressure_test.go
+6
@@ -299,6 +299,12 @@ func TestStBackpressureStreamWrite(t *testing.T) {
299
}
300
}
301
302
+ // trigger lazy connection handshaking
303
+ _, err = s.Read(nil)
304
+ if err != nil {
305
+ t.Fatal(err)
306
+ }
307
+
308
// 500ms rounds of lockstep write + drain
309
roundsStart := time.Now()
310
roundsTotal := 0