wire contexts into bitswap requests more deeply
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Nov 20, 2015 at 15:24 UTC
1f4d3300b0efc93e4d565dd9507d126b77c8492f
7 files changed
+106
-85
blockservice/blockservice.go
+4
@@ -122,6 +122,10 @@ func (s *BlockService) GetBlocks(ctx context.Context, ks []key.Key) <-chan *bloc
122
}
123
}
124
125
+ if len(misses) == 0 {
126
+ return
127
+ }
128
+
129
rblocks, err := s.Exchange.GetBlocks(ctx, misses)
130
if err != nil {
131
log.Debugf("Error with GetBlocks: %s", err)
exchange/bitswap/bitswap.go
+17
-34
@@ -86,7 +86,7 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
86
notifications: notif,
87
engine: decision.NewEngine(ctx, bstore), // TODO close the engine with Close() method
88
network: network,
89
- findKeys: make(chan *blockRequest, sizeBatchRequestChan),
89
+ findKeys: make(chan *wantlist.Entry, sizeBatchRequestChan),
90
process: px,
91
newBlocks: make(chan *blocks.Block, HasBlockBufferSize),
92
provideKeys: make(chan key.Key, provideKeysBufferSize),
@@ -129,7 +129,7 @@ type Bitswap struct {
129
notifications notifications.PubSub
130
131
// send keys to a worker to find and connect to providers for them
132
- findKeys chan *blockRequest
132
+ findKeys chan *wantlist.Entry
133
134
engine *decision.Engine
135
@@ -146,8 +146,8 @@ type Bitswap struct {
146
}
147
148
type blockRequest struct {
149
- keys []key.Key
150
- ctx context.Context
149
+ key key.Key
150
+ ctx context.Context
151
}
152
153
// GetBlock attempts to retrieve a particular block from peers within the
@@ -208,6 +208,12 @@ func (bs *Bitswap) WantlistForPeer(p peer.ID) []key.Key {
208
// resources, provide a context with a reasonably short deadline (ie. not one
209
// that lasts throughout the lifetime of the server)
210
func (bs *Bitswap) GetBlocks(ctx context.Context, keys []key.Key) (<-chan *blocks.Block, error) {
211
+ if len(keys) == 0 {
212
+ out := make(chan *blocks.Block)
213
+ close(out)
214
+ return out, nil
215
+ }
216
+
217
select {
218
case <-bs.process.Closing():
219
return nil, errors.New("bitswap is closed")
@@ -219,11 +225,14 @@ func (bs *Bitswap) GetBlocks(ctx context.Context, keys []key.Key) (<-chan *block
225
log.Event(ctx, "Bitswap.GetBlockRequest.Start", &k)
226
}
227
222
- bs.wm.WantBlocks(keys)
228
+ bs.wm.WantBlocks(ctx, keys)
229
224
- req := &blockRequest{
225
- keys: keys,
226
- ctx: ctx,
230
+ // NB: Optimization. Assumes that providers of key[0] are likely to
231
+ // be able to provide for all keys. This currently holds true in most
232
+ // every situation. Later, this assumption may not hold as true.
233
+ req := &wantlist.Entry{
234
+ Key: keys[0],
235
+ Ctx: ctx,
236
}
237
select {
238
case bs.findKeys <- req:
@@ -276,32 +285,6 @@ func (bs *Bitswap) tryPutBlock(blk *blocks.Block, attempts int) error {
285
return err
286
}
287
279
-func (bs *Bitswap) connectToProviders(ctx context.Context, entries []wantlist.Entry) {
280
-
281
- ctx, cancel := context.WithCancel(ctx)
282
- defer cancel()
283
-
284
- // Get providers for all entries in wantlist (could take a while)
285
- wg := sync.WaitGroup{}
286
- for _, e := range entries {
287
- wg.Add(1)
288
- go func(k key.Key) {
289
- defer wg.Done()
290
-
291
- child, cancel := context.WithTimeout(ctx, providerRequestTimeout)
292
- defer cancel()
293
- providers := bs.network.FindProvidersAsync(child, k, maxProvidersPerRequest)
294
- for prov := range providers {
295
- go func(p peer.ID) {
296
- bs.network.ConnectTo(ctx, p)
297
- }(prov)
298
- }
299
- }(e.Key)
300
- }
301
-
302
- wg.Wait() // make sure all our children do finish.
303
-}
304
-
288
func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) {
289
// This call records changes to wantlists, blocks received,
290
// and number of bytes transfered.
exchange/bitswap/decision/engine.go
+1
-1
@@ -217,7 +217,7 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
217
e.peerRequestQueue.Remove(entry.Key, p)
218
} else {
219
log.Debugf("wants %s - %d", entry.Key, entry.Priority)
220
- l.Wants(entry.Key, entry.Priority)
220
+ l.Wants(entry.Ctx, entry.Key, entry.Priority)
221
if exists, err := e.bs.Has(entry.Key); err == nil && exists {
222
e.peerRequestQueue.Push(entry.Entry, p)
223
newWorkExists = true
exchange/bitswap/decision/ledger.go
+4
-2
@@ -6,6 +6,8 @@ import (
6
key "github.com/ipfs/go-ipfs/blocks/key"
7
wl "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
8
peer "gx/ipfs/QmZwZjMVGss5rqYsJVGy18gNbkTJffFyq2x1uJ4e4p3ZAt/go-libp2p-peer"
9
+
10
+ "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
11
)
12
13
// keySet is just a convenient alias for maps of keys, where we only care
@@ -68,9 +70,9 @@ func (l *ledger) ReceivedBytes(n int) {
70
}
71
72
// TODO: this needs to be different. We need timeouts.
71
-func (l *ledger) Wants(k key.Key, priority int) {
73
+func (l *ledger) Wants(ctx context.Context, k key.Key, priority int) {
74
log.Debugf("peer %s wants %s", l.Partner, k)
73
- l.wantList.Add(k, priority)
75
+ l.wantList.Add(ctx, k, priority)
76
}
77
78
func (l *ledger) CancelWant(k key.Key) {
exchange/bitswap/wantlist/wantlist.go
+22
-8
@@ -3,9 +3,12 @@
3
package wantlist
4
5
import (
6
- key "github.com/ipfs/go-ipfs/blocks/key"
6
"sort"
7
"sync"
8
+
9
+ key "github.com/ipfs/go-ipfs/blocks/key"
10
+
11
+ "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
12
)
13
14
type ThreadSafe struct {
@@ -16,7 +19,6 @@ type ThreadSafe struct {
19
// not threadsafe
20
type Wantlist struct {
21
set map[key.Key]Entry
19
- // TODO provide O(1) len accessor if cost becomes an issue
22
}
23
24
type Entry struct {
@@ -24,6 +26,7 @@ type Entry struct {
26
// slices can be copied efficiently.
27
Key key.Key
28
Priority int
29
+ Ctx context.Context
30
}
31
32
type entrySlice []Entry
@@ -44,22 +47,25 @@ func New() *Wantlist {
47
}
48
}
49
47
-func (w *ThreadSafe) Add(k key.Key, priority int) {
48
- // TODO rm defer for perf
50
+func (w *ThreadSafe) Add(ctx context.Context, k key.Key, priority int) {
51
+ w.lk.Lock()
52
+ defer w.lk.Unlock()
53
+ w.Wantlist.Add(ctx, k, priority)
54
+}
55
+
56
+func (w *ThreadSafe) AddEntry(e Entry) {
57
w.lk.Lock()
58
defer w.lk.Unlock()
51
- w.Wantlist.Add(k, priority)
59
+ w.Wantlist.AddEntry(e)
60
}
61
62
func (w *ThreadSafe) Remove(k key.Key) {
55
- // TODO rm defer for perf
63
w.lk.Lock()
64
defer w.lk.Unlock()
65
w.Wantlist.Remove(k)
66
}
67
68
func (w *ThreadSafe) Contains(k key.Key) (Entry, bool) {
62
- // TODO rm defer for perf
69
w.lk.RLock()
70
defer w.lk.RUnlock()
71
return w.Wantlist.Contains(k)
@@ -87,14 +93,22 @@ func (w *Wantlist) Len() int {
93
return len(w.set)
94
}
95
90
-func (w *Wantlist) Add(k key.Key, priority int) {
96
+func (w *Wantlist) Add(ctx context.Context, k key.Key, priority int) {
97
if _, ok := w.set[k]; ok {
98
return
99
}
100
w.set[k] = Entry{
101
Key: k,
102
Priority: priority,
103
+ Ctx: ctx,
104
+ }
105
+}
106
+
107
+func (w *Wantlist) AddEntry(e Entry) {
108
+ if _, ok := w.set[e.Key]; ok {
109
+ return
110
}
111
+ w.set[e.Key] = e
112
}
113
114
func (w *Wantlist) Remove(k key.Key) {
exchange/bitswap/wantmanager.go
+14
-5
@@ -64,16 +64,16 @@ type msgQueue struct {
64
done chan struct{}
65
}
66
67
-func (pm *WantManager) WantBlocks(ks []key.Key) {
67
+func (pm *WantManager) WantBlocks(ctx context.Context, ks []key.Key) {
68
log.Infof("want blocks: %s", ks)
69
- pm.addEntries(ks, false)
69
+ pm.addEntries(ctx, ks, false)
70
}
71
72
func (pm *WantManager) CancelWants(ks []key.Key) {
73
- pm.addEntries(ks, true)
73
+ pm.addEntries(context.TODO(), ks, true)
74
}
75
76
-func (pm *WantManager) addEntries(ks []key.Key, cancel bool) {
76
+func (pm *WantManager) addEntries(ctx context.Context, ks []key.Key, cancel bool) {
77
var entries []*bsmsg.Entry
78
for i, k := range ks {
79
entries = append(entries, &bsmsg.Entry{
@@ -81,6 +81,7 @@ func (pm *WantManager) addEntries(ks []key.Key, cancel bool) {
81
Entry: wantlist.Entry{
82
Key: k,
83
Priority: kMaxPriority - i,
84
+ Ctx: ctx,
85
},
86
})
87
}
@@ -224,7 +225,7 @@ func (pm *WantManager) Run() {
225
if e.Cancel {
226
pm.wl.Remove(e.Key)
227
} else {
227
- pm.wl.Add(e.Key, e.Priority)
228
+ pm.wl.AddEntry(e.Entry)
229
}
230
}
231
@@ -237,6 +238,14 @@ func (pm *WantManager) Run() {
238
// resend entire wantlist every so often (REALLY SHOULDNT BE NECESSARY)
239
var es []*bsmsg.Entry
240
for _, e := range pm.wl.Entries() {
241
+ select {
242
+ case <-e.Ctx.Done():
243
+ // entry has been cancelled
244
+ // simply continue, the entry will be removed from the
245
+ // wantlist soon enough
246
+ continue
247
+ default:
248
+ }
249
es = append(es, &bsmsg.Entry{Entry: e})
250
}
251
for _, p := range pm.peers {
exchange/bitswap/workers.go
+44
-35
@@ -1,6 +1,7 @@
1
package bitswap
2
3
import (
4
+ "sync"
5
"time"
6
7
process "gx/ipfs/QmQopLATEYMNg7dVqZRNDfeE2S1yKy8zrRh5xnYiuqeZBn/goprocess"
@@ -8,6 +9,8 @@ import (
9
context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
10
11
key "github.com/ipfs/go-ipfs/blocks/key"
12
+ wantlist "github.com/ipfs/go-ipfs/exchange/bitswap/wantlist"
13
+ peer "gx/ipfs/QmZwZjMVGss5rqYsJVGy18gNbkTJffFyq2x1uJ4e4p3ZAt/go-libp2p-peer"
14
logging "gx/ipfs/Qmazh5oNUVsDZTs2g59rq8aYQqwpss8tcUWQzor5sCCEuH/go-log"
15
)
16
@@ -16,7 +19,7 @@ var TaskWorkerCount = 8
19
func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
20
// Start up a worker to handle block requests this node is making
21
px.Go(func(px process.Process) {
19
- bs.providerConnector(ctx)
22
+ bs.providerQueryManager(ctx)
23
})
24
25
// Start up workers to handle requests from other nodes for the data on this node
@@ -149,37 +152,6 @@ func (bs *Bitswap) provideCollector(ctx context.Context) {
152
}
153
}
154
152
-// connects to providers for the given keys
153
-func (bs *Bitswap) providerConnector(parent context.Context) {
154
- defer log.Info("bitswap client worker shutting down...")
155
-
156
- for {
157
- log.Event(parent, "Bitswap.ProviderConnector.Loop")
158
- select {
159
- case req := <-bs.findKeys:
160
- keys := req.keys
161
- if len(keys) == 0 {
162
- log.Warning("Received batch request for zero blocks")
163
- continue
164
- }
165
- log.Event(parent, "Bitswap.ProviderConnector.Work", logging.LoggableMap{"Keys": keys})
166
-
167
- // NB: Optimization. Assumes that providers of key[0] are likely to
168
- // be able to provide for all keys. This currently holds true in most
169
- // every situation. Later, this assumption may not hold as true.
170
- child, cancel := context.WithTimeout(req.ctx, providerRequestTimeout)
171
- providers := bs.network.FindProvidersAsync(child, keys[0], maxProvidersPerRequest)
172
- for p := range providers {
173
- go bs.network.ConnectTo(req.ctx, p)
174
- }
175
- cancel()
176
-
177
- case <-parent.Done():
178
- return
179
- }
180
- }
181
-}
182
-
155
func (bs *Bitswap) rebroadcastWorker(parent context.Context) {
156
ctx, cancel := context.WithCancel(parent)
157
defer cancel()
@@ -200,12 +172,49 @@ func (bs *Bitswap) rebroadcastWorker(parent context.Context) {
172
}
173
case <-broadcastSignal.C: // resend unfulfilled wantlist keys
174
log.Event(ctx, "Bitswap.Rebroadcast.active")
203
- entries := bs.wm.wl.Entries()
204
- if len(entries) > 0 {
205
- bs.connectToProviders(ctx, entries)
175
+ for _, e := range bs.wm.wl.Entries() {
176
+ bs.findKeys <- &e
177
}
178
case <-parent.Done():
179
return
180
}
181
}
182
}
183
+
184
+func (bs *Bitswap) providerQueryManager(ctx context.Context) {
185
+ var activeLk sync.Mutex
186
+ active := make(map[key.Key]*wantlist.Entry)
187
+
188
+ for {
189
+ select {
190
+ case e := <-bs.findKeys:
191
+ activeLk.Lock()
192
+ if _, ok := active[e.Key]; ok {
193
+ activeLk.Unlock()
194
+ continue
195
+ }
196
+ active[e.Key] = e
197
+ activeLk.Unlock()
198
+
199
+ go func(e *wantlist.Entry) {
200
+ child, cancel := context.WithTimeout(e.Ctx, providerRequestTimeout)
201
+ defer cancel()
202
+ providers := bs.network.FindProvidersAsync(child, e.Key, maxProvidersPerRequest)
203
+ for p := range providers {
204
+ go func(p peer.ID) {
205
+ err := bs.network.ConnectTo(child, p)
206
+ if err != nil {
207
+ log.Debug("failed to connect to provider %s: %s", p, err)
208
+ }
209
+ }(p)
210
+ }
211
+ activeLk.Lock()
212
+ delete(active, e.Key)
213
+ activeLk.Unlock()
214
+ }(e)
215
+
216
+ case <-ctx.Done():
217
+ return
218
+ }
219
+ }
220
+}