ad_bgl_read.c

fortran并行计算包

/* ---------------------------------------------------------------- */
/* (C)Copyright IBM Corp.  2007, 2008                               */
/* ---------------------------------------------------------------- */
/**
 * \file ad_bgl_read.c
 * \brief ???
 */

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/* 
 *   Copyright (C) 1997 University of Chicago. 
 *   See COPYRIGHT notice in top-level directory.
 */

#include "ad_bgl.h"
#include "adio_extern.h"

#include "ad_bgl_tuning.h"

void ADIOI_BGL_ReadContig(ADIO_File fd, void *buf, int count, 
                     MPI_Datatype datatype, int file_ptr_type,
		     ADIO_Offset offset, ADIO_Status *status, int *error_code)
{
    int err=-1, datatype_size, len;
    static char myname[] = "ADIOI_BGL_READCONTIG";

#if BGL_PROFILE
		/* timing */
		double io_time, io_time2;

		if (bglmpio_timing) {
		    io_time = MPI_Wtime();
		    bglmpio_prof_cr[ BGLMPIO_CIO_DATA_SIZE ] += len;
		}
#endif

    MPI_Type_size(datatype, &datatype_size);
    len = datatype_size * count;

#if BGL_PROFILE

    if (file_ptr_type == ADIO_EXPLICIT_OFFSET) {
        	if (bglmpio_timing2) io_time2 = MPI_Wtime();
	if (fd->fp_sys_posn != offset)
	    lseek(fd->fd_sys, offset, SEEK_SET);
        	if (bglmpio_timing2) bglmpio_prof_cr[ BGLMPIO_CIO_T_SEEK ] += (MPI_Wtime() - io_time2);
	if (fd->atomicity)
	    ADIOI_WRITE_LOCK(fd, offset, SEEK_SET, len);
	else ADIOI_READ_LOCK(fd, offset, SEEK_SET, len);
        	if (bglmpio_timing2) io_time2 = MPI_Wtime();
	err = read(fd->fd_sys, buf, len);
        	if (bglmpio_timing2) bglmpio_prof_cr[ BGLMPIO_CIO_T_POSI_RW ] += (MPI_Wtime() - io_time2);
	ADIOI_UNLOCK(fd, offset, SEEK_SET, len);
	fd->fp_sys_posn = offset + err;
	/* individual file pointer not updated */        
    }
    else {  /* read from curr. location of ind. file pointer */
	offset = fd->fp_ind;
        	if (bglmpio_timing2) io_time2 = MPI_Wtime();
	if (fd->fp_sys_posn != fd->fp_ind)
	    lseek(fd->fd_sys, fd->fp_ind, SEEK_SET);
        	if (bglmpio_timing2) bglmpio_prof_cr[ BGLMPIO_CIO_T_SEEK ] += (MPI_Wtime() - io_time2);
	if (fd->atomicity)
	    ADIOI_WRITE_LOCK(fd, offset, SEEK_SET, len);
	else ADIOI_READ_LOCK(fd, offset, SEEK_SET, len);
        	if (bglmpio_timing2) io_time2 = MPI_Wtime();
	err = read(fd->fd_sys, buf, len);
        	if (bglmpio_timing2) bglmpio_prof_cr[ BGLMPIO_CIO_T_POSI_RW ] += (MPI_Wtime() - io_time2);
	ADIOI_UNLOCK(fd, offset, SEEK_SET, len);
	fd->fp_ind += err;
	fd->fp_sys_posn = fd->fp_ind;
    }

#else	/* BGL_PROFILE */

    if (file_ptr_type == ADIO_EXPLICIT_OFFSET) {
	if (fd->fp_sys_posn != offset)
	    lseek(fd->fd_sys, offset, SEEK_SET);
	if (fd->atomicity)
	    ADIOI_WRITE_LOCK(fd, offset, SEEK_SET, len);
	else ADIOI_READ_LOCK(fd, offset, SEEK_SET, len);
	err = read(fd->fd_sys, buf, len);
	ADIOI_UNLOCK(fd, offset, SEEK_SET, len);
	fd->fp_sys_posn = offset + err;
	/* individual file pointer not updated */        
    }
    else {  /* read from curr. location of ind. file pointer */
	offset = fd->fp_ind;
	if (fd->fp_sys_posn != fd->fp_ind)
	    lseek(fd->fd_sys, fd->fp_ind, SEEK_SET);
	if (fd->atomicity)
	    ADIOI_WRITE_LOCK(fd, offset, SEEK_SET, len);
	else ADIOI_READ_LOCK(fd, offset, SEEK_SET, len);
	err = read(fd->fd_sys, buf, len);
	ADIOI_UNLOCK(fd, offset, SEEK_SET, len);
	fd->fp_ind += err;
	fd->fp_sys_posn = fd->fp_ind;
    }

#endif   /* BGL_PROFILE */

#if BGL_PROFILE
    		if (bglmpio_timing) bglmpio_prof_cr[ BGLMPIO_CIO_T_MPIO_RW ] += (MPI_Wtime() - io_time);
#endif

    /* --BEGIN ERROR HANDLING-- */
    if (err == -1) {
	*error_code = MPIO_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE,
					   myname, __LINE__, MPI_ERR_IO,
					   "**io", "**io %s", strerror(errno));
	return;
    }
    /* --END ERROR HANDLING-- */

