mpid_nem_inline.h

fortran并行计算包

MPID_nem_mpich2_dequeue_fastbox (int local_rank)
{
    int mpi_errno = MPI_SUCCESS;
    MPID_nem_fboxq_elem_t *el;

    el = &MPID_nem_fboxq_elem_list[local_rank];    

    MPIU_ERR_CHKANDJUMP (!el->usage, mpi_errno, MPI_ERR_OTHER, "**intern");

    --el->usage;
    if (el->usage == 0)
    {
	if (el->prev == NULL)
	    MPID_nem_fboxq_head = el->next;
	else
	    el->prev->next = el->next;

	if (el->next == NULL)
	    MPID_nem_fboxq_tail = el->prev;
	else
	    el->next->prev = el->prev;

	if (el == MPID_nem_curr_fboxq_elem)
	{
	    if (el->next == NULL)
		MPID_nem_curr_fboxq_elem = MPID_nem_fboxq_head;
	    else
		MPID_nem_curr_fboxq_elem = el->next;
	}
    }
    
 fn_exit:
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}

/*
  MPID_nem_mpich2_enqueue_fastbox (int local_rank)
  enqueues fastbox for process with local rank local_rank on fbox queue
  This function is called whenever a receive is posted for a process on this node.
  Fastboxes on fbox queue are polled regularly for incoming messages.
*/
#undef FUNCNAME
#define FUNCNAME MPID_nem_mpich2_dequeue_fastbox
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
MPID_NEM_INLINE_DECL
int MPID_nem_mpich2_enqueue_fastbox (int local_rank)
{
    int mpi_errno = MPI_SUCCESS;
    MPID_nem_fboxq_elem_t *el;

    el = &MPID_nem_fboxq_elem_list[local_rank];
	
    if (el->usage)
    {
	++el->usage;
    }
    else
    {
	el->usage = 1;
	if (MPID_nem_fboxq_tail == NULL)
	{
	    el->prev = NULL;
	    MPID_nem_curr_fboxq_elem = MPID_nem_fboxq_head = el;
	}
	else
	{
	    el->prev = MPID_nem_fboxq_tail;
	    MPID_nem_fboxq_tail->next = el;
	}
	    
	el->next = NULL;
	MPID_nem_fboxq_tail = el;
    }    

    return mpi_errno;
}
/*
  MPID_nem_recv_seqno_matches (MPID_nem_queue_ptr_t qhead)
  check whether the sequence number for the cell at the head of qhead is the one
  expected from the sender of that cell
  We only check these for processes in COMM_WORLD (i.e. the ones initially allocated)
*/
#undef FUNCNAME
#define FUNCNAME MPID_nem_recv_seqno_matches
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
MPID_NEM_INLINE_DECL int
MPID_nem_recv_seqno_matches (MPID_nem_queue_ptr_t qhead)
{
    MPID_nem_cell_ptr_t cell = MPID_NEM_REL_TO_ABS(qhead->my_head);
    int source = cell->pkt.mpich2.source;
    
    return (cell->pkt.mpich2.seqno == MPID_nem_recv_seqno[source]);
}

/*
  int MPID_nem_mpich2_test_recv (MPID_nem_cell_ptr_t *cell, int *in_fbox);

  non-blocking receive
  sets cell to the received cell, or NULL if there is nothing to receive. in_fbox is true iff the cell was found in a fbox
  the cell must be released back to the subsystem with MPID_nem_mpich2_release_cell() once the packet has been copied out
*/
#undef FUNCNAME
#define FUNCNAME MPID_nem_mpich2_test_recv
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
MPID_NEM_INLINE_DECL int
MPID_nem_mpich2_test_recv (MPID_nem_cell_ptr_t *cell, int *in_fbox)
{
    int mpi_errno = MPI_SUCCESS;
    
    DO_PAPI (PAPI_reset (PAPI_EventSet));

#ifdef ENABLED_CHECKPOINTING
    MPID_nem_ckpt_maybe_take_checkpoint();

    if (MPID_nem_ckpt_message_log)
    {
	MPID_nem_ckpt_replay_message (cell);
	MPIU_Assert ((*cell)->pkt.mpich2.seqno == MPID_nem_recv_seqno[(*cell)->pkt.mpich2.source]);
	++MPID_nem_recv_seqno[(*cell)->pkt.mpich2.source];
	*in_fbox = 0;
	goto fn_exit;
    }
#endif
    
#ifdef USE_FASTBOX
    poll_fboxes (cell, goto fbox_l);
#endif/* USE_FASTBOX     */

    if ((MPID_NEM_NET_MODULE != MPID_NEM_NO_MODULE) && (MPID_nem_mem_region.ext_procs > 0))
    {
	mpi_errno = MPID_nem_network_poll (MPID_NEM_POLL_IN);
        if (mpi_errno) MPIU_ERR_POP (mpi_errno);
    }

    if (MPID_nem_queue_empty (MPID_nem_mem_region.my_recvQ) || !MPID_nem_recv_seqno_matches (MPID_nem_mem_region.my_recvQ))
    {
#ifdef USE_FASTBOX
	poll_all_fboxes (cell, goto fbox_l);
#endif/* USE_FASTBOX     */
	*cell = NULL;
	goto fn_exit;
    }
    
    MPID_nem_queue_dequeue (MPID_nem_mem_region.my_recvQ, cell);

    ++MPID_nem_recv_seqno[(*cell)->pkt.mpich2.source];
    *in_fbox = 0;

 fn_exit:
#ifdef ENABLED_CHECKPOINTING
    if ((*cell)->pkt.header.type == MPID_NEM_PKT_CKPT)
	MPID_nem_ckpt_got_marker (cell, in_fbox);
    else if (MPID_nem_ckpt_logging_messages)
	MPID_nem_ckpt_log_message (*cell);
#endif
    DO_PAPI (PAPI_accum_var (PAPI_EventSet, PAPI_vvalues6));
    
    MPIU_DBG_STMT (CH3_CHANNEL, VERBOSE, {
	if (*cell)
	{
	    MPIU_DBG_MSG_S (CH3_CHANNEL, VERBOSE, "<-- Recv %s", (*in_fbox) ? "fbox " : "queue");
	    MPIU_DBG_STMT (CH3_CHANNEL, VERBOSE, MPID_nem_dbg_dump_cell (*cell));
	}
    });

 fn_fail:
    return mpi_errno;

 fbox_l:
    *in_fbox = 1;
    goto fn_exit;

}

