net/diag: recursively decrement timeouts.
Not sure this works. we dont have tests for net diag. We should make some. cc @whyrusleeping.
Juan Batiz-Benet committed
Jan 17, 2015 at 16:44 UTC
f6278735ef6ad4e8f3fda47ac6c58f33c46d248e
5 files changed
+78
-55
Godeps/_workspace/src/github.com/jbenet/go-logging/examples/example.go
+1
-1
@@ -3,7 +3,7 @@ package main
3
import (
4
"os"
5
6
- "github.com/op/go-logging"
6
+ "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-logging"
7
)
8
9
var log = logging.MustGetLogger("example")
diagnostics/diag.go
+51
-53
@@ -7,6 +7,7 @@ import (
7
"bytes"
8
"encoding/json"
9
"errors"
10
+ "fmt"
11
"io"
12
"sync"
13
"time"
@@ -33,6 +34,7 @@ var log = util.Logger("diagnostics")
34
var ProtocolDiag protocol.ID = "/ipfs/diagnostics"
35
36
const ResponseTimeout = time.Second * 10
37
+const HopTimeoutDecrement = time.Second * 2
38
39
// Diagnostics is a net service that manages requesting and responding to diagnostic
40
// requests
@@ -149,39 +151,24 @@ func (d *Diagnostics) GetDiagnostic(timeout time.Duration) ([]*DiagInfo, error)
151
peers := d.getPeers()
152
log.Debugf("Sending diagnostic request to %d peers.", len(peers))
153
152
- var out []*DiagInfo
153
- di := d.getDiagInfo()
154
- out = append(out, di)
155
-
154
pmes := newMessage(diagID)
155
158
- respdata := make(chan []byte)
159
- sends := 0
160
- for p, _ := range peers {
161
- log.Debugf("Sending getDiagnostic to: %s", p)
162
- sends++
163
- go func(p peer.ID) {
164
- data, err := d.getDiagnosticFromPeer(ctx, p, pmes)
165
- if err != nil {
166
- log.Errorf("GetDiagnostic error: %v", err)
167
- respdata <- nil
168
- return
169
- }
170
- respdata <- data
171
- }(p)
156
+ pmes.SetTimeoutDuration(timeout - HopTimeoutDecrement) // decrease timeout per hop
157
+ dpeers, err := d.getDiagnosticFromPeers(ctx, d.getPeers(), pmes)
158
+ if err != nil {
159
+ return nil, fmt.Errorf("diagnostic from peers err: %s", err)
160
}
161
174
- for i := 0; i < sends; i++ {
175
- data := <-respdata
176
- if data == nil {
177
- continue
178
- }
179
- out = appendDiagnostics(data, out)
162
+ var out []*DiagInfo
163
+ di := d.getDiagInfo()
164
+ out = append(out, di)
165
+ for _, dpi := range dpeers {
166
+ out = appendDiagnostics(out, dpi)
167
}
168
return out, nil
169
}
170
184
-func appendDiagnostics(data []byte, cur []*DiagInfo) []*DiagInfo {
171
+func appendDiagnostics(cur []*DiagInfo, data []byte) []*DiagInfo {
172
buf := bytes.NewBuffer(data)
173
dec := json.NewDecoder(buf)
174
for {
@@ -198,6 +185,38 @@ func appendDiagnostics(data []byte, cur []*DiagInfo) []*DiagInfo {
185
return cur
186
}
187
188
+func (d *Diagnostics) getDiagnosticFromPeers(ctx context.Context, peers map[peer.ID]int, pmes *pb.Message) ([][]byte, error) {
189
+ timeout := pmes.GetTimeoutDuration()
190
+ if timeout < 1 {
191
+ return nil, fmt.Errorf("timeout too short: %s", timeout)
192
+ }
193
+ ctx, _ = context.WithTimeout(ctx, timeout)
194
+
195
+ respdata := make(chan []byte)
196
+ sendcount := 0
197
+ for p, _ := range peers {
198
+ log.Debugf("Sending diagnostic request to peer: %s", p)
199
+ sendcount++
200
+ go func(p peer.ID) {
201
+ out, err := d.getDiagnosticFromPeer(ctx, p, pmes)
202
+ if err != nil {
203
+ log.Errorf("getDiagnostic error: %v", err)
204
+ respdata <- nil
205
+ return
206
+ }
207
+ respdata <- out
208
+ }(p)
209
+ }
210
+
211
+ outall := make([][]byte, 0, len(peers))
212
+ for i := 0; i < sendcount; i++ {
213
+ out := <-respdata
214
+ outall = append(outall, out)
215
+ }
216
+
217
+ return outall, nil
218
+}
219
+
220
// TODO: this method no longer needed.
221
func (d *Diagnostics) getDiagnosticFromPeer(ctx context.Context, p peer.ID, mes *pb.Message) ([]byte, error) {
222
rpmes, err := d.sendRequest(ctx, p, mes)
@@ -259,38 +278,17 @@ func (d *Diagnostics) handleDiagnostic(p peer.ID, pmes *pb.Message) (*pb.Message
278
d.diagMap[pmes.GetDiagID()] = time.Now()
279
d.diagLock.Unlock()
280
262
- buf := new(bytes.Buffer)
281
di := d.getDiagInfo()
264
- buf.Write(di.Marshal())
265
-
266
- ctx, _ := context.WithTimeout(context.TODO(), ResponseTimeout)
267
-
268
- respdata := make(chan []byte)
269
- sendcount := 0
270
- for p, _ := range d.getPeers() {
271
- log.Debugf("Sending diagnostic request to peer: %s", p)
272
- sendcount++
273
- go func(p peer.ID) {
274
- out, err := d.getDiagnosticFromPeer(ctx, p, pmes)
275
- if err != nil {
276
- log.Errorf("getDiagnostic error: %v", err)
277
- respdata <- nil
278
- return
279
- }
280
- respdata <- out
281
- }(p)
282
- }
283
-
284
- for i := 0; i < sendcount; i++ {
285
- out := <-respdata
286
- _, err := buf.Write(out)
287
- if err != nil {
288
- log.Errorf("getDiagnostic write output error: %v", err)
289
- continue
282
+ resp.Data = di.Marshal()
283
+ dpeers, err := d.getDiagnosticFromPeers(context.TODO(), d.getPeers(), pmes)
284
+ if err != nil {
285
+ log.Errorf("diagnostic from peers err: %s", err)
286
+ } else {
287
+ for _, b := range dpeers {
288
+ resp.Data = append(resp.Data, b...) // concatenate them all.
289
}
290
}
291
293
- resp.Data = buf.Bytes()
292
return resp, nil
293
}
294
diagnostics/internal/pb/diagnostics.pb.go
+11
-1
@@ -14,15 +14,18 @@ It has these top-level messages:
14
package diagnostics_pb
15
16
import proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/proto"
17
+import json "encoding/json"
18
import math "math"
19
19
-// Reference imports to suppress errors if they are not otherwise used.
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 Message struct {
26
DiagID *string `protobuf:"bytes,1,req" json:"DiagID,omitempty"`
27
Data []byte `protobuf:"bytes,2,opt" json:"Data,omitempty"`
28
+ Timeout *int64 `protobuf:"varint,3,opt" json:"Timeout,omitempty"`
29
XXX_unrecognized []byte `json:"-"`
30
}
31
@@ -44,5 +47,12 @@ func (m *Message) GetData() []byte {
47
return nil
48
}
49
50
+func (m *Message) GetTimeout() int64 {
51
+ if m != nil && m.Timeout != nil {
52
+ return *m.Timeout
53
+ }
54
+ return 0
55
+}
56
+
57
func init() {
58
}
diagnostics/internal/pb/diagnostics.proto
+1
@@ -3,4 +3,5 @@ package diagnostics.pb;
3
message Message {
4
required string DiagID = 1;
5
optional bytes Data = 2;
6
+ optional int64 Timeout = 3; // in nanoseconds
7
}
diagnostics/internal/pb/timeout.go
new
+14
@@ -0,0 +1,14 @@
1
+package diagnostics_pb
2
+
3
+import (
4
+ "time"
5
+)
6
+
7
+func (m *Message) GetTimeoutDuration() time.Duration {
8
+ return time.Duration(m.GetTimeout())
9
+}
10
+
11
+func (m *Message) SetTimeoutDuration(t time.Duration) {
12
+ it := int64(t)
13
+ m.Timeout = &it
14
+}