ibu_wait.ibal.c

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

				else 
				{
				    /* switch to EAGER sending with special packet type (set in switch routine)*/
				    MPIDI_CH3_Pkt_t pkt;
				    MPIDI_CH3_Pkt_rndv_eager_send_t* rndv_eager_pkt = &pkt.rndv_eager_send;
				    MPIU_DBG_PRINTF(("sending eager packet instead of rndv failed sender reloading\n"));

				    /* send new eager packet to the receiver */
#if defined(MPID_USE_SEQUENCE_NUMBERS)
				    rndv_eager_pkt->seqnum		= recv_vc_ptr->seqnum_send;
#endif    
				    rndv_eager_pkt->match.rank		= sreq->comm->rank;
				    rndv_eager_pkt->match.tag		= sreq->dev.match.tag;
				    rndv_eager_pkt->match.context_id	= sreq->dev.match.context_id;
				    pkt.rndv_eager_send.sender_req_id	= sreq->handle;
				    pkt.rndv_eager_send.receiver_req_id	= ((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->rreq; 

				    pkt.rndv_eager_send.type = MPIDI_CH3_PKT_RNDV_EAGER_SEND;
				    mpi_errno = MPIDI_CH3I_Switch_rndv_to_eager(recv_vc_ptr, sreq, &pkt);

				    /* --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);
					MPIU_DBG_PRINTFX(("exiting ibu_wait k\n"));
					MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
					return mpi_errno;
				    }
				    /* --END ERROR HANDLING-- */								
				}
			    }
			}
			else if (((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->send_recv == MPIDI_CH3_PKT_RELOAD_RECV)
			{
			    MPIU_DBG_PRINTF(("received reload recv packet (rreq=0x%x).\n", ((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->rreq));
			    MPID_Request_get_ptr(((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->rreq, rreq);
			    MPIU_DBG_PRINTF(("unregistering the receiver's iov.\n"));
			    rreq->ch.rndv_status = IBU_RNDV_SUCCESS;
			    for (i=0; i<rreq->dev.iov_count; i++)
			    {
				ibu_reg_status = ibu_deregister_memory(
				    rreq->dev.iov[i].MPID_IOV_BUF, 
				    rreq->dev.iov[i].MPID_IOV_LEN,
				    &rreq->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 g\n"));
				    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				    return mpi_errno;
				}
				/* --END ERROR HANDLING-- */
			    }
			    mpi_errno = MPIDI_CH3U_Handle_recv_req(recv_vc_ptr, rreq, &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 request after receiving a reload packet");
				MPIU_DBG_PRINTFX(("exiting ibu_wait o\n"));
				MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				return mpi_errno;
			    }
			    if (!complete)
			    {
				/* send a new iov */
				MPID_Request * cts_sreq;
				MPIDI_CH3_Pkt_t pkt;

				MPIU_DBG_PRINTF(("registering the receiver's iov.\n"));
				rreq->ch.rndv_status = IBU_RNDV_NO_DEREG;
				for (i=0; i<rreq->dev.iov_count; i++)
				{
				    rreq->ch.rndv_status = IBU_RNDV_SUCCESS;
				    ibu_reg_status = ibu_register_memory(
					rreq->dev.iov[i].MPID_IOV_BUF, 
					rreq->dev.iov[i].MPID_IOV_LEN,
					&rreq->ch.local_iov_mem[i]);
				    if (ibu_reg_status != IBU_SUCCESS) break;
				}
				if (ibu_reg_status != IBU_SUCCESS)
				{
				    rreq->ch.rndv_status = IBU_RNDV_CTS_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(
						rreq->dev.iov[i].MPID_IOV_BUF, 
						rreq->dev.iov[i].MPID_IOV_LEN, 
						&rreq->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 h\n"));
						MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
						return mpi_errno;
					    }
					    /* --END ERROR HANDLING-- */
					}
				    }								
				}
				if (rreq->ch.rndv_status == IBU_RNDV_SUCCESS)
				{
				    MPIU_DBG_PRINTF(("sending reloaded recv iov of length %d\n", rreq->dev.iov_count));								
				    MPIDI_Pkt_init(&pkt.iov, MPIDI_CH3_PKT_IOV);
				    pkt.iov.send_recv = MPIDI_CH3_PKT_RELOAD_RECV;
				    pkt.iov.req = ((MPIDI_CH3_Pkt_rdma_reload_t*)mem_ptr)->sreq;
				    pkt.iov.iov_len = rreq->dev.iov_count;

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

				    mpi_errno = MPIDI_CH3_iStartMsgv(recv_vc_ptr, rreq->dev.rdma_iov, 3, &cts_sreq);
				    /* --BEGIN ERROR HANDLING-- */
				    if (mpi_errno != MPI_SUCCESS)
				    {
					/* This destruction probably isn't correct. */
					/* I think it needs to save the error in the request, complete the request and return */
					MPIU_Object_set_ref(rreq, 0);
					MPIDI_CH3_Request_destroy(rreq);
					rreq = NULL;
					mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**ch3|ctspkt", 0);
					MPIU_DBG_PRINTFX(("exiting ibu_wait p\n"));
					MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
					return mpi_errno;
				    }
				    /* --END ERROR HANDLING-- */
				    if (cts_sreq != NULL)
				    {
					/* The sender doesn't need to know when the message has been sent.  So release the request immediately */
					MPID_Request_release(cts_sreq);
				    }
				}
				else
				{
				    /* Send a CTS_IOV_REG_ERROR packet. */
				    mpi_errno = ibui_post_rndv_cts_iov_reg_err(ibu, 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);
					MPIU_DBG_PRINTFX(("exiting ibu_wait k\n"));
					MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
					return mpi_errno;
				    }
				    /* --END ERROR HANDLING-- */	
				}
			    }
			}
			else
			{
			    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", 0);
			    MPIU_DBG_PRINTFX(("exiting ibu_wait q\n"));
			    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			    return mpi_errno;
			}

			recv_vc_ptr->ch.recv_active = NULL;
			recv_vc_ptr->ch.reading_pkt = TRUE;
			if (num_bytes > sizeof(MPIDI_CH3_Pkt_t))
			{
			    MPIU_DBG_PRINTF(("pkt handled with %d bytes remaining to be buffered.\n", num_bytes));
			    ibui_buffer_unex_read(ibu, mem_ptr_orig, sizeof(MPIDI_CH3_Pkt_t), num_bytes);
			}
			else
			{
			    /* Added by Mellanox, dafna April 11th removed post_receive */
			    /* ibuBlockFreeIB(ibu->allocator, mem_ptr_orig); 
			    ibui_post_receive(ibu);*/
			}
			/* return from the wait */
			*num_bytes_ptr = 0;
			*vc_pptr = recv_vc_ptr;
			*op_ptr = IBU_OP_WAKEUP;
			MPIU_DBG_PRINTFX(("exiting ibu_wait q\n"));
			MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			return MPI_SUCCESS;
		    }
		    else if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type == MPIDI_CH3_PKT_RNDV_CTS_IOV_REG_ERR)
		    {
			MPIDI_CH3_Pkt_t pkt;
			MPIDI_CH3_Pkt_rndv_eager_send_t* rndv_eager_pkt = &pkt.rndv_eager_send;
			MPIU_DBG_PRINTF(("received cts iov registration failure packet(sreq=0x%x).\n", ((MPIDI_CH3_Pkt_rndv_reg_error_t*)mem_ptr)->sreq));					
			MPID_Request_get_ptr(((MPIDI_CH3_Pkt_rndv_reg_error_t*)mem_ptr)->sreq, sreq);
			sreq->dev.rdma_request = ((MPIDI_CH3_Pkt_rndv_reg_error_t*)mem_ptr)->rreq;

			MPIU_DBG_PRINTF(("sending eager packet instead of rndv that failed at receiver\n"));

			/* send new eager packet to the receiver */
#if defined(MPID_USE_SEQUENCE_NUMBERS)
			rndv_eager_pkt->seqnum		    = recv_vc_ptr->seqnum_send;
#endif    					
			rndv_eager_pkt->match.rank	    = sreq->comm->rank;
			rndv_eager_pkt->match.tag	    = sreq->dev.match.tag;
			rndv_eager_pkt->match.context_id    = sreq->dev.match.context_id;
			pkt.rndv_eager_send.sender_req_id   = sreq->handle;
			pkt.rndv_eager_send.receiver_req_id = sreq->dev.rdma_request;
#ifndef USE_RDMA_GET
			/* only in USE_RDMA_PUT sreq registration may have failed when receiving this packet type,
			otherwise - previous registration has definitely succeeded */
			if (sreq->ch.rndv_status == IBU_RNDV_SUCCESS)
#endif
			{
			    MPIU_DBG_PRINTF(("unregistering the sender's iov.\n"));
			    /* unpin the sender's iov */
			    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 i\n"));
				    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				    return mpi_errno;
				}
				/* --END ERROR HANDLING-- */
			    }
			    sreq->ch.rndv_status = IBU_RNDV_NO_DEREG;
			}
			pkt.rndv_eager_send.type = MPIDI_CH3_PKT_RNDV_EAGER_SEND;
			mpi_errno = MPIDI_CH3I_Switch_rndv_to_eager(recv_vc_ptr, sreq, &pkt);
			/* --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 failed swithching to rndv a\n"));
			    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			    return mpi_errno;
			}
			/* --END ERROR HANDLING-- */

		    }
		    else if (((MPIDI_CH3_Pkt_t*)mem_ptr)->type == MPIDI_CH3_PKT_RNDV_EAGER_SEND)
		    {
			MPIDI_CH3_Pkt_rndv_eager_send_t * rndv_eager_pkt = &((MPIDI_CH3_Pkt_t*)mem_ptr)->rndv_eager_send;
			MPIU_DBG_PRINTF(("received rndv eager packet(sreq=0x%x).\n",
			    ((MPIDI_CH3_Pkt_rndv_eager_send_t*)mem_ptr)->sender_req_id));

			MPID_Request_get_ptr(rndv_eager_pkt->receiver_req_id, rreq);

			if (rreq->ch.rndv_status == IBU_RNDV_SUCCESS) 
			{
			    MPIU_DBG_PRINTF(("deregistering buffers.\n"));
			    /* unpin the receiver's iov */
			    for (i=0; i<rreq->dev.iov_count; i++)
			    {							
				ibu_reg_status = ibu_deregister_memory(
				    rreq->dev.iov[i].MPID_IOV_BUF, 
				    rreq->dev.iov[i].MPID_IOV_LEN, &
				    rreq->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 j\n"));
				    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
				    return mpi_errno;
				}
				/* --END ERROR HANDLING-- */
			    }
			    rreq->ch.rndv_status = IBU_RNDV_NO_DEREG;
			}
			/* Need to receive the data on this rndv_eager_message */

			rreq->status.MPI_SOURCE = rndv_eager_pkt->match.rank;
			rreq->status.MPI_TAG	= rndv_eager_pkt->match.tag;
			rreq->status.count	= rndv_eager_pkt->data_sz;
			rreq->dev.sender_req_id = rndv_eager_pkt->sender_req_id;
			rreq->dev.recv_data_sz	= rndv_eager_pkt->data_sz;
			MPIDI_Request_set_seqnum(rreq, rndv_eager_pkt->seqnum);
			MPIDI_Request_set_msg_type(rreq, MPIDI_REQUEST_EAGER_MSG);
			recv_vc_ptr->ch.recv_active = rreq;
			mpi_errno = MPIDI_CH3U_Post_data_receive(TRUE, &recv_vc_ptr->ch.recv_active);
			if (mpi_errno != MPI_SUCCESS)
			{
			    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "infiniband read progress unable to handle incoming packet");
			    MPIU_DBG_PRINTFX(("exiting ibu_wait rs\n"));
			    MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			    return mpi_errno;
			}
		    }
		    else
		    {
			mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "shared memory read progress unable to handle unknown rdma packet");
			MPIU_DBG_PRINTFX(("exiting ibu_wait r\n"));
			MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			return mpi_errno;
		    }
		}
		else
#endif /* MPIDI_CH3_CHANNEL_RNDV */
		{
		    mpi_errno = MPIDI_CH3U_Handle_recv_pkt(recv_vc_ptr, (MPIDI_CH3_Pkt_t*)mem_ptr, &recv_vc_ptr->ch.recv_active);
		    if (mpi_errno != MPI_SUCCESS)
		    {
			mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**fail", "**fail %s", "infiniband read progress unable to handle incoming packet");
			MPIU_DBG_PRINTFX(("exiting ibu_wait s\n"));
			MPIDI_FUNC_EXIT(MPID_STATE_IBU_WAIT);
			return mpi_errno;
		    }
		}
		if (recv_vc_ptr->ch.recv_active == NULL)
		{
		    MPIU_DBG_PRINTF(("packet with no data handled.\n"));
		    recv_vc_ptr->ch.reading_pkt = TRUE;
		}
		else
		{
		    /*
		    int z;