static inline
int mca_atomic_basic_fop(void *target,
                         void *prev,
                         uint64_t value,
                         size_t size,
                         int pe,
                         struct oshmem_op_t *op)
{
    int rc = OSHMEM_SUCCESS;
    long long temp_value = 0;

    atomic_basic_lock(pe);

    rc = MCA_SPML_CALL(get(target, size, (void*)&temp_value, pe));

    memcpy(prev, (void*) &temp_value, size);

    op->o_func.c_fn((void*) value,
                    (void*) &temp_value,
                    size / op->dt_size);

    if (rc == OSHMEM_SUCCESS) {
        rc = MCA_SPML_CALL(put(target, size, (void*)&temp_value, pe));
        shmem_quiet();
    }

    atomic_basic_unlock(pe);

    return rc;
}

static int do_mkey_req(opal_buffer_t *msg, int pe, int seg)
{
    uint8_t msg_type;
    oshmem_proc_t *proc;
    int i, n, tr_id;
    mca_spml_mkey_t *mkey;

    msg_type = MEMHEAP_RKEY_RESP;
    opal_dss.pack(msg, &msg_type, 1, OPAL_UINT8);

    /* go over all transports to remote pe and pack mkeys */
    n = oshmem_get_transport_count(pe);
    proc = oshmem_proc_group_find(oshmem_group_all, pe);
    opal_dss.pack(msg, &n, 1, OPAL_UINT32);
    MEMHEAP_VERBOSE(5, "found %d transports to %d", n, pe);
    for (i = 0; i < n; i++) {
        tr_id = proc->transport_ids[i];

        mkey = mca_memheap_base_get_mkey(__seg2base_va(seg), tr_id);
        if (!mkey) {
            MEMHEAP_ERROR("seg#%d tr_id: %d failed to find local mkey",
                          seg, tr_id);
            return OSHMEM_ERROR;
        }
        opal_dss.pack(msg, &tr_id, 1, OPAL_UINT32);
        opal_dss.pack(msg, &mkey->key, 1, OPAL_UINT64);
        opal_dss.pack(msg, &mkey->va_base, 1, OPAL_UINT64);

        if (NULL != MCA_SPML_CALL(get_remote_context_size)) {
            uint32_t context_size =
                    (mkey->spml_context == NULL ) ?
                            0 :
                            (uint32_t) MCA_SPML_CALL(get_remote_context_size(mkey->spml_context));
            opal_dss.pack(msg, &context_size, 1, OPAL_UINT32);
            if (0 != context_size) {
                opal_dss.pack(msg,
                              MCA_SPML_CALL(get_remote_context(mkey->spml_context)),
                              context_size,
                              OPAL_BYTE);
            }
        }

        MEMHEAP_VERBOSE(5,
                        "seg#%d tr_id: %d key %llx base_va %p",
                        seg, tr_id, (unsigned long long)mkey->key, mkey->va_base);
    }
    return OSHMEM_SUCCESS;
}

/**
 * note: we have to reg memory directly with btl because no proc will have a full btl list in proc_bml
 */
int mca_spml_yoda_deregister(sshmem_mkey_t *mkeys)
{
    int i;
    struct yoda_btl *ybtl;
    mca_spml_yoda_context_t* yoda_context;

    MCA_SPML_CALL(fence());
    mca_spml_yoda_wait_gets();

    if (!mkeys) {
        return OSHMEM_SUCCESS;
    }

    for (i = 0; i < mca_spml_yoda.n_btls; i++) {
        ybtl = &mca_spml_yoda.btl_type_map[i];
        yoda_context = (mca_spml_yoda_context_t*) mkeys[i].spml_context;
        if (NULL == yoda_context) {
            continue;
        }
        if (yoda_context->btl_src_descriptor) {
            ybtl->btl->btl_free(ybtl->btl, yoda_context->btl_src_descriptor);
            yoda_context->btl_src_descriptor = NULL;
        }
        if (yoda_context->registration) {
            ybtl->btl->btl_deregister_mem (ybtl->btl, yoda_context->registration);
        }

    }
    free(mkeys);

    return OSHMEM_SUCCESS;
}

