ch3_progress.c

fortran并行计算包

		    case MPIDI_CH3_PKT_CANCEL_SEND_REQ:
			{
			    MPIDI_CH3_Pkt_cancel_send_req_t * req_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->cancel_send_req;
			    MPID_Request * rreq;
			    int ack;
			    MPIDI_CH3_Pkt_t upkt;
			    MPIDI_CH3_Pkt_cancel_send_resp_t * resp_pkt = &upkt.cancel_send_resp;
			    MPID_Request * resp_sreq;
			    MPIDI_VC_t     *vc;

			    rreq = MPIDI_CH3U_Recvq_FDU(req_pkt->sender_req_id, &req_pkt->match);
			    if (rreq != NULL)
				{
				    if (MPIDI_Request_get_msg_type(rreq) == MPIDI_REQUEST_EAGER_MSG && rreq->dev.recv_data_sz > 0)
					{
					    MPIU_Free(rreq->dev.tmpbuf);
					}
				    MPID_Request_release(rreq);
				    ack = TRUE;
				}
			    else
				{
				    ack = FALSE;
				}
			    MPIDI_Pkt_init(resp_pkt, MPIDI_CH3_PKT_CANCEL_SEND_RESP);
			    resp_pkt->sender_req_id = req_pkt->sender_req_id;
			    resp_pkt->ack = ack;
			    if (in_fbox)
				{
				    MPIDI_PG_Get_vc (MPIDI_Process.my_pg, MPID_NEM_FBOX_SOURCE (cell), &vc);
				}
			    else
				{
				    MPIDI_PG_Get_vc (MPIDI_Process.my_pg, MPID_NEM_CELL_SOURCE (cell), &vc);
				}
			    mpi_errno = MPIDI_CH3_iStartMsg(vc, resp_pkt, sizeof(*resp_pkt), &resp_sreq);
			    /* --BEGIN ERROR HANDLING-- */
			    if (mpi_errno != MPI_SUCCESS)
				{
				    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**\
ch3|cancelresp", 0);
				    goto exit_fn;
				}
			    /* --END ERROR HANDLING-- */
			    if (resp_sreq != NULL)
				{
				    MPID_Request_release(resp_sreq);
				}
			}
			break;
		    case MPIDI_CH3_PKT_CANCEL_SEND_RESP:
			{
			    MPIDI_CH3_Pkt_cancel_send_resp_t * resp_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->cancel_send_resp;
			    MPID_Request * sreq;

			    MPID_Request_get_ptr(resp_pkt->sender_req_id, sreq);
			    if (resp_pkt->ack)
				{
				    sreq->status.cancelled = TRUE;
				    if (MPIDI_Request_get_msg_type(sreq) == MPIDI_REQUEST_RNDV_MSG ||
					MPIDI_Request_get_type(sreq) == MPIDI_REQUEST_TYPE_SSEND)
					{
					    int cc;
					    MPIDI_CH3U_Request_decrement_cc(sreq, &cc);
					}
				}
			    else
				{
				    MPIDI_DBG_PRINTF((35, FCNAME, "unable to cancel message"));
				}
			    MPIDI_CH3U_Request_complete(sreq);
			}
			break;
		    case MPIDI_CH3_PKT_PUT:
			{
			    MPIDI_CH3_Pkt_put_t * put_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->put;
			    fprintf(stdout,"ERROR : MPIDI_CH3_PKT_PUT not handled (yet) \n");
			}
			break;
		    case MPIDI_CH3_PKT_ACCUMULATE:
			{
			    MPIDI_CH3_Pkt_accum_t * accum_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->accum;
			    fprintf(stdout,"ERROR : MPIDI_CH3_PKT_ACCUMULATE not handled (yet) \n");
			}
			break;			
		    case MPIDI_CH3_PKT_GET:
			{
			    MPIDI_CH3_Pkt_get_t * get_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->get;
			    fprintf(stdout,"ERROR : MPIDI_CH3_PKT_GET not handled (yet) \n");
			}
			break;			
		    case MPIDI_CH3_PKT_GET_RESP:
			{
			    MPIDI_CH3_Pkt_get_resp_t * get_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->get_resp;
			    fprintf(stdout,"ERROR : MPIDI_CH3_PKT_GET_RESP not handled (yet) \n");
			}
			break;			
		    case MPIDI_CH3_PKT_LOCK:
			{
			    MPIDI_CH3_Pkt_lock_t * lock_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->lock;
			    MPID_Win *win_ptr;
			    MPIDI_VC_t     *vc;

			    if (in_fbox)
				{
				    MPIDI_PG_Get_vc (MPIDI_Process.my_pg, MPID_NEM_FBOX_SOURCE (cell), &vc);
				}
			    else
				{
				    MPIDI_PG_Get_vc (MPIDI_Process.my_pg, MPID_NEM_CELL_SOURCE (cell), &vc);
				}

			    MPID_Win_get_ptr(lock_pkt->target_win_handle, win_ptr);
			    if (MPIDI_CH3I_Try_acquire_win_lock(win_ptr,lock_pkt->lock_type) == 1)
				{
				    mpi_errno = MPIDI_CH3I_Send_lock_granted_pkt(vc,lock_pkt->source_win_handle);
				}
			    else
				{
				    /* queue the lock information */
				    MPIDI_Win_lock_queue *curr_ptr, *prev_ptr, *new_ptr;

				    /* FIXME: MT: This may need to be done atomically. */

				    curr_ptr = (MPIDI_Win_lock_queue *) win_ptr->lock_queue;
				    prev_ptr = curr_ptr;
				    while (curr_ptr != NULL)
					{
					    prev_ptr = curr_ptr;
					    curr_ptr = curr_ptr->next;
					}

				    new_ptr = (MPIDI_Win_lock_queue *) MPIU_Malloc(sizeof(MPIDI_Win_lock_queue));
				    /* --BEGIN ERROR HANDLING-- */
				    if (!new_ptr)
					{
					    mpi_errno = MPIR_Err_create_code( MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPI_ERR_OTHER,
									      "**nomem", 0 );
					    goto exit_fn;
					}
				    /* --END ERROR HANDLING-- */
				if (prev_ptr != NULL)
				    prev_ptr->next = new_ptr;
				else
				    win_ptr->lock_queue = new_ptr;
				
				new_ptr->next = NULL;
				new_ptr->lock_type = lock_pkt->lock_type;
				new_ptr->source_win_handle = lock_pkt->source_win_handle;
				new_ptr->vc = vc;
				new_ptr->pt_single_op = NULL;
				}
			}
			break;			
		    case MPIDI_CH3_PKT_LOCK_GRANTED:
			{
			    MPIDI_CH3_Pkt_lock_granted_t * lock_granted_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->lock_granted;
			    MPID_Win *win_ptr;
			    MPID_Win_get_ptr(lock_granted_pkt->source_win_handle, win_ptr);
			    win_ptr->lock_granted = 1;
			    MPIDI_CH3_Progress_signal_completion();
			}
			break;			
		    case MPIDI_CH3_PKT_PT_RMA_DONE:
			{
			    MPIDI_CH3_Pkt_pt_rma_done_t * pt_rma_done_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->pt_rma_done;
			    MPID_Win *win_ptr;
			    MPID_Win_get_ptr(pt_rma_done_pkt->source_win_handle, win_ptr);
			    win_ptr->lock_granted = 0;
			    MPIDI_CH3_Progress_signal_completion();
			}
			break;
		    case MPIDI_CH3_PKT_LOCK_PUT_UNLOCK:
                        {
                            MPIDI_CH3_Pkt_lock_put_unlock_t * lock_put_unlock_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->lock_put_unlock;
			    fprintf(stdout,"ERROR : MPIDI_CH3_PKT_LOCK_PUT_UNLOCK not handled (yet) \n");
			}
			break;
		    case MPIDI_CH3_PKT_LOCK_GET_UNLOCK:
                        {
                            MPIDI_CH3_Pkt_lock_get_unlock_t * lock_get_unlock_pkt = &((MPIDI_CH3_Pkt_t *)cell_buf)->lock_get_unlock;
			    fprintf(stdout,"ERROR : MPIDI_CH3_PKT_LOCK_GET_UNLOCK not handled (yet) \n");
			}
			break;						
		    default:
			{
			    /* nothing */
			}
		    }
		
		if (!in_fbox)
		    {
			MPIDI_VC_t *vc;
			MPIDI_PG_Get_vc (MPIDI_Process.my_pg, MPID_NEM_CELL_SOURCE (cell), &vc);
			MPID_nem_mpich2_release_cell (cell, vc);
		    }
		else
		    {
			MPID_nem_mpich2_release_fbox (cell);
		    }

		if(*foundp == FALSE)
		    {
			/* the cell does not match the request: create one */
			/* this is the request that sould be returned !    */
			goto make_req;
		    }
	    }
	else
	    {
	    make_req:
		rreq  = MPID_Request_create();
		if (rreq != NULL)
		    {
			MPIU_Object_set_ref(rreq, 2);
			rreq->kind                 = MPID_REQUEST_RECV;
			rreq->dev.match.tag        = tag;
			rreq->dev.match.rank       = source;
			rreq->dev.match.context_id = context_id;
			rreq->dev.next             = NULL;
			if (MPIDI_Process.recvq_posted_tail != NULL)
			    {
				MPIDI_Process.recvq_posted_tail->dev.next = rreq;
			    }
			else
			    {
				MPIDI_Process.recvq_posted_head = rreq;
			    }
			MPIDI_Process.recvq_posted_tail = rreq;
			MPIDI_POSTED_RECV_ENQUEUE_HOOK (rreq);
		    }
	    }	
    }
 exit_fn:
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_CH3I_PROGRESS_IPOKE_WITH_MATCHING);
    return rreq;
}

