ib_write_aggressive.c

刚才是说明 现在是安装程序在 LINUX环境下进行编程的MPICH安装文件

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

#include "ibimpl.h"

#ifdef WITH_METHOD_IB

/* prototypes */
#ifdef WITH_METHOD_SHM
int ib_stuff_vector_shm(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
#endif
#ifdef WITH_METHOD_VIA
int ib_stuff_vector_via(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
#endif
#ifdef WITH_METHOD_VIA_RDMA
int ib_stuff_vector_via_rdma(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
#endif
#ifdef WITH_METHOD_IB
int ib_stuff_vector_ib(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
#endif
#ifdef WITH_METHOD_NEW
int ib_stuff_vector_new(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
#endif
int ib_stuff_vector_vec(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
int ib_stuff_vector_tmp(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
int ib_stuff_vector_simple(MPID_IOV *, int *, MM_Car *, MM_Segment_buffer *);
int ib_update_car_num_written(MM_Car *, int *);

#ifndef min
#define min(a, b) ((a) < (b) ? (a) : (b))
#endif

#ifdef WITH_METHOD_SHM
int ib_stuff_vector_shm(MPID_IOV *vec, int *cur_pos, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_SHM);
    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_SHM);
    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_SHM);
    return FALSE;
}
#endif

#ifdef WITH_METHOD_VIA
int ib_stuff_vector_via(MPID_IOV *vec, int *cur_pos, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_VIA);
    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_VIA);
    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_VIA);
    return FALSE;
}
#endif

#ifdef WITH_METHOD_VIA_RDMA
int ib_stuff_vector_via_rdma(MPID_IOV *vec, int *cur_pos, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_VIA_RDMA);
    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_VIA_RDMA);
    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_VIA_RDMA);
    return FALSE;
}
#endif

int ib_stuff_vector_vec(MPID_IOV *vec, int *cur_pos_ptr, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    int cur_pos, cur_index, num_avail, final_segment;
    MPID_IOV *car_vec, *buf_vec;
    int num_left, i;
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_VEC);

    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_VEC);

    /* check to see that there is space in the vector to put data */
    if (*cur_pos_ptr == MPID_IOV_LIMIT)
    {
	/* cur_pos has run off the end of the vector */
	MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_VEC);
	return FALSE;
    }

    cur_index = car_ptr->data.ib.buf.vec_write.cur_index;
    car_vec = car_ptr->data.ib.buf.vec_write.vec;
    if (car_ptr->data.ib.buf.vec_write.num_read_copy != buf_ptr->vec.num_read)
    {
	/* update car vector */
	buf_vec = buf_ptr->vec.vec;
	
	/* update num_read_copy */
	car_ptr->data.ib.buf.vec_write.num_read_copy = buf_ptr->vec.num_read;
	
	/* copy the buf vector into the car vector from the current index to the end */
	memcpy(&car_vec[cur_index], &buf_vec[cur_index], 
	    (buf_ptr->vec.vec_size - cur_index) * sizeof(MPID_IOV));
	car_vec[cur_index].MPID_IOV_BUF = 
	    (char*)car_vec[cur_index].MPID_IOV_BUF + car_ptr->data.ib.buf.vec_write.num_written_at_cur_index;
	car_vec[cur_index].MPID_IOV_LEN = car_vec[cur_index].MPID_IOV_LEN - car_ptr->data.ib.buf.vec_write.num_written_at_cur_index;
	
	/* modify the vector copied from buf_ptr->vec to represent only the data that has been read 
	* This is done by traversing the vector, subtracting the lengths of each buffer until all the read
	* data is accounted for.
	*/
	
	/* set the size of the car vector to zero */
	car_ptr->data.ib.buf.vec_write.vec_size = 0;
	
	/* add vector elements to the size until all the read data is accounted for */
	num_left = car_ptr->data.ib.buf.vec_write.num_read_copy - car_ptr->data.ib.buf.vec_write.cur_num_written;
	i = cur_index;
	while (num_left > 0)
	{
	    car_ptr->data.ib.buf.vec_write.vec_size++;
	    num_left -= car_vec[i].MPID_IOV_LEN;
	    i++;
	}
	/* if the last vector buffer is larger than the amount of data read into that buffer,
	update the length field in the car's copy of the vector to represent only the read data */
	if (num_left < 0)
	{
	    car_vec[i].MPID_IOV_LEN += num_left;
	}
	
	/* at this point the vec in the car describes all the currently read data */
    }

    if (car_ptr->data.ib.buf.vec_write.cur_num_written == car_ptr->data.ib.buf.vec_write.num_read_copy)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_VEC);
	return FALSE;
    }

    /* copy as much of the car vector into the stuffed vector as possible */
    cur_pos = *cur_pos_ptr;
    /*cur_index = car_ptr->data.ib.buf.vec_write.cur_index;*/
    /* The amount available is the amount read minus the amount previously written. */
    num_avail = buf_ptr->vec.num_read - car_ptr->data.ib.buf.vec_write.cur_num_written;
    /* If the amount available for writing plus the total amount previously written is equal to the total size
       of the segment (segment_last) then set final_segment to TRUE */
    final_segment = (num_avail + car_ptr->data.ib.buf.vec_write.total_num_written) == buf_ptr->vec.segment_last;

    /*msg_printf("ib_stuff_vector_vec: num_avail = %d\n", num_avail);*/

    while ((cur_pos < MPID_IOV_LIMIT) && num_avail)
    {
	vec[cur_pos].MPID_IOV_BUF = car_vec[cur_index].MPID_IOV_BUF;
	num_avail -= (vec[cur_pos].MPID_IOV_LEN = car_vec[cur_index].MPID_IOV_LEN);
	cur_index++;
	cur_pos++;
    }
    *cur_pos_ptr = cur_pos;

    /* return TRUE if this segment is completely copied into the vec array */
    if (num_avail == 0 && final_segment)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_VEC);
	return TRUE;
    }

    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_VEC);
    return FALSE;
}

