p2p/net/conn/listener: ignore certain errors
This should handle early breakages, where a failing connection would take out the listener entirely. There are probably other errors we should be handling here, like secure connection failures.
Juan Batiz-Benet committed
Jan 11, 2015 at 11:59 UTC
4525269cd830ac2ba338d9e1281ea80d66e189cf
2 files changed
+137
-17
p2p/net/conn/dial_test.go
+91
@@ -1,8 +1,10 @@
1
package conn
2
3
import (
4
+ "fmt"
5
"io"
6
"net"
7
+ "strings"
8
"testing"
9
"time"
10
@@ -103,6 +105,9 @@ func testDialer(t *testing.T, secure bool) {
105
if !secure {
106
key1 = nil
107
key2 = nil
108
+ t.Log("testing insecurely")
109
+ } else {
110
+ t.Log("testing securely")
111
}
112
113
ctx, cancel := context.WithCancel(context.Background())
@@ -164,3 +169,89 @@ func TestDialerSecure(t *testing.T) {
169
// t.Skip("Skipping in favor of another test")
170
testDialer(t, true)
171
}
172
+
173
+func testDialerCloseEarly(t *testing.T, secure bool) {
174
+ // t.Skip("Skipping in favor of another test")
175
+
176
+ p1 := tu.RandPeerNetParamsOrFatal(t)
177
+ p2 := tu.RandPeerNetParamsOrFatal(t)
178
+
179
+ key1 := p1.PrivKey
180
+ if !secure {
181
+ key1 = nil
182
+ t.Log("testing insecurely")
183
+ } else {
184
+ t.Log("testing securely")
185
+ }
186
+
187
+ ctx, cancel := context.WithCancel(context.Background())
188
+ l1, err := Listen(ctx, p1.Addr, p1.ID, key1)
189
+ if err != nil {
190
+ t.Fatal(err)
191
+ }
192
+ p1.Addr = l1.Multiaddr() // Addr has been determined by kernel.
193
+
194
+ // lol nesting
195
+ d2 := &Dialer{
196
+ LocalPeer: p2.ID,
197
+ // PrivateKey: key2, -- dont give it key. we'll just close the conn.
198
+ }
199
+
200
+ errs := make(chan error, 100)
201
+ done := make(chan struct{}, 1)
202
+ gotclosed := make(chan struct{}, 1)
203
+ go func() {
204
+ defer func() { done <- struct{}{} }()
205
+
206
+ _, err := l1.Accept()
207
+ if err != nil {
208
+ if strings.Contains(err.Error(), "closed") {
209
+ gotclosed <- struct{}{}
210
+ return
211
+ }
212
+ errs <- err
213
+ }
214
+ errs <- fmt.Errorf("got conn")
215
+ }()
216
+
217
+ c, err := d2.Dial(ctx, p1.Addr, p1.ID)
218
+ if err != nil {
219
+ errs <- err
220
+ }
221
+ c.Close() // close it early.
222
+
223
+ readerrs := func() {
224
+ for {
225
+ select {
226
+ case e := <-errs:
227
+ t.Error(e)
228
+ default:
229
+ return
230
+ }
231
+ }
232
+ }
233
+ readerrs()
234
+
235
+ l1.Close()
236
+ <-done
237
+ cancel()
238
+ readerrs()
239
+ close(errs)
240
+
241
+ select {
242
+ case <-gotclosed:
243
+ default:
244
+ t.Error("did not get closed")
245
+ }
246
+}
247
+
248
+// we dont do a handshake with singleConn, so cant "close early."
249
+// func TestDialerCloseEarlyInsecure(t *testing.T) {
250
+// // t.Skip("Skipping in favor of another test")
251
+// testDialerCloseEarly(t, false)
252
+// }
253
+
254
+func TestDialerCloseEarlySecure(t *testing.T) {
255
+ // t.Skip("Skipping in favor of another test")
256
+ testDialerCloseEarly(t, true)
257
+}
p2p/net/conn/listen.go
+46
-17
@@ -2,13 +2,14 @@ package conn
2
3
import (
4
"fmt"
5
+ "io"
6
"net"
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
10
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
11
manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
11
-
12
+ tec "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-temp-err-catcher"
13
ic "github.com/jbenet/go-ipfs/p2p/crypto"
14
peer "github.com/jbenet/go-ipfs/p2p/peer"
15
)
@@ -46,25 +47,53 @@ func (l *listener) Accept() (net.Conn, error) {
47
// Contexts and io don't mix.
48
ctx := context.Background()
49
49
- maconn, err := l.Listener.Accept()
50
- if err != nil {
51
- return nil, err
50
+ var catcher tec.TempErrCatcher
51
+
52
+ catcher.IsTemp = func(e error) bool {
53
+ // ignore connection breakages up to this point. but log them
54
+ if e == io.EOF {
55
+ log.Debugf("listener ignoring conn with EOF: %s", e)
56
+ return true
57
+ }
58
+
59
+ te, ok := e.(tec.Temporary)
60
+ if ok {
61
+ log.Debugf("listener ignoring conn with temporary err: %s", e)
62
+ return te.Temporary()
63
+ }
64
+ return false
65
}
66
54
- c, err := newSingleConn(ctx, l.local, "", maconn)
55
- if err != nil {
56
- return nil, fmt.Errorf("Error accepting connection: %v", err)
57
- }
58
-
59
- if l.privk == nil {
60
- log.Warning("listener %s listening INSECURELY!", l)
61
- return c, nil
62
- }
63
- sc, err := newSecureConn(ctx, l.privk, c)
64
- if err != nil {
65
- return nil, fmt.Errorf("Error securing connection: %v", err)
67
+ for {
68
+ maconn, err := l.Listener.Accept()
69
+ if err != nil {
70
+ if catcher.IsTemporary(err) {
71
+ continue
72
+ }
73
+ return nil, err
74
+ }
75
+
76
+ c, err := newSingleConn(ctx, l.local, "", maconn)
77
+ if err != nil {
78
+ if catcher.IsTemporary(err) {
79
+ continue
80
+ }
81
+ return nil, err
82
+ }
83
+
84
+ if l.privk == nil {
85
+ log.Warning("listener %s listening INSECURELY!", l)
86
+ return c, nil
87
+ }
88
+ sc, err := newSecureConn(ctx, l.privk, c)
89
+ if err != nil {
90
+ if catcher.IsTemporary(err) {
91
+ continue
92
+ }
93
+ return nil, err
94
+ }
95
+ return sc, nil
96
}
67
- return sc, nil
97
}
98
99
func (l *listener) Addr() net.Addr {