bitswap.go 12 KB
Newer Older
Brian Tiger Chow's avatar
Brian Tiger Chow committed
1 2
// package bitswap implements the IPFS Exchange interface with the BitSwap
// bilateral exchange protocol.
3 4 5
package bitswap

import (
6
	"math"
7
	"sync"
Jeromy's avatar
Jeromy committed
8 9
	"time"

10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25
	process "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
	blocks "github.com/ipfs/go-ipfs/blocks"
	blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
	exchange "github.com/ipfs/go-ipfs/exchange"
	decision "github.com/ipfs/go-ipfs/exchange/bitswap/decision"
	bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
	bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
	notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
	wantlist "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
	peer "github.com/ipfs/go-ipfs/p2p/peer"
	"github.com/ipfs/go-ipfs/thirdparty/delay"
	eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
	u "github.com/ipfs/go-ipfs/util"
	errors "github.com/ipfs/go-ipfs/util/debugerror"
	pset "github.com/ipfs/go-ipfs/util/peerset" // TODO move this to peerstore
26 27
)

28
var log = eventlog.Logger("bitswap")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
29

Brian Tiger Chow's avatar
Brian Tiger Chow committed
30
const (
Brian Tiger Chow's avatar
Brian Tiger Chow committed
31 32 33
	// maxProvidersPerRequest specifies the maximum number of providers desired
	// from the network. This value is specified because the network streams
	// results.
Brian Tiger Chow's avatar
Brian Tiger Chow committed
34 35 36 37
	// TODO: if a 'non-nice' strategy is implemented, consider increasing this value
	maxProvidersPerRequest = 3
	providerRequestTimeout = time.Second * 10
	hasBlockTimeout        = time.Second * 15
38
	provideTimeout         = time.Second * 15
Brian Tiger Chow's avatar
Brian Tiger Chow committed
39
	sizeBatchRequestChan   = 32
40 41
	// kMaxPriority is the max priority as defined by the bitswap protocol
	kMaxPriority = math.MaxInt32
42

Jeromy's avatar
Jeromy committed
43
	HasBlockBufferSize = 256
44
	provideWorkers     = 4
Brian Tiger Chow's avatar
Brian Tiger Chow committed
45
)
Jeromy's avatar
Jeromy committed
46

Brian Tiger Chow's avatar
Brian Tiger Chow committed
47
var (
Brian Tiger Chow's avatar
Brian Tiger Chow committed
48
	rebroadcastDelay = delay.Fixed(time.Second * 10)
Brian Tiger Chow's avatar
Brian Tiger Chow committed
49
)
50

Brian Tiger Chow's avatar
Brian Tiger Chow committed
51 52 53 54
// New initializes a BitSwap instance that communicates over the provided
// BitSwapNetwork. This function registers the returned instance as the network
// delegate.
// Runs until context is cancelled.
55
func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
56
	bstore blockstore.Blockstore, nice bool) exchange.Interface {
57

58 59 60 61 62 63 64
	// important to use provided parent context (since it may include important
	// loggable data). It's probably not a good idea to allow bitswap to be
	// coupled to the concerns of the IPFS daemon in this way.
	//
	// FIXME(btc) Now that bitswap manages itself using a process, it probably
	// shouldn't accept a context anymore. Clients should probably use Close()
	// exclusively. We should probably find another way to share logging data
65 66
	ctx, cancelFunc := context.WithCancel(parent)

67
	notif := notifications.New()
68 69 70 71 72
	px := process.WithTeardown(func() error {
		notif.Shutdown()
		return nil
	})

73
	go func() {
74
		<-px.Closing() // process closes first
Jeromy's avatar
Jeromy committed
75
		cancelFunc()
76 77 78 79
	}()
	go func() {
		<-ctx.Done() // parent cancelled first
		px.Close()
80 81
	}()

82
	bs := &Bitswap{
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
83
		self:          p,
84
		blockstore:    bstore,
85
		notifications: notif,
86
		engine:        decision.NewEngine(ctx, bstore), // TODO close the engine with Close() method
87
		network:       network,
88
		wantlist:      wantlist.NewThreadSafe(),
89
		batchRequests: make(chan *blockRequest, sizeBatchRequestChan),
90
		process:       px,
Jeromy's avatar
Jeromy committed
91
		newBlocks:     make(chan *blocks.Block, HasBlockBufferSize),
92
		provideKeys:   make(chan u.Key),
93
	}
Brian Tiger Chow's avatar
Brian Tiger Chow committed
94
	network.SetDelegate(bs)
95

96 97
	// Start up bitswaps async worker routines
	bs.startWorkers(px, ctx)
98 99 100
	return bs
}

