Update thread.ParallelizeWithCancel to thread.ParallelizeWithCancelOnFailure that no longer takes a context.CancelFunc

private/buf/bufgen/generator.go

@@ -295,12 +295,10 @@ func (g *generator) execPlugins(
 	//      out: gen/proto
 	//    - name: insertion-point-writer
 	//      out: gen/proto
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
 	if err := thread.Parallelize(
 		ctx,
 		jobs,
-		thread.ParallelizeWithCancel(cancel),
+		thread.ParallelizeWithCancelOnFailure(),
 	); err != nil {
 		if errs := multierr.Errors(err); len(errs) > 0 {
 			return nil, errs[0]

private/bufpkg/bufprotoplugin/generator.go

@@ -72,9 +72,7 @@ func (g *generator) Generate(
 			)
 		}
 	}
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	if err := thread.Parallelize(ctx, jobs, thread.ParallelizeWithCancel(cancel)); err != nil {
+	if err := thread.Parallelize(ctx, jobs, thread.ParallelizeWithCancelOnFailure()); err != nil {
 		return nil, err
 	}
 	codeGeneratorResponse, err := protopluginResponseWriter.ToCodeGeneratorResponse()

private/pkg/bandeps/checker.go

@@ -144,9 +144,7 @@ func (c *checker) populateState(ctx context.Context, state *state, externalConfi
 			},
 		)
 	}
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	if err := thread.Parallelize(ctx, jobs, thread.ParallelizeWithCancel(cancel)); err != nil {
+	if err := thread.Parallelize(ctx, jobs, thread.ParallelizeWithCancelOnFailure()); err != nil {
 		return err
 	}
 
@@ -161,5 +159,5 @@ func (c *checker) populateState(ctx context.Context, state *state, externalConfi
 			},
 		)
 	}
-	return thread.Parallelize(ctx, jobs, thread.ParallelizeWithCancel(cancel))
+	return thread.Parallelize(ctx, jobs, thread.ParallelizeWithCancelOnFailure())
 }

private/pkg/thread/thread.go

@@ -68,6 +68,11 @@ func Parallelize(ctx context.Context, jobs []func(context.Context) error, option
 	if multiplier < 1 {
 		multiplier = 1
 	}
+	var cancel context.CancelFunc
+	if parallelizeOptions.cancelOnFailure {
+		ctx, cancel = context.WithCancel(ctx)
+		defer cancel()
+	}
 	semaphoreC := make(chan struct{}, Parallelism()*multiplier)
 	var retErr error
 	var wg sync.WaitGroup
@@ -100,8 +105,8 @@ func Parallelize(ctx context.Context, jobs []func(context.Context) error, option
 						lock.Lock()
 						retErr = multierr.Append(retErr, err)
 						lock.Unlock()
-						if parallelizeOptions.cancel != nil {
-							parallelizeOptions.cancel()
+						if cancel != nil {
+							cancel()
 						}
 					}
 					// This will never block.
@@ -129,17 +134,17 @@ func ParallelizeWithMultiplier(multiplier int) ParallelizeOption {
 	}
 }
 
-// ParallelizeWithCancel returns a new ParallelizeOption that will call the
-// given context.CancelFunc if any job fails.
-func ParallelizeWithCancel(cancel context.CancelFunc) ParallelizeOption {
+// ParallelizeWithCancelOnFailure returns a new ParallelizeOption that will attempt
+// to cancel all other jobs via context cancellation if any job fails.
+func ParallelizeWithCancelOnFailure() ParallelizeOption {
 	return func(parallelizeOptions *parallelizeOptions) {
-		parallelizeOptions.cancel = cancel
+		parallelizeOptions.cancelOnFailure = true
 	}
 }
 
 type parallelizeOptions struct {
-	multiplier int
-	cancel     context.CancelFunc
+	multiplier      int
+	cancelOnFailure bool
 }
 
 func newParallelizeOptions() *parallelizeOptions {