module executor

import context
import sync
import time

// try_post accepts f if the executor is open and queue space is available.
//
// It never waits for queue capacity. A full queue returns
// `executor: queue is full`, making backpressure explicit.
pub fn (mut e Executor) try_post(f JobFn) ! {
    if f == unsafe { nil } {
        return error(err_nil_job)
    }
    e.enqueue_job_now(ExecutorJob{
        f:            f
        report_error: true
    })!
}

// post_with_context waits until f can be admitted, parent is canceled, or the executor closes.
//
// The parent context bounds admission only. Once accepted, f will run on a
// future owner pump unless the application never pumps the executor.
// Active-owner submissions are accepted while capacity is available. If they
// would need to wait for capacity, they fail with
// `executor: owner thread cannot wait for queue capacity`.
pub fn (mut e Executor) post_with_context(parent context.Context, f JobFn) ! {
    if f == unsafe { nil } {
        return error(err_nil_job)
    }
    e.enqueue_job_with_context(parent, ExecutorJob{
        f:            f
        report_error: true
    })!
}

// post_with_timeout waits up to timeout for f to be admitted.
//
// The timeout bounds admission only and does not preempt a running callback.
// Active-owner submissions are accepted while capacity is available. If they
// would need to wait for capacity, they fail with
// `executor: owner thread cannot wait for queue capacity`.
pub fn (mut e Executor) post_with_timeout(timeout time.Duration, f JobFn) ! {
    if f == unsafe { nil } {
        return error(err_nil_job)
    }
    if timeout <= 0 {
        return error(err_timeout)
    }
    e.enqueue_job_with_timeout(time.now().add(timeout), ExecutorJob{
        f:            f
        report_error: true
    })!
}

fn (mut e Executor) enqueue_job_now(job ExecutorJob) ! {
    e.mutex.lock()
    if !e.accepting {
        e.mutex.unlock()
        return error(err_executor_closed)
    }
    if e.queue.len >= e.queue_size {
        e.mutex.unlock()
        return error(err_queue_full)
    }
    e.queue << job
    e.wake_job_ready_locked()
    e.mutex.unlock()
}

fn (mut e Executor) enqueue_job_with_context(parent context.Context, job ExecutorJob) ! {
    thread_id := sync.thread_id()
    mut submit_ctx := parent
    initial_err := submit_ctx.err()
    if initial_err !is none {
        return initial_err
    }
    done := submit_ctx.done()
    mut watch_done := true
    select {
        _ := <-done {
            err := submit_ctx.err()
            if err !is none {
                return err
            }
            watch_done = false
        }
        else {}
    }
    for {
        e.mutex.lock()
        if !e.accepting {
            e.mutex.unlock()
            return error(err_executor_closed)
        }
        if watch_done {
            err := submit_ctx.err()
            if err !is none {
                e.mutex.unlock()
                return err
            }
        }
        if e.queue.len < e.queue_size {
            e.queue << job
            e.wake_job_ready_locked()
            e.mutex.unlock()
            return
        }
        if e.owner_active && e.owner_thread_id == thread_id {
            e.mutex.unlock()
            return error(err_owner_submit_wait)
        }
        submit_ready := e.submit_ready
        e.mutex.unlock()
        if !watch_done {
            _ := <-submit_ready
            continue
        }
        select {
            _ := <-submit_ready {
                continue
            }
            _ := <-done {
                err := submit_ctx.err()
                if err !is none {
                    return err
                }
                watch_done = false
                continue
            }
        }
    }
}

fn (mut e Executor) enqueue_job_with_timeout(deadline time.Time, job ExecutorJob) ! {
    thread_id := sync.thread_id()
    for {
        e.mutex.lock()
        if !e.accepting {
            e.mutex.unlock()
            return error(err_executor_closed)
        }
        remaining := deadline - time.now()
        if remaining <= 0 {
            e.mutex.unlock()
            return error(err_timeout)
        }
        if e.queue.len < e.queue_size {
            e.queue << job
            e.wake_job_ready_locked()
            e.mutex.unlock()
            return
        }
        if e.owner_active && e.owner_thread_id == thread_id {
            e.mutex.unlock()
            return error(err_owner_submit_wait)
        }
        submit_ready := e.submit_ready
        e.mutex.unlock()
        select {
            _ := <-submit_ready {
                continue
            }
            remaining {
                return error(err_timeout)
            }
        }
    }
}