Project

General

Profile

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

root/software/fss-parallel-tools/fwg.c @ 7e8045d8

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 "fwg.h"
#include "malloc.h"
#include "hashfun.h"
#include <assert.h>
#include <pthread.h>
#include <string.h>
#include <stdio.h>

struct nodebag {
size_t n;
size_t size;
struct fwgnode **nodes;
};

static struct nodebag *mk_nodebag() {
const size_t initial_size = 2;
struct nodebag *MALLOC(result);
struct fwgnode **MALLOC_N(nodes, initial_size);
*result = (struct nodebag){0, initial_size, nodes};
return result;
}
static struct nodebag *nodebag_destroy(struct nodebag *nodebag) {
assert(nodebag->n == 0);
FREE(nodebag->nodes);
FREE(nodebag);
return NULL;
}
static void nodebag_push(struct nodebag *nodebag, struct fwgnode *node) {
if (nodebag->n * 2 >= nodebag->size) {
nodebag->size *= 2;
REALLOC(nodebag->nodes, nodebag->size);
}
assert(nodebag->n * 2 < nodebag->size);
nodebag->nodes[nodebag->n++] = node;
}
static int nodebag_is_empty(struct nodebag *nodebag) {
return nodebag->n == 0;
}
static int nodebag_pop_random(struct nodebag *nodebag, struct fwgnode **resultp) {
if (nodebag_is_empty(nodebag)) return 1;
size_t randv = (size_t)(random()) % nodebag->n;
struct fwgnode *result = nodebag->nodes[randv];
nodebag->nodes[randv] = nodebag->nodes[--nodebag->n];
*resultp = result;
return 0;
}
static int nodebag_pop_arbitrary(struct nodebag *nodebag, struct fwgnode **result) {
if (nodebag_is_empty(nodebag)) return 1;
*result = nodebag->nodes[--nodebag->n];
return 0;
}

struct fwgnode {
size_t ready_count; // how many predecessors are
// unfinished.
size_t cost;
char *name;
struct fwobject *fwobject;
struct nodebag *successors;
};

struct nodetable {
size_t n; // how many non-null items in nodes.
size_t size; // a power of two.
struct fwgnode **entries; // an array of length size.
};

static void nodetable_verify(const struct nodetable *n) {
assert(0 == (n->size & (n->size - 1)));
assert(n->n <= n->size);
if (0) {
size_t count = 0;
for (size_t i = 0; i < n->size; i++) {
if (n->entries[i]) count++;
}
assert(count == n->n);
}
}

static struct nodetable *mk_nodetable(void) {
struct nodetable *MALLOC(result);
const size_t initsize = 2;
struct fwgnode **MALLOC_N(nodeps, initsize);
for (size_t i = 0; i < initsize; i++) {
nodeps[i] = NULL;
}
*result = (struct nodetable){0, initsize, nodeps};
nodetable_verify(result);
return result;
}
static struct nodetable *nodetable_destroy(struct nodetable *nodetable) {
nodetable_verify(nodetable);
assert(nodetable->n == 0);
FREE(nodetable->entries);
FREE(nodetable);
return NULL;
}
static int nodetable_is_empty(struct nodetable *nodetable) {
return nodetable->n == 0;
}
static void nodetable_insert(struct nodetable *nodetable,
struct fwgnode *node);
static void nodetable_rehash(struct nodetable *nodetable, size_t new_size) {
const size_t old_size = nodetable->size;
struct fwgnode **old_entries = nodetable->entries;
MALLOC_N(nodetable->entries, new_size);
for (size_t i = 0; i < new_size; i++) {
nodetable->entries[i] = NULL;
}
nodetable->n = 0;
nodetable->size = new_size;
for (size_t i = 0; i < old_size; i++) {
if (old_entries[i]) {
nodetable_insert(nodetable, old_entries[i]);
}
}
FREE(old_entries);
}