void shmem_get128_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *len, MPI_Fint *pe)
{
    MCA_SPML_CALL(get(oshmem_ctx_default, FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*len) * 16,
        FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*pe)));
}

void shmem_put128_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *length, MPI_Fint *pe)
{
    MCA_SPML_CALL(put(FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*length) * 16,
        FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*pe)));
}

int mca_spml_ikrit_deregister(sshmem_mkey_t *mkeys)
{
    int i;

    MCA_SPML_CALL(fence(oshmem_ctx_default));
    if (!mkeys)
        return OSHMEM_SUCCESS;

    for (i = 0; i < MXM_PTL_LAST; i++) {
        switch (i) {
        case MXM_PTL_SHM:
            break;
        case MXM_PTL_RDMA:
            /* dereg memory */
            if (!mkeys[i].spml_context)
                break;
            mxm_mem_unmap(mca_spml_ikrit.mxm_context,
                    (void *)mkeys[i].va_base,
                    (unsigned long)mkeys[i].spml_context,
                    0);
            if (0 < mkeys[i].len) {
                free(mkeys[i].u.data);
            }
            break;
        }
    }
    free(mkeys);

    return OSHMEM_SUCCESS;

}

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)
        return OSHMEM_SUCCESS;

    mem_seg  = memheap_find_va(mkeys[0].va_base);
    ucx_mkey = (spml_ucx_mkey_t*)mkeys[0].spml_context;

    if (OPAL_UNLIKELY(NULL == mem_seg)) {
        return OSHMEM_ERROR;
    }
    
    if (MAP_SEGMENT_ALLOC_UCX != mem_seg->type) {
        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;
}

void shmem_wait_until_f(ompi_fortran_integer_t *var, MPI_Fint *cmp, ompi_fortran_integer_t *value)
{
    MCA_SPML_CALL(wait((void*)var, 
        OMPI_FINT_2_INT(*cmp), 
        (void*)value, 
        SHMEM_FINT));
}

static int __dereg_segment(map_segment_t *s)
{
    int rc = OSHMEM_SUCCESS;
    int j;
    int nprocs, my_pe;

    nprocs = oshmem_num_procs();
    my_pe = oshmem_my_proc_id();

    MCA_SPML_CALL(deregister(s->mkeys));

    if (s->mkeys_cache) {
        for (j = 0; j < nprocs; j++) {
            if (j == my_pe)
                continue;
            if (s->mkeys_cache[j]) {
                free(s->mkeys_cache[j]);
                s->mkeys_cache[j] = NULL;
            }
        }
        free(s->mkeys_cache);
        s->mkeys_cache = NULL;
    }

    s->is_active = 0;

    return rc;
}

void shmem_get64_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *len, MPI_Fint *pe)
{
    MCA_SPML_CALL(get(FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*len) * 8,
        FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*pe)));
}

void shmem_putmem_nbi_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *length, MPI_Fint *pe)
{
    MCA_SPML_CALL(put_nb(FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*length),
        FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*pe), NULL));
}

