ibu.vapi.c

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

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

#include "mpidimpl.h"
#include "ibu.h"
#ifdef HAVE_STDIO_H
#include <stdio.h>
#endif
#ifdef HAVE_MALLOC_H
#include <malloc.h>
#endif
#include "mpidi_ch3_impl.h"
#ifdef HAVE_WINDOWS_H
#ifdef USE_DEBUG_ALLOCATION_HOOK
#include <crtdbg.h>
#endif
#endif

#ifdef USE_IB_VAPI

#include "ibuimpl.vapi.h"

IBU_Global IBU_Process;

/*============== static functions prototypes ==================*/
static VAPI_ret_t ibui_get_list_of_hcas(VAPI_hca_id_t **hca_id_buf_p, u_int32_t *num_of_hcas);

/* utility ibu functions */
#undef FUNCNAME
#define FUNCNAME getMaxInlineSize
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
VAPI_ret_t getMaxInlineSize( ibu_t ibu )
{
    VAPI_ret_t status;
    VAPI_qp_attr_t qp_attr;
    VAPI_qp_init_attr_t qp_init_attr;
    VAPI_qp_attr_mask_t qp_attr_mask;
    MPIDI_STATE_DECL(MPID_STATE_IBU_GETMAXINLINESIZE);

    MPIDI_FUNC_ENTER(MPID_STATE_IBU_GETMAXINLINESIZE);

    MPIU_DBG_PRINTF(("entering getMaxInlineSize\n"));

    QP_ATTR_MASK_CLR_ALL(qp_attr_mask);
    status = VAPI_query_qp(IBU_Process.hca_handle, 
	ibu->qp_handle, 
	&qp_attr,
	&qp_attr_mask, 
	&qp_init_attr );
    if ( status != VAPI_OK )
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_GETMAXINLINESIZE);
	return status;
    }
    ibu->max_inline_size = qp_attr.cap.max_inline_data_sq;

    MPIU_DBG_PRINTF(("exiting getMaxInlineSize\n"));
    MPIDI_FUNC_EXIT(MPID_STATE_IBU_GETMAXINLINESIZE);
    return IBU_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME modifyQP
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
VAPI_ret_t modifyQP( ibu_t ibu, VAPI_qp_state_t qp_state )
{
    VAPI_ret_t status;
    VAPI_qp_attr_t qp_attr;
    VAPI_qp_cap_t qp_cap;
    VAPI_qp_attr_mask_t qp_attr_mask;
    MPIDI_STATE_DECL(MPID_STATE_IBU_MODIFYQP);

    MPIDI_FUNC_ENTER(MPID_STATE_IBU_MODIFYQP);

    MPIU_DBG_PRINTF(("entering modifyQP\n"));
    if (qp_state == VAPI_INIT)
    {
	QP_ATTR_MASK_CLR_ALL(qp_attr_mask);
	qp_attr.qp_state = VAPI_INIT;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_STATE);
	qp_attr.pkey_ix = 0;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PKEY_IX);
	qp_attr.port = IBU_Process.port;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PORT);
	qp_attr.remote_atomic_flags = VAPI_EN_REM_WRITE | VAPI_EN_REM_READ;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_REMOTE_ATOMIC_FLAGS);
	MPIU_DBG_PRINTF(("moving to INIT on port %d\n", IBU_Process.port));
    }
    else if (qp_state == VAPI_RTR) 
    {
	QP_ATTR_MASK_CLR_ALL(qp_attr_mask);
	qp_attr.qp_state         = VAPI_RTR;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_STATE);
	qp_attr.qp_ous_rd_atom   = 1;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_OUS_RD_ATOM);
	qp_attr.path_mtu         = MTU1024;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PATH_MTU);
	qp_attr.rq_psn           = 0;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_RQ_PSN);
	qp_attr.pkey_ix          = 0;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PKEY_IX);
	qp_attr.min_rnr_timer    = 0;/*5;*/
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_MIN_RNR_TIMER);
	qp_attr.dest_qp_num = ibu->dest_qp_num;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_DEST_QP_NUM);
	qp_attr.av.sl            = 0;
	qp_attr.av.grh_flag      = 0;
	qp_attr.av.static_rate   = 0;
	qp_attr.av.src_path_bits = 0;
	qp_attr.av.dlid = ibu->dlid;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_AV);
	MPIU_DBG_PRINTF(("moving to RTR qp(%d:%d => %d:%d)\n", IBU_Process.lid, ibu->qp_num, ibu->dlid, ibu->dest_qp_num));
    }
    else if (qp_state == VAPI_RTS)
    {
	QP_ATTR_MASK_CLR_ALL(qp_attr_mask);
	qp_attr.qp_state         = VAPI_RTS;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_STATE);
	qp_attr.sq_psn           = 0;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_SQ_PSN);
	qp_attr.timeout          = /*0x20;*/ 10;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_TIMEOUT);
	qp_attr.retry_count      = 1; /*5;*/
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_RETRY_COUNT);
	qp_attr.rnr_retry        = 3; /*1;*/
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_RNR_RETRY);
	qp_attr.ous_dst_rd_atom  = 1; /*255;*/
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_OUS_DST_RD_ATOM);
	MPIU_DBG_PRINTF(("moving to RTS\n"));
    }
    else if (qp_state == VAPI_RESET)
    {
	QP_ATTR_MASK_CLR_ALL(qp_attr_mask);
	qp_attr.qp_state = VAPI_RESET;
	QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_STATE);
    }
    else
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_MODIFYQP);
	return IBU_FAIL;
    }

    status = VAPI_modify_qp(IBU_Process.hca_handle, 
	ibu->qp_handle, 
	&qp_attr,
	&qp_attr_mask, 
	&qp_cap );
    if ( status != VAPI_OK )
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_MODIFYQP);
	return status;
    }

    MPIU_DBG_PRINTF(("exiting modifyQP\n"));
    MPIDI_FUNC_EXIT(MPID_STATE_IBU_MODIFYQP);
    return IBU_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME createQP
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
VAPI_ret_t createQP(ibu_t ibu, ibu_set_t set)
{
    VAPI_ret_t status;
    VAPI_qp_init_attr_t qp_init_attr;
    VAPI_qp_prop_t qp_prop;
    MPIDI_STATE_DECL(MPID_STATE_IBU_CREATEQP);

    MPIDI_FUNC_ENTER(MPID_STATE_IBU_CREATEQP);
    MPIU_DBG_PRINTF(("entering createQP\n"));
    qp_init_attr.cap.max_oust_wr_rq = IBU_DEFAULT_MAX_WQE; /*10000;*/
    qp_init_attr.cap.max_oust_wr_sq = IBU_DEFAULT_MAX_WQE; /*10000;*/
    qp_init_attr.cap.max_sg_size_rq = 8;
    qp_init_attr.cap.max_sg_size_sq = 8;
    qp_init_attr.pd_hndl = IBU_Process.pd_handle;
    qp_init_attr.rdd_hndl = 0;
    qp_init_attr.rq_cq_hndl = set;
    qp_init_attr.sq_cq_hndl = set;
    qp_init_attr.rq_sig_type = /*VAPI_SIGNAL_ALL_WR;*/ VAPI_SIGNAL_REQ_WR;
    qp_init_attr.sq_sig_type = /*VAPI_SIGNAL_ALL_WR;*/ VAPI_SIGNAL_REQ_WR;
    qp_init_attr.ts_type = IB_TS_RC;

    status = VAPI_create_qp(IBU_Process.hca_handle, &qp_init_attr, &ibu->qp_handle, &qp_prop);
    if (status != VAPI_OK)
    {
	MPIU_Internal_error_printf("VAPI_create_qp failed, error %s\n", VAPI_strerror(status));
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_CREATEQP);
	return status;
    }
    ibu->qp_num = qp_prop.qp_num;
    MPIU_DBG_PRINTF(("exiting createQP\n"));
    MPIDI_FUNC_EXIT(MPID_STATE_IBU_CREATEQP);
    return IBU_SUCCESS;
}


