dht.go 10 KB
Newer Older
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
1
// Package dht implements a distributed hash table that satisfies the ipfs routing
2
// interface. This DHT is modeled after kademlia with Coral and S/Kademlia modifications.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
3 4
package dht

5
import (
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
6
	"bytes"
Jeromy's avatar
Jeromy committed
7
	"context"
8
	"errors"
9
	"fmt"
10 11
	"sync"
	"time"
12

13 14 15 16
	pb "github.com/libp2p/go-libp2p-kad-dht/pb"
	providers "github.com/libp2p/go-libp2p-kad-dht/providers"
	routing "github.com/libp2p/go-libp2p-routing"

17
	proto "github.com/gogo/protobuf/proto"
18
	cid "github.com/ipfs/go-cid"
19 20 21 22
	ds "github.com/ipfs/go-datastore"
	logging "github.com/ipfs/go-log"
	goprocess "github.com/jbenet/goprocess"
	goprocessctx "github.com/jbenet/goprocess/context"
23 24
	ci "github.com/libp2p/go-libp2p-crypto"
	host "github.com/libp2p/go-libp2p-host"
George Antoniadis's avatar
George Antoniadis committed
25
	kb "github.com/libp2p/go-libp2p-kbucket"
26 27 28
	peer "github.com/libp2p/go-libp2p-peer"
	pstore "github.com/libp2p/go-libp2p-peerstore"
	protocol "github.com/libp2p/go-libp2p-protocol"
George Antoniadis's avatar
George Antoniadis committed
29
	record "github.com/libp2p/go-libp2p-record"
George Antoniadis's avatar
George Antoniadis committed
30
	recpb "github.com/libp2p/go-libp2p-record/pb"
31
	base32 "github.com/whyrusleeping/base32"
32 33
)

Jeromy's avatar
Jeromy committed
34
var log = logging.Logger("dht")
35

Jeromy's avatar
Jeromy committed
36 37
var ProtocolDHT protocol.ID = "/ipfs/kad/1.0.0"
var ProtocolDHTOld protocol.ID = "/ipfs/dht"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
38

39 40 41 42
// NumBootstrapQueries defines the number of random dht queries to do to
// collect members of the routing table.
const NumBootstrapQueries = 5

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
43 44 45
// IpfsDHT is an implementation of Kademlia with Coral and S/Kademlia modifications.
// It is used to implement the base IpfsRouting module.
type IpfsDHT struct {
Jeromy's avatar
Jeromy committed
46 47 48
	host      host.Host        // the network services we need
	self      peer.ID          // Local peer (yourself)
	peerstore pstore.Peerstore // Peer Registry
49

50
	datastore ds.Datastore // Local data
51

52
	routingTable *kb.RoutingTable // Array of routing tables for differently distanced nodes
53
	providers    *providers.ProviderManager
54

Jeromy's avatar
Jeromy committed
55
	birth time.Time // When this peer started up
56

57
	Validator record.Validator // record validator funcs
58
	Selector  record.Selector  // record selection funcs
59

60 61
	ctx  context.Context
	proc goprocess.Process
62 63 64

	strmap map[peer.ID]*messageSender
	smlk   sync.Mutex
65

Steven Allen's avatar
Steven Allen committed
66
	plk sync.Mutex
67 68
}

Jeromy's avatar
Jeromy committed
69
// NewDHT creates a new DHT object with the given peer as the 'local' host
70
func NewDHT(ctx context.Context, h host.Host, dstore ds.Batching) *IpfsDHT {
Justin Drake's avatar
Justin Drake committed
71
	dht := NewDHTClient(ctx, h, dstore)
72

73
	h.SetStreamHandler(ProtocolDHT, dht.handleNewStream)
Jeromy's avatar
Jeromy committed
74
	h.SetStreamHandler(ProtocolDHTOld, dht.handleNewStream)
Jeromy's avatar
Jeromy committed
75

Jeromy's avatar
Jeromy committed
76
	return dht
77 78
}