static size_t nodetable_find_index(const struct nodetable *nodetable,
const char *name) {
assert(0 == (nodetable->size & (nodetable->size - 1))); // power of 2
size_t hstart = hash_string(name);
for (size_t count = 0; count < nodetable->size; count++) {
size_t hi = (hstart+count) & (nodetable->size -1);
if (nodetable->entries[hi] == NULL) return hi;
if (strcmp(nodetable->entries[hi]->name, name) == 0) {
return hi;
}
}
assert(0); // unreachable
}
static void nodetable_insert(struct nodetable *nodetable,
struct fwgnode *node) {
if (0) printf("Inserting %s, n=%lu size=%lu\n", node->name, nodetable->n, nodetable->size);
nodetable_verify(nodetable);
if (0) printf("Verified\n");
if (nodetable->n * 2 >= nodetable->size) {
if (0) printf("rehashing\n");
nodetable_rehash(nodetable, nodetable->size * 2);
}
nodetable_verify(nodetable);
assert(nodetable->n * 2 < nodetable->size);
size_t hi = nodetable_find_index(nodetable, node->name);
assert(nodetable->entries[hi] == NULL);
nodetable->entries[hi] = node;
nodetable->n++;
if (0) printf("%s:%d n=%lu\n", __FILE__, __LINE__, nodetable->n);
nodetable_verify(nodetable);
if (0) printf("%s:%d ok\n", __FILE__, __LINE__);
}
static void nodetable_reinsert_from(struct nodetable *nodetable,
const size_t start) {
for (size_t count = 0; count < nodetable->size; count++) {
size_t hi = (start + count) & (nodetable->size - 1);
struct fwgnode *node = nodetable->entries[hi];
if (node == NULL) return;
nodetable->entries[hi] = NULL;
nodetable->n--;
nodetable_insert(nodetable, node);
}
}
static void nodetable_remove(struct nodetable *nodetable,
struct fwgnode *node) {
if (0) printf("%s:%d removing\n", __FILE__, __LINE__);
nodetable_verify(nodetable);
size_t hi = nodetable_find_index(nodetable, node->name);
assert(nodetable->entries[hi] == node);
nodetable->n--;
nodetable->entries[hi] = NULL;
nodetable_reinsert_from(nodetable, hi+1);
if (0) printf("%s:%d removed\n", __FILE__, __LINE__);
nodetable_verify(nodetable);
}
static int nodetable_find(struct nodetable *nodetable,
const char *name,
struct fwgnode **result) {
size_t hi = nodetable_find_index(nodetable, name);
if (nodetable->entries[hi] == NULL) return 1;
assert(0 == strcmp(nodetable->entries[hi]->name, name));
*result = nodetable->entries[hi];
return 0;
}

struct fwg {
pthread_mutex_t mutex;
pthread_cond_t wait_for_ready;
pthread_cond_t wait_for_notfull;
int done;
size_t budget;
size_t spent;
struct nodebag *ready_nodes;
struct nodetable *nodes; // A map from names to nodes
};
struct fwg_nodehandle;
struct fwobject; /* Opaque */

static void fwg_verify(const struct fwg *fwg) {
nodetable_verify(fwg->nodes);
}

struct fwg *mk_fwg(size_t budget) {
struct fwg *MALLOC(result);
pthread_mutex_init(&result->mutex, NULL);
pthread_cond_init(&result->wait_for_ready, NULL);
pthread_cond_init(&result->wait_for_notfull, NULL);
result->done = 0;
result->budget = budget;
result->spent = 0;
result->ready_nodes = mk_nodebag();
result->nodes = mk_nodetable();
fwg_verify(result);
return result;
}

struct fwg *fwg_destroy(struct fwg *fwg) {
//printf("%s:%d\n", __FILE__, __LINE__); fwg_verify(fwg);
pthread_mutex_destroy(&fwg->mutex);
pthread_cond_destroy(&fwg->wait_for_ready);
pthread_cond_destroy(&fwg->wait_for_notfull);
assert(nodebag_is_empty(fwg->ready_nodes));
fwg->ready_nodes = nodebag_destroy(fwg->ready_nodes);
if (!nodetable_is_empty(fwg->nodes)) {
printf("Nodetable->n=%ld\n", fwg->nodes->n);
for (size_t i = 0; i < fwg->nodes->size; i++ ){
if (fwg->nodes->entries[i]) {
printf(" %p: %s:\n", fwg->nodes->entries[i], fwg->nodes->entries[i]->name);
}
}
}
assert(nodetable_is_empty(fwg->nodes));
fwg->nodes = nodetable_destroy(fwg->nodes);
FREE(fwg);
return NULL;
}

static void add_dependency(struct fwg *fwg,
const char *prevname,
struct fwgnode *node) {
// NULL or empty previous is not a dependency
if (prevname==NULL || 0 == *prevname) return;
struct fwgnode *prevnode;
if (0) printf("adding dependency from %s to %p\n", prevname, node);
if (nodetable_find(fwg->nodes, prevname, &prevnode) == 0) {
if (0) printf(" Found: adding dependency from %p to %p\n", prevnode, node);
nodebag_push(prevnode->successors, node);
node->ready_count++;
if (0) printf(" Found prev node (ready_count=%lu)\n", node->ready_count);
} else {
if (0) printf(" No prev node (ready_count=%lu)\n", node->ready_count);
}
}

static void note_maybe_node_is_ready(struct fwg *fwg,
struct fwgnode *node) {
if (0) printf("ready_count=%lu\n", node->ready_count);
if (node->ready_count == 0) {
nodebag_push(fwg->ready_nodes, node);
pthread_cond_signal(&fwg->wait_for_ready);
if (0) printf("%s:%d n ready = %lu\n", __FILE__, __LINE__, fwg->ready_nodes->n);
}
}

