May 2010 - Triton-commits - mailman3.cels.anl.gov

Triton Repository branch, master, updated. 0bd8ff1c5443255bea44a2999a820de4862a1a4d
by noreply＠mcs.anl.gov 30 May '10

30 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 0bd8ff1c5443255bea44a2999a820de4862a1a4d (commit) via c81bb947d028f75977755eac89b39bb459f34f02 (commit) via 4483a71bc997fcd5dcd29976e8a8f105b076c7fe (commit) via 20e8346db0c0e82af16e233d86a8d9bef1df1112 (commit) via 0e093e62cfc8521b072e588a22a07f6971145771 (commit) via 2e5462317908cb165107dbbd68dbb7d95230892a (commit) from ac8dc9e40ca010f6d1a3c5b8a17a7639430774f0 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 0bd8ff1c5443255bea44a2999a820de4862a1a4d Author: Justin Wozniak <wozniak(a)mcs.anl.gov> Date: Sun May 30 19:36:03 2010 -0500 New fakess API and test commit c81bb947d028f75977755eac89b39bb459f34f02 Author: Justin Wozniak <wozniak(a)mcs.anl.gov> Date: Sun May 30 19:35:03 2010 -0500 Operator cleanup commit 4483a71bc997fcd5dcd29976e8a8f105b076c7fe Author: Justin Wozniak <wozniak(a)mcs.anl.gov> Date: Sun May 30 19:34:29 2010 -0500 Debuggability fix commit 20e8346db0c0e82af16e233d86a8d9bef1df1112 Author: Justin Wozniak <wozniak(a)mcs.anl.gov> Date: Thu May 27 23:34:53 2010 -0500 Draft of queue-based fakess functionality commit 0e093e62cfc8521b072e588a22a07f6971145771 Author: Justin Wozniak <wozniak(a)mcs.anl.gov> Date: Wed May 19 14:12:57 2010 -0500 Rename list-oriented wait() to waitany() commit 2e5462317908cb165107dbbd68dbb7d95230892a Author: Justin Wozniak <wozniak(a)mcs.anl.gov> Date: Wed May 19 14:09:00 2010 -0500 Clean out things we will probably not use ----------------------------------------------------------------------- Summary of changes: code/src/common/triton-hash.h | 4 +- code/src/fakess/bqueue.ae | 32 ++-- code/src/fakess/fakess.ae | 473 ++++++++++++++++--------------------- code/src/fakess/fakess.hae | 88 ++------ code/src/fakess/module.mk.in | 2 +- code/src/fakess/test/module.mk.in | 6 +- code/src/fakess/test/wait.ae | 2 +- code/src/net/triton-node.h | 2 +- 8 files changed, 247 insertions(+), 362 deletions(-) Diff of changes: diff --git a/code/src/common/triton-hash.h b/code/src/common/triton-hash.h index b8dde48..7721cd3 100644 --- a/code/src/common/triton-hash.h +++ b/code/src/common/triton-hash.h @@ -301,10 +301,12 @@ static inline int triton_hash_string_hash(void *k, int table_size) { uint32_t pc = 0, pb = 0; triton_string_t *s; + int result = -1; s = (triton_string_t *)k; assert(s->string); bj_hashlittle2(s->string, s->size, &pc, &pb); - return pc & (table_size - 1); + result = pc & (table_size - 1); + return result; } #endif /* TRITON_HASH_H */ diff --git a/code/src/fakess/bqueue.ae b/code/src/fakess/bqueue.ae index 52ef5b6..dea1c15 100644 --- a/code/src/fakess/bqueue.ae +++ b/code/src/fakess/bqueue.ae @@ -3,39 +3,37 @@ void triton_bqueue_init(triton_bqueue_t *bq) { - triton_list_init(&(bq->list)); - triton_mutex_init(&(bq->mutex), NULL); - triton_sched_reset(&(bq->sched)); + triton_list_init(bq->list); + triton_mutex_init(bq->mutex, NULL); + triton_sched_reset(bq->sched); } triton_ret_t triton_bqueue_enqueue(triton_list_link_t *link, triton_bqueue_t *queue) { - triton_mutex_lock(&(queue->mutex)); - triton_list_add_back(link, &(queue->list)); - triton_mutex_unlock(&(queue->mutex)); - triton_sched_notify(&(queue->sched), - TRITON_SCHED_ACTION_NOTIFY_ALL); + triton_mutex_lock(&queue->mutex); + triton_list_add_back(link, &queue->list); + triton_mutex_unlock(&queue->mutex); + triton_sched_notify(queue->sched); return TRITON_SUCCESS; } triton_ret_t triton_bqueue_push(triton_list_link_t *link, triton_bqueue_t *queue) { - triton_mutex_lock(&(queue->mutex)); - triton_list_add_front(link, &(queue->list)); - triton_mutex_unlock(&(queue->mutex)); - triton_sched_notify(&(queue->sched), - TRITON_SCHED_ACTION_NOTIFY_ALL); + triton_mutex_lock(&queue->mutex); + triton_list_add_front(link, &queue->list); + triton_mutex_unlock(&queue->mutex); + triton_sched_notify(queue->sched); return TRITON_SUCCESS; } triton_list_link_t *_triton_bqueue_dequeue(triton_bqueue_t *queue) { triton_list_link_t *result = NULL; - triton_mutex_lock(&(queue->mutex)); - result = triton_queue_dequeue(&(queue->list)); - triton_mutex_unlock(&(queue->mutex)); + triton_mutex_lock(&queue->mutex); + result = triton_queue_dequeue(&queue->list); + triton_mutex_unlock(&queue->mutex); return result; } @@ -45,7 +43,7 @@ __blocking triton_list_link_t *triton_bqueue_dequeue(triton_bqueue_t *queue) while (! (result = _triton_bqueue_dequeue(queue))) { - triton_sched_wait(&(queue->sched)); + triton_sched_wait(&queue->sched); } return result; diff --git a/code/src/fakess/fakess.ae b/code/src/fakess/fakess.ae index 8239fbe..4b2baaa 100644 --- a/code/src/fakess/fakess.ae +++ b/code/src/fakess/fakess.ae @@ -1,397 +1,326 @@ /** * Triton fake system state + * TODO: synchronize updates and queries */ #include "mpi.h" +#include "src/fakess/bqueue.hae" #include "src/fakess/fakess.hae" #include "src/net/mpi/mpi.hae" #include "src/common/resources/scheduling/sched.hae" -typedef struct triton_hash_table triton_state_table; +#include "src/common/triton-hash.h" #define TRITON_FAKESS_TOKEN 64 -struct triton_fakess_s -{ - triton_state_table *globals; - - /** - Map from triton-nodes to state_tables - */ - struct triton_hash_table *nodes; -}; - -struct triton_fakess_s state; - -triton_list_link_t empty_link = {NULL, NULL, NULL}; +/** + Map from triton-nodes to state_tables +*/ +struct triton_hash_table *fakess_nodes; /** - List of waiters + Table of waiters */ -triton_list_t *waiters; +struct triton_hash_table *waiters; /** Uniquify each call to wait(). TODO: Protect with mutex? */ -int unique_component = 0; +int unique_status = 0; -typedef struct +struct triton_state_table { triton_node_t node; - triton_string_t *key; - triton_sched_t *sched; - triton_list_t *workq; - int component; + struct triton_hash_table *table; triton_list_link_t link; -} waiter; +}; -static int string_entry_compare(void *key, struct triton_hash_link *link) +/** + The data items stored in the fakess state tables. +*/ +typedef struct { - triton_string_t *value1 = (triton_string_t *)key; - triton_string_entry *value2 = - triton_hash_get_entry(link, triton_string_entry, link); - assert(key); - assert(link); + triton_string_t key; + void *value; + triton_mutex_t mutex; + triton_sched_t sched; + triton_list_t observers; + struct triton_hash_link link; +} fakess_data_t; - return (!strcmp(value2->key->string, value1->string)); -} +typedef struct +{ + // f(); + triton_list_link_t link; +} fakess_observer_t; static int node_compare(void *key, struct triton_hash_link *link) { triton_node_t *value1 = (triton_node_t *)key; - triton_node_entry *value2 = - triton_hash_get_entry(link, triton_node_entry, link); + struct triton_state_table *value2 = + triton_hash_get_entry(link, struct triton_state_table, link); assert(key); assert(link); - return (value2->node.l == value1->l && - value2->node.u == value1->u); + return triton_node_eq(value2->node, *value1); } -__blocking triton_ret_t triton_fakess_init(void) +static int node_hash(void *k, int table_size) { - state.globals = triton_hash_init(string_entry_compare, - triton_hash_string_hash, - 113); - state.nodes = triton_hash_init(node_compare, - triton_hash_128bit_hash, - 23); + struct triton_state_table *t = (struct triton_state_table *)k; + int result; - waiters = malloc(sizeof(triton_list_t)); - triton_list_init(waiters); + result = triton_hash_128bit_hash(&t->node, table_size); + return result; +} - return TRITON_SUCCESS; +static int data_compare(void *key, struct triton_hash_link *link) +{ + fakess_data_t *value1 = (fakess_data_t *)key; + fakess_data_t *value2 = + triton_hash_get_entry(link, fakess_data_t, link); + int result; + assert(key); + assert(link); + assert(value2); + + result = (!strcmp(value1->key.string, value2->key.string)); + return result; } -/** - Register on the global key. - @return TRITON_ERR_NOENT if the key is not in globals. -*/ -triton_ret_t triton_fakess_listen_global(triton_string_t *key, - triton_listener_t listener) +static int data_hash(void *k, int table_size) { - triton_list_link_t *link; - triton_string_entry *entry; + fakess_data_t *data = (fakess_data_t *)k; + int result; - printf("register listener: %s\n", key->string); + result = triton_hash_string_hash(&data->key, table_size); + return result; +} - link = triton_hash_search(state.globals, key); +triton_ret_t triton_fakess_node_add(triton_node_t node); - if (!link) - { - printf("triton_fakess_listen_global(): global not found: %s\n", - key->string); - return TRITON_ERR_NOENT; - } +triton_ret_t triton_fakess_init(void) +{ + triton_ret_t result; + fakess_nodes = triton_hash_init(node_compare, + node_hash, + 23); + result = triton_fakess_node_add(TRITON_NODE_NULL); - entry = triton_list_get_entry(link, triton_string_entry, link); + return result; +} - triton_noter_listen(entry->noter, listener); +fakess_data_t *lookup(triton_node_t node, + triton_string_t *key); - return TRITON_SUCCESS; +void *triton_fakess_lock(triton_node_t node, + triton_string_t *key) +{ + fakess_data_t *data; + void *result; + + data = lookup(node, key); + triton_mutex_lock(&data->mutex); + result = data->value; + + return result; } -triton_sched_action_t fakess_trigger(struct triton_sched *sched, - triton_sched_trigger_event_t op) +void triton_fakess_unlock(triton_node_t node, + triton_string_t *key) { - return TRITON_SCHED_ACTION_NOTIFY_ALL; + fakess_data_t *data; + void *result; + + data = lookup(node, key); + triton_mutex_unlock(&data->mutex); } /** - @param items List of triton_fakess_items to wait on - @param workq List of work returned by this function - @param component Uniquely identify this call */ -__blocking triton_ret_t triton_fakess_wait(triton_list_t *items, - triton_list_t *workq, - int *component) +__blocking void triton_fakess_wait(triton_node_t node, + triton_string_t *key) { - triton_sched_t *sched = malloc(sizeof(triton_sched_t)); - triton_fakess_item *item, *s; - int c; - triton_sched_reset(sched); - - c = unique_component++; - triton_list_for_each_entry(item, s, items, link) - { - waiter *w = malloc(sizeof(waiter)); - - w->node = item->node; - w->key = item->key; - w->sched = sched; - w->workq = workq; - w->component = c; - // triton_sched_set_trigger(sched, fakess_trigger); - triton_list_link_clear(&(w->link)); - triton_list_add_back(&(w->link), waiters); - } - - triton_sched_wait(sched); - - *component = c; - return TRITON_SUCCESS; + fakess_data_t *data; + triton_fakess_lock(node, key); + data = lookup(node, key); + triton_sched_wait(data->sched, &data->mutex); + triton_fakess_unlock(node, key); } -void triton_fakess_notify(triton_node_t node, - triton_string_t *key) -{ - waiter *w, *s; - triton_fakess_item *result; - - triton_list_for_each_entry(w, s, waiters, link) - { - if (triton_node_eq(w->node, node) && - ! strcmp(w->key->string, key->string)) - { - printf("matched!\n"); - result = malloc(sizeof(triton_fakess_item)); - result->node = w->node; - result->key = w->key; - triton_list_link_clear(&(result->link)); - triton_queue_enqueue(&(result->link), w->workq); - triton_sched_notify_all(w->sched); - } - } -} +struct triton_state_table *lookup_node(triton_node_t node); +fakess_data_t *lookup_data(struct triton_state_table *state, + triton_string_t *key); -__blocking triton_ret_t triton_fakess_finalize(void) +fakess_data_t *lookup(triton_node_t node, + triton_string_t *key) { - return TRITON_SUCCESS; + struct triton_state_table *state; + fakess_data_t *data; + state = lookup_node(node); + data = lookup_data(state, key); + return data; } -__blocking triton_ret_t triton_fakess_node_count(int *count) +struct triton_state_table *lookup_node(triton_node_t node) { - *count = state.nodes->table_count; - - return TRITON_SUCCESS; + struct triton_state_table *result = NULL; + struct triton_hash_link *link; + link = triton_hash_search(fakess_nodes, &node); + if (link) + result = triton_hash_get_entry(link, + struct triton_state_table, + link); + return result; } -__blocking triton_ret_t triton_fakess_node_add(triton_node_t *node) +fakess_data_t *lookup_data(struct triton_state_table *state, + triton_string_t *key) { - - return TRITON_SUCCESS; + fakess_data_t *result = NULL; + struct triton_hash_link *link; + link = triton_hash_search(state->table, key); + if (link) + result = triton_hash_get_entry(link, fakess_data_t, link); + return result; } /** - @return The new entry for convenience. + @return TRITON_ERR_EXIST if the key already exists. */ -triton_string_entry *triton_fakess_store(triton_string_t *key, - triton_string_t *value) +triton_ret_t triton_fakess_node_add(triton_node_t node) { + struct triton_state_table *state; struct triton_hash_link *link; - triton_string_entry *entry; - printf("storing global item: %s=%s\n", key->string, value->string); + link = triton_hash_search(fakess_nodes, &node); + if (link) + return TRITON_ERR_EXIST; - entry = malloc(sizeof(triton_string_entry)); - entry->key = key; - entry->value = value; - entry->noter = triton_noter_init(entry); - entry->link = empty_link; + state = malloc(sizeof(struct triton_state_table)); + state->node = node; + state->table = triton_hash_init(data_compare, + data_hash, + 23); + triton_list_link_clear(&state->link); - triton_hash_add(state.globals, key, &(entry->link)); + triton_hash_add(fakess_nodes, &node, &state->link); - return entry; + return TRITON_SUCCESS; } /** - @return TRITON_ERR_EXIST if the key already exists. + Assist triton_fakess_item_add(). Copies key. + @return The new data entry for convenience. */ -triton_ret_t triton_fakess_global_item_add(triton_string_t *key, - triton_string_t *value) +fakess_data_t *triton_fakess_item_store(struct triton_state_table *state, + triton_string_t *key) { struct triton_hash_link *link; + fakess_data_t *data; - printf("adding global item: %s=%s\n", key->string, value->string); - - link = triton_hash_search(state.globals, key); - if (link) - return TRITON_ERR_EXIST; + data = malloc(sizeof(fakess_data_t)); + triton_string_init(&data->key, "%s", key->string); + data->value = TRITON_FAKESS_UNSET; + triton_list_init(&data->observers); + triton_sched_init(&data->sched); + triton_mutex_init(&data->mutex, NULL); + triton_list_link_clear(&data->link); - triton_fakess_store(key, value); + triton_hash_add(state->table, key, &data->link); - return TRITON_SUCCESS; + return data; } /** + Makes an internal copy of given key. @return TRITON_ERR_EXIST if the key already exists. */ triton_ret_t triton_fakess_item_add(triton_node_t node, - triton_string_t *key, - triton_string_t *value) + triton_string_t *key) { - struct triton_hash_link *link; + struct triton_state_table *state; + fakess_data_t *data; - printf("adding global item: %s=%s\n", key->string, value->string); + printf("adding item: (%li.%li) %s\n", + node.u, node.l, key->string); - link = triton_hash_search(state.globals, key); - if (link) + state = lookup_node(node); + assert(state); + + data = lookup_data(state, key); + if (data) return TRITON_ERR_EXIST; - triton_fakess_store(key, value); + triton_fakess_item_store(state, key); return TRITON_SUCCESS; } -__blocking triton_ret_t triton_fakess_post_global(triton_string_t *key, - triton_string_t *value) -{ - int i,rank,nodes; - int length; - char *buffer; - int error; - - buffer = malloc(TRITON_FAKESS_SIZE_MESSAGE); - length = sprintf(buffer, "GLOBAL: %s %s", - key->string, value->string); - printf("triton_fakess_post_global: %s\n", buffer); - - MPI_Comm_rank(MPI_COMM_WORLD, &rank); - MPI_Comm_size(MPI_COMM_WORLD, &nodes); - for(i = 0; i < nodes; i++) - { - if(i != rank) - { - printf("send to: %i %i\n", i, length); - error = triton_mpi_send(buffer, length+1, MPI_CHAR, i, - TRITON_TAG_FAKESS, MPI_COMM_WORLD); - printf("sent\n"); - if(error != 0) - { - return TRITON_ERR_MPI; - } - } - } - return TRITON_SUCCESS; -} - /** - Parse buffer to obtain array of pointers to triton_string_t's + Assist triton_fakess_set() */ -triton_string_t **triton_fakess_strings(char *buffer, - int *lengths, - int max) +void update(fakess_data_t *data, void *value) { - int i, j, k; - char *chars; - triton_string_t **results; - - results = malloc(max*sizeof(triton_string_t*)); - - i = 0; /* index into source */ - k = 0; - while(k < max) - { - chars = malloc(lengths[k]*sizeof(char)); - - j = 0; /* index into sink */ - while(buffer[i] != ' ' && - buffer[i] != '\0') - { - chars[j++] = buffer[i]; - i++; - } - chars[j] = '\0'; - - results[k] = malloc(sizeof(triton_string_t)); - triton_string_init_inplace_size(results[k], chars, j); - k++; - i++; - } - for ( ; k < max; k++) - { - results[k] = NULL; - } - return results; + triton_mutex_lock(&data->mutex); + + if (data->value && + data->value != TRITON_FAKESS_UNSET) + free(value); + data->value = value; + + triton_mutex_unlock(&data->mutex); } /** - Update the local system state structure for globals. + Frees the old value if non-NULL */ -triton_ret_t triton_fakess_local_update_global(triton_string_t *key, - triton_string_t *value) +triton_ret_t triton_fakess_set(triton_node_t node, + triton_string_t *key, + void *value) { - struct triton_hash_link *link; - triton_string_entry *entry; - - printf("setting global: %s=%s\n", - key->string, value->string); + fakess_data_t *data; - link = triton_hash_search(state.globals, key); + printf("triton_fakess_set(): (%li.%li) %s\n", + node.u, node.l, key->string); - if (link == NULL) - { - printf("was not found: %s\n", key->string); - return TRITON_ERR_NOENT; - } + data = lookup(node, key); + update(data, value); - entry = triton_hash_get_entry(link, triton_string_entry, link); - entry->value = value; + triton_sched_notify_all(data->sched); - triton_notify(entry->noter, NULL); + /* + // Call listeners: + triton_list_for_each_entry() + {} + */ + return TRITON_SUCCESS; } /** - Scan in a global key/value pair from MPI and do the local update. + Returns aa locked triton_fakess_result_t */ -void triton_fakess_notify_global(char *buffer) +triton_fakess_result_t *triton_fakess_get(triton_node_t node, + triton_string_t *key) { - int lengths[3] = {TRITON_FAKESS_TOKEN, - TRITON_FAKESS_SIZE_KEY, - TRITON_FAKESS_SIZE_VALUE}; - triton_string_t **results; - results = triton_fakess_strings(buffer, lengths, 3); - triton_fakess_local_update_global(results[1], results[2]); - /* destroy results[0] */ + triton_fakess_result_t *result = + malloc(sizeof(triton_fakess_result_t)); + fakess_data_t *data; + + data = lookup(node, key); + + triton_mutex_lock(&data->mutex); + result->value = data->value; + result->mutex = &data->mutex; + + return result; } -__blocking triton_ret_t triton_fakess_recv(void) +triton_ret_t triton_fakess_finalize(void) { - char recvbuf[TRITON_FAKESS_SIZE_MESSAGE]; - MPI_Status status; - int recvsize; - int ret; - - printf("%d: receiving message from %d\n", 1, 0); - ret = triton_mpi_recv(recvbuf, TRITON_FAKESS_SIZE_MESSAGE, MPI_CHAR, 0, - TRITON_TAG_FAKESS, MPI_COMM_WORLD, &status); - MPI_Get_count(&status, MPI_CHAR, &recvsize); - printf("%d: received message from %d: %.*s\n", 1, - 0, recvsize, recvbuf); - - if (strstr(recvbuf, "GLOBAL:")) - { - triton_fakess_notify_global(recvbuf); - } - else - { - printf("triton_fakess_recv(): unknown message\n"); - exit(1); - } - return TRITON_SUCCESS; } diff --git a/code/src/fakess/fakess.hae b/code/src/fakess/fakess.hae index 50d95f3..4f9fef7 100644 --- a/code/src/fakess/fakess.hae +++ b/code/src/fakess/fakess.hae @@ -15,85 +15,37 @@ #define TRITON_FAKESS_SIZE_VALUE 4096 #define TRITON_TAG_FAKESS 47 -typedef struct -{ - triton_node_t node; - struct triton_hash_link *link; -} triton_node_entry; +#define TRITON_FAKESS_UNSET ((void *)1) typedef struct { - triton_string_t* key; - triton_string_t* value; - triton_noter_t *noter; - struct triton_hash_link link; -} triton_string_entry; - -/** - These items are passed into wait() and returned in the workq. -*/ -typedef struct -{ - triton_node_t node; - triton_string_t *key; - triton_list_link_t link; -} triton_fakess_item; + void *value; + triton_mutex_t *mutex; +} triton_fakess_result_t; -__blocking triton_ret_t triton_fakess_init(void); -__blocking triton_ret_t triton_fakess_finalize(void); +triton_ret_t triton_fakess_init(void); +triton_ret_t triton_fakess_finalize(void); -triton_ret_t triton_fakess_listen_globals(triton_listener_t listener); +__blocking void triton_fakess_wait(triton_node_t node, + triton_string_t *key); -/** - Query for number of triton_nodes. -*/ -__blocking triton_ret_t triton_fakess_node_count(int *count); +void *triton_fakess_lock(triton_node_t node, + triton_string_t *key); -/** - Add a triton_node to the fakess space. -*/ -__blocking triton_ret_t triton_fakess_node_add(triton_node_t *node); +void triton_fakess_unlock(triton_node_t node, + triton_string_t *key); -triton_ret_t triton_fakess_global_item_add(triton_string_t *key, - triton_string_t *value); +triton_ret_t triton_fakess_node_add(triton_node_t node); triton_ret_t triton_fakess_item_add(triton_node_t node, - triton_string_t *key, - triton_string_t *value); - -/** - Perform the local update. - (Internal method, only accessible here for debugging.) -*/ -triton_ret_t triton_fakess_local_update_global(triton_string_t *key, - triton_string_t *value); - -/** - Posts change to global state. -*/ -__blocking triton_ret_t triton_fakess_post_global(triton_string_t *key, - triton_string_t *value); - -triton_ret_t triton_fakess_listen_global(triton_string_t *key, - triton_listener_t listener); - -/** - Block until a change occurs on a given item. -*/ -__blocking triton_ret_t triton_fakess_wait(triton_list_t *items, - triton_list_t *workq, - int *component); - -/** - Notify all pbranches waiting on the node/key. -*/ -void triton_fakess_notify(triton_node_t node, - triton_string_t *key); + triton_string_t *key); + +triton_ret_t triton_fakess_set(triton_node_t node, + triton_string_t *key, + void *value); -/** - Receive a fakess update. -*/ -__blocking triton_ret_t triton_fakess_recv(void); +triton_fakess_result_t *triton_fakess_get(triton_node_t node, + triton_string_t *key); #endif diff --git a/code/src/fakess/module.mk.in b/code/src/fakess/module.mk.in index f59e8d0..c1efa56 100644 --- a/code/src/fakess/module.mk.in +++ b/code/src/fakess/module.mk.in @@ -3,7 +3,7 @@ DIR := src/fakess ifneq (,$(BUILD_MPI)) AELIBSRC += $(DIR)/fakess.ae -AELIBSRC += $(DIR)/bqueue.ae +# AELIBSRC += $(DIR)/bqueue.ae MODCFLAGS_$(DIR)/fakess = $(MPICFLAGS) MODLDFLAGS_$(DIR)/fakess = $(MPILDFLAGS) diff --git a/code/src/fakess/test/module.mk.in b/code/src/fakess/test/module.mk.in index c28a38a..362ba53 100644 --- a/code/src/fakess/test/module.mk.in +++ b/code/src/fakess/test/module.mk.in @@ -4,7 +4,11 @@ DIR = src/fakess/test #AETESTSRC += $(DIR)/wait.ae #AETESTSRC += $(DIR)/post-global.ae #AETESTSRC += $(DIR)/listen-global.ae -AETESTSRC += $(DIR)/testbqueue.ae +#AETESTSRC += $(DIR)/testbqueue.ae +AETESTSRC += $(DIR)/fakess03.ae + +MODLIBS_$(DIR)/tables = -lpthread +MODLIBS_$(DIR)/fakess03 = -lpthread MODCFLAGS_$(DIR)/setup = $(MPICFLAGS) MODLDFLAGS_$(DIR)/setup = $(MPILDFLAGS) diff --git a/code/src/fakess/test/wait.ae b/code/src/fakess/test/wait.ae index bad93d5..3707231 100644 --- a/code/src/fakess/test/wait.ae +++ b/code/src/fakess/test/wait.ae @@ -59,7 +59,7 @@ static __blocking int test_wait(void) triton_list_add_back(&(item->link), items); printf("waiter: enter wait\n"); - code = triton_fakess_wait(items, workq, &component); + code = triton_fakess_waitany(items, workq, &component); assert(code == TRITON_SUCCESS); printf("waiter: notified\n"); diff --git a/code/src/net/triton-node.h b/code/src/net/triton-node.h index c898d8a..a0cc07a 100644 --- a/code/src/net/triton-node.h +++ b/code/src/net/triton-node.h @@ -26,7 +26,7 @@ triton_string_t *triton_node_to_string(triton_msg_ctx_t ctx, triton_node_t node) triton_node_t triton_node_lookup(triton_msg_ctx_t ctx, const char *name); -#define triton_node_eq(n1,n2) (n1.l==n2.l && n1.u==n2.u) +#define triton_node_eq(n1,n2) ((n1).l==(n2).l && (n1).u==(n2).u) #define TRITON_NODE_ANY TRITON_NODE_NULL hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. ac8dc9e40ca010f6d1a3c5b8a17a7639430774f0
by noreply＠mcs.anl.gov 28 May '10

