some dht cleanup, and make DHTs take a master context
Jeromy committed
Oct 15, 2014 at 12:30 UTC
60cd0f1cf00fa1526c079e21b77ddcc57905fede
7 files changed
+45
-106
core/core.go
+1
-1
@@ -141,7 +141,7 @@ func NewIpfsNode(cfg *config.Config, online bool) (*IpfsNode, error) {
141
diagnostics = diag.NewDiagnostics(local, net, diagService)
142
diagService.SetHandler(diagnostics)
143
144
- route = dht.NewDHT(local, peerstore, net, dhtService, d)
144
+ route = dht.NewDHT(ctx, local, peerstore, net, dhtService, d)
145
// TODO(brian): perform this inside NewDHT factory method
146
dhtService.SetHandler(route) // wire the handler to the service.
147
routing/dht/dht.go
+28
-3
@@ -53,16 +53,19 @@ type IpfsDHT struct {
53
54
//lock to make diagnostics work better
55
diaglock sync.Mutex
56
+
57
+ ctx context.Context
58
}
59
60
// NewDHT creates a new DHT object with the given peer as the 'local' host
59
-func NewDHT(p *peer.Peer, ps peer.Peerstore, net inet.Network, sender inet.Sender, dstore ds.Datastore) *IpfsDHT {
61
+func NewDHT(ctx context.Context, p *peer.Peer, ps peer.Peerstore, net inet.Network, sender inet.Sender, dstore ds.Datastore) *IpfsDHT {
62
dht := new(IpfsDHT)
63
dht.network = net
64
dht.sender = sender
65
dht.datastore = dstore
66
dht.self = p
67
dht.peerstore = ps
68
+ dht.ctx = ctx
69
70
dht.providers = NewProviderManager(p.ID)
71
@@ -71,6 +74,8 @@ func NewDHT(p *peer.Peer, ps peer.Peerstore, net inet.Network, sender inet.Sende
74
dht.routingTables[1] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Millisecond*1000)
75
dht.routingTables[2] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Hour)
76
dht.birth = time.Now()
77
+
78
+ go dht.PingRoutine(time.Second * 10)
79
return dht
80
}
81
@@ -137,7 +142,6 @@ func (dht *IpfsDHT) HandleMessage(ctx context.Context, mes msg.NetMessage) msg.N
142
// get handler for this msg type.
143
handler := dht.handlerForMsgType(pmes.GetType())
144
if handler == nil {
140
- // TODO handle/log err
145
log.Error("got back nil handler from handlerForMsgType")
146
return nil
147
}
@@ -350,7 +354,7 @@ func (dht *IpfsDHT) getLocal(key u.Key) ([]byte, error) {
354
355
byt, ok := v.([]byte)
356
if !ok {
353
- return byt, errors.New("value stored in datastore not []byte")
357
+ return nil, errors.New("value stored in datastore not []byte")
358
}
359
return byt, nil
360
}
@@ -533,6 +537,27 @@ func (dht *IpfsDHT) loadProvidableKeys() error {
537
return nil
538
}
539
540
+func (dht *IpfsDHT) PingRoutine(t time.Duration) {
541
+ tick := time.Tick(t)
542
+ for {
543
+ select {
544
+ case <-tick:
545
+ id := make([]byte, 16)
546
+ rand.Read(id)
547
+ peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(u.Key(id)), 5)
548
+ for _, p := range peers {
549
+ ctx, _ := context.WithTimeout(dht.ctx, time.Second*5)
550
+ err := dht.Ping(ctx, p)
551
+ if err != nil {
552
+ log.Error("Ping error: %s", err)
553
+ }
554
+ }
555
+ case <-dht.ctx.Done():
556
+ return
557
+ }
558
+ }
559
+}
560
+
561
// Bootstrap builds up list of peers by requesting random peer IDs
562
func (dht *IpfsDHT) Bootstrap(ctx context.Context) {
563
id := make([]byte, 16)
routing/dht/dht_logger.go
+6
-1
@@ -2,6 +2,7 @@ package dht
2
3
import (
4
"encoding/json"
5
+ "fmt"
6
"time"
7
)
8
@@ -29,12 +30,16 @@ func (l *logDhtRPC) EndLog() {
30
func (l *logDhtRPC) Print() {
31
b, err := json.Marshal(l)
32
if err != nil {
32
- log.Debug(err.Error())
33
+ log.Debug("Error marshaling logDhtRPC object: %s", err)
34
} else {
35
log.Debug(string(b))
36
}
37
}
38
39
+func (l *logDhtRPC) String() string {
40
+ return fmt.Sprintf("DHT RPC: %s took %s, success = %s", l.Type, l.Duration, l.Success)
41
+}
42
+
43
func (l *logDhtRPC) EndAndPrint() {
44
l.EndLog()
45
l.Print()
routing/dht/handlers.go
-54
@@ -5,9 +5,7 @@ import (
5
"fmt"
6
"time"
7
8
- msg "github.com/jbenet/go-ipfs/net/message"
8
peer "github.com/jbenet/go-ipfs/peer"
10
- kb "github.com/jbenet/go-ipfs/routing/kbucket"
9
u "github.com/jbenet/go-ipfs/util"
10
11
ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/datastore.go"
@@ -32,8 +30,6 @@ func (dht *IpfsDHT) handlerForMsgType(t Message_MessageType) dhtHandler {
30
return dht.handleGetProviders
31
case Message_PING:
32
return dht.handlePing
35
- case Message_DIAGNOSTIC:
36
- return dht.handleDiagnostic
33
default:
34
return nil
35
}
@@ -211,53 +207,3 @@ func (dht *IpfsDHT) handleAddProvider(p *peer.Peer, pmes *Message) (*Message, er
207
func (dht *IpfsDHT) Halt() {
208
dht.providers.Halt()
209
}
214
-
215
-// NOTE: not yet finished, low priority
216
-func (dht *IpfsDHT) handleDiagnostic(p *peer.Peer, pmes *Message) (*Message, error) {
217
- seq := dht.routingTables[0].NearestPeers(kb.ConvertPeerID(dht.self.ID), 10)
218
-
219
- for _, ps := range seq {
220
- _, err := msg.FromObject(ps, pmes)
221
- if err != nil {
222
- log.Error("handleDiagnostics error creating message: %v\n", err)
223
- continue
224
- }
225
- // dht.sender.SendRequest(context.TODO(), mes)
226
- }
227
- return nil, errors.New("not yet ported back")
228
-
229
- // buf := new(bytes.Buffer)
230
- // di := dht.getDiagInfo()
231
- // buf.Write(di.Marshal())
232
- //
233
- // // NOTE: this shouldnt be a hardcoded value
234
- // after := time.After(time.Second * 20)
235
- // count := len(seq)
236
- // for count > 0 {
237
- // select {
238
- // case <-after:
239
- // //Timeout, return what we have
240
- // goto out
241
- // case reqResp := <-listenChan:
242
- // pmesOut := new(Message)
243
- // err := proto.Unmarshal(reqResp.Data, pmesOut)
244
- // if err != nil {
245
- // // It broke? eh, whatever, keep going
246
- // continue
247
- // }
248
- // buf.Write(reqResp.Data)
249
- // count--
250
- // }
251
- // }
252
- //
253
- // out:
254
- // resp := Message{
255
- // Type: Message_DIAGNOSTIC,
256
- // ID: pmes.GetId(),
257
- // Value: buf.Bytes(),
258
- // Response: true,
259
- // }
260
- //
261
- // mes := swarm.NewMessage(p, resp.ToProtobuf())
262
- // dht.netChan.Outgoing <- mes
263
-}
routing/dht/messages.pb.go
+6
-11
@@ -1,4 +1,4 @@
1
-// Code generated by protoc-gen-gogo.
1
+// Code generated by protoc-gen-go.
2
// source: messages.proto
3
// DO NOT EDIT!
4
@@ -13,13 +13,11 @@ It has these top-level messages:
13
*/
14
package dht
15
16
-import proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/proto"
17
-import json "encoding/json"
16
+import proto "code.google.com/p/goprotobuf/proto"
17
import math "math"
18
20
-// Reference proto, json, and math imports to suppress error if they are not otherwise used.
19
+// Reference imports to suppress errors if they are not otherwise used.
20
var _ = proto.Marshal
22
-var _ = &json.SyntaxError{}
21
var _ = math.Inf
22
23
type Message_MessageType int32
@@ -31,7 +29,6 @@ const (
29
Message_GET_PROVIDERS Message_MessageType = 3
30
Message_FIND_NODE Message_MessageType = 4
31
Message_PING Message_MessageType = 5
34
- Message_DIAGNOSTIC Message_MessageType = 6
32
)
33
34
var Message_MessageType_name = map[int32]string{
@@ -41,7 +38,6 @@ var Message_MessageType_name = map[int32]string{
38
3: "GET_PROVIDERS",
39
4: "FIND_NODE",
40
5: "PING",
44
- 6: "DIAGNOSTIC",
41
}
42
var Message_MessageType_value = map[string]int32{
43
"PUT_VALUE": 0,
@@ -50,7 +46,6 @@ var Message_MessageType_value = map[string]int32{
46
"GET_PROVIDERS": 3,
47
"FIND_NODE": 4,
48
"PING": 5,
53
- "DIAGNOSTIC": 6,
49
}
50
51
func (x Message_MessageType) Enum() *Message_MessageType {
@@ -72,7 +67,7 @@ func (x *Message_MessageType) UnmarshalJSON(data []byte) error {
67
68
type Message struct {
69
// defines what type of message it is.
75
- Type *Message_MessageType `protobuf:"varint,1,req,name=type,enum=dht.Message_MessageType" json:"type,omitempty"`
70
+ Type *Message_MessageType `protobuf:"varint,1,opt,name=type,enum=dht.Message_MessageType" json:"type,omitempty"`
71
// defines what coral cluster level this query/response belongs to.
72
ClusterLevelRaw *int32 `protobuf:"varint,10,opt,name=clusterLevelRaw" json:"clusterLevelRaw,omitempty"`
73
// Used to specify the key associated with this message.
@@ -137,8 +132,8 @@ func (m *Message) GetProviderPeers() []*Message_Peer {
132
}
133
134
type Message_Peer struct {
140
- Id *string `protobuf:"bytes,1,req,name=id" json:"id,omitempty"`
141
- Addr *string `protobuf:"bytes,2,req,name=addr" json:"addr,omitempty"`
135
+ Id *string `protobuf:"bytes,1,opt,name=id" json:"id,omitempty"`
136
+ Addr *string `protobuf:"bytes,2,opt,name=addr" json:"addr,omitempty"`
137
XXX_unrecognized []byte `json:"-"`
138
}
139
routing/dht/messages.proto
+3
-4
@@ -10,16 +10,15 @@ message Message {
10
GET_PROVIDERS = 3;
11
FIND_NODE = 4;
12
PING = 5;
13
- DIAGNOSTIC = 6;
13
}
14
15
message Peer {
17
- required string id = 1;
18
- required string addr = 2;
16
+ optional string id = 1;
17
+ optional string addr = 2;
18
}
19
20
// defines what type of message it is.
22
- required MessageType type = 1;
21
+ optional MessageType type = 1;
22
23
// defines what coral cluster level this query/response belongs to.
24
optional int32 clusterLevelRaw = 10;
routing/dht/routing.go
+1
-32
@@ -1,8 +1,6 @@
1
package dht
2
3
import (
4
- "bytes"
5
- "encoding/json"
4
"sync"
5
6
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
@@ -62,6 +60,7 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key u.Key) ([]byte, error) {
60
routeLevel := 0
61
closest := dht.routingTables[routeLevel].NearestPeers(kb.ConvertKey(key), PoolSize)
62
if closest == nil || len(closest) == 0 {
63
+ log.Warning("Got no peers back from routing table!")
64
return nil, nil
65
}
66
@@ -282,33 +281,3 @@ func (dht *IpfsDHT) Ping(ctx context.Context, p *peer.Peer) error {
281
log.Info("ping %s end (err = %s)", p, err)
282
return err
283
}
285
-
286
-func (dht *IpfsDHT) getDiagnostic(ctx context.Context) ([]*diagInfo, error) {
287
-
288
- log.Info("Begin Diagnostic")
289
- peers := dht.routingTables[0].NearestPeers(kb.ConvertPeerID(dht.self.ID), 10)
290
- var out []*diagInfo
291
-
292
- query := newQuery(dht.self.Key(), func(ctx context.Context, p *peer.Peer) (*dhtQueryResult, error) {
293
- pmes := newMessage(Message_DIAGNOSTIC, "", 0)
294
- rpmes, err := dht.sendRequest(ctx, p, pmes)
295
- if err != nil {
296
- return nil, err
297
- }
298
-
299
- dec := json.NewDecoder(bytes.NewBuffer(rpmes.GetValue()))
300
- for {
301
- di := new(diagInfo)
302
- err := dec.Decode(di)
303
- if err != nil {
304
- break
305
- }
306
-
307
- out = append(out, di)
308
- }
309
- return &dhtQueryResult{success: true}, nil
310
- })
311
-
312
- _, err := query.Run(ctx, peers)
313
- return out, err
314
-}