v/vlib/x/executor/executor.v

module executor

import sync

// JobFn is the function signature executed by an Executor on the owner thread.
pub type JobFn = fn () !

// ExecutorConfig configures a bounded owner-thread executor.
@[params]
pub struct ExecutorConfig {
pub:
	queue_size int
}

struct ExecutorJobResult {
mut:
	err IError = none
}

struct ExecutorJob {
	f            fn () ! @[required]
	result_ch    chan ExecutorJobResult
	has_result   bool
	report_error bool
}

// Executor serializes accepted jobs on whichever thread actively pumps it.
//
// The executor does not own a thread. The owner identity is recorded only while
// run(), run_one(), or drain_pending() is actively pumping callbacks.
@[heap]
pub struct Executor {
mut:
	queue_size int
	queue      []ExecutorJob
	active     int

	mutex        &sync.Mutex = sync.new_mutex()
	job_ready    chan int
	submit_ready chan int
	done         chan int

	accepting bool = true
	stopped   bool
	waited    bool

	pumping    bool
	run_active bool

	owner_active    bool
	owner_thread_id u64

	first_err IError = none
}

// new creates an Executor with a fixed bounded queue.
//
// The owner thread is not captured at construction time. It is recorded only
// while an owner pump method is actively running.
pub fn new(config ExecutorConfig) !&Executor {
	if config.queue_size <= 0 {
		return error(err_queue_size_invalid)
	}
	return &Executor{
		queue_size:   config.queue_size
		queue:        []ExecutorJob{cap: config.queue_size}
		mutex:        sync.new_mutex()
		job_ready:    chan int{cap: 1}
		submit_ready: chan int{cap: 1}
		done:         chan int{cap: 1}
		accepting:    true
	}
}

fn (mut e Executor) pop_job_locked() ?ExecutorJob {
	if e.queue.len == 0 {
		return none
	}
	job := e.queue[0]
	e.queue.delete(0)
	e.active++
	e.wake_submitters_locked()
	return job
}

fn (mut e Executor) wake_job_ready_locked() {
	if !e.job_ready.closed {
		e.job_ready.close()
	}
	e.job_ready = chan int{cap: 1}
}

fn (mut e Executor) wake_submitters_locked() {
	if !e.submit_ready.closed {
		e.submit_ready.close()
	}
	if e.accepting {
		e.submit_ready = chan int{cap: 1}
	}
}

fn (mut e Executor) close_admission_locked() {
	if e.accepting {
		e.accepting = false
		e.wake_submitters_locked()
		e.wake_job_ready_locked()
	}
}

fn (mut e Executor) finish_terminal_locked() IError {
	if !e.stopped {
		e.stopped = true
		e.close_admission_locked()
		if !e.done.closed {
			e.done.close()
		}
	}
	return e.first_err
}

fn (mut e Executor) get_first_error() IError {
	e.mutex.lock()
	err := e.first_err
	e.mutex.unlock()
	return err
}

fn (mut e Executor) set_first_error(err IError) {
	e.mutex.lock()
	e.set_first_error_locked(err)
	e.mutex.unlock()
}

fn (mut e Executor) set_first_error_locked(err IError) {
	if e.first_err is none {
		e.first_err = err
	}
	e.close_admission_locked()
}

fn (mut e Executor) execute_job(job ExecutorJob) IError {
	defer {
		e.finish_job()
	}
	first_err_before := e.get_first_error()
	had_first_err := first_err_before !is none
	mut result := ExecutorJobResult{}
	job.f() or { result.err = err }
	if result.err is none && !had_first_err {
		err := e.get_first_error()
		if err !is none {
			result.err = err
		}
	}
	if result.err !is none && job.report_error {
		e.set_first_error(result.err)
	}
	if job.has_result {
		job.result_ch <- result
	}
	return result.err
}

fn (mut e Executor) finish_job() {
	e.mutex.lock()
	if e.active > 0 {
		e.active--
	}
	e.mutex.unlock()
}