#include "threaded_waits.h"

void cch_threaded_waits_init(struct cch_threaded_waits *thw)
{
    thw->disabled = false;
    nn_mutex_init(&thw->lock);
    al_array_init(thw->active);
}

static bool wait_range_satisfied(struct cch_backing *backing, struct cch_wait *wait)
{
    if (wait->end < 0 || (backing->size > 0 && wait->end > backing->size)) {
        return true;
    }
    struct cch_range *range;
    al_array_foreach_ptr(backing->available, i, range) {
        if (wait->start >= range->start && range->end >= wait->end) {
            return true;
        }
    }
    return false;
}

bool cch_threaded_wait_for_range(struct cch_threaded_waits *thw, struct cch_backing *backing,
    struct cch_wait *wait)
{
    off_t size = backing->get_size_if_known(backing);

    // Lock before checking wait_range_satisfied() in case the result changes
    // before wait is possibly added to active.
    nn_mutex_lock(&thw->lock);

    backing->lock(backing);
    bool satisfied = size >= 0 && wait_range_satisfied(backing, wait);
    backing->unlock(backing);

    nn_mutex_lock(&wait->lock);
    bool canceled = thw->disabled || wait->disabled;
    if (!canceled && !satisfied) {
        al_array_push(thw->active, wait);
        // Unlock handler to wait.
        nn_mutex_unlock(&thw->lock);
        nn_cond_wait(&wait->cond, &wait->lock);
        // A successful wait will have been removed before signaling.
        if (wait->disabled) {
            canceled = true;
            // Re-lock to remove.
            nn_mutex_lock(&thw->lock);
            al_array_remove(thw->active, wait);
        }
    }
    nn_mutex_unlock(&wait->lock);

    if (canceled || satisfied) {
        // Else, we already unlocked to wait.
        nn_mutex_unlock(&thw->lock);
    }

    return !canceled;
}

void cch_threaded_waits_signal_anys(struct cch_threaded_waits *thw)
{
    nn_mutex_lock(&thw->lock);
    struct cch_wait *wait;
    al_array_foreach_rev(thw->active, i, wait) {
        if (wait->end < 0) {
            // This will be considered a successful wait.
            al_array_remove_at(thw->active, i);
            nn_mutex_lock(&wait->lock);
            nn_cond_signal(&wait->cond);
            nn_mutex_unlock(&wait->lock);
        }
    }
    nn_mutex_unlock(&thw->lock);
}

void cch_threaded_waits_evaluate(struct cch_threaded_waits *thw, struct cch_backing *backing)
{
    nn_mutex_lock(&thw->lock);
    backing->lock(backing);
    struct cch_wait *wait;
    al_array_foreach_rev(thw->active, i, wait) {
        if (wait_range_satisfied(backing, wait)) {
            al_array_remove_at(thw->active, i);
            nn_mutex_lock(&wait->lock);
            nn_cond_signal(&wait->cond);
            nn_mutex_unlock(&wait->lock);
        }
    }
    backing->unlock(backing);
    nn_mutex_unlock(&thw->lock);
}

void cch_threaded_wait_disable(struct cch_wait *wait)
{
    // If this wait is in active, cond_is_waiting() will be true.
    // So, signaling cond while holding the lock here will ensure it's removed before
    // unlocking. This enables synchronization of successive disable() -> enable() calls.
    nn_mutex_lock(&wait->lock);
    wait->disabled = true;
    if (nn_cond_is_waiting(&wait->cond)) {
        nn_cond_signal(&wait->cond);
    }
    nn_mutex_unlock(&wait->lock);
}

void cch_threaded_wait_enable(struct cch_wait *wait)
{
    nn_mutex_lock(&wait->lock);
    wait->disabled = false;
    nn_mutex_unlock(&wait->lock);
}

void cch_threaded_waits_disable_all(struct cch_threaded_waits *thw)
{
    nn_mutex_lock(&thw->lock);

    if (thw->disabled) {
        nn_mutex_unlock(&thw->lock);
        return;
    }

    // Disallow any further waits.
    thw->disabled = true;

    struct cch_wait *wait;
    al_array_foreach_rev(thw->active, i, wait) {
        nn_mutex_lock(&wait->lock);
        wait->disabled = true;
        nn_cond_signal(&wait->cond);
        nn_mutex_unlock(&wait->lock);
    }

    nn_mutex_unlock(&thw->lock);
}

void cch_threaded_waits_close(struct cch_threaded_waits *thw)
{
    nn_mutex_destroy(&thw->lock);
    al_array_free(thw->active);
}