/**
 * Copyright © 2019, Oracle and/or its affiliates. All rights reserved.
 *
 * Licensed under the Universal Permissive License v 1.0 as shown at
 * http://oss.oracle.com/licenses/upl
 *
 */
#include "sched.h"
#include "malloc.h"
#include <assert.h>
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include "common/spew.h"

enum state { RUNNING, SYNCHING, READY };

struct frame {
    enum state state;

    // If READY we are in a list of all the ready and the continuation
    struct frame *next, *prev;
    continuation_t cont;

    // If SYNCING, we have a sync count
    size_t sync_count;

    // If RUNNING there is no additional information.

    struct frame *parent;
    inlet_t       parent_return_inlet;

    void *app_frame;
};

static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  cond  = PTHREAD_COND_INITIALIZER;

static struct frame *ready;
static int all_done;

static void push_ready_frame(struct frame *f) {
    f->next = ready;
    f->prev = NULL;
    if (ready) ready->prev = f;
    ready = f;
    pthread_cond_signal(&cond);
}

static struct frame *pop_ready_frame(void) {
    while (!all_done && !ready) {
        pthread_cond_wait(&cond, &mutex);
    }
    if (ready) {
        assert(!all_done);
        struct frame *result = ready;
        ready->state = RUNNING;
        if (result->next) {
            result->next->prev = NULL;
        }
        ready = result->next;
        return result;
    } else {
        assert(all_done);
        return NULL;
    }
}


struct frame * sched_spawn(struct frame   *parent,
                           continuation_t  parent_cont,
                           void           *sub_app_frame,
                           inlet_t         return_inlet)
{
    pthread_mutex_lock(&mutex);
    struct frame *MALLOC(result);
    spew(SPEW_DEBUG, "frame==%p", result);
    *result = (struct frame){.state               = RUNNING,
                             .next                = NULL,
                             .prev                = NULL,
                             .cont                = NULL,
                             .sync_count          = 0,
                             .parent              = parent,
                             .parent_return_inlet = return_inlet,
                             .app_frame           = sub_app_frame};
    parent->state = READY;

    push_ready_frame(parent);

    parent->cont       = parent_cont;

    parent->sync_count++;

    pthread_mutex_unlock(&mutex);
    return result;
}

struct frame *sched_sync(struct frame *node,
                         continuation_t cont) {
    pthread_mutex_lock(&mutex);
    assert(node->state == RUNNING);
    node->cont = cont;
    struct frame *result;
    if (node->sync_count == 0) {
        result = node;
    } else {
        node->state = SYNCHING;
        result = pop_ready_frame();
    }
    pthread_mutex_unlock(&mutex);
    return result;
}

struct frame *sched_return(struct frame *node, size_t resultint) {
    pthread_mutex_lock(&mutex);
    struct frame *parent = node->parent;
    assert(parent);
    node->parent_return_inlet(parent->app_frame, resultint);
    assert(parent->sync_count);
    parent->sync_count--;

    struct frame *result;
    switch(parent->state) {
        case READY: /* nothing to do */
        case RUNNING:
            result = pop_ready_frame();
            break;
        case SYNCHING:
            if (parent->sync_count == 0) {
                parent->state = RUNNING;
                result = parent;
            } else {
                result = pop_ready_frame();
            }
            break;
        default: abort();
    }
    free(node);
    pthread_mutex_unlock(&mutex);
    return result;
}

static void* worker(void*ignore __attribute__((unused))) {
    pthread_mutex_lock(&mutex);
    struct frame *f = pop_ready_frame();
    pthread_mutex_unlock(&mutex);
    while (f) {
        spew(SPEW_DEBUG, "f=%p cont=%p", f, f->cont);
        assert(f->state == RUNNING);
        f = f->cont(f, f->app_frame);
    }
    pthread_mutex_lock(&mutex);
    assert(all_done);
    pthread_mutex_unlock(&mutex);
    return NULL;
}

static void run_workers(size_t n_workers) {
    pthread_t *MALLOC_N(workers, n_workers);
    for (size_t i = 0; i < n_workers; i++) {
        int r = pthread_create(&workers[i], NULL, worker, NULL);
        assert(r == 0);
    }
    spew(SPEW_DEBUG, "starting workers");
    for (size_t i = 0; i < n_workers; i++) {
        void *nullv = NULL;
        int r = pthread_join(workers[i], nullv);
        assert(r==0);
        assert(nullv == NULL);
    }
    free(workers);
}

magic_t sched_app_frame_magic = "app_frame_magic";
struct sched_app_frame {
    magic_t  *magic;
    continuation_t fun;
    void          *fun_app_frame;
    size_t  have_answer;
    size_t  answer;
};
static void inlet_to_saf (void *safv, size_t r) {
    struct sched_app_frame *saf = safv;
    assert(saf->magic == &sched_app_frame_magic);
    assert(!saf->have_answer);
    saf->have_answer = 1;
    saf->answer = r;
}
static struct frame *prun_wait(struct frame *frame, void *app_frame);
static struct frame *prun_finish(struct frame *frame, void *app_frame);
static struct frame *prun_start(struct frame *frame, void *app_frame) {
    struct sched_app_frame *saf = app_frame;
    spew(SPEW_DEBUG, "%p prun_start inlet_to_saf=%p", frame, inlet_to_saf);
    assert(saf->magic == &sched_app_frame_magic);
    spew(SPEW_DEBUG, "prun_wait=%p", prun_wait);
    struct frame *fib_frame = sched_spawn(frame, prun_wait, saf->fun_app_frame, inlet_to_saf);
    spew(SPEW_DEBUG, "fib_frame=%p, calling %p", fib_frame, saf->fun);
    struct frame *next = saf->fun(fib_frame, saf->fun_app_frame);
    spew(SPEW_DEBUG, "next=%p", next);
    return next;
}
static struct frame *prun_wait(struct frame *frame, void *app_frame __attribute__((unused))) {
    spew(SPEW_DEBUG, "prun_finish=%p", prun_finish);
    return sched_sync(frame, prun_finish);
}
static struct frame *prun_finish(struct frame *frame __attribute__((unused)), void *app_frame __attribute__((unused))) {
    assert(ready == NULL);
    pthread_mutex_lock(&mutex);
    all_done = 1;
    pthread_cond_broadcast(&cond);
    pthread_mutex_unlock(&mutex);
    return NULL;
}
size_t prun(size_t n_workers, continuation_t fun, void *faf) {
    struct sched_app_frame *MALLOC(saf);
    *saf = (struct sched_app_frame){.magic         = &sched_app_frame_magic,
                                    .fun           = fun,
                                    .fun_app_frame = faf,
                                    .have_answer   = 0,
                                    .answer        = 0};
    struct frame *MALLOC(f);
    spew(SPEW_DEBUG, "prun frame = %p", f);
    *f = (struct frame){.state = READY,
                        .next  = NULL,
                        .prev  = NULL,
                        .cont  = prun_start,
                        .sync_count = 0,
                        .parent = NULL,
                        .parent_return_inlet = NULL,
                        .app_frame = saf};
    ready = f;
    all_done = 0;
    spew(SPEW_DEBUG, "calling run_workers number of workers: %ld", n_workers);
    run_workers(n_workers);
    spew(SPEW_DEBUG, "back from run_workers workers");
    assert(saf->have_answer);
    size_t r = saf->answer;
    free(f);
    free(saf);
    return r;
}