Project

General

Profile

Download (7.21 KB) Statistics
| Branch: | Tag: | Revision:

root/software/fss-parallel-tools/sched.c @ b1d8e8c9

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