rewrite FindProvidersAsync
Jeromy committed
Dec 11, 2014 at 05:08 UTC
2d94b784802f66210279d5dc3599914b5c760e0d
2 files changed
+65
-31
routing/dht/handlers.go
-1
@@ -182,7 +182,6 @@ func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.Peer, pmes *p
182
providers = append(providers, dht.self)
183
}
184
185
- // if we've got providers, send thos those.
185
if providers != nil && len(providers) > 0 {
186
resp.ProviderPeers = pb.PeersToPBPeers(dht.dialer, providers)
187
}
routing/dht/routing.go
+65
-30
@@ -138,43 +138,78 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
138
func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int) <-chan peer.Peer {
139
log.Event(ctx, "findProviders", &key)
140
peerOut := make(chan peer.Peer, count)
141
- go func() {
142
- defer close(peerOut)
143
-
144
- ps := newPeerSet()
145
- // TODO may want to make this function async to hide latency
146
- provs := dht.providers.GetProviders(ctx, key)
147
- for _, p := range provs {
148
- count--
149
- // NOTE: assuming that this list of peers is unique
150
- ps.Add(p)
141
+ go dht.findProvidersAsyncRoutine(ctx, key, count, peerOut)
142
+ return peerOut
143
+}
144
+
145
+func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, count int, peerOut chan peer.Peer) {
146
+ defer close(peerOut)
147
+
148
+ ps := newPeerSet()
149
+ provs := dht.providers.GetProviders(ctx, key)
150
+ for _, p := range provs {
151
+ count--
152
+ // NOTE: assuming that this list of peers is unique
153
+ ps.Add(p)
154
+ select {
155
+ case peerOut <- p:
156
+ case <-ctx.Done():
157
+ return
158
+ }
159
+ if count <= 0 {
160
+ return
161
+ }
162
+ }
163
+
164
+ // setup the Query
165
+ query := newQuery(key, dht.dialer, func(ctx context.Context, p peer.Peer) (*dhtQueryResult, error) {
166
+
167
+ pmes, err := dht.findProvidersSingle(ctx, p, key, 0)
168
+ if err != nil {
169
+ return nil, err
170
+ }
171
+
172
+ provs, errs := pb.PBPeersToPeers(dht.peerstore, pmes.GetProviderPeers())
173
+ for _, err := range errs {
174
+ if err != nil {
175
+ log.Warning(err)
176
+ }
177
+ }
178
+
179
+ // Add unique providers from request, up to 'count'
180
+ for _, prov := range provs {
181
+ if ps.Contains(prov) {
182
+ continue
183
+ }
184
select {
152
- case peerOut <- p:
185
+ case peerOut <- prov:
186
case <-ctx.Done():
154
- return
187
+ log.Error("Context timed out sending more providers")
188
+ return nil, ctx.Err()
189
}
156
- if count <= 0 {
157
- return
190
+ ps.Add(prov)
191
+ if ps.Size() >= count {
192
+ return &dhtQueryResult{success: true}, nil
193
}
194
}
195
161
- var wg sync.WaitGroup
162
- peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue)
163
- for _, pp := range peers {
164
- wg.Add(1)
165
- go func(p peer.Peer) {
166
- defer wg.Done()
167
- pmes, err := dht.findProvidersSingle(ctx, p, key, 0)
168
- if err != nil {
169
- log.Error(err)
170
- return
171
- }
172
- dht.addPeerListAsync(ctx, key, pmes.GetProviderPeers(), ps, count, peerOut)
173
- }(pp)
196
+ // Give closer peers back to the query to be queried
197
+ closer := pmes.GetCloserPeers()
198
+ clpeers, errs := pb.PBPeersToPeers(dht.peerstore, closer)
199
+ for _, err := range errs {
200
+ if err != nil {
201
+ log.Warning(err)
202
+ }
203
}
175
- wg.Wait()
176
- }()
177
- return peerOut
204
+
205
+ return &dhtQueryResult{closerPeers: clpeers}, nil
206
+ })
207
+
208
+ peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue)
209
+ _, err := query.Run(ctx, peers)
210
+ if err != nil {
211
+ log.Errorf("FindProviders Query error: %s", err)
212
+ }
213
}
214
215
func (dht *IpfsDHT) addPeerListAsync(ctx context.Context, k u.Key, peers []*pb.Message_Peer, ps *peerSet, count int, out chan peer.Peer) {