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

import (
6
	"context"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
7
	"errors"
dirkmc's avatar
dirkmc committed
8

9
	"sync"
Jeromy's avatar
Jeromy committed
10 11
	"time"

12
	delay "github.com/ipfs/go-ipfs-delay"
13

14 15 16 17 18 19 20 21 22 23 24 25
	bsbpm "github.com/ipfs/go-bitswap/internal/blockpresencemanager"
	decision "github.com/ipfs/go-bitswap/internal/decision"
	bsgetter "github.com/ipfs/go-bitswap/internal/getter"
	bsmq "github.com/ipfs/go-bitswap/internal/messagequeue"
	notifications "github.com/ipfs/go-bitswap/internal/notifications"
	bspm "github.com/ipfs/go-bitswap/internal/peermanager"
	bspqm "github.com/ipfs/go-bitswap/internal/providerquerymanager"
	bssession "github.com/ipfs/go-bitswap/internal/session"
	bssim "github.com/ipfs/go-bitswap/internal/sessioninterestmanager"
	bssm "github.com/ipfs/go-bitswap/internal/sessionmanager"
	bsspm "github.com/ipfs/go-bitswap/internal/sessionpeermanager"
	bswm "github.com/ipfs/go-bitswap/internal/wantmanager"
26 27
	bsmsg "github.com/ipfs/go-bitswap/message"
	bsnet "github.com/ipfs/go-bitswap/network"
Jeromy's avatar
Jeromy committed
28 29 30 31 32 33 34 35
	blocks "github.com/ipfs/go-block-format"
	cid "github.com/ipfs/go-cid"
	blockstore "github.com/ipfs/go-ipfs-blockstore"
	exchange "github.com/ipfs/go-ipfs-exchange-interface"
	logging "github.com/ipfs/go-log"
	metrics "github.com/ipfs/go-metrics-interface"
	process "github.com/jbenet/goprocess"
	procctx "github.com/jbenet/goprocess/context"
Raúl Kripalani's avatar
Raúl Kripalani committed
36
	peer "github.com/libp2p/go-libp2p-core/peer"
37 38
)

Jeromy's avatar
Jeromy committed
39
var log = logging.Logger("bitswap")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
40

41 42
var _ exchange.SessionExchange = (*Bitswap)(nil)

Brian Tiger Chow's avatar
Brian Tiger Chow committed
43
const (
44
	// these requests take at _least_ two minutes at the moment.
45 46
	provideTimeout         = time.Minute * 3
	defaultProvSearchDelay = time.Second
Jeromy's avatar
Jeromy committed
47
)
48

Jeromy's avatar
Jeromy committed
49
var (
50 51 52 53
	// HasBlockBufferSize is the buffer size of the channel for new blocks
	// that need to be provided. They should get pulled over by the
	// provideCollector even before they are actually provided.
	// TODO: Does this need to be this large givent that?
54 55
	HasBlockBufferSize    = 256
	provideKeysBufferSize = 2048
Steven Allen's avatar
Steven Allen committed
56
	provideWorkerMax      = 6
57 58 59

	// the 1<<18+15 is to observe old file chunks that are 1<<18 + 14 in size
	metricsBuckets = []float64{1 << 6, 1 << 10, 1 << 14, 1 << 18, 1<<18 + 15, 1 << 22}
Brian Tiger Chow's avatar
Brian Tiger Chow committed
60
)
Jeromy's avatar
Jeromy committed
61

62 63 64 65 66 67 68 69 70 71 72
// Option defines the functional option type that can be used to configure
// bitswap instances
type Option func(*Bitswap)

// ProvideEnabled is an option for enabling/disabling provide announcements
func ProvideEnabled(enabled bool) Option {
	return func(bs *Bitswap) {
		bs.provideEnabled = enabled
	}
}

73 74 75 76 77 78 79 80 81 82 83 84 85 86
// ProviderSearchDelay overwrites the global provider search delay
func ProviderSearchDelay(newProvSearchDelay time.Duration) Option {
	return func(bs *Bitswap) {
		bs.provSearchDelay = newProvSearchDelay
	}
}

// RebroadcastDelay overwrites the global provider rebroadcast delay
func RebroadcastDelay(newRebroadcastDelay delay.D) Option {
	return func(bs *Bitswap) {
		bs.rebroadcastDelay = newRebroadcastDelay
	}
}

87 88 89 90 91 92 93 94 95 96 97
// SetSendDontHaves indicates what to do when the engine receives a want-block
// for a block that is not in the blockstore. Either
// - Send a DONT_HAVE message
// - Simply don't respond
// This option is only used for testing.
func SetSendDontHaves(send bool) Option {
	return func(bs *Bitswap) {
		bs.engine.SetSendDontHaves(send)
	}
}

