Function Parallelize

private/pkg/thread/thread.go:55–125

Parallelize runs the jobs in parallel. A max of Parallelism jobs will be run at once. Returns the combined error from the jobs.

(ctx context.Context, jobs []func(context.Context) error, options ...ParallelizeOption)

// A max of Parallelism jobs will be run at once.
// Returns the combined error from the jobs.
func Parallelize(ctx context.Context, jobs []func(context.Context) error, options ...ParallelizeOption) error {
	parallelizeOptions := newParallelizeOptions()
	for _, option := range options {
		option(parallelizeOptions)
	}
	switch len(jobs) {
	case 0:
		return nil
	case 1:
		return jobs[0](ctx)
	}
	multiplier := max(parallelizeOptions.multiplier, 1)
	var cancel context.CancelFunc
	if parallelizeOptions.cancelOnFailure {
		ctx, cancel = context.WithCancel(ctx)
		defer cancel()
	}
	semaphoreC := make(chan struct{}, Parallelism()*multiplier)
	var errs []error
	var lock sync.Mutex
	addError := func(err error) {
		lock.Lock()
		errs = append(errs, err)
		lock.Unlock()
	}
	var wg sync.WaitGroup
	var stop bool
	for _, job := range jobs {
		if stop {
			break
		}
		// We always want context cancellation/deadline expiration to take
		// precedence over the semaphore unblocking, but select statements choose
		// among the unblocked non-default cases pseudorandomly. To correctly
		// enforce precedence, use a similar pattern to the check-lock-check
		// pattern common with sync.RWMutex: check the context twice, and only do
		// the semaphore-protected work in the innermost default case.
		select {
		case <-ctx.Done():
			stop = true
			addError(ctx.Err())
		case semaphoreC <- struct{}{}:
			select {
			case <-ctx.Done():
				stop = true
				addError(ctx.Err())
			default:
				job := job
				wg.Go(func() {
					if err := job(ctx); err != nil {
						addError(err)
						if cancel != nil {
							cancel()
						}
					}
					// This will never block.
					<-semaphoreC
				})