/*Mellanox - sleybo 
args:			
ibu - pointer to ibu_state_t 
retrun value:
p - memory hndl of the buffers , NULL if fail.

function that allocate local RDMA buffers and register them to the HCA


*/
#undef FUNCNAME
#define FUNCNAME ibui_update_remote_RDMA_buf
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int ibui_update_remote_RDMA_buf(ibu_t ibu, ibu_rdma_buf_t* buf, VAPI_rkey_t rkey)
{
    MPIDI_STATE_DECL(MPID_STATE_IBUI_UPDATE_REMOTE_RDMA_BUF);

    MPIDI_FUNC_ENTER(MPID_STATE_IBUI_UPDATE_REMOTE_RDMA_BUF);
    MPIU_DBG_PRINTF(("entering ibui_update_remote_RDMA_buf\n"));
    ibu->remote_RDMA_buf_base = buf;
    ibu->remote_RDMA_buf_hndl.rkey = rkey;
    ibu->remote_RDMA_head = 0;
    ibu->remote_RDMA_limit = IBU_NUM_OF_RDMA_BUFS - 1;

    MPIU_DBG_PRINTF(("ibu_update_RDMA_buf  rkey = %x  buf= %p\n",rkey,buf));
    MPIDI_FUNC_EXIT(MPID_STATE_IBUI_UPDATE_REMOTE_RDMA_BUF);
    return IBU_SUCCESS;
}




