bitswap.go 10.4 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
	"sync"
Jeromy's avatar
Jeromy committed
7 8
	"time"

9 10 11
	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"

	blocks "github.com/jbenet/go-ipfs/blocks"
12
	blockstore "github.com/jbenet/go-ipfs/blocks/blockstore"
13 14 15 16 17
	exchange "github.com/jbenet/go-ipfs/exchange"
	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"
	strategy "github.com/jbenet/go-ipfs/exchange/bitswap/strategy"
Jeromy's avatar
Jeromy committed
18
	wl "github.com/jbenet/go-ipfs/exchange/bitswap/wantlist"
19 20
	peer "github.com/jbenet/go-ipfs/peer"
	u "github.com/jbenet/go-ipfs/util"
Jeromy's avatar
Jeromy committed
21
	eventlog "github.com/jbenet/go-ipfs/util/eventlog"
Jeromy's avatar
Jeromy committed
22
	pset "github.com/jbenet/go-ipfs/util/peerset"
23 24
)

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

Jeromy's avatar
Jeromy committed
27
// Number of providers to request for sending a wantlist to
Jeromy's avatar
Jeromy committed
28 29
// TODO: if a 'non-nice' strategy is implemented, consider increasing this value
const maxProvidersPerRequest = 3
Jeromy's avatar
Jeromy committed
30

31 32 33
var providerRequestTimeout = time.Second * 10
var hasBlockTimeout = time.Second * 15
var rebroadcastDelay = time.Second * 10
34

35 36 37
// New initializes a BitSwap instance that communicates over the
// provided BitSwapNetwork. This function registers the returned instance as
// the network delegate.
38
// Runs until context is cancelled
39
func New(parent context.Context, p peer.Peer, network bsnet.BitSwapNetwork, routing bsnet.Routing,
40
	bstore blockstore.Blockstore, nice bool) exchange.Interface {
41

42 43
	ctx, cancelFunc := context.WithCancel(parent)

44 45
	notif := notifications.New()
	go func() {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
46
		<-ctx.Done()
Jeromy's avatar
Jeromy committed
47
		cancelFunc()
Brian Tiger Chow's avatar
Brian Tiger Chow committed
48
		notif.Shutdown()
49 50
	}()

51
	bs := &bitswap{
52
		blockstore:    bstore,
53
		cancelFunc:    cancelFunc,
54
		notifications: notif,
Jeromy's avatar
Jeromy committed
55
		ledgermanager: strategy.NewLedgerManager(bstore, ctx),
56
		routing:       routing,
Brian Tiger Chow's avatar
Brian Tiger Chow committed
57
		sender:        network,
58
		wantlist:      wl.New(),
59
		batchRequests: make(chan []u.Key, 32),
60
	}
Brian Tiger Chow's avatar
Brian Tiger Chow committed
61
	network.SetDelegate(bs)
62
	go bs.clientWorker(ctx)
Jeromy's avatar
Jeromy committed
63
	go bs.taskWorker(ctx)
64 65 66 67

	return bs
}

68 69 70 71
// bitswap instances implement the bitswap protocol.
type bitswap struct {

	// sender delivers messages on behalf of the session
72
	sender bsnet.BitSwapNetwork
73 74 75 76 77 78

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

	// routing interface for communication
79
	routing bsnet.Routing
80 81 82

	notifications notifications.PubSub

83 84 85 86
	// 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
	batchRequests chan []u.Key
Jeromy's avatar
Jeromy committed
87

88
	// strategy makes decisions about how to interact with partners.
Jeromy's avatar
Jeromy committed
89 90
	// TODO: strategy commented out until we have a use for it again
	//strategy strategy.Strategy
91

Jeromy's avatar
Jeromy committed
92
	ledgermanager *strategy.LedgerManager
93

Jeromy's avatar
Jeromy committed
94
	wantlist *wl.Wantlist
95 96 97

	// cancelFunc signals cancellation to the bitswap event loop
	cancelFunc func()
98 99
}

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

104 105 106 107
	// 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
