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

import (
4
	"bytes"
5
	"fmt"
Jeromy's avatar
Jeromy committed
6
	"sync"
Jeromy's avatar
Jeromy committed
7
	"time"
8

9
	key "github.com/ipfs/go-ipfs/blocks/key"
10 11 12 13 14
	notif "github.com/ipfs/go-ipfs/notifications"
	"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"
15
	pset "github.com/ipfs/go-ipfs/thirdparty/peerset"
Jeromy's avatar
Jeromy committed
16 17

	peer "gx/ipfs/QmQGwpJy9P4yXZySmqkZEXCmbBpJUb8xntCv8Ca4taZwDC/go-libp2p-peer"
Jeromy's avatar
Jeromy committed
18
	pstore "gx/ipfs/QmXHUpFsnpCmanRnacqYkFoLoFfEq5yS2nUgGkAjJ1Nj9j/go-libp2p-peerstore"
Jeromy's avatar
Jeromy committed
19
	context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
Jeromy's avatar
Jeromy committed
20
	inet "gx/ipfs/QmdBpVuSYuTGDA8Kn66CbKvEThXqKUh2nTANZEhzSxqrmJ/go-libp2p/p2p/net"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
21 22
)

23 24 25 26 27 28
// asyncQueryBuffer is the size of buffered channels in async queries. This
// buffer allows multiple queries to execute simultaneously, return their
// results and continue querying closer peers. Note that different query
// results will wait for the channel to drain.
var asyncQueryBuffer = 10

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
29 30 31 32 33
// This file implements the Routing interface for the IpfsDHT struct.

// Basic Put/Get

