mpid_nem_lmt_shm.c

fortran并行计算包

                if (copy_buf->receiver_present.val)
                {
                    SCHED_YIELD();
                    i = 0;
                }
                else
                {
                    req->dev.segment_first = first;
                    vc_ch->lmt_buf_num = buf_num;
                    *done = FALSE;
                    goto fn_exit;
                }
            }

            ++i;
        }

        MPID_NEM_READ_WRITE_BARRIER();


        /* we have a free buffer, fill it */
        if (data_sz <= PIPELINE_THRESHOLD)
            copy_limit = PIPELINE_MAX_SIZE;
        else
            copy_limit = MPID_NEM_COPY_BUF_LEN;
        last = (data_sz - first <= copy_limit) ? data_sz : first + copy_limit;
	MPID_Segment_pack(req->dev.segment_ptr, first, &last, (void *)copy_buf->buf[buf_num]); /* cast away volatile */
        MPID_NEM_WRITE_BARRIER();
        copy_buf->len[buf_num].val = last - first;

        first = last;
        buf_num = (buf_num+1) % NUM_BUFS;
    }
    while (last < data_sz);

    *done = TRUE;
    MPIDI_CH3U_Request_complete(req);

 fn_exit:
    copy_buf->sender_present.val = FALSE;
    MPIDI_FUNC_EXIT(MPID_STATE_LMT_SHM_SEND_PROGRESS);
    return mpi_errno;
}

/* Continued from note for lmt_shm_send_progress() above.  The
   receiver uses segment_unpack(), and just like segment_pack() it may
   not copy all the data you ask it to.  This means that we may leave
   some data in the buffer uncopied.  To handle this, we copy the
   remaining data to just before the next buffer.  Then when the next
   buffer is filled, we start copying where we just copied the
   leftover data from last time.  */

