From 289595e45dc3ebfe5ae1a9dc6f347b1b2d569c4a Mon Sep 17 00:00:00 2001 From: Xin Zhao Date: Thu, 3 Jan 2019 18:20:30 -0800 Subject: [PATCH] OMPI/OSHMEM: bug-fix: store mkeys for each oshmem ctx. Signed-off-by: Xin Zhao Signed-off-by: Tomislav Janjusic --- oshmem/mca/atomic/ucx/atomic_ucx_cswap.c | 2 +- oshmem/mca/atomic/ucx/atomic_ucx_module.c | 4 +- oshmem/mca/memheap/base/base.h | 8 +- oshmem/mca/memheap/base/memheap_base_mkey.c | 21 ++-- oshmem/mca/spml/base/base.h | 5 +- oshmem/mca/spml/base/spml_base.c | 4 +- oshmem/mca/spml/spml.h | 4 +- oshmem/mca/spml/ucx/spml_ucx.c | 124 +++++++++++++------- oshmem/mca/spml/ucx/spml_ucx.h | 16 ++- oshmem/mca/spml/ucx/spml_ucx_component.c | 7 ++ oshmem/mca/sshmem/sshmem_types.h | 1 - oshmem/shmem/c/shmem_addr_accessible.c | 3 +- oshmem/shmem/c/shmem_ptr.c | 3 +- 13 files changed, 133 insertions(+), 69 deletions(-) diff --git a/oshmem/mca/atomic/ucx/atomic_ucx_cswap.c b/oshmem/mca/atomic/ucx/atomic_ucx_cswap.c index 25fe992688..51b0762947 100644 --- a/oshmem/mca/atomic/ucx/atomic_ucx_cswap.c +++ b/oshmem/mca/atomic/ucx/atomic_ucx_cswap.c @@ -40,7 +40,7 @@ int mca_atomic_ucx_cswap(shmem_ctx_t ctx, assert(NULL != prev); *prev = value; - ucx_mkey = mca_spml_ucx_get_mkey(ucx_ctx, pe, target, (void *)&rva, mca_spml_self); + ucx_mkey = mca_spml_ucx_get_mkey(ctx, pe, target, (void *)&rva, mca_spml_self); status_ptr = ucp_atomic_fetch_nb(ucx_ctx->ucp_peers[pe].ucp_conn, UCP_ATOMIC_FETCH_OP_CSWAP, cond, prev, size, rva, ucx_mkey->rkey, diff --git a/oshmem/mca/atomic/ucx/atomic_ucx_module.c b/oshmem/mca/atomic/ucx/atomic_ucx_module.c index 4d269065cb..91d4551e45 100644 --- a/oshmem/mca/atomic/ucx/atomic_ucx_module.c +++ b/oshmem/mca/atomic/ucx/atomic_ucx_module.c @@ -47,7 +47,7 @@ int mca_atomic_ucx_op(shmem_ctx_t ctx, assert((8 == size) || (4 == size)); - ucx_mkey = mca_spml_ucx_get_mkey(ucx_ctx, pe, target, (void *)&rva, mca_spml_self); + ucx_mkey = mca_spml_ucx_get_mkey(ctx, pe, target, (void *)&rva, mca_spml_self); status = ucp_atomic_post(ucx_ctx->ucp_peers[pe].ucp_conn, op, value, size, rva, ucx_mkey->rkey); @@ -70,7 +70,7 @@ int mca_atomic_ucx_fop(shmem_ctx_t ctx, assert((8 == size) || (4 == size)); - ucx_mkey = mca_spml_ucx_get_mkey(ucx_ctx, pe, target, (void *)&rva, mca_spml_self); + ucx_mkey = mca_spml_ucx_get_mkey(ctx, pe, target, (void *)&rva, mca_spml_self); status_ptr = ucp_atomic_fetch_nb(ucx_ctx->ucp_peers[pe].ucp_conn, op, value, prev, size, rva, ucx_mkey->rkey, diff --git a/oshmem/mca/memheap/base/base.h b/oshmem/mca/memheap/base/base.h index 6b4a79fb9e..7178685f0a 100644 --- a/oshmem/mca/memheap/base/base.h +++ b/oshmem/mca/memheap/base/base.h @@ -69,7 +69,8 @@ void memheap_oob_destruct(void); OSHMEM_DECLSPEC int mca_memheap_base_is_symmetric_addr(const void* va); OSHMEM_DECLSPEC sshmem_mkey_t *mca_memheap_base_get_mkey(void* va, int tr_id); -OSHMEM_DECLSPEC sshmem_mkey_t * mca_memheap_base_get_cached_mkey_slow(map_segment_t *s, +OSHMEM_DECLSPEC sshmem_mkey_t * mca_memheap_base_get_cached_mkey_slow(shmem_ctx_t ctx, + map_segment_t *s, int pe, void* va, int btl_id, @@ -243,7 +244,8 @@ static inline map_segment_t *memheap_find_va(void* va) return s; } -static inline sshmem_mkey_t *mca_memheap_base_get_cached_mkey(int pe, +static inline sshmem_mkey_t *mca_memheap_base_get_cached_mkey(shmem_ctx_t ctx, + int pe, void* va, int btl_id, void** rva) @@ -273,7 +275,7 @@ static inline sshmem_mkey_t *mca_memheap_base_get_cached_mkey(int pe, return mkey; } - return mca_memheap_base_get_cached_mkey_slow(s, pe, va, btl_id, rva); + return mca_memheap_base_get_cached_mkey_slow(ctx, s, pe, va, btl_id, rva); } static inline int mca_memheap_base_num_transports(void) diff --git a/oshmem/mca/memheap/base/memheap_base_mkey.c b/oshmem/mca/memheap/base/memheap_base_mkey.c index 8d92293210..fea00694ba 100644 --- a/oshmem/mca/memheap/base/memheap_base_mkey.c +++ b/oshmem/mca/memheap/base/memheap_base_mkey.c @@ -55,6 +55,7 @@ struct oob_comm { oob_comm_request_t req_pool[MEMHEAP_RECV_REQS_MAX]; opal_list_t req_list; int is_inited; + shmem_ctx_t ctx; }; mca_memheap_map_t* memheap_map = NULL; @@ -66,7 +67,7 @@ static int send_buffer(int pe, opal_buffer_t *msg); static int oshmem_mkey_recv_cb(void); /* pickup list of rkeys and remote va */ -static int memheap_oob_get_mkeys(int pe, +static int memheap_oob_get_mkeys(shmem_ctx_t ctx, int pe, uint32_t va_seg_num, sshmem_mkey_t *mkey); @@ -142,7 +143,7 @@ static void memheap_attach_segment(sshmem_mkey_t *mkey, int tr_id) } -static void unpack_remote_mkeys(opal_buffer_t *msg, int remote_pe) +static void unpack_remote_mkeys(shmem_ctx_t ctx, opal_buffer_t *msg, int remote_pe) { int32_t cnt; int32_t n; @@ -182,7 +183,7 @@ static void unpack_remote_mkeys(opal_buffer_t *msg, int remote_pe) } else { memheap_oob.mkeys[tr_id].u.key = MAP_SEGMENT_SHM_INVALID; } - MCA_SPML_CALL(rmkey_unpack(&memheap_oob.mkeys[tr_id], memheap_oob.segno, remote_pe, tr_id)); + MCA_SPML_CALL(rmkey_unpack(ctx, &memheap_oob.mkeys[tr_id], memheap_oob.segno, remote_pe, tr_id)); } MEMHEAP_VERBOSE(5, @@ -242,7 +243,7 @@ static void do_recv(int source_pe, opal_buffer_t* buffer) case MEMHEAP_RKEY_RESP: MEMHEAP_VERBOSE(5, "*** RKEY RESP"); OPAL_THREAD_LOCK(&memheap_oob.lck); - unpack_remote_mkeys(buffer, source_pe); + unpack_remote_mkeys(memheap_oob.ctx, buffer, source_pe); memheap_oob.mkeys_rcvd = MEMHEAP_RKEY_RESP; opal_condition_broadcast(&memheap_oob.cond); OPAL_THREAD_UNLOCK(&memheap_oob.lck); @@ -455,14 +456,14 @@ static int send_buffer(int pe, opal_buffer_t *msg) return rc; } -static int memheap_oob_get_mkeys(int pe, uint32_t seg, sshmem_mkey_t *mkeys) +static int memheap_oob_get_mkeys(shmem_ctx_t ctx, int pe, uint32_t seg, sshmem_mkey_t *mkeys) { opal_buffer_t *msg; uint8_t cmd; int i; int rc; - if (OSHMEM_SUCCESS == MCA_SPML_CALL(oob_get_mkeys(pe, seg, mkeys))) { + if (OSHMEM_SUCCESS == MCA_SPML_CALL(oob_get_mkeys(ctx, pe, seg, mkeys))) { for (i = 0; i < memheap_map->num_transports; i++) { MEMHEAP_VERBOSE(5, "MKEY CALCULATED BY LOCAL SPML: pe: %d tr_id: %d %s", @@ -478,6 +479,7 @@ static int memheap_oob_get_mkeys(int pe, uint32_t seg, sshmem_mkey_t *mkeys) memheap_oob.mkeys = mkeys; memheap_oob.segno = seg; memheap_oob.mkeys_rcvd = 0; + memheap_oob.ctx = ctx; msg = OBJ_NEW(opal_buffer_t); if (!msg) { @@ -645,7 +647,7 @@ void mca_memheap_modex_recv_all(void) } memheap_oob.mkeys = s->mkeys_cache[i]; memheap_oob.segno = j; - unpack_remote_mkeys(msg, i); + unpack_remote_mkeys(oshmem_ctx_default, msg, i); } } @@ -674,7 +676,8 @@ exit_fatal: } } -sshmem_mkey_t * mca_memheap_base_get_cached_mkey_slow(map_segment_t *s, +sshmem_mkey_t * mca_memheap_base_get_cached_mkey_slow(shmem_ctx_t ctx, + map_segment_t *s, int pe, void* va, int btl_id, @@ -692,7 +695,7 @@ sshmem_mkey_t * mca_memheap_base_get_cached_mkey_slow(map_segment_t *s, if (!s->mkeys_cache[pe]) return NULL ; - rc = memheap_oob_get_mkeys(pe, + rc = memheap_oob_get_mkeys(ctx, pe, s - memheap_map->mem_segs, s->mkeys_cache[pe]); if (OSHMEM_SUCCESS != rc) diff --git a/oshmem/mca/spml/base/base.h b/oshmem/mca/spml/base/base.h index 4aeff7d760..e3ec1b6855 100644 --- a/oshmem/mca/spml/base/base.h +++ b/oshmem/mca/spml/base/base.h @@ -72,11 +72,12 @@ OSHMEM_DECLSPEC int mca_spml_base_test(void* addr, void* value, int datatype, int *out_value); -OSHMEM_DECLSPEC int mca_spml_base_oob_get_mkeys(int pe, +OSHMEM_DECLSPEC int mca_spml_base_oob_get_mkeys(shmem_ctx_t ctx, + int pe, uint32_t seg, sshmem_mkey_t *mkeys); -OSHMEM_DECLSPEC void mca_spml_base_rmkey_unpack(sshmem_mkey_t *mkey, uint32_t seg, int pe, int tr_id); +OSHMEM_DECLSPEC void mca_spml_base_rmkey_unpack(shmem_ctx_t ctx, sshmem_mkey_t *mkey, uint32_t seg, int pe, int tr_id); OSHMEM_DECLSPEC void mca_spml_base_rmkey_free(sshmem_mkey_t *mkey); OSHMEM_DECLSPEC void *mca_spml_base_rmkey_ptr(const void *dst_addr, sshmem_mkey_t *mkey, int pe); diff --git a/oshmem/mca/spml/base/spml_base.c b/oshmem/mca/spml/base/spml_base.c index 4e3331ae8c..48a7c1c712 100644 --- a/oshmem/mca/spml/base/spml_base.c +++ b/oshmem/mca/spml/base/spml_base.c @@ -247,12 +247,12 @@ int mca_spml_base_wait_nb(void* handle) return OSHMEM_SUCCESS; } -int mca_spml_base_oob_get_mkeys(int pe, uint32_t segno, sshmem_mkey_t *mkeys) +int mca_spml_base_oob_get_mkeys(shmem_ctx_t ctx, int pe, uint32_t segno, sshmem_mkey_t *mkeys) { return OSHMEM_ERROR; } -void mca_spml_base_rmkey_unpack(sshmem_mkey_t *mkey, uint32_t segno, int pe, int tr_id) +void mca_spml_base_rmkey_unpack(shmem_ctx_t ctx, sshmem_mkey_t *mkey, uint32_t segno, int pe, int tr_id) { } diff --git a/oshmem/mca/spml/spml.h b/oshmem/mca/spml/spml.h index c78ed6cbdd..fa992db91c 100644 --- a/oshmem/mca/spml/spml.h +++ b/oshmem/mca/spml/spml.h @@ -132,7 +132,7 @@ typedef int (*mca_spml_base_module_test_fn_t)(void* addr, * * @param mkey remote mkey */ -typedef void (*mca_spml_base_module_mkey_unpack_fn_t)(sshmem_mkey_t *, uint32_t segno, int remote_pe, int tr_id); +typedef void (*mca_spml_base_module_mkey_unpack_fn_t)(shmem_ctx_t ctx, sshmem_mkey_t *, uint32_t segno, int remote_pe, int tr_id); /** * If possible, get a pointer to the remote memory described by the mkey @@ -180,7 +180,7 @@ typedef int (*mca_spml_base_module_deregister_fn_t)(sshmem_mkey_t *mkeys); * * @return OSHMEM_SUCCSESS if keys are found */ -typedef int (*mca_spml_base_module_oob_get_mkeys_fn_t)(int pe, +typedef int (*mca_spml_base_module_oob_get_mkeys_fn_t)(shmem_ctx_t ctx, int pe, uint32_t seg, sshmem_mkey_t *mkeys); diff --git a/oshmem/mca/spml/ucx/spml_ucx.c b/oshmem/mca/spml/ucx/spml_ucx.c index a760557e68..56cc9199aa 100644 --- a/oshmem/mca/spml/ucx/spml_ucx.c +++ b/oshmem/mca/spml/ucx/spml_ucx.c @@ -42,7 +42,7 @@ #endif static -spml_ucx_mkey_t * mca_spml_ucx_get_mkey_slow(int pe, void *va, void **rva); +spml_ucx_mkey_t * mca_spml_ucx_get_mkey_slow(shmem_ctx_t ctx, int pe, void *va, void **rva); mca_spml_ucx_t mca_spml_ucx = { .super = { @@ -103,7 +103,7 @@ int mca_spml_ucx_enable(bool enable) int mca_spml_ucx_del_procs(ompi_proc_t** procs, size_t nprocs) { opal_common_ucx_del_proc_t *del_procs; - size_t i; + size_t i, j; int ret; oshmem_shmem_barrier(); @@ -118,6 +118,12 @@ int mca_spml_ucx_del_procs(ompi_proc_t** procs, size_t nprocs) } for (i = 0; i < nprocs; ++i) { + for (j = 0; j < MCA_MEMHEAP_SEG_COUNT; j++) { + if (mca_spml_ucx_ctx_default.ucp_peers[i].mkeys[j].key.rkey != NULL) { + ucp_rkey_destroy(mca_spml_ucx_ctx_default.ucp_peers[i].mkeys[j].key.rkey); + } + } + del_procs[i].ep = mca_spml_ucx_ctx_default.ucp_peers[i].ucp_conn; del_procs[i].vpid = i; @@ -217,7 +223,7 @@ static char spml_ucx_transport_ids[1] = { 0 }; int mca_spml_ucx_add_procs(ompi_proc_t** procs, size_t nprocs) { - size_t i, n; + size_t i, j, n; int rc = OSHMEM_ERROR; int my_rank = oshmem_my_proc_id(); ucs_status_t err; @@ -270,6 +276,10 @@ int mca_spml_ucx_add_procs(ompi_proc_t** procs, size_t nprocs) OSHMEM_PROC_DATA(procs[i])->num_transports = 1; OSHMEM_PROC_DATA(procs[i])->transport_ids = spml_ucx_transport_ids; + for (j = 0; j < MCA_MEMHEAP_SEG_COUNT; j++) { + mca_spml_ucx_ctx_default.ucp_peers[i].mkeys[j].key.rkey = NULL; + } + mca_spml_ucx.remote_addrs_tbl[i] = (char *)malloc(wk_rsizes[i]); memcpy(mca_spml_ucx.remote_addrs_tbl[i], (char *)(wk_raddrs + wk_roffs[i]), wk_rsizes[i]); @@ -308,37 +318,47 @@ error: static -spml_ucx_mkey_t * mca_spml_ucx_get_mkey_slow(int pe, void *va, void **rva) +spml_ucx_mkey_t * mca_spml_ucx_get_mkey_slow(shmem_ctx_t ctx, int pe, void *va, void **rva) { sshmem_mkey_t *r_mkey; + spml_ucx_mkey_t *ucx_mkey; + uint32_t segno; + mca_spml_ucx_ctx_t *ucx_ctx = (mca_spml_ucx_ctx_t *)ctx; + ucs_status_t err; - r_mkey = mca_memheap_base_get_cached_mkey(pe, va, 0, rva); + r_mkey = mca_memheap_base_get_cached_mkey(ctx, pe, va, 0, rva); if (OPAL_UNLIKELY(!r_mkey)) { SPML_UCX_ERROR("pe=%d: %p is not address of symmetric variable", pe, va); oshmem_shmem_abort(-1); return NULL; } - return (spml_ucx_mkey_t *)(r_mkey->spml_context); + + segno = memheap_find_segnum(va); + ucx_mkey = &ucx_ctx->ucp_peers[pe].mkeys[segno].key; + + if (ucx_mkey->rkey == NULL) { + err = ucp_ep_rkey_unpack(ucx_ctx->ucp_peers[pe].ucp_conn, + r_mkey->u.data, + &ucx_mkey->rkey); + mca_spml_ucx_cache_mkey(ucx_ctx, r_mkey, segno, pe); /* make sure it is properly cached */ + } + + return ucx_mkey; } void mca_spml_ucx_rmkey_free(sshmem_mkey_t *mkey) { - spml_ucx_mkey_t *ucx_mkey; - - if (!mkey->spml_context) { - return; - } - ucx_mkey = (spml_ucx_mkey_t *)(mkey->spml_context); - ucp_rkey_destroy(ucx_mkey->rkey); } -void *mca_spml_ucx_rmkey_ptr(const void *dst_addr, sshmem_mkey_t *mkey, int pe) +void *mca_spml_ucx_rmkey_ptr(const void *dst_addr, sshmem_mkey_t *key, int pe) { #if (((UCP_API_MAJOR >= 1) && (UCP_API_MINOR >= 3)) || (UCP_API_MAJOR >= 2)) void *rva; ucs_status_t err; - spml_ucx_mkey_t *ucx_mkey = (spml_ucx_mkey_t *)(mkey->spml_context); + mca_spml_ucx_ctx_t *ucx_ctx = (mca_spml_ucx_ctx_t *)&mca_spml_ucx_ctx_default; + uint32_t segno = memheap_find_segnum((void*)dst_addr); + spml_ucx_mkey_t *ucx_mkey = &ucx_ctx->ucp_peers[pe].mkeys[segno].key; err = ucp_rkey_ptr(ucx_mkey->rkey, (uint64_t)dst_addr, &rva); if (UCS_OK != err) { @@ -350,31 +370,23 @@ void *mca_spml_ucx_rmkey_ptr(const void *dst_addr, sshmem_mkey_t *mkey, int pe) #endif } -static void mca_spml_ucx_cache_mkey(mca_spml_ucx_ctx_t *ucx_ctx, sshmem_mkey_t *mkey, uint32_t segno, int dst_pe) -{ - ucp_peer_t *peer; - - peer = &(ucx_ctx->ucp_peers[dst_pe]); - mkey_segment_init(&peer->mkeys[segno].super, mkey, segno); -} - -void mca_spml_ucx_rmkey_unpack(sshmem_mkey_t *mkey, uint32_t segno, int pe, int tr_id) +void mca_spml_ucx_rmkey_unpack(shmem_ctx_t ctx, sshmem_mkey_t *mkey, uint32_t segno, int pe, int tr_id) { spml_ucx_mkey_t *ucx_mkey; + mca_spml_ucx_ctx_t *ucx_ctx = (mca_spml_ucx_ctx_t *)ctx; ucs_status_t err; - ucx_mkey = &mca_spml_ucx_ctx_default.ucp_peers[pe].mkeys[segno].key; + ucx_mkey = &ucx_ctx->ucp_peers[pe].mkeys[segno].key; - err = ucp_ep_rkey_unpack(mca_spml_ucx_ctx_default.ucp_peers[pe].ucp_conn, - mkey->u.data, + err = ucp_ep_rkey_unpack(ucx_ctx->ucp_peers[pe].ucp_conn, + mkey->u.data, &ucx_mkey->rkey); if (UCS_OK != err) { SPML_UCX_ERROR("failed to unpack rkey: %s", ucs_status_string(err)); goto error_fatal; } - mkey->spml_context = ucx_mkey; - mca_spml_ucx_cache_mkey(&mca_spml_ucx_ctx_default, mkey, segno, pe); + mca_spml_ucx_cache_mkey(ucx_ctx, mkey, segno, pe); return; error_fatal: @@ -436,7 +448,6 @@ sshmem_mkey_t *mca_spml_ucx_register(void* addr, mem_seg = memheap_find_seg(segno); ucx_mkey = &mca_spml_ucx_ctx_default.ucp_peers[my_pe].mkeys[segno].key; - mkeys[0].spml_context = ucx_mkey; /* if possible use mem handle already created by ucx allocator */ if (MAP_SEGMENT_ALLOC_UCX != mem_seg->type) { @@ -499,16 +510,16 @@ int mca_spml_ucx_deregister(sshmem_mkey_t *mkeys) { spml_ucx_mkey_t *ucx_mkey; map_segment_t *mem_seg; + int segno; + int my_pe = oshmem_my_proc_id(); MCA_SPML_CALL(quiet(oshmem_ctx_default)); - if (!mkeys) - return OSHMEM_SUCCESS; - - if (!mkeys[0].spml_context) + if (!mkeys || !mkeys[0].va_base) return OSHMEM_SUCCESS; mem_seg = memheap_find_va(mkeys[0].va_base); - ucx_mkey = (spml_ucx_mkey_t*)mkeys[0].spml_context; + segno = memheap_find_segnum(mkeys[0].va_base); + ucx_mkey = &mca_spml_ucx_ctx_default.ucp_peers[my_pe].mkeys[segno].key; if (OPAL_UNLIKELY(NULL == mem_seg)) { return OSHMEM_ERROR; @@ -518,11 +529,14 @@ int mca_spml_ucx_deregister(sshmem_mkey_t *mkeys) ucp_mem_unmap(mca_spml_ucx.ucp_context, ucx_mkey->mem_h); } ucp_rkey_destroy(ucx_mkey->rkey); + ucx_mkey->rkey = NULL; if (0 < mkeys[0].len) { ucp_rkey_buffer_release(mkeys[0].u.data); } + free(mkeys); + return OSHMEM_SUCCESS; } @@ -531,8 +545,12 @@ int mca_spml_ucx_ctx_create(long options, shmem_ctx_t *ctx) mca_spml_ucx_ctx_list_item_t *ctx_item; ucp_worker_params_t params; ucp_ep_params_t ep_params; - size_t i, nprocs = oshmem_num_procs(); + size_t i, j, nprocs = oshmem_num_procs(); ucs_status_t err; + int my_pe = oshmem_my_proc_id(); + size_t len; + spml_ucx_mkey_t *ucx_mkey; + sshmem_mkey_t *mkey; int rc = OSHMEM_ERROR; ctx_item = OBJ_NEW(mca_spml_ucx_ctx_list_item_t); @@ -567,6 +585,24 @@ int mca_spml_ucx_ctx_create(long options, shmem_ctx_t *ctx) ucs_status_string(err)); goto error2; } + + for (j = 0; j < MCA_MEMHEAP_SEG_COUNT; j++) { + ctx_item->ctx.ucp_peers[i].mkeys[j].key.rkey = NULL; + } + } + + for (i = 0; i < MCA_MEMHEAP_SEG_COUNT; i++) { + mkey = &memheap_map->mem_segs[i].mkeys_cache[my_pe][0]; + ucx_mkey = &ctx_item->ctx.ucp_peers[my_pe].mkeys[i].key; + err = ucp_ep_rkey_unpack(ctx_item->ctx.ucp_peers[my_pe].ucp_conn, + mkey->u.data, + &ucx_mkey->rkey); + if (UCS_OK != err) { + SPML_UCX_ERROR("failed to unpack rkey"); + goto error2; + } + + mca_spml_ucx_cache_mkey(&ctx_item->ctx, mkey, i, my_pe); } SHMEM_MUTEX_LOCK(mca_spml_ucx.internal_mutex); @@ -600,7 +636,7 @@ int mca_spml_ucx_ctx_create(long options, shmem_ctx_t *ctx) void mca_spml_ucx_ctx_destroy(shmem_ctx_t ctx) { mca_spml_ucx_ctx_list_item_t *ctx_item, *next; - size_t i, nprocs = oshmem_num_procs(); + size_t i, j, nprocs = oshmem_num_procs(); MCA_SPML_CALL(quiet(ctx)); @@ -618,6 +654,12 @@ void mca_spml_ucx_ctx_destroy(shmem_ctx_t ctx) del_procs = malloc(sizeof(*del_procs) * nprocs); for (i = 0; i < nprocs; ++i) { + for (j = 0; j < MCA_MEMHEAP_SEG_COUNT; j++) { + if (ctx_item->ctx.ucp_peers[i].mkeys[j].key.rkey != NULL) { + ucp_rkey_destroy(ctx_item->ctx.ucp_peers[i].mkeys[j].key.rkey); + } + } + del_procs[i].ep = ctx_item->ctx.ucp_peers[i].ucp_conn; del_procs[i].vpid = i; ctx_item->ctx.ucp_peers[i].ucp_conn = NULL; @@ -649,7 +691,7 @@ int mca_spml_ucx_get(shmem_ctx_t ctx, void *src_addr, size_t size, void *dst_add ucs_status_t status; #endif - ucx_mkey = mca_spml_ucx_get_mkey(ucx_ctx, src, src_addr, &rva, &mca_spml_ucx); + ucx_mkey = mca_spml_ucx_get_mkey(ctx, src, src_addr, &rva, &mca_spml_ucx); #if HAVE_DECL_UCP_GET_NB request = ucp_get_nb(ucx_ctx->ucp_peers[src].ucp_conn, dst_addr, size, (uint64_t)rva, ucx_mkey->rkey, opal_common_ucx_empty_complete_cb); @@ -668,7 +710,7 @@ int mca_spml_ucx_get_nb(shmem_ctx_t ctx, void *src_addr, size_t size, void *dst_ spml_ucx_mkey_t *ucx_mkey; mca_spml_ucx_ctx_t *ucx_ctx = (mca_spml_ucx_ctx_t *)ctx; - ucx_mkey = mca_spml_ucx_get_mkey(ucx_ctx, src, src_addr, &rva, &mca_spml_ucx); + ucx_mkey = mca_spml_ucx_get_mkey(ctx, src, src_addr, &rva, &mca_spml_ucx); status = ucp_get_nbi(ucx_ctx->ucp_peers[src].ucp_conn, dst_addr, size, (uint64_t)rva, ucx_mkey->rkey); @@ -686,7 +728,7 @@ int mca_spml_ucx_put(shmem_ctx_t ctx, void* dst_addr, size_t size, void* src_add ucs_status_t status; #endif - ucx_mkey = mca_spml_ucx_get_mkey(ucx_ctx, dst, dst_addr, &rva, &mca_spml_ucx); + ucx_mkey = mca_spml_ucx_get_mkey(ctx, dst, dst_addr, &rva, &mca_spml_ucx); #if HAVE_DECL_UCP_PUT_NB request = ucp_put_nb(ucx_ctx->ucp_peers[dst].ucp_conn, src_addr, size, (uint64_t)rva, ucx_mkey->rkey, opal_common_ucx_empty_complete_cb); @@ -705,7 +747,7 @@ int mca_spml_ucx_put_nb(shmem_ctx_t ctx, void* dst_addr, size_t size, void* src_ spml_ucx_mkey_t *ucx_mkey; mca_spml_ucx_ctx_t *ucx_ctx = (mca_spml_ucx_ctx_t *)ctx; - ucx_mkey = mca_spml_ucx_get_mkey(ucx_ctx, dst, dst_addr, &rva, &mca_spml_ucx); + ucx_mkey = mca_spml_ucx_get_mkey(ctx, dst, dst_addr, &rva, &mca_spml_ucx); status = ucp_put_nbi(ucx_ctx->ucp_peers[dst].ucp_conn, src_addr, size, (uint64_t)rva, ucx_mkey->rkey); diff --git a/oshmem/mca/spml/ucx/spml_ucx.h b/oshmem/mca/spml/ucx/spml_ucx.h index 48fe66e1d0..7fff74a3c5 100644 --- a/oshmem/mca/spml/ucx/spml_ucx.h +++ b/oshmem/mca/spml/ucx/spml_ucx.h @@ -80,7 +80,7 @@ struct mca_spml_ucx_ctx_list_item { }; typedef struct mca_spml_ucx_ctx_list_item mca_spml_ucx_ctx_list_item_t; -typedef spml_ucx_mkey_t * (*mca_spml_ucx_get_mkey_slow_fn_t)(int pe, void *va, void **rva); +typedef spml_ucx_mkey_t * (*mca_spml_ucx_get_mkey_slow_fn_t)(shmem_ctx_t ctx, int pe, void *va, void **rva); struct mca_spml_ucx { mca_spml_base_module_t super; @@ -142,7 +142,7 @@ extern int mca_spml_ucx_deregister(sshmem_mkey_t *mkeys); extern void mca_spml_ucx_memuse_hook(void *addr, size_t length); -extern void mca_spml_ucx_rmkey_unpack(sshmem_mkey_t *mkey, uint32_t segno, int pe, int tr_id); +extern void mca_spml_ucx_rmkey_unpack(shmem_ctx_t ctx, sshmem_mkey_t *mkey, uint32_t segno, int pe, int tr_id); extern void mca_spml_ucx_rmkey_free(sshmem_mkey_t *mkey); extern void *mca_spml_ucx_rmkey_ptr(const void *dst_addr, sshmem_mkey_t *, int pe); @@ -152,17 +152,25 @@ extern int mca_spml_ucx_fence(shmem_ctx_t ctx); extern int mca_spml_ucx_quiet(shmem_ctx_t ctx); extern int spml_ucx_progress(void); +static void mca_spml_ucx_cache_mkey(mca_spml_ucx_ctx_t *ucx_ctx, sshmem_mkey_t *mkey, uint32_t segno, int dst_pe) +{ + ucp_peer_t *peer; + + peer = &(ucx_ctx->ucp_peers[dst_pe]); + mkey_segment_init(&peer->mkeys[segno].super, mkey, segno); +} static inline spml_ucx_mkey_t * -mca_spml_ucx_get_mkey(mca_spml_ucx_ctx_t *ucx_ctx, int pe, void *va, void **rva, mca_spml_ucx_t* module) +mca_spml_ucx_get_mkey(shmem_ctx_t ctx, int pe, void *va, void **rva, mca_spml_ucx_t* module) { spml_ucx_cached_mkey_t *mkey; + mca_spml_ucx_ctx_t *ucx_ctx = (mca_spml_ucx_ctx_t *)ctx; mkey = ucx_ctx->ucp_peers[pe].mkeys; mkey = (spml_ucx_cached_mkey_t *)map_segment_find_va(&mkey->super.super, sizeof(*mkey), va); if (OPAL_UNLIKELY(NULL == mkey)) { assert(module->get_mkey_slow); - return module->get_mkey_slow(pe, va, rva); + return module->get_mkey_slow(ctx, pe, va, rva); } *rva = map_segment_va2rva(&mkey->super, va); return &mkey->key; diff --git a/oshmem/mca/spml/ucx/spml_ucx_component.c b/oshmem/mca/spml/ucx/spml_ucx_component.c index c91edbfc12..835223ff23 100644 --- a/oshmem/mca/spml/ucx/spml_ucx_component.c +++ b/oshmem/mca/spml/ucx/spml_ucx_component.c @@ -111,7 +111,14 @@ static int mca_spml_ucx_component_register(void) int spml_ucx_progress(void) { + mca_spml_ucx_ctx_list_item_t *ctx_item, *next; ucp_worker_progress(mca_spml_ucx_ctx_default.ucp_worker); + SHMEM_MUTEX_LOCK(mca_spml_ucx.internal_mutex); + OPAL_LIST_FOREACH_SAFE(ctx_item, next, &(mca_spml_ucx.ctx_list), + mca_spml_ucx_ctx_list_item_t) { + ucp_worker_progress(ctx_item->ctx.ucp_worker); + } + SHMEM_MUTEX_UNLOCK(mca_spml_ucx.internal_mutex); return 1; } diff --git a/oshmem/mca/sshmem/sshmem_types.h b/oshmem/mca/sshmem/sshmem_types.h index ccdf8995b5..88520d3d20 100644 --- a/oshmem/mca/sshmem/sshmem_types.h +++ b/oshmem/mca/sshmem/sshmem_types.h @@ -94,7 +94,6 @@ typedef struct sshmem_mkey { void *data; uint64_t key; } u; - void *spml_context; /* spml module can attach internal structures here */ } sshmem_mkey_t; typedef struct map_base_segment { diff --git a/oshmem/shmem/c/shmem_addr_accessible.c b/oshmem/shmem/c/shmem_addr_accessible.c index 8d44ff4181..724318a894 100644 --- a/oshmem/shmem/c/shmem_addr_accessible.c +++ b/oshmem/shmem/c/shmem_addr_accessible.c @@ -31,7 +31,8 @@ int shmem_addr_accessible(const void *addr, int pe) RUNTIME_CHECK_INIT(); for (i = 0; i < mca_memheap_base_num_transports(); i++) { - mkey = mca_memheap_base_get_cached_mkey(pe, (void *)addr, i, &rva); + /* TODO: iterate on all ctxs, try to get cached mkey */ + mkey = mca_memheap_base_get_cached_mkey(oshmem_ctx_default, pe, (void *)addr, i, &rva); if (mkey) { return 1; } diff --git a/oshmem/shmem/c/shmem_ptr.c b/oshmem/shmem/c/shmem_ptr.c index c7c4d49507..7bfb6d014f 100644 --- a/oshmem/shmem/c/shmem_ptr.c +++ b/oshmem/shmem/c/shmem_ptr.c @@ -52,7 +52,8 @@ void *shmem_ptr(const void *dst_addr, int pe) } for (i = 0; i < mca_memheap_base_num_transports(); i++) { - mkey = mca_memheap_base_get_cached_mkey(pe, (void *)dst_addr, i, &rva); + /* TODO: iterate on all ctxs, try to get cached mkeys */ + mkey = mca_memheap_base_get_cached_mkey(oshmem_ctx_default, pe, (void *)dst_addr, i, &rva); if (!mkey) { continue; }