Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

rewrite FindProvidersAsync #440

Merged
merged 2 commits into from
Dec 11, 2014
Merged

rewrite FindProvidersAsync #440

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2d94b78
rewrite FindProvidersAsync
whyrusleeping Dec 11, 2014
9e14bcf
changes from PR
whyrusleeping Dec 11, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion routing/dht/handlers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,6 @@ func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.Peer, pmes *p
providers = append(providers, dht.self)
}

// if we've got providers, send thos those.
if providers != nil && len(providers) > 0 {
resp.ProviderPeers = pb.PeersToPBPeers(dht.dialer, providers)
}
Expand Down
89 changes: 62 additions & 27 deletions routing/dht/routing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,43 +138,78 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int) <-chan peer.Peer {
log.Event(ctx, "findProviders", &key)
peerOut := make(chan peer.Peer, count)
go func() {
defer close(peerOut)

ps := newPeerSet()
// TODO may want to make this function async to hide latency
provs := dht.providers.GetProviders(ctx, key)
for _, p := range provs {
count--
// NOTE: assuming that this list of peers is unique
ps.Add(p)
go dht.findProvidersAsyncRoutine(ctx, key, count, peerOut)
return peerOut
}

func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, count int, peerOut chan peer.Peer) {
defer close(peerOut)

ps := newPeerSet()
provs := dht.providers.GetProviders(ctx, key)
for _, p := range provs {
// NOTE: assuming that this list of peers is unique
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what if this assumption isn't true? could it make it hang?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nah, it wont hang, we just might send a request to the same peer twice. But im very certain that GetProviders from our provider manager returns a unique list.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

easy to do the ps.Contains check here too.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that it is... i think not doing that is a result of my shitty-programmer-copy-paste-itis

if ps.AddIfSmallerThan(p, count) {
select {
case peerOut <- p:
case <-ctx.Done():
return
}
if count <= 0 {
return
}

// If we have enough peers locally, dont bother with remote RPC
if ps.Size() >= count {
return
}
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

afaict, this is an optimization in case we have enough locally, right? maybe comment it so?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thats exactly what it is, since the call asks for a certain number of providers, we dont ever want to send more than they expect.


// setup the Query
query := newQuery(key, dht.dialer, func(ctx context.Context, p peer.Peer) (*dhtQueryResult, error) {

pmes, err := dht.findProvidersSingle(ctx, p, key, 0)
if err != nil {
return nil, err
}

provs, errs := pb.PBPeersToPeers(dht.peerstore, pmes.GetProviderPeers())
for _, err := range errs {
if err != nil {
log.Warning(err)
}
}

var wg sync.WaitGroup
peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue)
for _, pp := range peers {
wg.Add(1)
go func(p peer.Peer) {
defer wg.Done()
pmes, err := dht.findProvidersSingle(ctx, p, key, 0)
if err != nil {
log.Error(err)
return
// Add unique providers from request, up to 'count'
for _, prov := range provs {
if ps.AddIfSmallerThan(prov, count) {
select {
case peerOut <- prov:
case <-ctx.Done():
log.Error("Context timed out sending more providers")
return nil, ctx.Err()
}
dht.addPeerListAsync(ctx, key, pmes.GetProviderPeers(), ps, count, peerOut)
}(pp)
}
if ps.Size() >= count {
return &dhtQueryResult{success: true}, nil
}
}
wg.Wait()
}()
return peerOut

// Give closer peers back to the query to be queried
closer := pmes.GetCloserPeers()
clpeers, errs := pb.PBPeersToPeers(dht.peerstore, closer)
for _, err := range errs {
if err != nil {
log.Warning(err)
}
}

return &dhtQueryResult{closerPeers: clpeers}, nil
})

peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

that [0] always sketches me out. (we should probably kill it)

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah, i can switch over to a single routing table in this PR if youre okay with that

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes pls

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

actually, just merge this and another PR. small PRs = easy to CR

_, err := query.Run(ctx, peers)
if err != nil {
log.Errorf("FindProviders Query error: %s", err)
}
}

func (dht *IpfsDHT) addPeerListAsync(ctx context.Context, k u.Key, peers []*pb.Message_Peer, ps *peerSet, count int, out chan peer.Peer) {
Expand Down