101 102
// Bitswap instances implement the bitswap protocol.
type Bitswap struct {
103

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
104 105 106
	// the ID of the peer to act on behalf of
	self peer.ID

107 108
	// network delivers messages on behalf of the session
	network bsnet.BitSwapNetwork
109 110 111 112 113 114 115

	// blockstore is the local database
	// NB: ensure threadsafety
	blockstore blockstore.Blockstore

	notifications notifications.PubSub

116 117 118
	// Requests for a set of related blocks
	// the assumption is made that the same peer is likely to
	// have more than a single block in the set
119
	batchRequests chan *blockRequest
Jeromy's avatar
Jeromy committed
120

121
	engine *decision.Engine
122

123
	wantlist *wantlist.ThreadSafe
124

125
	process process.Process
126 127

	newBlocks chan *blocks.Block
128 129

	provideKeys chan u.Key
130 131
}

132 133 134 135 136
type blockRequest struct {
	keys []u.Key
	ctx  context.Context
}

137
// GetBlock attempts to retrieve a particular block from peers within the
138
// deadline enforced by the context.
139
func (bs *Bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, error) {
140

141 142 143 144
	// Any async work initiated by this function must end when this function
	// returns. To ensure this, derive a new context. Note that it is okay to
	// listen on parent in this scope, but NOT okay to pass |parent| to
	// functions called by this one. Otherwise those functions won't return
145 146
	// when this context's cancel func is executed. This is difficult to
	// enforce. May this comment keep you safe.
147

148
	ctx, cancelFunc := context.WithCancel(parent)
149

Jeromy's avatar
Jeromy committed
150
	ctx = eventlog.ContextWithLoggable(ctx, eventlog.Uuid("GetBlockRequest"))
Jeromy's avatar
Jeromy committed
151
	defer log.EventBegin(ctx, "GetBlockRequest", &k).Done()
152 153 154 155

	defer func() {
		cancelFunc()
	}()
156

157
	promise, err := bs.GetBlocks(ctx, []u.Key{k})
158 159
	if err != nil {
		return nil, err
Jeromy's avatar
Jeromy committed
160
	}
161 162

	select {
163 164 165 166 167 168 169 170 171
	case block, ok := <-promise:
		if !ok {
			select {
			case <-ctx.Done():
				return nil, ctx.Err()
			default:
				return nil, errors.New("promise channel was closed")
			}
		}
Jeromy's avatar
Jeromy committed
172
		return block, nil
173 174
	case <-parent.Done():
		return nil, parent.Err()
175 176 177
	}
}

178 179 180 181 182 183 184
// GetBlocks returns a channel where the caller may receive blocks that
// correspond to the provided |keys|. Returns an error if BitSwap is unable to
// begin this request within the deadline enforced by the context.
//
// NB: Your request remains open until the context expires. To conserve
// resources, provide a context with a reasonably short deadline (ie. not one
// that lasts throughout the lifetime of the server)
185
func (bs *Bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) {
186 187 188 189 190
	select {
	case <-bs.process.Closing():
		return nil, errors.New("bitswap is closed")
	default:
	}
191
	promise := bs.notifications.Subscribe(ctx, keys...)
192 193 194 195 196

	req := &blockRequest{
		keys: keys,
		ctx:  ctx,
	}
197
	select {
198
	case bs.batchRequests <- req:
199
		return promise, nil
200 201 202
	case <-ctx.Done():
		return nil, ctx.Err()
	}
Jeromy's avatar
Jeromy committed
203 204
}

205 206
// HasBlock announces the existance of a block to this bitswap service. The
// service will potentially notify its peers.
207
func (bs *Bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
208
	log.Event(ctx, "hasBlock", blk)
209 210 211 212 213
	select {
	case <-bs.process.Closing():
		return errors.New("bitswap is closed")
	default:
	}
214 215 216 217 218
	if err := bs.blockstore.Put(blk); err != nil {
		return err
	}
	bs.wantlist.Remove(blk.Key())
	bs.notifications.Publish(blk)
219 220 221 222 223 224
	select {
	case bs.newBlocks <- blk:
	case <-ctx.Done():
		return ctx.Err()
	}
	return nil
225 226
}

