Skip to content

Instantly share code, notes, and snippets.

@goodell

goodell/gist:1839199

Created February 15, 2012 21:45
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save goodell/1839199 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save goodell/1839199 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
gistfile1.diff
commit 292bab7c2aa308bd00041374551090f7c0d30746
Author: Dave Goodell <[email protected]>
Date: Tue Feb 14 14:35:00 2012 -0600
implement MPIX_Mprobe and friends
This includes MPIX_Mprobe, MPIX_Improbe, MPIX_Mrecv, and MPIX_Imrecv.
!!!NOTE!!! This change adds three more functions to the ADI.
The current implementation of MPIX_Mprobe is much less efficient than it
could be. Right now it polls the unexpected queue (UQ), much like
MPI_Probe does. A better implementation would enqueue the "message"
object on the PQ so that the matching is handled by the progress engine
in the various packet handlers. Unfortunately, such an implementation
is surprisingly complicated at this point.
Includes only a simple test for now (test/mpi/pt2pt/mprobe), but it at
least shows that our Mprobe functionality friends basically all works.
Reviewed by buntinas@.
---
src/include/mpi.h.in | 14 +++
src/include/mpiimpl.h | 100 ++++++++++++++++++
src/mpi/errhan/errnames.txt | 13 +++
src/mpi/pt2pt/Makefile.mk | 4 +
src/mpi/pt2pt/improbe.c | 145 ++++++++++++++++++++++++++
src/mpi/pt2pt/imrecv.c | 159 +++++++++++++++++++++++++++++
src/mpi/pt2pt/mprobe.c | 136 +++++++++++++++++++++++++
src/mpi/pt2pt/mrecv.c | 181 +++++++++++++++++++++++++++++++++
src/mpi/pt2pt/waitall.c | 2 +
src/mpid/ch3/include/mpidimpl.h | 2 +
src/mpid/ch3/src/Makefile.mk | 3 +
src/mpid/ch3/src/ch3u_recvq.c | 97 ++++++++++++++++++
src/mpid/ch3/src/mpid_improbe.c | 57 +++++++++++
src/mpid/ch3/src/mpid_imrecv.c | 119 ++++++++++++++++++++++
src/mpid/ch3/src/mpid_mprobe.c | 69 +++++++++++++
test/mpi/pt2pt/Makefile.am | 3 +-
test/mpi/pt2pt/mprobe.c | 213 +++++++++++++++++++++++++++++++++++++++
test/mpi/pt2pt/testlist | 1 +
18 files changed, 1317 insertions(+), 1 deletions(-)
diff --git a/src/include/mpi.h.in b/src/include/mpi.h.in
index 990f550..f9aea64 100644
--- a/src/include/mpi.h.in
+++ b/src/include/mpi.h.in
@@ -192,6 +192,7 @@ typedef int MPI_Op;
#define MPI_DATATYPE_NULL ((MPI_Datatype)0x0c000000)
#define MPI_REQUEST_NULL ((MPI_Request)0x2c000000)
#define MPI_ERRHANDLER_NULL ((MPI_Errhandler)0x14000000)
+#define MPIX_MESSAGE_NULL ((MPIX_Message)MPI_REQUEST_NULL)
/* These are only guesses; make sure you change them in mpif.h as well */
#define MPI_MAX_PROCESSOR_NAME @MPI_MAX_PROCESSOR_NAME@
@@ -272,6 +273,9 @@ typedef int MPI_Errhandler;
/* MPI request opjects */
typedef int MPI_Request;
+/* MPI message objects for Mprobe and related functions */
+typedef int MPIX_Message;
+
/* User combination function */
typedef void (MPI_User_function) ( void *, void *, int *, MPI_Datatype * );
@@ -975,6 +979,12 @@ int MPI_Dist_graph_create(MPI_Comm comm_old, int n, int [], int [], int [], int
int MPI_Dist_graph_neighbors_count(MPI_Comm comm, int *indegree, int *outdegree, int *weighted);
int MPI_Dist_graph_neighbors(MPI_Comm comm, int maxindegree, int [], int [], int maxoutdegree, int [], int []);
+/* MPI-3 matched probe functionality, currently MPIX_ extensions */
+int MPIX_Improbe(int source, int tag, MPI_Comm comm, int *flag, MPIX_Message *message, MPI_Status *status);
+int MPIX_Imrecv(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Request *request);
+int MPIX_Mprobe(int source, int tag, MPI_Comm comm, MPIX_Message *message, MPI_Status *status);
+int MPIX_Mrecv(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Status *status);
+
/* MPI-3 nonblocking collectives, currently MPIX_ extensions */
int MPIX_Ibarrier(MPI_Comm comm, MPI_Request *request);
int MPIX_Ibcast(void *buffer, int count, MPI_Datatype datatype, int root, MPI_Comm comm, MPI_Request *request);
@@ -1323,6 +1333,10 @@ int PMPI_Dist_graph_create_adjacent(MPI_Comm comm_old, int indegree, int [], int
int PMPI_Dist_graph_create(MPI_Comm comm_old, int n, int [], int [], int [], int [], MPI_Info info, int reorder, MPI_Comm *comm_dist_graph);
int PMPI_Dist_graph_neighbors_count(MPI_Comm comm, int *indegree, int *outdegree, int *weighted);
int PMPI_Dist_graph_neighbors(MPI_Comm comm, int maxindegree, int [], int [], int maxoutdegree, int [], int []);
+int PMPIX_Improbe(int source, int tag, MPI_Comm comm, int *flag, MPIX_Message *message, MPI_Status *status);
+int PMPIX_Imrecv(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Request *request);
+int PMPIX_Mprobe(int source, int tag, MPI_Comm comm, MPIX_Message *message, MPI_Status *status);
+int PMPIX_Mrecv(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Status *status);
int PMPIX_Ibarrier(MPI_Comm comm, MPI_Request *request);
int PMPIX_Ibcast(void *buffer, int count, MPI_Datatype datatype, int root, MPI_Comm comm, MPI_Request *request);
int PMPIX_Igather(void *sendbuf, int sendcount, MPI_Datatype sendtype, void *recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm, MPI_Request *request);
diff --git a/src/include/mpiimpl.h b/src/include/mpiimpl.h
index 5284c40..e78a44b 100644
--- a/src/include/mpiimpl.h
+++ b/src/include/mpiimpl.h
@@ -1329,6 +1329,12 @@ void MPIR_Free_contextid( MPIR_Context_id_t );
/* Requests */
/* This currently defines a single structure type for all requests.
Eventually, we may want a union type, as used in MPICH-1 */
+/* NOTE-R1: MPID_REQUEST_MPROBE signifies that this is a request created by
+ * MPI_Mprobe or MPI_Improbe. Since we use MPI_Request objects as our
+ * MPI_Message objects, we use this separate kind in order to provide stronger
+ * error checking. Once a message (backed by a request) is promoted to a real
+ * request by calling MPI_Mrecv/MPI_Imrecv, we actually modify the kind to be
+ * MPID_REQUEST_RECV in order to keep completion logic as simple as possible. */
/*E
MPID_Request_kind - Kinds of MPI Requests
@@ -1344,6 +1350,7 @@ typedef enum MPID_Request_kind_t {
MPID_PREQUEST_RECV,
MPID_UREQUEST,
MPID_COLL_REQUEST,
+ MPID_REQUEST_MPROBE, /* see NOTE-R1 */
MPID_LAST_REQUEST_KIND
#ifdef MPID_DEV_REQUEST_KIND_DECL
, MPID_DEV_REQUEST_KIND_DECL
@@ -2905,6 +2912,93 @@ int MPID_Probe(int, int, MPID_Comm *, int, MPI_Status *);
int MPID_Iprobe(int, int, MPID_Comm *, int, int *, MPI_Status *);
/*@
+ MPID_Mprobe - Block until a matching request is found and return information
+ about it, including a message handle for later reception.
+
+ Input Parameters:
++ source - rank to match (or 'MPI_ANY_SOURCE')
+. tag - Tag to match (or 'MPI_ANY_TAG')
+. comm - communicator to match.
+- context_offset - context id offset of communicator to match
+
+ Output Parameter:
++ message - 'MPID_Request' (logically a message) set as defined by 'MPI_Mprobe'
+- status - 'MPI_Status' set as defined by 'MPI_Mprobe'
+
+ Return Value:
+ Error code.
+
+ Providing the 'context_offset' is necessary at this level to support the
+ way in which the MPICH implementation uses context ids in the implementation
+ of other operations. The communicator is present to allow the device
+ to use message-queues attached to particular communicators or connections
+ between processes.
+
+ Module:
+ Request
+
+ @*/
+int MPID_Mprobe(int source, int tag, MPID_Comm *comm, int context_offset,
+ MPID_Request **message, MPI_Status *status);
+
+/*@
+ MPID_Improbe - Look for a matching request in the receive queue and return
+ information about it, including a message handle for later reception.
+
+ Input Parameters:
++ source - rank to match (or 'MPI_ANY_SOURCE')
+. tag - Tag to match (or 'MPI_ANY_TAG')
+. comm - communicator to match.
+- context_offset - context id offset of communicator to match
+
+ Output Parameter:
++ flag - 'flag' set as defined by 'MPI_Improbe'
+. message - 'MPID_Request' (logically a message) set as defined by 'MPI_Improbe'
+- status - 'MPI_Status' set as defined by 'MPI_Improbe'
+
+ Return Value:
+ Error code.
+
+ Providing the 'context_offset' is necessary at this level to support the
+ way in which the MPICH implementation uses context ids in the implementation
+ of other operations. The communicator is present to allow the device
+ to use message-queues attached to particular communicators or connections
+ between processes.
+
+ Module:
+ Request
+
+ @*/
+int MPID_Improbe(int source, int tag, MPID_Comm *comm, int context_offset,
+ int *flag, MPID_Request **message, MPI_Status *status);
+
+/*@
+ MPID_Imrecv - Begin receiving the message indicated by the given message
+ handle and return a request object for later completion.
+
+ Input Parameters:
++ count - number of elements to receive
+. datatype - datatype of each recv buffer element
+- message - 'MPID_Request' (logically a message) set as defined by 'MPI_Mprobe'
+
+ Output Parameter:
++ buf - receive buffer
+- request - request object for completing the recv
+
+ Return Value:
+ Error code.
+
+ Module:
+ Request
+
+ NOTE: under most implementations the request object returned will
+ probably be some modified version of the "message" object passed in.
+
+ @*/
+int MPID_Imrecv(void *buf, int count, MPI_Datatype datatype,
+ MPID_Request *message, MPID_Request **rreqp);
+
+/*@
MPID_Cancel_send - Cancel the indicated send request
Input Parameter:
@@ -3736,6 +3830,12 @@ int MPIR_Comm_is_node_consecutive( MPID_Comm *);
void MPIR_Free_err_dyncodes( void );
+/* mprobe-related impl functions */
+int MPIR_Improbe_impl(int source, int tag, MPID_Comm *comm_ptr, int *flag, MPIX_Message *message, MPI_Status *status);
+int MPIR_Imrecv_impl(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Request *request);
+int MPIR_Mprobe_impl(int source, int tag, MPID_Comm *comm_ptr, MPIX_Message *message, MPI_Status *status);
+int MPIR_Mrecv_impl(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Status *status);
+
/* begin impl functions for NBC */
int MPIR_Ibarrier_impl(MPID_Comm *comm_ptr, MPI_Request *request);
int MPIR_Ibcast_impl(void *buffer, int count, MPI_Datatype datatype, int root, MPID_Comm *comm_ptr, MPI_Request *request);
diff --git a/src/mpi/errhan/errnames.txt b/src/mpi/errhan/errnames.txt
index 03c4193..7d05c5d 100644
--- a/src/mpi/errhan/errnames.txt
+++ b/src/mpi/errhan/errnames.txt
@@ -269,6 +269,9 @@ be in the range 0 to %d
**request_invalid_kind %d:The supplied request was invalid (kind=%d)
**request_invalid_kind %d %d:The supplied request in array element %d was \
invalid (kind=%d)
+**reqnotmsg: Invalid MPIX_Message (MPI_Request passed)
+**msgnotreq: Invalid MPI_Request (MPIX_Message passed)
+**msgnotreq %d: Invalid MPI_Request at array index %d (MPIX_Message passed)
**requestnull:Null MPI_Request
**requestnotpersist:Request is not persistent in MPI_Start or MPI_Startall.
**requestpersistactive:Persistent request passed to MPI_Start or MPI_Startall is already active.
@@ -1448,6 +1451,16 @@ is too big (> MPIU_SHMW_GHND_SZ)
**mpir_wingetattr:MPIR_WinGetAttr failed
**mpir_wingetattr %W %d %p %p:MPIR_WinGetAttr(%W, win_keyval=%d, attribute_val=%p, flag=%p) failed
+## MPI-3 Mprobe routines
+**mpix_improbe: MPIX_Improbe failed
+**mpix_improbe %d %d %C %p %p %p: MPIX_Improbe(source=%d, tag=%d, comm=%C, flag=%p, message=%p, status=%p)
+**mpix_mprobe: MPIX_Mprobe failed
+**mpix_mprobe %d %d %C %p %p: MPIX_Mprobe(source=%d, tag=%d, comm=%C, message=%p, status=%p)
+**mpix_mrecv: MPIX_Mrecv failed
+**mpix_mrecv %p %d %D %p %p: MPIX_Mrecv(buf=%p, count=%d, datatype=%D, message=%p, status=%p)
+**mpix_imrecv: MPIX_Imrecv failed
+**mpix_imrecv %p %d %D %p %p: MPIX_Imrecv(buf=%p, count=%d, datatype=%D, message=%p, request=%p)
+
## MPI-3 nonblocking collectives
**mpix_ibarrier: MPIX_Ibarrier failed
**mpix_ibarrier %C %p: MPIX_Ibarrier(comm=%C, request=%p)
diff --git a/src/mpi/pt2pt/Makefile.mk b/src/mpi/pt2pt/Makefile.mk
index b3cebf9..da058e5 100644
--- a/src/mpi/pt2pt/Makefile.mk
+++ b/src/mpi/pt2pt/Makefile.mk
@@ -14,11 +14,15 @@ mpi_sources += \
src/mpi/pt2pt/greq_start.c \
src/mpi/pt2pt/greq_complete.c \
src/mpi/pt2pt/ibsend.c \
+ src/mpi/pt2pt/improbe.c \
+ src/mpi/pt2pt/imrecv.c \
src/mpi/pt2pt/iprobe.c \
src/mpi/pt2pt/irecv.c \
src/mpi/pt2pt/irsend.c \
src/mpi/pt2pt/isend.c \
src/mpi/pt2pt/issend.c \
+ src/mpi/pt2pt/mprobe.c \
+ src/mpi/pt2pt/mrecv.c \
src/mpi/pt2pt/probe.c \
src/mpi/pt2pt/recv.c \
src/mpi/pt2pt/recv_init.c \
diff --git a/src/mpi/pt2pt/improbe.c b/src/mpi/pt2pt/improbe.c
new file mode 100644
index 0000000..1950778
--- /dev/null
+++ b/src/mpi/pt2pt/improbe.c
@@ -0,0 +1,145 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include "mpiimpl.h"
+
+/* -- Begin Profiling Symbol Block for routine MPIX_Improbe */
+#if defined(HAVE_PRAGMA_WEAK)
+#pragma weak MPIX_Improbe = PMPIX_Improbe
+#elif defined(HAVE_PRAGMA_HP_SEC_DEF)
+#pragma _HP_SECONDARY_DEF PMPIX_Improbe MPIX_Improbe
+#elif defined(HAVE_PRAGMA_CRI_DUP)
+#pragma _CRI duplicate MPIX_Improbe as PMPIX_Improbe
+#endif
+/* -- End Profiling Symbol Block */
+
+/* Define MPICH_MPI_FROM_PMPI if weak symbols are not supported to build
+ the MPI routines */
+#ifndef MPICH_MPI_FROM_PMPI
+#undef MPIX_Improbe
+#define MPIX_Improbe PMPIX_Improbe
+
+/* any non-MPI functions go here, especially non-static ones */
+
+#undef FUNCNAME
+#define FUNCNAME MPIR_Improbe_impl
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPIR_Improbe_impl(int source, int tag, MPID_Comm *comm_ptr, int *flag, MPIX_Message *message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_Request *msgp = NULL;
+
+ *message = MPIX_MESSAGE_NULL;
+ mpi_errno = MPID_Improbe(source, tag, comm_ptr, MPID_CONTEXT_INTRA_PT2PT, flag, &msgp, status);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ if (*flag) {
+ *message = msgp->handle;
+ MPIU_Assert(msgp->kind == MPID_REQUEST_MPROBE);
+ }
+ else {
+ /* the standard says that if flag is false then message and status are
+ * undefined */
+ *message = MPIX_MESSAGE_NULL;
+ }
+
+fn_exit:
+ return mpi_errno;
+fn_fail:
+ goto fn_exit;
+}
+
+#endif /* MPICH_MPI_FROM_PMPI */
+
+#undef FUNCNAME
+#define FUNCNAME MPIX_Improbe
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+/*@
+MPIX_Improbe - XXX description here
+
+Input Parameters:
++ source - rank of source or MPI_ANY_SOURCE (integer)
+. tag - message tag or MPI_ANY_TAG (integer)
+- comm - communicator (handle)
+
+Output Parameters:
++ flag - flag (logical)
+. message - returned message (handle)
+- status - status object (status)
+
+.N ThreadSafe
+
+.N Fortran
+
+.N Errors
+@*/
+int MPIX_Improbe(int source, int tag, MPI_Comm comm, int *flag, MPIX_Message *message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_Comm *comm_ptr = NULL;
+ MPID_MPI_STATE_DECL(MPID_STATE_MPIX_IMPROBE);
+
+ MPIU_THREAD_CS_ENTER(ALLFUNC,);
+ MPID_MPI_FUNC_ENTER(MPID_STATE_MPIX_IMPROBE);
+
+ /* Validate parameters, especially handles needing to be converted */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ MPIR_ERRTEST_COMM(comm, mpi_errno);
+
+ /* TODO more checks may be appropriate */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* Convert MPI object handles to object pointers */
+ MPID_Comm_get_ptr(comm, comm_ptr);
+
+ /* Validate parameters and objects (post conversion) */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ MPID_Comm_valid_ptr(comm_ptr, mpi_errno);
+ MPIR_ERRTEST_ARGNULL(flag, "flag", mpi_errno);
+ /* TODO more checks may be appropriate (counts, in_place, buffer aliasing, etc) */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* ... body of routine ... */
+
+ mpi_errno = MPIR_Improbe_impl(source, tag, comm_ptr, flag, message, status);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ /* ... end of body of routine ... */
+
+fn_exit:
+ MPID_MPI_FUNC_EXIT(MPID_STATE_MPIX_IMPROBE);
+ MPIU_THREAD_CS_EXIT(ALLFUNC,);
+ return mpi_errno;
+
+fn_fail:
+ /* --BEGIN ERROR HANDLING-- */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ mpi_errno = MPIR_Err_create_code(
+ mpi_errno, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPI_ERR_OTHER,
+ "**mpix_improbe", "**mpix_improbe %d %d %C %p %p %p", source, tag, comm, flag, message, status);
+ }
+# endif
+ mpi_errno = MPIR_Err_return_comm(NULL, FCNAME, mpi_errno);
+ goto fn_exit;
+ /* --END ERROR HANDLING-- */
+}
diff --git a/src/mpi/pt2pt/imrecv.c b/src/mpi/pt2pt/imrecv.c
new file mode 100644
index 0000000..4515d99
--- /dev/null
+++ b/src/mpi/pt2pt/imrecv.c
@@ -0,0 +1,159 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include "mpiimpl.h"
+
+/* -- Begin Profiling Symbol Block for routine MPIX_Imrecv */
+#if defined(HAVE_PRAGMA_WEAK)
+#pragma weak MPIX_Imrecv = PMPIX_Imrecv
+#elif defined(HAVE_PRAGMA_HP_SEC_DEF)
+#pragma _HP_SECONDARY_DEF PMPIX_Imrecv MPIX_Imrecv
+#elif defined(HAVE_PRAGMA_CRI_DUP)
+#pragma _CRI duplicate MPIX_Imrecv as PMPIX_Imrecv
+#endif
+/* -- End Profiling Symbol Block */
+
+/* Define MPICH_MPI_FROM_PMPI if weak symbols are not supported to build
+ the MPI routines */
+#ifndef MPICH_MPI_FROM_PMPI
+#undef MPIX_Imrecv
+#define MPIX_Imrecv PMPIX_Imrecv
+
+/* any non-MPI functions go here, especially non-static ones */
+
+#undef FUNCNAME
+#define FUNCNAME MPIR_Imrecv_impl
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPIR_Imrecv_impl(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Request *request)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_Request *msgp = NULL;
+ MPID_Request *rreq = NULL;
+
+ /* get the underlying MPID_Request object and validate it */
+ MPID_Request_get_ptr(*message, msgp);
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS;
+ {
+ MPID_Request_valid_ptr(msgp, mpi_errno);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ MPIU_ERR_CHKANDJUMP((msgp->kind != MPID_REQUEST_MPROBE),
+ mpi_errno, MPI_ERR_ARG, "**reqnotmsg")
+ }
+ MPID_END_ERROR_CHECKS;
+ }
+# endif
+
+ mpi_errno = MPID_Imrecv(buf, count, datatype, msgp, &rreq);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ MPIU_Assert(rreq != NULL);
+ *request = rreq->handle;
+ *message = MPIX_MESSAGE_NULL;
+
+fn_exit:
+ return mpi_errno;
+fn_fail:
+ goto fn_exit;
+}
+
+#endif /* MPICH_MPI_FROM_PMPI */
+
+#undef FUNCNAME
+#define FUNCNAME MPIX_Imrecv
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+/*@
+MPIX_Imrecv - XXX description here
+
+Input/Output Parameters:
+. message - message (handle)
+
+Input Parameters:
++ count - number of elements in the receive buffer (non-negative integer)
+- datatype - datatype of each receive buffer element (handle)
+
+Output Parameters:
++ buf - initial address of the receive buffer (choice)
+- request - communication request (handle)
+
+.N ThreadSafe
+
+.N Fortran
+
+.N Errors
+@*/
+int MPIX_Imrecv(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Request *request)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_MPI_STATE_DECL(MPID_STATE_MPIX_IMRECV);
+
+ MPIU_THREAD_CS_ENTER(ALLFUNC,);
+ MPID_MPI_FUNC_ENTER(MPID_STATE_MPIX_IMRECV);
+
+ /* Validate parameters, especially handles needing to be converted */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ MPIR_ERRTEST_DATATYPE(datatype, "datatype", mpi_errno);
+
+ /* TODO more checks may be appropriate */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* Convert MPI object handles to object pointers */
+
+ /* Validate parameters and objects (post conversion) */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ if (HANDLE_GET_KIND(datatype) != HANDLE_KIND_BUILTIN) {
+ MPID_Datatype *datatype_ptr = NULL;
+ MPID_Datatype_get_ptr(datatype, datatype_ptr);
+ MPID_Datatype_valid_ptr(datatype_ptr, mpi_errno);
+ MPID_Datatype_committed_ptr(datatype_ptr, mpi_errno);
+ }
+
+ MPIR_ERRTEST_ARGNULL(request, "request", mpi_errno);
+ /* TODO more checks may be appropriate (counts, in_place, buffer aliasing, etc) */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* ... body of routine ... */
+
+ mpi_errno = MPIR_Imrecv_impl(buf, count, datatype, message, request);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ /* ... end of body of routine ... */
+
+fn_exit:
+ MPID_MPI_FUNC_EXIT(MPID_STATE_MPIX_IMRECV);
+ MPIU_THREAD_CS_EXIT(ALLFUNC,);
+ return mpi_errno;
+
+fn_fail:
+ /* --BEGIN ERROR HANDLING-- */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ mpi_errno = MPIR_Err_create_code(
+ mpi_errno, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPI_ERR_OTHER,
+ "**mpix_imrecv", "**mpix_imrecv %p %d %D %p %p", buf, count, datatype, message, request);
+ }
+# endif
+ mpi_errno = MPIR_Err_return_comm(NULL, FCNAME, mpi_errno);
+ goto fn_exit;
+ /* --END ERROR HANDLING-- */
+}
diff --git a/src/mpi/pt2pt/mprobe.c b/src/mpi/pt2pt/mprobe.c
new file mode 100644
index 0000000..8ab9bed
--- /dev/null
+++ b/src/mpi/pt2pt/mprobe.c
@@ -0,0 +1,136 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include "mpiimpl.h"
+
+/* -- Begin Profiling Symbol Block for routine MPIX_Mprobe */
+#if defined(HAVE_PRAGMA_WEAK)
+#pragma weak MPIX_Mprobe = PMPIX_Mprobe
+#elif defined(HAVE_PRAGMA_HP_SEC_DEF)
+#pragma _HP_SECONDARY_DEF PMPIX_Mprobe MPIX_Mprobe
+#elif defined(HAVE_PRAGMA_CRI_DUP)
+#pragma _CRI duplicate MPIX_Mprobe as PMPIX_Mprobe
+#endif
+/* -- End Profiling Symbol Block */
+
+/* Define MPICH_MPI_FROM_PMPI if weak symbols are not supported to build
+ the MPI routines */
+#ifndef MPICH_MPI_FROM_PMPI
+#undef MPIX_Mprobe
+#define MPIX_Mprobe PMPIX_Mprobe
+
+/* any non-MPI functions go here, especially non-static ones */
+
+#undef FUNCNAME
+#define FUNCNAME MPIR_Mprobe_impl
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPIR_Mprobe_impl(int source, int tag, MPID_Comm *comm_ptr, MPIX_Message *message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_Request *msgp = NULL;
+
+ *message = MPIX_MESSAGE_NULL;
+ mpi_errno = MPID_Mprobe(source, tag, comm_ptr, MPID_CONTEXT_INTRA_PT2PT, &msgp, status);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ *message = msgp->handle;
+ MPIU_Assert(msgp->kind == MPID_REQUEST_MPROBE);
+
+fn_exit:
+ return mpi_errno;
+fn_fail:
+ goto fn_exit;
+}
+
+#endif /* MPICH_MPI_FROM_PMPI */
+
+#undef FUNCNAME
+#define FUNCNAME MPIX_Mprobe
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+/*@
+MPIX_Mprobe - XXX description here
+
+Input Parameters:
++ source - rank of source or MPI_ANY_SOURCE (integer)
+. tag - message tag or MPI_ANY_TAG (integer)
+- comm - communicator (handle)
+
+Output Parameters:
++ message - returned message (handle)
+- status - status object (status)
+
+.N ThreadSafe
+
+.N Fortran
+
+.N Errors
+@*/
+int MPIX_Mprobe(int source, int tag, MPI_Comm comm, MPIX_Message *message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_Comm *comm_ptr = NULL;
+ MPID_MPI_STATE_DECL(MPID_STATE_MPIX_MPROBE);
+
+ MPIU_THREAD_CS_ENTER(ALLFUNC,);
+ MPID_MPI_FUNC_ENTER(MPID_STATE_MPIX_MPROBE);
+
+ /* Validate parameters, especially handles needing to be converted */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ MPIR_ERRTEST_COMM(comm, mpi_errno);
+
+ /* TODO more checks may be appropriate */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* Convert MPI object handles to object pointers */
+ MPID_Comm_get_ptr(comm, comm_ptr);
+
+ /* Validate parameters and objects (post conversion) */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ MPID_Comm_valid_ptr(comm_ptr, mpi_errno);
+ /* TODO more checks may be appropriate (counts, in_place, buffer aliasing, etc) */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* ... body of routine ... */
+
+ mpi_errno = MPIR_Mprobe_impl(source, tag, comm_ptr, message, status);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ /* ... end of body of routine ... */
+
+fn_exit:
+ MPID_MPI_FUNC_EXIT(MPID_STATE_MPIX_MPROBE);
+ MPIU_THREAD_CS_EXIT(ALLFUNC,);
+ return mpi_errno;
+
+fn_fail:
+ /* --BEGIN ERROR HANDLING-- */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ mpi_errno = MPIR_Err_create_code(
+ mpi_errno, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPI_ERR_OTHER,
+ "**mpix_mprobe", "**mpix_mprobe %d %d %C %p %p", source, tag, comm, message, status);
+ }
+# endif
+ mpi_errno = MPIR_Err_return_comm(NULL, FCNAME, mpi_errno);
+ goto fn_exit;
+ /* --END ERROR HANDLING-- */
+}
diff --git a/src/mpi/pt2pt/mrecv.c b/src/mpi/pt2pt/mrecv.c
new file mode 100644
index 0000000..956d822
--- /dev/null
+++ b/src/mpi/pt2pt/mrecv.c
@@ -0,0 +1,181 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include "mpiimpl.h"
+
+/* -- Begin Profiling Symbol Block for routine MPIX_Mrecv */
+#if defined(HAVE_PRAGMA_WEAK)
+#pragma weak MPIX_Mrecv = PMPIX_Mrecv
+#elif defined(HAVE_PRAGMA_HP_SEC_DEF)
+#pragma _HP_SECONDARY_DEF PMPIX_Mrecv MPIX_Mrecv
+#elif defined(HAVE_PRAGMA_CRI_DUP)
+#pragma _CRI duplicate MPIX_Mrecv as PMPIX_Mrecv
+#endif
+/* -- End Profiling Symbol Block */
+
+/* Define MPICH_MPI_FROM_PMPI if weak symbols are not supported to build
+ the MPI routines */
+#ifndef MPICH_MPI_FROM_PMPI
+#undef MPIX_Mrecv
+#define MPIX_Mrecv PMPIX_Mrecv
+
+/* any non-MPI functions go here, especially non-static ones */
+
+#undef FUNCNAME
+#define FUNCNAME MPIR_Mrecv_impl
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPIR_Mrecv_impl(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPI_Request req_handle; /* dummy for MPIR_Request_complete */
+ int active_flag; /* dummy for MPIR_Request_complete */
+ MPID_Request *msgp = NULL;
+ MPID_Request *rreq = NULL;
+
+ /* get the underlying MPID_Request object and validate it */
+ MPID_Request_get_ptr(*message, msgp);
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS;
+ {
+ MPID_Request_valid_ptr(msgp, mpi_errno);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ MPIU_ERR_CHKANDJUMP((msgp->kind != MPID_REQUEST_MPROBE),
+ mpi_errno, MPI_ERR_ARG, "**reqnotmsg")
+ }
+ MPID_END_ERROR_CHECKS;
+ }
+# endif
+
+ /* There is no MPID_Mrecv at this time because there is no real optimization
+ * potential in that case. MPID_Recv exists to prevent creating a request
+ * unnecessarily for messages that are already present and eligible for
+ * immediate completion. */
+ mpi_errno = MPID_Imrecv(buf, count, datatype, msgp, &rreq);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ if (!MPID_Request_is_complete(rreq)) {
+ MPID_Progress_state progress_state;
+
+ MPID_Progress_start(&progress_state);
+ while (!MPID_Request_is_complete(rreq))
+ {
+ mpi_errno = MPID_Progress_wait(&progress_state);
+ if (mpi_errno) {
+ /* --BEGIN ERROR HANDLING-- */
+ MPID_Progress_end(&progress_state);
+ MPIU_ERR_POP(mpi_errno);
+ /* --END ERROR HANDLING-- */
+ }
+ }
+ MPID_Progress_end(&progress_state);
+ }
+ mpi_errno = MPIR_Request_complete(&req_handle, rreq, status, &active_flag);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ *message = MPIX_MESSAGE_NULL;
+
+fn_exit:
+ return mpi_errno;
+fn_fail:
+ goto fn_exit;
+}
+
+#endif /* MPICH_MPI_FROM_PMPI */
+
+#undef FUNCNAME
+#define FUNCNAME MPIX_Mrecv
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+/*@
+MPIX_Mrecv - XXX description here
+
+Input/Output Parameters:
+. message - message (handle)
+
+Input Parameters:
++ count - number of elements in the receive buffer (non-negative integer)
+- datatype - datatype of each receive buffer element (handle)
+
+Output Parameters:
++ buf - initial address of the receive buffer (choice)
+- status - status object (status)
+
+.N ThreadSafe
+
+.N Fortran
+
+.N Errors
+@*/
+int MPIX_Mrecv(void *buf, int count, MPI_Datatype datatype, MPIX_Message *message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_MPI_STATE_DECL(MPID_STATE_MPIX_MRECV);
+
+ MPIU_THREAD_CS_ENTER(ALLFUNC,);
+ MPID_MPI_FUNC_ENTER(MPID_STATE_MPIX_MRECV);
+
+ /* Validate parameters, especially handles needing to be converted */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ MPIR_ERRTEST_DATATYPE(datatype, "datatype", mpi_errno);
+
+ /* TODO more checks may be appropriate */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* Convert MPI object handles to object pointers */
+
+ /* Validate parameters and objects (post conversion) */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ MPID_BEGIN_ERROR_CHECKS
+ {
+ if (HANDLE_GET_KIND(datatype) != HANDLE_KIND_BUILTIN) {
+ MPID_Datatype *datatype_ptr = NULL;
+ MPID_Datatype_get_ptr(datatype, datatype_ptr);
+ MPID_Datatype_valid_ptr(datatype_ptr, mpi_errno);
+ MPID_Datatype_committed_ptr(datatype_ptr, mpi_errno);
+ }
+
+ /* TODO more checks may be appropriate (counts, in_place, buffer aliasing, etc) */
+ if (mpi_errno != MPI_SUCCESS) goto fn_fail;
+ }
+ MPID_END_ERROR_CHECKS
+ }
+# endif /* HAVE_ERROR_CHECKING */
+
+ /* ... body of routine ... */
+
+ mpi_errno = MPIR_Mrecv_impl(buf, count, datatype, message, status);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ /* ... end of body of routine ... */
+
+fn_exit:
+ MPID_MPI_FUNC_EXIT(MPID_STATE_MPIX_MRECV);
+ MPIU_THREAD_CS_EXIT(ALLFUNC,);
+ return mpi_errno;
+
+fn_fail:
+ /* --BEGIN ERROR HANDLING-- */
+# ifdef HAVE_ERROR_CHECKING
+ {
+ mpi_errno = MPIR_Err_create_code(
+ mpi_errno, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPI_ERR_OTHER,
+ "**mpix_mrecv", "**mpix_mrecv %p %d %D %p %p", buf, count, datatype, message, status);
+ }
+# endif
+ mpi_errno = MPIR_Err_return_comm(NULL, FCNAME, mpi_errno);
+ goto fn_exit;
+ /* --END ERROR HANDLING-- */
+}
diff --git a/src/mpi/pt2pt/waitall.c b/src/mpi/pt2pt/waitall.c
index 32d170b..7784464 100644
--- a/src/mpi/pt2pt/waitall.c
+++ b/src/mpi/pt2pt/waitall.c
@@ -69,6 +69,8 @@ int MPIR_Waitall_impl(int count, MPI_Request array_of_requests[],
{
MPID_Request_valid_ptr( request_ptrs[i], mpi_errno );
if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ MPIU_ERR_CHKANDJUMP1((request_ptrs[i]->kind == MPID_REQUEST_MPROBE),
+ mpi_errno, MPI_ERR_ARG, "**msgnotreq", "**msgnotreq %d", i)
}
MPID_END_ERROR_CHECKS;
}
diff --git a/src/mpid/ch3/include/mpidimpl.h b/src/mpid/ch3/include/mpidimpl.h
index 04d85b9..453c194 100644
--- a/src/mpid/ch3/include/mpidimpl.h
+++ b/src/mpid/ch3/include/mpidimpl.h
@@ -1445,6 +1445,8 @@ int MPIDI_CH3_Connect_to_root(const char *, MPIDI_VC_t **);
int MPIDI_CH3U_Recvq_init(void);
int MPIDI_CH3U_Recvq_FU(int, int, int, MPI_Status * );
MPID_Request * MPIDI_CH3U_Recvq_FDU(MPI_Request, MPIDI_Message_match *);
+MPID_Request * MPIDI_CH3U_Recvq_FDU_matchonly(int source, int tag, int context_id, MPID_Comm *comm,
+ int *foundp);
MPID_Request * MPIDI_CH3U_Recvq_FDU_or_AEP(int source, int tag,
int context_id, MPID_Comm *comm, void *user_buf,
int user_count, MPI_Datatype datatype, int * foundp);
diff --git a/src/mpid/ch3/src/Makefile.mk b/src/mpid/ch3/src/Makefile.mk
index 80e72e1..14032fb 100644
--- a/src/mpid/ch3/src/Makefile.mk
+++ b/src/mpid/ch3/src/Makefile.mk
@@ -35,12 +35,15 @@ lib_lib@MPILIBNAME@_la_SOURCES += \
src/mpid/ch3/src/mpid_finalize.c \
src/mpid/ch3/src/mpid_get_universe_size.c \
src/mpid/ch3/src/mpid_getpname.c \
+ src/mpid/ch3/src/mpid_improbe.c \
+ src/mpid/ch3/src/mpid_imrecv.c \
src/mpid/ch3/src/mpid_init.c \
src/mpid/ch3/src/mpid_iprobe.c \
src/mpid/ch3/src/mpid_irecv.c \
src/mpid/ch3/src/mpid_irsend.c \
src/mpid/ch3/src/mpid_isend.c \
src/mpid/ch3/src/mpid_issend.c \
+ src/mpid/ch3/src/mpid_mprobe.c \
src/mpid/ch3/src/mpid_port.c \
src/mpid/ch3/src/mpid_probe.c \
src/mpid/ch3/src/mpid_recv.c \
diff --git a/src/mpid/ch3/src/ch3u_recvq.c b/src/mpid/ch3/src/ch3u_recvq.c
index 3e162c8..fdee9b4 100644
--- a/src/mpid/ch3/src/ch3u_recvq.c
+++ b/src/mpid/ch3/src/ch3u_recvq.c
@@ -313,6 +313,103 @@ MPID_Request * MPIDI_CH3U_Recvq_FDU(MPI_Request sreq_id,
return rreq;
}
+/* TODO rename the old FDU and use that name for this one */
+/* This is the routine that you expect to be named "_FDU". It implements the
+ * behavior needed for improbe; specifically, searching the receive queue for
+ * the first matching request and dequeueing it. */
+#undef FUNCNAME
+#define FUNCNAME MPIDI_CH3U_Recvq_FDU_matchonly
+#undef FCNAME
+#define FCNAME MPIDI_QUOTE(FUNCNAME)
+MPID_Request * MPIDI_CH3U_Recvq_FDU_matchonly(int source, int tag, int context_id, MPID_Comm *comm, int *foundp)
+{
+ int found = FALSE;
+ MPID_Request *rreq, *prev_rreq;
+ MPIDI_Message_match match;
+ MPIDI_Message_match mask;
+ MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3U_RECVQ_FDU_MATCHONLY);
+
+ MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3U_RECVQ_FDU_MATCHONLY);
+
+ MPIU_THREAD_CS_ASSERT_HELD(MSGQUEUE);
+
+ /* Optimize this loop for an empty unexpected receive queue */
+ rreq = recvq_unexpected_head;
+ if (rreq) {
+ prev_rreq = NULL;
+
+ match.parts.context_id = context_id;
+ match.parts.tag = tag;
+ match.parts.rank = source;
+
+ if (tag != MPI_ANY_TAG && source != MPI_ANY_SOURCE) {
+ do {
+ if (MATCH_WITH_NO_MASK(rreq->dev.match, match)) {
+ if (prev_rreq != NULL) {
+ prev_rreq->dev.next = rreq->dev.next;
+ }
+ else {
+ recvq_unexpected_head = rreq->dev.next;
+ }
+
+ if (rreq->dev.next == NULL) {
+ recvq_unexpected_tail = prev_rreq;
+ }
+#ifdef ENABLE_RECVQ_STATISTICS
+ --unexpected_qlen;
+#endif
+
+ rreq->comm = comm;
+ MPIR_Comm_add_ref(comm);
+ /* don't have the (buf,count,type) info right now, can't add
+ * it to the request */
+ found = TRUE;
+ goto lock_exit;
+ }
+ prev_rreq = rreq;
+ rreq = rreq->dev.next;
+ } while (rreq);
+ }
+ else {
+ mask.parts.context_id = mask.parts.rank = mask.parts.tag = ~0;
+ if (tag == MPI_ANY_TAG)
+ match.parts.tag = mask.parts.tag = 0;
+ if (source == MPI_ANY_SOURCE)
+ match.parts.rank = mask.parts.rank = 0;
+
+ do {
+ if (MATCH_WITH_LEFT_MASK(rreq->dev.match, match, mask)) {
+ if (prev_rreq != NULL) {
+ prev_rreq->dev.next = rreq->dev.next;
+ }
+ else {
+ recvq_unexpected_head = rreq->dev.next;
+ }
+ if (rreq->dev.next == NULL) {
+ recvq_unexpected_tail = prev_rreq;
+ }
+#ifdef ENABLE_RECVQ_STATISTICS
+ --unexpected_qlen;
+#endif
+ rreq->comm = comm;
+ MPIR_Comm_add_ref(comm);
+ /* don't have the (buf,count,type) info right now, can't add
+ * it to the request */
+ found = TRUE;
+ goto lock_exit;
+ }
+ prev_rreq = rreq;
+ rreq = rreq->dev.next;
+ } while (rreq);
+ }
+ }
+
+lock_exit:
+ *foundp = found;
+
+ MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_CH3U_RECVQ_FDU_MATCHONLY);
+ return rreq;
+}
/*
* MPIDI_CH3U_Recvq_FDU_or_AEP()
diff --git a/src/mpid/ch3/src/mpid_improbe.c b/src/mpid/ch3/src/mpid_improbe.c
new file mode 100644
index 0000000..50cbae6
--- /dev/null
+++ b/src/mpid/ch3/src/mpid_improbe.c
@@ -0,0 +1,57 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include "mpidimpl.h"
+
+/* FIXME I think we still need to handle the anysource probe case for
+ * channel/netmod override. See MPID_Iprobe for more info. */
+
+#undef FUNCNAME
+#define FUNCNAME MPID_Improbe
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPID_Improbe(int source, int tag, MPID_Comm *comm, int context_offset,
+ int *flag, MPID_Request **message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ int context_id = comm->recvcontext_id + context_offset;
+
+ *message = NULL;
+
+ if (source == MPI_PROC_NULL)
+ {
+ MPIR_Status_set_procnull(status);
+ *flag = TRUE;
+ goto fn_exit;
+ }
+
+ MPIU_THREAD_CS_ENTER(MSGQUEUE,);
+ *message = MPIDI_CH3U_Recvq_FDU_matchonly(source, tag, context_id, comm, flag);
+ MPIU_THREAD_CS_EXIT(MSGQUEUE,);
+
+ if (!*flag) {
+ /* Always try to advance progress before returning failure
+ from the improbe test. */
+ /* FIXME: It would be helpful to know if the Progress_poke
+ operation causes any change in state; we could then avoid
+ a second test of the receive queue if we knew that nothing
+ had changed */
+ mpi_errno = MPID_Progress_poke();
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ MPIU_THREAD_CS_ENTER(MSGQUEUE,);
+ *message = MPIDI_CH3U_Recvq_FDU_matchonly(source, tag, context_id, comm, flag);
+ MPIU_THREAD_CS_EXIT(MSGQUEUE,);
+ }
+
+ if (*flag && *message)
+ (*message)->kind = MPID_REQUEST_MPROBE;
+
+fn_exit:
+ return mpi_errno;
+fn_fail:
+ goto fn_exit;
+}
+
diff --git a/src/mpid/ch3/src/mpid_imrecv.c b/src/mpid/ch3/src/mpid_imrecv.c
new file mode 100644
index 0000000..8b5ecef
--- /dev/null
+++ b/src/mpid/ch3/src/mpid_imrecv.c
@@ -0,0 +1,119 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include "mpidimpl.h"
+
+#undef FUNCNAME
+#define FUNCNAME MPID_Imrecv
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPID_Imrecv(void *buf, int count, MPI_Datatype datatype,
+ MPID_Request *message, MPID_Request **rreqp)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_Request *rreq;
+ MPID_Comm *comm;
+ MPIDI_VC_t *vc = NULL;
+
+ /* promote the request object to be a "real" recv request */
+ message->kind = MPID_REQUEST_RECV;
+
+ *rreqp = rreq = message;
+
+ comm = rreq->comm;
+
+ /* the following code was adapted from FDU_or_AEP and MPID_Irecv */
+ /* comm was already added at FDU (mprobe) time */
+ rreq->dev.user_buf = buf;
+ rreq->dev.user_count = count;
+ rreq->dev.datatype = datatype;
+
+ if (MPIDI_Request_get_msg_type(rreq) == MPIDI_REQUEST_EAGER_MSG)
+ {
+ int recv_pending;
+
+ /* This is an eager message */
+ MPIU_DBG_MSG(CH3_OTHER,VERBOSE,"eager message in the request");
+
+ /* If this is an eager synchronous message, then we need to send an
+ acknowledgement back to the sender. */
+ if (MPIDI_Request_get_sync_send_flag(rreq))
+ {
+ MPIDI_Comm_get_vc_set_active(comm, rreq->dev.match.parts.rank, &vc);
+ mpi_errno = MPIDI_CH3_EagerSyncAck(vc, rreq);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ }
+
+ /* the request was found in the unexpected queue, so it has a
+ recv_pending_count of at least 1 */
+ MPIDI_Request_decr_pending(rreq);
+ MPIDI_Request_check_pending(rreq, &recv_pending);
+
+ if (MPID_Request_is_complete(rreq)) {
+ /* is it ever possible to have (cc==0 && recv_pending>0) ? */
+ MPIU_Assert(!recv_pending);
+
+ /* All of the data has arrived, we need to copy the data and
+ then free the buffer. */
+ if (rreq->dev.recv_data_sz > 0)
+ {
+ MPIDI_CH3U_Request_unpack_uebuf(rreq);
+ MPIU_Free(rreq->dev.tmpbuf);
+ }
+
+ mpi_errno = rreq->status.MPI_ERROR;
+ goto fn_exit;
+ }
+ else
+ {
+ /* there should never be outstanding completion events for an unexpected
+ * recv without also having a "pending recv" */
+ MPIU_Assert(recv_pending);
+ /* The data is still being transfered across the net. We'll
+ leave it to the progress engine to handle once the
+ entire message has arrived. */
+ if (HANDLE_GET_KIND(datatype) != HANDLE_KIND_BUILTIN)
+ {
+ MPID_Datatype_get_ptr(datatype, rreq->dev.datatype_ptr);
+ MPID_Datatype_add_ref(rreq->dev.datatype_ptr);
+ }
+
+ }
+ }
+ else if (MPIDI_Request_get_msg_type(rreq) == MPIDI_REQUEST_RNDV_MSG)
+ {
+ MPIDI_Comm_get_vc_set_active(comm, rreq->dev.match.parts.rank, &vc);
+
+ mpi_errno = vc->rndvRecv_fn(vc, rreq);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ if (HANDLE_GET_KIND(datatype) != HANDLE_KIND_BUILTIN)
+ {
+ MPID_Datatype_get_ptr(datatype, rreq->dev.datatype_ptr);
+ MPID_Datatype_add_ref(rreq->dev.datatype_ptr);
+ }
+ }
+ else if (MPIDI_Request_get_msg_type(rreq) == MPIDI_REQUEST_SELF_MSG)
+ {
+ mpi_errno = MPIDI_CH3_RecvFromSelf(rreq, buf, count, datatype);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ }
+ else
+ {
+ /* --BEGIN ERROR HANDLING-- */
+ int msg_type = MPIDI_Request_get_msg_type(rreq);
+ MPID_Request_release(rreq);
+ rreq = NULL;
+ MPIU_ERR_SETANDJUMP1(mpi_errno,MPI_ERR_INTERN, "**ch3|badmsgtype",
+ "**ch3|badmsgtype %d", msg_type);
+ /* --END ERROR HANDLING-- */
+ }
+
+fn_exit:
+ return mpi_errno;
+fn_fail:
+ goto fn_exit;
+}
+
diff --git a/src/mpid/ch3/src/mpid_mprobe.c b/src/mpid/ch3/src/mpid_mprobe.c
new file mode 100644
index 0000000..1191317
--- /dev/null
+++ b/src/mpid/ch3/src/mpid_mprobe.c
@@ -0,0 +1,69 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include "mpidimpl.h"
+
+/* FIXME I think we still need to handle the anysource probe case for
+ * channel/netmod override. See MPID_Iprobe for more info. */
+
+#undef FUNCNAME
+#define FUNCNAME MPID_Mprobe
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPID_Mprobe(int source, int tag, MPID_Comm *comm, int context_offset,
+ MPID_Request **message, MPI_Status *status)
+{
+ int mpi_errno = MPI_SUCCESS;
+ MPID_Progress_state progress_state;
+ int found = FALSE;
+ int context_id = comm->recvcontext_id + context_offset;
+
+ *message = NULL;
+
+ if (source == MPI_PROC_NULL)
+ {
+ MPIR_Status_set_procnull(status);
+ found = TRUE;
+ goto fn_exit;
+ }
+
+ /* Inefficient implementation: we poll the unexpected queue looking for a
+ * matching request, interleaved with calls to progress. If there are many
+ * non-matching unexpected messages in the queue then we will end up
+ * needlessly scanning the UQ.
+ *
+ * A smarter implementation would enqueue a partial request (one lacking the
+ * recv buffer triple) onto the PQ. Unfortunately, this is a lot harder to
+ * do than it seems at first because of the spread-out nature of callers to
+ * various CH3U_Recvq routines and especially because of the enqueue/dequeue
+ * hooks for native MX tag matching support. */
+
+ MPIDI_CH3_Progress_start(&progress_state);
+ do
+ {
+ MPIU_THREAD_CS_ENTER(MSGQUEUE,);
+ *message = MPIDI_CH3U_Recvq_FDU_matchonly(source, tag, context_id, comm, &found);
+ MPIU_THREAD_CS_EXIT(MSGQUEUE,);
+ if (found)
+ break;
+
+ mpi_errno = MPIDI_CH3_Progress_wait(&progress_state);
+ }
+ while(mpi_errno == MPI_SUCCESS);
+ MPIDI_CH3_Progress_end(&progress_state);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+fn_exit:
+ if (*message) {
+ (*message)->kind = MPID_REQUEST_MPROBE;
+ MPIR_Request_extract_status((*message), status);
+ }
+
+ return mpi_errno;
+fn_fail:
+ goto fn_exit;
+}
+
diff --git a/test/mpi/pt2pt/Makefile.am b/test/mpi/pt2pt/Makefile.am
index b3a8638..5031100 100644
--- a/test/mpi/pt2pt/Makefile.am
+++ b/test/mpi/pt2pt/Makefile.am
@@ -47,5 +47,6 @@ noinst_PROGRAMS = \
inactivereq \
waittestnull \
sendall \
- large_message
+ large_message \
+ mprobe
diff --git a/test/mpi/pt2pt/mprobe.c b/test/mpi/pt2pt/mprobe.c
new file mode 100644
index 0000000..c161bae
--- /dev/null
+++ b/test/mpi/pt2pt/mprobe.c
@@ -0,0 +1,213 @@
+/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
+/*
+ * (C) 2012 by Argonne National Laboratory.
+ * See COPYRIGHT in top-level directory.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include "mpi.h"
+
+/* This is a temporary #ifdef to control whether we test this functionality. A
+ * configure-test or similar would be better. Eventually the MPI-3 standard
+ * will be released and this can be gated on a MPI_VERSION check */
+#if !defined(USE_STRICT_MPI) && defined(MPICH2)
+#define TEST_MPROBE_ROUTINES 1
+#endif
+
+/* assert-like macro that bumps the err count and emits a message */
+#define check(x_) \
+ do { \
+ if (!(x_)) { \
+ ++errs; \
+ if (errs < 10) { \
+ fprintf(stderr, "check failed: (%s), line %d\n", #x_, __LINE__); \
+ } \
+ } \
+ } while (0)
+
+int main(int argc, char **argv)
+{
+ int errs = 0;
+ int found;
+ int rank, size;
+ int sendbuf[8], recvbuf[8];
+ int count;
+ MPIX_Message msg;
+ MPI_Request rreq;
+ MPI_Status s1, s2;
+
+ MPI_Init(&argc, &argv);
+
+ MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+ MPI_Comm_size(MPI_COMM_WORLD, &size);
+
+ if (size < 2) {
+ printf("this test requires at least 2 processes\n");
+ MPI_Abort(MPI_COMM_WORLD, 1);
+ }
+
+ /* all processes besides ranks 0 & 1 aren't used by this test */
+ if (rank >= 2) {
+ goto epilogue;
+ }
+
+#ifdef TEST_MPROBE_ROUTINES
+ /* test 0: simple send & mprobe+mrecv */
+ if (rank == 0) {
+ sendbuf[0] = 0xdeadbeef;
+ sendbuf[1] = 0xfeedface;
+ MPI_Send(sendbuf, 2, MPI_INT, 1, 5, MPI_COMM_WORLD);
+ }
+ else {
+ /* the error field should remain unmodified */
+ s1.MPI_ERROR = MPI_ERR_DIMS;
+ s2.MPI_ERROR = MPI_ERR_TOPOLOGY;
+
+ msg = MPIX_MESSAGE_NULL;
+ MPIX_Mprobe(0, 5, MPI_COMM_WORLD, &msg, &s1);
+ check(s1.MPI_SOURCE == 0);
+ check(s1.MPI_TAG == 5);
+ check(s1.MPI_ERROR == MPI_ERR_DIMS);
+
+ count = -1;
+ MPI_Get_count(&s1, MPI_INT, &count);
+ check(count == 2);
+
+ MPIX_Mrecv(recvbuf, count, MPI_INT, &msg, &s2);
+ check(recvbuf[0] == 0xdeadbeef);
+ check(recvbuf[1] == 0xfeedface);
+ check(s2.MPI_SOURCE == 0);
+ check(s2.MPI_TAG == 5);
+ check(s2.MPI_ERROR == MPI_ERR_TOPOLOGY);
+ check(msg == MPIX_MESSAGE_NULL);
+ }
+
+ /* test 1: simple send & mprobe+imrecv */
+ if (rank == 0) {
+ sendbuf[0] = 0xdeadbeef;
+ sendbuf[1] = 0xfeedface;
+ MPI_Send(sendbuf, 2, MPI_INT, 1, 5, MPI_COMM_WORLD);
+ }
+ else {
+ /* the error field should remain unmodified */
+ s1.MPI_ERROR = MPI_ERR_DIMS;
+ s2.MPI_ERROR = MPI_ERR_TOPOLOGY;
+
+ msg = MPIX_MESSAGE_NULL;
+ MPIX_Mprobe(0, 5, MPI_COMM_WORLD, &msg, &s1);
+ check(s1.MPI_SOURCE == 0);
+ check(s1.MPI_TAG == 5);
+ check(s1.MPI_ERROR == MPI_ERR_DIMS);
+
+ count = -1;
+ MPI_Get_count(&s1, MPI_INT, &count);
+ check(count == 2);
+
+ rreq = MPI_REQUEST_NULL;
+ MPIX_Imrecv(recvbuf, count, MPI_INT, &msg, &rreq);
+ check(rreq != MPI_REQUEST_NULL);
+ MPI_Wait(&rreq, &s2);
+ check(recvbuf[0] == 0xdeadbeef);
+ check(recvbuf[1] == 0xfeedface);
+ check(s2.MPI_SOURCE == 0);
+ check(s2.MPI_TAG == 5);
+ check(s2.MPI_ERROR == MPI_ERR_TOPOLOGY);
+ check(msg == MPIX_MESSAGE_NULL);
+ }
+
+ /* test 2: simple send & improbe+mrecv */
+ if (rank == 0) {
+ sendbuf[0] = 0xdeadbeef;
+ sendbuf[1] = 0xfeedface;
+ MPI_Send(sendbuf, 2, MPI_INT, 1, 5, MPI_COMM_WORLD);
+ }
+ else {
+ /* the error field should remain unmodified */
+ s1.MPI_ERROR = MPI_ERR_DIMS;
+ s2.MPI_ERROR = MPI_ERR_TOPOLOGY;
+
+ msg = MPIX_MESSAGE_NULL;
+ do {
+ check(msg == MPIX_MESSAGE_NULL);
+ MPIX_Improbe(0, 5, MPI_COMM_WORLD, &found, &msg, &s1);
+ } while (!found);
+ check(msg != MPIX_MESSAGE_NULL);
+ check(s1.MPI_SOURCE == 0);
+ check(s1.MPI_TAG == 5);
+ check(s1.MPI_ERROR == MPI_ERR_DIMS);
+
+ count = -1;
+ MPI_Get_count(&s1, MPI_INT, &count);
+ check(count == 2);
+
+ MPIX_Mrecv(recvbuf, count, MPI_INT, &msg, &s2);
+ check(recvbuf[0] == 0xdeadbeef);
+ check(recvbuf[1] == 0xfeedface);
+ check(s2.MPI_SOURCE == 0);
+ check(s2.MPI_TAG == 5);
+ check(s2.MPI_ERROR == MPI_ERR_TOPOLOGY);
+ check(msg == MPIX_MESSAGE_NULL);
+ }
+
+ /* test 3: simple send & improbe+imrecv */
+ if (rank == 0) {
+ sendbuf[0] = 0xdeadbeef;
+ sendbuf[1] = 0xfeedface;
+ MPI_Send(sendbuf, 2, MPI_INT, 1, 5, MPI_COMM_WORLD);
+ }
+ else {
+ /* the error field should remain unmodified */
+ s1.MPI_ERROR = MPI_ERR_DIMS;
+ s2.MPI_ERROR = MPI_ERR_TOPOLOGY;
+
+ msg = MPIX_MESSAGE_NULL;
+ do {
+ check(msg == MPIX_MESSAGE_NULL);
+ MPIX_Improbe(0, 5, MPI_COMM_WORLD, &found, &msg, &s1);
+ } while (!found);
+ check(msg != MPIX_MESSAGE_NULL);
+ check(s1.MPI_SOURCE == 0);
+ check(s1.MPI_TAG == 5);
+ check(s1.MPI_ERROR == MPI_ERR_DIMS);
+
+ count = -1;
+ MPI_Get_count(&s1, MPI_INT, &count);
+ check(count == 2);
+
+ rreq = MPI_REQUEST_NULL;
+ MPIX_Imrecv(recvbuf, count, MPI_INT, &msg, &rreq);
+ check(rreq != MPI_REQUEST_NULL);
+ MPI_Wait(&rreq, &s2);
+ check(recvbuf[0] == 0xdeadbeef);
+ check(recvbuf[1] == 0xfeedface);
+ check(s2.MPI_SOURCE == 0);
+ check(s2.MPI_TAG == 5);
+ check(s2.MPI_ERROR == MPI_ERR_TOPOLOGY);
+ check(msg == MPIX_MESSAGE_NULL);
+ }
+
+ /* TODO MPI_ANY_SOURCE and MPI_ANY_TAG should be tested as well */
+ /* TODO a full range of message sizes should be tested too */
+ /* TODO threaded tests are also needed, but they should go in a separate
+ * program */
+
+#endif /* TEST_MPROBE_ROUTINES */
+
+epilogue:
+ MPI_Reduce((rank == 0 ? MPI_IN_PLACE : &errs), &errs, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
+ if (rank == 0) {
+ if (errs) {
+ printf("found %d errors\n", errs);
+ }
+ else {
+ printf(" No errors\n");
+ }
+ }
+
+ MPI_Finalize();
+
+ return 0;
+}
+
diff --git a/test/mpi/pt2pt/testlist b/test/mpi/pt2pt/testlist
index efb71f4..1300f28 100644
--- a/test/mpi/pt2pt/testlist
+++ b/test/mpi/pt2pt/testlist
@@ -36,3 +36,4 @@ waitany-null 1
# this should be run only on machines with large amount of memory (>=8GB)
# perhaps disable in the release tarball
large_message 3
+mprobe 2
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment