bitswap.go 11.2 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
	process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
11
	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
12
	blocks "github.com/jbenet/go-ipfs/blocks"
13
	blockstore "github.com/jbenet/go-ipfs/blocks/blockstore"
14
	exchange "github.com/jbenet/go-ipfs/exchange"
15
	decision "github.com/jbenet/go-ipfs/exchange/bitswap/decision"
16 17 18
	bsmsg "github.com/jbenet/go-ipfs/exchange/bitswap/message"
	bsnet "github.com/jbenet/go-ipfs/exchange/bitswap/network"
	notifications "github.com/jbenet/go-ipfs/exchange/bitswap/notifications"
19
	wantlist "github.com/jbenet/go-ipfs/exchange/bitswap/wantlist"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
20
	peer "github.com/jbenet/go-ipfs/p2p/peer"
21 22
	"github.com/jbenet/go-ipfs/thirdparty/delay"
	eventlog "github.com/jbenet/go-ipfs/thirdparty/eventlog"
23
	u "github.com/jbenet/go-ipfs/util"
Brian Tiger Chow's avatar
Brian Tiger Chow committed
24 25
	errors "github.com/jbenet/go-ipfs/util/debugerror"
	pset "github.com/jbenet/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 43 44

	hasBlockBufferSize = 256
	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,
91
		newBlocks:     make(chan *blocks.Block, hasBlockBufferSize),
92
	}
Brian Tiger Chow's avatar
Brian Tiger Chow committed
93
	network.SetDelegate(bs)
94

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

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

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

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

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

	notifications notifications.PubSub

115 116 117
	// 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
118
	batchRequests chan *blockRequest
Jeromy's avatar
Jeromy committed
119

120
	engine *decision.Engine
121

122
	wantlist *wantlist.ThreadSafe
123

124
	process process.Process
125 126

	newBlocks chan *blocks.Block
127 128
}

129 130 131 132 133
type blockRequest struct {
	keys []u.Key
	ctx  context.Context
}

134
// GetBlock attempts to retrieve a particular block from peers within the
135
// deadline enforced by the context.
Jeromy's avatar
Jeromy committed
136
func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, error) {
137

138 139 140 141
	// 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
142 143
	// when this context's cancel func is executed. This is difficult to
	// enforce. May this comment keep you safe.
144

145
	ctx, cancelFunc := context.WithCancel(parent)
146

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

	defer func() {
		cancelFunc()
	}()
153

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

	select {
160 161 162 163 164 165 166 167 168
	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
169
		return block, nil
170 171
	case <-parent.Done():
		return nil, parent.Err()
172 173 174
	}
}

175 176 177 178 179 180 181
// 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)
Jeromy's avatar
Jeromy committed
182
func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) {
183 184 185 186 187
	select {
	case <-bs.process.Closing():
		return nil, errors.New("bitswap is closed")
	default:
	}
188
	promise := bs.notifications.Subscribe(ctx, keys...)
189 190 191 192 193

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

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

224 225
func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error {
	set := pset.New()
226
	wg := sync.WaitGroup{}
227
	for peerToQuery := range peers {
228 229 230 231 232

		if !set.TryAdd(peerToQuery) { //Do once per peer
			continue
		}

233
		wg.Add(1)
234
		go func(p peer.ID) {
235
			defer wg.Done()
236
			if err := bs.send(ctx, p, m); err != nil {
237
				log.Debug(err) // TODO remove if too verbose
238
			}
239
		}(peerToQuery)
Jeromy's avatar
Jeromy committed
240
	}
241
	wg.Wait()
Jeromy's avatar
Jeromy committed
242 243 244
	return nil
}

245
func (bs *bitswap) sendWantlistToPeers(ctx context.Context, peers <-chan peer.ID) error {
Jeromy's avatar
Jeromy committed
246 247
	message := bsmsg.New()
	message.SetFull(true)
248 249
	for _, wanted := range bs.wantlist.Entries() {
		message.AddEntry(wanted.Key, wanted.Priority)
Jeromy's avatar
Jeromy committed
250
	}
251 252
	return bs.sendWantlistMsgToPeers(ctx, message, peers)
}
Jeromy's avatar
Jeromy committed
253

Brian Tiger Chow's avatar
Brian Tiger Chow committed
254
func (bs *bitswap) sendWantlistToProviders(ctx context.Context, entries []wantlist.Entry) {
Jeromy's avatar
Jeromy committed
255

256 257 258 259 260
	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
261

262
	// Get providers for all entries in wantlist (could take a while)
Jeromy's avatar
Jeromy committed
263
	wg := sync.WaitGroup{}
Jeromy's avatar
Jeromy committed
264
	for _, e := range entries {
265
		wg.Add(1)
Jeromy's avatar
Jeromy committed
266
		go func(k u.Key) {
Jeromy's avatar
Jeromy committed
267
			defer wg.Done()
268

269
			child, _ := context.WithTimeout(ctx, providerRequestTimeout)
270
			providers := bs.network.FindProvidersAsync(child, k, maxProvidersPerRequest)
271
			for prov := range providers {
272
				sendToPeers <- prov
Jeromy's avatar
Jeromy committed
273
			}
274
		}(e.Key)
Jeromy's avatar
Jeromy committed
275
	}
276 277 278 279 280 281 282 283

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

	err := bs.sendWantlistToPeers(ctx, sendToPeers)
	if err != nil {
284
		log.Debugf("sendWantlistToPeers error: %s", err)
285
	}
Jeromy's avatar
Jeromy committed
286 287
}

288
// TODO(brian): handle errors
289 290
func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) (
	peer.ID, bsmsg.BitSwapMessage) {
291
	defer log.EventBegin(ctx, "receiveMessage", p, incoming).Done()
292

293
	if p == "" {
294
		log.Debug("Received message from nil peer!")
295
		// TODO propagate the error upward
296
		return "", nil
297 298
	}
	if incoming == nil {
299
		log.Debug("Got nil bitswap message!")
300
		// TODO propagate the error upward
301
		return "", nil
302
	}
303

Jeromy's avatar
Jeromy committed
304 305
	// This call records changes to wantlists, blocks received,
	// and number of bytes transfered.
306
	bs.engine.MessageReceived(p, incoming)
Jeromy's avatar
Jeromy committed
307 308
	// TODO: this is bad, and could be easily abused.
	// Should only track *useful* messages in ledger
309

Brian Tiger Chow's avatar
Brian Tiger Chow committed
310
	for _, block := range incoming.Blocks() {
311 312
		hasBlockCtx, _ := context.WithTimeout(ctx, hasBlockTimeout)
		if err := bs.HasBlock(hasBlockCtx, block); err != nil {
313
			log.Debug(err)
Jeromy's avatar
Jeromy committed
314
		}
Brian Tiger Chow's avatar
Brian Tiger Chow committed
315
	}
316

317
	var keys []u.Key
Brian Tiger Chow's avatar
Brian Tiger Chow committed
318
	for _, block := range incoming.Blocks() {
319
		keys = append(keys, block.Key())
320
	}
321
	bs.cancelBlocks(ctx, keys)
322

Jeromy's avatar
Jeromy committed
323
	// TODO: consider changing this function to not return anything
324
	return "", nil
325 326
}

