laborers.go

package komi

import (
	"errors"
	"fmt"
	"sync"
)

// Submit sends a job to the pool for processing.
func (p Pool[I, _]) Submit(job I) error {
	if p.IsClosed() {
		return errors.New("can't submit a job to the closed pool")
	}
	p.inputs <- job
	p.jobsWaiting.Add(1)
	return nil
}

// Outputs returns a channel of outputs generated by the pool, nil
// if pool is closed or doesn't produce outputs.
func (p Pool[_, O]) Outputs() (chan O, error) {
	if p.IsClosed() {
		return nil, errors.New("no outputs from a closed pool")
	}
	if !p.producesOutputs() {
		return nil, errors.New("the pool doesn't produce outputs")
	}
	if p.IsConnected() {
		return nil, fmt.Errorf("the pool is connected to a parent %s", p.parent.Name())
	}
	return p.outputs, nil
}

// Errors returns a channel of errors generated by the pool, nil
// if pool is closed, doesn't produce errors, or an error handler is set.
func (p Pool[I, _]) Errors() (chan PoolError[I], error) {
	if p.IsClosed() {
		return nil, errors.New("no errors from a closed pool")
	}
	if !p.producesErrors() {
		return nil, errors.New("the pool doesn't produce errors")
	}
	return p.errors, nil
}

// startLaborers will start laborers.
func (p *Pool[I, O]) startLaborers() {
	// Create the group to wait on.
	p.laborersActive = &sync.WaitGroup{}

	// Create the channel
	p.laborersStopSignal = make(chan Signal)

	// Create the number given by the settings.
	for i := 0; i < p.settings.Laborers; i++ {
		go func(p *Pool[I, O]) {
			for {
				select {
				case job := <-p.inputs:
					// Run the work performer on each new job.
					p.workPerformer(job)
					continue
				case <-p.laborersStopSignal:
					// When stop signal received, mark the laborer as inactive
					// and kill the current scope.
					p.laborersActive.Done()
					return
				}
			}
		}(p)

		// Record the laborer as an active laborer.
		p.laborersActive.Add(1)
	}
	p.log.Debug("Started laborers", "count", p.settings.Laborers)
}

// stopLaborers will send blocking closure signals to all laborers and wait (blocking)
// until they all gracefully leave.
func (p *Pool[_, _]) stopLaborers() {
	p.log.Debug("Sending signals to kill laborers...", "count", p.settings.Laborers)
	for i := 0; i < p.settings.Laborers; i++ {
		p.laborersStopSignal <- signal
	}

	// Wait for all the laborers to quit.
	p.laborersActive.Wait()

	// After laborers quit, close the channel.
	close(p.laborersStopSignal)

	// Log the laborers closure.
	p.log.Debug("All laborers quit", "count", p.settings.Laborers)
}

// Wait wil block until the pool has no waiting jobs, see `With...` options.
func (p Pool[_, _]) Wait() {
	p.currentlyWaitingForJobs.Store(true)
	defer p.currentlyWaitingForJobs.Store(false)
	// If no jobs waiting, leave immediately
	if p.JobsWaiting() == 0 {
		return
	}
	// Wait for the `performedWork` to send a signal
	<-p.noJobsCurrentlyWaitingSignal
}

// JobsCompleted will return the number of jobs completed by the pool.
func (p Pool[_, _]) JobsCompleted() int64 {
	return p.jobsCompleted.Load()
}

// JobsWaiting will return the number of jobs currently waiting and executing
// by the pool.
func (p Pool[_, _]) JobsWaiting() int64 {
	return p.jobsWaiting.Load()
}

// JobsSucceeded will return the number of jobs succeeded (non-nil errors).
func (p Pool[_, _]) JobsSucceeded() int64 {
	return p.jobsSucceeded.Load()
}