#ifdef HAVE_STATUS_SET_BYTES
    MPIR_Status_set_bytes(status, datatype, err);
#endif

    *error_code = MPI_SUCCESS;
}



#define ADIOI_BUFFERED_READ \
{ \
    if (req_off >= readbuf_off + readbuf_len) { \
	readbuf_off = req_off; \
	readbuf_len = (int) (ADIOI_MIN(max_bufsize, end_offset-readbuf_off+1));\
	lseek(fd->fd_sys, readbuf_off, SEEK_SET);\
        if (!(fd->atomicity)) ADIOI_READ_LOCK(fd, readbuf_off, SEEK_SET, readbuf_len);\
        err = read(fd->fd_sys, readbuf, readbuf_len);\
        if (!(fd->atomicity)) ADIOI_UNLOCK(fd, readbuf_off, SEEK_SET, readbuf_len);\
        if (err == -1) err_flag = 1; \
    } \
    while (req_len > readbuf_off + readbuf_len - req_off) { \
	partial_read = (int) (readbuf_off + readbuf_len - req_off); \
	tmp_buf = (char *) ADIOI_Malloc(partial_read); \
	memcpy(tmp_buf, readbuf+readbuf_len-partial_read, partial_read); \
	ADIOI_Free(readbuf); \
	readbuf = (char *) ADIOI_Malloc(partial_read + max_bufsize); \
	memcpy(readbuf, tmp_buf, partial_read); \
	ADIOI_Free(tmp_buf); \
	readbuf_off += readbuf_len-partial_read; \
	readbuf_len = (int) (partial_read + ADIOI_MIN(max_bufsize, \
				       end_offset-readbuf_off+1)); \
	lseek(fd->fd_sys, readbuf_off+partial_read, SEEK_SET);\
        if (!(fd->atomicity)) ADIOI_READ_LOCK(fd, readbuf_off+partial_read, SEEK_SET, readbuf_len-partial_read);\
        err = read(fd->fd_sys, readbuf+partial_read, readbuf_len-partial_read);\
        if (!(fd->atomicity)) ADIOI_UNLOCK(fd, readbuf_off+partial_read, SEEK_SET, readbuf_len-partial_read);\
        if (err == -1) err_flag = 1; \
    } \
    memcpy((char *)buf + userbuf_off, readbuf+req_off-readbuf_off, req_len); \
}


