module executor

import sync

// run blocks on the caller thread and executes accepted jobs serially.
//
// It exits after stop() or the first job error closes admission and all accepted
// jobs have been drained.
pub fn (mut e Executor) run() ! {
	e.begin_owner_pump(true)!
	defer {
		e.end_owner_pump()
	}
	for {
		e.mutex.lock()
		job := e.pop_job_locked() or {
			if !e.accepting {
				err := e.finish_terminal_locked()
				e.mutex.unlock()
				if err !is none {
					return err
				}
				return
			}
			job_ready := e.job_ready
			e.mutex.unlock()
			_ := <-job_ready
			continue
		}
		e.mutex.unlock()
		e.execute_job(job)
	}
}

// run_one executes at most one accepted job and reports whether a job ran.
pub fn (mut e Executor) run_one() !bool {
	e.begin_owner_pump(false)!
	defer {
		e.end_owner_pump()
	}
	e.mutex.lock()
	job := e.pop_job_locked() or {
		if !e.accepting {
			err := e.finish_terminal_locked()
			e.mutex.unlock()
			if err !is none {
				return err
			}
			return false
		}
		e.mutex.unlock()
		return false
	}
	e.mutex.unlock()

	job_err := e.execute_job(job)
	terminal_err := e.finish_if_drained()
	if job_err !is none {
		return job_err
	}
	if terminal_err !is none {
		return terminal_err
	}
	return true
}

// drain_pending executes up to max_jobs jobs that are already accepted.
pub fn (mut e Executor) drain_pending(max_jobs int) !int {
	if max_jobs <= 0 {
		return error(err_drain_limit_invalid)
	}
	e.begin_owner_pump(false)!
	defer {
		e.end_owner_pump()
	}
	mut ran := 0
	mut first_err := IError(none)
	for ran < max_jobs {
		e.mutex.lock()
		job := e.pop_job_locked() or {
			if !e.accepting {
				err := e.finish_terminal_locked()
				e.mutex.unlock()
				if err !is none {
					return err
				}
				if first_err !is none {
					return first_err
				}
				return ran
			}
			e.mutex.unlock()
			if first_err !is none {
				return first_err
			}
			return ran
		}
		e.mutex.unlock()

		job_err := e.execute_job(job)
		ran++
		if job_err !is none && first_err is none {
			first_err = job_err
		}
	}
	terminal_err := e.finish_if_drained()
	if terminal_err !is none {
		return terminal_err
	}
	if first_err !is none {
		return first_err
	}
	return ran
}

// stop closes admission and wakes any blocked submitters or owner run loop.
//
// It is idempotent and non-blocking. Accepted jobs are drained by the owner
// pump; stop() itself never executes user callbacks.
pub fn (mut e Executor) stop() {
	e.mutex.lock()
	e.close_admission_locked()
	if e.queue.len == 0 && e.active == 0 {
		e.finish_terminal_locked()
	}
	e.mutex.unlock()
}

// wait waits for a running owner pump to reach terminal state.
//
// A valid wait is one-shot. Calling it from the owner callback, or before a
// terminal pump is active, returns a stable precondition error without consuming
// that one allowed wait.
pub fn (mut e Executor) wait() ! {
	thread_id := sync.thread_id()
	e.mutex.lock()
	if e.owner_active && e.owner_thread_id == thread_id {
		e.mutex.unlock()
		return error(err_wait_owner_thread)
	}
	if e.waited {
		e.mutex.unlock()
		return error(err_wait_called)
	}
	if e.stopped {
		e.waited = true
		err := e.first_err
		e.mutex.unlock()
		if err !is none {
			return err
		}
		return
	}
	if !e.run_active {
		e.mutex.unlock()
		return error(err_wait_before_terminal)
	}
	e.waited = true
	done := e.done
	e.mutex.unlock()

	_ := <-done
	err := e.get_first_error()
	if err !is none {
		return err
	}
}

fn (mut e Executor) finish_if_drained() IError {
	e.mutex.lock()
	if !e.accepting && e.queue.len == 0 && e.active == 0 {
		err := e.finish_terminal_locked()
		e.mutex.unlock()
		return err
	}
	e.mutex.unlock()
	return none
}