// PutValue adds value corresponding to given Key.
34
// This is the top level "Store" operation of the DHT
35
func (dht *IpfsDHT) PutValue(ctx context.Context, key key.Key, value []byte) error {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
36
	log.Debugf("PutValue %s", key)
Jeromy's avatar
Jeromy committed
37
	sk, err := dht.getOwnPrivateKey()
Jeromy's avatar
Jeromy committed
38 39 40
	if err != nil {
		return err
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
41

42 43 44 45 46
	sign, err := dht.Validator.IsSigned(key)
	if err != nil {
		return err
	}

Jeromy's avatar
Jeromy committed
47
	rec, err := record.MakePutRecord(sk, key, value, sign)
Jeromy's avatar
Jeromy committed
48
	if err != nil {
49
		log.Debug("creation of record failed!")
Jeromy's avatar
Jeromy committed
50 51 52
		return err
	}

Jeromy's avatar
Jeromy committed
53
	err = dht.putLocal(key, rec)
54 55 56 57
	if err != nil {
		return err
	}

58
	pchan, err := dht.GetClosestPeers(ctx, key)
59 60 61
	if err != nil {
		return err
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
62

63 64 65 66
	wg := sync.WaitGroup{}
	for p := range pchan {
		wg.Add(1)
		go func(p peer.ID) {
Jeromy's avatar
Jeromy committed
67 68
			ctx, cancel := context.WithCancel(ctx)
			defer cancel()
69
			defer wg.Done()
Jeromy's avatar
Jeromy committed
70 71 72 73 74
			notif.PublishQueryEvent(ctx, &notif.QueryEvent{
				Type: notif.Value,
				ID:   p,
			})

Jeromy's avatar
Jeromy committed
75
			err := dht.putValueToPeer(ctx, p, key, rec)
76
			if err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
77
				log.Debugf("failed putting value to peer: %s", err)
78 79 80 81 82
			}
		}(p)
	}
	wg.Wait()
	return nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
83 84 85
}

// GetValue searches for the value corresponding to given Key.
86
func (dht *IpfsDHT) GetValue(ctx context.Context, key key.Key) ([]byte, error) {
Jeromy's avatar
Jeromy committed
87 88 89 90
	ctx, cancel := context.WithTimeout(ctx, time.Minute)
	defer cancel()

	vals, err := dht.GetValues(ctx, key, 16)
91 92 93 94 95 96
	if err != nil {
		return nil, err
	}

	var recs [][]byte
	for _, v := range vals {
97 98 99
		if v.Val != nil {
			recs = append(recs, v.Val)
		}
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
	}

	i, err := dht.Selector.BestRecord(key, recs)
	if err != nil {
		return nil, err
	}

	best := recs[i]
	log.Debugf("GetValue %v %v", key, best)
	if best == nil {
		log.Errorf("GetValue yielded correct record with nil value.")
		return nil, routing.ErrNotFound
	}

	fixupRec, err := record.MakePutRecord(dht.peerstore.PrivKey(dht.self), key, best, true)
	if err != nil {
		// probably shouldnt actually 'error' here as we have found a value we like,
		// but this call failing probably isnt something we want to ignore
		return nil, err
	}

	for _, v := range vals {
		// if someone sent us a different 'less-valid' record, lets correct them
		if !bytes.Equal(v.Val, best) {
			go func(v routing.RecvdVal) {
Jeromy's avatar
Jeromy committed
125 126
				ctx, cancel := context.WithTimeout(dht.Context(), time.Second*30)
				defer cancel()
127 128 129 130 131 132 133 134 135 136 137 138 139 140 141
				err := dht.putValueToPeer(ctx, v.From, key, fixupRec)
				if err != nil {
					log.Error("Error correcting DHT entry: ", err)
				}
			}(v)
		}
	}

	return best, nil
}

func (dht *IpfsDHT) GetValues(ctx context.Context, key key.Key, nvals int) ([]routing.RecvdVal, error) {
	var vals []routing.RecvdVal
	var valslock sync.Mutex

Jeromy's avatar
Jeromy committed
142
	// If we have it local, dont bother doing an RPC!
143
	lrec, err := dht.getLocal(key)
Jeromy's avatar
Jeromy committed
144
	if err == nil {
145 146
		// TODO: this is tricky, we dont always want to trust our own value
		// what if the authoritative source updated it?
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
147
		log.Debug("have it locally")
148 149 150 151 152 153 154 155 156 157
		vals = append(vals, routing.RecvdVal{
			Val:  lrec.GetValue(),
			From: dht.self,
		})

		if nvals <= 1 {
			return vals, nil
		}
	} else if nvals == 0 {
		return nil, err
Jeromy's avatar
Jeromy committed
158 159
	}

160
	// get closest peers in the routing table
Jeromy's avatar
Jeromy committed
161
	rtp := dht.routingTable.NearestPeers(kb.ConvertKey(key), KValue)
Jeromy's avatar
Jeromy committed
162
	log.Debugf("peers in rt: %s", len(rtp), rtp)
163
	if len(rtp) == 0 {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
164
		log.Warning("No peers from routing table!")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
165
		return nil, kb.ErrLookupFailure
166 167
	}

168
	// setup the Query
Jeromy's avatar
Jeromy committed
169
	parent := ctx
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
170
	query := dht.newQuery(key, func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
Jeromy's avatar
Jeromy committed
171
		notif.PublishQueryEvent(parent, &notif.QueryEvent{
Jeromy's avatar
Jeromy committed
172 173 174 175
			Type: notif.SendingQuery,
			ID:   p,
		})

176
		rec, peers, err := dht.getValueOrPeers(ctx, p, key)
177 178 179 180 181 182 183 184 185
		switch err {
		case routing.ErrNotFound:
			// in this case, they responded with nothing,
			// still send a notification so listeners can know the
			// request has completed 'successfully'
			notif.PublishQueryEvent(parent, &notif.QueryEvent{
				Type: notif.PeerResponse,
				ID:   p,
			})
186
			return nil, err
187 188 189 190 191
		default:
			return nil, err

		case nil, errInvalidRecord:
			// in either of these cases, we want to keep going
192
		}
193

194 195
		res := &dhtQueryResult{closerPeers: peers}

196
		if rec.GetValue() != nil || err == errInvalidRecord {
197 198 199 200 201 202 203 204 205 206 207 208
			rv := routing.RecvdVal{
				Val:  rec.GetValue(),
				From: p,
			}
			valslock.Lock()
			vals = append(vals, rv)

			// If weve collected enough records, we're done
			if len(vals) >= nvals {
				res.success = true
			}
			valslock.Unlock()
209 210
		}

Jeromy's avatar
Jeromy committed
211
		notif.PublishQueryEvent(parent, &notif.QueryEvent{
Jeromy's avatar
Jeromy committed
212 213 214 215 216
			Type:      notif.PeerResponse,
			ID:        p,
			Responses: pointerizePeerInfos(peers),
		})

217 218
		return res, nil
	})
219

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
220
	// run it!
221 222 223 224 225
	_, err = query.Run(ctx, rtp)
	if len(vals) == 0 {
		if err != nil {
			return nil, err
		}
226 227
	}

228
	return vals, nil
229

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
230 231 232 233 234
}