108 109
	// when this context's cancel func is executed. This is difficult to
	// enforce. May this comment keep you safe.
110

111
	ctx, cancelFunc := context.WithCancel(parent)
112

Jeromy's avatar
Jeromy committed
113
	ctx = eventlog.ContextWithLoggable(ctx, eventlog.Uuid("GetBlockRequest"))
114
	log.Event(ctx, "GetBlockRequestBegin", &k)
115 116 117 118 119

	defer func() {
		cancelFunc()
		log.Event(ctx, "GetBlockRequestEnd", &k)
	}()
120

121
	promise, err := bs.GetBlocks(ctx, []u.Key{k})
122 123
	if err != nil {
		return nil, err
Jeromy's avatar
Jeromy committed
124
	}
125 126

	select {
127
	case block := <-promise:
Jeromy's avatar
Jeromy committed
128
		return block, nil
129 130
	case <-parent.Done():
		return nil, parent.Err()
131
	}
132

133 134
}

135 136 137 138 139 140 141
// 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
142
func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) {
143 144 145 146 147
	// TODO log the request

	promise := bs.notifications.Subscribe(ctx, keys...)
	select {
	case bs.batchRequests <- keys:
148
		return promise, nil
149 150 151
	case <-ctx.Done():
		return nil, ctx.Err()
	}
Jeromy's avatar
Jeromy committed
152 153
}

154 155 156 157 158 159 160 161 162 163 164 165
// 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 {
	if err := bs.blockstore.Put(blk); err != nil {
		return err
	}
	bs.wantlist.Remove(blk.Key())
	bs.notifications.Publish(blk)
	child, _ := context.WithTimeout(ctx, hasBlockTimeout)
	return bs.routing.Provide(child, blk.Key())
}

Jeromy's avatar
Jeromy committed
166
func (bs *bitswap) sendWantListTo(ctx context.Context, peers <-chan peer.Peer) error {
Jeromy's avatar
Jeromy committed
167 168 169
	if peers == nil {
		panic("Cant send wantlist to nil peerchan")
	}
Jeromy's avatar
Jeromy committed
170
	message := bsmsg.New()
Jeromy's avatar
Jeromy committed
171
	for _, wanted := range bs.wantlist.Entries() {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
172
		message.AddEntry(wanted.Value, wanted.Priority)
Jeromy's avatar
Jeromy committed
173
	}
174
	wg := sync.WaitGroup{}
Jeromy's avatar
Jeromy committed
175
	for peerToQuery := range peers {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
176
		log.Event(ctx, "PeerToQuery", peerToQuery)
177
		wg.Add(1)
Jeromy's avatar
Jeromy committed
178
		go func(p peer.Peer) {
179
			defer wg.Done()
Jeromy's avatar
Jeromy committed
180

Brian Tiger Chow's avatar
Brian Tiger Chow committed
181
			log.Event(ctx, "DialPeer", p)
Jeromy's avatar
Jeromy committed
182 183
			err := bs.sender.DialPeer(ctx, p)
			if err != nil {
184
				log.Errorf("Error sender.DialPeer(%s): %s", p, err)
Jeromy's avatar
Jeromy committed
185 186 187
				return
			}

188
			err = bs.sender.SendMessage(ctx, p, message)
Jeromy's avatar
Jeromy committed
189
			if err != nil {
190
				log.Errorf("Error sender.SendMessage(%s) = %s", p, err)
Jeromy's avatar
Jeromy committed
191 192 193 194 195
				return
			}
			// FIXME ensure accounting is handled correctly when
			// communication fails. May require slightly different API to
			// get better guarantees. May need shared sequence numbers.
Jeromy's avatar
Jeromy committed
196
			bs.ledgermanager.MessageSent(p, message)
Jeromy's avatar
Jeromy committed
197 198
		}(peerToQuery)
	}
199
	wg.Wait()
Jeromy's avatar
Jeromy committed
200 201 202
	return nil
}