79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99
// NewDHTClient creates a new DHT object with the given peer as the 'local' host
func NewDHTClient(ctx context.Context, h host.Host, dstore ds.Batching) *IpfsDHT {
	dht := makeDHT(ctx, h, dstore)

	// register for network notifs.
	dht.host.Network().Notify((*netNotifiee)(dht))

	dht.proc = goprocessctx.WithContextAndTeardown(ctx, func() error {
		// remove ourselves from network notifs.
		dht.host.Network().StopNotify((*netNotifiee)(dht))
		return nil
	})

	dht.proc.AddChild(dht.providers.Process())

	dht.Validator["pk"] = record.PublicKeyValidator
	dht.Selector["pk"] = record.PublicKeySelector

	return dht
}

Jeromy's avatar
Jeromy committed
100
func makeDHT(ctx context.Context, h host.Host, dstore ds.Batching) *IpfsDHT {
101 102 103 104 105 106 107 108 109 110
	rt := kb.NewRoutingTable(KValue, kb.ConvertPeerID(h.ID()), time.Minute, h.Peerstore())

	cmgr := h.ConnManager()
	rt.PeerAdded = func(p peer.ID) {
		cmgr.TagPeer(p, "kbucket", 5)
	}
	rt.PeerRemoved = func(p peer.ID) {
		cmgr.UntagPeer(p, "kbucket")
	}

Jeromy's avatar
Jeromy committed
111 112 113 114 115 116 117 118 119
	return &IpfsDHT{
		datastore:    dstore,
		self:         h.ID(),
		peerstore:    h.Peerstore(),
		host:         h,
		strmap:       make(map[peer.ID]*messageSender),
		ctx:          ctx,
		providers:    providers.NewProviderManager(ctx, h.ID(), dstore),
		birth:        time.Now(),
120
		routingTable: rt,
Jeromy's avatar
Jeromy committed
121 122 123 124 125 126

		Validator: make(record.Validator),
		Selector:  make(record.Selector),
	}
}

Jeromy's avatar
Jeromy committed
127 128
// putValueToPeer stores the given key/value pair at the peer 'p'
func (dht *IpfsDHT) putValueToPeer(ctx context.Context, p peer.ID,
129
	key string, rec *recpb.Record) error {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
130

131
	pmes := pb.NewMessage(pb.Message_PUT_VALUE, key, 0)
132
	pmes.Record = rec
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
133
	rpmes, err := dht.sendRequest(ctx, p, pmes)
134 135
	switch err {
	case ErrReadTimeout:
Jeromy's avatar
Jeromy committed
136
		log.Warningf("read timeout: %s %s", p.Pretty(), key)
137 138 139 140 141 142 143
		fallthrough
	default:
		return err
	case nil:
		break
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
144 145 146
	if err != nil {
		return err
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
147

148
	if !bytes.Equal(rpmes.GetRecord().Value, pmes.GetRecord().Value) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
149 150 151
		return errors.New("value not put correctly")
	}
	return nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
152 153
}

154 155
var errInvalidRecord = errors.New("received invalid record")

156 157
// getValueOrPeers queries a particular peer p for the value for
// key. It returns either the value or a list of closer peers.
158
// NOTE: It will update the dht's peerstore with any new addresses
159
// it finds for the given peer.
Jeromy's avatar
Jeromy committed
160
func (dht *IpfsDHT) getValueOrPeers(ctx context.Context, p peer.ID, key string) (*recpb.Record, []*pstore.PeerInfo, error) {
161

162
	pmes, err := dht.getValueSingle(ctx, p, key)
163
	if err != nil {
164
		return nil, nil, err
165 166
	}

167 168 169
	// Perhaps we were given closer peers
	peers := pb.PBPeersToPeerInfos(pmes.GetCloserPeers())

170
	if record := pmes.GetRecord(); record != nil {
171
		// Success! We were given the value
Jeromy's avatar
Jeromy committed
172
		log.Debug("getValueOrPeers: got value")
173

174
		// make sure record is valid.
175
		err = dht.Validator.VerifyRecord(record)
176
		if err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
177
			log.Info("Received invalid record! (discarded)")
178 179
			// return a sentinal to signify an invalid record was received
			err = errInvalidRecord
George Antoniadis's avatar
George Antoniadis committed
180
			record = new(recpb.Record)
181
		}
182
		return record, peers, err
183
	}
184

185
	if len(peers) > 0 {
186
		log.Debug("getValueOrPeers: peers")
187 188 189
		return nil, peers, nil
	}

190 191
	log.Warning("getValueOrPeers: routing.ErrNotFound")
	return nil, nil, routing.ErrNotFound
192 193
}