327 328 329
// Connected/Disconnected warns bitswap about peer connections
func (bs *bitswap) PeerConnected(p peer.ID) {
	// TODO: add to clientWorker??
330 331 332 333 334
	peers := make(chan peer.ID, 1)
	peers <- p
	close(peers)
	err := bs.sendWantlistToPeers(context.TODO(), peers)
	if err != nil {
335
		log.Debugf("error sending wantlist: %s", err)
336
	}
337 338 339
}

// Connected/Disconnected warns bitswap about peer connections
340 341
func (bs *bitswap) PeerDisconnected(p peer.ID) {
	bs.engine.PeerDisconnected(p)
342 343
}

Jeromy's avatar
Jeromy committed
344
func (bs *bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
345 346 347
	if len(bkeys) < 1 {
		return
	}
Jeromy's avatar
Jeromy committed
348 349 350
	message := bsmsg.New()
	message.SetFull(false)
	for _, k := range bkeys {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
351
		message.Cancel(k)
Jeromy's avatar
Jeromy committed
352
	}
353
	for _, p := range bs.engine.Peers() {
Jeromy's avatar
Jeromy committed
354 355
		err := bs.send(ctx, p, message)
		if err != nil {
356
			log.Debugf("Error sending message: %s", err)
Jeromy's avatar
Jeromy committed
357 358 359 360
		}
	}
}

361 362 363 364 365 366 367 368 369 370
func (bs *bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) {
	if len(bkeys) < 1 {
		return
	}

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

	wg := sync.WaitGroup{}
373
	for _, p := range bs.engine.Peers() {
374
		wg.Add(1)
375
		go func(p peer.ID) {
376
			defer wg.Done()
377 378 379 380 381
			err := bs.send(ctx, p, message)
			if err != nil {
				log.Debugf("Error sending message: %s", err)
			}
		}(p)
382
	}
383
	wg.Wait()
384 385
}

386
func (bs *bitswap) ReceiveError(err error) {
387
	log.Debugf("Bitswap ReceiveError: %s", err)
388 389
	// TODO log the network error
	// TODO bubble the network error up to the parent context/error logger
390 391
}

392 393
// send strives to ensure that accounting is always performed when a message is
// sent
394
func (bs *bitswap) send(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage) error {
395
	defer log.EventBegin(ctx, "sendMessage", p, m).Done()
396
	if err := bs.network.SendMessage(ctx, p, m); err != nil {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
397
		return errors.Wrap(err)
398
	}
399
	return bs.engine.MessageSent(p, m)
400
}
401 402

func (bs *bitswap) Close() error {
403
	return bs.process.Close()
404
}
405 406 407 408 409 410 411 412

func (bs *bitswap) GetWantlist() []u.Key {
	var out []u.Key
	for _, e := range bs.wantlist.Entries() {
		out = append(out, e.Key)
	}
	return out
}