Jeromy's avatar
Jeromy committed
203
func (bs *bitswap) sendWantlistToProviders(ctx context.Context, wantlist *wl.Wantlist) {
204 205 206
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

Jeromy's avatar
Jeromy committed
207 208 209
	message := bsmsg.New()
	message.SetFull(true)
	for _, e := range bs.wantlist.Entries() {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
210
		message.AddEntry(e.Value, e.Priority)
Jeromy's avatar
Jeromy committed
211 212 213 214
	}

	ps := pset.NewPeerSet()

215
	// Get providers for all entries in wantlist (could take a while)
Jeromy's avatar
Jeromy committed
216
	wg := sync.WaitGroup{}
Jeromy's avatar
Jeromy committed
217
	for _, e := range wantlist.Entries() {
218
		wg.Add(1)
Jeromy's avatar
Jeromy committed
219
		go func(k u.Key) {
Jeromy's avatar
Jeromy committed
220
			defer wg.Done()
221 222
			child, _ := context.WithTimeout(ctx, providerRequestTimeout)
			providers := bs.routing.FindProvidersAsync(child, k, maxProvidersPerRequest)
Jeromy's avatar
Jeromy committed
223

224
			for prov := range providers {
Jeromy's avatar
Jeromy committed
225
				if ps.TryAdd(prov) { //Do once per peer
Jeromy's avatar
Jeromy committed
226 227
					bs.send(ctx, prov, message)
				}
Jeromy's avatar
Jeromy committed
228
			}
Jeromy's avatar
Jeromy committed
229
		}(e.Value)
Jeromy's avatar
Jeromy committed
230
	}
Jeromy's avatar
Jeromy committed
231
	wg.Wait()
Jeromy's avatar
Jeromy committed
232 233
}

Jeromy's avatar
Jeromy committed
234
func (bs *bitswap) taskWorker(ctx context.Context) {
Jeromy's avatar
Jeromy committed
235 236 237 238
	for {
		select {
		case <-ctx.Done():
			return
Jeromy's avatar
Jeromy committed
239 240
		case task := <-bs.ledgermanager.GetTaskChan():
			block, err := bs.blockstore.Get(task.Key)
Jeromy's avatar
Jeromy committed
241
			if err != nil {
Jeromy's avatar
Jeromy committed
242 243
				log.Errorf("Expected to have block %s, but it was not found!", task.Key)
				continue
244
			}
Jeromy's avatar
Jeromy committed
245 246 247

			message := bsmsg.New()
			message.AddBlock(block)
Jeromy's avatar
Jeromy committed
248 249 250
			// TODO: maybe add keys from our wantlist?

			bs.send(ctx, task.Target, message)
Jeromy's avatar
Jeromy committed
251 252 253 254
		}
	}
}

255
// TODO ensure only one active request per key
256
func (bs *bitswap) clientWorker(parent context.Context) {
257 258

	ctx, cancel := context.WithCancel(parent)
Jeromy's avatar
Jeromy committed
259

260 261
	broadcastSignal := time.After(rebroadcastDelay)
	defer cancel()
Brian Tiger Chow's avatar
Brian Tiger Chow committed
262

Jeromy's avatar
Jeromy committed
263 264
	for {
		select {
265
		case <-broadcastSignal:
Jeromy's avatar
Jeromy committed
266
			// Resend unfulfilled wantlist keys
Jeromy's avatar
Jeromy committed
267
			bs.sendWantlistToProviders(ctx, bs.wantlist)
268
			broadcastSignal = time.After(rebroadcastDelay)
269
		case ks := <-bs.batchRequests:
270 271 272 273
			if len(ks) == 0 {
				log.Warning("Received batch request for zero blocks")
				continue
			}
274 275
			for i, k := range ks {
				bs.wantlist.Add(k, len(ks)-i)
276
			}
Jeromy's avatar
Jeromy committed
277 278 279 280 281 282 283
			// NB: send want list to providers for the first peer in this list.
			//		the assumption is made that the providers of the first key in
			//		the set are likely to have others as well.
			//		This currently holds true in most every situation, since when
			//		pinning a file, you store and provide all blocks associated with
			//		it. Later, this assumption may not hold as true if we implement
			//		newer bitswap strategies.
284 285
			child, _ := context.WithTimeout(ctx, providerRequestTimeout)
			providers := bs.routing.FindProvidersAsync(child, ks[0], maxProvidersPerRequest)
286 287 288 289

			err := bs.sendWantListTo(ctx, providers)
			if err != nil {
				log.Errorf("error sending wantlist: %s", err)
Jeromy's avatar
Jeromy committed
290
			}
291
		case <-parent.Done():
Jeromy's avatar
Jeromy committed
292 293 294 295 296
			return
		}
	}
}

