ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность

add ability to buffer put/accumulate messages during an epoch

Просмотр исходного кода 
This commit was SVN r15295.
Этот коммит содержится в:
Brian Barrett 2007-07-05 21:40:06 +00:00
родитель 5bbee1482e
Коммит 25e52238ab
7 изменённых файлов: 670 добавлений и 474 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

11
ompi/mca/osc/rdma/osc_rdma.h
Просмотреть файл

@ -31,6 +31,13 @@
BEGIN_C_DECLS BEGIN_C_DECLS
struct ompi_osc_rdma_buffer_t {
mca_btl_base_descriptor_t* descriptor;
size_t remain_len;
mca_bml_base_btl_t *bml_btl;
};
typedef struct ompi_osc_rdma_buffer_t ompi_osc_rdma_buffer_t;
struct ompi_osc_rdma_component_t { struct ompi_osc_rdma_component_t {
/** Extend the basic osc component interface */ /** Extend the basic osc component interface */
ompi_osc_base_component_t super; ompi_osc_base_component_t super;
@ -182,6 +189,10 @@ struct ompi_osc_rdma_module_t {
ompi_osc_rdma_peer_info_t *m_peer_info; ompi_osc_rdma_peer_info_t *m_peer_info;
int32_t m_rdma_num_pending; int32_t m_rdma_num_pending;
/*** buffering ***/
bool m_use_buffers;
ompi_osc_rdma_buffer_t *m_pending_buffers;
/* ********************* FENCE data ************************ */ /* ********************* FENCE data ************************ */
/* an array of <sizeof(m_comm)> ints, each containing the value /* an array of <sizeof(m_comm)> ints, each containing the value
1. */ 1. */

83
ompi/mca/osc/rdma/osc_rdma_component.c
Просмотреть файл

@ -144,6 +144,13 @@ component_open(void)
"Info key of same name overrides this value.", "Info key of same name overrides this value.",
false, false, 1, NULL); false, false, 1, NULL);
mca_base_param_reg_int(&mca_osc_rdma_component.super.osc_version,
"use_buffers",
"Coalesce messages during an epoch to reduce "
"network utilization. Info key of same name "
"overrides this value.",
false, false, 0, NULL);
mca_base_param_reg_int(&mca_osc_rdma_component.super.osc_version, mca_base_param_reg_int(&mca_osc_rdma_component.super.osc_version,
"use_rdma", "use_rdma",
"Use real RDMA operations to transfer data. " "Use real RDMA operations to transfer data. "
@ -355,6 +362,13 @@ ompi_osc_rdma_component_select(ompi_win_t *win,
module->m_setup_info = NULL; module->m_setup_info = NULL;
module->m_peer_info = NULL; module->m_peer_info = NULL;
/* buffer setup */
module->m_use_buffers = check_config_value_bool("use_buffers", info);
module->m_pending_buffers = malloc(sizeof(ompi_osc_rdma_buffer_t) *
ompi_comm_size(module->m_comm));
memset(module->m_pending_buffers, 0,
sizeof(ompi_osc_rdma_buffer_t) * ompi_comm_size(module->m_comm));
/* fence data */ /* fence data */
module->m_fence_coll_counts = (int*) module->m_fence_coll_counts = (int*)
malloc(sizeof(int) * ompi_comm_size(module->m_comm)); malloc(sizeof(int) * ompi_comm_size(module->m_comm));
@ -491,23 +505,22 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
int ret; int ret;
ompi_osc_rdma_module_t *module; ompi_osc_rdma_module_t *module;
void *payload; void *payload;
uint8_t hdr_type; bool done = false;
ompi_osc_rdma_base_header_t *base_header =
(ompi_osc_rdma_base_header_t*) descriptor->des_dst[0].seg_addr.pval;
assert(descriptor->des_dst[0].seg_len >= assert(descriptor->des_dst[0].seg_len >=
sizeof(ompi_osc_rdma_base_header_t)); sizeof(ompi_osc_rdma_base_header_t));
hdr_type = ((ompi_osc_rdma_base_header_t*)
descriptor->des_dst[0].seg_addr.pval)->hdr_type;
/* handle message */ /* handle message */
switch (hdr_type) { while (!done) {
switch (base_header->hdr_type) {
case OMPI_OSC_RDMA_HDR_PUT: case OMPI_OSC_RDMA_HDR_PUT:
{ {
ompi_osc_rdma_send_header_t *header; ompi_osc_rdma_send_header_t *header;
/* get our header and payload */ /* get our header and payload */
header = (ompi_osc_rdma_send_header_t*) header = (ompi_osc_rdma_send_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
payload = (void*) (header + 1); payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
@ -530,7 +543,7 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
} }
} }
ret = ompi_osc_rdma_sendreq_recv_put(module, header, payload); ret = ompi_osc_rdma_sendreq_recv_put(module, header, &payload);
} }
break; break;
@ -539,8 +552,7 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
ompi_osc_rdma_send_header_t *header; ompi_osc_rdma_send_header_t *header;
/* get our header and payload */ /* get our header and payload */
header = (ompi_osc_rdma_send_header_t*) header = (ompi_osc_rdma_send_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
payload = (void*) (header + 1); payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
@ -564,7 +576,7 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
} }
/* receive into temporary buffer */ /* receive into temporary buffer */
ret = ompi_osc_rdma_sendreq_recv_accum(module, header, payload); ret = ompi_osc_rdma_sendreq_recv_accum(module, header, &payload);
} }
break; break;
@ -576,8 +588,7 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
ompi_proc_t *proc; ompi_proc_t *proc;
/* get our header and payload */ /* get our header and payload */
header = (ompi_osc_rdma_send_header_t*) header = (ompi_osc_rdma_send_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
payload = (void*) (header + 1); payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
@ -633,8 +644,7 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
ompi_osc_rdma_sendreq_t *sendreq; ompi_osc_rdma_sendreq_t *sendreq;
/* get our header and payload */ /* get our header and payload */
header = (ompi_osc_rdma_reply_header_t*) header = (ompi_osc_rdma_reply_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
payload = (void*) (header + 1); payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
@ -648,15 +658,15 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
module = sendreq->req_module; module = sendreq->req_module;
/* receive data */ /* receive data */
ompi_osc_rdma_replyreq_recv(module, sendreq, header, payload); ompi_osc_rdma_replyreq_recv(module, sendreq, header, &payload);
} }
break; break;
case OMPI_OSC_RDMA_HDR_POST: case OMPI_OSC_RDMA_HDR_POST:
{ {
ompi_osc_rdma_control_header_t *header = ompi_osc_rdma_control_header_t *header =
(ompi_osc_rdma_control_header_t*) (ompi_osc_rdma_control_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
int32_t count; int32_t count;
payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -706,12 +716,13 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
} }
} }
break; break;
case OMPI_OSC_RDMA_HDR_COMPLETE: case OMPI_OSC_RDMA_HDR_COMPLETE:
{ {
ompi_osc_rdma_control_header_t *header = ompi_osc_rdma_control_header_t *header =
(ompi_osc_rdma_control_header_t*) (ompi_osc_rdma_control_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
int32_t count; int32_t count;
payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -737,9 +748,9 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
case OMPI_OSC_RDMA_HDR_LOCK_REQ: case OMPI_OSC_RDMA_HDR_LOCK_REQ:
{ {
ompi_osc_rdma_control_header_t *header = ompi_osc_rdma_control_header_t *header =
(ompi_osc_rdma_control_header_t*) (ompi_osc_rdma_control_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
int32_t count; int32_t count;
payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -767,8 +778,8 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
case OMPI_OSC_RDMA_HDR_UNLOCK_REQ: case OMPI_OSC_RDMA_HDR_UNLOCK_REQ:
{ {
ompi_osc_rdma_control_header_t *header = ompi_osc_rdma_control_header_t *header =
(ompi_osc_rdma_control_header_t*) (ompi_osc_rdma_control_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval; payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -788,9 +799,9 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
case OMPI_OSC_RDMA_HDR_UNLOCK_REPLY: case OMPI_OSC_RDMA_HDR_UNLOCK_REPLY:
{ {
ompi_osc_rdma_control_header_t *header = ompi_osc_rdma_control_header_t *header =
(ompi_osc_rdma_control_header_t*) (ompi_osc_rdma_control_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
int32_t count; int32_t count;
payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -812,9 +823,9 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
case OMPI_OSC_RDMA_HDR_RDMA_COMPLETE: case OMPI_OSC_RDMA_HDR_RDMA_COMPLETE:
{ {
ompi_osc_rdma_control_header_t *header = ompi_osc_rdma_control_header_t *header =
(ompi_osc_rdma_control_header_t*) (ompi_osc_rdma_control_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
int32_t count; int32_t count;
payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -836,13 +847,13 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
case OMPI_OSC_RDMA_HDR_RDMA_INFO: case OMPI_OSC_RDMA_HDR_RDMA_INFO:
{ {
ompi_osc_rdma_rdma_info_header_t *header = ompi_osc_rdma_rdma_info_header_t *header =
(ompi_osc_rdma_rdma_info_header_t*) (ompi_osc_rdma_rdma_info_header_t*) base_header;
descriptor->des_dst[0].seg_addr.pval;
ompi_proc_t *proc = NULL; ompi_proc_t *proc = NULL;
mca_bml_base_endpoint_t *endpoint = NULL; mca_bml_base_endpoint_t *endpoint = NULL;
mca_bml_base_btl_t *bml_btl; mca_bml_base_btl_t *bml_btl;
ompi_osc_rdma_btl_t *rdma_btl; ompi_osc_rdma_btl_t *rdma_btl;
int origin, index; int origin, index;
payload = (void*) (header + 1);
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -883,11 +894,23 @@ component_fragment_cb(struct mca_btl_base_module_t *btl,
} }
break; break;
case OMPI_OSC_RDMA_HDR_MULTI_END:
payload = base_header;
done = true;
break;
default: default:
/* BWB - FIX ME - this sucks */ /* BWB - FIX ME - this sucks */
opal_output(ompi_osc_base_output, opal_output(ompi_osc_base_output,
"received packet for Window with unknown type"); "received packet for Window with unknown type");
} }
if ((base_header->hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_MULTI) != 0) {
base_header = (ompi_osc_rdma_base_header_t*) payload;
} else {
done = true;
}
}
} }
int int

212
ompi/mca/osc/rdma/osc_rdma_data_move.c
Просмотреть файл

@ -75,7 +75,55 @@ inmsg_mark_complete(ompi_osc_rdma_module_t *module)
} }
} }
/**********************************************************************
*
* Multi-buffer support
*
**********************************************************************/
static int
send_multi_buffer(ompi_osc_rdma_module_t *module, int rank)
{
ompi_osc_rdma_base_header_t *header = (ompi_osc_rdma_base_header_t*)
((char*) module->m_pending_buffers[rank].descriptor->des_src[0].seg_addr.pval +
module->m_pending_buffers[rank].descriptor->des_src[0].seg_len);
header->hdr_type = OMPI_OSC_RDMA_HDR_MULTI_END;
header->hdr_flags = 0;
module->m_pending_buffers[rank].descriptor->des_src[0].seg_len +=
sizeof(ompi_osc_rdma_base_header_t);
mca_bml_base_send(module->m_pending_buffers[rank].bml_btl,
module->m_pending_buffers[rank].descriptor,
MCA_BTL_TAG_OSC_RDMA);
module->m_pending_buffers[rank].descriptor = NULL;
module->m_pending_buffers[rank].bml_btl = NULL;
module->m_pending_buffers[rank].remain_len = 0;
return OMPI_SUCCESS;
}
int
ompi_osc_rdma_flush(ompi_osc_rdma_module_t *module)
{
int i;
for (i = 0 ; i < ompi_comm_size(module->m_comm) ; ++i) {
if (module->m_pending_buffers[i].descriptor != NULL) {
send_multi_buffer(module, i);
}
}
return OMPI_SUCCESS;
}
/**********************************************************************
*
* RDMA data transfers (put / get)
*
**********************************************************************/
static void static void
rdma_cb(struct mca_btl_base_module_t* btl, rdma_cb(struct mca_btl_base_module_t* btl,
struct mca_btl_base_endpoint_t* endpoint, struct mca_btl_base_endpoint_t* endpoint,
@ -240,12 +288,12 @@ ompi_osc_rdma_sendreq_send_cb(struct mca_btl_base_module_t* btl,
struct mca_btl_base_descriptor_t* descriptor, struct mca_btl_base_descriptor_t* descriptor,
int status) int status)
{ {
ompi_osc_rdma_sendreq_t *sendreq =
(ompi_osc_rdma_sendreq_t*) descriptor->des_cbdata;
ompi_osc_rdma_send_header_t *header = ompi_osc_rdma_send_header_t *header =
(ompi_osc_rdma_send_header_t*) descriptor->des_src[0].seg_addr.pval; (ompi_osc_rdma_send_header_t*) descriptor->des_src[0].seg_addr.pval;
ompi_osc_rdma_module_t *module = sendreq->req_module; ompi_osc_rdma_sendreq_t *sendreq = NULL;
ompi_osc_rdma_module_t *module = NULL;
int32_t count; int32_t count;
bool done = false;
if (OMPI_SUCCESS != status) { if (OMPI_SUCCESS != status) {
/* requeue and return */ /* requeue and return */
@ -254,13 +302,22 @@ ompi_osc_rdma_sendreq_send_cb(struct mca_btl_base_module_t* btl,
return; return;
} }
/* have to look at header, and not the sendreq because in the case if (header->hdr_base.hdr_type == OMPI_OSC_RDMA_HDR_MULTI_END) {
of get, it's possible that the sendreq has been freed already done = true;
(if the remote side replies before we get our send completion }
callback) and already allocated to another request. We don't
wait for this completion before exiting a synchronization point while (!done) {
in the case of get, as we really don't care when it completes - sendreq = (ompi_osc_rdma_sendreq_t*) header->hdr_origin_sendreq.pval;
only when the data arrives. */ module = sendreq->req_module;
/* have to look at header, and not the sendreq because in the
case of get, it's possible that the sendreq has been freed
already (if the remote side replies before we get our send
completion callback) and already allocated to another
request. We don't wait for this completion before exiting
a synchronization point in the case of get, as we really
don't care when it completes - only when the data
arrives. */
if (OMPI_OSC_RDMA_HDR_GET != header->hdr_base.hdr_type) { if (OMPI_OSC_RDMA_HDR_GET != header->hdr_base.hdr_type) {
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT #if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) { if (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_NBO) {
@ -274,7 +331,9 @@ ompi_osc_rdma_sendreq_send_cb(struct mca_btl_base_module_t* btl,
count = sendreq->req_module->m_num_pending_out -= 1; count = sendreq->req_module->m_num_pending_out -= 1;
OPAL_THREAD_UNLOCK(&sendreq->req_module->m_lock); OPAL_THREAD_UNLOCK(&sendreq->req_module->m_lock);
ompi_osc_rdma_sendreq_free(sendreq); ompi_osc_rdma_sendreq_free(sendreq);
if (0 == count) opal_condition_broadcast(&sendreq->req_module->m_cond); if (0 == count) {
opal_condition_broadcast(&sendreq->req_module->m_cond);
}
} else { } else {
ompi_osc_rdma_longreq_t *longreq; ompi_osc_rdma_longreq_t *longreq;
ompi_osc_rdma_longreq_alloc(&longreq); ompi_osc_rdma_longreq_alloc(&longreq);
@ -302,6 +361,22 @@ ompi_osc_rdma_sendreq_send_cb(struct mca_btl_base_module_t* btl,
&(longreq->super.super)); &(longreq->super.super));
OPAL_THREAD_UNLOCK(&mca_osc_rdma_component.c_lock); OPAL_THREAD_UNLOCK(&mca_osc_rdma_component.c_lock);
} }
} else {
ompi_osc_rdma_sendreq_free(sendreq);
}
if (0 == (header->hdr_base.hdr_flags & OMPI_OSC_RDMA_HDR_FLAG_MULTI)) {
done = true;
} else {
header = (ompi_osc_rdma_send_header_t*)
(((char*) header) +
sizeof(ompi_osc_rdma_send_header_t) +
ompi_ddt_pack_description_length(sendreq->req_target_datatype) +
header->hdr_msg_length);
if (header->hdr_base.hdr_type == OMPI_OSC_RDMA_HDR_MULTI_END) {
done = true;
}
}
} }
/* release the descriptor and sendreq */ /* release the descriptor and sendreq */
@ -341,7 +416,7 @@ ompi_osc_rdma_sendreq_send(ompi_osc_rdma_module_t *module,
size_t written_data = 0; size_t written_data = 0;
size_t needed_len = sizeof(ompi_osc_rdma_send_header_t); size_t needed_len = sizeof(ompi_osc_rdma_send_header_t);
const void *packed_ddt; const void *packed_ddt;
size_t packed_ddt_len; size_t packed_ddt_len, remain;
if ((module->m_eager_send_active) && if ((module->m_eager_send_active) &&
(module->m_use_rdma) && (module->m_use_rdma) &&
@ -353,20 +428,34 @@ ompi_osc_rdma_sendreq_send(ompi_osc_rdma_module_t *module,
if (OPAL_LIKELY(OMPI_SUCCESS == ret)) return ret; if (OPAL_LIKELY(OMPI_SUCCESS == ret)) return ret;
} }
packed_ddt_len = ompi_ddt_pack_description_length(sendreq->req_target_datatype);
/* we always need to send the ddt */ /* we always need to send the ddt */
packed_ddt_len = ompi_ddt_pack_description_length(sendreq->req_target_datatype);
needed_len += packed_ddt_len; needed_len += packed_ddt_len;
if (OMPI_OSC_RDMA_GET != sendreq->req_type) { if (OMPI_OSC_RDMA_GET != sendreq->req_type) {
needed_len += sendreq->req_origin_bytes_packed; needed_len += sendreq->req_origin_bytes_packed;
} }
/* Get a BTL so we have the eager limit */ /* see if we already have a buffer */
if ((module->m_pending_buffers[sendreq->req_target_rank].remain_len >=
sizeof(ompi_osc_rdma_send_header_t) + sendreq->req_origin_bytes_packed) ||
(0 < module->m_pending_buffers[sendreq->req_target_rank].remain_len &&
sendreq->req_origin_bytes_packed > 2048)) {
bml_btl = module->m_pending_buffers[sendreq->req_target_rank].bml_btl;
descriptor = module->m_pending_buffers[sendreq->req_target_rank].descriptor;
remain = module->m_pending_buffers[sendreq->req_target_rank].remain_len;
} else {
/* send the existing buffer */
if (module->m_pending_buffers[sendreq->req_target_rank].descriptor) {
send_multi_buffer(module, sendreq->req_target_rank);
}
assert(OMPI_SUCCESS == ret);
/* get a buffer... */
endpoint = (mca_bml_base_endpoint_t*) sendreq->req_target_proc->proc_bml; endpoint = (mca_bml_base_endpoint_t*) sendreq->req_target_proc->proc_bml;
bml_btl = mca_bml_base_btl_array_get_next(&endpoint->btl_eager); bml_btl = mca_bml_base_btl_array_get_next(&endpoint->btl_eager);
descriptor = bml_btl->btl_alloc(bml_btl->btl, descriptor = bml_btl->btl_alloc(bml_btl->btl,
MCA_BTL_NO_ORDER, MCA_BTL_NO_ORDER,
needed_len < bml_btl->btl_eager_limit ? needed_len : module->m_use_buffers ? bml_btl->btl_eager_limit : needed_len < bml_btl->btl_eager_limit ? needed_len :
bml_btl->btl_eager_limit); bml_btl->btl_eager_limit);
if (NULL == descriptor) { if (NULL == descriptor) {
ret = OMPI_ERR_TEMP_OUT_OF_RESOURCE; ret = OMPI_ERR_TEMP_OUT_OF_RESOURCE;
@ -381,11 +470,18 @@ ompi_osc_rdma_sendreq_send(ompi_osc_rdma_module_t *module,
/* setup descriptor */ /* setup descriptor */
descriptor->des_cbfunc = ompi_osc_rdma_sendreq_send_cb; descriptor->des_cbfunc = ompi_osc_rdma_sendreq_send_cb;
descriptor->des_cbdata = (void*) sendreq;
descriptor->des_flags = MCA_BTL_DES_FLAGS_PRIORITY; descriptor->des_flags = MCA_BTL_DES_FLAGS_PRIORITY;
module->m_pending_buffers[sendreq->req_target_rank].bml_btl = bml_btl;
module->m_pending_buffers[sendreq->req_target_rank].descriptor = descriptor;
module->m_pending_buffers[sendreq->req_target_rank].remain_len = descriptor->des_src[0].seg_len - sizeof(ompi_osc_rdma_base_header_t);
remain = module->m_pending_buffers[sendreq->req_target_rank].remain_len;
descriptor->des_src[0].seg_len = 0;
}
/* pack header */ /* pack header */
header = (ompi_osc_rdma_send_header_t*) descriptor->des_src[0].seg_addr.pval; header = (ompi_osc_rdma_send_header_t*)
((char*) descriptor->des_src[0].seg_addr.pval + descriptor->des_src[0].seg_len);
written_data += sizeof(ompi_osc_rdma_send_header_t); written_data += sizeof(ompi_osc_rdma_send_header_t);
header->hdr_base.hdr_flags = 0; header->hdr_base.hdr_flags = 0;
header->hdr_windx = sendreq->req_module->m_comm->c_contextid; header->hdr_windx = sendreq->req_module->m_comm->c_contextid;
@ -413,26 +509,26 @@ ompi_osc_rdma_sendreq_send(ompi_osc_rdma_module_t *module,
#if OMPI_ENABLE_MEM_DEBUG #if OMPI_ENABLE_MEM_DEBUG
header->hdr_target_op = 0; header->hdr_target_op = 0;
#endif #endif
sendreq->req_refcount++;
break; break;
} }
/* Set datatype id and / or pack datatype */ /* Set datatype id and / or pack datatype */
ret = ompi_ddt_get_pack_description(sendreq->req_target_datatype, &packed_ddt); ret = ompi_ddt_get_pack_description(sendreq->req_target_datatype, &packed_ddt);
if (OMPI_SUCCESS != ret) goto cleanup; if (OMPI_SUCCESS != ret) goto cleanup;
memcpy((unsigned char*) descriptor->des_src[0].seg_addr.pval + written_data, memcpy((unsigned char*) descriptor->des_src[0].seg_addr.pval + descriptor->des_src[0].seg_len + written_data,
packed_ddt, packed_ddt_len); packed_ddt, packed_ddt_len);
written_data += packed_ddt_len; written_data += packed_ddt_len;
if (OMPI_OSC_RDMA_GET != sendreq->req_type) { if (OMPI_OSC_RDMA_GET != sendreq->req_type) {
/* if sending data and it fits, pack payload */ /* if sending data and it fits, pack payload */
if (descriptor->des_src[0].seg_len >= if (remain >= written_data + sendreq->req_origin_bytes_packed) {
written_data + sendreq->req_origin_bytes_packed) {
struct iovec iov; struct iovec iov;
uint32_t iov_count = 1; uint32_t iov_count = 1;
size_t max_data = sendreq->req_origin_bytes_packed; size_t max_data = sendreq->req_origin_bytes_packed;
iov.iov_len = max_data; iov.iov_len = max_data;
iov.iov_base = (IOVBASE_TYPE*)((unsigned char*) descriptor->des_src[0].seg_addr.pval + written_data); iov.iov_base = (IOVBASE_TYPE*)((unsigned char*) descriptor->des_src[0].seg_addr.pval + descriptor->des_src[0].seg_len + written_data);
ret = ompi_convertor_pack(&sendreq->req_origin_convertor, &iov, &iov_count, ret = ompi_convertor_pack(&sendreq->req_origin_convertor, &iov, &iov_count,
&max_data ); &max_data );
@ -441,20 +537,44 @@ ompi_osc_rdma_sendreq_send(ompi_osc_rdma_module_t *module,
goto cleanup; goto cleanup;
} }
assert(max_data == sendreq->req_origin_bytes_packed);
written_data += max_data; written_data += max_data;
descriptor->des_src[0].seg_len = written_data; descriptor->des_src[0].seg_len += written_data;
header->hdr_msg_length = sendreq->req_origin_bytes_packed; header->hdr_msg_length = sendreq->req_origin_bytes_packed;
} else { } else {
descriptor->des_src[0].seg_len += written_data;
header->hdr_msg_length = 0; header->hdr_msg_length = 0;
header->hdr_origin_tag = create_send_tag(module); header->hdr_origin_tag = create_send_tag(module);
} }
} else { } else {
descriptor->des_src[0].seg_len = written_data; descriptor->des_src[0].seg_len += written_data;
header->hdr_msg_length = 0; header->hdr_msg_length = 0;
} }
module->m_pending_buffers[sendreq->req_target_rank].remain_len -= written_data;
if (module->m_use_buffers) {
header->hdr_base.hdr_flags |= OMPI_OSC_RDMA_HDR_FLAG_MULTI;
#ifdef WORDS_BIGENDIAN
header->hdr_base.hdr_flags |= OMPI_OSC_RDMA_HDR_FLAG_NBO;
#elif OMPI_ENABLE_HETEROGENEOUS_SUPPORT
if (sendreq->req_target_proc->proc_arch & OMPI_ARCH_ISBIGENDIAN) {
header->hdr_base.hdr_flags |= OMPI_OSC_RDMA_HDR_FLAG_NBO;
OMPI_OSC_RDMA_SEND_HDR_HTON(*header);
}
#endif
if (module->m_pending_buffers[sendreq->req_target_rank].remain_len <
sizeof(ompi_osc_rdma_send_header_t) + 128) {
/* not enough space left - send now */
ret = send_multi_buffer(module, sendreq->req_target_rank);
} else {
ret = OMPI_SUCCESS;
}
goto done;
} else {
#ifdef WORDS_BIGENDIAN #ifdef WORDS_BIGENDIAN
header->hdr_base.hdr_flags |= OMPI_OSC_RDMA_HDR_FLAG_NBO; header->hdr_base.hdr_flags |= OMPI_OSC_RDMA_HDR_FLAG_NBO;
#elif OMPI_ENABLE_HETEROGENEOUS_SUPPORT #elif OMPI_ENABLE_HETEROGENEOUS_SUPPORT
@ -470,8 +590,13 @@ ompi_osc_rdma_sendreq_send(ompi_osc_rdma_module_t *module,
sendreq->req_module->m_comm->c_my_rank, sendreq->req_module->m_comm->c_my_rank,
sendreq->req_target_rank)); sendreq->req_target_rank));
module->m_pending_buffers[sendreq->req_target_rank].bml_btl = NULL;
module->m_pending_buffers[sendreq->req_target_rank].descriptor = NULL;
module->m_pending_buffers[sendreq->req_target_rank].remain_len = 0;
ret = mca_bml_base_send(bml_btl, descriptor, MCA_BTL_TAG_OSC_RDMA); ret = mca_bml_base_send(bml_btl, descriptor, MCA_BTL_TAG_OSC_RDMA);
goto done; goto done;
}
cleanup: cleanup:
if (descriptor != NULL) { if (descriptor != NULL) {
@ -660,6 +785,10 @@ ompi_osc_rdma_sendreq_recv_put_long_cb(ompi_osc_rdma_longreq_t *longreq)
OBJ_RELEASE(longreq->req_datatype); OBJ_RELEASE(longreq->req_datatype);
ompi_osc_rdma_longreq_free(longreq); ompi_osc_rdma_longreq_free(longreq);
OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_output,
"%d finished receiving long put message",
longreq->req_module->m_comm->c_my_rank));
inmsg_mark_complete(longreq->req_module); inmsg_mark_complete(longreq->req_module);
} }
@ -667,14 +796,14 @@ ompi_osc_rdma_sendreq_recv_put_long_cb(ompi_osc_rdma_longreq_t *longreq)
int int
ompi_osc_rdma_sendreq_recv_put(ompi_osc_rdma_module_t *module, ompi_osc_rdma_sendreq_recv_put(ompi_osc_rdma_module_t *module,
ompi_osc_rdma_send_header_t *header, ompi_osc_rdma_send_header_t *header,
void *inbuf) void **inbuf)
{ {
int ret = OMPI_SUCCESS; int ret = OMPI_SUCCESS;
void *target = (unsigned char*) module->m_win->w_baseptr + void *target = (unsigned char*) module->m_win->w_baseptr +
(header->hdr_target_disp * module->m_win->w_disp_unit); (header->hdr_target_disp * module->m_win->w_disp_unit);
ompi_proc_t *proc = ompi_comm_peer_lookup( module->m_comm, header->hdr_origin ); ompi_proc_t *proc = ompi_comm_peer_lookup( module->m_comm, header->hdr_origin );
struct ompi_datatype_t *datatype = struct ompi_datatype_t *datatype =
ompi_osc_rdma_datatype_create(proc, &inbuf); ompi_osc_rdma_datatype_create(proc, inbuf);
if (NULL == datatype) { if (NULL == datatype) {
opal_output(ompi_osc_base_output, opal_output(ompi_osc_base_output,
@ -701,7 +830,7 @@ ompi_osc_rdma_sendreq_recv_put(ompi_osc_rdma_module_t *module,
0, 0,
&convertor); &convertor);
iov.iov_len = header->hdr_msg_length; iov.iov_len = header->hdr_msg_length;
iov.iov_base = (IOVBASE_TYPE*)inbuf; iov.iov_base = (IOVBASE_TYPE*)*inbuf;
max_data = iov.iov_len; max_data = iov.iov_len;
ompi_convertor_unpack(&convertor, ompi_convertor_unpack(&convertor,
&iov, &iov,
@ -710,6 +839,13 @@ ompi_osc_rdma_sendreq_recv_put(ompi_osc_rdma_module_t *module,
OBJ_DESTRUCT(&convertor); OBJ_DESTRUCT(&convertor);
OBJ_RELEASE(datatype); OBJ_RELEASE(datatype);
inmsg_mark_complete(module); inmsg_mark_complete(module);
*inbuf = ((char*) *inbuf) + header->hdr_msg_length;
OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_output,
"%d received put message from %d",
module->m_comm->c_my_rank,
header->hdr_origin));
} else { } else {
ompi_osc_rdma_longreq_t *longreq; ompi_osc_rdma_longreq_t *longreq;
ompi_osc_rdma_longreq_alloc(&longreq); ompi_osc_rdma_longreq_alloc(&longreq);
@ -727,6 +863,12 @@ ompi_osc_rdma_sendreq_recv_put(ompi_osc_rdma_module_t *module,
module->m_comm, module->m_comm,
&(longreq->request)); &(longreq->request));
OPAL_OUTPUT_VERBOSE((50, ompi_osc_base_output,
"%d started long recv put message from %d (%d)",
module->m_comm->c_my_rank,
header->hdr_origin,
header->hdr_origin_tag));
/* put the send request in the waiting list */ /* put the send request in the waiting list */
OPAL_THREAD_LOCK(&mca_osc_rdma_component.c_lock); OPAL_THREAD_LOCK(&mca_osc_rdma_component.c_lock);
opal_list_append(&mca_osc_rdma_component.c_pending_requests, opal_list_append(&mca_osc_rdma_component.c_pending_requests,
@ -788,13 +930,13 @@ ompi_osc_rdma_sendreq_recv_accum_long_cb(ompi_osc_rdma_longreq_t *longreq)
int int
ompi_osc_rdma_sendreq_recv_accum(ompi_osc_rdma_module_t *module, ompi_osc_rdma_sendreq_recv_accum(ompi_osc_rdma_module_t *module,
ompi_osc_rdma_send_header_t *header, ompi_osc_rdma_send_header_t *header,
void *payload) void **payload)
{ {
int ret = OMPI_SUCCESS; int ret = OMPI_SUCCESS;
struct ompi_op_t *op = ompi_osc_rdma_op_create(header->hdr_target_op); struct ompi_op_t *op = ompi_osc_rdma_op_create(header->hdr_target_op);
ompi_proc_t *proc = ompi_comm_peer_lookup( module->m_comm, header->hdr_origin ); ompi_proc_t *proc = ompi_comm_peer_lookup( module->m_comm, header->hdr_origin );
struct ompi_datatype_t *datatype = struct ompi_datatype_t *datatype =
ompi_osc_rdma_datatype_create(proc, &payload); ompi_osc_rdma_datatype_create(proc, payload);
if (NULL == datatype) { if (NULL == datatype) {
opal_output(ompi_osc_base_output, opal_output(ompi_osc_base_output,
@ -807,7 +949,7 @@ ompi_osc_rdma_sendreq_recv_accum(ompi_osc_rdma_module_t *module,
OPAL_THREAD_LOCK(&module->m_acc_lock); OPAL_THREAD_LOCK(&module->m_acc_lock);
/* copy the data from the temporary buffer into the user window */ /* copy the data from the temporary buffer into the user window */
ret = ompi_osc_rdma_process_op(module, header, datatype, op, payload, ret = ompi_osc_rdma_process_op(module, header, datatype, op, *payload,
header->hdr_msg_length); header->hdr_msg_length);
/* unlock the window for accumulates */ /* unlock the window for accumulates */
@ -823,6 +965,8 @@ ompi_osc_rdma_sendreq_recv_accum(ompi_osc_rdma_module_t *module,
"%d received accum message from %d", "%d received accum message from %d",
module->m_comm->c_my_rank, module->m_comm->c_my_rank,
header->hdr_origin)); header->hdr_origin));
*payload = ((char*) *payload) + header->hdr_msg_length;
} else { } else {
ompi_osc_rdma_longreq_t *longreq; ompi_osc_rdma_longreq_t *longreq;
ptrdiff_t lb, extent, true_lb, true_extent; ptrdiff_t lb, extent, true_lb, true_extent;
@ -902,7 +1046,7 @@ int
ompi_osc_rdma_replyreq_recv(ompi_osc_rdma_module_t *module, ompi_osc_rdma_replyreq_recv(ompi_osc_rdma_module_t *module,
ompi_osc_rdma_sendreq_t *sendreq, ompi_osc_rdma_sendreq_t *sendreq,
ompi_osc_rdma_reply_header_t *header, ompi_osc_rdma_reply_header_t *header,
void *payload) void **payload)
{ {
int ret = OMPI_SUCCESS; int ret = OMPI_SUCCESS;
@ -916,7 +1060,7 @@ ompi_osc_rdma_replyreq_recv(ompi_osc_rdma_module_t *module,
int32_t count; int32_t count;
iov.iov_len = header->hdr_msg_length; iov.iov_len = header->hdr_msg_length;
iov.iov_base = (IOVBASE_TYPE*)payload; iov.iov_base = (IOVBASE_TYPE*)*payload;
max_data = iov.iov_len; max_data = iov.iov_len;
ompi_convertor_unpack(&sendreq->req_origin_convertor, ompi_convertor_unpack(&sendreq->req_origin_convertor,
&iov, &iov,
@ -925,6 +1069,8 @@ ompi_osc_rdma_replyreq_recv(ompi_osc_rdma_module_t *module,
count = sendreq->req_module->m_num_pending_out -= 1; count = sendreq->req_module->m_num_pending_out -= 1;
ompi_osc_rdma_sendreq_free(sendreq); ompi_osc_rdma_sendreq_free(sendreq);
*payload = ((char*) *payload) + header->hdr_msg_length;
if (0 == count) opal_condition_broadcast(&sendreq->req_module->m_cond); if (0 == count) opal_condition_broadcast(&sendreq->req_module->m_cond);
} else { } else {
ompi_osc_rdma_longreq_t *longreq; ompi_osc_rdma_longreq_t *longreq;

11
ompi/mca/osc/rdma/osc_rdma_data_move.h
Просмотреть файл

@ -33,27 +33,30 @@ int ompi_osc_rdma_replyreq_send(ompi_osc_rdma_module_t *module,
/* receive the target side of a sendreq for a put, directly into the user's window */ /* receive the target side of a sendreq for a put, directly into the user's window */
int ompi_osc_rdma_sendreq_recv_put(ompi_osc_rdma_module_t *module, int ompi_osc_rdma_sendreq_recv_put(ompi_osc_rdma_module_t *module,
ompi_osc_rdma_send_header_t *header, ompi_osc_rdma_send_header_t *header,
void *payload); void **payload);
/* receive the target side of a sendreq for an accumulate, possibly /* receive the target side of a sendreq for an accumulate, possibly
using a temproart buffer, then calling the reduction functions */ using a temproart buffer, then calling the reduction functions */
int ompi_osc_rdma_sendreq_recv_accum(ompi_osc_rdma_module_t *module, int ompi_osc_rdma_sendreq_recv_accum(ompi_osc_rdma_module_t *module,
ompi_osc_rdma_send_header_t *header, ompi_osc_rdma_send_header_t *header,
void *payload); void **payload);
/* receive the origin side of a replyreq (the reply part of an /* receive the origin side of a replyreq (the reply part of an
MPI_Get), directly into the user's window */ MPI_Get), directly into the user's window */
int ompi_osc_rdma_replyreq_recv(ompi_osc_rdma_module_t *module, int ompi_osc_rdma_replyreq_recv(ompi_osc_rdma_module_t *module,
ompi_osc_rdma_sendreq_t *sendreq, ompi_osc_rdma_sendreq_t *sendreq,
ompi_osc_rdma_reply_header_t *header, ompi_osc_rdma_reply_header_t *header,
void *payload); void **payload);
int ompi_osc_rdma_control_send(ompi_osc_rdma_module_t *module, int ompi_osc_rdma_control_send(ompi_osc_rdma_module_t *module,
ompi_proc_t *proc, ompi_proc_t *proc,
uint8_t type, int32_t value0, int32_t value1); uint8_t type,
int32_t value0, int32_t value1);
int ompi_osc_rdma_rdma_ack_send(ompi_osc_rdma_module_t *module, int ompi_osc_rdma_rdma_ack_send(ompi_osc_rdma_module_t *module,
ompi_proc_t *proc, ompi_proc_t *proc,
ompi_osc_rdma_btl_t *rdma_btl); ompi_osc_rdma_btl_t *rdma_btl);
int ompi_osc_rdma_flush(ompi_osc_rdma_module_t *module);
#endif #endif

6
ompi/mca/osc/rdma/osc_rdma_header.h
Просмотреть файл

@ -25,7 +25,7 @@
#include "opal/types.h" #include "opal/types.h"
/* Note -- 0x05 to 0x0A are of control_hdr type */ /* Note -- 0x05 to 0x0C are of control_hdr type */
#define OMPI_OSC_RDMA_HDR_PUT 0x01 #define OMPI_OSC_RDMA_HDR_PUT 0x01
#define OMPI_OSC_RDMA_HDR_ACC 0x02 #define OMPI_OSC_RDMA_HDR_ACC 0x02
#define OMPI_OSC_RDMA_HDR_GET 0x03 #define OMPI_OSC_RDMA_HDR_GET 0x03
@ -36,9 +36,11 @@
#define OMPI_OSC_RDMA_HDR_UNLOCK_REQ 0x08 #define OMPI_OSC_RDMA_HDR_UNLOCK_REQ 0x08
#define OMPI_OSC_RDMA_HDR_UNLOCK_REPLY 0x09 #define OMPI_OSC_RDMA_HDR_UNLOCK_REPLY 0x09
#define OMPI_OSC_RDMA_HDR_RDMA_COMPLETE 0x0A #define OMPI_OSC_RDMA_HDR_RDMA_COMPLETE 0x0A
#define OMPI_OSC_RDMA_HDR_RDMA_INFO 0x0B #define OMPI_OSC_RDMA_HDR_MULTI_END 0x0B
#define OMPI_OSC_RDMA_HDR_RDMA_INFO 0x0C
#define OMPI_OSC_RDMA_HDR_FLAG_NBO 0x01 #define OMPI_OSC_RDMA_HDR_FLAG_NBO 0x01
#define OMPI_OSC_RDMA_HDR_FLAG_MULTI 0x02
struct ompi_osc_rdma_base_header_t { struct ompi_osc_rdma_base_header_t {
uint8_t hdr_type; uint8_t hdr_type;

5
ompi/mca/osc/rdma/osc_rdma_sendreq.h
Просмотреть файл

@ -41,6 +41,8 @@ typedef enum {
struct ompi_osc_rdma_sendreq_t { struct ompi_osc_rdma_sendreq_t {
ompi_request_t super; ompi_request_t super;
int req_refcount;
/** type of sendreq (from ompi_osc_rdma_req_type_t) */ /** type of sendreq (from ompi_osc_rdma_req_type_t) */
ompi_osc_rdma_req_type_t req_type; ompi_osc_rdma_req_type_t req_type;
/** pointer to the module that created the sendreq */ /** pointer to the module that created the sendreq */
@ -106,6 +108,7 @@ ompi_osc_rdma_sendreq_alloc(ompi_osc_rdma_module_t *module,
(*sendreq)->req_module = module; (*sendreq)->req_module = module;
(*sendreq)->req_target_rank = target_rank; (*sendreq)->req_target_rank = target_rank;
(*sendreq)->req_target_proc = proc; (*sendreq)->req_target_proc = proc;
(*sendreq)->req_refcount = 1;
return OMPI_SUCCESS; return OMPI_SUCCESS;
} }
@ -165,6 +168,7 @@ ompi_osc_rdma_sendreq_init_target(ompi_osc_rdma_sendreq_t *sendreq,
static inline int static inline int
ompi_osc_rdma_sendreq_free(ompi_osc_rdma_sendreq_t *sendreq) ompi_osc_rdma_sendreq_free(ompi_osc_rdma_sendreq_t *sendreq)
{ {
if (0 == (--sendreq->req_refcount)) {
ompi_convertor_cleanup(&sendreq->req_origin_convertor); ompi_convertor_cleanup(&sendreq->req_origin_convertor);
OBJ_RELEASE(sendreq->req_target_datatype); OBJ_RELEASE(sendreq->req_target_datatype);
@ -172,6 +176,7 @@ ompi_osc_rdma_sendreq_free(ompi_osc_rdma_sendreq_t *sendreq)
OPAL_FREE_LIST_RETURN(&mca_osc_rdma_component.c_sendreqs, OPAL_FREE_LIST_RETURN(&mca_osc_rdma_component.c_sendreqs,
(opal_list_item_t*) sendreq); (opal_list_item_t*) sendreq);
}
return OMPI_SUCCESS; return OMPI_SUCCESS;
} }

6
ompi/mca/osc/rdma/osc_rdma_sync.c
Просмотреть файл

@ -150,6 +150,8 @@ ompi_osc_rdma_module_fence(int assert, ompi_win_t *win)
} }
} }
ompi_osc_rdma_flush(module);
OPAL_THREAD_LOCK(&module->m_lock); OPAL_THREAD_LOCK(&module->m_lock);
/* if some requests couldn't be started, push into the /* if some requests couldn't be started, push into the
"queued" list, where we will try to restart them later. */ "queued" list, where we will try to restart them later. */
@ -338,6 +340,8 @@ ompi_osc_rdma_module_complete(ompi_win_t *win)
} }
} }
ompi_osc_rdma_flush(module);
OPAL_THREAD_LOCK(&module->m_lock); OPAL_THREAD_LOCK(&module->m_lock);
/* if some requests couldn't be started, push into the /* if some requests couldn't be started, push into the
"queued" list, where we will try to restart them later. */ "queued" list, where we will try to restart them later. */
@ -558,6 +562,8 @@ ompi_osc_rdma_module_unlock(int target,
} }
} }
ompi_osc_rdma_flush(module);
OPAL_THREAD_LOCK(&module->m_lock); OPAL_THREAD_LOCK(&module->m_lock);
/* if some requests couldn't be started, push into the /* if some requests couldn't be started, push into the
"queued" list, where we will try to restart them later. */ "queued" list, where we will try to restart them later. */