Method Process

internal/parallelwork/parallel_work_queue.go:61–95

Process starts N workers, which will be processing elements in the queue until the queue is empty and all workers are idle or until any of the workers returns an error.

(ctx context.Context, workers int)

// Process starts N workers, which will be processing elements in the queue until the queue
// is empty and all workers are idle or until any of the workers returns an error.
func (v *Queue) Process(ctx context.Context, workers int) error {
	defer v.reportProgress(ctx)

	eg, ctx := errgroup.WithContext(ctx)

	for range workers {
		eg.Go(func() error {
			for {
				select {
				case <-ctx.Done():
					// context canceled - some other worker returned an error.
					return ctx.Err()

				default:
					callback := v.dequeue(ctx)
					if callback == nil {
						// no more work, shut down.
						return nil
					}

					err := callback()

					v.completed(ctx)

					if err != nil {
						return err
					}
				}
			}
		})
	}

	//nolint:wrapcheck
	return eg.Wait()
}

func (v *Queue) dequeue(ctx context.Context) CallbackFunc {
	v.monitor.L.Lock()