arrow::Status Connection::SendAMIov()

in dissociated-ipc/ucx_conn.cc [273:310]

• 30 lines of code
• 7 McCabe index (conditional complexity)

arrow::Status Connection::SendAMIov(unsigned int id, const ucp_dt_iov_t* iov,
                                    const size_t iov_cnt, void* user_data,
                                    ucp_send_nbx_callback_t cb,
                                    const ucs_memory_type_t memory_type) {
  ARROW_RETURN_NOT_OK(CheckClosed());

  ucp_request_param_t request_param;
  request_param.op_attr_mask = UCP_OP_ATTR_FIELD_FLAGS | UCP_OP_ATTR_FIELD_DATATYPE |
                               UCP_OP_ATTR_FIELD_MEMORY_TYPE;
  request_param.flags = UCP_AM_SEND_FLAG_REPLY;
  request_param.datatype = UCP_DATATYPE_IOV;
  if (cb) {
    request_param.cb.send = cb;
    request_param.op_attr_mask |= UCP_OP_ATTR_FIELD_CALLBACK;
  }
  if (user_data) {
    request_param.user_data = user_data;
    request_param.op_attr_mask |= UCP_OP_ATTR_FIELD_USER_DATA;
  }
  request_param.memory_type = memory_type;

  void* request =
      ucp_am_send_nbx(remote_endpoint_, id, nullptr, 0, iov, iov_cnt, &request_param);
  if (!request) {
    // request completed immediately, call the cb manually if it exists
    // since it won't be called automatically
    if (cb) cb(request, UCS_OK, user_data);
  } else if (UCS_PTR_IS_ERR(request)) {
    // same thing, call it manually
    auto status = UCS_PTR_STATUS(request);
    if (cb) cb(request, status, user_data);
    return utils::FromUcsStatus("ucp_am_send_nbx", status);
  }

  // otherwise the callback will be called eventually when it completes
  // we can just return success.
  return arrow::Status::OK();
}