/**
 * 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 "todo.h"
#include "malloc.h"
#include <assert.h>
#include <pthread.h>

struct todo_entry {
    struct todo_item *item;
    struct todo_entry *next;
};

struct todo_list{
    pthread_mutex_t    mutex;
    pthread_cond_t     cond;
    struct todo_entry *entries;
    size_t             n_undone; // how many items haven't finished.
};

struct todo_list *
mk_todo_list(void) {
    struct todo_list *MALLOC(tl);
    pthread_mutex_init(&tl->mutex, NULL);
    pthread_cond_init(&tl->cond, NULL);
    tl->entries = NULL;
    tl->n_undone = 0;
    return tl;
}

struct todo_list *
todo_list_destroy(struct todo_list *tl) {
    assert(tl->entries == NULL);
    assert(tl->n_undone == 0);
    pthread_mutex_destroy(&tl->mutex);
    pthread_cond_destroy(&tl->cond);
    FREE(tl);
    return NULL;
}


void
todo_list_push(struct todo_list *tl,
               struct todo_item *item) {
    struct todo_entry *MALLOC(entry);
    pthread_mutex_lock(&tl->mutex);
    *entry = (struct todo_entry){item, tl->entries};
    tl->entries = entry;
    tl->n_undone++;
    pthread_cond_signal(&tl->cond);
    pthread_mutex_unlock(&tl->mutex);
}

static struct todo_item *
todo_list_pop(struct todo_list *tl) {
    pthread_mutex_lock(&tl->mutex);
    while (tl->n_undone > 0 && tl->entries == NULL) {
        pthread_cond_wait(&tl->cond, &tl->mutex);
    }
    struct todo_item *result;
    if (tl->entries) {
        struct todo_entry *entry = tl->entries;
        tl->entries = entry->next;
        result = entry->item;
        FREE(entry);
    } else {
        assert(tl->n_undone == 0);
        result = NULL;
    }
    pthread_mutex_unlock(&tl->mutex);
    return result;
}

static void
todo_list_decrement_undone(struct todo_list *tl) {
    pthread_mutex_lock(&tl->mutex);
    tl->n_undone--;
    if (tl->n_undone == 0) {
        pthread_cond_broadcast(&tl->cond);
    }
    pthread_mutex_unlock(&tl->mutex);
}

typedef const char *magic_t;
magic_t wpair_magic;

struct wpair {
    magic_t *magic;
    size_t threadnum;
    struct todo_list *tl;
    void (*work_on_item_function)(struct todo_list *, struct todo_item *);
};

static void* todo_worker(void *arg) {
    struct wpair *p = arg;
    assert(p->magic == &wpair_magic);
    while (1) {
        struct todo_item *item = todo_list_pop(p->tl);
        if (item == NULL) break;
        p->work_on_item_function(p->tl, item);
        todo_list_decrement_undone(p->tl);
    }
    return arg;
}

void todo_list_run(struct todo_list *tl,
                   size_t n_threads,
                   struct todo_item *first_item,
                   void (*work_on_item_function)(struct todo_list *tl,
                                                 struct todo_item *)) {
    todo_list_push(tl, first_item);
    pthread_t *MALLOC_N(threads, n_threads);
    struct wpair *MALLOC_N(wpairs, n_threads);
    for (size_t i = 0; i < n_threads; i++) {
        wpairs[i] = (struct wpair){&wpair_magic, i, tl, work_on_item_function};
        pthread_create(&threads[i], NULL, todo_worker, &wpairs[i]);
    }
    for (size_t i = 0; i < n_threads; i++) {
        void *retval;
        pthread_join(threads[i], &retval);
        assert(retval == &wpairs[i]);
    }
    assert(tl->entries == NULL);
    assert(tl->n_undone == 0);
    FREE(threads);
    FREE(wpairs);
}