/*Mellanox - sleybo 
args:			
ibu - pointer to ibu_state_t 
retrun value:
p - memory hndl of the buffers , NULL if fail.

function that allocate local RDMA buffers and register them to the HCA
*/
#undef FUNCNAME
#define FUNCNAME ibui_RDMA_buf_init
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
ibu_rdma_buf_t* ibui_RDMA_buf_init(ibu_t ibu, VAPI_rkey_t* rkey)
{
    ibu_rdma_buf_t *buf;
    ibu_mem_t mem_handle;
    MPIDI_STATE_DECL(MPID_STATE_IBUI_RDMA_BUF_INIT);

    MPIDI_FUNC_ENTER(MPID_STATE_IBUI_RDMA_BUF_INIT);
    MPIU_DBG_PRINTF(("entering ibui_RDMA_buf_init\n"));
    buf = ibuRDMAAllocInitIB(&mem_handle); 

    if (buf == NULL)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IBUI_RDMA_BUF_INIT);
	return NULL;
    }

    ibu->local_RDMA_buf_base = buf;
    ibu->local_RDMA_buf_hndl = mem_handle; 
    ibu->local_RDMA_head =  0;
    ibu->local_last_updated_RDMA_limit = IBU_NUM_OF_RDMA_BUFS - 1;

    *rkey = ibu->local_RDMA_buf_hndl.rkey;
    MPIU_DBG_PRINTF(("ibui_RDMA_buf_init   rkey = %x  buf= %p\n", *rkey, buf));
    MPIDI_FUNC_EXIT(MPID_STATE_IBUI_RDMA_BUF_INIT);
    return buf;
}