// Value provider layer of indirection.
// This is what DSHTs (Coral and MainlineDHT) do to store large values in a DHT.

235
// Provide makes this node announce that it can provide a value for the given key
236
func (dht *IpfsDHT) Provide(ctx context.Context, key key.Key) error {
Jeromy's avatar
Jeromy committed
237
	defer log.EventBegin(ctx, "provide", &key).Done()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
238 239

	// add self locally
240
	dht.providers.AddProvider(ctx, key, dht.self)
241

242
	peers, err := dht.GetClosestPeers(ctx, key)
243 244
	if err != nil {
		return err
245 246
	}

247 248 249 250 251
	mes, err := dht.makeProvRecord(key)
	if err != nil {
		return err
	}

Jeromy's avatar
Jeromy committed
252
	wg := sync.WaitGroup{}
253
	for p := range peers {
Jeromy's avatar
Jeromy committed
254 255 256
		wg.Add(1)
		go func(p peer.ID) {
			defer wg.Done()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
257
			log.Debugf("putProvider(%s, %s)", key, p)
258
			err := dht.sendMessage(ctx, p, mes)
Jeromy's avatar
Jeromy committed
259
			if err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
260
				log.Debug(err)
Jeromy's avatar
Jeromy committed
261 262
			}
		}(p)
263
	}
Jeromy's avatar
Jeromy committed
264
	wg.Wait()
265
	return nil
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
266
}
267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282
func (dht *IpfsDHT) makeProvRecord(skey key.Key) (*pb.Message, error) {
	pi := pstore.PeerInfo{
		ID:    dht.self,
		Addrs: dht.host.Addrs(),
	}

	// // only share WAN-friendly addresses ??
	// pi.Addrs = addrutil.WANShareableAddrs(pi.Addrs)
	if len(pi.Addrs) < 1 {
		return nil, fmt.Errorf("no known addresses for self. cannot put provider.")
	}

	pmes := pb.NewMessage(pb.Message_ADD_PROVIDER, string(skey), 0)
	pmes.ProviderPeers = pb.RawPeerInfosToPBPeers([]pstore.PeerInfo{pi})
	return pmes, nil
}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
283

Brian Tiger Chow's avatar
Brian Tiger Chow committed
284
// FindProviders searches until the context expires.
Jeromy's avatar
Jeromy committed
285 286
func (dht *IpfsDHT) FindProviders(ctx context.Context, key key.Key) ([]pstore.PeerInfo, error) {
	var providers []pstore.PeerInfo
Jeromy's avatar
Jeromy committed
287
	for p := range dht.FindProvidersAsync(ctx, key, KValue) {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
288 289 290 291 292
		providers = append(providers, p)
	}
	return providers, nil
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
293 294 295
// FindProvidersAsync is the same thing as FindProviders, but returns a channel.
// Peers will be returned on the channel as soon as they are found, even before
// the search query completes.
Jeromy's avatar
Jeromy committed
296
func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key key.Key, count int) <-chan pstore.PeerInfo {
297
	log.Event(ctx, "findProviders", &key)
Jeromy's avatar
Jeromy committed
298
	peerOut := make(chan pstore.PeerInfo, count)
Jeromy's avatar
Jeromy committed
299 300 301 302
	go dht.findProvidersAsyncRoutine(ctx, key, count, peerOut)
	return peerOut
}