void ADIOI_BGL_ReadStrided(ADIO_File fd, void *buf, int count,
                       MPI_Datatype datatype, int file_ptr_type,
                       ADIO_Offset offset, ADIO_Status *status, int
                       *error_code)
{
/* offset is in units of etype relative to the filetype. */

    ADIOI_Flatlist_node *flat_buf, *flat_file;
    int i, j, k, err=-1, brd_size, frd_size=0, st_index=0;
    int bufsize, num, size, sum, n_etypes_in_filetype, size_in_filetype;
    int n_filetypes, etype_in_filetype;
    ADIO_Offset abs_off_in_filetype=0;
    int filetype_size, etype_size, buftype_size, req_len, partial_read;
    MPI_Aint filetype_extent, buftype_extent; 
    int buf_count, buftype_is_contig, filetype_is_contig;
    ADIO_Offset userbuf_off;
    ADIO_Offset off, req_off, disp, end_offset=0, readbuf_off, start_off;
    char *readbuf, *tmp_buf, *value;
    int flag, st_frd_size, st_n_filetypes, readbuf_len;
    int new_brd_size, new_frd_size, err_flag=0, info_flag, max_bufsize;

    static char myname[] = "ADIOI_BGL_READSTRIDED";

    if (fd->hints->ds_read == ADIOI_HINT_DISABLE) {
  /* if user has disabled data sieving on reads, use naive
	 * approach instead.
	 */
      /*FPRINTF(stderr, "ADIOI_GEN_ReadStrided_naive(%d):\n", __LINE__);*/
      ADIOI_GEN_ReadStrided_naive(fd, 
				    buf,
				    count,
				    datatype,
				    file_ptr_type,
				    offset,
				    status,
				    error_code);
    	return;
    }
    /*FPRINTF(stderr, "%s(%d):\n",myname, __LINE__);*/

    ADIOI_Datatype_iscontig(datatype, &buftype_is_contig);
    ADIOI_Datatype_iscontig(fd->filetype, &filetype_is_contig);

    MPI_Type_size(fd->filetype, &filetype_size);
    if ( ! filetype_size ) {
	*error_code = MPI_SUCCESS; 
	return;
    }

    MPI_Type_extent(fd->filetype, &filetype_extent);
    MPI_Type_size(datatype, &buftype_size);
    MPI_Type_extent(datatype, &buftype_extent);
    etype_size = fd->etype_size;

    bufsize = buftype_size * count;

/* get max_bufsize from the info object. */

    value = (char *) ADIOI_Malloc((MPI_MAX_INFO_VAL+1)*sizeof(char));
    MPI_Info_get(fd->info, "ind_rd_buffer_size", MPI_MAX_INFO_VAL, value, 
                 &info_flag);
    max_bufsize = atoi(value);
    ADIOI_Free(value);

    if (!buftype_is_contig && filetype_is_contig) {

/* noncontiguous in memory, contiguous in file. */

	ADIOI_Flatten_datatype(datatype);
	flat_buf = ADIOI_Flatlist;
	while (flat_buf->type != datatype) flat_buf = flat_buf->next;

        off = (file_ptr_type == ADIO_INDIVIDUAL) ? fd->fp_ind : 
                 fd->disp + etype_size * offset;

	start_off = off;
	end_offset = off + bufsize - 1;
        readbuf_off = off;
        readbuf = (char *) ADIOI_Malloc(max_bufsize);
        readbuf_len = (int) (ADIOI_MIN(max_bufsize, end_offset-readbuf_off+1));

/* if atomicity is true, lock (exclusive) the region to be accessed */
        if (fd->atomicity)
            ADIOI_WRITE_LOCK(fd, start_off, SEEK_SET, end_offset-start_off+1);

	lseek(fd->fd_sys, readbuf_off, SEEK_SET);
        if (!(fd->atomicity)) ADIOI_READ_LOCK(fd, readbuf_off, SEEK_SET, readbuf_len);
        err = read(fd->fd_sys, readbuf, readbuf_len);
        if (!(fd->atomicity)) ADIOI_UNLOCK(fd, readbuf_off, SEEK_SET, readbuf_len);
        if (err == -1) err_flag = 1;

        for (j=0; j<count; j++) 
            for (i=0; i<flat_buf->count; i++) {
                userbuf_off = j*buftype_extent + flat_buf->indices[i];