/*
  int MPID_nem_mpich2_test_recv_wait (MPID_nem_cell_ptr_t *cell, int *in_fbox, int timeout);

  blocking receive with timeout
  waits up to timeout iterations to receive a cell
  sets cell to the received cell, or NULL if there is nothing to receive. in_fbox is true iff the cell was found in a fbox
  the cell must be released back to the subsystem with MPID_nem_mpich2_release_cell() once the packet has been copied out
*/
#undef FUNCNAME
#define FUNCNAME MPID_nem_mpich2_test_recv_wait
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
MPID_NEM_INLINE_DECL int
MPID_nem_mpich2_test_recv_wait (MPID_nem_cell_ptr_t *cell, int *in_fbox, int timeout)
{
    int mpi_errno = MPI_SUCCESS;
    
#ifdef USE_FASTBOX
    poll_fboxes (cell, goto fbox_l);
#endif/* USE_FASTBOX     */

    if ((MPID_NEM_NET_MODULE != MPID_NEM_NO_MODULE) && (MPID_nem_mem_region.ext_procs > 0))
    {
	mpi_errno = MPID_nem_network_poll (MPID_NEM_POLL_IN);
        if (mpi_errno) MPIU_ERR_POP (mpi_errno);
    }

    while ((--timeout > 0) && (MPID_nem_queue_empty (MPID_nem_mem_region.my_recvQ) || !MPID_nem_recv_seqno_matches (MPID_nem_mem_region.my_recvQ)))
    {
#ifdef USE_FASTBOX
	poll_all_fboxes (cell, goto fbox_l);
#endif/* USE_FASTBOX     */
	*cell = NULL;
	goto exit_l;
    }
    
    MPID_nem_queue_dequeue (MPID_nem_mem_region.my_recvQ, cell);

    ++MPID_nem_recv_seqno[(*cell)->pkt.mpich2.source];
    *in_fbox = 0;
 exit_l:
    
    MPIU_DBG_STMT (CH3_CHANNEL, VERBOSE, {
            if (*cell)
            {
                MPIU_DBG_MSG_S (CH3_CHANNEL, VERBOSE, "<-- Recv %s", (*in_fbox) ? "fbox " : "queue");
                MPIU_DBG_STMT (CH3_CHANNEL, VERBOSE, MPID_nem_dbg_dump_cell (*cell));
            }
        });

 fn_fail:
    return mpi_errno;

 fbox_l:
    *in_fbox = 1;
    goto exit_l;
}