227
func (bs *Bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error {
228
	set := pset.New()
229
	wg := sync.WaitGroup{}
230

231 232 233 234 235 236 237
loop:
	for {
		select {
		case peerToQuery, ok := <-peers:
			if !ok {
				break loop
			}
238

239 240
			if !set.TryAdd(peerToQuery) { //Do once per peer
				continue
241
			}
242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262

			wg.Add(1)
			go func(p peer.ID) {
				defer wg.Done()
				if err := bs.send(ctx, p, m); err != nil {
					log.Debug(err) // TODO remove if too verbose
				}
			}(peerToQuery)
		case <-ctx.Done():
			return nil
		}
	}
	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()

	select {
	case <-done:
	case <-ctx.Done():
263 264 265
		// NB: we may be abandoning goroutines here before they complete
		// this shouldnt be an issue because they will complete soon anyways
		// we just don't want their being slow to impact bitswap transfer speeds
Jeromy's avatar
Jeromy committed
266 267 268 269
	}
	return nil
}

270
func (bs *Bitswap) sendWantlistToPeers(ctx context.Context, peers <-chan peer.ID) error {
Jeromy's avatar
Jeromy committed
271 272
	message := bsmsg.New()
	message.SetFull(true)
273 274
	for _, wanted := range bs.wantlist.Entries() {
		message.AddEntry(wanted.Key, wanted.Priority)
Jeromy's avatar
Jeromy committed
275
	}
276 277
	return bs.sendWantlistMsgToPeers(ctx, message, peers)
}
Jeromy's avatar
Jeromy committed
278

279
func (bs *Bitswap) sendWantlistToProviders(ctx context.Context, entries []wantlist.Entry) {
Jeromy's avatar
Jeromy committed
280

281 282 283 284 285
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// prepare a channel to hand off to sendWantlistToPeers
	sendToPeers := make(chan peer.ID)
Jeromy's avatar
Jeromy committed
286

287
	// Get providers for all entries in wantlist (could take a while)
Jeromy's avatar
Jeromy committed
288
	wg := sync.WaitGroup{}
Jeromy's avatar
Jeromy committed
289
	for _, e := range entries {
290
		wg.Add(1)
Jeromy's avatar
Jeromy committed
291
		go func(k u.Key) {
Jeromy's avatar
Jeromy committed
292
			defer wg.Done()
293

294 295
			child, cancel := context.WithTimeout(ctx, providerRequestTimeout)
			defer cancel()
296
			providers := bs.network.FindProvidersAsync(child, k, maxProvidersPerRequest)
297
			for prov := range providers {
298
				sendToPeers <- prov
Jeromy's avatar
Jeromy committed
299
			}
300
		}(e.Key)
Jeromy's avatar
Jeromy committed
301
	}
302 303 304 305 306 307 308 309

	go func() {
		wg.Wait() // make sure all our children do finish.
		close(sendToPeers)
	}()

	err := bs.sendWantlistToPeers(ctx, sendToPeers)
	if err != nil {
310
		log.Debugf("sendWantlistToPeers error: %s", err)
311
	}
Jeromy's avatar
Jeromy committed
312 313
}

314
// TODO(brian): handle errors
315
func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) (
316
	peer.ID, bsmsg.BitSwapMessage) {
317
	defer log.EventBegin(ctx, "receiveMessage", p, incoming).Done()
318

319
	if p == "" {
320
		log.Debug("Received message from nil peer!")
321
		// TODO propagate the error upward
322
		return "", nil
323 324
	}
	if incoming == nil {
325
		log.Debug("Got nil bitswap message!")
326
		// TODO propagate the error upward
327
		return "", nil
328
	}
329

Jeromy's avatar
Jeromy committed
330 331
	// This call records changes to wantlists, blocks received,
	// and number of bytes transfered.
332
	bs.engine.MessageReceived(p, incoming)
Jeromy's avatar
Jeromy committed
333 334
	// TODO: this is bad, and could be easily abused.
	// Should only track *useful* messages in ledger
335

Brian Tiger Chow's avatar
Brian Tiger Chow committed
336
	for _, block := range incoming.Blocks() {
337
		hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
338
		if err := bs.HasBlock(hasBlockCtx, block); err != nil {
339
			log.Debug(err)
Jeromy's avatar
Jeromy committed
340
		}
341
		cancel()
Brian Tiger Chow's avatar
Brian Tiger Chow committed
342
	}
343

344
	var keys []u.Key
Brian Tiger Chow's avatar
Brian Tiger Chow committed
345
	for _, block := range incoming.Blocks() {
346
		keys = append(keys, block.Key())
347
	}
348
	bs.cancelBlocks(ctx, keys)
349

Jeromy's avatar
Jeromy committed
350
	// TODO: consider changing this function to not return anything
351
	return "", nil
352 353
}