int ib_stuff_vector_tmp(MPID_IOV *vec, int *cur_pos, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_TMP);
    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_TMP);

    /* check to see that there is data available and space in the vector to put it */
    if ((*cur_pos == MPID_IOV_LIMIT) ||
        (car_ptr->data.ib.buf.tmp.num_written == buf_ptr->tmp.num_read) ||
	(buf_ptr->tmp.num_read == 0))
    {
	/* cur_pos has run off the end of the vector OR
	   If num_written is equal to num_read then there is no data available to write. OR
	   No data has been read */
	MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_TMP);
	return FALSE;
    }

    /*msg_printf("ib_stuff_vector_tmp: added %d byte buffer\n", buf_ptr->tmp.num_read - car_ptr->data.ib.buf.tmp.num_written);*/

    /* add the tmp buffer to the vector */
    vec[*cur_pos].MPID_IOV_BUF = (char*)(buf_ptr->tmp.buf) + car_ptr->data.ib.buf.tmp.num_written;
    vec[*cur_pos].MPID_IOV_LEN = buf_ptr->tmp.num_read - car_ptr->data.ib.buf.tmp.num_written;
    (*cur_pos)++;

    /* if the entire segment is in the vector then return true else false */
    if (buf_ptr->tmp.num_read == buf_ptr->tmp.len)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_TMP);
	return TRUE;
    }

    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_TMP);
    return FALSE;
}