297
// TODO(brian): handle errors
298 299
func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.Peer, incoming bsmsg.BitSwapMessage) (
	peer.Peer, bsmsg.BitSwapMessage) {
Jeromy's avatar
Jeromy committed
300
	log.Debugf("ReceiveMessage from %s", p)
301

302
	if p == nil {
303
		log.Error("Received message from nil peer!")
304 305
		// TODO propagate the error upward
		return nil, nil
306 307
	}
	if incoming == nil {
308
		log.Error("Got nil bitswap message!")
309 310
		// TODO propagate the error upward
		return nil, nil
311
	}
312

Jeromy's avatar
Jeromy committed
313 314
	// This call records changes to wantlists, blocks received,
	// and number of bytes transfered.
Jeromy's avatar
Jeromy committed
315
	bs.ledgermanager.MessageReceived(p, incoming)
Jeromy's avatar
Jeromy committed
316 317
	// TODO: this is bad, and could be easily abused.
	// Should only track *useful* messages in ledger
318

Jeromy's avatar
Jeromy committed
319
	var blkeys []u.Key
Brian Tiger Chow's avatar
Brian Tiger Chow committed
320
	for _, block := range incoming.Blocks() {
Jeromy's avatar
Jeromy committed
321
		blkeys = append(blkeys, block.Key())
Brian Tiger Chow's avatar
Brian Tiger Chow committed
322 323
		if err := bs.HasBlock(ctx, block); err != nil {
			log.Error(err)
Jeromy's avatar
Jeromy committed
324
		}
Brian Tiger Chow's avatar
Brian Tiger Chow committed
325
	}
Jeromy's avatar
Jeromy committed
326 327
	if len(blkeys) > 0 {
		bs.cancelBlocks(ctx, blkeys)
328
	}
329

Jeromy's avatar
Jeromy committed
330
	// TODO: consider changing this function to not return anything
331
	return nil, nil
332 333
}

Jeromy's avatar
Jeromy committed
334 335 336 337
func (bs *bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
	message := bsmsg.New()
	message.SetFull(false)
	for _, k := range bkeys {
Brian Tiger Chow's avatar
Brian Tiger Chow committed
338
		message.Cancel(k)
Jeromy's avatar
Jeromy committed
339
	}
Jeromy's avatar
Jeromy committed
340
	for _, p := range bs.ledgermanager.Peers() {
Jeromy's avatar
Jeromy committed
341 342 343 344 345 346 347
		err := bs.send(ctx, p, message)
		if err != nil {
			log.Errorf("Error sending message: %s", err)
		}
	}
}

348
func (bs *bitswap) ReceiveError(err error) {
349
	log.Errorf("Bitswap ReceiveError: %s", err)
350 351
	// TODO log the network error
	// TODO bubble the network error up to the parent context/error logger
352 353
}

354 355
// send strives to ensure that accounting is always performed when a message is
// sent
356 357 358 359
func (bs *bitswap) send(ctx context.Context, p peer.Peer, m bsmsg.BitSwapMessage) error {
	if err := bs.sender.SendMessage(ctx, p, m); err != nil {
		return err
	}
Jeromy's avatar
Jeromy committed
360
	return bs.ledgermanager.MessageSent(p, m)
361
}
362 363 364 365 366

func (bs *bitswap) Close() error {
	bs.cancelFunc()
	return nil // to conform to Closer interface
}