Method workerBody

pkg/workerpool/workerpool.go:95–149

(initialWork *contextualItem[T])

}

func (wp *workerPool[T]) workerBody(initialWork *contextualItem[T]) func(context.Context) error {
	worker := func(ctx context.Context) error {
		var timeout bool
		defer func() {
			if timeout {
				return
			}
			atomic.AddInt32(&wp.workers, -1)
			select {
			case <-ctx.Done():
			case <-wp.Done():
			default:
				// Since the task did not finish due to a context cancellation
				// or a timeout, the worker body must have panicked. As such
				// we attempt to spawn a replacement worker in order to avoid
				// stalling the queue indefinitely.
				wp.spawnWorker(nil)
			}
		}()

		defer wp.wg.Done()
		defer registerWorkerStopped(wp.Name)

		registerWorkerIdle(wp.Name)
		defer registerWorkerBusy(wp.Name)

		if initialWork != nil {
			wp.handle(initialWork)
		}

		for {
			select {
			case <-wp.Done():
				return wp.Err()

			case <-ctx.Done():
				return ctx.Err()

			case <-time.After(wp.WorkerIdleTimeout):
				if decrementIfGreaterThan(&wp.workers, int32(wp.MinWorkers)) {
					timeout = true
					return nil
				}

			case item := <-wp.fastQueue:
				wp.handle(item)

			case item := <-wp.mainQueue:
				registerWorkDequeued(wp.Name, item.queuedAt)
				wp.handle(item)
			}
		}
	}
	return worker
}

// spawnWorker spawns a worker task, if a worker slot is available.
func (wp *workerPool[T]) spawnWorker(initialWork *contextualItem[T]) bool {