bitswap.go 12.3 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"
8
	"math"
9
	"sync"
Jeromy's avatar
Jeromy committed
10
	"sync/atomic"
Jeromy's avatar
Jeromy committed
11 12
	"time"

13 14 15 16 17 18
	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"
Jeromy's avatar
Jeromy committed
19
	flags "github.com/ipfs/go-ipfs/flags"
20
	"github.com/ipfs/go-ipfs/thirdparty/delay"
Jeromy's avatar
Jeromy committed
21

22
	metrics "gx/ipfs/QmRg1gKTHzc3CZXSKzem8aR4E3TubFhbgXwfVuWnSK5CC5/go-metrics-interface"
23 24
	process "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess"
	procctx "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess/context"
Jeromy's avatar
Jeromy committed
25
	logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
Jeromy's avatar
Jeromy committed
26
	blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
27
	cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
28
	peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
29 30
)

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

Brian Tiger Chow's avatar
Brian Tiger Chow committed
33
const (
Brian Tiger Chow's avatar
Brian Tiger Chow committed
34 35 36
	// 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
37 38 39
	// TODO: if a 'non-nice' strategy is implemented, consider increasing this value
	maxProvidersPerRequest = 3
	providerRequestTimeout = time.Second * 10
40 41
	provideTimeout         = time.Second * 15
	sizeBatchRequestChan   = 32
42 43
	// kMaxPriority is the max priority as defined by the bitswap protocol
	kMaxPriority = math.MaxInt32
Jeromy's avatar
Jeromy committed
44
)
45

Jeromy's avatar
Jeromy committed
46
var (
47 48 49
	HasBlockBufferSize    = 256
	provideKeysBufferSize = 2048
	provideWorkerMax      = 512
50 51 52

	// 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
53
)
Jeromy's avatar
Jeromy committed
54

Jeromy's avatar
Jeromy committed
55 56 57 58 59 60 61 62
func init() {
	if flags.LowMemMode {
		HasBlockBufferSize = 64
		provideKeysBufferSize = 512
		provideWorkerMax = 16
	}
}

63
var rebroadcastDelay = delay.Fixed(time.Minute)
64

Brian Tiger Chow's avatar
Brian Tiger Chow committed
65 66 67 68
// 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.
69
func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
70
	bstore blockstore.Blockstore, nice bool) exchange.Interface {
71

72 73
	// 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
74
	// coupled to the concerns of the ipfs daemon in this way.
75 76 77 78
	//
	// 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
79
	ctx, cancelFunc := context.WithCancel(parent)
80
	ctx = metrics.CtxSubScope(ctx, "bitswap")
81
	dupHist := metrics.NewCtx(ctx, "recv_dup_blocks_bytes", "Summary of duplicate"+
82
		" data blocks recived").Histogram(metricsBuckets)
83
	allHist := metrics.NewCtx(ctx, "recv_all_blocks_bytes", "Summary of all"+
84
		" data blocks recived").Histogram(metricsBuckets)
85

86
	notif := notifications.New()
87 88 89 90 91
	px := process.WithTeardown(func() error {
		notif.Shutdown()
		return nil
	})

92
	bs := &Bitswap{
93
		blockstore:    bstore,
94
		notifications: notif,
95
		engine:        decision.NewEngine(ctx, bstore), // TODO close the engine with Close() method
96
		network:       network,
97
		findKeys:      make(chan *blockRequest, sizeBatchRequestChan),
98
		process:       px,
99 100
		newBlocks:     make(chan *cid.Cid, HasBlockBufferSize),
		provideKeys:   make(chan *cid.Cid, provideKeysBufferSize),
101
		wm:            NewWantManager(ctx, network),
102
		counters:      new(counters),
103 104 105

		dupMetric: dupHist,
		allMetric: allHist,
106
	}
107
	go bs.wm.Run()
Brian Tiger Chow's avatar
Brian Tiger Chow committed
108
	network.SetDelegate(bs)
109

110 111
	// Start up bitswaps async worker routines
	bs.startWorkers(px, ctx)
112 113 114 115 116 117 118 119 120

	// 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()
	}()
	procctx.CloseAfterContext(px, ctx) // parent cancelled first

