gps: use golang.org/x/sync/errgroup

internal/gps/manager_test.go

@@ -343,7 +343,7 @@ func TestMgrMethodsFailWithBadPath(t *testing.T) {
 	if _, _, err = sm.GetManifestAndLock(bad, nil, naiveAnalyzer{}); err == nil {
 		t.Error("GetManifestAndLock() did not error on bad input")
 	}
-	if err = sm.ExportProject(bad, nil, ""); err == nil {
+	if err = sm.ExportProject(context.Background(), bad, nil, ""); err == nil {
 		t.Error("ExportProject() did not error on bad input")
 	}
 }
@@ -758,7 +758,7 @@ func TestErrAfterRelease(t *testing.T) {
 		t.Errorf("GetManifestAndLock errored after Release(), but with unexpected error: %T %s", terr, terr.Error())
 	}
 
-	err = sm.ExportProject(id, nil, "")
+	err = sm.ExportProject(context.Background(), id, nil, "")
 	if err == nil {
 		t.Errorf("ExportProject did not error after calling Release()")
 	} else if terr, ok := err.(smIsReleased); !ok {

internal/gps/maybe_source.go

@@ -28,21 +28,29 @@ type maybeSource interface {
 	getURL() string
 }
 
+type errorSlice []error
+
+func (errs *errorSlice) Error() string {
+	var buf bytes.Buffer
+	for _, err := range *errs {
+		fmt.Fprintf(&buf, "\n\t%s", err)
+	}
+	return buf.String()
+}
+
 type maybeSources []maybeSource
 
 func (mbs maybeSources) try(ctx context.Context, cachedir string, c singleSourceCache, superv *supervisor) (source, sourceState, error) {
-	var e sourceFailures
+	var errs errorSlice
 	for _, mb := range mbs {
 		src, state, err := mb.try(ctx, cachedir, c, superv)
 		if err == nil {
 			return src, state, nil
 		}
-		e = append(e, sourceSetupFailure{
-			ident: mb.getURL(),
-			err:   err,
-		})
+		errs = append(errs, errors.Wrapf(err, "failed to set up %q", mb.getURL()))
 	}
-	return nil, 0, e
+
+	return nil, 0, errors.Wrap(&errs, "no valid source could be created")
 }
 
 // This really isn't generally intended to be used - the interface is for
@@ -56,27 +64,6 @@ func (mbs maybeSources) getURL() string {
 	return strings.Join(strslice, "\n")
 }
 
-type sourceSetupFailure struct {
-	ident string
-	err   error
-}
-
-func (e sourceSetupFailure) Error() string {
-	return fmt.Sprintf("failed to set up %q, error %s", e.ident, e.err.Error())
-}
-
-type sourceFailures []sourceSetupFailure
-
-func (sf sourceFailures) Error() string {
-	var buf bytes.Buffer
-	fmt.Fprintf(&buf, "no valid source could be created:")
-	for _, e := range sf {
-		fmt.Fprintf(&buf, "\n\t%s", e.Error())
-	}
-
-	return buf.String()
-}
-
 // sourceCachePath returns a url-sanitized source cache dir path.
 func sourceCachePath(cacheDir, sourceURL string) string {
 	return filepath.Join(cacheDir, "sources", sanitizer.Replace(sourceURL))

internal/gps/result.go

@@ -5,13 +5,15 @@
 package gps
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"os"
 	"path/filepath"
 	"sync"
 
 	"github.com/pkg/errors"
+	"golang.org/x/sync/errgroup"
 )
 
 // A Solution is returned by a solver run. It is mostly just a Lock, with some
@@ -62,103 +64,76 @@ func WriteDepTree(basedir string, l Lock, sm SourceManager, sv bool, logger *log
 		return fmt.Errorf("must provide non-nil Lock to WriteDepTree")
 	}
 
-	err := os.MkdirAll(basedir, 0777)
-	if err != nil {
+	if err := os.MkdirAll(basedir, 0777); err != nil {
 		return err
 	}
 
+	g, ctx := errgroup.WithContext(context.TODO())
 	lps := l.Projects()
-
-	type resp struct {
-		i   int
-		err error
+	sem := make(chan struct{}, concurrentWriters)
+	var cnt struct {
+		sync.Mutex
+		i int
 	}
-	respCh := make(chan resp, len(lps))
-	writeCh := make(chan int, len(lps))
-	cancel := make(chan struct{})
 
-	// Queue work.
 	for i := range lps {
-		writeCh <- i
-	}
-	close(writeCh)
-	// Launch writers.
-	writers := concurrentWriters
-	if len(lps) < writers {
-		writers = len(lps)
-	}
-	var wg sync.WaitGroup
-	wg.Add(writers)
-	for i := 0; i < writers; i++ {
-		go func() {
-			defer wg.Done()
+		p := lps[i] // per-iteration copy
 
-			for i := range writeCh {
+		g.Go(func() error {
+			err := func() error {
 				select {
-				case <-cancel:
-					return
-				default:
+				case sem <- struct{}{}:
+					defer func() { <-sem }()
+				case <-ctx.Done():
+					return ctx.Err()
 				}
 
-				p := lps[i]
-				to := filepath.FromSlash(filepath.Join(basedir, string(p.Ident().ProjectRoot)))
+				ident := p.Ident()
+				projectRoot := string(ident.ProjectRoot)
+				to := filepath.FromSlash(filepath.Join(basedir, projectRoot))
 
-				if err := sm.ExportProject(p.Ident(), p.Version(), to); err != nil {
-					respCh <- resp{i, errors.Wrapf(err, "failed to export %s", p.Ident().ProjectRoot)}
-					continue
+				if err := sm.ExportProject(ctx, ident, p.Version(), to); err != nil {
+					return errors.Wrapf(err, "failed to export %s", projectRoot)
 				}
 
 				if sv {
-					select {
-					case <-cancel:
-						return
-					default:
+					if err := ctx.Err(); err != nil {
+						return err
 					}
 
 					if err := filepath.Walk(to, stripVendor); err != nil {
-						respCh <- resp{i, errors.Wrapf(err, "failed to strip vendor from %s", p.Ident().ProjectRoot)}
-						continue
+						return errors.Wrapf(err, "failed to strip vendor from %s", projectRoot)
 					}
 				}
 
-				respCh <- resp{i, nil}
-			}
-		}()
-	}
-	// Monitor writers
-	go func() {
-		wg.Wait()
-		close(respCh)
-	}()
-
-	// Log results and collect errors
-	var errs []error
-	var cnt int
-	for resp := range respCh {
-		cnt++
-		msg := "Wrote"
-		if resp.err != nil {
-			if len(errs) == 0 {
-				close(cancel)
+				return nil
+			}()
+
+			switch err {
+			case context.Canceled, context.DeadlineExceeded:
+				// Don't log "secondary" errors.
+			default:
+				msg := "Wrote"
+				if err != nil {
+					msg = "Failed to write"
+				}
+
+				// Log and increment atomically to prevent re-ordering.
+				cnt.Lock()
+				cnt.i++
+				logger.Printf("(%d/%d) %s %s@%s\n", cnt.i, len(lps), msg, p.Ident(), p.Version())
+				cnt.Unlock()
 			}
-			errs = append(errs, resp.err)
-			msg = "Failed to write"
-		}
-		p := lps[resp.i]
-		logger.Printf("(%d/%d) %s %s@%s\n", cnt, len(lps), msg, p.Ident(), p.Version())
-	}
 
-	if len(errs) > 0 {
-		logger.Println("Failed to write dep tree. The following errors occurred:")
-		for i, err := range errs {
-			logger.Printf("(%d/%d) %s\n", i+1, len(errs), err)
-		}
+			return err
+		})
+	}
 
+	err := g.Wait()
+	if err != nil {
 		os.RemoveAll(basedir)
-
-		return errors.New("failed to write dep tree")
 	}
-	return nil
+	return errors.Wrap(err, "failed to write dep tree")
 }
 
 func (r solution) Projects() []LockedProject {

internal/gps/solve_basic_test.go

@@ -5,6 +5,7 @@
 package gps
 
 import (
+	"context"
 	"fmt"
 	"regexp"
 	"strings"
@@ -1460,7 +1461,7 @@ func (sm *depspecSourceManager) SyncSourceFor(id ProjectIdentifier) error {
 
 func (sm *depspecSourceManager) Release() {}
 
-func (sm *depspecSourceManager) ExportProject(id ProjectIdentifier, v Version, to string) error {
+func (sm *depspecSourceManager) ExportProject(context.Context, ProjectIdentifier, Version, string) error {
 	return fmt.Errorf("dummy sm doesn't support exporting")
 }
 

internal/gps/source_manager.go

@@ -65,7 +65,7 @@ type SourceManager interface {
 
 	// ExportProject writes out the tree of the provided import path, at the
 	// provided version, to the provided directory.
-	ExportProject(ProjectIdentifier, Version, string) error
+	ExportProject(context.Context, ProjectIdentifier, Version, string) error
 
 	// DeduceRootProject takes an import path and deduces the corresponding
 	// project/source root.
@@ -460,17 +460,17 @@ func (sm *SourceMgr) SyncSourceFor(id ProjectIdentifier) error {
 
 // ExportProject writes out the tree of the provided ProjectIdentifier's
 // ProjectRoot, at the provided version, to the provided directory.
-func (sm *SourceMgr) ExportProject(id ProjectIdentifier, v Version, to string) error {
+func (sm *SourceMgr) ExportProject(ctx context.Context, id ProjectIdentifier, v Version, to string) error {
 	if atomic.LoadInt32(&sm.releasing) == 1 {
 		return smIsReleased{}
 	}
 
-	srcg, err := sm.srcCoord.getSourceGatewayFor(context.TODO(), id)
+	srcg, err := sm.srcCoord.getSourceGatewayFor(ctx, id)
 	if err != nil {
 		return err
 	}
 
-	return srcg.exportVersionTo(context.TODO(), v, to)
+	return srcg.exportVersionTo(ctx, v, to)
 }
 
 // DeduceProjectRoot takes an import path and deduces the corresponding