ibu_wait.vapi.c

mpi并行计算的c++代码 可用vc或gcc编译通过 可以用来搭建并行计算试验环境

		    *num_bytes_ptr = 0;
		    *vc_pptr = ibu->vc_ptr;
		    *op_ptr = IBU_OP_WAKEUP;
		    MPIU_DBG_PRINTFX(("exiting ibu_wait h\n"));
		    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
		    return MPI_SUCCESS;
		}
	    }
	    break;
#endif /* MPIDI_CH3_CHANNEL_RNDV */
	case VAPI_CQE_SQ_SEND_DATA:
	    if (completion_data.status != VAPI_SUCCESS)
	    {
		MPIU_Internal_error_printf("%s: send completion status = %s\n",
		    FCNAME, VAPI_wc_status_sym(completion_data.status));
		mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", VAPI_wc_status_sym(completion_data.status));
		PrintWC(&completion_data);
		MPIU_DBG_PRINTFX(("exiting ibu_wait 40\n"));
		MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
		return mpi_errno;
	    }
	    if (mem_ptr == (void*)-1)
	    {
		MPIU_DBG_PRINTF(("ack sent\n"));
		/* flow control ack completed, no user data so break out here */
		MPIU_DBG_PRINTF(("ack sent.\n"));
		break;
	    }
	    num_bytes = send_length;
	    MPIDI_DBG_PRINTF((60, FCNAME, "send num_bytes = %d\n", num_bytes));
	    ibuBlockFreeIB(ibu->allocator, mem_ptr);

	    *num_bytes_ptr = num_bytes;
	    *op_ptr = IBU_OP_TIMEOUT;
	    *vc_pptr = ibu->vc_ptr;
	    MPIU_DBG_PRINTFX(("exiting ibu_wait 5\n"));
	    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
	    return MPI_SUCCESS;
	    break;
	case VAPI_CQE_RQ_SEND_DATA:
	    if (completion_data.status != VAPI_SUCCESS)
	    {
		MPIU_Internal_error_printf("%s: recv completion status = %s\n",
		    FCNAME, VAPI_wc_status_sym(completion_data.status));
		mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", VAPI_wc_status_sym(completion_data.status));
		PrintWC(&completion_data);
		MPIU_DBG_PRINTFX(("exiting ibu_wait 41\n"));
		MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
		return mpi_errno;
	    }
	    if (completion_data.imm_data_valid)
	    {
		MPIU_Assert(completion_data.imm_data_valid != 1); /* Error! */
		break;
	    }
	    num_bytes = completion_data.byte_len;
	    recv_vc_ptr = ibu->vc_ptr;
	    pkt_offset = 0;
	    if (recv_vc_ptr->ch.reading_pkt)
	    {
#ifdef MPIDI_CH3_CHANNEL_RNDV
		if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type > MPIDI_CH3_PKT_END_CH3)
		{
		    if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type == MPIDI_CH3_PKT_RTS_IOV)
		    {
			MPIU_DBG_PRINTF(("received rts packet(sreq=0x%x).\n",
			    ((MPIDI_CH3_Pkt_rdma_rts_iov_t*)mem_ptr)->sreq));
			rreq = MPID_Request_create();
			if (rreq == NULL)
			{
			    mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**nomem", 0);
			    MPIU_DBG_PRINTFX(("exiting ibu_wait h\n"));
			    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			    return mpi_errno;
			}
			MPIU_Object_set_ref(rreq, 1);
			rreq->kind = MPIDI_CH3I_RTS_IOV_READ_REQUEST;
			rreq->dev.rdma_request = ((MPIDI_CH3_Pkt_rdma_rts_iov_t*)mem_ptr)->sreq;
			rreq->dev.rdma_iov_count = ((MPIDI_CH3_Pkt_rdma_rts_iov_t*)mem_ptr)->iov_len;
			rreq->dev.iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&rreq->dev.rdma_iov;
			rreq->dev.iov[0].MPID_IOV_LEN = rreq->dev.rdma_iov_count * sizeof(MPID_IOV);
			rreq->dev.iov[1].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&rreq->ch.remote_iov_mem[0];
			rreq->dev.iov[1].MPID_IOV_LEN = rreq->dev.rdma_iov_count * sizeof(ibu_mem_t);
			rreq->dev.iov[2].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&rreq->ch.pkt;
			rreq->dev.iov[2].MPID_IOV_LEN = sizeof(MPIDI_CH3_Pkt_t);
			rreq->dev.iov_count = 3;		
			rreq->ch.req = NULL;
			recv_vc_ptr->ch.recv_active = rreq;
		    }
		    else if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type == MPIDI_CH3_PKT_RTS_PUT)
		    {
			int found;
			MPIU_DBG_PRINTF(("received rts put packet(sreq=0x%x).\n",
			    ((MPIDI_CH3_Pkt_rndv_req_to_send_t*)mem_ptr)->sender_req_id));

			mpi_errno = MPIDI_CH3U_Handle_recv_rndv_pkt(recv_vc_ptr,
			    (MPIDI_CH3_Pkt_t*)mem_ptr,
			    &rreq, &found);
			/* --BEGIN ERROR HANDLING-- */
			if (mpi_errno != MPI_SUCCESS)
			{
			    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "ibu read progress unable to handle incoming rts(put) packet");
			    MPIU_DBG_PRINTFX(("exiting ibu_wait v\n"));
			    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			    return mpi_errno;
			}
			/* --END ERROR HANDLING-- */

			if (found)
			{
			    mpi_errno = MPIDI_CH3U_Post_data_receive(found, &rreq);
			    /* --BEGIN ERROR HANDLING-- */
			    if (mpi_errno != MPI_SUCCESS)
			    {
				mpi_errno = MPIR_Err_create_code (mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", 0);
				MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				return mpi_errno;
			    }
			    /* --END ERROR HANDLING-- */
			    mpi_errno = MPIDI_CH3_iStartRndvTransfer(recv_vc_ptr, rreq);
			    /* --BEGIN ERROR HANDLING-- */
			    if (mpi_errno != MPI_SUCCESS)
			    {
				mpi_errno = MPIR_Err_create_code (mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", 0);
				MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				return mpi_errno;
			    }
			    /* --END ERROR HANDLING-- */
			}

			recv_vc_ptr->ch.recv_active = NULL;
		    }
		    else if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type == MPIDI_CH3_PKT_CTS_IOV)
		    {
			MPIU_DBG_PRINTF(("received cts packet(sreq=0x%x, rreq=0x%x).\n",
			    ((MPIDI_CH3_Pkt_rdma_cts_iov_t*)mem_ptr)->sreq,
			    ((MPIDI_CH3_Pkt_rdma_cts_iov_t*)mem_ptr)->rreq));
			MPID_Request_get_ptr(((MPIDI_CH3_Pkt_rdma_cts_iov_t*)mem_ptr)->sreq, sreq);
			sreq->dev.rdma_request = ((MPIDI_CH3_Pkt_rdma_cts_iov_t*)mem_ptr)->rreq;
			sreq->dev.rdma_iov_count = ((MPIDI_CH3_Pkt_rdma_cts_iov_t*)mem_ptr)->iov_len;
			rreq = MPID_Request_create();
			if (rreq == NULL)
			{
			    mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**nomem", 0);
			    MPIU_DBG_PRINTFX(("exiting ibu_wait i\n"));
			    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			    return mpi_errno;
			}
			MPIU_Object_set_ref(rreq, 1);
			rreq->kind = MPIDI_CH3I_IOV_WRITE_REQUEST;
			rreq->dev.iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&sreq->dev.rdma_iov;
			rreq->dev.iov[0].MPID_IOV_LEN = sreq->dev.rdma_iov_count * sizeof(MPID_IOV);
			rreq->dev.iov[1].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&sreq->ch.remote_iov_mem[0];
			rreq->dev.iov[1].MPID_IOV_LEN = sreq->dev.rdma_iov_count * sizeof(ibu_mem_t);
			rreq->dev.iov_count = 2;
			rreq->ch.req = sreq;
			recv_vc_ptr->ch.recv_active = rreq;
		    }
		    else if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type == MPIDI_CH3_PKT_IOV)
		    {
			if ( ((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->send_recv == MPIDI_CH3_PKT_RELOAD_SEND )
			{
			    MPIU_DBG_PRINTF(("received sender's iov packet, posting a read of %d iovs.\n", ((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->iov_len));
			    MPID_Request_get_ptr(((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->req, sreq);
			    sreq->dev.rdma_iov_count = ((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->iov_len;
			    rreq = MPID_Request_create();
			    if (rreq == NULL)
			    {
				mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**nomem", 0);
				MPIU_DBG_PRINTFX(("exiting ibu_wait j\n"));
				MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				return mpi_errno;
			    }
			    MPIU_Object_set_ref(rreq, 1);
			    rreq->kind = MPIDI_CH3I_IOV_READ_REQUEST;
			    rreq->dev.iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&sreq->dev.rdma_iov;
			    rreq->dev.iov[0].MPID_IOV_LEN = sreq->dev.rdma_iov_count * sizeof(MPID_IOV);
			    rreq->dev.iov[1].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&sreq->ch.remote_iov_mem[0];
			    rreq->dev.iov[1].MPID_IOV_LEN = sreq->dev.rdma_iov_count * sizeof(ibu_mem_t);
			    rreq->dev.iov_count = 2;
			    rreq->ch.req = sreq;
			    recv_vc_ptr->ch.recv_active = rreq;
			}
			else if ( ((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->send_recv == MPIDI_CH3_PKT_RELOAD_RECV )
			{
			    MPIU_DBG_PRINTF(("received receiver's iov packet, posting a read of %d iovs.\n", ((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->iov_len));
			    MPID_Request_get_ptr(((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->req, rreq);
			    rreq->dev.rdma_iov_count = ((MPIDI_CH3_Pkt_rdma_iov_t*)mem_ptr)->iov_len;
			    sreq = MPID_Request_create();
			    if (sreq == NULL)
			    {
				mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**nomem", 0);
				MPIU_DBG_PRINTFX(("exiting ibu_wait k\n"));
				MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				return mpi_errno;
			    }
			    MPIU_Object_set_ref(sreq, 1);
			    sreq->kind = MPIDI_CH3I_IOV_WRITE_REQUEST;
			    sreq->dev.iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&rreq->dev.rdma_iov;
			    sreq->dev.iov[0].MPID_IOV_LEN = rreq->dev.rdma_iov_count * sizeof(MPID_IOV);
			    sreq->dev.iov[1].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&rreq->ch.remote_iov_mem[0];
			    sreq->dev.iov[1].MPID_IOV_LEN = rreq->dev.rdma_iov_count * sizeof(ibu_mem_t);
			    sreq->dev.iov_count = 2;
			    sreq->ch.req = rreq;
			    recv_vc_ptr->ch.recv_active = sreq;
			}
			else
			{
			    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "received invalid MPIDI_CH3_PKT_IOV packet");
			    MPIU_DBG_PRINTFX(("exiting ibu_wait l\n"));
			    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			    return mpi_errno;
			}
		    }
		    else if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type == MPIDI_CH3_PKT_RELOAD)
		    {
			if (((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->send_recv == MPIDI_CH3_PKT_RELOAD_SEND)
			{
			    MPIU_DBG_PRINTF(("received reload send packet (sreq=0x%x).\n", ((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->sreq));
			    MPID_Request_get_ptr(((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->sreq, sreq);
			    MPIU_DBG_PRINTF(("unregistering the sender's iov.\n"));
			    sreq->ch.rndv_status = IBU_RNDV_SUCCESS;
			    for (i=0; i<sreq->dev.iov_count; i++)
			    {
				ibu_reg_status = ibu_deregister_memory(
				    sreq->dev.iov[i].MPID_IOV_BUF, 
				    sreq->dev.iov[i].MPID_IOV_LEN,
				    &sreq->ch.local_iov_mem[i]);
				/* --BEGIN ERROR HANDLING-- */
				if (ibu_reg_status != IBU_SUCCESS)
				{
				    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "unable to deregister");
				    MPIU_DBG_PRINTFX(("exiting ibu_wait deregister failed e\n"));
				    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				    return mpi_errno;
				}
				/* --END ERROR HANDLING-- */
			    }
			    mpi_errno = MPIDI_CH3U_Handle_send_req(recv_vc_ptr, sreq, &complete);
			    if (mpi_errno != MPI_SUCCESS)
			    {
				mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "unable to update send request after receiving a reload packet");
				MPIU_DBG_PRINTFX(("exiting ibu_wait m\n"));
				MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				return mpi_errno;
			    }
			    if (!complete)
			    {
				/* send a new iov */
				MPID_Request * rts_sreq;
				MPIDI_CH3_Pkt_t pkt;

				MPIU_DBG_PRINTF(("registering the sender's iov.\n"));
				sreq->ch.rndv_status = IBU_RNDV_NO_DEREG;
				for (i=0; i<sreq->dev.iov_count; i++)
				{
				    sreq->ch.rndv_status = IBU_RNDV_SUCCESS;
				    ibu_reg_status = ibu_register_memory(
					sreq->dev.iov[i].MPID_IOV_BUF, 
					sreq->dev.iov[i].MPID_IOV_LEN,
					&sreq->ch.local_iov_mem[i]);
				    if (ibu_reg_status != IBU_SUCCESS) break;
				}

				if (ibu_reg_status != IBU_SUCCESS)
				{
				    sreq->ch.rndv_status = IBU_RNDV_RTS_IOV_FAIL;
				    /* Deregister all register buffers we are not going to use */ 
				    MPIU_DBG_PRINTF(("ibu_register_memory failed. deregistering the sender's iov.\n"));
				    if (i > 0) 
				    {
					for (i-=1; i==0; i--) /* take last i's value one down, since last did not succeed*/
					{
					    ibu_reg_status = ibu_deregister_memory(
						sreq->dev.iov[i].MPID_IOV_BUF, 
						sreq->dev.iov[i].MPID_IOV_LEN, 
						&sreq->ch.local_iov_mem[i]);
					    /* --BEGIN ERROR HANDLING-- */
					    if (ibu_reg_status != IBU_SUCCESS)
					    {
						mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "unable to deregister");
						MPIU_DBG_PRINTFX(("exiting ibu_wait deregister failed f\n"));
						MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
						return mpi_errno;
					    }
					    /* --END ERROR HANDLING-- */
					}
				    }								
				}
				if (sreq->ch.rndv_status == IBU_RNDV_SUCCESS) 
				{
				    MPIU_DBG_PRINTF(("sending reloaded send iov of length %d\n", sreq->dev.iov_count));								
				    MPIDI_Pkt_init(&pkt.iov, MPIDI_CH3_PKT_IOV);
				    pkt.iov.send_recv = MPIDI_CH3_PKT_RELOAD_SEND;
				    pkt.iov.req = ((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->rreq;
				    pkt.iov.iov_len = sreq->dev.iov_count;

				    sreq->dev.rdma_iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&pkt;
				    sreq->dev.rdma_iov[0].MPID_IOV_LEN = sizeof(MPIDI_CH3_Pkt_t);
				    sreq->dev.rdma_iov[1].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)sreq->dev.iov;
				    sreq->dev.rdma_iov[1].MPID_IOV_LEN = sreq->dev.iov_count * sizeof(MPID_IOV);
				    sreq->dev.rdma_iov[2].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)&sreq->ch.local_iov_mem[0];
				    sreq->dev.rdma_iov[2].MPID_IOV_LEN = sreq->dev.iov_count * sizeof(ibu_mem_t);

				    mpi_errno = MPIDI_CH3_iStartMsgv(recv_vc_ptr, sreq->dev.rdma_iov, 3, &rts_sreq);
				    /* --BEGIN ERROR HANDLING-- */
				    if (mpi_errno != MPI_SUCCESS)
				    {
					MPIU_Object_set_ref(sreq, 0);
					MPIDI_CH3_Request_destroy(sreq);
					sreq = NULL;
					mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**ch3|rtspkt", 0);
					MPIU_DBG_PRINTFX(("exiting ibu_wait n\n"));
					MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
					return mpi_errno;
				    }
				    /* --END ERROR HANDLING-- */
				    if (rts_sreq != NULL)
				    {
					/* The sender doesn't need to know when the message has been sent.  So release the request immediately */
					MPID_Request_release(rts_sreq);
				    }
				}