increase limit by multiplier

Author: patrick-ogrady

storage/modules/balance_storage.go

@@ -23,8 +23,6 @@ import (
 	"runtime"
 	"sync"
 
-	"github.com/neilotoole/errgroup"
-
 	"github.com/coinbase/rosetta-sdk-go/asserter"
 	"github.com/coinbase/rosetta-sdk-go/parser"
 	"github.com/coinbase/rosetta-sdk-go/storage/database"
@@ -198,7 +196,7 @@ func (b *BalanceStorage) AddingBlock(
 	var newAccountsLock sync.Mutex
 
 	// Concurrent execution limited to runtime.NumCPU
-	g, gctx := errgroup.WithContextN(ctx, b.numCPU, b.numCPU)
+	g, gctx := utils.ErrGroupWithContext(ctx, b.numCPU)
 	for i := range changes {
 		// We need to set variable before calling goroutine
 		// to avoid getting an updated pointer as loop iteration
@@ -273,7 +271,7 @@ func (b *BalanceStorage) RemovingBlock(
 	var staleAccountsMutex sync.Mutex
 
 	// Concurrent execution limited to runtime.NumCPU
-	g, gctx := errgroup.WithContextN(ctx, b.numCPU, b.numCPU)
+	g, gctx := utils.ErrGroupWithContext(ctx, b.numCPU)
 	for i := range changes {
 		// We need to set variable before calling goroutine
 		// to avoid getting an updated pointer as loop iteration

storage/modules/block_storage.go

@@ -23,8 +23,6 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/neilotoole/errgroup"
-
 	"github.com/coinbase/rosetta-sdk-go/storage/database"
 	"github.com/coinbase/rosetta-sdk-go/storage/encoder"
 	storageErrs "github.com/coinbase/rosetta-sdk-go/storage/errors"
@@ -252,7 +250,7 @@ func (b *BlockStorage) pruneBlock(
 		blockIdentifier := blockResponse.Block.BlockIdentifier
 
 		// Remove all transaction hashes
-		g, gctx := errgroup.WithContextN(ctx, b.numCPU, b.numCPU)
+		g, gctx := utils.ErrGroupWithContext(ctx, b.numCPU)
 		for i := range blockResponse.OtherTransactions {
 			// We need to set variable before calling goroutine
 			// to avoid getting an updated pointer as loop iteration
@@ -644,7 +642,7 @@ func (b *BlockStorage) AddBlock(
 		return fmt.Errorf("%w: %v", storageErrs.ErrBlockStoreFailed, err)
 	}
 
-	g, gctx := errgroup.WithContextN(ctx, b.numCPU, b.numCPU)
+	g, gctx := utils.ErrGroupWithContext(ctx, b.numCPU)
 	for i := range block.Transactions {
 		// We need to set variable before calling goroutine
 		// to avoid getting an updated pointer as loop iteration
@@ -727,7 +725,7 @@ func (b *BlockStorage) RemoveBlock(
 	}
 
 	// Remove all transaction hashes
-	g, gctx := errgroup.WithContextN(ctx, b.numCPU, b.numCPU)
+	g, gctx := utils.ErrGroupWithContext(ctx, b.numCPU)
 	for i := range block.Transactions {
 		// We need to set variable before calling goroutine
 		// to avoid getting an updated pointer as loop iteration

storage/modules/coin_storage.go

@@ -21,8 +21,6 @@ import (
 	"runtime"
 	"strings"
 
-	"github.com/neilotoole/errgroup"
-
 	"github.com/coinbase/rosetta-sdk-go/asserter"
 	"github.com/coinbase/rosetta-sdk-go/storage/database"
 	"github.com/coinbase/rosetta-sdk-go/storage/errors"
@@ -298,7 +296,7 @@ func (c *CoinStorage) updateCoins( // nolint:gocognit
 		}
 	}
 
-	g, gctx := errgroup.WithContextN(ctx, c.numCPU, c.numCPU)
+	g, gctx := utils.ErrGroupWithContext(ctx, c.numCPU)
 	for identifier, val := range addCoins {
 		if _, ok := removeCoins[identifier]; ok {
 			continue

utils/errgroup.go

@@ -0,0 +1,19 @@
+package utils
+
+import (
+	"context"
+
+	"github.com/neilotoole/errgroup"
+)
+
+const (
+	workerMultiplier = 3
+	queueMultiplier  = 20
+)
+
+// ErrGroupWithContext returns a drop-in replacement for `sync/errgroup` that
+// limits the number of concurrently running goroutines to some function
+// of the number of CPUs.
+func ErrGroupWithContext(ctx context.Context, numCPU int) (*errgroup.Group, context.Context) {
+	return errgroup.WithContextN(ctx, numCPU*workerMultiplier, numCPU*queueMultiplier)
+}