_rndv_zcopy_config_t *zcopy)
{
    return /* is the current message greater than the minimal GET/PUT Zcopy? */
           (length >= zcopy->min) &&
           /* is the current message less than the maximal GET/PUT Zcopy? */
           ((length <= zcopy->max) ||
            /* or can the message be split? */ zcopy->split);
}

UCS_PROFILE_FUNC_VOID(ucp_rndv_receive, (worker, rreq, rndv_rts_hdr, rkey_buf),
                      ucp_worker_h worker, ucp_request_t *rreq,
                      const ucp_rndv_rts_hdr_t *rndv_rts_hdr,
                      const void *rkey_buf)
{
    ucp_rndv_mode_t rndv_mode;
    ucp_request_t *rndv_req;
    ucp_ep_h ep;
    ucp_ep_config_t *ep_config;
    ucs_status_t status;
    int is_get_zcopy_supported;
    int is_get_zcopy_failed;
    ucp_ep_rndv_zcopy_config_t *get_zcopy;
    ucp_ep_rndv_zcopy_config_t *put_zcopy;
    ucs_memory_type_t src_mem_type;

    UCS_ASYNC_BLOCK(&worker->async);

    UCS_PROFILE_REQUEST_EVENT(rreq, "rndv_receive", 0);

    /* if receiving a message on an already closed endpoint, stop processing */
    UCP_WORKER_GET_VALID_EP_BY_ID(&ep, worker, rndv_rts_hdr->sreq.ep_id,
                                  { status = UCS_ERR_CANCELED; goto err; },
                                  "RNDV rts");

    /* the internal send request allocated on receiver side (to perform a "get"
     * operation, send "ATS" and "RTR") */
    rndv_req = ucp_request_get(worker);
    if (rndv_req == NULL) {
        ucs_error("failed to allocate rendezvous reply");
        status = UCS_ERR_NO_MEMORY;
        goto err;
    }

    rndv_req->flags           = 0;
    rndv_req->send.ep         = ep;
    rndv_req->send.rndv.mdesc = NULL;
    is_get_zcopy_failed       = 0;
    src_mem_type              = UCS_MEMORY_TYPE_HOST;

    ucp_trace_req(rreq,
                  "rndv matched remote {address 0x%"PRIx64" size %zu sreq_id "
                  "0x%"PRIx64"} rndv_sreq %p", rndv_rts_hdr->address,
                  rndv_rts_hdr->size, rndv_rts_hdr->sreq.req_id, rndv_req);

    if (ucs_unlikely(rreq->recv.length < rndv_rts_hdr->size)) {
        ucp_trace_req(rndv_req,
                      "rndv truncated remote size %zu local size %zu rreq %p",
                      rndv_rts_hdr->size, rreq->recv.length, rreq);
        ucp_rndv_req_send_ack(rndv_req, rndv_rts_hdr->size,
                              rndv_rts_hdr->sreq.req_id, UCS_OK,
                              UCP_AM_ID_RNDV_ATS, "send_ats");
        ucp_request_recv_generic_dt_finish(rreq);
        ucp_rndv_zcopy_recv_req_complete(rreq, UCS_ERR_MESSAGE_TRUNCATED);
        goto out;
    }

    /* if the receive side is not connected yet then the RTS was received on a stub ep */
    ep_config = ucp_ep_config(ep);
    get_zcopy = &ep_config->rndv.get_zcopy;
    rndv_mode = worker->context->config.ext.rndv_mode;

    if (ucp_rndv_is_rkey_ptr(rndv_rts_hdr, rkey_buf, ep, rreq->recv.mem_type,
                             rndv_mode)) {
        ucp_rndv_do_rkey_ptr(rndv_req, rreq, rndv_rts_hdr, rkey_buf);
        goto out;
    }

    if (UCP_DT_IS_CONTIG(rreq->recv.datatype)) {
        is_get_zcopy_supported =
                (rndv_rts_hdr->address!= 0) &&
                ucp_rndv_test_zcopy_scheme_support(rndv_rts_hdr->size,
                                                   get_zcopy);
        if (is_get_zcopy_supported) {
            /* try to fetch the data with a get_zcopy operation */
            status = ucp_rndv_req_send_rma_get(rndv_req, rreq, rndv_rts_hdr,
                                               rkey_buf);
            if (status == UCS_OK) {
                goto out;
            }

            /* fallback to non get zcopy protocol */
            is_get_zcopy_failed = 1;
            src_mem_type        = ucp_rkey_packed_mem_type(rkey_buf);
        }

        /* Check if pipelined memtype staging is needed */
        if ((rndv_mode == UCP_RNDV_MODE_AUTO) &&
            UCP_MEM_IS_GPU(rreq->recv.mem_type) &&
            ucp_rndv_is_recv_pipeline_needed(rndv_req, rndv_rts_hdr, rkey_buf,
                                             rreq->recv.mem_type,
                                             is_get_zcopy_failed)) {
            put_zcopy = &ep_config->rndv.put_zcopy;
            ucp_rndv_recv_data_init(rreq, rndv_rts_hdr->size);
            if (ucp_rndv_is_put_pipeline_needed(rndv_rts_hdr->address,
                                                rndv_rts_hdr->size, rkey_buf,
                                                get_zcopy, put_zcopy,
                                                is_get_zcopy_failed)) {
                /* send FRAG RTR for sender to PUT the fragment. */
                ucp_rndv_send_frag_rtr(worker, rndv_req, rreq, rndv_rts_hdr);
                goto out;
            } else if (is_get_zcopy_supported) {
                ucs_assert(is_get_zcopy_failed);

                /* sender address is present. do GET pipeline */
                status = ucp_rndv_recv_start_get_pipeline(
                        worker, rndv_req, rreq, rndv_rts_hdr->sreq.req_id,
                        rkey_buf, rndv_rts_hdr->address, rndv_rts_hdr->size, 0);
                if (status == UCS_OK) {
                    goto out;
                }
            }
        }

        if (!is_get_zcopy_failed ||!UCP_MEM_IS_HOST(src_mem_type)) {
            if (rreq->flags & UCP_REQUEST_FLAG_USER_MEMH) {
                /* At this point we know the datatype is contig */
                ucp_request_init_dt_reg_from_memh(rreq,
                                                  ep_config->key.rma_bw_md_map,
                                                  rreq->recv.user_memh,
                                                  &rreq->recv.state.dt.contig);
            }

            /* register receive buffer for
             * put protocol (or) pipeline rndv for non-host memory type
             */
            ucs_assert(rndv_rts_hdr->size <= rreq->recv.length);
            ucp_