Brian Tiger Chow's avatar
Brian Tiger Chow committed
98 99
// New initializes a BitSwap instance that communicates over the provided
// BitSwapNetwork. This function registers the returned instance as the network
100
// delegate. Runs until context is cancelled or bitswap.Close is called.
Łukasz Magiera's avatar
Łukasz Magiera committed
101
func New(parent context.Context, network bsnet.BitSwapNetwork,
102
	bstore blockstore.Blockstore, options ...Option) exchange.Interface {
103

104 105
	// 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
106
	// coupled to the concerns of the ipfs daemon in this way.
107 108 109 110
	//
	// 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
111
	ctx, cancelFunc := context.WithCancel(parent)
112
	ctx = metrics.CtxSubScope(ctx, "bitswap")
113
	dupHist := metrics.NewCtx(ctx, "recv_dup_blocks_bytes", "Summary of duplicate"+
114
		" data blocks recived").Histogram(metricsBuckets)
115
	allHist := metrics.NewCtx(ctx, "recv_all_blocks_bytes", "Summary of all"+
116
		" data blocks recived").Histogram(metricsBuckets)
117

118 119 120
	sentHistogram := metrics.NewCtx(ctx, "sent_all_blocks_bytes", "Histogram of blocks sent by"+
		" this bitswap").Histogram(metricsBuckets)

121 122 123 124
	px := process.WithTeardown(func() error {
		return nil
	})

125 126 127 128 129 130 131 132
	var wm *bswm.WantManager
	// onDontHaveTimeout is called when a want-block is sent to a peer that
	// has an old version of Bitswap that doesn't support DONT_HAVE messages,
	// and no response is received within a timeout.
	onDontHaveTimeout := func(p peer.ID, dontHaves []cid.Cid) {
		// Simulate a DONT_HAVE message arriving to the WantManager
		wm.ReceiveFrom(ctx, p, nil, nil, dontHaves)
	}
133
	peerQueueFactory := func(ctx context.Context, p peer.ID) bspm.PeerQueue {
134
		return bsmq.New(ctx, p, network, onDontHaveTimeout)
135 136
	}

dirkmc's avatar
dirkmc committed
137 138 139
	sim := bssim.New()
	bpm := bsbpm.New()
	pm := bspm.New(ctx, peerQueueFactory, network.Self())
140
	wm = bswm.New(ctx, pm, sim, bpm)
141 142
	pqm := bspqm.New(ctx, network)

dirkmc's avatar
dirkmc committed
143 144 145 146
	sessionFactory := func(ctx context.Context, id uint64, spm bssession.SessionPeerManager,
		sim *bssim.SessionInterestManager,
		pm bssession.PeerManager,
		bpm *bsbpm.BlockPresenceManager,
147
		notif notifications.PubSub,
148
		provSearchDelay time.Duration,
dirkmc's avatar
dirkmc committed
149 150 151
		rebroadcastDelay delay.D,
		self peer.ID) bssm.Session {
		return bssession.New(ctx, id, wm, spm, sim, pm, bpm, notif, provSearchDelay, rebroadcastDelay, self)
152
	}
dirkmc's avatar
dirkmc committed
153
	sessionPeerManagerFactory := func(ctx context.Context, id uint64) bssession.SessionPeerManager {
154
		return bsspm.New(ctx, id, network.ConnectionManager(), pqm)
155
	}
156
	notif := notifications.New()
dirkmc's avatar
dirkmc committed
157 158 159
	sm := bssm.New(ctx, sessionFactory, sim, sessionPeerManagerFactory, bpm, pm, notif, network.Self())
	wm.SetSessionManager(sm)
	engine := decision.NewEngine(ctx, bstore, network.ConnectionManager(), network.Self())
160

161
	bs := &Bitswap{
162
		blockstore:       bstore,
163
		engine:           engine,
164 165 166 167 168
		network:          network,
		process:          px,
		newBlocks:        make(chan cid.Cid, HasBlockBufferSize),
		provideKeys:      make(chan cid.Cid, provideKeysBufferSize),
		wm:               wm,
dirkmc's avatar
dirkmc committed
169
		pm:               pm,
170
		pqm:              pqm,
dirkmc's avatar
dirkmc committed
171 172
		sm:               sm,
		sim:              sim,
173
		notif:            notif,
174 175 176 177 178 179 180
		counters:         new(counters),
		dupMetric:        dupHist,
		allMetric:        allHist,
		sentHistogram:    sentHistogram,
		provideEnabled:   true,
		provSearchDelay:  defaultProvSearchDelay,
		rebroadcastDelay: delay.Fixed(time.Minute),
181 182 183 184 185
	}

	// apply functional options before starting and running bitswap
	for _, option := range options {
		option(bs)
186
	}
187

188
	bs.pqm.Startup()
Brian Tiger Chow's avatar
Brian Tiger Chow committed
189
	network.SetDelegate(bs)
190

191
	// Start up bitswaps async worker routines
192
	bs.startWorkers(ctx, px)
193
	engine.StartWorkers(ctx, px)
194 195 196 197 198 199

	// bind the context and process.
	// do it over here to avoid closing before all setup is done.
	go func() {
		<-px.Closing() // process closes first
		cancelFunc()
200
		notif.Shutdown()
201 202 203
	}()
	procctx.CloseAfterContext(px, ctx) // parent cancelled first

204 205 206
	return bs
}