#endif  /* BYPASS_PROGRESS */

#undef FUNCNAME
#define FUNCNAME MPIDI_CH3I_Progress_init
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3I_Progress_init(void)
{
    int mpi_errno = MPI_SUCCESS;
    int i;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3I_PROGRESS_INIT);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3I_PROGRESS_INIT);

    for (i = 0; i < CH3_NUM_QUEUES; ++i)
    {
	MPIDI_CH3I_sendq_head[i] = NULL;
	MPIDI_CH3I_sendq_tail[i] = NULL;
    }

    /* Initialize the code to handle incoming packets */
    mpi_errno = MPIDI_CH3_PktHandler_Init(pktArray, PKTARRAY_SIZE);
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);
    mpi_errno = MPID_nem_lmt_pkthandler_init(pktArray, PKTARRAY_SIZE);
    if (mpi_errno) MPIU_ERR_POP(mpi_errno);

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

#undef FUNCNAME
#define FUNCNAME MPIDI_CH3I_Progress_finalize
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3I_Progress_finalize(void)
{
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3I_PROGRESS_FINALIZE);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3I_PROGRESS_FINALIZE);

    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_CH3I_PROGRESS_FINALIZE);
    return MPI_SUCCESS;
}


#undef FUNCNAME
#define FUNCNAME MPIDI_CH3_Connection_terminate
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3_Connection_terminate (MPIDI_VC_t * vc)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3_CONNECTION_TERMINATE);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3_CONNECTION_TERMINATE);
    if (((MPIDI_CH3I_VC *)vc->channel_private)->is_local)
        mpi_errno = MPID_nem_vc_terminate(vc);
    else
        mpi_errno = MPID_nem_net_module_vc_terminate(vc);
    if(mpi_errno) MPIU_ERR_POP(mpi_errno);

    mpi_errno = MPIDI_CH3U_Handle_connection (vc, MPIDI_VC_EVENT_TERMINATED);
    if(mpi_errno) MPIU_ERR_POP(mpi_errno);

fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_CH3_CONNECTION_TERMINATE);
    return mpi_errno;
fn_fail:
    goto fn_exit;
}
/* end MPIDI_CH3_Connection_terminate() */



#undef FUNCNAME
#define FUNCNAME MPIDI_CH3I_Posted_recv_enqueued
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3I_Posted_recv_enqueued (MPID_Request *rreq)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3I_POSTED_RECV_ENQUEUED);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3I_POSTED_RECV_ENQUEUED);
    /* don't enqueue for anysource */
    if (rreq->dev.match.rank < 0)
	goto fn_exit;
    /* don't enqueue a fastbox for yourself */
    if (rreq->dev.match.rank == MPIDI_CH3I_my_rank)
	goto fn_exit;
    /* don't enqueue non-local processes */
    if (!MPID_NEM_IS_LOCAL (rreq->dev.match.rank))
	goto fn_exit;

    mpi_errno = MPID_nem_mpich2_enqueue_fastbox (MPID_NEM_LOCAL_RANK (rreq->dev.match.rank));
    if (mpi_errno) MPIU_ERR_POP (mpi_errno);

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

#undef FUNCNAME
#define FUNCNAME MPIDI_CH3I_Posted_recv_dequeued
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3I_Posted_recv_dequeued (MPID_Request *rreq)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3I_POSTED_RECV_DEQUEUED);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3I_POSTED_RECV_DEQUEUED);
    if (rreq->dev.match.rank < 0)
	goto fn_exit;
    if (rreq->dev.match.rank == MPIDI_CH3I_my_rank)
	goto fn_exit;
    if (!MPID_NEM_IS_LOCAL (rreq->dev.match.rank))
	goto fn_exit;

    mpi_errno = MPID_nem_mpich2_dequeue_fastbox (MPID_NEM_LOCAL_RANK (rreq->dev.match.rank));
    if (mpi_errno) MPIU_ERR_POP (mpi_errno);

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