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 3d203e8e837f8f5cf86ba3781db95d21636c3aba (commit) from 63274c9ccf59b18ffdd909e6fc67036ce327e252 (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 3d203e8e837f8f5cf86ba3781db95d21636c3aba Author: Philip Carns <[email protected]> Date: Tue Aug 31 16:22:53 2010 -0500 short term reorg of some mpi mutexes etc. ----------------------------------------------------------------------- Summary of changes: code/src/net/mpi/mpi.c | 64 +++++++++++++++++++++++++++++++++++++---------- 1 files changed, 50 insertions(+), 14 deletions(-) Diff of changes: diff --git a/code/src/net/mpi/mpi.c b/code/src/net/mpi/mpi.c index 4f8ccbc..01805cf 100644 --- a/code/src/net/mpi/mpi.c +++ b/code/src/net/mpi/mpi.c @@ -58,12 +58,10 @@ static int mpi_initialized = 0; static int double_array_size(struct mpi_context *ctx) { int i; - triton_mutex_lock(&ctx->array_mutex); ctx->array_size *= 2; ctx->reqarray = realloc(ctx->reqarray, sizeof(*ctx->reqarray) * ctx->array_size); if(!ctx->reqarray) { - triton_mutex_unlock(&ctx->array_mutex); return -ENOMEM; } for(i = ctx->array_size/2; i < ctx->array_size; ++i) @@ -74,16 +72,13 @@ static int double_array_size(struct mpi_context *ctx) ctx->indexarray = realloc(ctx->indexarray, sizeof(*ctx->indexarray) * ctx->array_size); if(!ctx->indexarray) { - triton_mutex_unlock(&ctx->array_mutex); return -ENOMEM; } ctx->statusarray = realloc(ctx->statusarray, sizeof(*ctx->statusarray) * ctx->array_size); if(!ctx->statusarray) { - triton_mutex_unlock(&ctx->array_mutex); return -ENOMEM; } - triton_mutex_unlock(&ctx->array_mutex); return 0; } @@ -123,10 +118,12 @@ static int add_op(MPI_Request req, /* kind of a hack for mpi case, if size of opcache doubles, we double * the request and index arrays too */ + triton_mutex_lock(&mpi_ctx->array_mutex); if(ae_opcache_size(mpi_ctx->opcache) > mpi_ctx->array_size) { double_array_size(mpi_ctx); } + triton_mutex_unlock(&mpi_ctx->array_mutex); mpi_op = ae_op_entry(op, struct mpi_op, op); mpi_op->status = status; @@ -136,9 +133,12 @@ static int add_op(MPI_Request req, mpi_op->op_id = ae_id_gen(mpi_resource_id, (uint64_t)(op->cache_id)); triton_mutex_lock(&mpi_ctx->oplist_mutex); ae_ops_enqueue(op, &mpi_ctx->oplist); + triton_mutex_unlock(&mpi_ctx->oplist_mutex); + + triton_mutex_lock(&mpi_ctx->array_mutex); mpi_ctx->reqarray[op->cache_id] = req; mpi_op->request = &(mpi_ctx->reqarray[op->cache_id]); - triton_mutex_unlock(&mpi_ctx->oplist_mutex); + triton_mutex_unlock(&mpi_ctx->array_mutex); *op_id = mpi_op->op_id; return 0; } @@ -184,13 +184,13 @@ static int set_ops_in_progress(ae_context_t ctx) /* if no ops in progress, the poll function may be sleeping, so we * send a signal to wake it up and test */ + triton_mutex_lock(&mpi_ctx->wait_mutex); if(mpi_ctx->ops_in_progress == 0) { - triton_mutex_lock(&mpi_ctx->wait_mutex); mpi_ctx->ops_in_progress = 1; triton_cond_signal(&mpi_ctx->wait_cond); - triton_mutex_unlock(&mpi_ctx->wait_mutex); } + triton_mutex_unlock(&mpi_ctx->wait_mutex); return 0; } @@ -397,13 +397,25 @@ ae_define_post(int, triton_mpi_recv, return TRITON_SUCCESS; } +/* struct to track callback data when we have multiple mpi completions */ +struct mpi_callback_info +{ + int mpi_errno; + void (*callback)(void *, int); + void *user_ptr; +}; + static triton_ret_t mpi_context_poll(struct mpi_context *mpi_ctx, int millisecs) { int ret, outcount, test_count; struct timespec t; struct ae_op *op; struct mpi_op *mpi_op; + struct mpi_callback_info* callback_array = NULL; + int callback_count = 0; + int i; + triton_mutex_lock(&mpi_ctx->array_mutex); test_count = ae_opcache_count(mpi_ctx->opcache); /* Test for completion of operations */ @@ -435,6 +447,7 @@ static triton_ret_t mpi_context_poll(struct mpi_context *mpi_ctx, int millisecs) /* ok, no ops posted, so we can wait */ mpi_ctx->ops_in_progress = 0; + triton_mutex_unlock(&mpi_ctx->array_mutex); /* timedwait takes absolute time */ TRITON_REL_MSECS_TO_ABS_TIMESPEC(millisecs, t); ret = triton_cond_timedwait(&mpi_ctx->wait_cond, &mpi_ctx->wait_mutex, &t); @@ -447,6 +460,7 @@ static triton_ret_t mpi_context_poll(struct mpi_context *mpi_ctx, int millisecs) return triton_error_from_errno(ret); } } + triton_mutex_lock(&mpi_ctx->array_mutex); test_count = ae_opcache_count(mpi_ctx->opcache); @@ -464,15 +478,17 @@ static triton_ret_t mpi_context_poll(struct mpi_context *mpi_ctx, int millisecs) if(ret == MPI_SUCCESS || ret == MPI_ERR_IN_STATUS) { - int i; + if(outcount > 0) + { + callback_array = malloc(outcount*sizeof(*callback_array)); + assert(callback_array); + } for(i = 0; i < outcount; ++i) { int mpi_errno = MPI_SUCCESS; if(mpi_ctx->reqarray[mpi_ctx->indexarray[i]] == MPI_REQUEST_NULL) { - void (*callback)(void *, int); - void *user_ptr; int cancelled; MPI_Test_cancelled(&mpi_ctx->statusarray[i], &cancelled); if(cancelled) @@ -493,10 +509,17 @@ static triton_ret_t mpi_context_poll(struct mpi_context *mpi_ctx, int millisecs) memcpy((mpi_op->status), &(mpi_ctx->statusarray[i]), sizeof(MPI_Status)); } - callback = op->callback; - user_ptr = op->user_ptr; remove_op(mpi_ctx, mpi_op->op_id); - callback(user_ptr, mpi_errno); + + /* Copy information needed for each callback so that we + * can trigger them after releasing mutex that protects + * reqarray. + */ + callback_array[callback_count].mpi_errno = mpi_errno; + callback_array[callback_count].callback = op->callback; + callback_array[callback_count].user_ptr = op->user_ptr; + callback_count++; + /* callback(user_ptr, mpi_errno); */ } } } @@ -506,9 +529,22 @@ static triton_ret_t mpi_context_poll(struct mpi_context *mpi_ctx, int millisecs) goto error_exit; } + triton_mutex_unlock(&mpi_ctx->array_mutex); + + /* trigger callbacks for all completed operations */ + for(i=0; i<callback_count; i++) + { + callback_array[i].callback(callback_array[i].user_ptr, callback_array[i].mpi_errno); + } + if(callback_array) + free(callback_array); + return TRITON_SUCCESS; error_exit: + triton_mutex_unlock(&mpi_ctx->array_mutex); + if(callback_array) + free(callback_array); return triton_error_from_mpi(ret); } hooks/post-receive -- Triton Repository