207 208
// Bitswap instances implement the bitswap protocol.
type Bitswap struct {
209
	// the wantlist tracks global wants for bitswap
210
	wm *bswm.WantManager
211

dirkmc's avatar
dirkmc committed
212 213
	pm *bspm.PeerManager

214 215 216
	// the provider query manager manages requests to find providers
	pqm *bspqm.ProviderQueryManager

217 218
	// the engine is the bit of logic that decides who to send which blocks to
	engine *decision.Engine
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
219

220 221
	// network delivers messages on behalf of the session
	network bsnet.BitSwapNetwork
222 223 224 225 226

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

227 228 229
	// manages channels of outgoing blocks for sessions
	notif notifications.PubSub

230 231 232
	// newBlocks is a channel for newly added blocks to be provided to the
	// network.  blocks pushed down this channel get buffered and fed to the
	// provideKeys channel later on to avoid too much network activity
233
	newBlocks chan cid.Cid
234
	// provideKeys directly feeds provide workers
235
	provideKeys chan cid.Cid
236

237 238 239
	process process.Process

	// Counters for various statistics
240 241
	counterLk sync.Mutex
	counters  *counters
242 243

	// Metrics interface metrics
244 245 246
	dupMetric     metrics.Histogram
	allMetric     metrics.Histogram
	sentHistogram metrics.Histogram
Jeromy's avatar
Jeromy committed
247

dirkmc's avatar
dirkmc committed
248
	// the SessionManager routes requests to interested sessions
249
	sm *bssm.SessionManager
250

dirkmc's avatar
dirkmc committed
251 252 253 254
	// the SessionInterestManager keeps track of which sessions are interested
	// in which CIDs
	sim *bssim.SessionInterestManager

255 256
	// whether or not to make provide announcements
	provideEnabled bool
257 258 259 260 261 262

	// how long to wait before looking for providers in a session
	provSearchDelay time.Duration

	// how often to rebroadcast providing requests to find more optimized providers
	rebroadcastDelay delay.D
263 264
}

265 266 267 268 269 270 271 272 273 274
type counters struct {
	blocksRecvd    uint64
	dupBlocksRecvd uint64
	dupDataRecvd   uint64
	blocksSent     uint64
	dataSent       uint64
	dataRecvd      uint64
	messagesRecvd  uint64
}

275
// GetBlock attempts to retrieve a particular block from peers within the
276
// deadline enforced by the context.
277
func (bs *Bitswap) GetBlock(parent context.Context, k cid.Cid) (blocks.Block, error) {
278
	return bsgetter.SyncGetBlock(parent, k, bs.GetBlocks)
279 280
}

281 282
// WantlistForPeer returns the currently understood list of blocks requested by a
// given peer.
283 284
func (bs *Bitswap) WantlistForPeer(p peer.ID) []cid.Cid {
	var out []cid.Cid
285
	for _, e := range bs.engine.WantlistForPeer(p) {
286
		out = append(out, e.Cid)
287 288 289 290
	}
	return out
}

291 292
// LedgerForPeer returns aggregated data about blocks swapped and communication
// with a given peer.
293 294 295 296
func (bs *Bitswap) LedgerForPeer(p peer.ID) *decision.Receipt {
	return bs.engine.LedgerForPeer(p)
}

297 298 299 300 301 302 303
// 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)
304
func (bs *Bitswap) GetBlocks(ctx context.Context, keys []cid.Cid) (<-chan blocks.Block, error) {
305
	session := bs.sm.NewSession(ctx, bs.provSearchDelay, bs.rebroadcastDelay)
306
	return session.GetBlocks(ctx, keys)
Jeromy's avatar
Jeromy committed
307 308
}

