| 235 | } |
| 236 | |
| 237 | static ucs_status_t SendData(ucp_dt_iov_t* iov, size_t iovCnt, size_t* header, size_t headerSize, uint64_t amID, ucp_worker_h* dataWorker, ucp_ep_h* ep, bool needReply = false, ucs_memory_type_t memType = UCS_MEMORY_TYPE_HOST, bool forcedEager = false) { |
| 238 | const char* myName = "SendData"; |
| 239 | Request_t send_request_ctx = {.type = 0, .complete = 0}; |
| 240 | |
| 241 | tool::Logging(LOG_COMM, myName, "iovCnt=%zu, headerSize=%zu, amID=%lu\n", iovCnt, headerSize, amID); |
| 242 | |
| 243 | ucp_request_param_t param; |
| 244 | param.op_attr_mask = UCP_OP_ATTR_FIELD_CALLBACK | |
| 245 | UCP_OP_ATTR_FIELD_DATATYPE | |
| 246 | UCP_OP_ATTR_FIELD_USER_DATA; |
| 247 | if (needReply | forcedEager) { |
| 248 | param.op_attr_mask |= UCP_OP_ATTR_FIELD_FLAGS; |
| 249 | } |
| 250 | |
| 251 | // param.op_attr_mask |= UCP_OP_ATTR_FLAG_NO_IMM_CMPL; |
| 252 | // param.flags = UCP_AM_SEND_FLAG_RNDV; |
| 253 | param.datatype = (iovCnt == 1) ? ucp_dt_make_contig(1) : UCP_DATATYPE_IOV; |
| 254 | param.user_data = &send_request_ctx; |
| 255 | param.cb.send = (ucp_send_nbx_callback_t)SendRecvCommonCallBack; |
| 256 | param.memory_type = memType; |
| 257 | param.flags = UCP_AM_SEND_FLAG_REPLY; |
| 258 | if (forcedEager) { |
| 259 | param.flags |= UCP_AM_SEND_FLAG_EAGER; |
| 260 | } |
| 261 | |
| 262 | void* msg = (iovCnt == 1) ? iov[0].buffer : iov; |
| 263 | size_t msg_length = (iovCnt == 1) ? iov[0].length : iovCnt; |
| 264 | Request_t* send_request = (Request_t*)ucp_am_send_nbx( |
| 265 | *ep, amID, header, headerSize, |
| 266 | msg, msg_length, ¶m); |
| 267 | |
| 268 | tool::Logging(LOG_COMM, myName, "waiting for the send request to be completed.\n"); |
| 269 | ucs_status_t status = Wait(send_request, &send_request_ctx, dataWorker); |
| 270 | if (status != UCS_OK) { |
| 271 | tool::Logging(LOG_ERROR, myName, "failed to send data: %s\n", ucs_status_string(status)); |
| 272 | return status; |
| 273 | } |
| 274 | else { |
| 275 | tool::Logging(LOG_COMM, myName, "send request completed successfully.\n"); |
| 276 | } |
| 277 | if (send_request != NULL) { |
| 278 | ucp_request_free(send_request); |
| 279 | } |
| 280 | return status; |
| 281 | } |
| 282 | |
| 283 | static ucs_status_t ReceiveData(ucp_dt_iov_t* iov, size_t* iovCnt, void* dataBuffer, uint64_t amID, ucp_worker_h* dataWorker, ucs_memory_type_t memType, bool* isClosed) { |
| 284 | // iovCnt is set by ActiveMessageRecvCallback |
no test coverage detected