|
@@ -38,24 +38,25 @@ static struct _starpu_mpi_req *_starpu_mpi_isend_common(starpu_data_handle_t dat
|
|
|
int dest, int mpi_tag, MPI_Comm comm,
|
|
int dest, int mpi_tag, MPI_Comm comm,
|
|
|
unsigned detached, void (*callback)(void *), void *arg);
|
|
unsigned detached, void (*callback)(void *), void *arg);
|
|
|
static struct _starpu_mpi_req *_starpu_mpi_irecv_common(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, unsigned detached, void (*callback)(void *), void *arg);
|
|
static struct _starpu_mpi_req *_starpu_mpi_irecv_common(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, unsigned detached, void (*callback)(void *), void *arg);
|
|
|
|
|
+static void _starpu_mpi_handle_detached_request(struct _starpu_mpi_req *req);
|
|
|
|
|
|
|
|
/* The list of requests that have been newly submitted by the application */
|
|
/* The list of requests that have been newly submitted by the application */
|
|
|
static struct _starpu_mpi_req_list *new_requests;
|
|
static struct _starpu_mpi_req_list *new_requests;
|
|
|
|
|
|
|
|
/* The list of detached requests that have already been submitted to MPI */
|
|
/* The list of detached requests that have already been submitted to MPI */
|
|
|
static struct _starpu_mpi_req_list *detached_requests;
|
|
static struct _starpu_mpi_req_list *detached_requests;
|
|
|
-static pthread_mutex_t detached_requests_mutex;
|
|
|
|
|
|
|
+static _starpu_pthread_mutex_t detached_requests_mutex;
|
|
|
|
|
|
|
|
/* Condition to wake up progression thread */
|
|
/* Condition to wake up progression thread */
|
|
|
-static pthread_cond_t cond_progression;
|
|
|
|
|
|
|
+static _starpu_pthread_cond_t cond_progression;
|
|
|
/* Condition to wake up waiting for all current MPI requests to finish */
|
|
/* Condition to wake up waiting for all current MPI requests to finish */
|
|
|
-static pthread_cond_t cond_finished;
|
|
|
|
|
-static pthread_mutex_t mutex;
|
|
|
|
|
|
|
+static _starpu_pthread_cond_t cond_finished;
|
|
|
|
|
+static _starpu_pthread_mutex_t mutex;
|
|
|
static pthread_t progress_thread;
|
|
static pthread_t progress_thread;
|
|
|
static int running = 0;
|
|
static int running = 0;
|
|
|
|
|
|
|
|
/* Count requests posted by the application and not yet submitted to MPI, i.e pushed into the new_requests list */
|
|
/* Count requests posted by the application and not yet submitted to MPI, i.e pushed into the new_requests list */
|
|
|
-static pthread_mutex_t mutex_posted_requests;
|
|
|
|
|
|
|
+static _starpu_pthread_mutex_t mutex_posted_requests;
|
|
|
static int posted_requests = 0, newer_requests, barrier_running = 0;
|
|
static int posted_requests = 0, newer_requests, barrier_running = 0;
|
|
|
|
|
|
|
|
#define _STARPU_MPI_INC_POSTED_REQUESTS(value) { _STARPU_PTHREAD_MUTEX_LOCK(&mutex_posted_requests); posted_requests += value; _STARPU_PTHREAD_MUTEX_UNLOCK(&mutex_posted_requests); }
|
|
#define _STARPU_MPI_INC_POSTED_REQUESTS(value) { _STARPU_PTHREAD_MUTEX_LOCK(&mutex_posted_requests); posted_requests += value; _STARPU_PTHREAD_MUTEX_UNLOCK(&mutex_posted_requests); }
|
|
@@ -74,11 +75,11 @@ static struct _starpu_mpi_req *_starpu_mpi_isend_irecv_common(starpu_data_handle
|
|
|
enum starpu_access_mode mode)
|
|
enum starpu_access_mode mode)
|
|
|
{
|
|
{
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
struct _starpu_mpi_req *req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
struct _starpu_mpi_req *req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
|
STARPU_ASSERT(req);
|
|
STARPU_ASSERT(req);
|
|
|
|
|
|
|
|
- _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
|
|
|
|
+ _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
|
|
|
|
|
/* Initialize the request structure */
|
|
/* Initialize the request structure */
|
|
|
req->submitted = 0;
|
|
req->submitted = 0;
|
|
@@ -101,10 +102,10 @@ static struct _starpu_mpi_req *_starpu_mpi_isend_irecv_common(starpu_data_handle
|
|
|
|
|
|
|
|
/* Asynchronously request StarPU to fetch the data in main memory: when
|
|
/* Asynchronously request StarPU to fetch the data in main memory: when
|
|
|
* it is available in main memory, _starpu_mpi_submit_new_mpi_request(req) is called and
|
|
* it is available in main memory, _starpu_mpi_submit_new_mpi_request(req) is called and
|
|
|
- * the request is actually submitted */
|
|
|
|
|
|
|
+ * the request is actually submitted */
|
|
|
starpu_data_acquire_cb(data_handle, mode, _starpu_mpi_submit_new_mpi_request, (void *)req);
|
|
starpu_data_acquire_cb(data_handle, mode, _starpu_mpi_submit_new_mpi_request, (void *)req);
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return req;
|
|
return req;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -116,16 +117,16 @@ static struct _starpu_mpi_req *_starpu_mpi_isend_irecv_common(starpu_data_handle
|
|
|
|
|
|
|
|
static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
|
|
static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
|
STARPU_ASSERT(req->ptr);
|
|
STARPU_ASSERT(req->ptr);
|
|
|
|
|
|
|
|
- _STARPU_MPI_DEBUG("post MPI isend tag %d dst %d ptr %p datatype %p count %d req %p\n", req->mpi_tag, req->srcdst, req->ptr, req->datatype, (int)req->count, &req->request);
|
|
|
|
|
|
|
+ _STARPU_MPI_DEBUG("post MPI isend tag %d dst %d ptr %p datatype %p count %d req %p\n", req->mpi_tag, req->srcdst, req->ptr, req->datatype, (int)req->count, &req->request);
|
|
|
|
|
|
|
|
_starpu_mpi_comm_amounts_inc(req->comm, req->srcdst, req->datatype, req->count);
|
|
_starpu_mpi_comm_amounts_inc(req->comm, req->srcdst, req->datatype, req->count);
|
|
|
|
|
|
|
|
- req->ret = MPI_Isend(req->ptr, req->count, req->datatype, req->srcdst, req->mpi_tag, req->comm, &req->request);
|
|
|
|
|
- STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
+ req->ret = MPI_Isend(req->ptr, req->count, req->datatype, req->srcdst, req->mpi_tag, req->comm, &req->request);
|
|
|
|
|
+ STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
|
TRACE_MPI_ISEND(req->srcdst, req->mpi_tag, 0);
|
|
TRACE_MPI_ISEND(req->srcdst, req->mpi_tag, 0);
|
|
|
|
|
|
|
@@ -134,7 +135,10 @@ static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
|
|
|
req->submitted = 1;
|
|
req->submitted = 1;
|
|
|
_STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
|
|
_STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+
|
|
|
|
|
+ _starpu_mpi_handle_detached_request(req);
|
|
|
|
|
+
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
static void _starpu_mpi_isend_size_callback(void *arg)
|
|
static void _starpu_mpi_isend_size_callback(void *arg)
|
|
@@ -145,8 +149,8 @@ static void _starpu_mpi_isend_size_callback(void *arg)
|
|
|
|
|
|
|
|
static void _starpu_mpi_isend_size_func(struct _starpu_mpi_req *req)
|
|
static void _starpu_mpi_isend_size_func(struct _starpu_mpi_req *req)
|
|
|
{
|
|
{
|
|
|
- req->needs_unpacking = starpu_mpi_handle_to_datatype(req->data_handle, &req->datatype);
|
|
|
|
|
- if (!req->needs_unpacking)
|
|
|
|
|
|
|
+ _starpu_mpi_handle_allocate_datatype(req->data_handle, &req->datatype, &req->user_datatype);
|
|
|
|
|
+ if (req->user_datatype == 0)
|
|
|
{
|
|
{
|
|
|
req->count = 1;
|
|
req->count = 1;
|
|
|
req->ptr = starpu_handle_get_local_ptr(req->data_handle);
|
|
req->ptr = starpu_handle_get_local_ptr(req->data_handle);
|
|
@@ -172,7 +176,7 @@ static struct _starpu_mpi_req *_starpu_mpi_isend_common(starpu_data_handle_t dat
|
|
|
|
|
|
|
|
int starpu_mpi_isend(starpu_data_handle_t data_handle, starpu_mpi_req *public_req, int dest, int mpi_tag, MPI_Comm comm)
|
|
int starpu_mpi_isend(starpu_data_handle_t data_handle, starpu_mpi_req *public_req, int dest, int mpi_tag, MPI_Comm comm)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
STARPU_ASSERT(public_req);
|
|
STARPU_ASSERT(public_req);
|
|
|
|
|
|
|
|
struct _starpu_mpi_req *req;
|
|
struct _starpu_mpi_req *req;
|
|
@@ -181,17 +185,17 @@ int starpu_mpi_isend(starpu_data_handle_t data_handle, starpu_mpi_req *public_re
|
|
|
STARPU_ASSERT(req);
|
|
STARPU_ASSERT(req);
|
|
|
*public_req = req;
|
|
*public_req = req;
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int starpu_mpi_isend_detached(starpu_data_handle_t data_handle,
|
|
int starpu_mpi_isend_detached(starpu_data_handle_t data_handle,
|
|
|
int dest, int mpi_tag, MPI_Comm comm, void (*callback)(void *), void *arg)
|
|
int dest, int mpi_tag, MPI_Comm comm, void (*callback)(void *), void *arg)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
_starpu_mpi_isend_common(data_handle, dest, mpi_tag, comm, 1, callback, arg);
|
|
_starpu_mpi_isend_common(data_handle, dest, mpi_tag, comm, 1, callback, arg);
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -200,13 +204,13 @@ int starpu_mpi_send(starpu_data_handle_t data_handle, int dest, int mpi_tag, MPI
|
|
|
starpu_mpi_req req;
|
|
starpu_mpi_req req;
|
|
|
MPI_Status status;
|
|
MPI_Status status;
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
memset(&status, 0, sizeof(MPI_Status));
|
|
memset(&status, 0, sizeof(MPI_Status));
|
|
|
|
|
|
|
|
starpu_mpi_isend(data_handle, &req, dest, mpi_tag, comm);
|
|
starpu_mpi_isend(data_handle, &req, dest, mpi_tag, comm);
|
|
|
starpu_mpi_wait(&req, &status);
|
|
starpu_mpi_wait(&req, &status);
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -218,39 +222,52 @@ int starpu_mpi_send(starpu_data_handle_t data_handle, int dest, int mpi_tag, MPI
|
|
|
|
|
|
|
|
static void _starpu_mpi_irecv_data_func(struct _starpu_mpi_req *req)
|
|
static void _starpu_mpi_irecv_data_func(struct _starpu_mpi_req *req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
|
STARPU_ASSERT(req->ptr);
|
|
STARPU_ASSERT(req->ptr);
|
|
|
|
|
|
|
|
_STARPU_MPI_DEBUG("post MPI irecv tag %d src %d data %p ptr %p datatype %p count %d req %p \n", req->mpi_tag, req->srcdst, req->data_handle, req->ptr, req->datatype, (int)req->count, &req->request);
|
|
_STARPU_MPI_DEBUG("post MPI irecv tag %d src %d data %p ptr %p datatype %p count %d req %p \n", req->mpi_tag, req->srcdst, req->data_handle, req->ptr, req->datatype, (int)req->count, &req->request);
|
|
|
|
|
|
|
|
- req->ret = MPI_Irecv(req->ptr, req->count, req->datatype, req->srcdst, req->mpi_tag, req->comm, &req->request);
|
|
|
|
|
- STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
+ req->ret = MPI_Irecv(req->ptr, req->count, req->datatype, req->srcdst, req->mpi_tag, req->comm, &req->request);
|
|
|
|
|
+ STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
|
/* somebody is perhaps waiting for the MPI request to be posted */
|
|
/* somebody is perhaps waiting for the MPI request to be posted */
|
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&req->req_mutex);
|
|
|
req->submitted = 1;
|
|
req->submitted = 1;
|
|
|
_STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
|
|
_STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+
|
|
|
|
|
+ _starpu_mpi_handle_detached_request(req);
|
|
|
|
|
+
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+struct _starpu_mpi_irecv_size_callback
|
|
|
|
|
+{
|
|
|
|
|
+ starpu_data_handle_t handle;
|
|
|
|
|
+ struct _starpu_mpi_req *req;
|
|
|
|
|
+};
|
|
|
|
|
+
|
|
|
static void _starpu_mpi_irecv_size_callback(void *arg)
|
|
static void _starpu_mpi_irecv_size_callback(void *arg)
|
|
|
{
|
|
{
|
|
|
- struct _starpu_mpi_req *req = (struct _starpu_mpi_req *) arg;
|
|
|
|
|
|
|
+ struct _starpu_mpi_irecv_size_callback *callback = (struct _starpu_mpi_irecv_size_callback *)arg;
|
|
|
|
|
+
|
|
|
|
|
+ starpu_data_unregister(callback->handle);
|
|
|
|
|
+ callback->req->ptr = malloc(callback->req->count);
|
|
|
#ifdef STARPU_DEVEL
|
|
#ifdef STARPU_DEVEL
|
|
|
-# warning TODO: are we sure that req->count can be used as we have not released count_handle?
|
|
|
|
|
|
|
+#warning TODO: in some cases, callback->req->count is incorrect, we need to fix that
|
|
|
#endif
|
|
#endif
|
|
|
- req->ptr = malloc(req->count);
|
|
|
|
|
- _starpu_mpi_irecv_data_func(req);
|
|
|
|
|
|
|
+ STARPU_ASSERT_MSG(callback->req->ptr, "cannot allocate message of size %ld\n", callback->req->count);
|
|
|
|
|
+ _starpu_mpi_irecv_data_func(callback->req);
|
|
|
|
|
+ free(callback);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
static void _starpu_mpi_irecv_size_func(struct _starpu_mpi_req *req)
|
|
static void _starpu_mpi_irecv_size_func(struct _starpu_mpi_req *req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
|
- req->needs_unpacking = starpu_mpi_handle_to_datatype(req->data_handle, &req->datatype);
|
|
|
|
|
- if (!req->needs_unpacking)
|
|
|
|
|
|
|
+ _starpu_mpi_handle_allocate_datatype(req->data_handle, &req->datatype, &req->user_datatype);
|
|
|
|
|
+ if (req->user_datatype == 0)
|
|
|
{
|
|
{
|
|
|
req->count = 1;
|
|
req->count = 1;
|
|
|
req->ptr = starpu_handle_get_local_ptr(req->data_handle);
|
|
req->ptr = starpu_handle_get_local_ptr(req->data_handle);
|
|
@@ -260,9 +277,11 @@ static void _starpu_mpi_irecv_size_func(struct _starpu_mpi_req *req)
|
|
|
{
|
|
{
|
|
|
starpu_data_handle_t count_handle;
|
|
starpu_data_handle_t count_handle;
|
|
|
|
|
|
|
|
|
|
+ struct _starpu_mpi_irecv_size_callback *callback = malloc(sizeof(struct _starpu_mpi_irecv_size_callback));
|
|
|
starpu_variable_data_register(&count_handle, 0, (uintptr_t)&req->count, sizeof(req->count));
|
|
starpu_variable_data_register(&count_handle, 0, (uintptr_t)&req->count, sizeof(req->count));
|
|
|
- _starpu_mpi_irecv_common(count_handle, req->srcdst, req->mpi_tag, req->comm, 1, _starpu_mpi_irecv_size_callback, req);
|
|
|
|
|
- starpu_data_unregister_submit(count_handle);
|
|
|
|
|
|
|
+ callback->handle = count_handle;
|
|
|
|
|
+ callback->req = req;
|
|
|
|
|
+ _starpu_mpi_irecv_common(count_handle, req->srcdst, req->mpi_tag, req->comm, 1, _starpu_mpi_irecv_size_callback, callback);
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -273,7 +292,7 @@ static struct _starpu_mpi_req *_starpu_mpi_irecv_common(starpu_data_handle_t dat
|
|
|
|
|
|
|
|
int starpu_mpi_irecv(starpu_data_handle_t data_handle, starpu_mpi_req *public_req, int source, int mpi_tag, MPI_Comm comm)
|
|
int starpu_mpi_irecv(starpu_data_handle_t data_handle, starpu_mpi_req *public_req, int source, int mpi_tag, MPI_Comm comm)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
STARPU_ASSERT(public_req);
|
|
STARPU_ASSERT(public_req);
|
|
|
|
|
|
|
|
struct _starpu_mpi_req *req;
|
|
struct _starpu_mpi_req *req;
|
|
@@ -282,15 +301,15 @@ int starpu_mpi_irecv(starpu_data_handle_t data_handle, starpu_mpi_req *public_re
|
|
|
STARPU_ASSERT(req);
|
|
STARPU_ASSERT(req);
|
|
|
*public_req = req;
|
|
*public_req = req;
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int starpu_mpi_irecv_detached(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, void (*callback)(void *), void *arg)
|
|
int starpu_mpi_irecv_detached(starpu_data_handle_t data_handle, int source, int mpi_tag, MPI_Comm comm, void (*callback)(void *), void *arg)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
_starpu_mpi_irecv_common(data_handle, source, mpi_tag, comm, 1, callback, arg);
|
|
_starpu_mpi_irecv_common(data_handle, source, mpi_tag, comm, 1, callback, arg);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -298,11 +317,11 @@ int starpu_mpi_recv(starpu_data_handle_t data_handle, int source, int mpi_tag, M
|
|
|
{
|
|
{
|
|
|
starpu_mpi_req req;
|
|
starpu_mpi_req req;
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
starpu_mpi_irecv(data_handle, &req, source, mpi_tag, comm);
|
|
starpu_mpi_irecv(data_handle, &req, source, mpi_tag, comm);
|
|
|
starpu_mpi_wait(&req, status);
|
|
starpu_mpi_wait(&req, status);
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -314,26 +333,26 @@ int starpu_mpi_recv(starpu_data_handle_t data_handle, int source, int mpi_tag, M
|
|
|
|
|
|
|
|
static void _starpu_mpi_wait_func(struct _starpu_mpi_req *waiting_req)
|
|
static void _starpu_mpi_wait_func(struct _starpu_mpi_req *waiting_req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
/* Which is the mpi request we are waiting for ? */
|
|
/* Which is the mpi request we are waiting for ? */
|
|
|
struct _starpu_mpi_req *req = waiting_req->other_request;
|
|
struct _starpu_mpi_req *req = waiting_req->other_request;
|
|
|
|
|
|
|
|
req->ret = MPI_Wait(&req->request, waiting_req->status);
|
|
req->ret = MPI_Wait(&req->request, waiting_req->status);
|
|
|
- STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
+ STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
|
_starpu_mpi_handle_request_termination(req);
|
|
_starpu_mpi_handle_request_termination(req);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int starpu_mpi_wait(starpu_mpi_req *public_req, MPI_Status *status)
|
|
int starpu_mpi_wait(starpu_mpi_req *public_req, MPI_Status *status)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
int ret;
|
|
int ret;
|
|
|
struct _starpu_mpi_req *waiting_req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
struct _starpu_mpi_req *waiting_req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
|
STARPU_ASSERT(waiting_req);
|
|
STARPU_ASSERT(waiting_req);
|
|
|
struct _starpu_mpi_req *req = *public_req;
|
|
struct _starpu_mpi_req *req = *public_req;
|
|
|
|
|
|
|
|
- _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
|
|
|
|
+ _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
|
|
|
|
|
/* We cannot try to complete a MPI request that was not actually posted
|
|
/* We cannot try to complete a MPI request that was not actually posted
|
|
|
* to MPI yet. */
|
|
* to MPI yet. */
|
|
@@ -364,8 +383,8 @@ int starpu_mpi_wait(starpu_mpi_req *public_req, MPI_Status *status)
|
|
|
*public_req = NULL;
|
|
*public_req = NULL;
|
|
|
free(req);
|
|
free(req);
|
|
|
|
|
|
|
|
- //free(waiting_req);
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ free(waiting_req);
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return ret;
|
|
return ret;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -377,13 +396,13 @@ int starpu_mpi_wait(starpu_mpi_req *public_req, MPI_Status *status)
|
|
|
|
|
|
|
|
static void _starpu_mpi_test_func(struct _starpu_mpi_req *testing_req)
|
|
static void _starpu_mpi_test_func(struct _starpu_mpi_req *testing_req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
/* Which is the mpi request we are testing for ? */
|
|
/* Which is the mpi request we are testing for ? */
|
|
|
struct _starpu_mpi_req *req = testing_req->other_request;
|
|
struct _starpu_mpi_req *req = testing_req->other_request;
|
|
|
|
|
|
|
|
- _STARPU_MPI_DEBUG("Test request %p - mpitag %d - TYPE %s %d\n", &req->request, req->mpi_tag, _starpu_mpi_request_type(req->request_type), req->srcdst);
|
|
|
|
|
|
|
+ _STARPU_MPI_DEBUG("Test request %p - mpitag %d - TYPE %s %d\n", &req->request, req->mpi_tag, _starpu_mpi_request_type(req->request_type), req->srcdst);
|
|
|
req->ret = MPI_Test(&req->request, testing_req->flag, testing_req->status);
|
|
req->ret = MPI_Test(&req->request, testing_req->flag, testing_req->status);
|
|
|
- STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
+ STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
|
if (*testing_req->flag)
|
|
if (*testing_req->flag)
|
|
|
{
|
|
{
|
|
@@ -395,12 +414,12 @@ static void _starpu_mpi_test_func(struct _starpu_mpi_req *testing_req)
|
|
|
testing_req->completed = 1;
|
|
testing_req->completed = 1;
|
|
|
_STARPU_PTHREAD_COND_SIGNAL(&testing_req->req_cond);
|
|
_STARPU_PTHREAD_COND_SIGNAL(&testing_req->req_cond);
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&testing_req->req_mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&testing_req->req_mutex);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int starpu_mpi_test(starpu_mpi_req *public_req, int *flag, MPI_Status *status)
|
|
int starpu_mpi_test(starpu_mpi_req *public_req, int *flag, MPI_Status *status)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
int ret = 0;
|
|
int ret = 0;
|
|
|
|
|
|
|
|
STARPU_ASSERT(public_req);
|
|
STARPU_ASSERT(public_req);
|
|
@@ -416,8 +435,8 @@ int starpu_mpi_test(starpu_mpi_req *public_req, int *flag, MPI_Status *status)
|
|
|
if (submitted)
|
|
if (submitted)
|
|
|
{
|
|
{
|
|
|
struct _starpu_mpi_req *testing_req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
struct _starpu_mpi_req *testing_req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
|
- STARPU_ASSERT(testing_req);
|
|
|
|
|
- // memset(testing_req, 0, sizeof(struct _starpu_mpi_req));
|
|
|
|
|
|
|
+ STARPU_ASSERT(testing_req);
|
|
|
|
|
+ // memset(testing_req, 0, sizeof(struct _starpu_mpi_req));
|
|
|
|
|
|
|
|
/* Initialize the request structure */
|
|
/* Initialize the request structure */
|
|
|
_STARPU_PTHREAD_MUTEX_INIT(&(testing_req->req_mutex), NULL);
|
|
_STARPU_PTHREAD_MUTEX_INIT(&(testing_req->req_mutex), NULL);
|
|
@@ -427,15 +446,15 @@ int starpu_mpi_test(starpu_mpi_req *public_req, int *flag, MPI_Status *status)
|
|
|
testing_req->other_request = req;
|
|
testing_req->other_request = req;
|
|
|
testing_req->func = _starpu_mpi_test_func;
|
|
testing_req->func = _starpu_mpi_test_func;
|
|
|
testing_req->completed = 0;
|
|
testing_req->completed = 0;
|
|
|
- testing_req->request_type = TEST_REQ;
|
|
|
|
|
|
|
+ testing_req->request_type = TEST_REQ;
|
|
|
|
|
|
|
|
- _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
|
|
- _starpu_mpi_submit_new_mpi_request(testing_req);
|
|
|
|
|
|
|
+ _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
|
|
+ _starpu_mpi_submit_new_mpi_request(testing_req);
|
|
|
|
|
|
|
|
/* We wait for the test request to finish */
|
|
/* We wait for the test request to finish */
|
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&(testing_req->req_mutex));
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&(testing_req->req_mutex));
|
|
|
while (!(testing_req->completed))
|
|
while (!(testing_req->completed))
|
|
|
- _STARPU_PTHREAD_COND_WAIT(&(testing_req->req_cond), &(testing_req->req_mutex));
|
|
|
|
|
|
|
+ _STARPU_PTHREAD_COND_WAIT(&(testing_req->req_cond), &(testing_req->req_mutex));
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&(testing_req->req_mutex));
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&(testing_req->req_mutex));
|
|
|
|
|
|
|
|
ret = testing_req->ret;
|
|
ret = testing_req->ret;
|
|
@@ -448,13 +467,15 @@ int starpu_mpi_test(starpu_mpi_req *public_req, int *flag, MPI_Status *status)
|
|
|
*public_req = NULL;
|
|
*public_req = NULL;
|
|
|
free(req);
|
|
free(req);
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ free(testing_req);
|
|
|
}
|
|
}
|
|
|
else
|
|
else
|
|
|
{
|
|
{
|
|
|
*flag = 0;
|
|
*flag = 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return ret;
|
|
return ret;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -466,18 +487,18 @@ int starpu_mpi_test(starpu_mpi_req *public_req, int *flag, MPI_Status *status)
|
|
|
|
|
|
|
|
static void _starpu_mpi_barrier_func(struct _starpu_mpi_req *barrier_req)
|
|
static void _starpu_mpi_barrier_func(struct _starpu_mpi_req *barrier_req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
|
barrier_req->ret = MPI_Barrier(barrier_req->comm);
|
|
barrier_req->ret = MPI_Barrier(barrier_req->comm);
|
|
|
- STARPU_ASSERT(barrier_req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
+ STARPU_ASSERT(barrier_req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
|
_starpu_mpi_handle_request_termination(barrier_req);
|
|
_starpu_mpi_handle_request_termination(barrier_req);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int starpu_mpi_barrier(MPI_Comm comm)
|
|
int starpu_mpi_barrier(MPI_Comm comm)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
int ret;
|
|
int ret;
|
|
|
struct _starpu_mpi_req *barrier_req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
struct _starpu_mpi_req *barrier_req = calloc(1, sizeof(struct _starpu_mpi_req));
|
|
|
STARPU_ASSERT(barrier_req);
|
|
STARPU_ASSERT(barrier_req);
|
|
@@ -513,7 +534,7 @@ int starpu_mpi_barrier(MPI_Comm comm)
|
|
|
barrier_req->request_type = BARRIER_REQ;
|
|
barrier_req->request_type = BARRIER_REQ;
|
|
|
barrier_req->comm = comm;
|
|
barrier_req->comm = comm;
|
|
|
|
|
|
|
|
- _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
|
|
|
|
+ _STARPU_MPI_INC_POSTED_REQUESTS(1);
|
|
|
_starpu_mpi_submit_new_mpi_request(barrier_req);
|
|
_starpu_mpi_submit_new_mpi_request(barrier_req);
|
|
|
|
|
|
|
|
/* We wait for the MPI request to finish */
|
|
/* We wait for the MPI request to finish */
|
|
@@ -524,8 +545,8 @@ int starpu_mpi_barrier(MPI_Comm comm)
|
|
|
|
|
|
|
|
ret = barrier_req->ret;
|
|
ret = barrier_req->ret;
|
|
|
|
|
|
|
|
- //free(waiting_req);
|
|
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ //free(waiting_req);
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
return ret;
|
|
return ret;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -538,31 +559,39 @@ int starpu_mpi_barrier(MPI_Comm comm)
|
|
|
#ifdef STARPU_MPI_VERBOSE
|
|
#ifdef STARPU_MPI_VERBOSE
|
|
|
static char *_starpu_mpi_request_type(enum _starpu_mpi_request_type request_type)
|
|
static char *_starpu_mpi_request_type(enum _starpu_mpi_request_type request_type)
|
|
|
{
|
|
{
|
|
|
- switch (request_type)
|
|
|
|
|
- {
|
|
|
|
|
- case SEND_REQ: return "SEND_REQ";
|
|
|
|
|
- case RECV_REQ: return "RECV_REQ";
|
|
|
|
|
- case WAIT_REQ: return "WAIT_REQ";
|
|
|
|
|
- case TEST_REQ: return "TEST_REQ";
|
|
|
|
|
- case BARRIER_REQ: return "BARRIER_REQ";
|
|
|
|
|
- default: return "unknown request type";
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ switch (request_type)
|
|
|
|
|
+ {
|
|
|
|
|
+ case SEND_REQ: return "SEND_REQ";
|
|
|
|
|
+ case RECV_REQ: return "RECV_REQ";
|
|
|
|
|
+ case WAIT_REQ: return "WAIT_REQ";
|
|
|
|
|
+ case TEST_REQ: return "TEST_REQ";
|
|
|
|
|
+ case BARRIER_REQ: return "BARRIER_REQ";
|
|
|
|
|
+ default: return "unknown request type";
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
#endif
|
|
#endif
|
|
|
|
|
|
|
|
static void _starpu_mpi_handle_request_termination(struct _starpu_mpi_req *req)
|
|
static void _starpu_mpi_handle_request_termination(struct _starpu_mpi_req *req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
|
_STARPU_MPI_DEBUG("complete MPI (%s %d) data %p req %p - tag %d\n", _starpu_mpi_request_type(req->request_type), req->srcdst, req->data_handle, &req->request, req->mpi_tag);
|
|
_STARPU_MPI_DEBUG("complete MPI (%s %d) data %p req %p - tag %d\n", _starpu_mpi_request_type(req->request_type), req->srcdst, req->data_handle, &req->request, req->mpi_tag);
|
|
|
- if (req->request_type != BARRIER_REQ)
|
|
|
|
|
|
|
+ if (req->request_type == RECV_REQ || req->request_type == SEND_REQ)
|
|
|
{
|
|
{
|
|
|
- if (req->needs_unpacking)
|
|
|
|
|
- starpu_handle_unpack_data(req->data_handle, req->ptr, req->count);
|
|
|
|
|
|
|
+ if (req->user_datatype == 1)
|
|
|
|
|
+ {
|
|
|
|
|
+ if (req->request_type == RECV_REQ)
|
|
|
|
|
+ // req->ptr is freed by starpu_handle_unpack_data
|
|
|
|
|
+ starpu_handle_unpack_data(req->data_handle, req->ptr, req->count);
|
|
|
|
|
+ else
|
|
|
|
|
+ free(req->ptr);
|
|
|
|
|
+ }
|
|
|
else
|
|
else
|
|
|
- MPI_Type_free(&req->datatype);
|
|
|
|
|
- starpu_data_release(req->data_handle);
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ {
|
|
|
|
|
+ _starpu_mpi_handle_free_datatype(req->data_handle, &req->datatype);
|
|
|
|
|
+ }
|
|
|
|
|
+ starpu_data_release(req->data_handle);
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
if (req->request_type == RECV_REQ)
|
|
if (req->request_type == RECV_REQ)
|
|
|
{
|
|
{
|
|
@@ -579,23 +608,23 @@ static void _starpu_mpi_handle_request_termination(struct _starpu_mpi_req *req)
|
|
|
req->completed = 1;
|
|
req->completed = 1;
|
|
|
_STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
|
|
_STARPU_PTHREAD_COND_BROADCAST(&req->req_cond);
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&req->req_mutex);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
static void _starpu_mpi_submit_new_mpi_request(void *arg)
|
|
static void _starpu_mpi_submit_new_mpi_request(void *arg)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
struct _starpu_mpi_req *req = arg;
|
|
struct _starpu_mpi_req *req = arg;
|
|
|
|
|
|
|
|
- _STARPU_MPI_INC_POSTED_REQUESTS(-1);
|
|
|
|
|
|
|
+ _STARPU_MPI_INC_POSTED_REQUESTS(-1);
|
|
|
|
|
|
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
|
_starpu_mpi_req_list_push_front(new_requests, req);
|
|
_starpu_mpi_req_list_push_front(new_requests, req);
|
|
|
newer_requests = 1;
|
|
newer_requests = 1;
|
|
|
- _STARPU_MPI_DEBUG("Pushing new request type %s\n", _starpu_mpi_request_type(req->request_type));
|
|
|
|
|
|
|
+ _STARPU_MPI_DEBUG("Pushing new request type %s\n", _starpu_mpi_request_type(req->request_type));
|
|
|
_STARPU_PTHREAD_COND_BROADCAST(&cond_progression);
|
|
_STARPU_PTHREAD_COND_BROADCAST(&cond_progression);
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
#ifdef USE_STARPU_ACTIVITY
|
|
#ifdef USE_STARPU_ACTIVITY
|
|
@@ -617,7 +646,7 @@ static unsigned _starpu_mpi_progression_hook_func(void *arg __attribute__((unuse
|
|
|
|
|
|
|
|
static void _starpu_mpi_test_detached_requests(void)
|
|
static void _starpu_mpi_test_detached_requests(void)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
int flag;
|
|
int flag;
|
|
|
MPI_Status status;
|
|
MPI_Status status;
|
|
|
struct _starpu_mpi_req *req, *next_req;
|
|
struct _starpu_mpi_req *req, *next_req;
|
|
@@ -632,7 +661,7 @@ static void _starpu_mpi_test_detached_requests(void)
|
|
|
|
|
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
|
|
|
|
|
|
|
|
- //_STARPU_MPI_DEBUG("Test detached request %p - mpitag %d - TYPE %s %d\n", &req->request, req->mpi_tag, _starpu_mpi_request_type(req->request_type), req->srcdst);
|
|
|
|
|
|
|
+ //_STARPU_MPI_DEBUG("Test detached request %p - mpitag %d - TYPE %s %d\n", &req->request, req->mpi_tag, _starpu_mpi_request_type(req->request_type), req->srcdst);
|
|
|
req->ret = MPI_Test(&req->request, &flag, &status);
|
|
req->ret = MPI_Test(&req->request, &flag, &status);
|
|
|
STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
STARPU_ASSERT(req->ret == MPI_SUCCESS);
|
|
|
|
|
|
|
@@ -644,29 +673,19 @@ static void _starpu_mpi_test_detached_requests(void)
|
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&detached_requests_mutex);
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&detached_requests_mutex);
|
|
|
|
|
|
|
|
if (flag)
|
|
if (flag)
|
|
|
|
|
+ {
|
|
|
_starpu_mpi_req_list_erase(detached_requests, req);
|
|
_starpu_mpi_req_list_erase(detached_requests, req);
|
|
|
|
|
+ free(req);
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
-#ifdef STARPU_DEVEL
|
|
|
|
|
-#warning TODO fix memleak
|
|
|
|
|
-#endif
|
|
|
|
|
- /* Detached requests are automatically allocated by the lib */
|
|
|
|
|
- //if (req->detached)
|
|
|
|
|
- // free(req);
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&detached_requests_mutex);
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-static void _starpu_mpi_handle_new_request(struct _starpu_mpi_req *req)
|
|
|
|
|
|
|
+static void _starpu_mpi_handle_detached_request(struct _starpu_mpi_req *req)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_LOG_IN();
|
|
|
|
|
- STARPU_ASSERT(req);
|
|
|
|
|
-
|
|
|
|
|
- /* submit the request to MPI */
|
|
|
|
|
- _STARPU_MPI_DEBUG("Handling new request type %s\n", _starpu_mpi_request_type(req->request_type));
|
|
|
|
|
- req->func(req);
|
|
|
|
|
-
|
|
|
|
|
if (req->detached)
|
|
if (req->detached)
|
|
|
{
|
|
{
|
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
@@ -681,59 +700,68 @@ static void _starpu_mpi_handle_new_request(struct _starpu_mpi_req *req)
|
|
|
_STARPU_PTHREAD_COND_SIGNAL(&cond_progression);
|
|
_STARPU_PTHREAD_COND_SIGNAL(&cond_progression);
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
|
}
|
|
}
|
|
|
- _STARPU_MPI_LOG_OUT();
|
|
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+static void _starpu_mpi_handle_new_request(struct _starpu_mpi_req *req)
|
|
|
|
|
+{
|
|
|
|
|
+ _STARPU_MPI_LOG_IN();
|
|
|
|
|
+ STARPU_ASSERT(req);
|
|
|
|
|
+
|
|
|
|
|
+ /* submit the request to MPI */
|
|
|
|
|
+ _STARPU_MPI_DEBUG("Handling new request type %s\n", _starpu_mpi_request_type(req->request_type));
|
|
|
|
|
+ req->func(req);
|
|
|
|
|
+
|
|
|
|
|
+ _STARPU_MPI_LOG_OUT();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
struct _starpu_mpi_argc_argv
|
|
struct _starpu_mpi_argc_argv
|
|
|
{
|
|
{
|
|
|
|
|
+ int initialize_mpi;
|
|
|
int *argc;
|
|
int *argc;
|
|
|
char ***argv;
|
|
char ***argv;
|
|
|
};
|
|
};
|
|
|
|
|
|
|
|
static void _starpu_mpi_print_thread_level_support(int thread_level, char *msg)
|
|
static void _starpu_mpi_print_thread_level_support(int thread_level, char *msg)
|
|
|
{
|
|
{
|
|
|
- switch (thread_level)
|
|
|
|
|
- {
|
|
|
|
|
- case MPI_THREAD_SERIALIZED:
|
|
|
|
|
- {
|
|
|
|
|
- _STARPU_DISP("MPI%s MPI_THREAD_SERIALIZED; Multiple threads may make MPI calls, but only one at a time.\n", msg);
|
|
|
|
|
- break;
|
|
|
|
|
- }
|
|
|
|
|
- case MPI_THREAD_FUNNELED:
|
|
|
|
|
- {
|
|
|
|
|
- _STARPU_DISP("MPI%s MPI_THREAD_FUNNELED; The application can safely make calls to StarPU-MPI functions, but should not call directly MPI communication functions.\n", msg);
|
|
|
|
|
- break;
|
|
|
|
|
- }
|
|
|
|
|
- case MPI_THREAD_SINGLE:
|
|
|
|
|
- {
|
|
|
|
|
- _STARPU_DISP("MPI%s MPI_THREAD_SINGLE; MPI does not have multi-thread support, this might cause problems. The application can make calls to StarPU-MPI functions, but not call directly MPI Communication functions.\n", msg);
|
|
|
|
|
- break;
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ switch (thread_level)
|
|
|
|
|
+ {
|
|
|
|
|
+ case MPI_THREAD_SERIALIZED:
|
|
|
|
|
+ {
|
|
|
|
|
+ _STARPU_DISP("MPI%s MPI_THREAD_SERIALIZED; Multiple threads may make MPI calls, but only one at a time.\n", msg);
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+ case MPI_THREAD_FUNNELED:
|
|
|
|
|
+ {
|
|
|
|
|
+ _STARPU_DISP("MPI%s MPI_THREAD_FUNNELED; The application can safely make calls to StarPU-MPI functions, but should not call directly MPI communication functions.\n", msg);
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+ case MPI_THREAD_SINGLE:
|
|
|
|
|
+ {
|
|
|
|
|
+ _STARPU_DISP("MPI%s MPI_THREAD_SINGLE; MPI does not have multi-thread support, this might cause problems. The application can make calls to StarPU-MPI functions, but not call directly MPI Communication functions.\n", msg);
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
static void *_starpu_mpi_progress_thread_func(void *arg)
|
|
static void *_starpu_mpi_progress_thread_func(void *arg)
|
|
|
{
|
|
{
|
|
|
struct _starpu_mpi_argc_argv *argc_argv = (struct _starpu_mpi_argc_argv *) arg;
|
|
struct _starpu_mpi_argc_argv *argc_argv = (struct _starpu_mpi_argc_argv *) arg;
|
|
|
- int flag;
|
|
|
|
|
|
|
|
|
|
- MPI_Initialized(&flag);
|
|
|
|
|
- _STARPU_DEBUG("MPI_Initialized %d\n", flag);
|
|
|
|
|
- if (flag == 0)
|
|
|
|
|
|
|
+ if (argc_argv->initialize_mpi)
|
|
|
{
|
|
{
|
|
|
int thread_support;
|
|
int thread_support;
|
|
|
- _STARPU_DEBUG("Calling MPI_Init_thread\n");
|
|
|
|
|
|
|
+ _STARPU_DEBUG("Calling MPI_Init_thread\n");
|
|
|
if (MPI_Init_thread(argc_argv->argc, argc_argv->argv, MPI_THREAD_SERIALIZED, &thread_support) != MPI_SUCCESS)
|
|
if (MPI_Init_thread(argc_argv->argc, argc_argv->argv, MPI_THREAD_SERIALIZED, &thread_support) != MPI_SUCCESS)
|
|
|
{
|
|
{
|
|
|
_STARPU_ERROR("MPI_Init_thread failed\n");
|
|
_STARPU_ERROR("MPI_Init_thread failed\n");
|
|
|
- }
|
|
|
|
|
|
|
+ }
|
|
|
_starpu_mpi_print_thread_level_support(thread_support, "_Init_thread level =");
|
|
_starpu_mpi_print_thread_level_support(thread_support, "_Init_thread level =");
|
|
|
- }
|
|
|
|
|
|
|
+ }
|
|
|
else
|
|
else
|
|
|
{
|
|
{
|
|
|
- int provided;
|
|
|
|
|
- MPI_Query_thread(&provided);
|
|
|
|
|
- _starpu_mpi_print_thread_level_support(provided, " has been initialized with");
|
|
|
|
|
|
|
+ int provided;
|
|
|
|
|
+ MPI_Query_thread(&provided);
|
|
|
|
|
+ _starpu_mpi_print_thread_level_support(provided, " has been initialized with");
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/* notify the main thread that the progression thread is ready */
|
|
/* notify the main thread that the progression thread is ready */
|
|
@@ -754,7 +782,7 @@ static void *_starpu_mpi_progress_thread_func(void *arg)
|
|
|
|
|
|
|
|
if (block)
|
|
if (block)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_DEBUG("NO MORE REQUESTS TO HANDLE\n");
|
|
|
|
|
|
|
+ _STARPU_MPI_DEBUG("NO MORE REQUESTS TO HANDLE\n");
|
|
|
if (barrier_running)
|
|
if (barrier_running)
|
|
|
/* Tell mpi_barrier */
|
|
/* Tell mpi_barrier */
|
|
|
_STARPU_PTHREAD_COND_SIGNAL(&cond_finished);
|
|
_STARPU_PTHREAD_COND_SIGNAL(&cond_finished);
|
|
@@ -775,7 +803,7 @@ static void *_starpu_mpi_progress_thread_func(void *arg)
|
|
|
/* handling a request is likely to block for a while
|
|
/* handling a request is likely to block for a while
|
|
|
* (on a sync_data_with_mem call), we want to let the
|
|
* (on a sync_data_with_mem call), we want to let the
|
|
|
* application submit requests in the meantime, so we
|
|
* application submit requests in the meantime, so we
|
|
|
- * release the lock. */
|
|
|
|
|
|
|
+ * release the lock. */
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
|
_starpu_mpi_handle_new_request(req);
|
|
_starpu_mpi_handle_new_request(req);
|
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
@@ -784,13 +812,13 @@ static void *_starpu_mpi_progress_thread_func(void *arg)
|
|
|
|
|
|
|
|
STARPU_ASSERT(_starpu_mpi_req_list_empty(detached_requests));
|
|
STARPU_ASSERT(_starpu_mpi_req_list_empty(detached_requests));
|
|
|
STARPU_ASSERT(_starpu_mpi_req_list_empty(new_requests));
|
|
STARPU_ASSERT(_starpu_mpi_req_list_empty(new_requests));
|
|
|
- STARPU_ASSERT(posted_requests == 0);
|
|
|
|
|
|
|
+ STARPU_ASSERT(posted_requests == 0);
|
|
|
|
|
|
|
|
- if (flag == 0)
|
|
|
|
|
|
|
+ if (argc_argv->initialize_mpi)
|
|
|
{
|
|
{
|
|
|
- _STARPU_MPI_DEBUG("Calling MPI_Finalize()\n");
|
|
|
|
|
- MPI_Finalize();
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ _STARPU_MPI_DEBUG("Calling MPI_Finalize()\n");
|
|
|
|
|
+ MPI_Finalize();
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_UNLOCK(&mutex);
|
|
|
|
|
|
|
@@ -835,12 +863,12 @@ static void _starpu_mpi_add_sync_point_in_fxt(void)
|
|
|
|
|
|
|
|
TRACE_MPI_BARRIER(rank, worldsize, random_number);
|
|
TRACE_MPI_BARRIER(rank, worldsize, random_number);
|
|
|
|
|
|
|
|
- _STARPU_MPI_DEBUG("unique key %x\n", random_number);
|
|
|
|
|
|
|
+ _STARPU_MPI_DEBUG("unique key %x\n", random_number);
|
|
|
#endif
|
|
#endif
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
static
|
|
static
|
|
|
-int _starpu_mpi_initialize(int *argc, char ***argv)
|
|
|
|
|
|
|
+int _starpu_mpi_initialize(int *argc, char ***argv, int initialize_mpi)
|
|
|
{
|
|
{
|
|
|
_STARPU_PTHREAD_MUTEX_INIT(&mutex, NULL);
|
|
_STARPU_PTHREAD_MUTEX_INIT(&mutex, NULL);
|
|
|
_STARPU_PTHREAD_COND_INIT(&cond_progression, NULL);
|
|
_STARPU_PTHREAD_COND_INIT(&cond_progression, NULL);
|
|
@@ -850,12 +878,13 @@ int _starpu_mpi_initialize(int *argc, char ***argv)
|
|
|
_STARPU_PTHREAD_MUTEX_INIT(&detached_requests_mutex, NULL);
|
|
_STARPU_PTHREAD_MUTEX_INIT(&detached_requests_mutex, NULL);
|
|
|
detached_requests = _starpu_mpi_req_list_new();
|
|
detached_requests = _starpu_mpi_req_list_new();
|
|
|
|
|
|
|
|
- _STARPU_PTHREAD_MUTEX_INIT(&mutex_posted_requests, NULL);
|
|
|
|
|
|
|
+ _STARPU_PTHREAD_MUTEX_INIT(&mutex_posted_requests, NULL);
|
|
|
|
|
|
|
|
struct _starpu_mpi_argc_argv *argc_argv = malloc(sizeof(struct _starpu_mpi_argc_argv));
|
|
struct _starpu_mpi_argc_argv *argc_argv = malloc(sizeof(struct _starpu_mpi_argc_argv));
|
|
|
|
|
+ argc_argv->initialize_mpi = initialize_mpi;
|
|
|
argc_argv->argc = argc;
|
|
argc_argv->argc = argc;
|
|
|
argc_argv->argv = argv;
|
|
argc_argv->argv = argv;
|
|
|
- _STARPU_PTHREAD_CREATE(&progress_thread, NULL, _starpu_mpi_progress_thread_func, argc_argv);
|
|
|
|
|
|
|
+ _STARPU_PTHREAD_CREATE("MPI progress", &progress_thread, NULL, _starpu_mpi_progress_thread_func, argc_argv);
|
|
|
|
|
|
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
_STARPU_PTHREAD_MUTEX_LOCK(&mutex);
|
|
|
while (!running)
|
|
while (!running)
|
|
@@ -875,25 +904,25 @@ int _starpu_mpi_initialize(int *argc, char ***argv)
|
|
|
|
|
|
|
|
_starpu_mpi_add_sync_point_in_fxt();
|
|
_starpu_mpi_add_sync_point_in_fxt();
|
|
|
_starpu_mpi_comm_amounts_init(MPI_COMM_WORLD);
|
|
_starpu_mpi_comm_amounts_init(MPI_COMM_WORLD);
|
|
|
- _starpu_mpi_tables_init(MPI_COMM_WORLD);
|
|
|
|
|
|
|
+ _starpu_mpi_cache_init(MPI_COMM_WORLD);
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-int starpu_mpi_init(int *argc, char ***argv)
|
|
|
|
|
|
|
+int starpu_mpi_init(int *argc, char ***argv, int initialize_mpi)
|
|
|
{
|
|
{
|
|
|
- return _starpu_mpi_initialize(argc, argv);
|
|
|
|
|
|
|
+ return _starpu_mpi_initialize(argc, argv, initialize_mpi);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int starpu_mpi_initialize(void)
|
|
int starpu_mpi_initialize(void)
|
|
|
{
|
|
{
|
|
|
- return _starpu_mpi_initialize(NULL, NULL);
|
|
|
|
|
|
|
+ return _starpu_mpi_initialize(NULL, NULL, 0);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int starpu_mpi_initialize_extended(int *rank, int *world_size)
|
|
int starpu_mpi_initialize_extended(int *rank, int *world_size)
|
|
|
{
|
|
{
|
|
|
int ret;
|
|
int ret;
|
|
|
|
|
|
|
|
- ret = _starpu_mpi_initialize(NULL, NULL);
|
|
|
|
|
|
|
+ ret = _starpu_mpi_initialize(NULL, NULL, 1);
|
|
|
if (ret == 0)
|
|
if (ret == 0)
|
|
|
{
|
|
{
|
|
|
_STARPU_DEBUG("Calling MPI_Comm_rank\n");
|
|
_STARPU_DEBUG("Calling MPI_Comm_rank\n");
|
|
@@ -908,7 +937,7 @@ int starpu_mpi_shutdown(void)
|
|
|
void *value;
|
|
void *value;
|
|
|
int rank, world_size;
|
|
int rank, world_size;
|
|
|
|
|
|
|
|
- /* We need to get the rank before calling MPI_Finalize to pass to _starpu_mpi_comm_amounts_display() */
|
|
|
|
|
|
|
+ /* We need to get the rank before calling MPI_Finalize to pass to _starpu_mpi_comm_amounts_display() */
|
|
|
MPI_Comm_rank(MPI_COMM_WORLD, &rank);
|
|
MPI_Comm_rank(MPI_COMM_WORLD, &rank);
|
|
|
MPI_Comm_size(MPI_COMM_WORLD, &world_size);
|
|
MPI_Comm_size(MPI_COMM_WORLD, &world_size);
|
|
|
|
|
|
|
@@ -930,7 +959,7 @@ int starpu_mpi_shutdown(void)
|
|
|
|
|
|
|
|
_starpu_mpi_comm_amounts_display(rank);
|
|
_starpu_mpi_comm_amounts_display(rank);
|
|
|
_starpu_mpi_comm_amounts_free();
|
|
_starpu_mpi_comm_amounts_free();
|
|
|
- _starpu_mpi_tables_free(world_size);
|
|
|
|
|
|
|
+ _starpu_mpi_cache_free(world_size);
|
|
|
|
|
|
|
|
return 0;
|
|
return 0;
|
|
|
}
|
|
}
|