root/software/fss-parallel-tools/todo.c @ ed0ec7ca
| 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 "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);
|
|||
}
|