Jeromy's avatar
Jeromy committed
303
func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key key.Key, count int, peerOut chan pstore.PeerInfo) {
Jeromy's avatar
Jeromy committed
304
	defer log.EventBegin(ctx, "findProvidersAsync", &key).Done()
Jeromy's avatar
Jeromy committed
305 306
	defer close(peerOut)

Jeromy's avatar
Jeromy committed
307
	ps := pset.NewLimited(count)
Jeromy's avatar
Jeromy committed
308 309
	provs := dht.providers.GetProviders(ctx, key)
	for _, p := range provs {
310
		// NOTE: Assuming that this list of peers is unique
Jeromy's avatar
Jeromy committed
311
		if ps.TryAdd(p) {
Jeromy's avatar
Jeromy committed
312
			select {
313
			case peerOut <- dht.peerstore.PeerInfo(p):
Jeromy's avatar
Jeromy committed
314 315 316
			case <-ctx.Done():
				return
			}
Jeromy's avatar
Jeromy committed
317
		}
Jeromy's avatar
Jeromy committed
318 319 320

		// If we have enough peers locally, dont bother with remote RPC
		if ps.Size() >= count {
Jeromy's avatar
Jeromy committed
321 322 323 324 325
			return
		}
	}

	// setup the Query
Jeromy's avatar
Jeromy committed
326
	parent := ctx
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
327
	query := dht.newQuery(key, func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
Jeromy's avatar
Jeromy committed
328
		notif.PublishQueryEvent(parent, &notif.QueryEvent{
329 330 331
			Type: notif.SendingQuery,
			ID:   p,
		})
332
		pmes, err := dht.findProvidersSingle(ctx, p, key)
Jeromy's avatar
Jeromy committed
333 334 335 336
		if err != nil {
			return nil, err
		}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
337
		log.Debugf("%d provider entries", len(pmes.GetProviderPeers()))
338
		provs := pb.PBPeersToPeerInfos(pmes.GetProviderPeers())
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
339
		log.Debugf("%d provider entries decoded", len(provs))
Jeromy's avatar
Jeromy committed
340 341 342

		// Add unique providers from request, up to 'count'
		for _, prov := range provs {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
343
			log.Debugf("got provider: %s", prov)
344
			if ps.TryAdd(prov.ID) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
345
				log.Debugf("using provider: %s", prov)
Jeromy's avatar
Jeromy committed
346 347 348
				select {
				case peerOut <- prov:
				case <-ctx.Done():
349
					log.Debug("context timed out sending more providers")
Jeromy's avatar
Jeromy committed
350 351
					return nil, ctx.Err()
				}
352
			}
Jeromy's avatar
Jeromy committed
353
			if ps.Size() >= count {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
354
				log.Debugf("got enough providers (%d/%d)", ps.Size(), count)
Jeromy's avatar
Jeromy committed
355
				return &dhtQueryResult{success: true}, nil
356 357 358
			}
		}

Jeromy's avatar
Jeromy committed
359 360
		// Give closer peers back to the query to be queried
		closer := pmes.GetCloserPeers()
361
		clpeers := pb.PBPeersToPeerInfos(closer)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
362
		log.Debugf("got closer peers: %d %s", len(clpeers), clpeers)
363

Jeromy's avatar
Jeromy committed
364
		notif.PublishQueryEvent(parent, &notif.QueryEvent{
365 366 367 368
			Type:      notif.PeerResponse,
			ID:        p,
			Responses: pointerizePeerInfos(clpeers),
		})
Jeromy's avatar
Jeromy committed
369 370 371
		return &dhtQueryResult{closerPeers: clpeers}, nil
	})

Jeromy's avatar
Jeromy committed
372
	peers := dht.routingTable.NearestPeers(kb.ConvertKey(key), KValue)
Jeromy's avatar
Jeromy committed
373 374
	_, err := query.Run(ctx, peers)
	if err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
375
		log.Debugf("Query error: %s", err)
376 377 378 379
		notif.PublishQueryEvent(ctx, &notif.QueryEvent{
			Type:  notif.QueryError,
			Extra: err.Error(),
		})
Jeromy's avatar
Jeromy committed
380
	}
