dht.go 12.6 KB
Newer Older
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
1 2
package dht

3
import (
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
4
	"bytes"
5
	"crypto/rand"
6
	"errors"
7
	"fmt"
8 9
	"sync"
	"time"
10

11
	inet "github.com/jbenet/go-ipfs/net"
12
	msg "github.com/jbenet/go-ipfs/net/message"
13
	peer "github.com/jbenet/go-ipfs/peer"
Jeromy's avatar
Jeromy committed
14
	kb "github.com/jbenet/go-ipfs/routing/kbucket"
15
	u "github.com/jbenet/go-ipfs/util"
16

17
	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
18
	ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/datastore.go"
19
	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
Jeromy's avatar
Jeromy committed
20

21
	"github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
22 23
)

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
24 25 26 27 28
// 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 {
29 30
	// Array of routing tables for differently distanced nodes
	// NOTE: (currently, only a single table is used)
31
	routingTables []*kb.RoutingTable
32

33
	// the network interface. service
34
	network inet.Network
35
	sender  inet.Sender
36

Jeromy's avatar
Jeromy committed
37 38 39
	// Local peer (yourself)
	self *peer.Peer

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
40 41 42
	// Other peers
	peerstore peer.Peerstore

Jeromy's avatar
Jeromy committed
43 44
	// Local data
	datastore ds.Datastore
45
	dslock    sync.Mutex
46

47
	providers *ProviderManager
48

49 50
	// When this peer started up
	birth time.Time
51 52 53

	//lock to make diagnostics work better
	diaglock sync.Mutex
54 55
}

Jeromy's avatar
Jeromy committed
56
// NewDHT creates a new DHT object with the given peer as the 'local' host
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
57
func NewDHT(p *peer.Peer, ps peer.Peerstore, net inet.Network, sender inet.Sender, dstore ds.Datastore) *IpfsDHT {
58
	dht := new(IpfsDHT)
Jeromy's avatar
Jeromy committed
59
	dht.network = net
60
	dht.sender = sender
Jeromy's avatar
Jeromy committed
61
	dht.datastore = dstore
62
	dht.self = p
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
63
	dht.peerstore = ps
64

65
	dht.providers = NewProviderManager(p.ID)
66

67 68 69 70
	dht.routingTables = make([]*kb.RoutingTable, 3)
	dht.routingTables[0] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Millisecond*30)
	dht.routingTables[1] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Millisecond*100)
	dht.routingTables[2] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Hour)
71
	dht.birth = time.Now()
Jeromy's avatar
Jeromy committed
72
	return dht
73 74
}

