|
/**
|
|
* 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;
|
|
}
|