static int memheap_oob_get_mkeys(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))) {
        for (i = 0; i < memheap_map->num_transports; i++) {
            mkeys[i].va_base = mca_memheap_seg2base_va(seg);
            MEMHEAP_VERBOSE(5,
                            "MKEY CALCULATED BY LOCAL SPML: pe: %d tr_id: %d %s",
                            pe,
                            i,
                            mca_spml_base_mkey2str(&mkeys[i]));
        }
        return OSHMEM_SUCCESS;
    }

    OPAL_THREAD_LOCK(&memheap_oob.lck);

    memheap_oob.mkeys = mkeys;
    memheap_oob.mkeys_rcvd = 0;

    msg = OBJ_NEW(opal_buffer_t);
    if (!msg) {
        OPAL_THREAD_UNLOCK(&memheap_oob.lck);
        MEMHEAP_ERROR("failed to get msg buffer");
        return OSHMEM_ERROR;
    }

    OPAL_THREAD_LOCK(&memheap_oob.lck);
    cmd = MEMHEAP_RKEY_REQ;
    opal_dss.pack(msg, &cmd, 1, OPAL_UINT8);
    opal_dss.pack(msg, &seg, 1, OPAL_UINT32);

    rc = send_buffer(pe, msg);
    if (MPI_SUCCESS != rc) {
        OPAL_THREAD_UNLOCK(&memheap_oob.lck);
        MEMHEAP_ERROR("FAILED to send rml message %d", rc);
        return OSHMEM_ERROR;
    }

    while (!memheap_oob.mkeys_rcvd) {
        opal_condition_wait(&memheap_oob.cond, &memheap_oob.lck);
    }

    if (MEMHEAP_RKEY_RESP == memheap_oob.mkeys_rcvd) {
        rc = OSHMEM_SUCCESS;
    } else {
        MEMHEAP_ERROR("failed to get rkey seg#%d pe=%d", seg, pe);
        rc = OSHMEM_ERROR;
    }

    OPAL_THREAD_UNLOCK(&memheap_oob.lck);
    return rc;
}

void shmem_integer_get_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *len, MPI_Fint *pe)
{
    size_t integer_type_size = 0;
    ompi_datatype_type_size(&ompi_mpi_integer.dt, &integer_type_size);

    MCA_SPML_CALL(get(oshmem_ctx_default, FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*len) * integer_type_size,
        FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*pe)));
}

void shmem_character_get_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *len, MPI_Fint *pe)
{
    size_t character_type_size = 0;
    ompi_datatype_type_size(&ompi_mpi_character.dt, &character_type_size);

    MCA_SPML_CALL(get(FPTR_2_VOID_PTR(source), 
        OMPI_FINT_2_INT(*len) * character_type_size, 
        FPTR_2_VOID_PTR(target), 
        OMPI_FINT_2_INT(*pe)));
}

void shmem_complex_put_nbi_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *length, MPI_Fint *pe)
{
    size_t complex_type_size = 0;
    ompi_datatype_type_size(&ompi_mpi_cplex.dt, &complex_type_size);

    MCA_SPML_CALL(put_nb(FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*length) * complex_type_size,
        FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*pe), NULL));
}

void shmem_logical_put_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *length, MPI_Fint *pe)
{
    size_t logical_type_size = 0;
    ompi_datatype_type_size(&ompi_mpi_logical.dt, &logical_type_size);

    MCA_SPML_CALL(put(oshmem_ctx_default, FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*length) * logical_type_size,
        FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*pe)));
}

void shmem_double_put_nbi_f(FORTRAN_POINTER_T target, FORTRAN_POINTER_T source, MPI_Fint *length, MPI_Fint *pe)
{
    size_t double_precision_type_size = 0;
    ompi_datatype_type_size(&ompi_mpi_dblprec.dt, &double_precision_type_size);

    MCA_SPML_CALL(put_nb(FPTR_2_VOID_PTR(target),
        OMPI_FINT_2_INT(*length) * double_precision_type_size,
        FPTR_2_VOID_PTR(source),
        OMPI_FINT_2_INT(*pe), NULL));
}