Łukasz Magiera's avatar
Łukasz Magiera committed
309
// HasBlock announces the existence of a block to this bitswap service. The
310
// service will potentially notify its peers.
311
func (bs *Bitswap) HasBlock(blk blocks.Block) error {
dirkmc's avatar
dirkmc committed
312
	return bs.receiveBlocksFrom(context.Background(), "", []blocks.Block{blk}, nil, nil)
313 314 315 316 317 318
}

// TODO: Some of this stuff really only needs to be done when adding a block
// from the user, not when receiving it from the network.
// In case you run `git blame` on this comment, I'll save you some time: ask
// @whyrusleeping, I don't know the answers you seek.
dirkmc's avatar
dirkmc committed
319
func (bs *Bitswap) receiveBlocksFrom(ctx context.Context, from peer.ID, blks []blocks.Block, haves []cid.Cid, dontHaves []cid.Cid) error {
320 321 322 323 324
	select {
	case <-bs.process.Closing():
		return errors.New("bitswap is closed")
	default:
	}
325

326 327 328 329
	wanted := blks

	// If blocks came from the network
	if from != "" {
dirkmc's avatar
dirkmc committed
330 331 332 333
		var notWanted []blocks.Block
		wanted, notWanted = bs.sim.SplitWantedUnwanted(blks)
		for _, b := range notWanted {
			log.Debugf("[recv] block not in wantlist; cid=%s, peer=%s", b.Cid(), from)
334 335 336 337
		}
	}

	// Put wanted blocks into blockstore
dirkmc's avatar
dirkmc committed
338 339 340 341 342 343
	if len(wanted) > 0 {
		err := bs.blockstore.PutMany(wanted)
		if err != nil {
			log.Errorf("Error writing %d blocks to datastore: %s", len(wanted), err)
			return err
		}
344
	}
345

346 347 348 349 350
	// NOTE: There exists the possiblity for a race condition here.  If a user
	// creates a node, then adds it to the dagservice while another goroutine
	// is waiting on a GetBlock for that object, they will receive a reference
	// to the same node. We should address this soon, but i'm not going to do
	// it now as it requires more thought and isnt causing immediate problems.
Jeromy's avatar
Jeromy committed
351

352 353 354 355 356 357
	allKs := make([]cid.Cid, 0, len(blks))
	for _, b := range blks {
		allKs = append(allKs, b.Cid())
	}

	// Send all block keys (including duplicates) to any sessions that want them.
358
	// (The duplicates are needed by sessions for accounting purposes)
dirkmc's avatar
dirkmc committed
359
	bs.wm.ReceiveFrom(ctx, from, allKs, haves, dontHaves)
360

dirkmc's avatar
dirkmc committed
361 362
	// Send wanted blocks to decision engine
	bs.engine.ReceiveFrom(from, wanted, haves)
363

364
	// Publish the block to any Bitswap clients that had requested blocks.
dirkmc's avatar
dirkmc committed
365
	// (the sessions use this pubsub mechanism to inform clients of incoming
366 367 368 369 370
	// blocks)
	for _, b := range wanted {
		bs.notif.Publish(b)
	}

371
	// If the reprovider is enabled, send wanted blocks to reprovider
372
	if bs.provideEnabled {
dirkmc's avatar
dirkmc committed
373
		for _, blk := range wanted {
374
			select {
dirkmc's avatar
dirkmc committed
375
			case bs.newBlocks <- blk.Cid():
376 377 378 379
				// send block off to be reprovided
			case <-bs.process.Closing():
				return bs.process.Close()
			}
380
		}
381
	}
382

383 384 385 386 387 388
	if from != "" {
		for _, b := range wanted {
			log.Event(ctx, "Bitswap.GetBlockRequest.End", b.Cid())
		}
	}

389
	return nil
390 391
}