int ib_stuff_vector_simple(MPID_IOV *vec, int *cur_pos, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_SIMPLE);
    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_SIMPLE);

    /* check to see that there is data available and space in the vector to put it */
    if ((*cur_pos == MPID_IOV_LIMIT) ||
        (car_ptr->data.ib.buf.simple.num_written == buf_ptr->simple.num_read) ||
	(buf_ptr->simple.num_read == 0))
    {
	/* cur_pos has run off the end of the vector OR
	   If num_written is equal to num_read then there is no data available to write. OR
	   No data has been read */
	MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_SIMPLE);
	return FALSE;
    }

    /*msg_printf("ib_stuff_vector_simple: added %d byte buffer\n", buf_ptr->simple.num_read - car_ptr->data.ib.buf.simple.num_written);*/

    /* add the simple buffer to the vector */
    vec[*cur_pos].MPID_IOV_BUF = (char*)(buf_ptr->simple.buf) + car_ptr->data.ib.buf.simple.num_written;
    vec[*cur_pos].MPID_IOV_LEN = buf_ptr->simple.num_read - car_ptr->data.ib.buf.simple.num_written;
    (*cur_pos)++;

    /* if the entire segment is in the vector then return true else false */
    if (buf_ptr->simple.num_read == buf_ptr->simple.len)
    {
	MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_SIMPLE);
	return TRUE;
    }

    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_SIMPLE);
    return FALSE;
}

#ifdef WITH_METHOD_IB
int ib_stuff_vector_ib(MPID_IOV *vec, int *cur_pos, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_IB);
    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_IB);
    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_IB);
    return FALSE;
}
#endif

#ifdef WITH_METHOD_NEW
int ib_stuff_vector_new(MPID_IOV *vec, int *cur_pos, MM_Car *car_ptr, MM_Segment_buffer *buf_ptr)
{
    MPIDI_STATE_DECL(MPID_STATE_IB_STUFF_VECTOR_NEW);
    MPIDI_FUNC_ENTER(MPID_STATE_IB_STUFF_VECTOR_NEW);
    MPIDI_FUNC_EXIT(MPID_STATE_IB_STUFF_VECTOR_NEW);
    return FALSE;
}
#endif

int ib_update_car_num_written(MM_Car *car_ptr, int *num_written_ptr)
{
    MM_Segment_buffer *buf_ptr;
    int num_written;
    int num_left, i;
    MPIDI_STATE_DECL(MPID_STATE_IB_UPDATE_CAR_NUM_WRITTEN);

    MPIDI_FUNC_ENTER(MPID_STATE_IB_UPDATE_CAR_NUM_WRITTEN);

#ifdef MPICH_DEV_BUILD
    if (car_ptr == NULL)
    {
	err_printf("Error: ib_update_car_num_written called on a NULL car pointer, num_written = %d\n", *num_written_ptr);
    }
#endif

    buf_ptr = car_ptr->buf_ptr;
    num_written = *num_written_ptr;

    switch (buf_ptr->type)
    {
#ifdef WITH_METHOD_SHM
    case MM_SHM_BUFFER:
	break;
#endif
#ifdef WITH_METHOD_VIA
    case MM_VIA_BUFFER:
	break;
#endif
#ifdef WITH_METHOD_VIA_RDMA
    case MM_VIA_RDMA_BUFFER:
	break;
#endif
    case MM_VEC_BUFFER:
	num_written = min(
	    /* the maximum amout of buffer space in this buffer that could have been written */
	    buf_ptr->vec.segment_last - car_ptr->data.ib.buf.vec_write.total_num_written,
	    /* the actual amount written */
	    num_written);

	/*msg_printf("num_written vec: %d\n", num_written);*/
	
	/* update vector */
	car_ptr->data.ib.buf.vec_write.cur_num_written += num_written;
	car_ptr->data.ib.buf.vec_write.total_num_written += num_written;
	if (car_ptr->data.ib.buf.vec_write.cur_num_written == buf_ptr->vec.buf_size)
	{
	    /* reset this car */
	    car_ptr->data.ib.buf.vec_write.cur_index = 0;
	    car_ptr->data.ib.buf.vec_write.num_read_copy = 0;
	    car_ptr->data.ib.buf.vec_write.cur_num_written = 0;
	    car_ptr->data.ib.buf.vec_write.num_written_at_cur_index = 0;
	    car_ptr->data.ib.buf.vec_write.vec_size = 0;
	    /* signal that we have finished writing the current vector */
	    mm_dec_atomic(&(buf_ptr->vec.num_cars_outstanding));
	}
	else
	{
	    num_left = num_written;