381 382
}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
383
// FindPeer searches for a peer with given ID.
Jeromy's avatar
Jeromy committed
384
func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (pstore.PeerInfo, error) {
Jeromy's avatar
Jeromy committed
385
	defer log.EventBegin(ctx, "FindPeer", id).Done()
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
386

387
	// Check if were already connected to them
Jeromy's avatar
Jeromy committed
388
	if pi := dht.FindLocal(id); pi.ID != "" {
389
		return pi, nil
390 391
	}

Jeromy's avatar
Jeromy committed
392
	peers := dht.routingTable.NearestPeers(kb.ConvertPeerID(id), KValue)
393
	if len(peers) == 0 {
Jeromy's avatar
Jeromy committed
394
		return pstore.PeerInfo{}, kb.ErrLookupFailure
395
	}
396

Jeromy's avatar
Jeromy committed
397
	// Sanity...
398
	for _, p := range peers {
399
		if p == id {
400
			log.Debug("found target peer in list of closest peers...")
401
			return dht.peerstore.PeerInfo(p), nil
402
		}
403
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
404

Jeromy's avatar
Jeromy committed
405
	// setup the Query
Jeromy's avatar
Jeromy committed
406
	parent := ctx
407
	query := dht.newQuery(key.Key(id), func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
Jeromy's avatar
Jeromy committed
408
		notif.PublishQueryEvent(parent, &notif.QueryEvent{
409 410 411
			Type: notif.SendingQuery,
			ID:   p,
		})
Jeromy's avatar
Jeromy committed
412

413
		pmes, err := dht.findPeerSingle(ctx, p, id)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
414
		if err != nil {
415
			return nil, err
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
416
		}
417

Jeromy's avatar
Jeromy committed
418
		closer := pmes.GetCloserPeers()
419
		clpeerInfos := pb.PBPeersToPeerInfos(closer)
420

421
		// see it we got the peer here
422 423
		for _, npi := range clpeerInfos {
			if npi.ID == id {
Jeromy's avatar
Jeromy committed
424
				return &dhtQueryResult{
425
					peer:    npi,
Jeromy's avatar
Jeromy committed
426 427 428
					success: true,
				}, nil
			}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
429 430
		}

Jeromy's avatar
Jeromy committed
431
		notif.PublishQueryEvent(parent, &notif.QueryEvent{
432 433 434 435
			Type:      notif.PeerResponse,
			Responses: pointerizePeerInfos(clpeerInfos),
		})

436
		return &dhtQueryResult{closerPeers: clpeerInfos}, nil
437
	})
438

Jeromy's avatar
Jeromy committed
439
	// run it!
440
	result, err := query.Run(ctx, peers)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
441
	if err != nil {
Jeromy's avatar
Jeromy committed
442
		return pstore.PeerInfo{}, err
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
443 444
	}

445
	log.Debugf("FindPeer %v %v", id, result.success)
446
	if result.peer.ID == "" {
Jeromy's avatar
Jeromy committed
447
		return pstore.PeerInfo{}, routing.ErrNotFound
448
	}
Jeromy's avatar
Jeromy committed
449

450
	return result.peer, nil
451 452
}

453
// FindPeersConnectedToPeer searches for peers directly connected to a given peer.
Jeromy's avatar
Jeromy committed
454
func (dht *IpfsDHT) FindPeersConnectedToPeer(ctx context.Context, id peer.ID) (<-chan pstore.PeerInfo, error) {
455

Jeromy's avatar
Jeromy committed
456 457
	peerchan := make(chan pstore.PeerInfo, asyncQueryBuffer)
	peersSeen := make(map[peer.ID]struct{})
458

Jeromy's avatar
Jeromy committed
459
	peers := dht.routingTable.NearestPeers(kb.ConvertPeerID(id), KValue)
460
	if len(peers) == 0 {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
461
		return nil, kb.ErrLookupFailure
462 463 464
	}

	// setup the Query
465
	query := dht.newQuery(key.Key(id), func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
466

467
		pmes, err := dht.findPeerSingle(ctx, p, id)
468 469 470 471
		if err != nil {
			return nil, err
		}

Jeromy's avatar
Jeromy committed
472
		var clpeers []pstore.PeerInfo
473 474
		closer := pmes.GetCloserPeers()
		for _, pbp := range closer {
475
			pi := pb.PBPeerToPeerInfo(pbp)
476

477 478
			// skip peers already seen
			if _, found := peersSeen[pi.ID]; found {
479 480
				continue
			}
481
			peersSeen[pi.ID] = struct{}{}
482 483 484 485 486 487

			// if peer is connected, send it to our client.
			if pb.Connectedness(*pbp.Connection) == inet.Connected {
				select {
				case <-ctx.Done():
					return nil, ctx.Err()
488
				case peerchan <- pi:
489 490 491 492
				}
			}

			// if peer is the peer we're looking for, don't bother querying it.
493
			// TODO maybe query it?
494
			if pb.Connectedness(*pbp.Connection) != inet.Connected {
495
				clpeers = append(clpeers, pi)
496 497 498 499 500 501 502 503 504
			}
		}

		return &dhtQueryResult{closerPeers: clpeers}, nil
	})

	// run it! run it asynchronously to gen peers as results are found.
	// this does no error checking
	go func() {
505
		if _, err := query.Run(ctx, peers); err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
506
			log.Debug(err)
507 508 509 510 511 512 513 514
		}

		// close the peerchan channel when done.
		close(peerchan)
	}()

	return peerchan, nil
}