-
Notifications
You must be signed in to change notification settings - Fork 0
/
dht.go
371 lines (305 loc) · 10.6 KB
/
dht.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
// Package dht implements a distributed hash table that satisfies the ipfs routing
// interface. This DHT is modeled after kademlia with Coral and S/Kademlia modifications.
package dht
import (
"bytes"
"crypto/rand"
"errors"
"fmt"
"sync"
"time"
ci "github.com/ipfs/go-ipfs/p2p/crypto"
host "github.com/ipfs/go-ipfs/p2p/host"
peer "github.com/ipfs/go-ipfs/p2p/peer"
protocol "github.com/ipfs/go-ipfs/p2p/protocol"
routing "github.com/ipfs/go-ipfs/routing"
pb "github.com/ipfs/go-ipfs/routing/dht/pb"
kb "github.com/ipfs/go-ipfs/routing/kbucket"
record "github.com/ipfs/go-ipfs/routing/record"
"github.com/ipfs/go-ipfs/thirdparty/eventlog"
u "github.com/ipfs/go-ipfs/util"
"github.com/ipfs/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
ctxgroup "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
)
var log = eventlog.Logger("dht")
var ProtocolDHT protocol.ID = "/ipfs/dht"
const doPinging = false
// NumBootstrapQueries defines the number of random dht queries to do to
// collect members of the routing table.
const NumBootstrapQueries = 5
// TODO. SEE https://github.com/jbenet/node-ipfs/blob/master/submodules/ipfs-dht/index.js
// IpfsDHT is an implementation of Kademlia with Coral and S/Kademlia modifications.
// It is used to implement the base IpfsRouting module.
type IpfsDHT struct {
host host.Host // the network services we need
self peer.ID // Local peer (yourself)
peerstore peer.Peerstore // Peer Registry
datastore ds.ThreadSafeDatastore // Local data
routingTable *kb.RoutingTable // Array of routing tables for differently distanced nodes
providers *ProviderManager
birth time.Time // When this peer started up
diaglock sync.Mutex // lock to make diagnostics work better
Validator record.Validator // record validator funcs
ctxgroup.ContextGroup
}
// NewDHT creates a new DHT object with the given peer as the 'local' host
func NewDHT(ctx context.Context, h host.Host, dstore ds.ThreadSafeDatastore) *IpfsDHT {
dht := new(IpfsDHT)
dht.datastore = dstore
dht.self = h.ID()
dht.peerstore = h.Peerstore()
dht.host = h
// register for network notifs.
dht.host.Network().Notify((*netNotifiee)(dht))
dht.ContextGroup = ctxgroup.WithContextAndTeardown(ctx, func() error {
// remove ourselves from network notifs.
dht.host.Network().StopNotify((*netNotifiee)(dht))
return nil
})
h.SetStreamHandler(ProtocolDHT, dht.handleNewStream)
dht.providers = NewProviderManager(dht.Context(), dht.self)
dht.AddChild(dht.providers)
dht.routingTable = kb.NewRoutingTable(20, kb.ConvertPeerID(dht.self), time.Minute, dht.peerstore)
dht.birth = time.Now()
dht.Validator = make(record.Validator)
dht.Validator["pk"] = record.PublicKeyValidator
if doPinging {
dht.Children().Add(1)
go dht.PingRoutine(time.Second * 10)
}
return dht
}
// LocalPeer returns the peer.Peer of the dht.
func (dht *IpfsDHT) LocalPeer() peer.ID {
return dht.self
}
// log returns the dht's logger
func (dht *IpfsDHT) log() eventlog.EventLogger {
return log // TODO rm
}
// Connect to a new peer at the given address, ping and add to the routing table
func (dht *IpfsDHT) Connect(ctx context.Context, npeer peer.ID) error {
// TODO: change interface to accept a PeerInfo as well.
if err := dht.host.Connect(ctx, peer.PeerInfo{ID: npeer}); err != nil {
return err
}
// Ping new peer to register in their routing table
// NOTE: this should be done better...
if _, err := dht.Ping(ctx, npeer); err != nil {
return fmt.Errorf("failed to ping newly connected peer: %s", err)
}
log.Event(ctx, "connect", dht.self, npeer)
dht.Update(ctx, npeer)
return nil
}
// putValueToPeer stores the given key/value pair at the peer 'p'
func (dht *IpfsDHT) putValueToPeer(ctx context.Context, p peer.ID,
key u.Key, rec *pb.Record) error {
pmes := pb.NewMessage(pb.Message_PUT_VALUE, string(key), 0)
pmes.Record = rec
rpmes, err := dht.sendRequest(ctx, p, pmes)
if err != nil {
return err
}
if !bytes.Equal(rpmes.GetRecord().Value, pmes.GetRecord().Value) {
return errors.New("value not put correctly")
}
return nil
}
// putProvider sends a message to peer 'p' saying that the local node
// can provide the value of 'key'
func (dht *IpfsDHT) putProvider(ctx context.Context, p peer.ID, key string) error {
// add self as the provider
pi := peer.PeerInfo{
ID: dht.self,
Addrs: dht.host.Addrs(),
}
// // only share WAN-friendly addresses ??
// pi.Addrs = addrutil.WANShareableAddrs(pi.Addrs)
if len(pi.Addrs) < 1 {
// log.Infof("%s putProvider: %s for %s error: no wan-friendly addresses", dht.self, p, u.Key(key), pi.Addrs)
return fmt.Errorf("no known addresses for self. cannot put provider.")
}
pmes := pb.NewMessage(pb.Message_ADD_PROVIDER, string(key), 0)
pmes.ProviderPeers = pb.RawPeerInfosToPBPeers([]peer.PeerInfo{pi})
err := dht.sendMessage(ctx, p, pmes)
if err != nil {
return err
}
log.Debugf("%s putProvider: %s for %s (%s)", dht.self, p, u.Key(key), pi.Addrs)
return nil
}
// getValueOrPeers queries a particular peer p for the value for
// key. It returns either the value or a list of closer peers.
// NOTE: it will update the dht's peerstore with any new addresses
// it finds for the given peer.
func (dht *IpfsDHT) getValueOrPeers(ctx context.Context, p peer.ID,
key u.Key) ([]byte, []peer.PeerInfo, error) {
pmes, err := dht.getValueSingle(ctx, p, key)
if err != nil {
return nil, nil, err
}
if record := pmes.GetRecord(); record != nil {
// Success! We were given the value
log.Debug("getValueOrPeers: got value")
// make sure record is valid.
err = dht.verifyRecordOnline(ctx, record)
if err != nil {
log.Info("Received invalid record! (discarded)")
return nil, nil, err
}
return record.GetValue(), nil, nil
}
// Perhaps we were given closer peers
peers := pb.PBPeersToPeerInfos(pmes.GetCloserPeers())
if len(peers) > 0 {
log.Debug("getValueOrPeers: peers")
return nil, peers, nil
}
log.Warning("getValueOrPeers: routing.ErrNotFound")
return nil, nil, routing.ErrNotFound
}
// getValueSingle simply performs the get value RPC with the given parameters
func (dht *IpfsDHT) getValueSingle(ctx context.Context, p peer.ID,
key u.Key) (*pb.Message, error) {
defer log.EventBegin(ctx, "getValueSingle", p, &key).Done()
pmes := pb.NewMessage(pb.Message_GET_VALUE, string(key), 0)
return dht.sendRequest(ctx, p, pmes)
}
// getLocal attempts to retrieve the value from the datastore
func (dht *IpfsDHT) getLocal(key u.Key) ([]byte, error) {
log.Debug("getLocal %s", key)
v, err := dht.datastore.Get(key.DsKey())
if err != nil {
return nil, err
}
log.Debug("found in db")
byt, ok := v.([]byte)
if !ok {
return nil, errors.New("value stored in datastore not []byte")
}
rec := new(pb.Record)
err = proto.Unmarshal(byt, rec)
if err != nil {
return nil, err
}
// TODO: 'if paranoid'
if u.Debug {
err = dht.verifyRecordLocally(rec)
if err != nil {
log.Debugf("local record verify failed: %s (discarded)", err)
return nil, err
}
}
return rec.GetValue(), nil
}
// getOwnPrivateKey attempts to load the local peers private
// key from the peerstore.
func (dht *IpfsDHT) getOwnPrivateKey() (ci.PrivKey, error) {
sk := dht.peerstore.PrivKey(dht.self)
if sk == nil {
log.Warningf("%s dht cannot get own private key!", dht.self)
return nil, fmt.Errorf("cannot get private key to sign record!")
}
return sk, nil
}
// putLocal stores the key value pair in the datastore
func (dht *IpfsDHT) putLocal(key u.Key, rec *pb.Record) error {
data, err := proto.Marshal(rec)
if err != nil {
return err
}
return dht.datastore.Put(key.DsKey(), data)
}
// Update signals the routingTable to Update its last-seen status
// on the given peer.
func (dht *IpfsDHT) Update(ctx context.Context, p peer.ID) {
log.Event(ctx, "updatePeer", p)
dht.routingTable.Update(p)
}
// FindLocal looks for a peer with a given ID connected to this dht and returns the peer and the table it was found in.
func (dht *IpfsDHT) FindLocal(id peer.ID) peer.PeerInfo {
p := dht.routingTable.Find(id)
if p != "" {
return dht.peerstore.PeerInfo(p)
}
return peer.PeerInfo{}
}
// findPeerSingle asks peer 'p' if they know where the peer with id 'id' is
func (dht *IpfsDHT) findPeerSingle(ctx context.Context, p peer.ID, id peer.ID) (*pb.Message, error) {
defer log.EventBegin(ctx, "findPeerSingle", p, id).Done()
pmes := pb.NewMessage(pb.Message_FIND_NODE, string(id), 0)
return dht.sendRequest(ctx, p, pmes)
}
func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key u.Key) (*pb.Message, error) {
defer log.EventBegin(ctx, "findProvidersSingle", p, &key).Done()
pmes := pb.NewMessage(pb.Message_GET_PROVIDERS, string(key), 0)
return dht.sendRequest(ctx, p, pmes)
}
// nearestPeersToQuery returns the routing tables closest peers.
func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID {
key := u.Key(pmes.GetKey())
closer := dht.routingTable.NearestPeers(kb.ConvertKey(key), count)
return closer
}
// betterPeerToQuery returns nearestPeersToQuery, but iff closer than self.
func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, p peer.ID, count int) []peer.ID {
closer := dht.nearestPeersToQuery(pmes, count)
// no node? nil
if closer == nil {
return nil
}
// == to self? thats bad
for _, p := range closer {
if p == dht.self {
log.Debug("Attempted to return self! this shouldnt happen...")
return nil
}
}
var filtered []peer.ID
for _, clp := range closer {
// Dont send a peer back themselves
if p == clp {
continue
}
// must all be closer than self
key := u.Key(pmes.GetKey())
if !kb.Closer(dht.self, clp, key) {
filtered = append(filtered, clp)
}
}
// ok seems like closer nodes
return filtered
}
func (dht *IpfsDHT) ensureConnectedToPeer(ctx context.Context, p peer.ID) error {
if p == dht.self {
return errors.New("attempting to ensure connection to self")
}
// dial connection
return dht.host.Connect(ctx, peer.PeerInfo{ID: p})
}
// PingRoutine periodically pings nearest neighbors.
func (dht *IpfsDHT) PingRoutine(t time.Duration) {
defer dht.Children().Done()
tick := time.Tick(t)
for {
select {
case <-tick:
id := make([]byte, 16)
rand.Read(id)
peers := dht.routingTable.NearestPeers(kb.ConvertKey(u.Key(id)), 5)
for _, p := range peers {
ctx, cancel := context.WithTimeout(dht.Context(), time.Second*5)
_, err := dht.Ping(ctx, p)
if err != nil {
log.Debugf("Ping error: %s", err)
}
cancel()
}
case <-dht.Closing():
return
}
}
}