This source file includes following definitions.
- completion_fn
- ompi_mtl_portals4_iprobe
- ompi_mtl_portals4_improbe
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 #include "ompi_config.h"
21 #include "ompi/communicator/communicator.h"
22 #include "ompi/message/message.h"
23
24 #include "mtl_portals4.h"
25 #include "mtl_portals4_endpoint.h"
26 #include "mtl_portals4_request.h"
27 #include "mtl_portals4_message.h"
28
29 static int
30 completion_fn(ptl_event_t *ev, ompi_mtl_portals4_base_request_t *ptl_base_request)
31 {
32 ompi_mtl_portals4_probe_request_t *ptl_request =
33 (ompi_mtl_portals4_probe_request_t*) ptl_base_request;
34
35 opal_output_verbose(10, ompi_mtl_base_framework.framework_output,
36 "%s:%d: completion_fn: %d %d",
37 __FILE__, __LINE__, ev->type, ev->ni_fail_type);
38
39 if (OPAL_UNLIKELY(ev->ni_fail_type == PTL_OK)) {
40 ptl_request->found_match = 1;
41 ptl_request->status.MPI_SOURCE = MTL_PORTALS4_GET_SOURCE(ev->match_bits);
42 ptl_request->status.MPI_TAG = MTL_PORTALS4_GET_TAG(ev->match_bits);
43 ptl_request->status.MPI_ERROR = MPI_SUCCESS;
44 ptl_request->status._ucount = MTL_PORTALS4_GET_LENGTH(ev->hdr_data);
45 if (ev->type != PTL_EVENT_SEARCH) {
46 ptl_request->message = ompi_mtl_portals4_message_alloc(ev);
47 }
48 } else {
49 ptl_request->found_match = 0;
50 }
51 opal_atomic_wmb();
52 ptl_request->req_complete = 1;
53
54 return OMPI_SUCCESS;
55 }
56
57 int
58 ompi_mtl_portals4_iprobe(struct mca_mtl_base_module_t* mtl,
59 struct ompi_communicator_t *comm,
60 int src,
61 int tag,
62 int *flag,
63 struct ompi_status_public_t *status)
64 {
65 struct ompi_mtl_portals4_probe_request_t request;
66 ptl_me_t me;
67 ptl_process_t remote_proc;
68 ptl_match_bits_t match_bits, ignore_bits;
69 int ret;
70
71 if (MPI_ANY_SOURCE == src) {
72 if (ompi_mtl_portals4.use_logical) {
73 remote_proc.rank = PTL_RANK_ANY;
74 } else {
75 remote_proc.phys.nid = PTL_NID_ANY;
76 remote_proc.phys.pid = PTL_PID_ANY;
77 }
78 } else if ((ompi_mtl_portals4.use_logical) && (MPI_COMM_WORLD == comm)) {
79 remote_proc.rank = src;
80 } else {
81 ompi_proc_t* ompi_proc = ompi_comm_peer_lookup( comm, src );
82 remote_proc = *((ptl_process_t*) ompi_mtl_portals4_get_endpoint (mtl, ompi_proc));
83 }
84
85 MTL_PORTALS4_SET_RECV_BITS(match_bits, ignore_bits, comm->c_contextid,
86 src, tag);
87
88 me.start = NULL;
89 me.length = 0;
90 me.ct_handle = PTL_CT_NONE;
91 me.min_free = 0;
92 me.uid = ompi_mtl_portals4.uid;
93 me.options = PTL_ME_OP_PUT | PTL_ME_USE_ONCE;
94 me.match_id = remote_proc;
95 me.match_bits = match_bits;
96 me.ignore_bits = ignore_bits;
97
98 request.super.type = portals4_req_probe;
99 request.super.event_callback = completion_fn;
100 request.req_complete = 0;
101 request.found_match = 0;
102
103 opal_atomic_wmb();
104
105 ret = PtlMESearch(ompi_mtl_portals4.ni_h,
106 ompi_mtl_portals4.recv_idx,
107 &me,
108 PTL_SEARCH_ONLY,
109 &request);
110 if (OPAL_UNLIKELY(PTL_OK != ret)) {
111 opal_output_verbose(1, ompi_mtl_base_framework.framework_output,
112 "%s:%d: PtlMESearch failed: %d",
113 __FILE__, __LINE__, ret);
114 return ompi_mtl_portals4_get_error(ret);
115 }
116
117 while (0 == request.req_complete) {
118 opal_progress();
119 }
120
121 *flag = request.found_match;
122 if (1 == *flag) {
123 *status = request.status;
124 }
125
126 return OMPI_SUCCESS;
127 }
128
129
130 int
131 ompi_mtl_portals4_improbe(struct mca_mtl_base_module_t *mtl,
132 struct ompi_communicator_t *comm,
133 int src,
134 int tag,
135 int *matched,
136 struct ompi_message_t **message,
137 struct ompi_status_public_t *status)
138 {
139 struct ompi_mtl_portals4_probe_request_t request;
140 ptl_me_t me;
141 ptl_process_t remote_proc;
142 ptl_match_bits_t match_bits, ignore_bits;
143 int ret;
144
145 opal_output_verbose(1, ompi_mtl_base_framework.framework_output,
146 "%s:%d: improbe %d %d %d",
147 __FILE__, __LINE__, comm->c_contextid, src, tag);
148
149 if (MPI_ANY_SOURCE == src) {
150 if (ompi_mtl_portals4.use_logical) {
151 remote_proc.rank = PTL_RANK_ANY;
152 } else {
153 remote_proc.phys.nid = PTL_NID_ANY;
154 remote_proc.phys.pid = PTL_PID_ANY;
155 }
156 } else if ((ompi_mtl_portals4.use_logical) && (MPI_COMM_WORLD == comm)) {
157 remote_proc.rank = src;
158 } else {
159 ompi_proc_t* ompi_proc = ompi_comm_peer_lookup( comm, src );
160 remote_proc = *((ptl_process_t*) ompi_mtl_portals4_get_endpoint (mtl, ompi_proc));
161 }
162
163 MTL_PORTALS4_SET_RECV_BITS(match_bits, ignore_bits, comm->c_contextid,
164 src, tag);
165
166 me.start = NULL;
167 me.length = 0;
168 me.ct_handle = PTL_CT_NONE;
169 me.min_free = 0;
170 me.uid = ompi_mtl_portals4.uid;
171 me.options = PTL_ME_OP_PUT | PTL_ME_USE_ONCE;
172 me.match_id = remote_proc;
173 me.match_bits = match_bits;
174 me.ignore_bits = ignore_bits;
175
176 request.super.type = portals4_req_probe;
177 request.super.event_callback = completion_fn;
178 request.req_complete = 0;
179 request.found_match = 0;
180
181 opal_atomic_wmb();
182
183 ret = PtlMESearch(ompi_mtl_portals4.ni_h,
184 ompi_mtl_portals4.recv_idx,
185 &me,
186 PTL_SEARCH_DELETE,
187 &request);
188 if (OPAL_UNLIKELY(PTL_OK != ret)) {
189 opal_output_verbose(1, ompi_mtl_base_framework.framework_output,
190 "%s:%d: PtlMESearch failed: %d",
191 __FILE__, __LINE__, ret);
192 return ompi_mtl_portals4_get_error(ret);
193 }
194
195 while (0 == request.req_complete) {
196 opal_progress();
197 }
198
199 *matched = request.found_match;
200 if (1 == *matched) {
201 *status = request.status;
202
203 (*message) = ompi_message_alloc();
204 if (NULL == (*message)) {
205 return OMPI_ERR_OUT_OF_RESOURCE;
206 }
207
208 (*message)->comm = comm;
209 (*message)->req_ptr = request.message;
210 (*message)->peer = status->MPI_SOURCE;
211 (*message)->count = status->_ucount;
212
213 if (NULL == (*message)->req_ptr) {
214 ompi_message_return(*message);
215 *message = NULL;
216 return OMPI_ERR_OUT_OF_RESOURCE;
217 }
218 } else {
219 (*message) = MPI_MESSAGE_NULL;
220 }
221
222 return OMPI_SUCCESS;
223 }