28 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via ac8dc9e40ca010f6d1a3c5b8a17a7639430774f0 (commit) from 5d9650205b1d38e3ae308d5d9ef762985b0909f7 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit ac8dc9e40ca010f6d1a3c5b8a17a7639430774f0 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Fri May 28 15:00:45 2010 -0400 switch to log_flush() for db coalescing ----------------------------------------------------------------------- Summary of changes: .../prototype/bdb-resource/bdb-resource.c | 54 ++++++++++++++++++++ .../prototype/bdb-resource/bdb-resource.hae | 7 +++ code/src/versioned-osd/prototype/versioned-osd.ae | 5 +- 3 files changed, 64 insertions(+), 2 deletions(-) Diff of changes: diff --git a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c index eb1ba0f..8e7e800 100644 --- a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c +++ b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c @@ -120,6 +120,11 @@ struct bdb_op u_int32_t min; u_int32_t flags; } txn_checkpoint; + struct log_flush_op + { + DB_ENV *env; + DB_LSN *lsn; + } log_flush; }u; }; @@ -134,6 +139,7 @@ static int put_work_fn(struct ae_op* op); static int del_work_fn(struct ae_op* op); static int cursor_work_fn(struct ae_op* op); static int txn_checkpoint_work_fn(struct ae_op* op); +static int log_flush_work_fn(struct ae_op* op); static void* thread_fn(void* foo) { @@ -423,6 +429,54 @@ triton_ret_t bdb_txn_begin( return(TRITON_SUCCESS); } +triton_ret_t bdb_log_flush( + DB_ENV *env, + DB_LSN *lsn, + void (*callback)(void *ptr, triton_ret_t ret), + void *user_ptr, + ae_hints_t hints, + ae_context_t ctx, + ae_op_id_t *op_id) +{ + struct ae_op *op; + struct bdb_op *b_op; + + op = ae_opcache_get(bdb_opcache); + if(!op) + return(TRITON_ERR_NOMEM); + ae_op_fill(op, callback, user_ptr, hints, ctx); + + b_op = ae_op_entry(op, struct bdb_op, op); + b_op->op_id = ae_id_gen(triton_bdb_resource_id, (uint64_t)(op->cache_id)); + b_op->u.log_flush.env = env; + b_op->u.log_flush.lsn = lsn; + b_op->work_fn = log_flush_work_fn; + b_op->cleanup_fn = NULL; + + *op_id = b_op->op_id; + bdb_launch_op(op, triton_bdb_progress_mode); + + return(TRITON_SUCCESS); +} + +static int log_flush_work_fn(struct ae_op* op) +{ + struct bdb_op* b_op; + int ret; + + b_op = ae_op_entry(op, struct bdb_op, op); + assert(b_op); + + ret = b_op->u.log_flush.env->log_flush( + b_op->u.log_flush.env, + b_op->u.log_flush.lsn); + triton_debug(bdb_r_mask, "bdb resource flushed log\n"); + + b_op->return_code = triton_ret_from_bdb(ret); + + return(1); +} + triton_ret_t bdb_txn_checkpoint( DB_ENV *env, u_int32_t kbyte, diff --git a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae index 7b2219d..aaf0f47 100644 --- a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae +++ b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae @@ -128,6 +128,13 @@ __blocking triton_ret_t bdb_txn_checkpoint( u_int32_t min, u_int32_t flags); +/** + * Flush bdb log to disk + */ +__blocking triton_ret_t bdb_log_flush( + DB_ENV *env, + DB_LSN *lsn); + /** * Converts a BDB error code into a triton error code */ diff --git a/code/src/versioned-osd/prototype/versioned-osd.ae b/code/src/versioned-osd/prototype/versioned-osd.ae index 58ed74c..ddd19b7 100644 --- a/code/src/versioned-osd/prototype/versioned-osd.ae +++ b/code/src/versioned-osd/prototype/versioned-osd.ae @@ -197,8 +197,9 @@ static __blocking triton_ret_t end_db_write(struct coalesce_obj *obj) } triton_mutex_unlock(&coalesce_mutex); - tret = bdb_txn_checkpoint(envp, 0, 0, 0); + tret = bdb_log_flush(envp, NULL); /* TODO: need to propigate this return code out somehow */ + assert(tret == TRITON_SUCCESS); /* notify waiters */ triton_mutex_lock(&coalesce_mutex); @@ -1250,7 +1251,7 @@ triton_ret_t vosd_init(const char* path, int db_coalesce_highwater) /* Coalescing enabled. Make bdb transactions non-durable so that we * can explicitly checkpoint them via coalescing */ - ret = envp->set_flags(envp, DB_TXN_NOSYNC, 1); + ret = envp->set_flags(envp, DB_TXN_WRITE_NOSYNC, 1); if(ret != 0) { tret = triton_ret_from_bdb(errno); hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 5d9650205b1d38e3ae308d5d9ef762985b0909f7
by noreply＠mcs.anl.gov 27 May '10

27 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 5d9650205b1d38e3ae308d5d9ef762985b0909f7 (commit) from 2267c0a000bf465a07cd6531aa6278212dc1e4fb (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 5d9650205b1d38e3ae308d5d9ef762985b0909f7 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Thu May 27 10:59:26 2010 -0400 db coalescing for vosd ----------------------------------------------------------------------- Summary of changes: .../prototype/bdb-resource/bdb-resource.c | 64 ++++++++++++++++++++ .../prototype/bdb-resource/bdb-resource.hae | 9 +++ .../prototype/tests/vosd-bench-concurrent.ae | 20 +++++-- code/src/versioned-osd/prototype/versioned-osd.ae | 32 ++++------ 4 files changed, 101 insertions(+), 24 deletions(-) Diff of changes: diff --git a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c index 1ef899b..eb1ba0f 100644 --- a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c +++ b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.c @@ -113,6 +113,13 @@ struct bdb_op DBC **cursorp; u_int32_t flags; } cursor; + struct txn_checkpoint_op + { + DB_ENV *env; + u_int32_t kbyte; + u_int32_t min; + u_int32_t flags; + } txn_checkpoint; }u; }; @@ -126,6 +133,7 @@ static int cursor_close_work_fn(struct ae_op* op); static int put_work_fn(struct ae_op* op); static int del_work_fn(struct ae_op* op); static int cursor_work_fn(struct ae_op* op); +static int txn_checkpoint_work_fn(struct ae_op* op); static void* thread_fn(void* foo) { @@ -415,6 +423,62 @@ triton_ret_t bdb_txn_begin( return(TRITON_SUCCESS); } +triton_ret_t bdb_txn_checkpoint( + DB_ENV *env, + u_int32_t kbyte, + u_int32_t min, + u_int32_t flags, + void (*callback)(void *ptr, triton_ret_t ret), + void *user_ptr, + ae_hints_t hints, + ae_context_t ctx, + ae_op_id_t *op_id) +{ + struct ae_op *op; + struct bdb_op *b_op; + + op = ae_opcache_get(bdb_opcache); + if(!op) + return(TRITON_ERR_NOMEM); + ae_op_fill(op, callback, user_ptr, hints, ctx); + + b_op = ae_op_entry(op, struct bdb_op, op); + b_op->op_id = ae_id_gen(triton_bdb_resource_id, (uint64_t)(op->cache_id)); + b_op->u.txn_checkpoint.env = env; + b_op->u.txn_checkpoint.kbyte = kbyte; + b_op->u.txn_checkpoint.min = min; + b_op->u.txn_checkpoint.flags = flags; + b_op->work_fn = txn_checkpoint_work_fn; + b_op->cleanup_fn = NULL; + + *op_id = b_op->op_id; + bdb_launch_op(op, triton_bdb_progress_mode); + + return(TRITON_SUCCESS); +} + + +static int txn_checkpoint_work_fn(struct ae_op* op) +{ + struct bdb_op* b_op; + int ret; + + b_op = ae_op_entry(op, struct bdb_op, op); + assert(b_op); + + ret = b_op->u.txn_checkpoint.env->txn_checkpoint( + b_op->u.txn_checkpoint.env, + b_op->u.txn_checkpoint.kbyte, + b_op->u.txn_checkpoint.min, + b_op->u.txn_checkpoint.flags); + triton_debug(bdb_r_mask, "bdb resource checkpointed\n"); + + b_op->return_code = triton_ret_from_bdb(ret); + + return(1); +} + + static int txn_begin_work_fn(struct ae_op* op) { struct bdb_op* b_op; diff --git a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae index 6f04fd4..7b2219d 100644 --- a/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae +++ b/code/src/versioned-osd/prototype/bdb-resource/bdb-resource.hae @@ -120,6 +120,15 @@ __blocking triton_ret_t bdb_cursor( u_int32_t flags); /** + * Checkpoint transactions + */ +__blocking triton_ret_t bdb_txn_checkpoint( + DB_ENV *env, + u_int32_t kbyte, + u_int32_t min, + u_int32_t flags); + +/** * Converts a BDB error code into a triton error code */ triton_ret_t triton_ret_from_bdb(int error); diff --git a/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae b/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae index 277ded3..91baa58 100644 --- a/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae +++ b/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae @@ -168,10 +168,11 @@ int main(int argc, char *argv[]) int ret; struct bench_op* tmp_op; triton_ret_t tret; + int highwater = 0; - if(argc != 5) + if(argc != 6) { - fprintf(stderr, "Usage: vosd-bench-concurrent <workload description file> <vosd dir> <concurrent ops> <k|d>\n"); + fprintf(stderr, "Usage: vosd-bench-concurrent <workload description file> <vosd dir> <concurrent ops> <k|d> <highwater mark>\n"); fprintf(stderr, " # k for key/val mode\n"); fprintf(stderr, " # d for data mode\n"); return(-1); @@ -180,7 +181,7 @@ int main(int argc, char *argv[]) ret = sscanf(argv[3], "%d", &concurrent); if(ret != 1 || concurrent < 1) { - fprintf(stderr, "Usage: vosd-bench-concurrent <workload description file> <vosd dir> <concurrent ops> <k|d>\n"); + fprintf(stderr, "Usage: vosd-bench-concurrent <workload description file> <vosd dir> <concurrent ops> <k|d> <highwater mark>\n"); fprintf(stderr, " # k for key/val mode\n"); fprintf(stderr, " # d for data mode\n"); return(-1); @@ -189,7 +190,16 @@ int main(int argc, char *argv[]) ret = sscanf(argv[4], "%c", &data_mode); if(ret != 1 || (data_mode != 'k' && data_mode != 'd')) { - fprintf(stderr, "Usage: vosd-bench-concurrent <workload description file> <vosd dir> <concurrent ops> <k|d>\n"); + fprintf(stderr, "Usage: vosd-bench-concurrent <workload description file> <vosd dir> <concurrent ops> <k|d> <highwater mark>\n"); + fprintf(stderr, " # k for key/val mode\n"); + fprintf(stderr, " # d for data mode\n"); + return(-1); + } + + ret = sscanf(argv[5], "%d", &highwater); + if(ret != 1 || highwater < 0) + { + fprintf(stderr, "Usage: vosd-bench-concurrent <workload description file> <vosd dir> <concurrent ops> <k|d> <highwater mark>\n"); fprintf(stderr, " # k for key/val mode\n"); fprintf(stderr, " # d for data mode\n"); return(-1); @@ -241,7 +251,7 @@ int main(int argc, char *argv[]) #endif assert(tret == TRITON_SUCCESS); - tret = vosd_init(argv[2], 0); + tret = vosd_init(argv[2], highwater); assert(tret == TRITON_SUCCESS); ae_context_create(&ctx, 3, "bdb", "file", "sched"); diff --git a/code/src/versioned-osd/prototype/versioned-osd.ae b/code/src/versioned-osd/prototype/versioned-osd.ae index d468668..58ed74c 100644 --- a/code/src/versioned-osd/prototype/versioned-osd.ae +++ b/code/src/versioned-osd/prototype/versioned-osd.ae @@ -32,8 +32,6 @@ * - audit what calls can return DB_LOCK_DEADLOCK (c_close() seems silly, * for example) * - deadlock avoidance - * - consider non-durable transactions in order to coalesce commits - * - use txn_checkpoint to flush when coalescing * - consider rmw locks where appropriate * - long term: consider fallocate hint for logs to keep them linear * - try computing cksum along the way, store in db @@ -164,6 +162,7 @@ static __blocking triton_ret_t end_db_write(struct coalesce_obj *obj) triton_list_t to_notify; triton_list_link_t *nop; struct coalesce_obj *cop; + triton_ret_t tret; /* do nothing if coalescing is off */ if(global_db_coalesce_highwater == 0) @@ -173,7 +172,6 @@ static __blocking triton_ret_t end_db_write(struct coalesce_obj *obj) triton_mutex_lock(&coalesce_mutex); /* remove myself from db_write_started_queue */ - printf("FOO: ending db write for obj %p\n", obj); triton_list_del(&obj->link); /* watermark check: @@ -199,13 +197,8 @@ static __blocking triton_ret_t end_db_write(struct coalesce_obj *obj) } triton_mutex_unlock(&coalesce_mutex); - printf("FOO: db write for obj %p is syncing\n", obj); -#if 0 - /* do fake sync */ - printf("hit low/high watermark,\tindex: %d synced\n", obj->ind); - ret = triton_timer(1000); - assert(ret == TRITON_SUCCESS); -#endif + tret = bdb_txn_checkpoint(envp, 0, 0, 0); + /* TODO: need to propigate this return code out somehow */ /* notify waiters */ triton_mutex_lock(&coalesce_mutex); @@ -221,7 +214,6 @@ static __blocking triton_ret_t end_db_write(struct coalesce_obj *obj) } else { - printf("FOO: db write for obj %p is coalescing\n", obj); triton_mutex_unlock(&db_write_started_mutex); /* Need to wait for more operations to coalesce. Add to queue and @@ -1253,16 +1245,18 @@ triton_ret_t vosd_init(const char* path, int db_coalesce_highwater) goto vosd_init_error; } -#if 0 - /* purely for experimental purposes at this point */ - /* make transactions non-durable */ - ret = envp->set_flags(envp, DB_TXN_NOSYNC, 1); - if(ret != 0) + if(global_db_coalesce_highwater > 0) { - tret = triton_ret_from_bdb(errno); - goto vosd_init_error; + /* Coalescing enabled. Make bdb transactions non-durable so that we + * can explicitly checkpoint them via coalescing + */ + ret = envp->set_flags(envp, DB_TXN_NOSYNC, 1); + if(ret != 0) + { + tret = triton_ret_from_bdb(errno); + goto vosd_init_error; + } } -#endif /* have db do deadlock detection and reject txn with fewest locks */ ret = envp->set_lk_detect(envp, DB_LOCK_MINWRITE); hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 2267c0a000bf465a07cd6531aa6278212dc1e4fb
by noreply＠mcs.anl.gov 27 May '10

27 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 2267c0a000bf465a07cd6531aa6278212dc1e4fb (commit) via 022a5d0b28c892b6e4abe41d1a7140dac6e7d34b (commit) via 390668dc558a73f32af5338230bf53a617b976a0 (commit) from 76fe742487888b3bfedc8ae08d0e3056fe4244cc (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 2267c0a000bf465a07cd6531aa6278212dc1e4fb Author: Phil Carns <carns(a)mcs.anl.gov> Date: Thu May 27 10:05:33 2010 -0400 don't use vosd db coalescing yet commit 022a5d0b28c892b6e4abe41d1a7140dac6e7d34b Author: Phil Carns <carns(a)mcs.anl.gov> Date: Thu May 27 09:52:59 2010 -0400 fix callback race in sched resource commit 390668dc558a73f32af5338230bf53a617b976a0 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Wed May 26 19:00:55 2010 -0400 partial infrastructure for coalescing db txns ----------------------------------------------------------------------- Summary of changes: code/src/common/resources/scheduling/sched.c | 12 +- .../prototype/tests/vosd-bench-concurrent.ae | 4 +- .../prototype/tests/vosd-multi-object.ae | 4 +- code/src/versioned-osd/prototype/tests/vosd1.ae | 4 +- code/src/versioned-osd/prototype/versioned-osd.ae | 287 +++++++++++++++++--- code/src/versioned-osd/prototype/versioned-osd.hae | 4 +- 6 files changed, 266 insertions(+), 49 deletions(-) Diff of changes: diff --git a/code/src/common/resources/scheduling/sched.c b/code/src/common/resources/scheduling/sched.c index 042a6ae..51e106a 100644 --- a/code/src/common/resources/scheduling/sched.c +++ b/code/src/common/resources/scheduling/sched.c @@ -123,11 +123,20 @@ static triton_ret_t triton_sched_poll(ae_context_t context, int millisecs) triton_mutex_lock(&sched_done_mutex); while(ae_ops_count(&sched_done_queue) > 0) { + void (*callback)(void *, triton_ret_t); + void *user_ptr; + op = ae_ops_dequeue(&sched_done_queue); triton_mutex_unlock(&sched_done_mutex); result = ae_op_entry(op, struct triton_sched_id_s, op); triton_mutex_lock(result->mutex); - ae_invoke_callback(op, triton_ret_t, TRITON_SUCCESS); + + /* save callback and user pointer so that we can release the op + * safely before triggering the callback + */ + callback = op->callback; + user_ptr = op->user_ptr; + if(result->internal) { ae_opcache_put(sched_opcache, op); @@ -136,6 +145,7 @@ static triton_ret_t triton_sched_poll(ae_context_t context, int millisecs) { ae_op_clear(op); } + callback(user_ptr, TRITON_SUCCESS); ++completed; if(--est_count_before_timeout <= 0) diff --git a/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae b/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae index 15d61ad..277ded3 100644 --- a/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae +++ b/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae @@ -241,10 +241,10 @@ int main(int argc, char *argv[]) #endif assert(tret == TRITON_SUCCESS); - tret = vosd_init(argv[2]); + tret = vosd_init(argv[2], 0); assert(tret == TRITON_SUCCESS); - ae_context_create(&ctx, 2, "bdb", "file"); + ae_context_create(&ctx, 3, "bdb", "file", "sched"); done = 0; do_vosd_test(done_callback, NULL, NULL, ctx, &op_id); diff --git a/code/src/versioned-osd/prototype/tests/vosd-multi-object.ae b/code/src/versioned-osd/prototype/tests/vosd-multi-object.ae index f94e6f4..014f3d8 100644 --- a/code/src/versioned-osd/prototype/tests/vosd-multi-object.ae +++ b/code/src/versioned-osd/prototype/tests/vosd-multi-object.ae @@ -154,14 +154,14 @@ int main(int argc, char *argv[]) return(-1); } - ret = vosd_init("/tmp/vosd"); + ret = vosd_init("/tmp/vosd", 0); if(ret != TRITON_SUCCESS) { triton_perror("vosd_init", ret); return(-1); } - ae_context_create(&ctx, 2, "bdb", "file"); + ae_context_create(&ctx, 3, "bdb", "file", "sched"); if(ret != TRITON_SUCCESS) { triton_perror("ae_context_create", ret); diff --git a/code/src/versioned-osd/prototype/tests/vosd1.ae b/code/src/versioned-osd/prototype/tests/vosd1.ae index 8d7a24e..4c9623b 100644 --- a/code/src/versioned-osd/prototype/tests/vosd1.ae +++ b/code/src/versioned-osd/prototype/tests/vosd1.ae @@ -545,14 +545,14 @@ int main(int argc, char *argv[]) return(-1); } - ret = vosd_init("/tmp/vosd"); + ret = vosd_init("/tmp/vosd", 0); if(ret != TRITON_SUCCESS) { triton_perror("vosd_init", ret); return(-1); } - ae_context_create(&ctx, 2, "bdb", "file"); + ae_context_create(&ctx, 3, "bdb", "file", "sched"); if(ret != TRITON_SUCCESS) { triton_perror("vosd_init", ret); diff --git a/code/src/versioned-osd/prototype/versioned-osd.ae b/code/src/versioned-osd/prototype/versioned-osd.ae index 9d6e57c..d468668 100644 --- a/code/src/versioned-osd/prototype/versioned-osd.ae +++ b/code/src/versioned-osd/prototype/versioned-osd.ae @@ -59,7 +59,19 @@ static DB *missing_ver_dbp = NULL; static DB *ver_dbp = NULL; static struct triton_hash_table *txn_table = NULL; static triton_mutex_t txn_mutex = TRITON_MUTEX_INITIALIZER; -static int txn_pending_count = 0; +static int global_db_coalesce_highwater = 0; +/* TODO: disable coalescing entirely if highwater is zero */ + +struct coalesce_obj +{ + triton_sched_id_t sched_id; + triton_list_link_t link; +}; +static triton_sched_t coalesce_sched; +static triton_mutex_t coalesce_mutex = TRITON_MUTEX_INITIALIZER; +static triton_list_t coalesce_queue = TRITON_LIST_STATIC_INITIALIZER(coalesce_queue); +static triton_mutex_t db_write_started_mutex = TRITON_MUTEX_INITIALIZER; +static triton_list_t db_write_started_queue = TRITON_LIST_STATIC_INITIALIZER(db_write_started_queue); /* db entry representing region in logical map */ /* indexed via last offset */ @@ -145,6 +157,84 @@ static triton_ret_t merge_logical_map( struct triton_list* list2_in, struct triton_list* list_out, void** free_ptr); static void free_logical_map_list(struct triton_list* list); +static void start_db_write(struct coalesce_obj *obj); + +static __blocking triton_ret_t end_db_write(struct coalesce_obj *obj) +{ + triton_list_t to_notify; + triton_list_link_t *nop; + struct coalesce_obj *cop; + + /* do nothing if coalescing is off */ + if(global_db_coalesce_highwater == 0) + return TRITON_SUCCESS; + + triton_mutex_lock(&db_write_started_mutex); + triton_mutex_lock(&coalesce_mutex); + + /* remove myself from db_write_started_queue */ + printf("FOO: ending db write for obj %p\n", obj); + triton_list_del(&obj->link); + + /* watermark check: + * low: If no more ops are coming in, then checkpoint what we have so far. + * high: Checkpoint regardless of other pending operations if we already + * have highwater mark operations waiting. + */ + if(triton_list_empty(&db_write_started_queue) || + triton_list_count(&coalesce_queue) >= global_db_coalesce_highwater) + { + triton_mutex_unlock(&db_write_started_mutex); + + /* high-watermark or low-watermark reached. Checkpoint berkeley db. */ + + /* Move the coalesced operations onto a local queue to make sure + * that we only release the operations that triggered this + * particular checkpoint. + */ + triton_list_init(&to_notify); + while(!triton_list_empty(&coalesce_queue)) + { + triton_queue_enqueue(triton_queue_dequeue(&coalesce_queue), &to_notify); + } + triton_mutex_unlock(&coalesce_mutex); + + printf("FOO: db write for obj %p is syncing\n", obj); +#if 0 + /* do fake sync */ + printf("hit low/high watermark,\tindex: %d synced\n", obj->ind); + ret = triton_timer(1000); + assert(ret == TRITON_SUCCESS); +#endif + + /* notify waiters */ + triton_mutex_lock(&coalesce_mutex); + while(!triton_list_empty(&to_notify)) + { + nop = triton_queue_dequeue(&to_notify); + cop = triton_list_get_entry(nop, struct coalesce_obj, link); + + /* notify this guy! */ + triton_sched_notify_specific(coalesce_sched, &cop->sched_id); + } + triton_mutex_unlock(&coalesce_mutex); + } + else + { + printf("FOO: db write for obj %p is coalescing\n", obj); + triton_mutex_unlock(&db_write_started_mutex); + + /* Need to wait for more operations to coalesce. Add to queue and + * then wait for other operations. + */ + + triton_queue_enqueue(&obj->link, &coalesce_queue); + triton_sched_obj_wait(coalesce_sched, &coalesce_mutex, &obj->sched_id); + triton_mutex_unlock(&coalesce_mutex); + } + + return TRITON_SUCCESS; +} /* TODO: prototype gives compile errors; have to define function here * instead @@ -780,6 +870,53 @@ static __blocking triton_ret_t __vosd_txn_close_retryable( } +static __blocking triton_ret_t vosd_txn_close_internal( + uint64_t oid, + uint64_t oid_fork, + uint64_t txn_number) +{ + triton_ret_t tret; + struct txn_accumulator* txn_acc; + struct triton_hash_link* hash_link; + int ret = 0; + + /* pull txn accumulator out of hash so no one can touch it */ + triton_mutex_lock(&txn_mutex); + hash_link = triton_hash_search(txn_table, &txn_number); + if(!hash_link) + { + triton_mutex_unlock(&txn_mutex); + return(TRITON_ERR_NO_TXN); + } + txn_acc = triton_hash_get_entry(hash_link, struct txn_accumulator, + hash_link); + + if(txn_acc->oid != oid || + txn_acc->oid_fork != oid_fork) + { + triton_mutex_unlock(&txn_mutex); + return(TRITON_ERR_INVAL); + } + + triton_list_del(&txn_acc->hash_link); + triton_mutex_unlock(&txn_mutex); + + /* actual work of closing the txn; retry if DB deadlock */ + for(tret = TRITON_ERR_DEADLOCK; tret == TRITON_ERR_DEADLOCK; ) + { + /* TODO: triton_error_destroy() causes segfaults. */ +#if 0 + triton_error_destroy(tret); +#endif + tret = __vosd_txn_close_retryable(txn_acc); + } + + /* TODO: any cleanup steps if we failed to close the txn? */ + /* For now just assume that we leave the txn as is. BDB semantics + * would destroy the txn. + */ + return(tret); +} #if 0 static __blocking triton_ret_t __get_read_region_retryable( struct logical_map_key* lmk, struct logical_map_entry* lme) @@ -1026,7 +1163,7 @@ static __blocking triton_ret_t read_segment( return(TRITON_SUCCESS); } -triton_ret_t vosd_init(const char* path) +triton_ret_t vosd_init(const char* path, int db_coalesce_highwater) { triton_ret_t tret; int ret; @@ -1034,6 +1171,9 @@ triton_ret_t vosd_init(const char* path) uint32_t env_flags; uint32_t open_flags; + global_db_coalesce_highwater = db_coalesce_highwater; + assert(global_db_coalesce_highwater >= 0); + /* initialize BDB resource */ tret = bdb_init(TRITON_BDB_PROG_THREAD_PER_OP); if(tret != TRITON_SUCCESS) @@ -1049,6 +1189,21 @@ triton_ret_t vosd_init(const char* path) return(tret); } + /* initialize scheduling resource */ + tret = triton_sched_resource_init(); + if(tret != TRITON_SUCCESS) + { + file_finalize(); + bdb_finalize(); + return(tret); + } + + tret = triton_sched_init(&coalesce_sched); + if(tret != TRITON_SUCCESS) + { + goto vosd_init_error; + } + /* make a directory for storage if not already present */ ret = mkdir(path, 0755); if(ret != 0 && errno == EEXIST) @@ -1220,6 +1375,9 @@ vosd_init_error: void vosd_finalize(void) { + triton_sched_destroy(coalesce_sched); + + triton_sched_resource_finalize(); file_finalize(); bdb_finalize(); @@ -1271,6 +1429,10 @@ __blocking triton_ret_t vosd_get_version(uint64_t oid, uint64_t* version) __blocking triton_ret_t vosd_create(uint64_t oid) { triton_ret_t tret; + triton_ret_t tret_tmp; + struct coalesce_obj co; + + start_db_write(&co); /* TODO: note that a do/while loop will compile here, but not work * correctly @@ -1283,9 +1445,16 @@ __blocking triton_ret_t vosd_create(uint64_t oid) #endif tret = __vosd_create_retryable(oid); } - /* delay creating log file until the first write() operation */ + tret_tmp = end_db_write(&co); + if(tret == TRITON_SUCCESS) + { + return(tret_tmp); + } +#if 0 + triton_error_destroy(tret_tmp); +#endif return(tret); } @@ -1534,7 +1703,6 @@ triton_ret_t vosd_txn_open( triton_list_init(&txn_acc->update_list); triton_mutex_lock(&txn_mutex); - txn_pending_count++; /* TODO: safety test for duplicate txn numbers */ triton_hash_add(txn_table, &txn_acc->txn_number, &txn_acc->hash_link); triton_mutex_unlock(&txn_mutex); @@ -1542,7 +1710,7 @@ triton_ret_t vosd_txn_open( return(TRITON_SUCCESS); } -__blocking triton_ret_t vosd_write( +static __blocking triton_ret_t vosd_write_internal( uint64_t oid, uint64_t oid_fork, uint64_t txn_number, @@ -1744,60 +1912,67 @@ __blocking triton_ret_t vosd_write( tret = TRITON_SUCCESS; if(flags & VOSD_FLAG_AUTO_TXN) { - tret = vosd_txn_close(oid, oid_fork, txn_number); + tret = vosd_txn_close_internal(oid, oid_fork, txn_number); } free_tmp_listio(); return(tret); } -__blocking triton_ret_t vosd_txn_close( +__blocking triton_ret_t vosd_write( uint64_t oid, uint64_t oid_fork, - uint64_t txn_number) + uint64_t txn_number, + char** mem_offsets, + int64_t* mem_sizes, + int mem_count, + int64_t* obj_offsets, + int64_t* obj_sizes, + int obj_count, + int flags) { triton_ret_t tret; - struct txn_accumulator* txn_acc; - struct triton_hash_link* hash_link; - int ret = 0; + triton_ret_t tret_tmp; + struct coalesce_obj co; - /* pull txn accumulator out of hash so no one can touch it */ - triton_mutex_lock(&txn_mutex); - hash_link = triton_hash_search(txn_table, &txn_number); - if(!hash_link) - { - triton_mutex_unlock(&txn_mutex); - return(TRITON_ERR_NO_TXN); - } - txn_acc = triton_hash_get_entry(hash_link, struct txn_accumulator, - hash_link); + start_db_write(&co); - if(txn_acc->oid != oid || - txn_acc->oid_fork != oid_fork) + tret = vosd_write_internal(oid, oid_fork, txn_number, mem_offsets, + mem_sizes, mem_count, obj_offsets, obj_sizes, obj_count, flags); + + tret_tmp = end_db_write(&co); + if(tret == TRITON_SUCCESS) { - triton_mutex_unlock(&txn_mutex); - return(TRITON_ERR_INVAL); + return(tret_tmp); } - - triton_list_del(&txn_acc->hash_link); - triton_mutex_unlock(&txn_mutex); +#if 0 + triton_error_destroy(tret_tmp); +#endif + return(tret); +} - /* actual work of closing the txn; retry if DB deadlock */ - for(tret = TRITON_ERR_DEADLOCK; tret == TRITON_ERR_DEADLOCK; ) + +__blocking triton_ret_t vosd_txn_close( + uint64_t oid, + uint64_t oid_fork, + uint64_t txn_number) +{ + triton_ret_t tret; + triton_ret_t tret_tmp; + struct coalesce_obj co; + + start_db_write(&co); + + tret = vosd_txn_close_internal(oid, oid_fork, txn_number); + + tret_tmp = end_db_write(&co); + if(tret == TRITON_SUCCESS) { - /* TODO: triton_error_destroy() causes segfaults. */ + return(tret_tmp); + } #if 0 - triton_error_destroy(tret); + triton_error_destroy(tret_tmp); #endif - tret = __vosd_txn_close_retryable(txn_acc); - } - - txn_pending_count--; - - /* TODO: any cleanup steps if we failed to close the txn? */ - /* For now just assume that we leave the txn as is. BDB semantics - * would destroy the txn. - */ return(tret); } @@ -1890,6 +2065,10 @@ __blocking triton_ret_t vosd_read( __blocking triton_ret_t vosd_remove(uint64_t oid) { triton_ret_t tret; + triton_ret_t tret_tmp; + struct coalesce_obj co; + + start_db_write(&co); for(tret = TRITON_ERR_DEADLOCK; tret == TRITON_ERR_DEADLOCK; ) { @@ -1899,6 +2078,14 @@ __blocking triton_ret_t vosd_remove(uint64_t oid) tret = __vosd_remove_retryable(oid); } + tret_tmp = end_db_write(&co); + if(tret == TRITON_SUCCESS) + { + return(tret_tmp); + } +#if 0 + triton_error_destroy(tret_tmp); +#endif return(tret); } @@ -2286,6 +2473,24 @@ static void free_logical_map_list(struct triton_list* list) return; } +/* must be called at the beginning of any vosd function that has the + * potential to modify bdb + */ +static void start_db_write(struct coalesce_obj *obj) +{ + /* do nothing if coalescing is off */ + if(global_db_coalesce_highwater == 0) + return; + + /* put all db writers in a queue to keep track of what operations are in + * progress that will modify bdb + */ + triton_list_link_clear(&obj->link); + triton_mutex_lock(&db_write_started_mutex); + triton_queue_enqueue(&obj->link, &db_write_started_queue); + triton_mutex_unlock(&db_write_started_mutex); +} + /* * Local variables: * c-indent-level: 4 diff --git a/code/src/versioned-osd/prototype/versioned-osd.hae b/code/src/versioned-osd/prototype/versioned-osd.hae index b2b191b..e94cf90 100644 --- a/code/src/versioned-osd/prototype/versioned-osd.hae +++ b/code/src/versioned-osd/prototype/versioned-osd.hae @@ -26,8 +26,10 @@ * Initializes VOSD interface */ triton_ret_t vosd_init( - const char* path /**< local path to use for storing data, + const char* path, /**< local path to use for storing data, will be created if needed */ + int db_coalesce_highwater /**< high water mark for db coalescing, + 0 disables coalescing */ ); /** hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 76fe742487888b3bfedc8ae08d0e3056fe4244cc
by noreply＠mcs.anl.gov 26 May '10

26 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 76fe742487888b3bfedc8ae08d0e3056fe4244cc (commit) from 29a07b872b14bd16335a021167bdfca1d880b918 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 76fe742487888b3bfedc8ae08d0e3056fe4244cc Author: Samuel Lang <slang(a)bblogin.mcs.anl.gov> Date: Wed May 26 20:27:08 2010 -0500 fix uninitialized errors in coalesce test ----------------------------------------------------------------------- Summary of changes: .../common/resources/scheduling/test/coalesce.ae | 1 + 1 files changed, 1 insertions(+), 0 deletions(-) Diff of changes: diff --git a/code/src/common/resources/scheduling/test/coalesce.ae b/code/src/common/resources/scheduling/test/coalesce.ae index 84f4b31..7e5d4c1 100644 --- a/code/src/common/resources/scheduling/test/coalesce.ae +++ b/code/src/common/resources/scheduling/test/coalesce.ae @@ -57,6 +57,7 @@ static __blocking int dostuff(void) void begin(struct coalesce_obj *obj) { /* add obj to queue of started ops. Allows us to keep track of what's in progress. */ + triton_list_link_clear(&obj->link); triton_mutex_lock(&ops_started_mutex); triton_queue_enqueue(&obj->link, &ops_started_queue); triton_mutex_unlock(&ops_started_mutex); hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 29a07b872b14bd16335a021167bdfca1d880b918
by noreply＠mcs.anl.gov 26 May '10

26 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 29a07b872b14bd16335a021167bdfca1d880b918 (commit) from deb1bfa33fcad2a84b6cb651a43002ac11e84e6a (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 29a07b872b14bd16335a021167bdfca1d880b918 Author: Samuel Lang <slang(a)bblogin.mcs.anl.gov> Date: Wed May 26 17:39:25 2010 -0500 fix minor ret value bug in caolesce test ----------------------------------------------------------------------- Summary of changes: .../common/resources/scheduling/test/coalesce.ae | 2 +- 1 files changed, 1 insertions(+), 1 deletions(-) Diff of changes: diff --git a/code/src/common/resources/scheduling/test/coalesce.ae b/code/src/common/resources/scheduling/test/coalesce.ae index 6142f4b..84f4b31 100644 --- a/code/src/common/resources/scheduling/test/coalesce.ae +++ b/code/src/common/resources/scheduling/test/coalesce.ae @@ -64,7 +64,7 @@ void begin(struct coalesce_obj *obj) __blocking int sync(struct coalesce_obj *obj) { - int ret; + triton_ret_t ret; triton_list_t to_notify; triton_list_link_t *nop; struct coalesce_obj *cop; hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. deb1bfa33fcad2a84b6cb651a43002ac11e84e6a
by noreply＠mcs.anl.gov 26 May '10

26 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via deb1bfa33fcad2a84b6cb651a43002ac11e84e6a (commit) from 663ef17326478d4869c894f4355dcb51c4598696 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit deb1bfa33fcad2a84b6cb651a43002ac11e84e6a Author: Phil Carns <carns(a)mcs.anl.gov> Date: Wed May 26 15:53:25 2010 -0400 remove references to old sched interface ----------------------------------------------------------------------- Summary of changes: code/src/versioned-osd/prototype/versioned-osd.ae | 48 --------------------- 1 files changed, 0 insertions(+), 48 deletions(-) Diff of changes: diff --git a/code/src/versioned-osd/prototype/versioned-osd.ae b/code/src/versioned-osd/prototype/versioned-osd.ae index d735175..9d6e57c 100644 --- a/code/src/versioned-osd/prototype/versioned-osd.ae +++ b/code/src/versioned-osd/prototype/versioned-osd.ae @@ -60,7 +60,6 @@ static DB *ver_dbp = NULL; static struct triton_hash_table *txn_table = NULL; static triton_mutex_t txn_mutex = TRITON_MUTEX_INITIALIZER; static int txn_pending_count = 0; -static triton_sched_t coalesce_sched; /* db entry representing region in logical map */ /* indexed via last offset */ @@ -146,7 +145,6 @@ static triton_ret_t merge_logical_map( struct triton_list* list2_in, struct triton_list* list_out, void** free_ptr); static void free_logical_map_list(struct triton_list* list); -static triton_sched_action_t coalesce_trigger(struct triton_sched *sched, triton_sched_trigger_event_t op); /* TODO: prototype gives compile errors; have to define function here * instead @@ -1036,9 +1034,6 @@ triton_ret_t vosd_init(const char* path) uint32_t env_flags; uint32_t open_flags; - triton_sched_reset(&coalesce_sched); - triton_sched_set_trigger(&coalesce_sched, coalesce_trigger); - /* initialize BDB resource */ tret = bdb_init(TRITON_BDB_PROG_THREAD_PER_OP); if(tret != TRITON_SUCCESS) @@ -1765,7 +1760,6 @@ __blocking triton_ret_t vosd_txn_close( struct txn_accumulator* txn_acc; struct triton_hash_link* hash_link; int ret = 0; - triton_sched_id sched_id; /* pull txn accumulator out of hash so no one can touch it */ triton_mutex_lock(&txn_mutex); @@ -1798,25 +1792,6 @@ __blocking triton_ret_t vosd_txn_close( tret = __vosd_txn_close_retryable(txn_acc); } - /* NOTE: need to think about race conditions before proceeding down - * the coalescing path - */ -#if 0 - /* TODO: why is the return code an int here? */ - ret = triton_sched_obj_wait(&coalesce_sched, &sched_id); - - /* TODO: error handling */ - assert(ret == 0); - if(sched_id.rank == 0) - { - /* checkpoint bdb */ - } - else - { - /* wait on rank 0 to checkpoint bdb */ - } -#endif - txn_pending_count--; /* TODO: any cleanup steps if we failed to close the txn? */ @@ -2311,29 +2286,6 @@ static void free_logical_map_list(struct triton_list* list) return; } -/* callback function to check when coalescing should be triggered */ -static triton_sched_action_t coalesce_trigger(struct triton_sched *sched, triton_sched_trigger_event_t op) -{ - int count; - triton_sched_action_t sret; - - triton_mutex_lock(&txn_mutex); - - count = ae_ops_count(&sched->opqueue); - /* if there are no more txns to wait for, _or_ if we have at least 8 - * txns waiting already, then go ahead and checkpoint - */ - if(count == txn_pending_count || count >= 8) - sret = TRITON_SCHED_ACTION_NOTIFY_ALL; - else - sret = TRITON_SCHED_ACTION_NONE; - - triton_mutex_unlock(&txn_mutex); - - return(sret); -} - - /* * Local variables: * c-indent-level: 4 hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 663ef17326478d4869c894f4355dcb51c4598696
by noreply＠mcs.anl.gov 26 May '10

26 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 663ef17326478d4869c894f4355dcb51c4598696 (commit) from 48ac98c8ebbd3e9a3d8d4c2a6bb355d2955bed6b (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 663ef17326478d4869c894f4355dcb51c4598696 Author: Samuel Lang <slang(a)bblogin.mcs.anl.gov> Date: Wed May 26 11:42:19 2010 -0500 new scheduling interface to get locking right. New coalescing example too. ----------------------------------------------------------------------- Summary of changes: code/src/common/resources/scheduling/sched.c | 263 ++++++++++---------- code/src/common/resources/scheduling/sched.hae | 74 ++---- .../common/resources/scheduling/test/coalesce.ae | 121 +++++++--- .../common/resources/scheduling/test/limitqueue.ae | 28 +-- .../common/resources/scheduling/test/module.mk.in | 2 + .../resources/scheduling/test/pthread-compare.ae | 21 +- code/src/common/resources/timer/timer.c | 8 +- code/src/common/resources/timer/timer.hae | 2 +- code/src/common/triton-thread.h | 2 + code/src/common/triton-types.h | 2 +- 10 files changed, 277 insertions(+), 246 deletions(-) Diff of changes: diff --git a/code/src/common/resources/scheduling/sched.c b/code/src/common/resources/scheduling/sched.c index c22235d..042a6ae 100644 --- a/code/src/common/resources/scheduling/sched.c +++ b/code/src/common/resources/scheduling/sched.c @@ -3,50 +3,39 @@ #include "src/aesop/aesop.h" #include "src/common/resources/scheduling/sched.h" -static int triton_sched_resource_id; -static ae_opcache_t sched_opcache; -static int triton_sched_notify_nolock(triton_sched_t *sched, triton_sched_action_t action); -static int triton_sched_notify_all_nolock(triton_sched_t *sched); -static int triton_sched_notify_next_nolock(triton_sched_t *sched); - -#define triton_sched_inotify(_sched, _action) \ - if(_action != TRITON_SCHED_ACTION_NONE) triton_sched_notify_nolock(_sched, _action) - -#define triton_sched_itrigger(_sched, _event) \ - do { \ - if((!((_sched)->innotify)) && (_sched)->trigger) \ - { \ - triton_sched_action_t a = (_sched)->trigger(sched, _event); \ - triton_sched_inotify(sched, a); \ - } \ - } while(0) - -static struct ae_resource triton_sched_resource = +struct triton_sched_s { - .resource_name = "sched", + ae_ops_t opqueue; }; - -triton_ret_t triton_sched_init(void) -{ - triton_ret_t ret; - ret = AE_OPCACHE_INIT(triton_sched_id, op, 1024, &sched_opcache); - if(ret != TRITON_SUCCESS) +static ae_opcache_t sched_opcache; + +static triton_mutex_t sched_done_mutex; +static ae_ops_t sched_done_queue; + +triton_ret_t triton_sched_init(triton_sched_t *sched) +{ + struct triton_sched_s *s; + s = malloc(sizeof(*s)); + if(s == NULL) { - return ret; + return TRITON_ERR_NOMEM; } - return ae_resource_register(&triton_sched_resource, &triton_sched_resource_id); + ae_ops_init(&s->opqueue); + *sched = s; + return TRITON_SUCCESS; } -void triton_sched_finalize(void) +void triton_sched_destroy(triton_sched_t sched) { - ae_resource_unregister(triton_sched_resource_id); - ae_opcache_destroy(sched_opcache); + assert(ae_ops_empty(&sched->opqueue)); + free(sched); } - -triton_ret_t triton_sched_obj_wait(triton_sched_t *sched, - triton_sched_id *result, - void (*callback)(void *ptr, int ret), + +triton_ret_t triton_sched_obj_wait(triton_sched_t sched, + triton_mutex_t *mutex, + triton_sched_id_t *result, + void (*callback)(void *ptr, triton_ret_t ret), void *user_ptr, ae_hints_t hints, ae_context_t ctx, @@ -56,7 +45,7 @@ triton_ret_t triton_sched_obj_wait(triton_sched_t *sched, if(result == NULL) { op = ae_opcache_get(sched_opcache); - result = ae_op_entry(op, triton_sched_id, op); + result = ae_op_entry(op, struct triton_sched_id_s, op); result->internal = 1; } else @@ -64,104 +53,81 @@ triton_ret_t triton_sched_obj_wait(triton_sched_t *sched, result->internal = 0; } + assert(triton_mutex_trylock(mutex) == EBUSY); + ae_op_clear(&result->op); ae_op_fill(&result->op, callback, user_ptr, hints, ctx); + result->mutex = mutex; - triton_mutex_lock(&sched->lock); - result->rank = ae_ops_count(&sched->opqueue); ae_ops_enqueue(&result->op, &sched->opqueue); - triton_sched_itrigger(sched, TRITON_SCHED_OP_ADDED); - triton_mutex_unlock(&sched->lock); + triton_mutex_unlock(result->mutex); return TRITON_SUCCESS; } -int triton_sched_notify(triton_sched_t *sched, triton_sched_action_t action) +triton_ret_t triton_sched_notify_all(triton_sched_t sched) { - switch(action) + struct ae_op *op; + while(!ae_ops_empty(&sched->opqueue)) { - case TRITON_SCHED_ACTION_NONE: - break; - case TRITON_SCHED_ACTION_NOTIFY_ALL: - triton_sched_notify_all(sched); - break; - case TRITON_SCHED_ACTION_NOTIFY_NEXT: - triton_sched_notify_next(sched); - break; - default: - return -EINVAL; - } - return 0; -} + /* remove op from scheduling queue */ + op = ae_ops_dequeue(&sched->opqueue); -static int triton_sched_notify_nolock(triton_sched_t *sched, triton_sched_action_t action) -{ - switch(action) - { - case TRITON_SCHED_ACTION_NONE: - break; - case TRITON_SCHED_ACTION_NOTIFY_ALL: - triton_sched_notify_all_nolock(sched); - break; - case TRITON_SCHED_ACTION_NOTIFY_NEXT: - triton_sched_notify_next_nolock(sched); - break; - default: - return -EINVAL; + /* add op to done queue */ + triton_mutex_lock(&sched_done_mutex); + ae_ops_enqueue(op, &sched_done_queue); + triton_mutex_unlock(&sched_done_mutex); } - return 0; -} - -static int triton_sched_notify_all_nolock(triton_sched_t *sched) + return TRITON_SUCCESS; +} + +triton_ret_t triton_sched_notify_next(triton_sched_t sched) { struct ae_op *op; - triton_sched_id *result; - if(ae_ops_empty(&sched->opqueue)) - { - return 0; - } - sched->innotify = 1; - while(!ae_ops_empty(&sched->opqueue)) + if(!ae_ops_empty(&sched->opqueue)) { op = ae_ops_dequeue(&sched->opqueue); - ae_invoke_callback(op, int, 0); - result = ae_op_entry(op, triton_sched_id, op); - if(result->internal) - { - ae_opcache_put(sched_opcache, op); - } - else - { - ae_op_clear(op); - } + + /* add op to done queue */ + triton_mutex_lock(&sched_done_mutex); + ae_ops_enqueue(op, &sched_done_queue); + triton_mutex_unlock(&sched_done_mutex); } - triton_sched_itrigger(sched, TRITON_SCHED_EMPTIED); - sched->innotify = 0; - return 0; + return TRITON_SUCCESS; } - -int triton_sched_notify_all(triton_sched_t *sched) -{ - int ret; - triton_mutex_lock(&sched->lock); - ret = triton_sched_notify_all_nolock(sched); - triton_mutex_unlock(&sched->lock); +triton_ret_t triton_sched_notify_specific(triton_sched_t sched, triton_sched_id_t *id) +{ + ae_ops_del(&id->op); + triton_mutex_lock(&sched_done_mutex); + ae_ops_enqueue(&id->op, &sched_done_queue); + triton_mutex_unlock(&sched_done_mutex); - return(ret); + return TRITON_SUCCESS; } -static int triton_sched_notify_next_nolock(triton_sched_t *sched) +static triton_ret_t triton_sched_poll(ae_context_t context, int millisecs) { struct ae_op *op; - triton_sched_id *result; - sched->innotify = 1; - if(!ae_ops_empty(&sched->opqueue)) + struct timeval start, adjust, now, timeout, diff; + struct triton_sched_id_s *result; + int est_count_before_timeout = 0; + int completed = 0; + + gettimeofday(&start, NULL); + adjust.tv_sec = (int)(millisecs / 1e3); + adjust.tv_usec = (millisecs % 1000) * 1e3; + timeradd(&start, &adjust, &timeout); + + triton_mutex_lock(&sched_done_mutex); + while(ae_ops_count(&sched_done_queue) > 0) { - op = ae_ops_dequeue(&sched->opqueue); - ae_invoke_callback(op, int, 0); - result = ae_op_entry(op, triton_sched_id, op); + op = ae_ops_dequeue(&sched_done_queue); + triton_mutex_unlock(&sched_done_mutex); + result = ae_op_entry(op, struct triton_sched_id_s, op); + triton_mutex_lock(result->mutex); + ae_invoke_callback(op, triton_ret_t, TRITON_SUCCESS); if(result->internal) { ae_opcache_put(sched_opcache, op); @@ -170,42 +136,69 @@ static int triton_sched_notify_next_nolock(triton_sched_t *sched) { ae_op_clear(op); } - triton_sched_itrigger(sched, TRITON_SCHED_OP_REMOVED); + + ++completed; + if(--est_count_before_timeout <= 0) + { + gettimeofday(&now, NULL); + if(timercmp(&timeout, &now, <)) + { + /* timeout has been reached */ + return TRITON_ERR_TIMEDOUT; + } + else + { + float mills_since_start, avg_mills; + /* didn't hit timeout, figure out how long the + * last callback took so that we know how many + * callbacks to call before checking the time + * again. We actually get the average time + * taken by all the callbacks so far, and figure + * out how many we can call in the time left. + * To avoid actually reaching the timeout, we + * only invoke 90% of the callbacks we have time + * for. + */ + timersub(&now, &start, &diff); + mills_since_start = diff.tv_sec * 1e3 + (diff.tv_usec * 1e-3); + avg_mills = mills_since_start / completed; + est_count_before_timeout = (int) (((millisecs - mills_since_start) / avg_mills) * 0.9); + } + } + + triton_mutex_lock(&sched_done_mutex); } - sched->innotify = 0; - return 0; + triton_mutex_unlock(&sched_done_mutex); + + return TRITON_SUCCESS; } -int triton_sched_notify_next(triton_sched_t *sched) +static int triton_sched_resource_id; +static struct ae_resource triton_sched_resource = +{ + .resource_name = "sched", + .poll_context = triton_sched_poll +}; + +triton_ret_t triton_sched_resource_init(void) { - int ret; + triton_ret_t ret; - triton_mutex_lock(&sched->lock); - ret = triton_sched_notify_next_nolock(sched); - triton_mutex_unlock(&sched->lock); + ret = AE_OPCACHE_INIT(struct triton_sched_id_s, op, 1024, &sched_opcache); + if(ret != TRITON_SUCCESS) + { + return ret; + } - return(ret); + ae_ops_init(&sched_done_queue); + triton_mutex_init(&sched_done_mutex, NULL); + return ae_resource_register(&triton_sched_resource, &triton_sched_resource_id); } -int triton_sched_notify_specific(triton_sched_t *sched, triton_sched_id *id) +void triton_sched_resource_finalize(void) { - triton_mutex_lock(&sched->lock); - sched->innotify = 1; - - ae_ops_del(&id->op); - ae_invoke_callback(&id->op, int, 0); - if(id->internal) - { - ae_opcache_put(sched_opcache, &id->op); - } - else - { - ae_op_clear(&id->op); - } - triton_sched_itrigger(sched, TRITON_SCHED_OP_REMOVED); - sched->innotify = 0; - triton_mutex_unlock(&sched->lock); - return 0; + ae_resource_unregister(triton_sched_resource_id); + ae_opcache_destroy(sched_opcache); } /* diff --git a/code/src/common/resources/scheduling/sched.hae b/code/src/common/resources/scheduling/sched.hae index 93931c4..edc09cc 100644 --- a/code/src/common/resources/scheduling/sched.hae +++ b/code/src/common/resources/scheduling/sched.hae @@ -2,67 +2,45 @@ #define __SCHED_HAE__ #include "src/aesop/aesop.h" +#include "src/common/triton-thread.h" -typedef struct triton_sched_entry +typedef struct triton_sched_id_s { - int rank; + triton_mutex_t *mutex; struct ae_op op; int internal; -} triton_sched_id; +} triton_sched_id_t; -typedef enum -{ - TRITON_SCHED_OP_ADDED, - TRITON_SCHED_OP_REMOVED, - TRITON_SCHED_EMPTIED -} triton_sched_trigger_event_t; - -typedef enum -{ - TRITON_SCHED_ACTION_NONE, - TRITON_SCHED_ACTION_NOTIFY_ALL, - TRITON_SCHED_ACTION_NOTIFY_NEXT -} triton_sched_action_t; - -struct triton_sched; -typedef triton_sched_action_t (*triton_sched_trigger_t)(struct triton_sched *sched, triton_sched_trigger_event_t op); - -typedef struct triton_sched -{ - ae_ops_t opqueue; - triton_mutex_t lock; - triton_sched_trigger_t trigger; - int innotify; -} triton_sched_t; - -#define triton_sched_reset(_sched) \ -do { \ - ae_ops_init(&(_sched)->opqueue); \ - triton_mutex_init(&(_sched)->lock, NULL); \ - (_sched)->trigger = NULL; \ -} while(0) - -#define triton_sched_set_trigger(sched, _trigger) (sched)->trigger = (_trigger); +typedef struct triton_sched_s *triton_sched_t; -int triton_sched_notify(triton_sched_t *sched, triton_sched_action_t action); +triton_ret_t triton_sched_resource_init(void); +void triton_sched_resource_finalize(void); -int triton_sched_notify_all(triton_sched_t *sched); +triton_ret_t triton_sched_init(triton_sched_t *sched); +void triton_sched_destroy(triton_sched_t sched); -int triton_sched_notify_next(triton_sched_t *sched); - -int triton_sched_notify_specific(triton_sched_t *sched, triton_sched_id *id); - -int triton_sched_get_entries(triton_sched_t *sched, int *count, triton_sched_id **entries); +triton_ret_t triton_sched_notify_all(triton_sched_t sched); +triton_ret_t triton_sched_notify_next(triton_sched_t sched); +triton_ret_t triton_sched_notify_specific(triton_sched_t sched, triton_sched_id_t *id); +/* gets the envelope structure pointer from the scheduling id: + * + * triton_sched_get_envelope(triton_sched_id_t id, envelope_type, sched_id_member_name); + */ #define triton_sched_get_envelope(_id, _type, _member) \ ((_type *)((char *)(_id) - (unsigned long)((&((_type *)0)->_member)))) -__blocking int triton_sched_obj_wait(triton_sched_t *sched, triton_sched_id *result); - -#define triton_sched_wait(_sched) triton_sched_obj_wait(_sched, NULL) +/** + * triton_sched_obj_wait waits on a scheduling object to be notified. The mutex passed in + * must be locked. + */ +__blocking triton_ret_t triton_sched_obj_wait(triton_sched_t sched, triton_mutex_t *mutex, triton_sched_id_t *result); -triton_ret_t triton_sched_init(void); -void triton_sched_finalize(void); +/** + * Just wait on a scheduling variable without a scheduling id. No way to wake up individuals, only notify_all or + * notify_next can be used. + */ +#define triton_sched_wait(_sched, _mutex) triton_sched_obj_wait(_sched, _mutex, NULL) #endif diff --git a/code/src/common/resources/scheduling/test/coalesce.ae b/code/src/common/resources/scheduling/test/coalesce.ae index 1647d5d..6142f4b 100644 --- a/code/src/common/resources/scheduling/test/coalesce.ae +++ b/code/src/common/resources/scheduling/test/coalesce.ae @@ -2,39 +2,51 @@ #include "src/common/resources/scheduling/sched.hae" #include "src/common/resources/timer/timer.hae" -static triton_sched_action_t coalesce_trigger(struct triton_sched *sched, triton_sched_trigger_event_t op) -{ - if(ae_ops_count(&sched->opqueue) >= 8) - { - return TRITON_SCHED_ACTION_NOTIFY_ALL; - } - return TRITON_SCHED_ACTION_NONE; -} - struct coalesce_obj { int ind; - triton_sched_id sched_id; + triton_sched_id_t sched_id; + triton_list_link_t link; }; static triton_sched_t coalesce_sched; -static triton_sched_t sync_sched; +static triton_mutex_t coalesce_mutex = TRITON_MUTEX_INITIALIZER; +static triton_list_t coalesce_queue = TRITON_LIST_STATIC_INITIALIZER(coalesce_queue); -__blocking int coalesce(struct coalesce_obj *obj); +static triton_mutex_t ops_started_mutex = TRITON_MUTEX_INITIALIZER; +static triton_list_t ops_started_queue = TRITON_LIST_STATIC_INITIALIZER(ops_started_queue); + +void begin(struct coalesce_obj *obj); +__blocking int sync(struct coalesce_obj *obj); static __blocking int dostuff(void) { - struct coalesce_obj obj[32]; + struct coalesce_obj obj[1000]; pwait { - int i; /* special pwait variable gets copied at pbranch invocation */ - for(i = 0; i < 32; ++i) + pprivate int i; /* special pwait variable gets copied at pbranch invocation */ + for(i = 0; i < 1000; ++i) { pbranch { obj[i].ind = i; - coalesce(&obj[i]); + + /* delay start of operations as if they were coming in randomly. We + * try to distribute the "operations" evenly over 100 seconds. + */ + triton_timer(rand() % 100000); + + /* needed for low watermark */ + begin(&obj[i]); + + /* ok now do some blocking thing for random amount of time, but no + * longer than 100 millisecs. + */ + triton_timer(rand() % 100); + + /* now sync! */ + sync(&obj[i]); } } } @@ -42,26 +54,71 @@ static __blocking int dostuff(void) return 0; } -__blocking int coalesce(struct coalesce_obj *obj) +void begin(struct coalesce_obj *obj) +{ + /* add obj to queue of started ops. Allows us to keep track of what's in progress. */ + triton_mutex_lock(&ops_started_mutex); + triton_queue_enqueue(&obj->link, &ops_started_queue); + triton_mutex_unlock(&ops_started_mutex); +} + +__blocking int sync(struct coalesce_obj *obj) { int ret; - ret = triton_sched_obj_wait(&coalesce_sched, &obj->sched_id); - assert(ret == 0); - printf("coalesce wait completed for index: %d\n", obj->ind); + triton_list_t to_notify; + triton_list_link_t *nop; + struct coalesce_obj *cop; + + triton_mutex_lock(&ops_started_mutex); + triton_mutex_lock(&coalesce_mutex); + + /* remove from ops_started_queue */ + triton_list_del(&obj->link); - if(obj->sched_id.rank == 0) + /* watermarks check: low: if no more ops are coming in, I need to sync the ones that are done. + * high: if more than 8 ops are coalesced, I need to sync these guys. + */ + if(triton_list_empty(&ops_started_queue) || triton_list_count(&coalesce_queue) > 8) { - printf("index: %d does the sync\n", obj->ind); + triton_mutex_unlock(&ops_started_mutex); + + /* high-watermark or low-watermark reached. Sync. */ + + /* get coalesced ops off coalescing queue, put onto local notify queue */ + triton_list_init(&to_notify); + while(!triton_list_empty(&coalesce_queue)) + { + triton_queue_enqueue(triton_queue_dequeue(&coalesce_queue), &to_notify); + } + triton_mutex_unlock(&coalesce_mutex); + + /* do faky sync */ + printf("hit low/high watermark,\tindex: %d synced\n", obj->ind); ret = triton_timer(1000); - assert(ret == 0); - ret = triton_sched_notify_all(&sync_sched); - assert(ret == 0); + assert(ret == TRITON_SUCCESS); + + /* notify waiters */ + while(!triton_list_empty(&to_notify)) + { + nop = triton_queue_dequeue(&to_notify); + cop = triton_list_get_entry(nop, struct coalesce_obj, link); + + /* notify this guy! */ + triton_mutex_lock(&coalesce_mutex); + triton_sched_notify_specific(coalesce_sched, &cop->sched_id); + triton_mutex_unlock(&coalesce_mutex); + printf("\t\tindex: %d synced\n", cop->ind); + } } else { - printf("index: %d waiting for sync to complete (rank=%d)\n", obj->ind, obj->sched_id.rank); - ret = triton_sched_wait(&sync_sched); - assert(ret == 0); + triton_mutex_unlock(&ops_started_mutex); + + /* I get coalesced. First add to coalesce queue, then wait */ + + triton_queue_enqueue(&obj->link, &coalesce_queue); + triton_sched_obj_wait(coalesce_sched, &coalesce_mutex, &obj->sched_id); + triton_mutex_unlock(&coalesce_mutex); } return 0; @@ -78,14 +135,12 @@ int main(int argc, char *argv[]) { ae_context_t ctx; - triton_sched_init(); + triton_sched_resource_init(); triton_timer_init(); ae_context_create(&ctx, 2, "sched", "timer"); - triton_sched_reset(&coalesce_sched); - triton_sched_reset(&sync_sched); - triton_sched_set_trigger(&coalesce_sched, coalesce_trigger); + triton_sched_init(&coalesce_sched); dostuff(cb,NULL,NULL,ctx,NULL); @@ -96,7 +151,7 @@ int main(int argc, char *argv[]) ae_context_destroy(ctx); triton_timer_finalize(); - triton_sched_finalize(); + triton_sched_resource_finalize(); return 0; } diff --git a/code/src/common/resources/scheduling/test/limitqueue.ae b/code/src/common/resources/scheduling/test/limitqueue.ae index 103464e..7871e10 100644 --- a/code/src/common/resources/scheduling/test/limitqueue.ae +++ b/code/src/common/resources/scheduling/test/limitqueue.ae @@ -5,7 +5,6 @@ struct limit_obj { int ind; - triton_sched_id sched_id; }; static triton_sched_t limit_sched; @@ -18,7 +17,7 @@ static __blocking int dostuff(void) pwait { - int i; /* special pwait variable gets copied at pbranch invocation */ + pprivate int i; /* special pwait variable gets copied at pbranch invocation */ for(i = 0; i < 100; ++i) { pbranch @@ -37,19 +36,16 @@ static triton_mutex_t progmut = TRITON_MUTEX_INITIALIZER; __blocking int dolimit(struct limit_obj *obj) { - int ret; + triton_ret_t ret; triton_mutex_lock(&progmut); if(inprogress > 16) { - triton_mutex_unlock(&progmut); - /* wait till an op finishes */ - ret = triton_sched_wait(&limit_sched); - assert(ret == 0); - printf("limit wait completed for index: %d\n", obj->ind); + ret = triton_sched_wait(limit_sched, &progmut); - triton_mutex_lock(&progmut); + assert(ret == TRITON_SUCCESS); + printf("limit wait completed for index: %d\n", obj->ind); } ++inprogress; @@ -58,14 +54,14 @@ __blocking int dolimit(struct limit_obj *obj) /* actually do something here, and then notify the next guy */ printf("index: %d sleeping\n", obj->ind); ret = triton_timer(100); - assert(ret == 0); + assert(ret == TRITON_SUCCESS); triton_mutex_lock(&progmut); --inprogress; - triton_mutex_lock(&progmut); - ret = triton_sched_notify_next(&limit_sched); - assert(ret == 0); + ret = triton_sched_notify_next(limit_sched); + triton_mutex_unlock(&progmut); + assert(ret == TRITON_SUCCESS); return 0; } @@ -81,12 +77,12 @@ int main(int argc, char *argv[]) { ae_context_t ctx; - triton_sched_init(); + triton_sched_resource_init(); triton_timer_init(); ae_context_create(&ctx, 2, "sched", "timer"); - triton_sched_reset(&limit_sched); + triton_sched_init(&limit_sched); dostuff(cb,NULL,NULL,ctx,NULL); @@ -97,7 +93,7 @@ int main(int argc, char *argv[]) ae_context_destroy(ctx); triton_timer_finalize(); - triton_sched_finalize(); + triton_sched_resource_finalize(); return 0; } diff --git a/code/src/common/resources/scheduling/test/module.mk.in b/code/src/common/resources/scheduling/test/module.mk.in index 8ca2fb2..a7d66d5 100644 --- a/code/src/common/resources/scheduling/test/module.mk.in +++ b/code/src/common/resources/scheduling/test/module.mk.in @@ -5,3 +5,5 @@ AETESTSRC += $(DIR)/coalesce.ae \ $(DIR)/pthread-compare.ae MODLIBS_$(DIR)/pthread-compare = -lpthread +MODLIBS_$(DIR)/coalesce = -lpthread +MODLIBS_$(DIR)/limitqueue = -lpthread diff --git a/code/src/common/resources/scheduling/test/pthread-compare.ae b/code/src/common/resources/scheduling/test/pthread-compare.ae index b6b084b..2ca6726 100644 --- a/code/src/common/resources/scheduling/test/pthread-compare.ae +++ b/code/src/common/resources/scheduling/test/pthread-compare.ae @@ -7,6 +7,7 @@ static pthread_cond_t cond; static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; static triton_sched_t sched; +static triton_mutex_t sched_mutex = TRITON_MUTEX_INITIALIZER; static __blocking int sched_test(void) { @@ -20,7 +21,9 @@ static __blocking int sched_test(void) { pbranch { - triton_sched_wait(&sched); + triton_mutex_lock(&sched_mutex); + triton_sched_wait(sched, &sched_mutex); + triton_mutex_unlock(&sched_mutex); } } } @@ -48,12 +51,12 @@ static void *thread_cb(void *up) static void thread_test(void) { int i, ret; - pthread_t tid[100]; + pthread_t tid[1000]; void *val; struct timeval t1, t2, diff; gettimeofday(&t1, NULL); - for(i = 0; i < 100; ++i) + for(i = 0; i < 1000; ++i) { ret = pthread_create(&tid[i], NULL, thread_cb, NULL); if(ret != 0) @@ -68,7 +71,7 @@ static void thread_test(void) pthread_cond_broadcast(&cond); pthread_mutex_unlock(&mutex); - for(i = 0; i < 100; ++i) + for(i = 0; i < 1000; ++i) { pthread_join(tid[i], &val); } @@ -93,21 +96,23 @@ int main(int argc, char *argv[]) { ae_context_t ctx; - triton_sched_init(); + triton_sched_resource_init(); ae_context_create(&ctx, 1, "sched"); - triton_sched_reset(&sched); + triton_sched_init(&sched); sched_test(donecb, NULL, NULL, ctx, NULL); - triton_sched_notify_all(&sched); + triton_mutex_lock(&sched_mutex); + triton_sched_notify_all(sched); + triton_mutex_unlock(&sched_mutex); while(!done) { ae_poll(ctx, 100); } ae_context_destroy(ctx); - triton_sched_finalize(); + triton_sched_resource_finalize(); pthread_cond_init(&cond, NULL); thread_test(); diff --git a/code/src/common/resources/timer/timer.c b/code/src/common/resources/timer/timer.c index ff65488..cdd9744 100644 --- a/code/src/common/resources/timer/timer.c +++ b/code/src/common/resources/timer/timer.c @@ -26,7 +26,7 @@ struct timer_op triton_ret_t triton_timer( int millisecs, - void (*callback)(void *ptr, int ret), + void (*callback)(void *ptr, triton_ret_t ret), void *user_ptr, ae_hints_t hints, ae_context_t ctx, @@ -103,7 +103,7 @@ static triton_ret_t triton_timer_poll(ae_context_t context, int millisecs) gop = ae_ops_dequeue(&timer_oplist); triton_mutex_unlock(&timer_mutex); - ae_opcache_complete_op(timer_opcache, gop, int, 0); + ae_opcache_complete_op(timer_opcache, gop, triton_ret_t, TRITON_SUCCESS); /* setup for next iteration */ triton_mutex_lock(&timer_mutex); @@ -148,7 +148,7 @@ static triton_ret_t triton_timer_poll(ae_context_t context, int millisecs) gop = ae_ops_dequeue(&timer_oplist); triton_mutex_unlock(&timer_mutex); - ae_opcache_complete_op(timer_opcache, gop, int, 0); + ae_opcache_complete_op(timer_opcache, gop, triton_ret_t, TRITON_SUCCESS); /* setup for next iteration */ triton_mutex_lock(&timer_mutex); @@ -167,7 +167,7 @@ static triton_ret_t triton_timer_poll(ae_context_t context, int millisecs) nanosleep(&ts_diff, &rem); } - return 0; + return TRITON_SUCCESS; } static triton_ret_t triton_timer_cancel(ae_context_t triton_ctx, ae_op_id_t op_id) diff --git a/code/src/common/resources/timer/timer.hae b/code/src/common/resources/timer/timer.hae index 3871380..b33069c 100644 --- a/code/src/common/resources/timer/timer.hae +++ b/code/src/common/resources/timer/timer.hae @@ -3,7 +3,7 @@ #include "src/aesop/aesop.h" -__blocking int triton_timer(int millisecs); +__blocking triton_ret_t triton_timer(int millisecs); triton_ret_t triton_timer_init(void); void triton_timer_finalize(void); diff --git a/code/src/common/triton-thread.h b/code/src/common/triton-thread.h index 9f54371..40dbfc1 100644 --- a/code/src/common/triton-thread.h +++ b/code/src/common/triton-thread.h @@ -10,6 +10,7 @@ typedef pthread_mutex_t triton_mutex_t; typedef pthread_mutexattr_t triton_mutexattr_t; #define triton_mutex_lock(__mut) pthread_mutex_lock(__mut) #define triton_mutex_unlock(__mut) pthread_mutex_unlock(__mut) +#define triton_mutex_trylock(__mut) pthread_mutex_trylock(__mut) #define triton_mutex_init(__mut, __attr) pthread_mutex_init(__mut, __attr) #define triton_mutex_destroy(__mut) pthread_mutex_destroy(__mut) #define TRITON_MUTEX_INITIALIZER PTHREAD_MUTEX_INITIALIZER @@ -33,6 +34,7 @@ typedef int triton_mutex_t; typedef int triton_mutexattr_t; static inline int triton_mutex_lock(triton_mutex_t *__mut) { return 0; } static inline int triton_mutex_unlock(triton_mutex_t *__mut) { return 0; } +static inline int triton_mutex_trylock(triton_mutex_t *__mut) { return 0; } static inline int triton_mutex_init(triton_mutex_t *__mut, triton_mutexattr_t *__attr) {return 0; } static inline int triton_mutex_destroy(triton_mutex_t *__mut) {return 0; } #define TRITON_MUTEX_INITIALIZER 0 diff --git a/code/src/common/triton-types.h b/code/src/common/triton-types.h index 4da1b79..64df71e 100644 --- a/code/src/common/triton-types.h +++ b/code/src/common/triton-types.h @@ -11,7 +11,7 @@ typedef struct { uint64_t l; uint64_t u; } uint128_t; #include <sys/time.h> -#define TRITON_REL_MSECS_TO_ABS_TIMESPEC(__msecs, __abs) \ +#define TRITON_REL_MSECS_TO_ABS_TIMESPEC(__msecs, __abs) \ do { \ struct timeval __now; \ gettimeofday(&__now, NULL); \ hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. 48ac98c8ebbd3e9a3d8d4c2a6bb355d2955bed6b
by noreply＠mcs.anl.gov 19 May '10

19 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via 48ac98c8ebbd3e9a3d8d4c2a6bb355d2955bed6b (commit) from cdd5d06ce4bc879d1be0d57f92d895a9e8c23d93 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit 48ac98c8ebbd3e9a3d8d4c2a6bb355d2955bed6b Author: Justin Wozniak <wozniak(a)mcs.anl.gov> Date: Wed May 19 13:23:50 2010 -0500 Bug fix to bqueue & test ----------------------------------------------------------------------- Summary of changes: code/src/fakess/bqueue.ae | 17 ++++++++++++----- code/src/fakess/module.mk.in | 2 +- code/src/fakess/test/module.mk.in | 2 +- code/src/fakess/test/testbqueue.ae | 2 +- 4 files changed, 15 insertions(+), 8 deletions(-) Diff of changes: diff --git a/code/src/fakess/bqueue.ae b/code/src/fakess/bqueue.ae index 5a8349c..52ef5b6 100644 --- a/code/src/fakess/bqueue.ae +++ b/code/src/fakess/bqueue.ae @@ -14,7 +14,6 @@ triton_ret_t triton_bqueue_enqueue(triton_list_link_t *link, triton_mutex_lock(&(queue->mutex)); triton_list_add_back(link, &(queue->list)); triton_mutex_unlock(&(queue->mutex)); - printf("notifying: %p\n", &(queue->sched)); triton_sched_notify(&(queue->sched), TRITON_SCHED_ACTION_NOTIFY_ALL); return TRITON_SUCCESS; @@ -31,17 +30,25 @@ triton_ret_t triton_bqueue_push(triton_list_link_t *link, return TRITON_SUCCESS; } +triton_list_link_t *_triton_bqueue_dequeue(triton_bqueue_t *queue) +{ + triton_list_link_t *result = NULL; + triton_mutex_lock(&(queue->mutex)); + result = triton_queue_dequeue(&(queue->list)); + triton_mutex_unlock(&(queue->mutex)); + return result; +} + __blocking triton_list_link_t *triton_bqueue_dequeue(triton_bqueue_t *queue) { triton_list_link_t *result = NULL; - while (result == NULL) + while (! (result = _triton_bqueue_dequeue(queue))) { - printf("waiting: %p\n", &(queue->sched)); triton_sched_wait(&(queue->sched)); - result = triton_queue_dequeue(&(queue->list)); } - return NULL; + + return result; } triton_ret_t triton_bqueue_finalize(triton_bqueue_t *queue) diff --git a/code/src/fakess/module.mk.in b/code/src/fakess/module.mk.in index 1ce5377..f59e8d0 100644 --- a/code/src/fakess/module.mk.in +++ b/code/src/fakess/module.mk.in @@ -3,7 +3,7 @@ DIR := src/fakess ifneq (,$(BUILD_MPI)) AELIBSRC += $(DIR)/fakess.ae -#AELIBSRC += $(DIR)/bqueue.ae +AELIBSRC += $(DIR)/bqueue.ae MODCFLAGS_$(DIR)/fakess = $(MPICFLAGS) MODLDFLAGS_$(DIR)/fakess = $(MPILDFLAGS) diff --git a/code/src/fakess/test/module.mk.in b/code/src/fakess/test/module.mk.in index e98a6b8..c28a38a 100644 --- a/code/src/fakess/test/module.mk.in +++ b/code/src/fakess/test/module.mk.in @@ -4,7 +4,7 @@ DIR = src/fakess/test #AETESTSRC += $(DIR)/wait.ae #AETESTSRC += $(DIR)/post-global.ae #AETESTSRC += $(DIR)/listen-global.ae -#AETESTSRC += $(DIR)/testbqueue.ae +AETESTSRC += $(DIR)/testbqueue.ae MODCFLAGS_$(DIR)/setup = $(MPICFLAGS) MODLDFLAGS_$(DIR)/setup = $(MPILDFLAGS) diff --git a/code/src/fakess/test/testbqueue.ae b/code/src/fakess/test/testbqueue.ae index 2ca73cd..6a03a7c 100644 --- a/code/src/fakess/test/testbqueue.ae +++ b/code/src/fakess/test/testbqueue.ae @@ -19,7 +19,7 @@ static __blocking int dostuff(void) pwait { int i; /* special pwait variable gets copied at pbranch invocation */ - for(i = 0; i < 2; ++i) + for(i = 0; i < 4; ++i) { pbranch { hooks/post-receive -- Triton Repository

1 0

Triton Repository branch, master, updated. cdd5d06ce4bc879d1be0d57f92d895a9e8c23d93
by noreply＠mcs.anl.gov 19 May '10

19 May '10

This is an automated email from the git hooks/post-receive script. It was generated because a ref change was pushed to the repository containing the project "Triton Repository". The branch, master has been updated via cdd5d06ce4bc879d1be0d57f92d895a9e8c23d93 (commit) via 8af429f8f3962768944a651f0b38c91665d122fc (commit) via 2f34b97cb2a7b56abebb9d706a289fdacc678387 (commit) via c2215396cc143d8faabafe1b8825c30df8db6a5a (commit) from 769a9e0f76910a0ef8934030914cdd94fde4b553 (commit) Those revisions listed above that are new to this repository have not appeared on any other notification email; so we list those revisions in full, below. - Log ----------------------------------------------------------------- commit cdd5d06ce4bc879d1be0d57f92d895a9e8c23d93 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Wed May 19 12:54:19 2010 -0500 steps towards using triton_sched for coalescing commit 8af429f8f3962768944a651f0b38c91665d122fc Author: Phil Carns <carns(a)mcs.anl.gov> Date: Wed May 19 12:40:55 2010 -0500 comment out debugging printfs commit 2f34b97cb2a7b56abebb9d706a289fdacc678387 Author: Phil Carns <carns(a)mcs.anl.gov> Date: Wed May 19 09:45:16 2010 -0500 reorganize locks to avoid deadlock in scheduler commit c2215396cc143d8faabafe1b8825c30df8db6a5a Author: Phil Carns <carns(a)mcs.anl.gov> Date: Tue May 18 22:13:33 2010 -0500 enable large file support by default ----------------------------------------------------------------------- Summary of changes: code/Makefile.in | 2 + code/src/common/resources/scheduling/sched.c | 55 ++++++++++++++++--- .../prototype/tests/vosd-bench-concurrent.ae | 3 +- code/src/versioned-osd/prototype/versioned-osd.ae | 58 +++++++++++++++++++- 4 files changed, 107 insertions(+), 11 deletions(-) Diff of changes: diff --git a/code/Makefile.in b/code/Makefile.in index 0974bb5..0b1731f 100644 --- a/code/Makefile.in +++ b/code/Makefile.in @@ -59,6 +59,8 @@ BUILD_LDFLAGS = @BUILD_LDFLAGS@ STRICT_CFLAGS = @STRICT_CFLAGS@ CFLAGS= -I$(srcdir)/include -I. -I$(srcdir) @CFLAGS@ @CPPFLAGS@ +# large file support +CFLAGS += -D_FILE_OFFSET_BITS=64 -D_LARGEFILE64_SOURCE CWARNS= ifneq (,$(STRICT_CFLAGS)) CWARNS += -Wcast-align -Wbad-function-cast diff --git a/code/src/common/resources/scheduling/sched.c b/code/src/common/resources/scheduling/sched.c index b7c2f7e..c22235d 100644 --- a/code/src/common/resources/scheduling/sched.c +++ b/code/src/common/resources/scheduling/sched.c @@ -5,9 +5,12 @@ static int triton_sched_resource_id; static ae_opcache_t sched_opcache; +static int triton_sched_notify_nolock(triton_sched_t *sched, triton_sched_action_t action); +static int triton_sched_notify_all_nolock(triton_sched_t *sched); +static int triton_sched_notify_next_nolock(triton_sched_t *sched); #define triton_sched_inotify(_sched, _action) \ - if(_action != TRITON_SCHED_ACTION_NONE) triton_sched_notify(_sched, _action) + if(_action != TRITON_SCHED_ACTION_NONE) triton_sched_notify_nolock(_sched, _action) #define triton_sched_itrigger(_sched, _event) \ do { \ @@ -91,14 +94,31 @@ int triton_sched_notify(triton_sched_t *sched, triton_sched_action_t action) return 0; } -int triton_sched_notify_all(triton_sched_t *sched) +static int triton_sched_notify_nolock(triton_sched_t *sched, triton_sched_action_t action) +{ + switch(action) + { + case TRITON_SCHED_ACTION_NONE: + break; + case TRITON_SCHED_ACTION_NOTIFY_ALL: + triton_sched_notify_all_nolock(sched); + break; + case TRITON_SCHED_ACTION_NOTIFY_NEXT: + triton_sched_notify_next_nolock(sched); + break; + default: + return -EINVAL; + } + return 0; +} + + +static int triton_sched_notify_all_nolock(triton_sched_t *sched) { struct ae_op *op; triton_sched_id *result; - triton_mutex_lock(&sched->lock); if(ae_ops_empty(&sched->opqueue)) { - triton_mutex_unlock(&sched->lock); return 0; } sched->innotify = 1; @@ -118,15 +138,24 @@ int triton_sched_notify_all(triton_sched_t *sched) } triton_sched_itrigger(sched, TRITON_SCHED_EMPTIED); sched->innotify = 0; - triton_mutex_unlock(&sched->lock); return 0; } -int triton_sched_notify_next(triton_sched_t *sched) +int triton_sched_notify_all(triton_sched_t *sched) +{ + int ret; + + triton_mutex_lock(&sched->lock); + ret = triton_sched_notify_all_nolock(sched); + triton_mutex_unlock(&sched->lock); + + return(ret); +} + +static int triton_sched_notify_next_nolock(triton_sched_t *sched) { struct ae_op *op; triton_sched_id *result; - triton_mutex_lock(&sched->lock); sched->innotify = 1; if(!ae_ops_empty(&sched->opqueue)) { @@ -144,10 +173,20 @@ int triton_sched_notify_next(triton_sched_t *sched) triton_sched_itrigger(sched, TRITON_SCHED_OP_REMOVED); } sched->innotify = 0; - triton_mutex_unlock(&sched->lock); return 0; } +int triton_sched_notify_next(triton_sched_t *sched) +{ + int ret; + + triton_mutex_lock(&sched->lock); + ret = triton_sched_notify_next_nolock(sched); + triton_mutex_unlock(&sched->lock); + + return(ret); +} + int triton_sched_notify_specific(triton_sched_t *sched, triton_sched_id *id) { triton_mutex_lock(&sched->lock); diff --git a/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae b/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae index 36eff31..15d61ad 100644 --- a/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae +++ b/code/src/versioned-osd/prototype/tests/vosd-bench-concurrent.ae @@ -132,13 +132,12 @@ static __blocking int do_vosd_test(void) } free(buffer); - printf("Finished branch %d of %d\n", (i+1), concurrent); + /* printf("Finished branch %d of %d\n", (i+1), concurrent); */ } } } - printf("After pwait.\n"); end_tm = Wtime(); #if 0 diff --git a/code/src/versioned-osd/prototype/versioned-osd.ae b/code/src/versioned-osd/prototype/versioned-osd.ae index d85eadd..d735175 100644 --- a/code/src/versioned-osd/prototype/versioned-osd.ae +++ b/code/src/versioned-osd/prototype/versioned-osd.ae @@ -20,6 +20,7 @@ #include "src/versioned-osd/prototype/file-resource/file-resource.hae" #include "src/versioned-osd/prototype/versioned-osd.hae" #include "src/versioned-osd/prototype/vosd-fd-cache.hae" +#include "src/common/resources/scheduling/sched.hae" /* This file contains a prototype implementation of the VOSD API that uses @@ -58,6 +59,8 @@ static DB *missing_ver_dbp = NULL; static DB *ver_dbp = NULL; static struct triton_hash_table *txn_table = NULL; static triton_mutex_t txn_mutex = TRITON_MUTEX_INITIALIZER; +static int txn_pending_count = 0; +static triton_sched_t coalesce_sched; /* db entry representing region in logical map */ /* indexed via last offset */ @@ -143,6 +146,7 @@ static triton_ret_t merge_logical_map( struct triton_list* list2_in, struct triton_list* list_out, void** free_ptr); static void free_logical_map_list(struct triton_list* list); +static triton_sched_action_t coalesce_trigger(struct triton_sched *sched, triton_sched_trigger_event_t op); /* TODO: prototype gives compile errors; have to define function here * instead @@ -1032,6 +1036,9 @@ triton_ret_t vosd_init(const char* path) uint32_t env_flags; uint32_t open_flags; + triton_sched_reset(&coalesce_sched); + triton_sched_set_trigger(&coalesce_sched, coalesce_trigger); + /* initialize BDB resource */ tret = bdb_init(TRITON_BDB_PROG_THREAD_PER_OP); if(tret != TRITON_SUCCESS) @@ -1532,6 +1539,7 @@ triton_ret_t vosd_txn_open( triton_list_init(&txn_acc->update_list); triton_mutex_lock(&txn_mutex); + txn_pending_count++; /* TODO: safety test for duplicate txn numbers */ triton_hash_add(txn_table, &txn_acc->txn_number, &txn_acc->hash_link); triton_mutex_unlock(&txn_mutex); @@ -1756,6 +1764,8 @@ __blocking triton_ret_t vosd_txn_close( triton_ret_t tret; struct txn_accumulator* txn_acc; struct triton_hash_link* hash_link; + int ret = 0; + triton_sched_id sched_id; /* pull txn accumulator out of hash so no one can touch it */ triton_mutex_lock(&txn_mutex); @@ -1788,8 +1798,31 @@ __blocking triton_ret_t vosd_txn_close( tret = __vosd_txn_close_retryable(txn_acc); } + /* NOTE: need to think about race conditions before proceeding down + * the coalescing path + */ +#if 0 + /* TODO: why is the return code an int here? */ + ret = triton_sched_obj_wait(&coalesce_sched, &sched_id); + + /* TODO: error handling */ + assert(ret == 0); + if(sched_id.rank == 0) + { + /* checkpoint bdb */ + } + else + { + /* wait on rank 0 to checkpoint bdb */ + } +#endif + + txn_pending_count--; + /* TODO: any cleanup steps if we failed to close the txn? */ - /* For now just assume that we leave the txn as is */ + /* For now just assume that we leave the txn as is. BDB semantics + * would destroy the txn. + */ return(tret); } @@ -2278,6 +2311,29 @@ static void free_logical_map_list(struct triton_list* list) return; } +/* callback function to check when coalescing should be triggered */ +static triton_sched_action_t coalesce_trigger(struct triton_sched *sched, triton_sched_trigger_event_t op) +{ + int count; + triton_sched_action_t sret; + + triton_mutex_lock(&txn_mutex); + + count = ae_ops_count(&sched->opqueue); + /* if there are no more txns to wait for, _or_ if we have at least 8 + * txns waiting already, then go ahead and checkpoint + */ + if(count == txn_pending_count || count >= 8) + sret = TRITON_SCHED_ACTION_NOTIFY_ALL; + else + sret = TRITON_SCHED_ACTION_NONE; + + triton_mutex_unlock(&txn_mutex); + + return(sret); +} + + /* * Local variables: * c-indent-level: 4 hooks/post-receive -- Triton Repository

1 0