rewrite sendWantlistToProviders
Jeromy committed
Dec 15, 2014 at 01:33 UTC
cfbe92bc8bcddef4f429438a42ce32abec5929a0
4 files changed
+66
-83
exchange/bitswap/bitswap.go
+15
-37
@@ -19,6 +19,7 @@ import (
19
peer "github.com/jbenet/go-ipfs/peer"
20
u "github.com/jbenet/go-ipfs/util"
21
eventlog "github.com/jbenet/go-ipfs/util/eventlog"
22
+ pset "github.com/jbenet/go-ipfs/util/peerset"
23
)
24
25
var log = eventlog.Logger("bitswap")
@@ -204,57 +205,34 @@ func (bs *bitswap) sendWantListTo(ctx context.Context, peers <-chan peer.Peer) e
205
}
206
207
func (bs *bitswap) sendWantlistToProviders(ctx context.Context, wantlist *wl.Wantlist) {
207
- provset := make(map[u.Key]peer.Peer)
208
- provcollect := make(chan peer.Peer)
209
-
208
ctx, cancel := context.WithCancel(ctx)
209
defer cancel()
210
213
- wg := sync.WaitGroup{}
211
+ message := bsmsg.New()
212
+ message.SetFull(true)
213
+ for _, e := range bs.wantlist.Entries() {
214
+ message.AddEntry(e.Value, e.Priority, false)
215
+ }
216
+
217
+ ps := pset.NewPeerSet()
218
+
219
// Get providers for all entries in wantlist (could take a while)
220
+ wg := sync.WaitGroup{}
221
for _, e := range wantlist.Entries() {
222
wg.Add(1)
223
go func(k u.Key) {
224
+ defer wg.Done()
225
child, _ := context.WithTimeout(ctx, providerRequestTimeout)
226
providers := bs.routing.FindProvidersAsync(child, k, maxProvidersPerRequest)
227
228
for prov := range providers {
222
- provcollect <- prov
229
+ if ps.AddIfSmallerThan(prov, -1) { //Do once per peer
230
+ bs.send(ctx, prov, message)
231
+ }
232
}
224
- wg.Done()
233
}(e.Value)
234
}
227
-
228
- // When all workers finish, close the providers channel
229
- go func() {
230
- wg.Wait()
231
- close(provcollect)
232
- }()
233
-
234
- // Filter out duplicates,
235
- // no need to send our wantlists out twice in a given time period
236
- for {
237
- select {
238
- case p, ok := <-provcollect:
239
- if !ok {
240
- break
241
- }
242
- provset[p.Key()] = p
243
- case <-ctx.Done():
244
- log.Error("Context cancelled before we got all the providers!")
245
- return
246
- }
247
- }
248
-
249
- message := bsmsg.New()
250
- message.SetFull(true)
251
- for _, e := range bs.wantlist.Entries() {
252
- message.AddEntry(e.Value, e.Priority, false)
253
- }
254
-
255
- for _, prov := range provset {
256
- bs.send(ctx, prov, message)
257
- }
235
+ wg.Wait()
236
}
237
238
func (bs *bitswap) roundWorker(ctx context.Context) {
routing/dht/routing.go
+3
-2
@@ -11,6 +11,7 @@ import (
11
pb "github.com/jbenet/go-ipfs/routing/dht/pb"
12
kb "github.com/jbenet/go-ipfs/routing/kbucket"
13
u "github.com/jbenet/go-ipfs/util"
14
+ pset "github.com/jbenet/go-ipfs/util/peerset"
15
)
16
17
// asyncQueryBuffer is the size of buffered channels in async queries. This
@@ -140,7 +141,7 @@ func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int
141
func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, count int, peerOut chan peer.Peer) {
142
defer close(peerOut)
143
143
- ps := newPeerSet()
144
+ ps := pset.NewPeerSet()
145
provs := dht.providers.GetProviders(ctx, key)
146
for _, p := range provs {
147
// NOTE: assuming that this list of peers is unique
@@ -207,7 +208,7 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, co
208
}
209
}
210
210
-func (dht *IpfsDHT) addPeerListAsync(ctx context.Context, k u.Key, peers []*pb.Message_Peer, ps *peerSet, count int, out chan peer.Peer) {
211
+func (dht *IpfsDHT) addPeerListAsync(ctx context.Context, k u.Key, peers []*pb.Message_Peer, ps *pset.PeerSet, count int, out chan peer.Peer) {
212
var wg sync.WaitGroup
213
for _, pbp := range peers {
214
wg.Add(1)
routing/dht/util.go
-44
@@ -2,8 +2,6 @@ package dht
2
3
import (
4
"sync"
5
-
6
- peer "github.com/jbenet/go-ipfs/peer"
5
)
6
7
// Pool size is the number of nodes used for group find/set RPC calls
@@ -39,45 +37,3 @@ func (c *counter) Size() (s int) {
37
c.mut.Unlock()
38
return
39
}
42
-
43
-// peerSet is a threadsafe set of peers
44
-type peerSet struct {
45
- ps map[string]bool
46
- lk sync.RWMutex
47
-}
48
-
49
-func newPeerSet() *peerSet {
50
- ps := new(peerSet)
51
- ps.ps = make(map[string]bool)
52
- return ps
53
-}
54
-
55
-func (ps *peerSet) Add(p peer.Peer) {
56
- ps.lk.Lock()
57
- ps.ps[string(p.ID())] = true
58
- ps.lk.Unlock()
59
-}
60
-
61
-func (ps *peerSet) Contains(p peer.Peer) bool {
62
- ps.lk.RLock()
63
- _, ok := ps.ps[string(p.ID())]
64
- ps.lk.RUnlock()
65
- return ok
66
-}
67
-
68
-func (ps *peerSet) Size() int {
69
- ps.lk.RLock()
70
- defer ps.lk.RUnlock()
71
- return len(ps.ps)
72
-}
73
-
74
-func (ps *peerSet) AddIfSmallerThan(p peer.Peer, maxsize int) bool {
75
- var success bool
76
- ps.lk.Lock()
77
- if _, ok := ps.ps[string(p.ID())]; !ok && len(ps.ps) < maxsize {
78
- success = true
79
- ps.ps[string(p.ID())] = true
80
- }
81
- ps.lk.Unlock()
82
- return success
83
-}
util/peerset/peerset.go
new
+48
@@ -0,0 +1,48 @@
1
+package peerset
2
+
3
+import (
4
+ peer "github.com/jbenet/go-ipfs/peer"
5
+ "sync"
6
+)
7
+
8
+// PeerSet is a threadsafe set of peers
9
+type PeerSet struct {
10
+ ps map[string]bool
11
+ lk sync.RWMutex
12
+}
13
+
14
+func NewPeerSet() *PeerSet {
15
+ ps := new(PeerSet)
16
+ ps.ps = make(map[string]bool)
17
+ return ps
18
+}
19
+
20
+func (ps *PeerSet) Add(p peer.Peer) {
21
+ ps.lk.Lock()
22
+ ps.ps[string(p.ID())] = true
23
+ ps.lk.Unlock()
24
+}
25
+
26
+func (ps *PeerSet) Contains(p peer.Peer) bool {
27
+ ps.lk.RLock()
28
+ _, ok := ps.ps[string(p.ID())]
29
+ ps.lk.RUnlock()
30
+ return ok
31
+}
32
+
33
+func (ps *PeerSet) Size() int {
34
+ ps.lk.RLock()
35
+ defer ps.lk.RUnlock()
36
+ return len(ps.ps)
37
+}
38
+
39
+func (ps *PeerSet) AddIfSmallerThan(p peer.Peer, maxsize int) bool {
40
+ var success bool
41
+ ps.lk.Lock()
42
+ if _, ok := ps.ps[string(p.ID())]; !ok && len(ps.ps) < maxsize {
43
+ success = true
44
+ ps.ps[string(p.ID())] = true
45
+ }
46
+ ps.lk.Unlock()
47
+ return success
48
+}