#undef FUNCNAME
#define FUNCNAME lmt_shm_recv_progress
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int lmt_shm_recv_progress(MPIDI_VC_t *vc, MPID_Request *req, int *done)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_CH3I_VC *vc_ch = (MPIDI_CH3I_VC *)vc->channel_private;
    volatile MPID_nem_copy_buf_t * const copy_buf = vc_ch->lmt_copy_buf;
    MPIDI_msg_sz_t first;
    MPIDI_msg_sz_t last, expected_last;
    int buf_num;
    MPIDI_msg_sz_t data_sz, len;
    int i;
    MPIDI_msg_sz_t surfeit;
    char *src_buf;
    char tmpbuf[MPID_NEM_CACHE_LINE_LEN];
    MPIDI_STATE_DECL(MPID_STATE_LMT_SHM_RECV_PROGRESS);

    MPIDI_FUNC_ENTER(MPID_STATE_LMT_SHM_RECV_PROGRESS);

    MPIU_Assert((vc_ch->lmt_copy_buf->owner_info.val.rank == MPIDI_Process.my_pg_rank &&
                 vc_ch->lmt_copy_buf->owner_info.val.remote_req_id == vc_ch->lmt_active_lmt->req->ch.lmt_req_id) ||
                (vc_ch->lmt_copy_buf->owner_info.val.rank == vc->pg_rank &&
                 vc_ch->lmt_copy_buf->owner_info.val.remote_req_id == vc_ch->lmt_active_lmt->req->handle));

    copy_buf->receiver_present.val = TRUE;

    surfeit = vc_ch->lmt_surfeit;
    data_sz = req->ch.lmt_data_sz;
    buf_num = vc_ch->lmt_buf_num;
    first = req->dev.segment_first;

    do
    {
        /* If the buffer is empty, wait.  If the sender is actively
           working on this transfer, yield the processor and keep
           waiting, otherwise wait for a bounded amount of time. */
        i = 0;
        while ((len = copy_buf->len[buf_num].val) == 0)
        {
            if (i == NUM_BUSY_POLLS)
            {
                if (copy_buf->sender_present.val)
                {
                    SCHED_YIELD();
                    i = 0;
                }
                else
                {
                    req->dev.segment_first = first;
                    vc_ch->lmt_buf_num = buf_num;
                    vc_ch->lmt_surfeit = surfeit;
                    *done = FALSE;
                    goto fn_exit;
                }
            }

            ++i;
        }

        MPID_NEM_READ_BARRIER();

        /* unpack data including any leftover from the previous buffer */
        src_buf = ((char *)copy_buf->buf[buf_num]) - surfeit; /* cast away volatile */
        last = expected_last = (data_sz - first <= surfeit + len) ? data_sz : first + surfeit + len;

	MPID_Segment_unpack(req->dev.segment_ptr, first, &last, src_buf);

        if (surfeit && buf_num > 0)
        {
            /* we had leftover data from the previous buffer, we can
               now mark that buffer as empty */

            MPID_NEM_READ_WRITE_BARRIER();
            copy_buf->len[(buf_num-1)].val = 0;
            /* Make sure we copied at least the leftover data from last time */
            MPIU_Assert(last - first > surfeit);
       }

        if (last < expected_last)
        {
            /* we have leftover data in the buffer that we couldn't copy out */
            char *surfeit_ptr;

            surfeit_ptr = (char *)src_buf + last - first;
            surfeit = expected_last - last;

            if (buf_num == NUM_BUFS-1)
            {
                /* if we're wrapping back to buf 0, then we can copy it directly */
                memcpy(((char *)copy_buf->buf[0]) - surfeit, surfeit_ptr, surfeit);

                MPID_NEM_READ_WRITE_BARRIER();
                copy_buf->len[buf_num].val = 0;
            }
            else
            {
                /* otherwise, we need to copy to a tmpbuf first to make sure the src and dest addresses don't overlap */
                memcpy(tmpbuf, surfeit_ptr, surfeit);
                memcpy(((char *)copy_buf->buf[buf_num+1]) - surfeit, tmpbuf, surfeit);
            }
        }
        else
        {
            /* all data was unpacked, we can mark this buffer as empty */
            surfeit = 0;

            MPID_NEM_READ_WRITE_BARRIER();
            copy_buf->len[buf_num].val = 0;	
        }

        first = last;
        buf_num = (buf_num+1) % NUM_BUFS;
    }
    while (last < data_sz);

    for (i = 0; i < NUM_BUFS; ++i)
        copy_buf->len[i].val = 0;

    copy_buf->owner_info.val.remote_req_id = MPI_REQUEST_NULL;
    MPID_NEM_WRITE_BARRIER();
    copy_buf->owner_info.val.rank          = NO_OWNER;

    *done = TRUE;
    MPIDI_CH3U_Request_complete(req);

 fn_exit:
    copy_buf->receiver_present.val = FALSE;
    MPIDI_FUNC_EXIT(MPID_STATE_LMT_SHM_RECV_PROGRESS);
    return mpi_errno;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_lmt_shm_handle_cookie
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPID_nem_lmt_shm_handle_cookie(MPIDI_VC_t *vc, MPID_Request *req, MPID_IOV cookie)
{
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_LMT_SHM_HANDLE_COOKIE);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_LMT_SHM_HANDLE_COOKIE);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_LMT_SHM_HANDLE_COOKIE);
    return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_lmt_shm_done_send
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPID_nem_lmt_shm_done_send(MPIDI_VC_t *vc, MPID_Request *req)
{
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_LMT_SHM_DONE_SEND);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_LMT_SHM_DONE_SEND);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_LMT_SHM_DONE_SEND);
    return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_lmt_shm_done_recv
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPID_nem_lmt_shm_done_recv(MPIDI_VC_t *vc, MPID_Request *req)
{
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_LMT_SHM_DONE_RECV);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_LMT_SHM_DONE_RECV);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_LMT_SHM_DONE_RECV);
    return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_lmt_shm_progress_vc
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static inline int lmt_shm_progress_vc(MPIDI_VC_t *vc, int *done)
{
    int mpi_errno = MPI_SUCCESS;
    int done_req = FALSE;
    MPID_nem_lmt_shm_wait_element_t *we;
    MPIDI_CH3I_VC *vc_ch = (MPIDI_CH3I_VC *)vc->channel_private;
    MPIDI_STATE_DECL(MPID_STATE_LMT_SHM_PROGRESS_VC);

    MPIDI_FUNC_ENTER(MPID_STATE_LMT_SHM_PROGRESS_VC);

    *done = FALSE;

    if (vc_ch->lmt_active_lmt == NULL)
    {
        mpi_errno = get_next_req(vc);
        if (mpi_errno) MPIU_ERR_POP(mpi_errno);

        if (vc_ch->lmt_active_lmt == NULL)
        {
            /* couldn't find an appropriate request, try again later */
            goto fn_exit;
        }
    }

    we = vc_ch->lmt_active_lmt;
    mpi_errno = we->progress(vc, we->req, &done_req);
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

    if (done_req)
    {
        MPIU_Free(vc_ch->lmt_active_lmt);
        vc_ch->lmt_active_lmt = NULL;

        if (LMT_SHM_Q_EMPTY(vc_ch->lmt_queue))
            *done = TRUE;
    }

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_LMT_SHM_PROGRESS_VC);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_lmt_shm_progress
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPID_nem_lmt_shm_progress()
{
    int mpi_errno = MPI_SUCCESS;
    lmt_shm_prog_element_t *pe;
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_LMT_SHM_PROGRESS);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_LMT_SHM_PROGRESS);

    pe = LMT_SHM_L_HEAD();

    while (pe)
    {
        int done = FALSE;

        mpi_errno = lmt_shm_progress_vc(pe->vc, &done);
        if (mpi_errno) MPIU_ERR_POP(mpi_errno);

        if (done)
        {
            lmt_shm_prog_element_t *f;
            MPIU_Assert(LMT_SHM_Q_EMPTY(((MPIDI_CH3I_VC *)pe->vc->channel_private)->lmt_queue));
            MPIU_Assert(((MPIDI_CH3I_VC *)pe->vc->channel_private)->lmt_active_lmt == NULL);
            MPIU_Assert(((MPIDI_CH3I_VC *)pe->vc->channel_private)->lmt_enqueued);
            ((MPIDI_CH3I_VC *)pe->vc->channel_private)->lmt_enqueued = FALSE;

            f = pe;
            pe = pe->next;
            LMT_SHM_L_REMOVE(f);
            MPIU_Free(f);
        }
        else
            pe = pe->next;
    }

    if (LMT_SHM_L_EMPTY())
        MPID_nem_lmt_shm_pending = FALSE;

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_LMT_SHM_PROGRESS);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_allocate_shm_region
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int MPID_nem_allocate_shm_region(volatile MPID_nem_copy_buf_t **buf_p, char *handle[])
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_ALLOCATE_SHM_REGION);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_ALLOCATE_SHM_REGION);

    if (*buf_p)
    {
        /* we're already attached */
        goto fn_exit;
    }

    mpi_errno = MPID_nem_allocate_shared_memory((char **)buf_p, sizeof (MPID_nem_copy_buf_t), handle);
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_ALLOCATE_SHM_REGION);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_attach_shm_region
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int MPID_nem_attach_shm_region(volatile MPID_nem_copy_buf_t **buf_p, const char handle[])
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_ATTACH_SHM_REGION);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_ATTACH_SHM_REGION);

    if(*buf_p)
    {
        /* we're already attached */
        goto fn_exit;
    }

    mpi_errno = MPID_nem_attach_shared_memory((char **)buf_p, sizeof (MPID_nem_copy_buf_t), handle);
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

    mpi_errno = MPID_nem_remove_shared_memory(handle);
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_ATTACH_SHM_REGION);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_detach_shm_region
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int MPID_nem_detach_shm_region(volatile MPID_nem_copy_buf_t *buf, char handle[])
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_DETACH_SHM_REGION);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_DETACH_SHM_REGION);

    /* for now never detach */
    goto fn_exit;

    MPIU_Free(handle);

    mpi_errno = MPID_nem_detach_shared_memory ((char *)buf, sizeof (MPID_nem_copy_buf_t));
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_DETACH_SHM_REGION);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_delete_shm_region
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int MPID_nem_delete_shm_region(volatile MPID_nem_copy_buf_t *buf, char handle[])
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_DELETE_SHM_REGION);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_DELETE_SHM_REGION);

    mpi_errno = MPID_nem_remove_shared_memory(handle);
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

    MPIU_Free(handle);

    mpi_errno = MPID_nem_detach_shared_memory ((char *)buf, sizeof (MPID_nem_copy_buf_t));
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_DELETE_SHM_REGION);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}