close_issue_2
/
tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor/importgraph: reduce I/O concurrency to avoid EMFILE 

Use an concurrency-limiting semaphore to reduce I/O parallelism in Import.

Also, start the producer in a new goroutine so that it runs in parallel
with the consumer.  Paradoxically, this reduces the peak number of
goroutines.

Also, in buildutil.ForEachPackage, make the concurrency limiting
semaphore global, since I/O parallelism is a process-wide resource.

Change-Id: I282b717c50603361826e5675077c9f464c874132
Reviewed-on: https://go-review.googlesource.com/18215
Reviewed-by: Michael Matloob <matloob@golang.org>
This commit is contained in:
Alan Donovan 2015-12-30 14:17:09 -05:00
parent 6e2f52e572
commit 3a85b8da38
2 changed files with 68 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
go/buildutil/allpackages.go
View File

@ -50,10 +50,6 @@ func AllPackages(ctxt *build.Context) []string {
// which must be concurrency-safe. // which must be concurrency-safe.
// //
func ForEachPackage(ctxt *build.Context, found func(importPath string, err error)) { func ForEachPackage(ctxt *build.Context, found func(importPath string, err error)) {
// We use a counting semaphore to limit
// the number of parallel calls to ReadDir.
sema := make(chan bool, 20)
ch := make(chan item) ch := make(chan item)
var wg sync.WaitGroup var wg sync.WaitGroup
@ -61,7 +57,7 @@ func ForEachPackage(ctxt *build.Context, found func(importPath string, err error
root := root root := root
wg.Add(1) wg.Add(1)
go func() { go func() {
allPackages(ctxt, sema, root, ch) allPackages(ctxt, root, ch)
wg.Done() wg.Done()
}() }()
} }
@ -81,7 +77,11 @@ type item struct {
err error // (optional) err error // (optional)
} }
func allPackages(ctxt *build.Context, sema chan bool, root string, ch chan<- item) { // We use a process-wide counting semaphore to limit
// the number of parallel calls to ReadDir.
var ioLimit = make(chan bool, 20)
func allPackages(ctxt *build.Context, root string, ch chan<- item) {
root = filepath.Clean(root) + string(os.PathSeparator) root = filepath.Clean(root) + string(os.PathSeparator)
var wg sync.WaitGroup var wg sync.WaitGroup
@ -102,9 +102,9 @@ func allPackages(ctxt *build.Context, sema chan bool, root string, ch chan<- ite
return return
} }
sema <- true ioLimit <- true
files, err := ReadDir(ctxt, dir) files, err := ReadDir(ctxt, dir)
<-sema <-ioLimit
if pkg != "" || err != nil { if pkg != "" || err != nil {
ch <- item{pkg, err} ch <- item{pkg, err}
} }

10
refactor/importgraph/graph.go
View File

@ -68,17 +68,19 @@ func Build(ctxt *build.Context) (forward, reverse Graph, errors map[string]error
ch := make(chan interface{}) ch := make(chan interface{})
go func() {
sema := make(chan int, 20) // I/O concurrency limiting semaphore sema := make(chan int, 20) // I/O concurrency limiting semaphore
var wg sync.WaitGroup var wg sync.WaitGroup
buildutil.ForEachPackage(ctxt, func(path string, err error) { buildutil.ForEachPackage(ctxt, func(path string, err error) {
wg.Add(1)
go func() {
defer wg.Done()
if err != nil { if err != nil {
ch <- pathError{path, err} ch <- pathError{path, err}
return return
} }
wg.Add(1)
go func() {
defer wg.Done()
sema <- 1 sema <- 1
bp, err := ctxt.Import(path, "", buildutil.AllowVendor) bp, err := ctxt.Import(path, "", buildutil.AllowVendor)
<-sema <-sema
@ -135,9 +137,9 @@ func Build(ctxt *build.Context) (forward, reverse Graph, errors map[string]error
ch <- importEdge{path, absolutize(imp)} ch <- importEdge{path, absolutize(imp)}
} }
} }
}() }()
}) })
go func() {
wg.Wait() wg.Wait()
close(ch) close(ch)
}() }()