From f0088aa7635d3fca6de6144b9eb21814c9bc2d4e Mon Sep 17 00:00:00 2001 From: Winter Date: Sat, 30 Jan 2016 13:09:34 +0100 Subject: [PATCH] mtapi_c: changed task state attribute to atomic mtapi_network_c: signaling of task completion is now atomic --- mtapi_c/src/embb_mtapi_scheduler_t.c | 11 +++++++---- mtapi_c/src/embb_mtapi_task_context_t.c | 2 +- mtapi_c/src/embb_mtapi_task_t.c | 4 ++-- mtapi_c/src/embb_mtapi_task_t.h | 2 +- mtapi_network_c/src/embb_mtapi_network.c | 4 ++-- 5 files changed, 13 insertions(+), 10 deletions(-) diff --git a/mtapi_c/src/embb_mtapi_scheduler_t.c b/mtapi_c/src/embb_mtapi_scheduler_t.c index acae760..2cdef50 100644 --- a/mtapi_c/src/embb_mtapi_scheduler_t.c +++ b/mtapi_c/src/embb_mtapi_scheduler_t.c @@ -302,7 +302,7 @@ int embb_mtapi_scheduler_worker(void * arg) { node->queue_pool, task->queue); } - switch (task->state) { + switch (embb_atomic_load_int(&task->state)) { case MTAPI_TASK_SCHEDULED: /* multi-instance task, another instance might be running */ case MTAPI_TASK_RUNNING: @@ -398,10 +398,11 @@ mtapi_boolean_t embb_mtapi_scheduler_wait_for_task( node->scheduler); /* now wait and schedule new tasks if we are on a worker */ + mtapi_task_state_t task_state = embb_atomic_load_int(&task->state); while ( - (MTAPI_TASK_SCHEDULED == task->state) || - (MTAPI_TASK_RUNNING == task->state) || - (MTAPI_TASK_RETAINED == task->state) ) { + (MTAPI_TASK_SCHEDULED == task_state) || + (MTAPI_TASK_RUNNING == task_state) || + (MTAPI_TASK_RETAINED == task_state) ) { if (MTAPI_INFINITE < timeout) { embb_time_t current_time; embb_time_now(¤t_time); @@ -416,6 +417,8 @@ mtapi_boolean_t embb_mtapi_scheduler_wait_for_task( node->scheduler, node, context); + + task_state = embb_atomic_load_int(&task->state); } return MTAPI_TRUE; diff --git a/mtapi_c/src/embb_mtapi_task_context_t.c b/mtapi_c/src/embb_mtapi_task_context_t.c index a74088a..1af899f 100644 --- a/mtapi_c/src/embb_mtapi_task_context_t.c +++ b/mtapi_c/src/embb_mtapi_task_context_t.c @@ -187,7 +187,7 @@ mtapi_task_state_t mtapi_context_taskstate_get( &(task_context->thread_context->tss_id)); if (local_context == task_context->thread_context) { - task_state = task_context->task->state; + task_state = embb_atomic_load_int(&task_context->task->state); local_status = MTAPI_SUCCESS; } else { local_status = MTAPI_ERR_CONTEXT_OUTOFCONTEXT; diff --git a/mtapi_c/src/embb_mtapi_task_t.c b/mtapi_c/src/embb_mtapi_task_t.c index 9a8649e..d59731c 100644 --- a/mtapi_c/src/embb_mtapi_task_t.c +++ b/mtapi_c/src/embb_mtapi_task_t.c @@ -79,7 +79,7 @@ void embb_mtapi_task_initialize(embb_mtapi_task_t* that) { that->action.id = EMBB_MTAPI_IDPOOL_INVALID_ID; that->job.id = EMBB_MTAPI_IDPOOL_INVALID_ID; - that->state = MTAPI_TASK_ERROR; + embb_atomic_store_int(&that->state, MTAPI_TASK_ERROR); that->task_id = MTAPI_TASK_ID_NONE; that->group.id = EMBB_MTAPI_IDPOOL_INVALID_ID; that->queue.id = EMBB_MTAPI_IDPOOL_INVALID_ID; @@ -159,7 +159,7 @@ void embb_mtapi_task_set_state( assert(MTAPI_NULL != that); embb_spin_lock(&that->state_lock); - that->state = state; + embb_atomic_store_int(&that->state, state); embb_atomic_memory_barrier(); embb_spin_unlock(&that->state_lock); } diff --git a/mtapi_c/src/embb_mtapi_task_t.h b/mtapi_c/src/embb_mtapi_task_t.h index ee48853..be47d6c 100644 --- a/mtapi_c/src/embb_mtapi_task_t.h +++ b/mtapi_c/src/embb_mtapi_task_t.h @@ -66,7 +66,7 @@ struct embb_mtapi_task_struct { mtapi_action_hndl_t action; embb_spinlock_t state_lock; - volatile mtapi_task_state_t state; + embb_atomic_int state; embb_atomic_unsigned_int current_instance; embb_atomic_unsigned_int instances_todo; diff --git a/mtapi_network_c/src/embb_mtapi_network.c b/mtapi_network_c/src/embb_mtapi_network.c index 598fe75..be78205 100644 --- a/mtapi_network_c/src/embb_mtapi_network.c +++ b/mtapi_network_c/src/embb_mtapi_network.c @@ -377,7 +377,7 @@ static int embb_mtapi_network_thread(void * args) { assert(err == results_size); local_task->error_code = (mtapi_status_t)task_status; - local_task->state = MTAPI_TASK_COMPLETED; + embb_atomic_store_int(&local_task->state, MTAPI_TASK_COMPLETED); embb_atomic_fetch_and_add_int(&local_action->num_tasks, -1); /* is task associated with a group? */ @@ -530,7 +530,7 @@ static void network_task_start( assert(err == send_buf->size); embb_atomic_fetch_and_add_int(&local_action->num_tasks, 1); - local_task->state = MTAPI_TASK_RUNNING; + embb_atomic_store_int(&local_task->state, MTAPI_TASK_RUNNING); embb_mtapi_network_buffer_clear(send_buf); -- libgit2 0.26.0