194
// getValueSingle simply performs the get value RPC with the given parameters
195 196 197 198 199 200
func (dht *IpfsDHT) getValueSingle(ctx context.Context, p peer.ID, key string) (*pb.Message, error) {
	meta := logging.LoggableMap{
		"key":  key,
		"peer": p,
	}

ForrestWeston's avatar
ForrestWeston committed
201 202
	eip := log.EventBegin(ctx, "getValueSingle", meta)
	defer eip.Done()
203

204
	pmes := pb.NewMessage(pb.Message_GET_VALUE, key, 0)
205 206 207 208 209
	resp, err := dht.sendRequest(ctx, p, pmes)
	switch err {
	case nil:
		return resp, nil
	case ErrReadTimeout:
Jeromy's avatar
Jeromy committed
210
		log.Warningf("read timeout: %s %s", p.Pretty(), key)
211 212
		fallthrough
	default:
ForrestWeston's avatar
ForrestWeston committed
213
		eip.SetError(err)
214 215
		return nil, err
	}
Jeromy's avatar
Jeromy committed
216 217
}

218
// getLocal attempts to retrieve the value from the datastore
219 220
func (dht *IpfsDHT) getLocal(key string) (*recpb.Record, error) {
	log.Debugf("getLocal %s", key)
221

222
	v, err := dht.datastore.Get(mkDsKey(key))
223 224 225
	if err != nil {
		return nil, err
	}
226
	log.Debugf("found %s in local datastore")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
227 228 229

	byt, ok := v.([]byte)
	if !ok {
230
		return nil, errors.New("value stored in datastore not []byte")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
231
	}
George Antoniadis's avatar
George Antoniadis committed
232
	rec := new(recpb.Record)
233 234 235 236 237
	err = proto.Unmarshal(byt, rec)
	if err != nil {
		return nil, err
	}

238
	err = dht.Validator.VerifyRecord(rec)
Jeromy's avatar
Jeromy committed
239 240 241
	if err != nil {
		log.Debugf("local record verify failed: %s (discarded)", err)
		return nil, err
242 243
	}

244
	return rec, nil
245 246
}

Jeromy's avatar
Jeromy committed
247 248
// getOwnPrivateKey attempts to load the local peers private
// key from the peerstore.
Jeromy's avatar
Jeromy committed
249 250 251
func (dht *IpfsDHT) getOwnPrivateKey() (ci.PrivKey, error) {
	sk := dht.peerstore.PrivKey(dht.self)
	if sk == nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
252
		log.Warningf("%s dht cannot get own private key!", dht.self)
Jeromy's avatar
Jeromy committed
253 254 255 256 257
		return nil, fmt.Errorf("cannot get private key to sign record!")
	}
	return sk, nil
}

258
// putLocal stores the key value pair in the datastore
259
func (dht *IpfsDHT) putLocal(key string, rec *recpb.Record) error {
260 261 262 263 264
	data, err := proto.Marshal(rec)
	if err != nil {
		return err
	}

265
	return dht.datastore.Put(mkDsKey(key), data)
266
}
267

268
// Update signals the routingTable to Update its last-seen status
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
269
// on the given peer.
270
func (dht *IpfsDHT) Update(ctx context.Context, p peer.ID) {
271
	log.Event(ctx, "updatePeer", p)
272
	dht.routingTable.Update(p)
273
}
Jeromy's avatar
Jeromy committed
274

Jeromy's avatar
Jeromy committed
275
// FindLocal 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
276
func (dht *IpfsDHT) FindLocal(id peer.ID) pstore.PeerInfo {
277
	p := dht.routingTable.Find(id)
278
	if p != "" {
Jeromy's avatar
Jeromy committed
279
		return dht.peerstore.PeerInfo(p)
Jeromy's avatar
Jeromy committed
280
	}
Jeromy's avatar
Jeromy committed
281
	return pstore.PeerInfo{}
Jeromy's avatar
Jeromy committed
282
}
283