#undef FUNCNAME
#define FUNCNAME ibu_start_qp
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
ibu_t ibu_start_qp(ibu_set_t set, int *qp_num_ptr)
{
    VAPI_ret_t status;
    ibu_t p;
    MPIDI_STATE_DECL(MPID_STATE_IBU_START_QP);

    MPIDI_FUNC_ENTER(MPID_STATE_IBU_START_QP);
    MPIU_DBG_PRINTF(("entering ibu_start_qp\n"));
    p = (ibu_t)MPIU_Malloc(sizeof(ibu_state_t));
    if (p == NULL)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_START_QP);
	return NULL;
    }

    memset(p, 0, sizeof(ibu_state_t));
    p->state = 0;
    p->allocator = ibuBlockAllocInitIB();

    /*MPIDI_DBG_PRINTF((60, FCNAME, "creating the queue pair\n"));*/
    /* Create the queue pair */
    status = createQP(p, set);
    if (status != IBU_SUCCESS)
    {
	MPIU_Internal_error_printf("ibu_create_qp: createQP failed, error %s\n", VAPI_strerror(status));
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_START_QP);
	return NULL;
    }
    *qp_num_ptr = p->qp_num;
    MPIDI_FUNC_EXIT(MPID_STATE_IBU_START_QP);
    return p;
}

#undef FUNCNAME
#define FUNCNAME ibu_finish_qp
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int ibu_finish_qp(ibu_t p, ibu_lid_t dest_lid, int dest_qp_num)
{
    VAPI_ret_t status;
    MPIDI_STATE_DECL(MPID_STATE_IBU_FINISH_QP);

    MPIDI_FUNC_ENTER(MPID_STATE_IBU_FINISH_QP);
    MPIU_DBG_PRINTF(("entering ibu_finish_qp\n"));

    p->dest_qp_num = dest_qp_num;
    p->dlid = dest_lid;
    /*MPIDI_DBG_PRINTF((60, FCNAME, "modifyQP(INIT)"));*/
    status = modifyQP(p, VAPI_INIT);
    if (status != IBU_SUCCESS)
    {
	MPIU_Internal_error_printf("ibu_finish_qp: modifyQP(INIT) failed, error %s\n", VAPI_strerror(status));
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_FINISH_QP);
	return -1;
    }
    /*MPIDI_DBG_PRINTF((60, FCNAME, "modifyQP(RTR)"));*/
    status = modifyQP(p, VAPI_RTR);
    if (status != IBU_SUCCESS)
    {
	MPIU_Internal_error_printf("ibu_finish_qp: modifyQP(RTR) failed, error %s\n", VAPI_strerror(status));
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_FINISH_QP);
	return -1;
    }

    status = modifyQP(p, VAPI_RTS);
    if (status != IBU_SUCCESS)
    {
	MPIU_Internal_error_printf("ibu_finish_qp: modifyQP(RTS) failed, error %s\n", VAPI_strerror(status));
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_FINISH_QP);
	return -1;
    }

    status = getMaxInlineSize(p);
    if (status != IBU_SUCCESS)
    {
	MPIU_Internal_error_printf("ibu_finish_qp: getMaxInlineSize() failed, error %s\n", VAPI_strerror(status));
	MPIDI_FUNC_EXIT(MPID_STATE_IBU_FINISH_QP);
	return -1;
    }

    /* Mellanox dafna July 11th*/
    p->send_wqe_info_fifo.head = 0;
    p->send_wqe_info_fifo.tail = 0;	
    /* Mellanox END dafna July 11th*/

    MPIU_DBG_PRINTF(("exiting ibu_create_qp\n"));    
    MPIDI_FUNC_EXIT(MPID_STATE_IBU_FINISH_QP);
    return IBU_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME ibui_post_ack_write
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int ibui_post_ack_write(ibu_t ibu)
{
    int mpi_errno = MPI_SUCCESS;
    int num_bytes;
    MPIDI_CH3_Pkt_t upkt;
    MPIDI_CH3_Pkt_rdma_limit_upt_t * const ack_pkt = &upkt.limit_upt;
    MPIDI_STATE_DECL(MPID_STATE_IBUI_POST_ACK_WRITE);

    MPIDI_FUNC_ENTER(MPID_STATE_IBUI_POST_ACK_WRITE);
    MPIU_DBG_PRINTF(("entering ibui_post_ack_write\n"));

    ack_pkt->iov_len = 0;
    ack_pkt->type = MPIDI_CH3_PKT_LMT_UPT;