void atomic_basic_lock(int pe)
{
    int index = -1;
    int me = oshmem_my_proc_id();
    int num_pe = oshmem_num_procs();
    char lock_required = ATOMIC_LOCK_WAITING;
    char lock_active = ATOMIC_LOCK_ACTIVE;
    int root_pe = pe;

    do {
        /* announce that we need the resource */
        do {
            MCA_SPML_CALL(put((void*)(atomic_lock_sync + me), sizeof(lock_required), (void*)&lock_required, root_pe));
            MCA_SPML_CALL(get((void*)atomic_lock_sync, num_pe * sizeof(*atomic_lock_sync), (void*)local_lock_sync, root_pe));
        } while (local_lock_sync[me] != lock_required);

        MCA_SPML_CALL(get((void*)atomic_lock_turn, sizeof(index), (void*)&index, root_pe));
        while (index != me) {
            if (local_lock_sync[index] != ATOMIC_LOCK_IDLE) {
                MCA_SPML_CALL(get((void*)atomic_lock_turn, sizeof(index), (void*)&index, root_pe));
                MCA_SPML_CALL(get((void*)atomic_lock_sync, num_pe * sizeof(*atomic_lock_sync), (void*)local_lock_sync, root_pe));
            } else {
                index = (index + 1) % num_pe;
            }
        }

        /* now tentatively claim the resource */
        do {
            MCA_SPML_CALL(put((void*)(atomic_lock_sync + me), sizeof(lock_active), (void*)&lock_active, root_pe));
            MCA_SPML_CALL(get((void*)atomic_lock_sync, num_pe * sizeof(*atomic_lock_sync), (void*)local_lock_sync, root_pe));
        } while (local_lock_sync[me] != lock_active);

        index = 0;
        while ((index < num_pe)
                && ((index == me)
                        || (local_lock_sync[index] != ATOMIC_LOCK_ACTIVE))) {
            index = index + 1;
        }

        MCA_SPML_CALL(get((void*)atomic_lock_turn, sizeof(*atomic_lock_turn), (void*)local_lock_turn, root_pe));
    } while (!((index >= num_pe)
            && ((*local_lock_turn == me)
                    || (local_lock_sync[*local_lock_turn] == ATOMIC_LOCK_IDLE))));

    MCA_SPML_CALL(put((void*)atomic_lock_turn, sizeof(me), (void*)&me, root_pe));
}

void mca_spml_ucx_ctx_destroy(shmem_ctx_t ctx)
{
    MCA_SPML_CALL(quiet(ctx));

    SHMEM_MUTEX_LOCK(mca_spml_ucx.internal_mutex);
    _ctx_remove(&mca_spml_ucx.active_array, (mca_spml_ucx_ctx_t *)ctx);
    _ctx_add(&mca_spml_ucx.idle_array, (mca_spml_ucx_ctx_t *)ctx);
    SHMEM_MUTEX_UNLOCK(mca_spml_ucx.internal_mutex);

    if (!mca_spml_ucx.active_array.ctxs_count) {
        opal_progress_unregister(spml_ucx_ctx_progress);
    }
}

void atomic_basic_unlock(int pe)
{
    int index = -1;
    int me = oshmem_my_proc_id();
    int num_pe = oshmem_num_procs();
    char lock_idle = ATOMIC_LOCK_IDLE;
    int root_pe = pe;

    MCA_SPML_CALL(get((void*)atomic_lock_sync, num_pe * sizeof(*atomic_lock_sync), (void*)local_lock_sync, root_pe));
    MCA_SPML_CALL(get((void*)atomic_lock_turn, sizeof(index), (void*)&index, root_pe));

    do {
        index = (index + 1) % num_pe;
    } while (local_lock_sync[index] == ATOMIC_LOCK_IDLE);

    MCA_SPML_CALL(put((void*)atomic_lock_turn, sizeof(index), (void*)&index, root_pe));

    do {
        MCA_SPML_CALL(put((void*)(atomic_lock_sync + me), sizeof(lock_idle), (void*)&lock_idle, root_pe));
        MCA_SPML_CALL(get((void*)atomic_lock_sync, num_pe * sizeof(*atomic_lock_sync), (void*)local_lock_sync, root_pe));
    } while (local_lock_sync[me] != lock_idle);
}