opensource/kubo
1
0
Fork 0
mirror of https://github.com/ipfs/kubo.git synced 2025-07-03 04:37:30 +08:00
Code Issues Projects Releases Wiki Activity

merkledag: add a concurrency limit to merkledag fetch graph

Browse Source 
License: MIT
Signed-off-by: Jeromy <why@ipfs.io>
This commit is contained in:
Jeromy
2016-12-09 14:58:12 -08:00
parent b6c1155ac8
commit 0c14b4162e
1 changed files with 14 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
merkledag/merkledag.go
View File

@ -449,6 +449,10 @@ func EnumerateChildrenAsync(ctx context.Context, ds DAGService, c *cid.Cid, visi
} }
} }
// FetchGraphConcurrency is total number of concurrenct fetches that
// 'fetchNodes' will start at a time
var FetchGraphConcurrency = 8
func fetchNodes(ctx context.Context, ds DAGService, in <-chan []*cid.Cid, out chan<- *NodeOption) { func fetchNodes(ctx context.Context, ds DAGService, in <-chan []*cid.Cid, out chan<- *NodeOption) {
var wg sync.WaitGroup var wg sync.WaitGroup
defer func() { defer func() {
@ -458,8 +462,13 @@ func fetchNodes(ctx context.Context, ds DAGService, in <-chan []*cid.Cid, out ch
close(out) close(out)
}() }()
rateLimit := make(chan struct{}, FetchGraphConcurrency)
get := func(ks []*cid.Cid) { get := func(ks []*cid.Cid) {
defer wg.Done() defer wg.Done()
defer func() {
<-rateLimit
}()
nodes := ds.GetMany(ctx, ks) nodes := ds.GetMany(ctx, ks)
for opt := range nodes { for opt := range nodes {
select { select {
@ -471,6 +480,11 @@ func fetchNodes(ctx context.Context, ds DAGService, in <-chan []*cid.Cid, out ch
} }
for ks := range in { for ks := range in {
select {
case rateLimit <- struct{}{}:
case <-ctx.Done():
return
}
wg.Add(1) wg.Add(1)
go get(ks) go get(ks)
} }