121 122 123
	return bs
}

124 125
// Bitswap instances implement the bitswap protocol.
type Bitswap struct {
126 127 128
	// the peermanager manages sending messages to peers in a way that
	// wont block bitswap operation
	wm *WantManager
129

130 131
	// 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
132

133 134
	// network delivers messages on behalf of the session
	network bsnet.BitSwapNetwork
135 136 137 138 139

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

140 141
	// notifications engine for receiving new blocks and routing them to the
	// appropriate user requests
142 143
	notifications notifications.PubSub

144
	// findKeys sends keys to a worker to find and connect to providers for them
145
	findKeys chan *blockRequest
146 147 148
	// 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
149
	newBlocks chan *cid.Cid
150
	// provideKeys directly feeds provide workers
151
	provideKeys chan *cid.Cid
152

153 154 155
	process process.Process

	// Counters for various statistics
156 157
	counterLk sync.Mutex
	counters  *counters
158 159 160 161

	// Metrics interface metrics
	dupMetric metrics.Histogram
	allMetric metrics.Histogram
Jeromy's avatar
Jeromy committed
162 163 164 165

	// Sessions
	sessions []*Session
	sessLk   sync.Mutex
Jeromy's avatar
Jeromy committed
166 167 168

	sessID   uint64
	sessIDLk sync.Mutex
169 170
}

171 172 173 174 175 176 177 178 179 180
type counters struct {
	blocksRecvd    uint64
	dupBlocksRecvd uint64
	dupDataRecvd   uint64
	blocksSent     uint64
	dataSent       uint64
	dataRecvd      uint64
	messagesRecvd  uint64
}

181
type blockRequest struct {
182
	Cid *cid.Cid
183
	Ctx context.Context
184 185
}

186
// GetBlock attempts to retrieve a particular block from peers within the
187
// deadline enforced by the context.
188
func (bs *Bitswap) GetBlock(parent context.Context, k *cid.Cid) (blocks.Block, error) {
Jeromy's avatar
Jeromy committed
189
	return getBlock(parent, k, bs.GetBlocks)
190 191
}

192 193
func (bs *Bitswap) WantlistForPeer(p peer.ID) []*cid.Cid {
	var out []*cid.Cid
194
	for _, e := range bs.engine.WantlistForPeer(p) {
195
		out = append(out, e.Cid)
196 197 198 199
	}
	return out
}

200 201 202 203
func (bs *Bitswap) LedgerForPeer(p peer.ID) *decision.Receipt {
	return bs.engine.LedgerForPeer(p)
}