/*
  int MPID_nem_mpich2_blocking_recv (MPID_nem_cell_ptr_t *cell, int *in_fbox);

  blocking receive waits until there is something to receive, or then
  sets cell to the received cell. in_fbox is true iff the cell was
  found in a fbox the cell must be released back to the subsystem with
  MPID_nem_mpich2_release_cell() once the packet has been copied out
*/
#undef FUNCNAME
#define FUNCNAME MPID_nem_mpich2_blocking_recv
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
MPID_NEM_INLINE_DECL int
MPID_nem_mpich2_blocking_recv(MPID_nem_cell_ptr_t *cell, int *in_fbox)
{
    int mpi_errno = MPI_SUCCESS;
    unsigned completions = MPIDI_CH3I_progress_completion_count;
#ifndef ENABLE_NO_SCHED_YIELD
    int pollcount = 0;
#endif
    DO_PAPI (PAPI_reset (PAPI_EventSet));

#ifdef MPICH_IS_THREADED
    /* We should never enter this function in a multithreaded app */
#ifdef HAVE_RUNTIME_THREADCHECK
    MPIU_Assert(!MPIR_ThreadInfo.isThreaded);
#else
    MPIU_Assert(0);
#endif
#endif


#ifdef ENABLED_CHECKPOINTING
    MPID_nem_ckpt_maybe_take_checkpoint();

 top_l:
    if (MPID_nem_ckpt_message_log)
    {
	MPID_nem_ckpt_replay_message (cell);
	MPIU_Assert ((*cell)->pkt.mpich2.seqno == MPID_nem_recv_seqno[(*cell)->pkt.mpich2.source]);
	++MPID_nem_recv_seqno[(*cell)->pkt.mpich2.source];
	*in_fbox = 0;
	return mpi_errno;
    }
#endif
    
    
#ifdef USE_FASTBOX
    poll_fboxes (cell, goto fbox_l);
#endif /*USE_FASTBOX */

    if ((MPID_NEM_NET_MODULE != MPID_NEM_NO_MODULE) && (MPID_nem_mem_region.ext_procs > 0))
    {
	mpi_errno = MPID_nem_network_poll (MPID_NEM_POLL_IN);
        if (mpi_errno) MPIU_ERR_POP (mpi_errno);
    }

    while (MPID_nem_queue_empty (MPID_nem_mem_region.my_recvQ) || !MPID_nem_recv_seqno_matches (MPID_nem_mem_region.my_recvQ))
    {
	DO_PAPI (PAPI_reset (PAPI_EventSet));

#ifdef USE_FASTBOX	
	poll_all_fboxes (cell, goto fbox_l);
	poll_fboxes (cell, goto fbox_l);
#endif /*USE_FASTBOX */

	if ((MPID_NEM_NET_MODULE != MPID_NEM_NO_MODULE) && (MPID_nem_mem_region.ext_procs > 0))
	{            
	    mpi_errno = MPID_nem_network_poll (MPID_NEM_POLL_IN);
            if (mpi_errno) MPIU_ERR_POP (mpi_errno);

            if (completions != MPIDI_CH3I_progress_completion_count || MPID_nem_lmt_shm_pending || MPIDI_CH3I_active_send[CH3_NORMAL_QUEUE]
                || MPIDI_CH3I_SendQ_head(CH3_NORMAL_QUEUE))
            {
                *cell = NULL;
                *in_fbox = 0;
                goto exit_l;
            }
	}
#ifndef ENABLE_NO_SCHED_YIELD
	if (pollcount >= MPID_NEM_POLLS_BEFORE_YIELD)
	{
	    pollcount = 0;
	    sched_yield();
	}
	++pollcount;
#endif
    }

    MPID_nem_queue_dequeue (MPID_nem_mem_region.my_recvQ, cell);

    ++MPID_nem_recv_seqno[(*cell)->pkt.mpich2.source];
    *in_fbox = 0;

 exit_l:    

#ifdef ENABLED_CHECKPOINTING
    if (*cell)
    {
        if ((*cell)->pkt.header.type == MPID_NEM_PKT_CKPT)
        {
            MPID_nem_ckpt_got_marker (cell, in_fbox);
            goto top_l;
        }
        else if (MPID_nem_ckpt_logging_messages)
            MPID_nem_ckpt_log_message (*cell);
    }
#endif

    DO_PAPI (PAPI_accum_var (PAPI_EventSet,PAPI_vvalues8));
    
    MPIU_DBG_STMT (CH3_CHANNEL, VERBOSE, {
            if (*cell)
            {
                MPIU_DBG_MSG_S (CH3_CHANNEL, VERBOSE, "<-- Recv %s", (*in_fbox) ? "fbox " : "queue");
                MPIU_DBG_STMT (CH3_CHANNEL, VERBOSE, MPID_nem_dbg_dump_cell(*cell));
            }
        });

 fn_fail:
    return mpi_errno;

 fbox_l:
    *in_fbox = 1;
    goto exit_l;
}

/*
  int MPID_nem_mpich2_release_cell (MPID_nem_cell_ptr_t cell, MPIDI_VC_t *vc);

  releases the cell back to the subsystem to be used for subsequent receives
*/
#undef FUNCNAME
#define FUNCNAME MPID_nem_mpich2_release_cell
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
MPID_NEM_INLINE_DECL int
MPID_nem_mpich2_release_cell (MPID_nem_cell_ptr_t cell, MPIDI_VC_t *vc)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_CH3I_VC *vc_ch = (MPIDI_CH3I_VC *)vc->channel_private;
    DO_PAPI (PAPI_reset (PAPI_EventSet));
#ifdef ENABLED_CHECKPOINTING
    if (cell->pkt.header.type == MPID_NEM_PKT_CKPT_REPLAY)
    {
	if (!MPID_nem_ckpt_message_log)
	    /* this is the last replayed message */
	    MPID_nem_ckpt_free_msg_log();
	return mpi_errno;
    }
#endif
    MPID_nem_queue_enqueue (vc_ch->free_queue, cell);
    DO_PAPI (PAPI_accum_var (PAPI_EventSet,PAPI_vvalues9));
    return mpi_errno;
}

#endif /*_MPID_NEM_INLINE_H*/