fix: bounded parallelization when saving cache (vercel#723)

Fix for vercel#711 (discussion vercel#709).

Bounded cache saving parallelization to `runtime.NumCPU()` which now gives us the maximum number of effective threads while staying under the OS limit of max processes count.

Cheers 🎉

cli/internal/cache/cache_fs.go

@@ -3,13 +3,12 @@ package cache
 import (
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"runtime"
 	"path/filepath"
-
+	"io/ioutil"
 	"github.com/vercel/turborepo/cli/internal/analytics"
 	"github.com/vercel/turborepo/cli/internal/config"
 	"github.com/vercel/turborepo/cli/internal/fs"
-
 	"golang.org/x/sync/errgroup"
 )
 
@@ -67,27 +66,36 @@ func (f *fsCache) logFetch(hit bool, hash string, duration int) {
 
 func (f *fsCache) Put(target, hash string, duration int, files []string) error {
 	g := new(errgroup.Group)
-	for i, file := range files {
-		_, file := i, file // https://golang.org/doc/faq#closures_and_goroutines
-		hash := hash
+
+  numDigesters := runtime.NumCPU()
+	fileQueue := make(chan string, numDigesters)
+
+	for i := 0; i < numDigesters; i++ {
 		g.Go(func() error {
-			rel, err := filepath.Rel(target, file)
-			if err != nil {
-				return fmt.Errorf("error constructing relative path from %v to %v: %w", target, file, err)
-			}
-			if !fs.IsDirectory(file) {
-				if err := fs.EnsureDir(filepath.Join(f.cacheDirectory, hash, rel)); err != nil {
-					return fmt.Errorf("error ensuring directory file from cache: %w", err)
+			for file := range fileQueue {
+				rel, err := filepath.Rel(target, file)
+				if err != nil {
+					return fmt.Errorf("error constructing relative path from %v to %v: %w", target, file, err)
 				}
-
-				if err := fs.CopyOrLinkFile(file, filepath.Join(f.cacheDirectory, hash, rel), fs.DirPermissions, fs.DirPermissions, true, true); err != nil {
-					return fmt.Errorf("error copying file from cache: %w", err)
+				if !fs.IsDirectory(file) {
+					if err := fs.EnsureDir(filepath.Join(f.cacheDirectory, hash, rel)); err != nil {
+						return fmt.Errorf("error ensuring directory file from cache: %w", err)
+					}
+
+					if err := fs.CopyOrLinkFile(file, filepath.Join(f.cacheDirectory, hash, rel), fs.DirPermissions, fs.DirPermissions, true, true); err != nil {
+						return fmt.Errorf("error copying file from cache: %w", err)
+					}
 				}
 			}
 			return nil
 		})
 	}
 
+	for _, file := range files {
+		fileQueue <- file
+	}
+	close(fileQueue)
+
 	if err := g.Wait(); err != nil {
 		return err
 	}