204 205 206 207 208 209 210
// 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)
211
func (bs *Bitswap) GetBlocks(ctx context.Context, keys []*cid.Cid) (<-chan blocks.Block, error) {
212
	if len(keys) == 0 {
213
		out := make(chan blocks.Block)
214 215 216 217
		close(out)
		return out, nil
	}

218 219 220 221 222
	select {
	case <-bs.process.Closing():
		return nil, errors.New("bitswap is closed")
	default:
	}
223
	promise := bs.notifications.Subscribe(ctx, keys...)
224

225
	for _, k := range keys {
226
		log.Event(ctx, "Bitswap.GetBlockRequest.Start", k)
227 228
	}

Jeromy's avatar
Jeromy committed
229 230 231
	mses := bs.getNextSessionID()

	bs.wm.WantBlocks(ctx, keys, nil, mses)
232

233 234 235
	// NB: Optimization. Assumes that providers of key[0] are likely to
	// be able to provide for all keys. This currently holds true in most
	// every situation. Later, this assumption may not hold as true.
236
	req := &blockRequest{
237
		Cid: keys[0],
238
		Ctx: ctx,
239
	}
240

241
	remaining := cid.NewSet()
242
	for _, k := range keys {
243
		remaining.Add(k)
244 245 246 247 248 249 250 251
	}

	out := make(chan blocks.Block)
	go func() {
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()
		defer close(out)
		defer func() {
252
			// can't just defer this call on its own, arguments are resolved *when* the defer is created
Jeromy's avatar
Jeromy committed
253
			bs.CancelWants(remaining.Keys(), mses)
254 255 256 257 258 259 260 261
		}()
		for {
			select {
			case blk, ok := <-promise:
				if !ok {
					return
				}

Jeromy's avatar
Jeromy committed
262
				bs.CancelWants([]*cid.Cid{blk.Cid()}, mses)
263
				remaining.Remove(blk.Cid())
264 265 266 267 268 269 270 271 272 273 274
				select {
				case out <- blk:
				case <-ctx.Done():
					return
				}
			case <-ctx.Done():
				return
			}
		}
	}()

275
	select {
Jeromy's avatar
Jeromy committed
276
	case bs.findKeys <- req:
277
		return out, nil
278 279 280
	case <-ctx.Done():
		return nil, ctx.Err()
	}
Jeromy's avatar
Jeromy committed
281 282
}

Jeromy's avatar
Jeromy committed
283 284 285 286 287 288 289
func (bs *Bitswap) getNextSessionID() uint64 {
	bs.sessIDLk.Lock()
	defer bs.sessIDLk.Unlock()
	bs.sessID++
	return bs.sessID
}

290
// CancelWant removes a given key from the wantlist
Jeromy's avatar
Jeromy committed
291
func (bs *Bitswap) CancelWants(cids []*cid.Cid, ses uint64) {
292 293 294
	if len(cids) == 0 {
		return
	}
Jeromy's avatar
Jeromy committed
295
	bs.wm.CancelWants(context.Background(), cids, nil, ses)
296 297
}

298 299
// HasBlock announces the existance of a block to this bitswap service. The
// service will potentially notify its peers.
300
func (bs *Bitswap) HasBlock(blk blocks.Block) error {
301 302 303 304 305
	select {
	case <-bs.process.Closing():
		return errors.New("bitswap is closed")
	default:
	}
306

307
	err := bs.blockstore.Put(blk)
308 309
	if err != nil {
		log.Errorf("Error writing block to datastore: %s", err)
310 311
		return err
	}
312

313 314 315 316 317
	// 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
318 319
	bs.notifications.Publish(blk)

320 321
	bs.engine.AddBlock(blk)

322
	select {
323
	case bs.newBlocks <- blk.Cid():
324
		// send block off to be reprovided
325 326
	case <-bs.process.Closing():
		return bs.process.Close()
327 328
	}
	return nil
329 330
}

Jeromy's avatar
Jeromy committed
331
// SessionsForBlock returns a slice of all sessions that may be interested in the given cid
Jeromy's avatar
Jeromy committed
332 333 334 335 336 337
func (bs *Bitswap) SessionsForBlock(c *cid.Cid) []*Session {
	bs.sessLk.Lock()
	defer bs.sessLk.Unlock()

	var out []*Session
	for _, s := range bs.sessions {
Jeromy's avatar
Jeromy committed
338
		if s.interestedIn(c) {
Jeromy's avatar
Jeromy committed
339 340 341 342 343 344
			out = append(out, s)
		}
	}
	return out
}