static void fwg_add_internal(struct fwg *fwg,
const char *name,
struct fwobject *fwobject,
size_t cost,
const char *prev1,
const char *prev2,
const char *prev3) {
struct fwgnode *MALLOC(node);
if (0) printf("Add node %p %s %s %s %s\n", node, name, prev1, prev2, prev3);
pthread_mutex_lock(&fwg->mutex);
//printf("%s:%d\n", __FILE__, __LINE__); fwg_verify(fwg);
*node = (struct fwgnode){0, cost, strdup(name), fwobject, mk_nodebag()};
add_dependency(fwg, name, node);
add_dependency(fwg, prev1, node);
add_dependency(fwg, prev2, node);
add_dependency(fwg, prev3, node);
if (0) printf("%s:%d Inserting\n", __FILE__, __LINE__);
nodetable_insert(fwg->nodes, node);
if (0) printf("%s:%d Inserted\n", __FILE__, __LINE__);
note_maybe_node_is_ready(fwg, node);
fwg->spent += cost;
while (fwg->spent > fwg->budget) {
pthread_cond_wait(&fwg->wait_for_notfull, &fwg->mutex);
}
//printf("%s:%d\n", __FILE__, __LINE__); fwg_verify(fwg);
pthread_mutex_unlock(&fwg->mutex);
}
void fwg_add1(struct fwg *fwg,
const char *name, struct fwobject *fwobject, size_t budget,
const char *prev1) {
fwg_add_internal(fwg, name, fwobject, budget, prev1, NULL, NULL);
}

void fwg_add2(struct fwg *fwg,
const char *name, struct fwobject *fwobject, size_t budget,
const char *prev1, const char *prev2) {
fwg_add_internal(fwg, name, fwobject, budget, prev1, prev2, NULL);
}
void fwg_add3(struct fwg *fwg,
const char *name, struct fwobject *fwobject, size_t budget,
const char *prev1, const char *prev2, const char *prev3) {
fwg_add_internal(fwg, name, fwobject, budget, prev1, prev2, prev3);
}
void fwg_end_of_nodes(struct fwg *fwg) {
if (0) printf("End of nodes\n");
pthread_mutex_lock(&fwg->mutex);
//printf("%s:%d\n", __FILE__, __LINE__); fwg_verify(fwg);
assert(!fwg->done);
fwg->done = 1;
pthread_cond_broadcast(&fwg->wait_for_ready);
//printf("%s:%d\n", __FILE__, __LINE__); fwg_verify(fwg);
pthread_mutex_unlock(&fwg->mutex);
}
int fwg_get_ready_node(struct fwg *fwg,
struct fwobject **fwobject,
struct fwgnode **handle) {
pthread_mutex_lock(&fwg->mutex);
//printf("%s:%d\n", __FILE__, __LINE__); fwg_verify(fwg);
while (!fwg->done && nodebag_is_empty(fwg->ready_nodes)) {
pthread_cond_wait(&fwg->wait_for_ready, &fwg->mutex);
if (0) printf("%s:%d awoke\n", __FILE__, __LINE__);
}
struct fwgnode *result;
int r = nodebag_pop_random(fwg->ready_nodes, &result);
//printf("r=%d\n", r);
if (r != 0) {
assert(fwg->done);
} else {
assert(result->cost <= fwg->spent);
if (fwg->spent > fwg->budget
&& fwg->spent - result->cost <= fwg->budget)
{
pthread_cond_broadcast(&fwg->wait_for_notfull);
}
fwg->spent -= result->cost;
*fwobject = result->fwobject;
*handle = result;
}
//printf("%s:%d\n", __FILE__, __LINE__); fwg_verify(fwg);
pthread_mutex_unlock(&fwg->mutex);
if (0) printf("%s:%d returning %d\n", __FILE__, __LINE__, r);
return r;
}

void fwg_finish_node(struct fwg *fwg, struct fwgnode *handle) {
pthread_mutex_lock(&fwg->mutex);
if (0) printf("Finishing %p which has %lu successors\n", handle, handle->successors->n);
while (1) {
struct fwgnode *succ;
int r = nodebag_pop_arbitrary(handle->successors, &succ);
if (r != 0) break;
if (0) printf(" Maybe wake up %p (pred=%lu)\n", succ, succ->ready_count);
assert(succ->ready_count > 0);
succ->ready_count--;
note_maybe_node_is_ready(fwg, succ);
if (0) printf(" succ->ready_count=%lu\n", succ->ready_count);
}
nodetable_remove(fwg->nodes, handle);
handle->successors = nodebag_destroy(handle->successors);
FREE(handle->name);
FREE(handle);
pthread_mutex_unlock(&fwg->mutex);
}