392 393
// ReceiveMessage is called by the network interface when a new message is
// received.
394
func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) {
Steven Allen's avatar
Steven Allen committed
395 396 397
	bs.counterLk.Lock()
	bs.counters.messagesRecvd++
	bs.counterLk.Unlock()
Jeromy's avatar
Jeromy committed
398

Jeromy's avatar
Jeromy committed
399 400
	// This call records changes to wantlists, blocks received,
	// and number of bytes transfered.
401
	bs.engine.MessageReceived(ctx, p, incoming)
Jeromy's avatar
Jeromy committed
402 403
	// TODO: this is bad, and could be easily abused.
	// Should only track *useful* messages in ledger
404

405 406
	iblocks := incoming.Blocks()

dirkmc's avatar
dirkmc committed
407 408 409 410 411
	if len(iblocks) > 0 {
		bs.updateReceiveCounters(iblocks)
		for _, b := range iblocks {
			log.Debugf("[recv] block; cid=%s, peer=%s", b.Cid(), p)
		}
412
	}
413

dirkmc's avatar
dirkmc committed
414 415 416 417 418 419 420 421 422
	haves := incoming.Haves()
	dontHaves := incoming.DontHaves()
	if len(iblocks) > 0 || len(haves) > 0 || len(dontHaves) > 0 {
		// Process blocks
		err := bs.receiveBlocksFrom(ctx, p, iblocks, haves, dontHaves)
		if err != nil {
			log.Warningf("ReceiveMessage recvBlockFrom error: %s", err)
			return
		}
423
	}
424 425 426 427 428 429 430
}

func (bs *Bitswap) updateReceiveCounters(blocks []blocks.Block) {
	// Check which blocks are in the datastore
	// (Note: any errors from the blockstore are simply logged out in
	// blockstoreHas())
	blocksHas := bs.blockstoreHas(blocks)
431

432 433 434
	bs.counterLk.Lock()
	defer bs.counterLk.Unlock()

435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452
	// Do some accounting for each block
	for i, b := range blocks {
		has := blocksHas[i]

		blkLen := len(b.RawData())
		bs.allMetric.Observe(float64(blkLen))
		if has {
			bs.dupMetric.Observe(float64(blkLen))
		}

		c := bs.counters

		c.blocksRecvd++
		c.dataRecvd += uint64(blkLen)
		if has {
			c.dupBlocksRecvd++
			c.dupDataRecvd += uint64(blkLen)
		}
453 454 455
	}
}

456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478
func (bs *Bitswap) blockstoreHas(blks []blocks.Block) []bool {
	res := make([]bool, len(blks))

	wg := sync.WaitGroup{}
	for i, block := range blks {
		wg.Add(1)
		go func(i int, b blocks.Block) {
			defer wg.Done()

			has, err := bs.blockstore.Has(b.Cid())
			if err != nil {
				log.Infof("blockstore.Has error: %s", err)
				has = false
			}

			res[i] = has
		}(i, block)
	}
	wg.Wait()

	return res
}

479 480
// PeerConnected is called by the network interface
// when a peer initiates a new connection to bitswap.
481
func (bs *Bitswap) PeerConnected(p peer.ID) {
482
	bs.wm.Connected(p)
483
	bs.engine.PeerConnected(p)
484 485
}

486 487
// PeerDisconnected is called by the network interface when a peer
// closes a connection
488
func (bs *Bitswap) PeerDisconnected(p peer.ID) {
489
	bs.wm.Disconnected(p)
490
	bs.engine.PeerDisconnected(p)
491 492
}

493 494
// ReceiveError is called by the network interface when an error happens
// at the network layer. Currently just logs error.
495
func (bs *Bitswap) ReceiveError(err error) {
496
	log.Infof("Bitswap ReceiveError: %s", err)
497 498
	// TODO log the network error
	// TODO bubble the network error up to the parent context/error logger
499 500
}

501
// Close is called to shutdown Bitswap
502
func (bs *Bitswap) Close() error {
503
	return bs.process.Close()
504
}
505

506
// GetWantlist returns the current local wantlist.
507
func (bs *Bitswap) GetWantlist() []cid.Cid {
dirkmc's avatar
dirkmc committed
508 509 510 511 512 513
	return bs.pm.CurrentWants()
}

// GetWanthaves returns the current list of want-haves.
func (bs *Bitswap) GetWantHaves() []cid.Cid {
	return bs.pm.CurrentWantHaves()
514
}
515

516
// IsOnline is needed to match go-ipfs-exchange-interface
517 518 519
func (bs *Bitswap) IsOnline() bool {
	return true
}
520

521 522 523 524 525 526
// NewSession generates a new Bitswap session. You should use this, rather
// that calling Bitswap.GetBlocks, any time you intend to do several related
// block requests in a row. The session returned will have it's own GetBlocks
// method, but the session will use the fact that the requests are related to
// be more efficient in its requests to peers. If you are using a session
// from go-blockservice, it will create a bitswap session automatically.
527
func (bs *Bitswap) NewSession(ctx context.Context) exchange.Fetcher {
528
	return bs.sm.NewSession(ctx, bs.provSearchDelay, bs.rebroadcastDelay)
529
}