345
func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) {
346
	atomic.AddUint64(&bs.counters.messagesRecvd, 1)
Jeromy's avatar
Jeromy committed
347

Jeromy's avatar
Jeromy committed
348 349
	// This call records changes to wantlists, blocks received,
	// and number of bytes transfered.
350
	bs.engine.MessageReceived(p, incoming)
Jeromy's avatar
Jeromy committed
351 352
	// TODO: this is bad, and could be easily abused.
	// Should only track *useful* messages in ledger
353

354 355 356
	iblocks := incoming.Blocks()

	if len(iblocks) == 0 {
357 358 359 360
		return
	}

	// quickly send out cancels, reduces chances of duplicate block receives
361
	var keys []*cid.Cid
362
	for _, block := range iblocks {
363
		if _, found := bs.wm.wl.Contains(block.Cid()); !found {
364
			log.Infof("received un-asked-for %s from %s", block, p)
365 366
			continue
		}
367
		keys = append(keys, block.Cid())
Jeromy's avatar
Jeromy committed
368
	}
Jeromy's avatar
Jeromy committed
369

Jeromy's avatar
Jeromy committed
370 371 372
	wg := sync.WaitGroup{}
	for _, block := range iblocks {
		wg.Add(1)
373
		go func(b blocks.Block) {
Jeromy's avatar
Jeromy committed
374
			defer wg.Done()
375

376
			bs.updateReceiveCounters(b)
377

378 379
			k := b.Cid()
			log.Event(ctx, "Bitswap.GetBlockRequest.End", k)
380

Jeromy's avatar
Jeromy committed
381
			for _, ses := range bs.SessionsForBlock(k) {
Jeromy's avatar
Jeromy committed
382 383
				ses.receiveBlockFrom(p, b)
				bs.CancelWants([]*cid.Cid{k}, ses.id)
Jeromy's avatar
Jeromy committed
384
			}
385
			log.Debugf("got block %s from %s", b, p)
386
			if err := bs.HasBlock(b); err != nil {
Jeromy's avatar
Jeromy committed
387 388 389
				log.Warningf("ReceiveMessage HasBlock error: %s", err)
			}
		}(block)
390
	}
Jeromy's avatar
Jeromy committed
391
	wg.Wait()
392 393
}

394 395
var ErrAlreadyHaveBlock = errors.New("already have block")

396
func (bs *Bitswap) updateReceiveCounters(b blocks.Block) {
397
	blkLen := len(b.RawData())
398
	has, err := bs.blockstore.Has(b.Cid())
399 400
	if err != nil {
		log.Infof("blockstore.Has error: %s", err)
401
		return
402
	}
403 404 405

	bs.allMetric.Observe(float64(blkLen))
	if has {
406
		bs.dupMetric.Observe(float64(blkLen))
407 408
	}

409 410
	bs.counterLk.Lock()
	defer bs.counterLk.Unlock()
411
	c := bs.counters
412

413 414
	c.blocksRecvd++
	c.dataRecvd += uint64(len(b.RawData()))
415
	if has {
416 417
		c.dupBlocksRecvd++
		c.dupDataRecvd += uint64(blkLen)
418 419 420
	}
}

421
// Connected/Disconnected warns bitswap about peer connections
422
func (bs *Bitswap) PeerConnected(p peer.ID) {
423
	bs.wm.Connected(p)
424
	bs.engine.PeerConnected(p)
425 426 427
}

// Connected/Disconnected warns bitswap about peer connections
428
func (bs *Bitswap) PeerDisconnected(p peer.ID) {
429
	bs.wm.Disconnected(p)
430
	bs.engine.PeerDisconnected(p)
431 432
}

433
func (bs *Bitswap) ReceiveError(err error) {
434
	log.Infof("Bitswap ReceiveError: %s", err)
435 436
	// TODO log the network error
	// TODO bubble the network error up to the parent context/error logger
437 438
}

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

443 444
func (bs *Bitswap) GetWantlist() []*cid.Cid {
	var out []*cid.Cid
445
	for _, e := range bs.wm.wl.Entries() {
446
		out = append(out, e.Cid)
447 448 449
	}
	return out
}
450 451 452 453

func (bs *Bitswap) IsOnline() bool {
	return true
}