75
// Connect to a new peer at the given address, ping and add to the routing table
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
76 77
func (dht *IpfsDHT) Connect(npeer *peer.Peer) (*peer.Peer, error) {
	u.DOut("Connect to new peer: %s\n", npeer.ID.Pretty())
78 79 80 81 82 83 84 85 86 87 88

	// TODO(jbenet,whyrusleeping)
	//
	// Connect should take in a Peer (with ID). In a sense, we shouldn't be
	// allowing connections to random multiaddrs without knowing who we're
	// speaking to (i.e. peer.ID). In terms of moving around simple addresses
	// -- instead of an (ID, Addr) pair -- we can use:
	//
	//   /ip4/10.20.30.40/tcp/1234/ipfs/Qxhxxchxzcncxnzcnxzcxzm
	//
	err := dht.network.DialPeer(npeer)
89
	if err != nil {
90
		return nil, err
91 92
	}

Jeromy's avatar
Jeromy committed
93 94
	// Ping new peer to register in their routing table
	// NOTE: this should be done better...
95
	err = dht.Ping(npeer, time.Second*2)
Jeromy's avatar
Jeromy committed
96
	if err != nil {
97
		return nil, fmt.Errorf("failed to ping newly connected peer: %s\n", err)
Jeromy's avatar
Jeromy committed
98 99
	}

100 101
	dht.Update(npeer)

102
	return npeer, nil
Jeromy's avatar
Jeromy committed
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
// HandleMessage implements the inet.Handler interface.
func (dht *IpfsDHT) HandleMessage(ctx context.Context, mes msg.NetMessage) (msg.NetMessage, error) {

	mData := mes.Data()
	if mData == nil {
		return nil, errors.New("message did not include Data")
	}

	mPeer := mes.Peer()
	if mPeer == nil {
		return nil, errors.New("message did not include a Peer")
	}

	// deserialize msg
	pmes := new(Message)
	err := proto.Unmarshal(mData, pmes)
	if err != nil {
		return nil, fmt.Errorf("Failed to decode protobuf message: %v\n", err)
	}

	// update the peer (on valid msgs only)
	dht.Update(mPeer)

	// Print out diagnostic
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
129
	u.DOut("[peer: %s] Got message type: '%s' [from = %s]\n",
130 131 132 133 134 135 136 137 138 139 140 141 142 143 144
		dht.self.ID.Pretty(),
		Message_MessageType_name[int32(pmes.GetType())], mPeer.ID.Pretty())

	// get handler for this msg type.
	handler := dht.handlerForMsgType(pmes.GetType())
	if handler == nil {
		return nil, errors.New("Recieved invalid message type")
	}

	// dispatch handler.
	rpmes, err := handler(mPeer, pmes)
	if err != nil {
		return nil, err
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
145 146 147 148 149
	// if nil response, return it before serializing
	if rpmes == nil {
		return nil, nil
	}

150 151 152 153 154 155 156 157 158
	// serialize response msg
	rmes, err := msg.FromObject(mPeer, rpmes)
	if err != nil {
		return nil, fmt.Errorf("Failed to encode protobuf message: %v\n", err)
	}

	return rmes, nil
}

159 160 161 162 163 164 165 166 167 168 169
// sendRequest sends out a request using dht.sender, but also makes sure to
// measure the RTT for latency measurements.
func (dht *IpfsDHT) sendRequest(ctx context.Context, p *peer.Peer, pmes *Message) (*Message, error) {

	mes, err := msg.FromObject(p, pmes)
	if err != nil {
		return nil, err
	}

	start := time.Now()

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
170 171 172 173 174
	// Print out diagnostic
	u.DOut("[peer: %s] Sent message type: '%s' [to = %s]\n",
		dht.self.ID.Pretty(),
		Message_MessageType_name[int32(pmes.GetType())], p.ID.Pretty())

175 176 177 178
	rmes, err := dht.sender.SendRequest(ctx, mes)
	if err != nil {
		return nil, err
	}
179 180 181
	if rmes == nil {
		return nil, errors.New("no response to request")
	}
182 183 184 185 186 187 188 189 190 191 192 193

	rtt := time.Since(start)
	rmes.Peer().SetLatency(rtt)

	rpmes := new(Message)
	if err := proto.Unmarshal(rmes.Data(), rpmes); err != nil {
		return nil, err
	}

	return rpmes, nil
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
194 195 196
func (dht *IpfsDHT) putValueToNetwork(ctx context.Context, p *peer.Peer,
	key string, value []byte) error {

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
197 198
	pmes := newMessage(Message_PUT_VALUE, string(key), 0)
	pmes.Value = value
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
199
	rpmes, err := dht.sendRequest(ctx, p, pmes)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
200 201 202
	if err != nil {
		return err
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
203 204 205 206 207

	if !bytes.Equal(rpmes.Value, pmes.Value) {
		return errors.New("value not put correctly")
	}
	return nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
208 209 210 211 212 213 214 215 216 217 218 219
}

func (dht *IpfsDHT) putProvider(ctx context.Context, p *peer.Peer, key string) error {
	pmes := newMessage(Message_ADD_PROVIDER, string(key), 0)

	mes, err := msg.FromObject(p, pmes)
	if err != nil {
		return err
	}
	return dht.sender.SendMessage(ctx, mes)
}

220 221 222 223
func (dht *IpfsDHT) getValueOrPeers(ctx context.Context, p *peer.Peer,
	key u.Key, level int) ([]byte, []*peer.Peer, error) {

	pmes, err := dht.getValueSingle(ctx, p, key, level)
224
	if err != nil {
225
		return nil, nil, err
226 227
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
228
	u.DOut("pmes.GetValue() %v\n", pmes.GetValue())
229
	if value := pmes.GetValue(); value != nil {
230
		// Success! We were given the value
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
231
		u.DOut("getValueOrPeers: got value\n")
232
		return value, nil, nil
233
	}
234

235
	// TODO decide on providers. This probably shouldn't be happening.
236 237 238 239 240
	if prv := pmes.GetProviderPeers(); prv != nil && len(prv) > 0 {
		val, err := dht.getFromPeerList(ctx, key, prv, level)
		if err != nil {
			return nil, nil, err
		}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
241
		u.DOut("getValueOrPeers: get from providers\n")
242 243
		return val, nil, nil
	}
244 245

	// Perhaps we were given closer peers
246
	var peers []*peer.Peer
247
	for _, pb := range pmes.GetCloserPeers() {
248 249 250
		if peer.ID(pb.GetId()).Equal(dht.self.ID) {
			continue
		}
251

252 253
		addr, err := ma.NewMultiaddr(pb.GetAddr())
		if err != nil {
254
			u.PErr("%v\n", err.Error())
255 256
			continue
		}
257

258 259 260 261 262
		// check if we already have this peer.
		pr, _ := dht.peerstore.Get(peer.ID(pb.GetId()))
		if pr == nil {
			pr = &peer.Peer{ID: peer.ID(pb.GetId())}
			dht.peerstore.Put(pr)
263
		}
264 265
		pr.AddAddress(addr) // idempotent
		peers = append(peers, pr)
266
	}
267 268

	if len(peers) > 0 {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
269
		u.DOut("getValueOrPeers: peers\n")
270 271 272
		return nil, peers, nil
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
273
	u.DOut("getValueOrPeers: u.ErrNotFound\n")
274
	return nil, nil, u.ErrNotFound
275 276
}

277
// getValueSingle simply performs the get value RPC with the given parameters
278 279 280
func (dht *IpfsDHT) getValueSingle(ctx context.Context, p *peer.Peer,
	key u.Key, level int) (*Message, error) {

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
281
	pmes := newMessage(Message_GET_VALUE, string(key), level)
282
	return dht.sendRequest(ctx, p, pmes)
Jeromy's avatar
Jeromy committed
283 284
}

285 286 287 288
// TODO: Im not certain on this implementation, we get a list of peers/providers
// from someone what do we do with it? Connect to each of them? randomly pick
// one to get the value from? Or just connect to one at a time until we get a
// successful connection and request the value from it?
289 290
func (dht *IpfsDHT) getFromPeerList(ctx context.Context, key u.Key,
	peerlist []*Message_Peer, level int) ([]byte, error) {
291

292
	for _, pinfo := range peerlist {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
293
		p, err := dht.ensureConnectedToPeer(pinfo)
294 295 296
		if err != nil {
			u.DErr("getFromPeers error: %s\n", err)
			continue
Jeromy's avatar
Jeromy committed
297
		}
298 299

		pmes, err := dht.getValueSingle(ctx, p, key, level)
Jeromy's avatar
Jeromy committed
300
		if err != nil {
301
			u.DErr("getFromPeers error: %s\n", err)
Jeromy's avatar
Jeromy committed
302 303 304
			continue
		}

305 306 307 308
		if value := pmes.GetValue(); value != nil {
			// Success! We were given the value
			dht.providers.AddProvider(key, p)
			return value, nil
309
		}
Jeromy's avatar
Jeromy committed
310 311 312 313
	}
	return nil, u.ErrNotFound
}

314
func (dht *IpfsDHT) getLocal(key u.Key) ([]byte, error) {
315 316
	dht.dslock.Lock()
	defer dht.dslock.Unlock()
317
	v, err := dht.datastore.Get(ds.NewKey(string(key)))
318 319 320
	if err != nil {
		return nil, err
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
321 322 323 324 325 326

	byt, ok := v.([]byte)
	if !ok {
		return byt, errors.New("value stored in datastore not []byte")
	}
	return byt, nil
327 328
}

329
func (dht *IpfsDHT) putLocal(key u.Key, value []byte) error {
330 331
	return dht.datastore.Put(ds.NewKey(string(key)), value)
}
332

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
333 334
// Update signals to all routingTables to Update their last-seen status
// on the given peer.
335
func (dht *IpfsDHT) Update(p *peer.Peer) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
336
	removedCount := 0
337
	for _, route := range dht.routingTables {
338
		removed := route.Update(p)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
339
		// Only close the connection if no tables refer to this peer
340
		if removed != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
341
			removedCount++
342
		}
343
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
344 345 346 347 348 349 350 351

	// Only close the connection if no tables refer to this peer
	// if removedCount == len(dht.routingTables) {
	// 	dht.network.ClosePeer(p)
	// }
	// ACTUALLY, no, let's not just close the connection. it may be connected
	// due to other things. it seems that we just need connection timeouts
	// after some deadline of inactivity.
352
}
Jeromy's avatar
Jeromy committed
353

354
// Find looks for a peer with a given ID connected to this dht and returns the peer and the table it was found in.
Jeromy's avatar
Jeromy committed
355
func (dht *IpfsDHT) Find(id peer.ID) (*peer.Peer, *kb.RoutingTable) {
356
	for _, table := range dht.routingTables {
Jeromy's avatar
Jeromy committed
357 358 359 360 361 362 363
		p := table.Find(id)
		if p != nil {
			return p, table
		}
	}
	return nil, nil
}
364

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
365 366 367
func (dht *IpfsDHT) findPeerSingle(ctx context.Context, p *peer.Peer, id peer.ID, level int) (*Message, error) {
	pmes := newMessage(Message_FIND_NODE, string(id), level)
	return dht.sendRequest(ctx, p, pmes)
368
}
369

370 371
func (dht *IpfsDHT) printTables() {
	for _, route := range dht.routingTables {
372 373 374
		route.Print()
	}
}
Jeromy's avatar
Jeromy committed
375

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
376 377 378
func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p *peer.Peer, key u.Key, level int) (*Message, error) {
	pmes := newMessage(Message_GET_PROVIDERS, string(key), level)
	return dht.sendRequest(ctx, p, pmes)
Jeromy's avatar
Jeromy committed
379 380
}

381
// TODO: Could be done async
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
382
func (dht *IpfsDHT) addProviders(key u.Key, peers []*Message_Peer) []*peer.Peer {
383
	var provArr []*peer.Peer
Jeromy's avatar
Jeromy committed
384
	for _, prov := range peers {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
385 386 387
		p, err := dht.peerFromInfo(prov)
		if err != nil {
			u.PErr("error getting peer from info: %v\n", err)
Jeromy's avatar
Jeromy committed
388 389
			continue
		}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
390 391 392 393

		// Dont add outselves to the list
		if p.ID.Equal(dht.self.ID) {
			continue
Jeromy's avatar
Jeromy committed
394
		}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
395 396

		// TODO(jbenet) ensure providers is idempotent
397
		dht.providers.AddProvider(key, p)
398
		provArr = append(provArr, p)
Jeromy's avatar
Jeromy committed
399
	}
400
	return provArr
Jeromy's avatar
Jeromy committed
401
}
Jeromy's avatar
Jeromy committed
402

403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428
// nearestPeerToQuery returns the routing tables closest peers.
func (dht *IpfsDHT) nearestPeerToQuery(pmes *Message) *peer.Peer {
	level := pmes.GetClusterLevel()
	cluster := dht.routingTables[level]

	key := u.Key(pmes.GetKey())
	closer := cluster.NearestPeer(kb.ConvertKey(key))
	return closer
}

// betterPeerToQuery returns nearestPeerToQuery, but iff closer than self.
func (dht *IpfsDHT) betterPeerToQuery(pmes *Message) *peer.Peer {
	closer := dht.nearestPeerToQuery(pmes)

	// no node? nil
	if closer == nil {
		return nil
	}

	// == to self? nil
	if closer.ID.Equal(dht.self.ID) {
		u.DOut("Attempted to return self! this shouldnt happen...\n")
		return nil
	}

	// self is closer? nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
429
	key := u.Key(pmes.GetKey())
430 431 432 433 434 435 436 437
	if kb.Closer(dht.self.ID, closer.ID, key) {
		return nil
	}

	// ok seems like a closer node.
	return closer
}

438 439 440
func (dht *IpfsDHT) peerFromInfo(pbp *Message_Peer) (*peer.Peer, error) {

	id := peer.ID(pbp.GetId())
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
441 442 443 444 445 446

	// continue if it's ourselves
	if id.Equal(dht.self.ID) {
		return nil, errors.New("found self")
	}

447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463
	p, _ := dht.peerstore.Get(id)
	if p == nil {
		p, _ = dht.Find(id)
		if p != nil {
			panic("somehow peer not getting into peerstore")
		}
	}

	if p == nil {
		maddr, err := ma.NewMultiaddr(pbp.GetAddr())
		if err != nil {
			return nil, err
		}

		// create new Peer
		p := &peer.Peer{ID: id}
		p.AddAddress(maddr)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
464 465 466 467 468 469 470 471 472
		dht.peerstore.Put(p)
	}
	return p, nil
}

func (dht *IpfsDHT) ensureConnectedToPeer(pbp *Message_Peer) (*peer.Peer, error) {
	p, err := dht.peerFromInfo(pbp)
	if err != nil {
		return nil, err
Jeromy's avatar
Jeromy committed
473 474
	}

475
	// dial connection
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
476
	err = dht.network.DialPeer(p)
477
	return p, err
Jeromy's avatar
Jeromy committed
478
}
479 480

func (dht *IpfsDHT) loadProvidableKeys() error {
481 482 483 484
	kl, err := dht.datastore.KeyList()
	if err != nil {
		return err
	}
485 486 487 488 489 490
	for _, k := range kl {
		dht.providers.AddProvider(u.Key(k.Bytes()), dht.self)
	}
	return nil
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
491
// Bootstrap builds up list of peers by requesting random peer IDs
492 493 494 495 496
func (dht *IpfsDHT) Bootstrap() {
	id := make([]byte, 16)
	rand.Read(id)
	dht.FindPeer(peer.ID(id), time.Second*10)
}