Jeromy's avatar
Jeromy committed
284
// findPeerSingle asks peer 'p' if they know where the peer with id 'id' is
285
func (dht *IpfsDHT) findPeerSingle(ctx context.Context, p peer.ID, id peer.ID) (*pb.Message, error) {
ForrestWeston's avatar
ForrestWeston committed
286 287
	eip := log.EventBegin(ctx, "findPeerSingle", p, id)
	defer eip.Done()
288

289
	pmes := pb.NewMessage(pb.Message_FIND_NODE, string(id), 0)
290 291 292 293 294
	resp, err := dht.sendRequest(ctx, p, pmes)
	switch err {
	case nil:
		return resp, nil
	case ErrReadTimeout:
Jeromy's avatar
Jeromy committed
295
		log.Warningf("read timeout: %s %s", p.Pretty(), id)
296 297
		fallthrough
	default:
ForrestWeston's avatar
ForrestWeston committed
298
		eip.SetError(err)
299 300
		return nil, err
	}
301
}
302

303
func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key *cid.Cid) (*pb.Message, error) {
ForrestWeston's avatar
ForrestWeston committed
304 305
	eip := log.EventBegin(ctx, "findProvidersSingle", p, key)
	defer eip.Done()
306

307
	pmes := pb.NewMessage(pb.Message_GET_PROVIDERS, key.KeyString(), 0)
308 309 310 311 312
	resp, err := dht.sendRequest(ctx, p, pmes)
	switch err {
	case nil:
		return resp, nil
	case ErrReadTimeout:
Jeromy's avatar
Jeromy committed
313
		log.Warningf("read timeout: %s %s", p.Pretty(), key)
314 315
		fallthrough
	default:
ForrestWeston's avatar
ForrestWeston committed
316
		eip.SetError(err)
317 318
		return nil, err
	}
Jeromy's avatar
Jeromy committed
319 320
}

321
// nearestPeersToQuery returns the routing tables closest peers.
322
func (dht *IpfsDHT) nearestPeersToQuery(pmes *pb.Message, count int) []peer.ID {
323
	closer := dht.routingTable.NearestPeers(kb.ConvertKey(pmes.GetKey()), count)
324 325 326
	return closer
}

327
// betterPeerToQuery returns nearestPeersToQuery, but iff closer than self.
328
func (dht *IpfsDHT) betterPeersToQuery(pmes *pb.Message, p peer.ID, count int) []peer.ID {
329
	closer := dht.nearestPeersToQuery(pmes, count)
330 331 332

	// no node? nil
	if closer == nil {
Jeromy's avatar
Jeromy committed
333
		log.Warning("no closer peers to send:", p)
334 335 336
		return nil
	}

Steven Allen's avatar
Steven Allen committed
337
	filtered := make([]peer.ID, 0, len(closer))
Jeromy's avatar
Jeromy committed
338 339 340
	for _, clp := range closer {

		// == to self? thats bad
Jeromy's avatar
Jeromy committed
341
		if clp == dht.self {
Jeromy's avatar
Jeromy committed
342
			log.Warning("attempted to return self! this shouldn't happen...")
343 344
			return nil
		}
345
		// Dont send a peer back themselves
Jeromy's avatar
Jeromy committed
346
		if clp == p {
347 348 349
			continue
		}

Jeromy's avatar
Jeromy committed
350
		filtered = append(filtered, clp)
351 352
	}

353 354
	// ok seems like closer nodes
	return filtered
355 356
}

357 358 359 360 361 362 363 364 365 366 367 368 369 370
// Context return dht's context
func (dht *IpfsDHT) Context() context.Context {
	return dht.ctx
}

// Process return dht's process
func (dht *IpfsDHT) Process() goprocess.Process {
	return dht.proc
}

// Close calls Process Close
func (dht *IpfsDHT) Close() error {
	return dht.proc.Close()
}
371 372 373 374

func mkDsKey(s string) ds.Key {
	return ds.NewKey(base32.RawStdEncoding.EncodeToString([]byte(s)))
}