354
// Connected/Disconnected warns bitswap about peer connections
355
func (bs *Bitswap) PeerConnected(p peer.ID) {
356
	// TODO: add to clientWorker??
357 358 359 360 361
	peers := make(chan peer.ID, 1)
	peers <- p
	close(peers)
	err := bs.sendWantlistToPeers(context.TODO(), peers)
	if err != nil {
362
		log.Debugf("error sending wantlist: %s", err)
363
	}
364 365 366
}

// Connected/Disconnected warns bitswap about peer connections
367
func (bs *Bitswap) PeerDisconnected(p peer.ID) {
368
	bs.engine.PeerDisconnected(p)
369 370
}

371
func (bs *Bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
372 373 374
	if len(bkeys) < 1 {
		return
	}
Jeromy's avatar
Jeromy committed
375 376 377
	message := bsmsg.New()
	message.SetFull(false)
	for _, k := range bkeys {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
378
		message.Cancel(k)
Jeromy's avatar
Jeromy committed
379
	}
380
	for _, p := range bs.engine.Peers() {
Jeromy's avatar
Jeromy committed
381 382
		err := bs.send(ctx, p, message)
		if err != nil {
383
			log.Debugf("Error sending message: %s", err)
Jeromy's avatar
Jeromy committed
384 385 386 387
		}
	}
}

388
func (bs *Bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) {
389 390 391 392 393 394 395 396 397
	if len(bkeys) < 1 {
		return
	}

	message := bsmsg.New()
	message.SetFull(false)
	for i, k := range bkeys {
		message.AddEntry(k, kMaxPriority-i)
	}
398 399

	wg := sync.WaitGroup{}
400
	for _, p := range bs.engine.Peers() {
401
		wg.Add(1)
402
		go func(p peer.ID) {
403
			defer wg.Done()
404 405 406 407 408
			err := bs.send(ctx, p, message)
			if err != nil {
				log.Debugf("Error sending message: %s", err)
			}
		}(p)
409
	}
410 411 412 413 414 415 416 417
	done := make(chan struct{})
	go func() {
		wg.Wait()
		close(done)
	}()
	select {
	case <-done:
	case <-ctx.Done():
418 419 420
		// NB: we may be abandoning goroutines here before they complete
		// this shouldnt be an issue because they will complete soon anyways
		// we just don't want their being slow to impact bitswap transfer speeds
421
	}
422 423
}

424
func (bs *Bitswap) ReceiveError(err error) {
425
	log.Debugf("Bitswap ReceiveError: %s", err)
426 427
	// TODO log the network error
	// TODO bubble the network error up to the parent context/error logger
428 429
}

430 431
// send strives to ensure that accounting is always performed when a message is
// sent
432
func (bs *Bitswap) send(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage) error {
433
	defer log.EventBegin(ctx, "sendMessage", p, m).Done()
434
	if err := bs.network.SendMessage(ctx, p, m); err != nil {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
435
		return errors.Wrap(err)
436
	}
437
	return bs.engine.MessageSent(p, m)
438
}
439

440
func (bs *Bitswap) Close() error {
441
	return bs.process.Close()
442
}
443

444
func (bs *Bitswap) GetWantlist() []u.Key {
445 446 447 448 449 450
	var out []u.Key
	for _, e := range bs.wantlist.Entries() {
		out = append(out, e.Key)
	}
	return out
}