track broadcasted wantlist entries
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
May 19, 2017 at 21:04 UTC
579fd46488576321252a3a2c96340a16ea2d62c6
10 files changed
+146
-38
blockservice/blockservice.go
+3
@@ -251,15 +251,18 @@ func (s *blockService) Close() error {
251
return s.exchange.Close()
252
}
253
254
+// Session is a helper type to provide higher level access to bitswap sessions
255
type Session struct {
256
bs blockstore.Blockstore
257
ses exchange.Fetcher
258
}
259
260
+// GetBlock gets a block in the context of a request session
261
func (s *Session) GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error) {
262
return getBlock(ctx, c, s.bs, s.ses)
263
}
264
265
+// GetBlocks gets blocks in the context of a request session
266
func (s *Session) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block {
267
return getBlocks(ctx, ks, s.bs, s.ses)
268
}
core/commands/bitswap.go
+5
@@ -111,6 +111,11 @@ Print out all blocks currently on the bitswap wantlist for the local peer.`,
111
res.SetError(err, cmds.ErrNormal)
112
return
113
}
114
+ if pid == nd.Identity {
115
+ res.SetOutput(&KeyList{bs.GetWantlist()})
116
+ return
117
+ }
118
+
119
res.SetOutput(&KeyList{bs.WantlistForPeer(pid)})
120
} else {
121
res.SetOutput(&KeyList{bs.GetWantlist()})
exchange/bitswap/bitswap.go
+1
@@ -323,6 +323,7 @@ func (bs *Bitswap) HasBlock(blk blocks.Block) error {
323
return nil
324
}
325
326
+// SessionsForBlock returns a slice of all sessions that may be interested in the given cid
327
func (bs *Bitswap) SessionsForBlock(c *cid.Cid) []*Session {
328
bs.sessLk.Lock()
329
defer bs.sessLk.Unlock()
exchange/bitswap/bitswap_test.go
+1
-1
@@ -332,7 +332,7 @@ func TestBasicBitswap(t *testing.T) {
332
t.Fatal(err)
333
}
334
335
- time.Sleep(time.Millisecond * 20)
335
+ time.Sleep(time.Millisecond * 25)
336
wl := instances[2].Exchange.WantlistForPeer(instances[1].Peer)
337
if len(wl) != 0 {
338
t.Fatal("should have no items in other peers wantlist")
exchange/bitswap/session.go
+41
-13
@@ -25,14 +25,14 @@ type Session struct {
25
activePeers map[peer.ID]struct{}
26
activePeersArr []peer.ID
27
28
- bs *Bitswap
29
- incoming chan blkRecv
30
- newReqs chan []*cid.Cid
31
- cancelKeys chan []*cid.Cid
28
+ bs *Bitswap
29
+ incoming chan blkRecv
30
+ newReqs chan []*cid.Cid
31
+ cancelKeys chan []*cid.Cid
32
+ interestReqs chan interestReq
33
34
interest *lru.Cache
35
liveWants map[string]time.Time
35
- liveCnt int
36
37
tick *time.Timer
38
baseTickDelay time.Duration
@@ -55,6 +55,7 @@ func (bs *Bitswap) NewSession(ctx context.Context) *Session {
55
liveWants: make(map[string]time.Time),
56
newReqs: make(chan []*cid.Cid),
57
cancelKeys: make(chan []*cid.Cid),
58
+ interestReqs: make(chan interestReq),
59
ctx: ctx,
60
bs: bs,
61
incoming: make(chan blkRecv),
@@ -85,8 +86,29 @@ func (s *Session) receiveBlockFrom(from peer.ID, blk blocks.Block) {
86
s.incoming <- blkRecv{from: from, blk: blk}
87
}
88
89
+type interestReq struct {
90
+ c *cid.Cid
91
+ resp chan bool
92
+}
93
+
94
+// TODO: PERF: this is using a channel to guard a map access against race
95
+// conditions. This is definitely much slower than a mutex, though its unclear
96
+// if it will actually induce any noticeable slowness. This is implemented this
97
+// way to avoid adding a more complex set of mutexes around the liveWants map.
98
+// note that in the average case (where this session *is* interested in the
99
+// block we received) this function will not be called, as the cid will likely
100
+// still be in the interest cache.
101
+func (s *Session) isLiveWant(c *cid.Cid) bool {
102
+ resp := make(chan bool)
103
+ s.interestReqs <- interestReq{
104
+ c: c,
105
+ resp: resp,
106
+ }
107
+ return <-resp
108
+}
109
+
110
func (s *Session) interestedIn(c *cid.Cid) bool {
89
- return s.interest.Contains(c.KeyString())
111
+ return s.interest.Contains(c.KeyString()) || s.isLiveWant(c)
112
}
113
114
const provSearchDelay = time.Second * 10
@@ -124,12 +146,11 @@ func (s *Session) run(ctx context.Context) {
146
for _, k := range keys {
147
s.interest.Add(k.KeyString(), nil)
148
}
127
- if s.liveCnt < activeWantsLimit {
128
- toadd := activeWantsLimit - s.liveCnt
149
+ if len(s.liveWants) < activeWantsLimit {
150
+ toadd := activeWantsLimit - len(s.liveWants)
151
if toadd > len(keys) {
152
toadd = len(keys)
153
}
132
- s.liveCnt += toadd
154
155
now := keys[:toadd]
156
keys = keys[toadd:]
@@ -152,15 +173,23 @@ func (s *Session) run(ctx context.Context) {
173
s.bs.wm.WantBlocks(ctx, live, nil, s.id)
174
175
if len(live) > 0 {
155
- go func() {
156
- for p := range s.bs.network.FindProvidersAsync(ctx, live[0], 10) {
176
+ go func(k *cid.Cid) {
177
+ // TODO: have a task queue setup for this to:
178
+ // - rate limit
179
+ // - manage timeouts
180
+ // - ensure two 'findprovs' calls for the same block don't run concurrently
181
+ // - share peers between sessions based on interest set
182
+ for p := range s.bs.network.FindProvidersAsync(ctx, k, 10) {
183
newpeers <- p
184
}
159
- }()
185
+ }(live[0])
186
}
187
s.resetTick()
188
case p := <-newpeers:
189
s.addActivePeer(p)
190
+ case lwchk := <-s.interestReqs:
191
+ _, ok := s.liveWants[lwchk.c.KeyString()]
192
+ lwchk.resp <- ok
193
case <-ctx.Done():
194
return
195
}
@@ -170,7 +199,6 @@ func (s *Session) run(ctx context.Context) {
199
func (s *Session) receiveBlock(ctx context.Context, blk blocks.Block) {
200
ks := blk.Cid().KeyString()
201
if _, ok := s.liveWants[ks]; ok {
173
- s.liveCnt--
202
tval := s.liveWants[ks]
203
s.latTotal += time.Since(tval)
204
s.fetchcnt++
exchange/bitswap/session_test.go
+52
@@ -150,3 +150,55 @@ func TestSessionSplitFetch(t *testing.T) {
150
}
151
}
152
}
153
+
154
+func TestInterestCacheOverflow(t *testing.T) {
155
+ ctx, cancel := context.WithCancel(context.Background())
156
+ defer cancel()
157
+
158
+ vnet := getVirtualNetwork()
159
+ sesgen := NewTestSessionGenerator(vnet)
160
+ defer sesgen.Close()
161
+ bgen := blocksutil.NewBlockGenerator()
162
+
163
+ blks := bgen.Blocks(2049)
164
+ inst := sesgen.Instances(2)
165
+
166
+ a := inst[0]
167
+ b := inst[1]
168
+
169
+ ses := a.Exchange.NewSession(ctx)
170
+ zeroch, err := ses.GetBlocks(ctx, []*cid.Cid{blks[0].Cid()})
171
+ if err != nil {
172
+ t.Fatal(err)
173
+ }
174
+
175
+ var restcids []*cid.Cid
176
+ for _, blk := range blks[1:] {
177
+ restcids = append(restcids, blk.Cid())
178
+ }
179
+
180
+ restch, err := ses.GetBlocks(ctx, restcids)
181
+ if err != nil {
182
+ t.Fatal(err)
183
+ }
184
+
185
+ // wait to ensure that all the above cids were added to the sessions cache
186
+ time.Sleep(time.Millisecond * 50)
187
+
188
+ if err := b.Exchange.HasBlock(blks[0]); err != nil {
189
+ t.Fatal(err)
190
+ }
191
+
192
+ select {
193
+ case blk, ok := <-zeroch:
194
+ if ok && blk.Cid().Equals(blks[0].Cid()) {
195
+ // success!
196
+ } else {
197
+ t.Fatal("failed to get the block")
198
+ }
199
+ case <-restch:
200
+ t.Fatal("should not get anything on restch")
201
+ case <-time.After(time.Second * 5):
202
+ t.Fatal("timed out waiting for block")
203
+ }
204
+}
exchange/bitswap/wantlist/wantlist.go
+14
@@ -53,6 +53,14 @@ func New() *Wantlist {
53
}
54
}
55
56
+// Add adds the given cid to the wantlist with the specified priority, governed
57
+// by the session ID 'ses'. if a cid is added under multiple session IDs, then
58
+// it must be removed by each of those sessions before it is no longer 'in the
59
+// wantlist'. Calls to Add are idempotent given the same arguments. Subsequent
60
+// calls with different values for priority will not update the priority
61
+// TODO: think through priority changes here
62
+// Add returns true if the cid did not exist in the wantlist before this call
63
+// (even if it was under a different session)
64
func (w *ThreadSafe) Add(c *cid.Cid, priority int, ses uint64) bool {
65
w.lk.Lock()
66
defer w.lk.Unlock()
@@ -84,6 +92,10 @@ func (w *ThreadSafe) AddEntry(e *Entry, ses uint64) bool {
92
return true
93
}
94
95
+// Remove removes the given cid from being tracked by the given session.
96
+// 'true' is returned if this call to Remove removed the final session ID
97
+// tracking the cid. (meaning true will be returned iff this call caused the
98
+// value of 'Contains(c)' to change from true to false)
99
func (w *ThreadSafe) Remove(c *cid.Cid, ses uint64) bool {
100
w.lk.Lock()
101
defer w.lk.Unlock()
@@ -101,6 +113,8 @@ func (w *ThreadSafe) Remove(c *cid.Cid, ses uint64) bool {
113
return false
114
}
115
116
+// Contains returns true if the given cid is in the wantlist tracked by one or
117
+// more sessions
118
func (w *ThreadSafe) Contains(k *cid.Cid) (*Entry, bool) {
119
w.lk.RLock()
120
defer w.lk.RUnlock()
exchange/bitswap/wantmanager.go
+25
-24
@@ -25,6 +25,7 @@ type WantManager struct {
25
// synchronized by Run loop, only touch inside there
26
peers map[peer.ID]*msgQueue
27
wl *wantlist.ThreadSafe
28
+ bcwl *wantlist.ThreadSafe
29
30
network bsnet.BitSwapNetwork
31
ctx context.Context
@@ -47,6 +48,7 @@ func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantMana
48
peerReqs: make(chan chan []peer.ID),
49
peers: make(map[peer.ID]*msgQueue),
50
wl: wantlist.NewThreadSafe(),
51
+ bcwl: wantlist.NewThreadSafe(),
52
network: network,
53
ctx: ctx,
54
cancel: cancel,
@@ -61,7 +63,7 @@ type msgQueue struct {
63
outlk sync.Mutex
64
out bsmsg.BitSwapMessage
65
network bsnet.BitSwapNetwork
64
- wl *wantlist.Wantlist
66
+ wl *wantlist.ThreadSafe
67
68
sender bsnet.MessageSender
69
@@ -71,11 +73,13 @@ type msgQueue struct {
73
done chan struct{}
74
}
75
76
+// WantBlocks adds the given cids to the wantlist, tracked by the given session
77
func (pm *WantManager) WantBlocks(ctx context.Context, ks []*cid.Cid, peers []peer.ID, ses uint64) {
78
log.Infof("want blocks: %s", ks)
79
pm.addEntries(ctx, ks, peers, false, ses)
80
}
81
82
+// CancelWants removes the given cids from the wantlist, tracked by the given session
83
func (pm *WantManager) CancelWants(ctx context.Context, ks []*cid.Cid, peers []peer.ID, ses uint64) {
84
pm.addEntries(context.Background(), ks, peers, true, ses)
85
}
@@ -134,9 +138,10 @@ func (pm *WantManager) startPeerHandler(p peer.ID) *msgQueue {
138
139
// new peer, we will want to give them our full wantlist
140
fullwantlist := bsmsg.New(true)
137
- for _, e := range pm.wl.Entries() {
138
- ne := *e
139
- mq.wl.AddEntry(&ne)
141
+ for _, e := range pm.bcwl.Entries() {
142
+ for k := range e.SesTrk {
143
+ mq.wl.AddEntry(e, k)
144
+ }
145
fullwantlist.AddEntry(e.Cid, e.Priority)
146
}
147
mq.out = fullwantlist
@@ -284,13 +289,23 @@ func (pm *WantManager) Run() {
289
select {
290
case ws := <-pm.incoming:
291
292
+ // is this a broadcast or not?
293
+ brdc := len(ws.targets) == 0
294
+
295
// add changes to our wantlist
296
for _, e := range ws.entries {
297
if e.Cancel {
298
+ if brdc {
299
+ pm.bcwl.Remove(e.Cid, ws.from)
300
+ }
301
+
302
if pm.wl.Remove(e.Cid, ws.from) {
303
pm.wantlistGauge.Dec()
304
}
305
} else {
306
+ if brdc {
307
+ pm.bcwl.AddEntry(e.Entry, ws.from)
308
+ }
309
if pm.wl.AddEntry(e.Entry, ws.from) {
310
pm.wantlistGauge.Inc()
311
}
@@ -300,7 +315,7 @@ func (pm *WantManager) Run() {
315
// broadcast those wantlist changes
316
if len(ws.targets) == 0 {
317
for _, p := range pm.peers {
303
- p.addMessage(ws.entries)
318
+ p.addMessage(ws.entries, ws.from)
319
}
320
} else {
321
for _, t := range ws.targets {
@@ -309,24 +324,10 @@ func (pm *WantManager) Run() {
324
log.Warning("tried sending wantlist change to non-partner peer")
325
continue
326
}
312
- p.addMessage(ws.entries)
327
+ p.addMessage(ws.entries, ws.from)
328
}
329
}
330
316
- case <-tock.C:
317
- // resend entire wantlist every so often (REALLY SHOULDNT BE NECESSARY)
318
- var es []*bsmsg.Entry
319
- for _, e := range pm.wl.Entries() {
320
- es = append(es, &bsmsg.Entry{Entry: e})
321
- }
322
-
323
- for _, p := range pm.peers {
324
- p.outlk.Lock()
325
- p.out = bsmsg.New(true)
326
- p.outlk.Unlock()
327
-
328
- p.addMessage(es)
329
- }
331
case p := <-pm.connect:
332
pm.startPeerHandler(p)
333
case p := <-pm.disconnect:
@@ -347,14 +348,14 @@ func (wm *WantManager) newMsgQueue(p peer.ID) *msgQueue {
348
return &msgQueue{
349
done: make(chan struct{}),
350
work: make(chan struct{}, 1),
350
- wl: wantlist.New(),
351
+ wl: wantlist.NewThreadSafe(),
352
network: wm.network,
353
p: p,
354
refcnt: 1,
355
}
356
}
357
357
-func (mq *msgQueue) addMessage(entries []*bsmsg.Entry) {
358
+func (mq *msgQueue) addMessage(entries []*bsmsg.Entry, ses uint64) {
359
var work bool
360
mq.outlk.Lock()
361
defer func() {
@@ -378,12 +379,12 @@ func (mq *msgQueue) addMessage(entries []*bsmsg.Entry) {
379
// one passed in
380
for _, e := range entries {
381
if e.Cancel {
381
- if mq.wl.Remove(e.Cid) {
382
+ if mq.wl.Remove(e.Cid, ses) {
383
work = true
384
mq.out.Cancel(e.Cid)
385
}
386
} else {
386
- if mq.wl.Add(e.Cid, e.Priority) {
387
+ if mq.wl.Add(e.Cid, e.Priority, ses) {
388
work = true
389
mq.out.AddEntry(e.Cid, e.Priority)
390
}
exchange/interface.go
+1
@@ -24,6 +24,7 @@ type Interface interface { // type Exchanger interface
24
io.Closer
25
}
26
27
+// Fetcher is an object that can be used to retrieve blocks
28
type Fetcher interface {
29
// GetBlock returns the block associated with a given key.
30
GetBlock(context.Context, *cid.Cid) (blocks.Block, error)
merkledag/merkledag.go
+3
@@ -155,6 +155,9 @@ func GetLinksDirect(serv node.NodeGetter) GetLinks {
155
return func(ctx context.Context, c *cid.Cid) ([]*node.Link, error) {
156
node, err := serv.Get(ctx, c)
157
if err != nil {
158
+ if err == bserv.ErrNotFound {
159
+ err = ErrNotFound
160
+ }
161
return nil, err
162
}
163
return node.Links(), nil