Skip to content

Commit c7c24e4

Browse files
hppritchahppritcha
authored
Merge pull request #9667 from hppritcha/topic/add_isendrecv_variants
MPI4: implement mpi_isendrecv and variant
2 parents 5782c0a + 47c8063 commit c7c24e4

25 files changed

+1107
-1
lines changed

‎ompi/communicator/comm_request.c

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@
88
* of Tennessee Research Foundation. All rights
99
* reserved.
1010
* Copyright (c) 2016 IBM Corporation. All rights reserved.
11+
* Copyright (c) 2021 Triad National Security, LLC. All rights
12+
* reserved.
1113
* $COPYRIGHT$
1214
*
1315
* Additional copyrights may follow
@@ -30,6 +32,7 @@ typedef struct ompi_comm_request_item_t {
3032
opal_list_item_t super;
3133
ompi_comm_request_callback_fn_t callback;
3234
ompi_request_t *subreqs[OMPI_COMM_REQUEST_MAX_SUBREQ];
35+
uint32_t flags;
3336
int subreq_count;
3437
} ompi_comm_request_item_t;
3538
OBJ_CLASS_DECLARATION(ompi_comm_request_item_t);
@@ -70,6 +73,12 @@ void ompi_comm_request_fini (void)
7073

7174
int ompi_comm_request_schedule_append (ompi_comm_request_t *request, ompi_comm_request_callback_fn_t callback,
7275
ompi_request_t *subreqs[], int subreq_count)
76+
{
77+
return ompi_comm_request_schedule_append_w_flags(request, callback, subreqs, subreq_count, 0);
78+
}
79+
80+
int ompi_comm_request_schedule_append_w_flags(ompi_comm_request_t *request, ompi_comm_request_callback_fn_t callback,
81+
ompi_request_t *subreqs[], int subreq_count, uint32_t flags)
7382
{
7483
ompi_comm_request_item_t *request_item;
7584
int i;
@@ -84,6 +93,7 @@ int ompi_comm_request_schedule_append (ompi_comm_request_t *request, ompi_comm_r
8493
}
8594

8695
request_item->callback = callback;
96+
request_item->flags = flags;
8797

8898
for (i = 0 ; i < subreq_count ; ++i) {
8999
request_item->subreqs[i] = subreqs[i];
@@ -125,7 +135,9 @@ static int ompi_comm_request_progress (void)
125135
* that it does some subreqs cleanup */
126136
request->super.req_status.MPI_ERROR = subreq->req_status.MPI_ERROR;
127137
}
128-
ompi_request_free (&subreq);
138+
if (!(request_item->flags & OMPI_COMM_REQ_FLAG_RETAIN_SUBREQ)) {
139+
ompi_request_free (&subreq);
140+
}
129141
request_item->subreq_count--;
130142
completed++;
131143
} else {

‎ompi/communicator/comm_request.h

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22
/*
33
* Copyright (c) 2013-2016 Los Alamos National Security, LLC. All rights
44
* reseved.
5+
* Copyright (c) 2021 Triad National Security, LLC. All rights
6+
* reserved.
57
* $COPYRIGHT$
68
*
79
* Additional copyrights may follow
@@ -18,6 +20,9 @@
1820
/* increase this number if more subrequests are needed */
1921
#define OMPI_COMM_REQUEST_MAX_SUBREQ 2
2022

23+
/* indicate that the caller will free subrequests */
24+
#define OMPI_COMM_REQ_FLAG_RETAIN_SUBREQ 0x00000001
25+
2126
typedef struct ompi_comm_request_t {
2227
ompi_request_t super;
2328

@@ -32,6 +37,8 @@ void ompi_comm_request_init (void);
3237
void ompi_comm_request_fini (void);
3338
int ompi_comm_request_schedule_append (ompi_comm_request_t *request, ompi_comm_request_callback_fn_t callback,
3439
ompi_request_t *subreqs[], int subreq_count);
40+
int ompi_comm_request_schedule_append_w_flags(ompi_comm_request_t *request, ompi_comm_request_callback_fn_t callback,
41+
ompi_request_t *subreqs[], int subreq_count, uint32_t flags);
3542
void ompi_comm_request_start (ompi_comm_request_t *request);
3643
ompi_comm_request_t *ompi_comm_request_get (void);
3744
void ompi_comm_request_return (ompi_comm_request_t *request);

‎ompi/include/mpi.h.in

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1703,6 +1703,13 @@ OMPI_DECLSPEC int MPI_Irsend(const void *buf, int count, MPI_Datatype datatype,
17031703
int tag, MPI_Comm comm, MPI_Request *request);
17041704
OMPI_DECLSPEC int MPI_Isend(const void *buf, int count, MPI_Datatype datatype, int dest,
17051705
int tag, MPI_Comm comm, MPI_Request *request);
1706+
OMPI_DECLSPEC int MPI_Isendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
1707+
int dest, int sendtag, void *recvbuf, int recvcount,
1708+
MPI_Datatype recvtype, int source, int recvtag,
1709+
MPI_Comm comm, MPI_Request *request);
1710+
OMPI_DECLSPEC int MPI_Isendrecv_replace(void * buf, int count, MPI_Datatype datatype,
1711+
int dest, int sendtag, int source, int recvtag,
1712+
MPI_Comm comm, MPI_Request *request);
17061713
OMPI_DECLSPEC int MPI_Issend(const void *buf, int count, MPI_Datatype datatype, int dest,
17071714
int tag, MPI_Comm comm, MPI_Request *request);
17081715
OMPI_DECLSPEC int MPI_Is_thread_main(int *flag);
@@ -2439,6 +2446,13 @@ OMPI_DECLSPEC int PMPI_Irsend(const void *buf, int count, MPI_Datatype datatype
24392446
int tag, MPI_Comm comm, MPI_Request *request);
24402447
OMPI_DECLSPEC int PMPI_Isend(const void *buf, int count, MPI_Datatype datatype, int dest,
24412448
int tag, MPI_Comm comm, MPI_Request *request);
2449+
OMPI_DECLSPEC int PMPI_Isendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
2450+
int dest, int sendtag, void *recvbuf, int recvcount,
2451+
MPI_Datatype recvtype, int source, int recvtag,
2452+
MPI_Comm comm, MPI_Request *request);
2453+
OMPI_DECLSPEC int PMPI_Isendrecv_replace(void * buf, int count, MPI_Datatype datatype,
2454+
int dest, int sendtag, int source, int recvtag,
2455+
MPI_Comm comm, MPI_Request *request);
24422456
OMPI_DECLSPEC int PMPI_Issend(const void *buf, int count, MPI_Datatype datatype, int dest,
24432457
int tag, MPI_Comm comm, MPI_Request *request);
24442458
OMPI_DECLSPEC int PMPI_Precv_init(void* buf, int partitions, MPI_Count count,

‎ompi/mpi/c/Makefile.am

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -284,6 +284,8 @@ interface_profile_sources = \
284284
irsend.c \
285285
is_thread_main.c \
286286
isend.c \
287+
isendrecv.c \
288+
isendrecv_replace.c \
287289
issend.c \
288290
lookup_name.c \
289291
message_f2c.c \

‎ompi/mpi/c/isendrecv.c

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
1+
/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
2+
/*
3+
* Copyright (c) 2004-2007 The Trustees of Indiana University and Indiana
4+
* University Research and Technology
5+
* Corporation. All rights reserved.
6+
* Copyright (c) 2004-2021 The University of Tennessee and The University
7+
* of Tennessee Research Foundation. All rights
8+
* reserved.
9+
* Copyright (c) 2004-2008 High Performance Computing Center Stuttgart,
10+
* University of Stuttgart. All rights reserved.
11+
* Copyright (c) 2004-2005 The Regents of the University of California.
12+
* All rights reserved.
13+
* Copyright (c) 2010-2012 Oak Ridge National Labs. All rights reserved.
14+
* Copyright (c) 2013 Los Alamos National Security, LLC. All rights
15+
* reserved.
16+
* Copyright (c) 2015 Research Organization for Information Science
17+
* and Technology (RIST). All rights reserved.
18+
* Copyright (c) 2021 Nanook Consulting. All rights reserved.
19+
* Copyright (c) 2021 Triad National Security, LLC. All rights
20+
* reserved.
21+
* $COPYRIGHT$
22+
*
23+
* Additional copyrights may follow
24+
*
25+
* $HEADER$
26+
*/
27+
#include "ompi_config.h"
28+
29+
#include "ompi/mpi/c/bindings.h"
30+
#include "ompi/runtime/params.h"
31+
#include "ompi/communicator/communicator.h"
32+
#include "ompi/communicator/comm_request.h"
33+
#include "ompi/errhandler/errhandler.h"
34+
#include "ompi/mca/pml/pml.h"
35+
#include "ompi/request/request.h"
36+
#include "ompi/memchecker.h"
37+
#include "ompi/runtime/ompi_spc.h"
38+
39+
#if OMPI_BUILD_MPI_PROFILING
40+
#if OPAL_HAVE_WEAK_SYMBOLS
41+
#pragma weak MPI_Isendrecv = PMPI_Isendrecv
42+
#endif
43+
#define MPI_Isendrecv PMPI_Isendrecv
44+
#endif
45+
46+
static const char FUNC_NAME[] = "MPI_Isendrecv";
47+
48+
struct ompi_isendrecv_context_t {
49+
opal_object_t super;
50+
int nreqs;
51+
int source;
52+
ompi_request_t *subreq[2];
53+
};
54+
55+
typedef struct ompi_isendrecv_context_t ompi_isendrecv_context_t;
56+
OBJ_CLASS_INSTANCE(ompi_isendrecv_context_t, opal_object_t, NULL, NULL);
57+
58+
static int ompi_isendrecv_complete_func (ompi_comm_request_t *request)
59+
{
60+
ompi_isendrecv_context_t *context =
61+
(ompi_isendrecv_context_t *) request->context;
62+
63+
/*
64+
* Copy the status from the receive side of the sendrecv request?
65+
* But what if the send failed?
66+
*
67+
* Probably need to bring up in the MPI forum.
68+
*/
69+
70+
if (MPI_PROC_NULL != context->source) {
71+
OMPI_COPY_STATUS(&request->super.req_status,
72+
context->subreq[0]->req_status, false);
73+
} else {
74+
OMPI_COPY_STATUS(&request->super.req_status,
75+
ompi_request_empty.req_status, false);
76+
}
77+
78+
if(NULL != context->subreq[0]) {
79+
ompi_request_free(&context->subreq[0]);
80+
}
81+
if(NULL != context->subreq[1]) {
82+
ompi_request_free(&context->subreq[1]);
83+
}
84+
85+
return OMPI_SUCCESS;
86+
}
87+
88+
89+
int MPI_Isendrecv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
90+
int dest, int sendtag, void *recvbuf, int recvcount,
91+
MPI_Datatype recvtype, int source, int recvtag,
92+
MPI_Comm comm, MPI_Request *request)
93+
{
94+
ompi_isendrecv_context_t *context = NULL;
95+
ompi_comm_request_t *crequest;
96+
int rc = MPI_SUCCESS;
97+
int nreqs = 0;
98+
uint32_t flags;
99+
100+
SPC_RECORD(OMPI_SPC_ISENDRECV, 1);
101+
102+
MEMCHECKER(
103+
memchecker_datatype(sendtype);
104+
memchecker_datatype(recvtype);
105+
memchecker_call(&opal_memchecker_base_isdefined, sendbuf, sendcount, sendtype);
106+
memchecker_comm(comm);
107+
);
108+
109+
if ( MPI_PARAM_CHECK ) {
110+
OMPI_ERR_INIT_FINALIZE(FUNC_NAME);
111+
OMPI_CHECK_DATATYPE_FOR_SEND(rc, sendtype, sendcount);
112+
OMPI_CHECK_DATATYPE_FOR_RECV(rc, recvtype, recvcount);
113+
OMPI_CHECK_USER_BUFFER(rc, sendbuf, sendtype, sendcount);
114+
OMPI_CHECK_USER_BUFFER(rc, recvbuf, recvtype, recvcount);
115+
116+
if (ompi_comm_invalid(comm)) {
117+
return OMPI_ERRHANDLER_NOHANDLE_INVOKE(MPI_ERR_COMM, FUNC_NAME);
118+
} else if (dest != MPI_PROC_NULL && ompi_comm_peer_invalid(comm, dest)) {
119+
rc = MPI_ERR_RANK;
120+
} else if (sendtag < 0 || sendtag > mca_pml.pml_max_tag) {
121+
rc = MPI_ERR_TAG;
122+
} else if (source != MPI_PROC_NULL && source != MPI_ANY_SOURCE && ompi_comm_peer_invalid(comm, source)) {
123+
rc = MPI_ERR_RANK;
124+
} else if (((recvtag < 0) && (recvtag != MPI_ANY_TAG)) || (recvtag > mca_pml.pml_max_tag)) {
125+
rc = MPI_ERR_TAG;
126+
} else if (request == NULL) {
127+
rc = MPI_ERR_REQUEST;
128+
}
129+
130+
OMPI_ERRHANDLER_CHECK(rc, comm, rc, FUNC_NAME);
131+
}
132+
133+
crequest = ompi_comm_request_get ();
134+
if (NULL == crequest) {
135+
return OMPI_ERR_OUT_OF_RESOURCE;
136+
}
137+
138+
context = OBJ_NEW(ompi_isendrecv_context_t);
139+
if (NULL == context) {
140+
ompi_comm_request_return (crequest);
141+
return OMPI_ERR_OUT_OF_RESOURCE;
142+
}
143+
144+
crequest->context = &context->super;
145+
context->subreq[0] = NULL;
146+
context->subreq[1] = NULL;
147+
context->source = source;
148+
149+
if (source != MPI_PROC_NULL) { /* post recv */
150+
rc = MCA_PML_CALL(irecv(recvbuf, recvcount, recvtype,
151+
source, recvtag, comm, &context->subreq[nreqs++]));
152+
if (MPI_SUCCESS != rc) {
153+
OBJ_RELEASE(context);
154+
ompi_comm_request_return (crequest);
155+
}
156+
OMPI_ERRHANDLER_CHECK(rc, comm, rc, FUNC_NAME);
157+
}
158+
159+
if (dest != MPI_PROC_NULL) { /* send */
160+
rc = MCA_PML_CALL(isend(sendbuf, sendcount, sendtype, dest,
161+
sendtag, MCA_PML_BASE_SEND_STANDARD, comm, &context->subreq[nreqs++]));
162+
if (MPI_SUCCESS != rc) {
163+
OBJ_RELEASE(context);
164+
ompi_comm_request_return (crequest);
165+
}
166+
OMPI_ERRHANDLER_CHECK(rc, comm, rc, FUNC_NAME);
167+
}
168+
169+
/*
170+
* schedule the operation
171+
*/
172+
173+
context->nreqs = nreqs;
174+
assert(nreqs <= 2);
175+
176+
flags = OMPI_COMM_REQ_FLAG_RETAIN_SUBREQ;
177+
178+
rc = ompi_comm_request_schedule_append_w_flags(crequest, ompi_isendrecv_complete_func,
179+
context->subreq, nreqs, flags);
180+
if (MPI_SUCCESS != rc) {
181+
OBJ_RELEASE(context);
182+
ompi_comm_request_return (crequest);
183+
}
184+
185+
OMPI_ERRHANDLER_CHECK(rc, comm, rc, FUNC_NAME);
186+
187+
/* kick off the request */
188+
189+
ompi_comm_request_start (crequest);
190+
*request = &crequest->super;
191+
192+
return rc;
193+
}

0 commit comments

Comments
 (0)