cmd/go: parallelize module load operations
Loading information about modules not yet cached
on the local system is dominated by network I/O time.
Load the information in parallel to overlap this I/O.
As a test case, I am using a real (not synthetic) but unpublished
repo with 90 transitive dependencies. In that repo,
I've measured the time it takes for
rm -rf $GOPATH/src/v && vgo mod -vendor
Before this CL, that operation averages about 220 seconds.
Adding caching of already-looked-up information
in modfetch/repo.go drops the average time to about 170 seconds.
Changing the MVS requirement scans to run in parallel
drops the average time to about 65 seconds.
This CL includes Bryan's comments for mvs.Reqs
from CL 116235, extended to document the new parallel
access requirements.
This CL also adds the build -x flag to 'vgo mod' so that it's
possible to see all the git commands being executed.
Fixes golang/go#24316.
Change-Id: Ia8228e850c6c60d8fa80d2fb3b6773c70e86137b
Reviewed-on: https://go-review.googlesource.com/119055
Run-TryBot: Russ Cox <rsc@golang.org>
Reviewed-by: Bryan C. Mills <bcmills@google.com>
TryBot-Result: Gobot Gobot <gobot@golang.org>
diff --git a/vendor/cmd/go/internal/modcmd/mod.go b/vendor/cmd/go/internal/modcmd/mod.go
index b97cd8c..47facee 100644
--- a/vendor/cmd/go/internal/modcmd/mod.go
+++ b/vendor/cmd/go/internal/modcmd/mod.go
@@ -178,6 +178,8 @@
CmdMod.Flag.Var(flagFunc(flagDropReplace), "dropreplace", "")
CmdMod.Flag.Var(flagFunc(flagAddReplace), "addreplace", "")
CmdMod.Flag.Var(flagFunc(flagDropExclude), "dropexclude", "")
+
+ base.AddBuildFlagsNX(&CmdMod.Flag)
}
func runMod(cmd *base.Command, args []string) {
diff --git a/vendor/cmd/go/internal/modfetch/cache.go b/vendor/cmd/go/internal/modfetch/cache.go
new file mode 100644
index 0000000..d140d47
--- /dev/null
+++ b/vendor/cmd/go/internal/modfetch/cache.go
@@ -0,0 +1,115 @@
+// Copyright 2018 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package modfetch
+
+import (
+ "sync"
+
+ "cmd/go/internal/par"
+)
+
+// A cachingRepo is a cache around an underlying Repo,
+// avoiding redundant calls to ModulePath, Versions, Stat, Latest, and GoMod (but not Zip).
+// It is also safe for simultaneous use by multiple goroutines
+// (so that it can be returned from Lookup multiple times).
+// It serializes calls to the underlying Repo.
+type cachingRepo struct {
+ path string
+ cache par.Cache // cache for all operations
+
+ mu sync.Mutex // protects r's methods
+ r Repo
+}
+
+func newCachingRepo(r Repo) *cachingRepo {
+ return &cachingRepo{
+ r: r,
+ path: r.ModulePath(),
+ }
+}
+
+func (r *cachingRepo) ModulePath() string {
+ return r.path
+}
+
+func (r *cachingRepo) Versions(prefix string) ([]string, error) {
+ type cached struct {
+ list []string
+ err error
+ }
+ c := r.cache.Do("versions:"+prefix, func() interface{} {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ list, err := r.r.Versions(prefix)
+ return cached{list, err}
+ }).(cached)
+
+ if c.err != nil {
+ return nil, c.err
+ }
+ return append([]string(nil), c.list...), nil
+}
+
+func (r *cachingRepo) Stat(rev string) (*RevInfo, error) {
+ type cached struct {
+ info *RevInfo
+ err error
+ }
+ c := r.cache.Do("stat:"+rev, func() interface{} {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ info, err := r.r.Stat(rev)
+ return cached{info, err}
+ }).(cached)
+
+ if c.err != nil {
+ return nil, c.err
+ }
+ info := *c.info
+ return &info, nil
+}
+
+func (r *cachingRepo) Latest() (*RevInfo, error) {
+ type cached struct {
+ info *RevInfo
+ err error
+ }
+ c := r.cache.Do("latest:", func() interface{} {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ info, err := r.r.Latest()
+ return cached{info, err}
+ }).(cached)
+
+ if c.err != nil {
+ return nil, c.err
+ }
+ info := *c.info
+ return &info, nil
+}
+
+func (r *cachingRepo) GoMod(rev string) ([]byte, error) {
+ type cached struct {
+ text []byte
+ err error
+ }
+ c := r.cache.Do("gomod:", func() interface{} {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ text, err := r.r.GoMod(rev)
+ return cached{text, err}
+ }).(cached)
+
+ if c.err != nil {
+ return nil, c.err
+ }
+ return append([]byte(nil), c.text...), nil
+}
+
+func (r *cachingRepo) Zip(version, tmpdir string) (string, error) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.r.Zip(version, tmpdir)
+}
diff --git a/vendor/cmd/go/internal/modfetch/repo.go b/vendor/cmd/go/internal/modfetch/repo.go
index b089981..83b2262 100644
--- a/vendor/cmd/go/internal/modfetch/repo.go
+++ b/vendor/cmd/go/internal/modfetch/repo.go
@@ -7,6 +7,7 @@
import (
"errors"
"fmt"
+ "os"
pathpkg "path"
"sort"
"strings"
@@ -18,9 +19,12 @@
"cmd/go/internal/modfetch/github"
"cmd/go/internal/modfetch/googlesource"
"cmd/go/internal/module"
+ "cmd/go/internal/par"
"cmd/go/internal/semver"
)
+const traceRepo = false // trace all repo actions, for debugging
+
// A Repo represents a repository storing all versions of a single module.
type Repo interface {
// ModulePath returns the module path.
@@ -60,8 +64,34 @@
Time time.Time // commit time
}
+var lookupCache par.Cache
+
// Lookup returns the module with the given module path.
func Lookup(path string) (Repo, error) {
+ if traceRepo {
+ defer logCall("Lookup(%q)", path)()
+ }
+
+ type cached struct {
+ r Repo
+ err error
+ }
+ c := lookupCache.Do(path, func() interface{} {
+ r, err := lookup(path)
+ if err == nil {
+ if traceRepo {
+ r = newLoggingRepo(r)
+ }
+ r = newCachingRepo(r)
+ }
+ return cached{r, err}
+ }).(cached)
+
+ return c.r, c.err
+}
+
+// lookup returns the module with the given module path.
+func lookup(path string) (r Repo, err error) {
if cfg.BuildGetmode != "" {
return nil, fmt.Errorf("module lookup disabled by -getmode=%s", cfg.BuildGetmode)
}
@@ -78,6 +108,9 @@
}
func Import(path string, allowed func(module.Version) bool) (Repo, *RevInfo, error) {
+ if traceRepo {
+ defer logCall("Import(%q, ...)", path)()
+ }
try := func(path string) (Repo, *RevInfo, error) {
r, err := Lookup(path)
if err != nil {
@@ -140,3 +173,59 @@
return list[i] < list[j]
})
}
+
+// A loggingRepo is a wrapper around an underlying Repo
+// that prints a log message at the start and end of each call.
+// It can be inserted when debugging.
+type loggingRepo struct {
+ r Repo
+}
+
+func newLoggingRepo(r Repo) *loggingRepo {
+ return &loggingRepo{r}
+}
+
+// logCall prints a log message using format and args and then
+// also returns a function that will print the same message again,
+// along with the elapsed time.
+// Typical usage is:
+//
+// defer logCall("hello %s", arg)()
+//
+// Note the final ().
+func logCall(format string, args ...interface{}) func() {
+ start := time.Now()
+ fmt.Fprintf(os.Stderr, "+++ %s\n", fmt.Sprintf(format, args...))
+ return func() {
+ fmt.Fprintf(os.Stderr, "%.3fs %s\n", time.Since(start).Seconds(), fmt.Sprintf(format, args...))
+ }
+}
+
+func (l *loggingRepo) ModulePath() string {
+ return l.r.ModulePath()
+}
+
+func (l *loggingRepo) Versions(prefix string) (tags []string, err error) {
+ defer logCall("Repo[%s]: Versions(%q)", l.r.ModulePath(), prefix)()
+ return l.r.Versions(prefix)
+}
+
+func (l *loggingRepo) Stat(rev string) (*RevInfo, error) {
+ defer logCall("Repo[%s]: Stat(%q)", l.r.ModulePath(), rev)()
+ return l.r.Stat(rev)
+}
+
+func (l *loggingRepo) Latest() (*RevInfo, error) {
+ defer logCall("Repo[%s]: Latest()", l.r.ModulePath())()
+ return l.r.Latest()
+}
+
+func (l *loggingRepo) GoMod(version string) ([]byte, error) {
+ defer logCall("Repo[%s]: GoMod(%q)", l.r.ModulePath(), version)()
+ return l.r.GoMod(version)
+}
+
+func (l *loggingRepo) Zip(version, tmpdir string) (string, error) {
+ defer logCall("Repo[%s]: Zip(%q, %q)", l.r.ModulePath(), version, tmpdir)()
+ return l.r.Zip(version, tmpdir)
+}
diff --git a/vendor/cmd/go/internal/mvs/mvs.go b/vendor/cmd/go/internal/mvs/mvs.go
index 47670ff..9e7816d 100644
--- a/vendor/cmd/go/internal/mvs/mvs.go
+++ b/vendor/cmd/go/internal/mvs/mvs.go
@@ -9,14 +9,47 @@
import (
"fmt"
"sort"
+ "sync"
"cmd/go/internal/module"
+ "cmd/go/internal/par"
)
+// A Reqs is the requirement graph on which Minimal Version Selection (MVS) operates.
+//
+// The version strings are opaque except for the special versions "" and "none"
+// (see the documentation for module.Version). In particular, MVS does not
+// assume that the version strings are semantic versions; instead, the Max method
+// gives access to the comparison operation.
+//
+// It must be safe to call methods on a Reqs from multiple goroutines simultaneously.
+// Because a Reqs may read the underlying graph from the network on demand,
+// the MVS algorithms parallelize the traversal to overlap network delays.
type Reqs interface {
+ // Required returns the module versions explicitly required by m itself.
+ // The caller must not modify the returned list.
Required(m module.Version) ([]module.Version, error)
+
+ // Max returns the maximum of v1 and v2 (it returns either v1 or v2).
+ // For all versions v, Max(v, "none") must be v.
+ // TODO(rsc,bcmills): For all versions v, Max(v, "") must be "" ? Maybe.
+ //
+ // Note that v1 < v2 can be written Max(v1, v2) != v1
+ // and similarly v1 <= v2 can be written Max(v1, v2) == v2.
Max(v1, v2 string) string
+
+ // Latest returns the latest known version of the module at path
+ // (the one to use during UpgradeAll).
+ //
+ // Latest never returns version "none": if no module exists at the given path,
+ // it returns a non-nil error instead.
+ //
+ // TODO(bcmills): If path is the current module, must Latest return version
+ // "", or the most recent prior version?
Latest(path string) (module.Version, error)
+
+ // Previous returns the version of m.Path immediately prior to m.Version,
+ // or "none" if no such version is known.
Previous(m module.Version) (module.Version, error)
}
@@ -34,30 +67,35 @@
}
func buildList(target module.Version, reqs Reqs, uses map[module.Version][]module.Version, vers map[string]string) ([]module.Version, error) {
+ // Explore work graph in parallel in case reqs.Required
+ // does high-latency network operations.
+ var work par.Work
+ work.Add(target)
var (
- min = map[string]string{target.Path: target.Version}
- todo = []module.Version{target}
- seen = map[module.Version]bool{target: true}
+ mu sync.Mutex
+ min = map[string]string{target.Path: target.Version}
)
- for len(todo) > 0 {
- m := todo[len(todo)-1]
- todo = todo[:len(todo)-1]
+ work.Do(10, func(item interface{}) {
+ m := item.(module.Version)
required, _ := reqs.Required(m)
+
+ for _, r := range required {
+ work.Add(r)
+ }
+
+ mu.Lock()
+ defer mu.Unlock()
for _, r := range required {
if uses != nil {
uses[r] = append(uses[r], m)
}
- if !seen[r] {
- if v, ok := min[r.Path]; !ok {
- min[r.Path] = r.Version
- } else if max := reqs.Max(v, r.Version); max != v {
- min[r.Path] = max
- }
- todo = append(todo, r)
- seen[r] = true
+ if v, ok := min[r.Path]; !ok {
+ min[r.Path] = r.Version
+ } else if max := reqs.Max(v, r.Version); max != v {
+ min[r.Path] = max
}
}
- }
+ })
if min[target.Path] != target.Version {
panic("unbuildable") // TODO
@@ -93,8 +131,12 @@
}
// Req returns the minimal requirement list for the target module
-// that result in the given build list.
+// that results in the given build list.
func Req(target module.Version, list []module.Version, reqs Reqs) ([]module.Version, error) {
+ // Note: Not running in parallel because we assume
+ // that list came from a previous operation that paged
+ // in all the requirements, so there's no I/O to overlap now.
+
// Compute postorder, cache requirements.
var postorder []module.Version
reqCache := map[module.Version][]module.Version{}
@@ -165,28 +207,56 @@
// UpgradeAll returns a build list for the target module
// in which every module is upgraded to its latest version.
func UpgradeAll(target module.Version, reqs Reqs) ([]module.Version, error) {
- have := map[string]bool{target.Path: true}
- list := []module.Version{target}
- for i := 0; i < len(list); i++ {
- m := list[i]
- required, err := reqs.Required(m)
- if err != nil {
- panic(err) // TODO
- }
- for _, r := range required {
- latest, err := reqs.Latest(r.Path)
+ // Explore work graph in parallel, like in buildList,
+ // but here the work item is only a path, not a path+version pair,
+ // because we always take the latest of any path.
+ var work par.Work
+ work.Add(target.Path)
+ var (
+ mu sync.Mutex
+ list []module.Version
+ min = map[string]string{target.Path: ""}
+ )
+ work.Do(10, func(item interface{}) {
+ path := item.(string)
+ m := module.Version{Path: path}
+ if path != target.Path {
+ latest, err := reqs.Latest(path)
if err != nil {
panic(err) // TODO
}
- if reqs.Max(latest.Version, r.Version) != latest.Version {
- panic("mistake") // TODO
- }
- if !have[r.Path] {
- have[r.Path] = true
- list = append(list, module.Version{Path: r.Path, Version: latest.Version})
+ m.Version = latest.Version
+ }
+
+ required, err := reqs.Required(m)
+ if err != nil {
+ panic("TODO")
+ }
+
+ mu.Lock()
+ // Important: must append to list before calling work.Add (below).
+ // We expect the first work item (target) to be first in list.
+ list = append(list, m)
+ for _, r := range required {
+ if v, ok := min[r.Path]; !ok {
+ min[r.Path] = r.Version
+ } else {
+ min[r.Path] = reqs.Max(v, r.Version)
}
}
+ mu.Unlock()
+
+ for _, r := range required {
+ work.Add(r.Path)
+ }
+ })
+
+ for _, m := range list {
+ if reqs.Max(m.Version, min[m.Path]) != m.Version {
+ panic("mistake") // TODO
+ }
}
+
tail := list[1:]
sort.Slice(tail, func(i, j int) bool {
return tail[i].Path < tail[j].Path
diff --git a/vendor/cmd/go/internal/par/work.go b/vendor/cmd/go/internal/par/work.go
new file mode 100644
index 0000000..31c29b1
--- /dev/null
+++ b/vendor/cmd/go/internal/par/work.go
@@ -0,0 +1,128 @@
+// Copyright 2018 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package par implements parallel execution helpers.
+package par
+
+import (
+ "math/rand"
+ "sync"
+)
+
+// Work manages a set of work items to be executed in parallel, at most once each.
+// The items in the set must all be valid map keys.
+type Work struct {
+ f func(interface{}) // function to run for each item
+ running int // total number of runners
+
+ mu sync.Mutex
+ added map[interface{}]bool // items added to set
+ todo []interface{} // items yet to be run
+ wait sync.Cond // wait when todo is empty
+ waiting int // number of runners waiting for todo
+}
+
+func (w *Work) init() {
+ if w.added == nil {
+ w.added = make(map[interface{}]bool)
+ w.wait.L = &w.mu
+ }
+}
+
+// Add adds item to the work set, if it hasn't already been added.
+func (w *Work) Add(item interface{}) {
+ w.mu.Lock()
+ w.init()
+ if !w.added[item] {
+ w.added[item] = true
+ w.todo = append(w.todo, item)
+ if w.waiting > 0 {
+ w.wait.Signal()
+ }
+ }
+ w.mu.Unlock()
+}
+
+// Do runs f in parallel on items from the work set,
+// with at most n invocations of f running at a time.
+// It returns when everything added to the work set has been processed.
+// At least one item should have been added to the work set
+// before calling Do (or else Do returns immediately),
+// but it is allowed for f(item) to add new items to the set.
+// Do should only be used once on a given Work.
+func (w *Work) Do(n int, f func(item interface{})) {
+ if n < 1 {
+ panic("par.Work.Do: n < 1")
+ }
+ if w.running >= 1 {
+ panic("par.Work.Do: already called Do")
+ }
+
+ w.running = n
+ w.f = f
+
+ for i := 0; i < n-1; i++ {
+ go w.runner()
+ }
+ w.runner()
+}
+
+// runner executes work in w until both nothing is left to do
+// and all the runners are waiting for work.
+// (Then all the runners return.)
+func (w *Work) runner() {
+ for {
+ // Wait for something to do.
+ w.mu.Lock()
+ for len(w.todo) == 0 {
+ w.waiting++
+ if w.waiting == w.running {
+ // All done.
+ w.wait.Broadcast()
+ w.mu.Unlock()
+ return
+ }
+ w.wait.Wait()
+ w.waiting--
+ }
+
+ // Pick something to do at random,
+ // to eliminate pathological contention
+ // in case items added at about the same time
+ // are most likely to contend.
+ i := rand.Intn(len(w.todo))
+ item := w.todo[i]
+ w.todo[i] = w.todo[len(w.todo)-1]
+ w.todo = w.todo[:len(w.todo)-1]
+ w.mu.Unlock()
+
+ w.f(item)
+ }
+}
+
+// Cache runs an action once per key and caches the result.
+type Cache struct {
+ m sync.Map
+}
+
+// Do calls the function f if and only if Do is being called for the first time with this key.
+// No call to Do with a given key returns until the one call to f returns.
+// Do returns the value returned by the one call to f.
+func (c *Cache) Do(key interface{}, f func() interface{}) interface{} {
+ type entry struct {
+ once sync.Once
+ result interface{}
+ }
+
+ entryIface, ok := c.m.Load(key)
+ if !ok {
+ entryIface, _ = c.m.LoadOrStore(key, new(entry))
+ }
+ e := entryIface.(*entry)
+
+ e.once.Do(func() {
+ e.result = f()
+ })
+ return e.result
+}
diff --git a/vendor/cmd/go/internal/par/work_test.go b/vendor/cmd/go/internal/par/work_test.go
new file mode 100644
index 0000000..71c0395
--- /dev/null
+++ b/vendor/cmd/go/internal/par/work_test.go
@@ -0,0 +1,53 @@
+// Copyright 2018 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package par
+
+import (
+ "sync/atomic"
+ "testing"
+)
+
+func TestWork(t *testing.T) {
+ var w Work
+
+ const N = 10000
+ n := int32(0)
+ w.Add(N)
+ w.Do(100, func(x interface{}) {
+ atomic.AddInt32(&n, 1)
+ i := x.(int)
+ if i >= 2 {
+ w.Add(i - 1)
+ w.Add(i - 2)
+ }
+ w.Add(i >> 1)
+ w.Add((i >> 1) ^ 1)
+ })
+ if n != N+1 {
+ t.Fatalf("ran %d items, expected %d", n, N+1)
+ }
+}
+
+func TestCache(t *testing.T) {
+ var cache Cache
+
+ n := 1
+ v := cache.Do(1, func() interface{} { n++; return n })
+ if v != 2 {
+ t.Fatalf("cache.Do(1) did not run f")
+ }
+ v = cache.Do(1, func() interface{} { n++; return n })
+ if v != 2 {
+ t.Fatalf("cache.Do(1) ran f again!")
+ }
+ v = cache.Do(2, func() interface{} { n++; return n })
+ if v != 3 {
+ t.Fatalf("cache.Do(2) did not run f")
+ }
+ v = cache.Do(1, func() interface{} { n++; return n })
+ if v != 2 {
+ t.Fatalf("cache.Do(1) did not returned saved value from original cache.Do(1)")
+ }
+}
diff --git a/vendor/cmd/go/internal/vgo/load.go b/vendor/cmd/go/internal/vgo/load.go
index 25cb0dd..0f69bf0 100644
--- a/vendor/cmd/go/internal/vgo/load.go
+++ b/vendor/cmd/go/internal/vgo/load.go
@@ -23,6 +23,7 @@
"cmd/go/internal/modfile"
"cmd/go/internal/module"
"cmd/go/internal/mvs"
+ "cmd/go/internal/par"
"cmd/go/internal/search"
"cmd/go/internal/semver"
)
@@ -401,6 +402,7 @@
type mvsReqs struct {
extra []module.Version
+ cache par.Cache
}
func newReqs(extra ...module.Version) *mvsReqs {
@@ -411,30 +413,40 @@
}
func (r *mvsReqs) Required(mod module.Version) ([]module.Version, error) {
- list, err := r.required(mod)
- if err != nil {
- return nil, err
+ type cached struct {
+ list []module.Version
+ err error
}
- if *getU {
- for i := range list {
- list[i].Version = "none"
+
+ c := r.cache.Do(mod, func() interface{} {
+ list, err := r.required(mod)
+ if err != nil {
+ return cached{nil, err}
}
- return list, nil
- }
- for i, mv := range list {
- for excluded[mv] {
- mv1, err := r.Next(mv)
- if err != nil {
- return nil, err
+ if *getU {
+ for i := range list {
+ list[i].Version = "none"
}
- if mv1.Version == "" {
- return nil, fmt.Errorf("%s(%s) depends on excluded %s(%s) with no newer version available", mod.Path, mod.Version, mv.Path, mv.Version)
- }
- mv = mv1
+ return cached{list, nil}
}
- list[i] = mv
- }
- return list, nil
+ for i, mv := range list {
+ for excluded[mv] {
+ mv1, err := r.next(mv)
+ if err != nil {
+ return cached{nil, err}
+ }
+ if mv1.Version == "" {
+ return cached{nil, fmt.Errorf("%s(%s) depends on excluded %s(%s) with no newer version available", mod.Path, mod.Version, mv.Path, mv.Version)}
+ }
+ mv = mv1
+ }
+ list[i] = mv
+ }
+
+ return cached{list, nil}
+ }).(cached)
+
+ return c.list, c.err
}
var vgoVersion = []byte(modconv.Prefix)
@@ -595,23 +607,14 @@
return module.Version{Path: path, Version: info.Version}, nil
}
-var versionCache = make(map[string][]string)
-
func versions(path string) ([]string, error) {
- list, ok := versionCache[path]
- if !ok {
- var err error
- repo, err := modfetch.Lookup(path)
- if err != nil {
- return nil, err
- }
- list, err = repo.Versions("")
- if err != nil {
- return nil, err
- }
- versionCache[path] = list
+ // Note: modfetch.Lookup and repo.Versions are cached,
+ // so there's no need for us to add extra caching here.
+ repo, err := modfetch.Lookup(path)
+ if err != nil {
+ return nil, err
}
- return list, nil
+ return repo.Versions("")
}
func (*mvsReqs) Previous(m module.Version) (module.Version, error) {
@@ -626,7 +629,10 @@
return module.Version{Path: m.Path, Version: "none"}, nil
}
-func (*mvsReqs) Next(m module.Version) (module.Version, error) {
+// next returns the next version of m.Path after m.Version.
+// It is only used by the exclusion processing in the Required method,
+// not called directly by MVS.
+func (*mvsReqs) next(m module.Version) (module.Version, error) {
list, err := versions(m.Path)
if err != nil {
return module.Version{}, err
