bugfix in spdystream
and updated go-peerstream cc @whyrusleeping
Juan Batiz-Benet committed
Dec 31, 2014 at 00:34 UTC
0fe9f0603b0e28494c4354507ba88d0b1ccb855d
18 files changed
+153
-31
Godeps/Godeps.json
+5
-5
@@ -69,10 +69,6 @@
69
"ImportPath": "github.com/coreos/go-semver/semver",
70
"Rev": "6fe83ccda8fb9b7549c9ab4ba47f47858bc950aa"
71
},
72
- {
73
- "ImportPath": "github.com/docker/spdystream",
74
- "Rev": "e9d52d77345028f700deb3799ec0b77d5df56dcd"
75
- },
72
{
73
"ImportPath": "github.com/facebookgo/stack",
74
"Rev": "4da6d991fc3c389efa512151354d643eb5fae4e2"
@@ -141,12 +137,16 @@
137
},
138
{
139
"ImportPath": "github.com/jbenet/go-peerstream",
144
- "Rev": "fc6b2a964210945ae2325e6678898d2222526f00"
140
+ "Rev": "c3ee65e805acc6a27036b9b892e0a95fc9769c3c"
141
},
142
{
143
"ImportPath": "github.com/jbenet/go-random",
144
"Rev": "2e83344e7dc7898f94501665af34edd4aa95a013"
145
},
146
+ {
147
+ "ImportPath": "github.com/jbenet/spdystream",
148
+ "Rev": "ebaae88f58b8042aae386bd3972d07cf9a892a4b"
149
+ },
150
{
151
"ImportPath": "github.com/kr/binarydist",
152
"Rev": "9955b0ab8708602d411341e55fffd7e0700f86bd"
Godeps/_workspace/src/github.com/jbenet/go-peerstream/conn.go
+1
-1
@@ -6,7 +6,7 @@ import (
6
"net/http"
7
"sync"
8
9
- ss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/docker/spdystream"
9
+ ss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/spdystream"
10
)
11
12
// ConnHandler is a function which receives a Conn. It allows
Godeps/_workspace/src/github.com/jbenet/go-peerstream/stream.go
+1
-1
@@ -1,7 +1,7 @@
1
package peerstream
2
3
import (
4
- ss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/docker/spdystream"
4
+ ss "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/spdystream"
5
)
6
7
// StreamHandler is a function which receives a Stream. It
Godeps/_workspace/src/github.com/jbenet/spdystream/CONTRIBUTING.md
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/LICENSE
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/MAINTAINERS
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/README.md
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/connection.go
renamed
+27
-21
@@ -1,9 +1,9 @@
1
package spdystream
2
3
import (
4
- "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/spdy"
4
"errors"
5
"fmt"
6
+ "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/spdy"
7
"io"
8
"net"
9
"net/http"
@@ -262,16 +262,17 @@ func (s *Connection) addStreamFrame(frame *spdy.SynStreamFrame) {
262
}
263
264
stream := &Stream{
265
- streamId: frame.StreamId,
266
- parent: parent,
267
- conn: s,
268
- startChan: make(chan error),
269
- headers: frame.Headers,
270
- finished: (frame.CFHeader.Flags & spdy.ControlFlagUnidirectional) != 0x00,
271
- replyCond: sync.NewCond(new(sync.Mutex)),
272
- dataChan: make(chan []byte),
273
- headerChan: make(chan http.Header),
274
- closeChan: make(chan bool),
265
+ streamId: frame.StreamId,
266
+ parent: parent,
267
+ conn: s,
268
+ startChan: make(chan error),
269
+ headers: frame.Headers,
270
+ finished: (frame.CFHeader.Flags & spdy.ControlFlagUnidirectional) != 0x00,
271
+ replyCond: sync.NewCond(new(sync.Mutex)),
272
+ dataChan: make(chan []byte),
273
+ headerChan: make(chan http.Header),
274
+ closeChan: make(chan bool),
275
+ shutdownChan: make(chan struct{}),
276
}
277
if frame.CFHeader.Flags&spdy.ControlFlagFin != 0x00 {
278
close(stream.dataChan)
@@ -415,8 +416,12 @@ func (s *Connection) handleDataFrame(frame *spdy.DataFrame) error {
416
break
417
default:
418
debugMessage("(%p) (%d) Data frame send chan", stream, stream.streamId)
418
- stream.dataChan <- frame.Data
419
- debugMessage("(%p) (%d) Data frame sent", stream, stream.streamId)
419
+ select {
420
+ case stream.dataChan <- frame.Data:
421
+ debugMessage("(%p) (%d) Data frame sent", stream, stream.streamId)
422
+ case <-stream.shutdownChan:
423
+ debugMessage("(%p) (%d) Data frame not sent (stream shut down)", stream, stream.streamId)
424
+ }
425
}
426
stream.dataLock.RUnlock()
427
}
@@ -495,14 +500,15 @@ func (s *Connection) CreateStream(headers http.Header, parent *Stream, fin bool)
500
}
501
502
stream := &Stream{
498
- streamId: streamId,
499
- parent: parent,
500
- conn: s,
501
- startChan: make(chan error),
502
- headers: headers,
503
- dataChan: make(chan []byte),
504
- headerChan: make(chan http.Header),
505
- closeChan: make(chan bool),
503
+ streamId: streamId,
504
+ parent: parent,
505
+ conn: s,
506
+ startChan: make(chan error),
507
+ headers: headers,
508
+ dataChan: make(chan []byte),
509
+ headerChan: make(chan http.Header),
510
+ closeChan: make(chan bool),
511
+ shutdownChan: make(chan struct{}),
512
}
513
514
debugMessage("(%p) (%p) Create stream", s, stream)
Godeps/_workspace/src/github.com/jbenet/spdystream/handlers.go
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/priority.go
renamed
+1
-1
@@ -1,8 +1,8 @@
1
package spdystream
2
3
import (
4
- "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/spdy"
4
"container/heap"
5
+ "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/spdy"
6
"sync"
7
)
8
Godeps/_workspace/src/github.com/jbenet/spdystream/priority_test.go
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/spdy_bench_test.go
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/spdy_test.go
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/stream.go
renamed
+11
-1
@@ -1,9 +1,9 @@
1
package spdystream
2
3
import (
4
- "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/spdy"
4
"errors"
5
"fmt"
6
+ "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/spdy"
7
"io"
8
"net"
9
"net/http"
@@ -33,6 +33,8 @@ type Stream struct {
33
replyCond *sync.Cond
34
replied bool
35
closeChan chan bool
36
+
37
+ shutdownChan chan struct{} // closed when Reset is called (no more R/W).
38
}
39
40
// WriteData writes data to stream, sending a dataframe per call
@@ -167,6 +169,14 @@ func (s *Stream) Close() error {
169
func (s *Stream) Reset() error {
170
s.conn.removeStream(s)
171
172
+ // only close it once.
173
+ select {
174
+ case <-s.shutdownChan:
175
+ // already was closed.
176
+ default:
177
+ close(s.shutdownChan)
178
+ }
179
+
180
s.finishLock.Lock()
181
if s.finished {
182
s.finishLock.Unlock()
Godeps/_workspace/src/github.com/jbenet/spdystream/stream_test.go
new
+106
@@ -0,0 +1,106 @@
1
+package spdystream
2
+
3
+import (
4
+ "net"
5
+ "net/http"
6
+ "sync"
7
+ "testing"
8
+)
9
+
10
+func TestStreamReset(t *testing.T) {
11
+ var wg sync.WaitGroup
12
+ listen := "localhost:7743"
13
+ server, serverErr := runServer(listen, &wg)
14
+ if serverErr != nil {
15
+ t.Fatalf("Error initializing server: %s", serverErr)
16
+ }
17
+
18
+ conn, dialErr := net.Dial("tcp", listen)
19
+ if dialErr != nil {
20
+ t.Fatalf("Error dialing server: %s", dialErr)
21
+ }
22
+
23
+ spdyConn, spdyErr := NewConnection(conn, false)
24
+ if spdyErr != nil {
25
+ t.Fatalf("Error creating spdy connection: %s", spdyErr)
26
+ }
27
+ go spdyConn.Serve(NoOpStreamHandler)
28
+
29
+ authenticated = true
30
+ stream, streamErr := spdyConn.CreateStream(http.Header{}, nil, false)
31
+ if streamErr != nil {
32
+ t.Fatalf("Error creating stream: %s", streamErr)
33
+ }
34
+
35
+ buf := []byte("dskjahfkdusahfkdsahfkdsafdkas")
36
+ for i := 0; i < 10; i++ {
37
+ if _, err := stream.Write(buf); err != nil {
38
+ t.Fatalf("Error writing to stream: %s", err)
39
+ }
40
+ }
41
+ for i := 0; i < 10; i++ {
42
+ if _, err := stream.Read(buf); err != nil {
43
+ t.Fatalf("Error reading from stream: %s", err)
44
+ }
45
+ }
46
+
47
+ // fmt.Printf("Resetting...\n")
48
+ if err := stream.Reset(); err != nil {
49
+ t.Fatalf("Error reseting stream: %s", err)
50
+ }
51
+
52
+ closeErr := server.Close()
53
+ if closeErr != nil {
54
+ t.Fatalf("Error shutting down server: %s", closeErr)
55
+ }
56
+ wg.Wait()
57
+}
58
+
59
+func TestStreamResetWithDataRemaining(t *testing.T) {
60
+ var wg sync.WaitGroup
61
+ listen := "localhost:7743"
62
+ server, serverErr := runServer(listen, &wg)
63
+ if serverErr != nil {
64
+ t.Fatalf("Error initializing server: %s", serverErr)
65
+ }
66
+
67
+ conn, dialErr := net.Dial("tcp", listen)
68
+ if dialErr != nil {
69
+ t.Fatalf("Error dialing server: %s", dialErr)
70
+ }
71
+
72
+ spdyConn, spdyErr := NewConnection(conn, false)
73
+ if spdyErr != nil {
74
+ t.Fatalf("Error creating spdy connection: %s", spdyErr)
75
+ }
76
+ go spdyConn.Serve(NoOpStreamHandler)
77
+
78
+ authenticated = true
79
+ stream, streamErr := spdyConn.CreateStream(http.Header{}, nil, false)
80
+ if streamErr != nil {
81
+ t.Fatalf("Error creating stream: %s", streamErr)
82
+ }
83
+
84
+ buf := []byte("dskjahfkdusahfkdsahfkdsafdkas")
85
+ for i := 0; i < 10; i++ {
86
+ if _, err := stream.Write(buf); err != nil {
87
+ t.Fatalf("Error writing to stream: %s", err)
88
+ }
89
+ }
90
+
91
+ // read a bit to make sure a goroutine gets to <-dataChan
92
+ if _, err := stream.Read(buf); err != nil {
93
+ t.Fatalf("Error reading from stream: %s", err)
94
+ }
95
+
96
+ // fmt.Printf("Resetting...\n")
97
+ if err := stream.Reset(); err != nil {
98
+ t.Fatalf("Error reseting stream: %s", err)
99
+ }
100
+
101
+ closeErr := server.Close()
102
+ if closeErr != nil {
103
+ t.Fatalf("Error shutting down server: %s", closeErr)
104
+ }
105
+ wg.Wait()
106
+}
Godeps/_workspace/src/github.com/jbenet/spdystream/utils.go
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/ws/connection.go
renamed
Godeps/_workspace/src/github.com/jbenet/spdystream/ws/ws_test.go
renamed
+1
-1
@@ -2,7 +2,7 @@ package ws
2
3
import (
4
"bytes"
5
- "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/docker/spdystream"
5
+ "github.com/docker/spdystream"
6
"github.com/gorilla/websocket"
7
"io"
8
"log"