int launch_turbine(MPI_Comm comm, char* cmd, int argc, char** argv) {
  int status = 0;
  char** argvc = (char**)malloc((argc+1)*sizeof(char*));
  int i;
  for(i=0; i<argc; i++) {
    argvc[i] = argv[i];
  }
  argvc[argc] = NULL;
  MPI_Info info;
  MPI_Info_create(&info);
  MPI_Info_set(info,"launcher","turbine");
  MPIX_Comm_launch(cmd, argvc, info, 0, comm, &status);
  MPI_Info_free(&info);
  free(argvc);
  if(comm != MPI_COMM_SELF) {
    MPI_Comm_free(&comm);
  }
  return status;
}

/* parse the file-of-hints.  Format is zero or more lines of "<key> <value>\n".
 * A # in collumn zero is a comment and the line will be ignored.  Do our best
 * to ignore badly formed lines too. 
 *
 * The caller provides an 'info' object.  Each key-value pair found by the
 * parser will get added to the info object.  any keys already set will be left
 * alone on the assumption that the caller knows best. 
 *
 * because MPI-IO hints are optional, we can get away with limited error
 * reporting.  */
static int file_to_info(int fd, MPI_Info info)
{
    char *buffer, *token, *key, *val, *garbage;
    char *pos1, *pos2;
    int flag, ret;
    char dummy;
    struct stat statbuf;

    /* assumption: config files will be small (less than 1MB) */
    fstat(fd, &statbuf);
    /* add 1 to size to make room for NULL termination */
    buffer = (char *)calloc(statbuf.st_size + 1, sizeof (char));
    if (buffer == NULL) return -1;

    ret = read(fd, buffer, statbuf.st_size);
    if (ret < 0) return -1;
    token = strtok_r(buffer, "\n", &pos1);
    do {
	if ( (key = strtok_r(token, " \t", &pos2)) == NULL) 
	    /* malformed line: found no items */
	    continue;
	if (token[0] == '#') 
	    /* ignore '#'-delimited comments */
	    continue;
	if ( (val = strtok_r(NULL, " \t", &pos2))  == NULL) 
	    /* malformed line: found key without value */
	    continue;
	if ( (garbage = strtok_r(NULL, " \t", &pos2)) != NULL) 
	    /* malformed line: more than two items */
	    continue;
	    
#ifdef SYSHINT_DEBUG
	printf("found: key=%s val=%s\n", key, val);
#endif
	/* don't actually care what the value is. only want to know if key
	 * exists: we leave it alone if so*/
	MPI_Info_get(info, key, 0, &dummy, &flag);
	if (flag == 1) continue;
	MPI_Info_set(info, key, val);
    } while ((token = strtok_r(NULL, "\n", &pos1)) != NULL);
    free(buffer);
    return 0;
}

void oshmpi_allock(MPI_Comm comm)
{
  MPI_Info lock_info=MPI_INFO_NULL;
  MPI_Info_create(&lock_info);

  /* We define the sheap size to be symmetric and assume it for the global static data. */
  MPI_Info_set(lock_info, "same_size", "true");

  MPI_Win_allocate (4 * sizeof (int), sizeof (int), lock_info,
		    comm, &oshmpi_lock_base, &oshmpi_lock_win);
  oshmpi_lock_base[NEXT_DISP] = -1;
  oshmpi_lock_base[PREV_DISP] = -1;
  oshmpi_lock_base[TAIL_DISP] = -1;
  oshmpi_lock_base[LOCK_DISP] = -1;
  MPI_Win_lock_all (TAIL, oshmpi_lock_win);

  MPI_Info_free(&lock_info);

  return;
}

long * allocate_memory (int me, MPI_Win * win)
{
	long * msg_buffer; 
	long * win_base ; /* base */
	
	MPI_Info info;
	MPI_Info_create(&info);
	MPI_Info_set(info, "same_size", "true");
	
	MPI_Alloc_mem((MAX_MSG_SZ * ITERS_LARGE) * sizeof(long), MPI_INFO_NULL, &msg_buffer);
	MPI_Win_allocate((MAX_MSG_SZ * ITERS_LARGE) * sizeof(long), sizeof(long), info, MPI_COMM_WORLD, &win_base, win);
	MPI_Win_lock_all (MPI_MODE_NOCHECK, *win);

        MPI_Info_free(&info);

	if (NULL == msg_buffer && MPI_BOTTOM == win_base) {
		fprintf(stderr, "Failed to allocate window (pe: %d)\n", me);
		exit(EXIT_FAILURE);
	}

	return msg_buffer;
}

        int MpiCommunicator::init( int minId, long thecomm_ )
        {
            VT_FUNC_I( "MpiCommunicator::init" );

            assert( sizeof(thecomm_) >= sizeof(MPI_Comm) );
            MPI_Comm thecomm = (MPI_Comm)thecomm_;

            // turn wait mode on for intel mpi if possible
            // this should greatly improve performance for intel mpi
            PAL_SetEnvVar( "I_MPI_WAIT_MODE", "enable", 0);

            int flag;
            MPI_Initialized( &flag );
            if ( ! flag ) {
                int p;
                //!! FIXME passing NULL ptr breaks mvapich1 mpi implementation
                MPI_Init_thread( 0, NULL, MPI_THREAD_MULTIPLE, &p );
                if( p != MPI_THREAD_MULTIPLE ) {
                    // can't use Speaker yet, need Channels to be inited
                    std::cerr << "[CnC] Warning: not MPI_THREAD_MULTIPLE (" << MPI_THREAD_MULTIPLE << "), but " << p << std::endl;
                }
            } else if( thecomm == 0 ) {
                CNC_ABORT( "Process has already been initialized" );
            }


            MPI_Comm myComm = MPI_COMM_WORLD;
            int rank;
            MPI_Comm parentComm;
            if( thecomm == 0 ) {
                MPI_Comm_get_parent( &parentComm );
            } else {
                m_customComm = true;
                m_exit0CallOk = false;
                myComm = thecomm;
            }
            MPI_Comm_rank( myComm, &rank );
            
            // father of all checks if he's requested to spawn processes:
            if ( rank == 0 && parentComm == MPI_COMM_NULL ) {
                // Ok, let's spawn the clients.
                // I need some information for the startup.
                // 1. Name of the executable (default is the current exe)
                const char * _tmp = getenv( "CNC_MPI_SPAWN" );
                if ( _tmp ) {
                    int nClientsToSpawn = atol( _tmp );
                    _tmp = getenv( "CNC_MPI_EXECUTABLE" );
                    std::string clientExe( _tmp ? _tmp : "" );
                    if( clientExe.empty() ) clientExe = PAL_GetProgname();
                    CNC_ASSERT( ! clientExe.empty() );
                    // 3. Special setting for MPI_Info: hosts
                    const char * clientHost = getenv( "CNC_MPI_HOSTS" );
                    
                    // Prepare MPI_Info object:
                    MPI_Info clientInfo = MPI_INFO_NULL;
                    if ( clientHost ) {
                        MPI_Info_create( &clientInfo );
                        if ( clientHost ) {
                            MPI_Info_set( clientInfo, const_cast< char * >( "host" ), const_cast< char * >( clientHost ) );
                            // can't use Speaker yet, need Channels to be inited
                            std::cerr << "[CnC " << rank << "] Set MPI_Info_set( \"host\", \"" << clientHost << "\" )\n";
                        }
                    }
                    // Now spawn the client processes:
                    // can't use Speaker yet, need Channels to be inited
                    std::cerr << "[CnC " << rank << "] Spawning " << nClientsToSpawn << " MPI processes" << std::endl;
                    int* errCodes = new int[nClientsToSpawn];
                    MPI_Comm interComm;
                    int err = MPI_Comm_spawn( const_cast< char * >( clientExe.c_str() ),
                                              MPI_ARGV_NULL, nClientsToSpawn,
                                              clientInfo, 0, MPI_COMM_WORLD,
                                              &interComm, errCodes );
                    delete [] errCodes;
                    if ( err ) {
                        // can't use Speaker yet, need Channels to be inited
                        std::cerr << "[CnC " << rank << "] Error in MPI_Comm_spawn. Skipping process spawning";
                    } else {
                        MPI_Intercomm_merge( interComm, 0, &myComm );
                    }
                } // else {
                // No process spawning
                // MPI-1 situation: all clients to be started by mpiexec
                //                    myComm = MPI_COMM_WORLD;
                //}
            }
            if ( thecomm == 0 && parentComm != MPI_COMM_NULL ) {
                // I am a child. Build intra-comm to the parent.
                MPI_Intercomm_merge( parentComm, 1, &myComm );
            }
            MPI_Comm_rank( myComm, &rank );

            CNC_ASSERT( m_channel == NULL );
            MpiChannelInterface* myChannel = new MpiChannelInterface( use_crc(), myComm );
            m_channel = myChannel;

            int size;
            MPI_Comm_size( myComm, &size );
            // Are we on the host or on the remote side?
            if ( rank == 0 ) {
                if( size <= 1 ) {
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		assert(y != NULL);
		z = y + 3;
		while (*z && !isspace((unsigned char)*z)) z++;
		chrNames[nbchr++] = strndup(y + 3, z - y - 3);
		assert(nbchr < MAXNBCHR - 2);
	}
	chrNames[nbchr++] = strdup(UNMAPPED);
	chrNames[nbchr++] = strdup(DISCORDANT);

	hsiz = x - xbuf;
	hbuf = strndup(xbuf, hsiz);

	if (rank == 0) {
		fprintf(stderr, "The size of the file is %zu bytes\n", (size_t)st.st_size);
		fprintf(stderr, "Header has %d+2 references\n", nbchr - 2);
	}
	asprintf(&header, "@HD\tVN:1.0\tSO:%s\n%s", sort_name, hbuf);

	free(hbuf);

	assert(munmap(xbuf, (size_t)st.st_size) != -1);
	assert(close(fd) != -1);

	//task FIRST FINE TUNING FINFO FOR READING OPERATIONS


	MPI_Info_create(&finfo);
	/*
	 * In this part you shall adjust the striping factor and unit according
	 * to the underlying filesystem.
	 * Harmless for other file system.
	 *
	 */
	MPI_Info_set(finfo,"striping_factor", STRIPING_FACTOR);
	MPI_Info_set(finfo,"striping_unit", STRIPING_UNIT); //2G striping
	MPI_Info_set(finfo,"ind_rd_buffer_size", STRIPING_UNIT); //2gb buffer
	MPI_Info_set(finfo,"romio_ds_read",DATA_SIEVING_READ);

	/*
	 * for collective reading and writing
	 * should be adapted too and tested according to the file system
	 * Harmless for other file system.
	 */
	MPI_Info_set(finfo,"nb_proc", NB_PROC);
	MPI_Info_set(finfo,"cb_nodes", CB_NODES);
	MPI_Info_set(finfo,"cb_block_size", CB_BLOCK_SIZE);
	MPI_Info_set(finfo,"cb_buffer_size", CB_BUFFER_SIZE);


	//we open the input file
	ierr = MPI_File_open(MPI_COMM_WORLD, file_name,  MPI_MODE_RDONLY , finfo, &mpi_filed);
	//assert(in != -1);
	if (ierr){
		if (rank == 0) fprintf(stderr, "%s: Failed to open file in process 0 %s\n", argv[0], argv[1]);
		MPI_Abort(MPI_COMM_WORLD, errorcode);
		exit(2);
	}
	ierr = MPI_File_get_size(mpi_filed, &fileSize);
	assert(ierr == MPI_SUCCESS);
	input_file_size = (long long)fileSize;

	/* Get chunk offset and size */
	fsiz = input_file_size;
	lsiz = fsiz / num_proc;
	loff = rank * lsiz;

int main(int argc, char *argv[])
{
  int i, j, k, length, my_rank, left, right, size, test_value, mid;    
  double start, finish, transfer_time; 
  float snd_buf_left[max_length], snd_buf_right[max_length];
  float *rcv_buf_left, *rcv_buf_right;
  float *rcv_buf_left_neighbor, *rcv_buf_right_neighbor;

  MPI_Win win_rcv_buf_left, win_rcv_buf_right;
  MPI_Info info_noncontig;
  MPI_Aint buf_size;
  int disp_unit;
  

/* Naming conventions                                                                */
/* Processes:                                                                        */
/*     my_rank-1                        my_rank                         my_rank+1    */
/* "left neighbor"                     "myself"                     "right neighbor" */
/*   ...    rcv_buf_right <--- snd_buf_left snd_buf_right ---> rcv_buf_left    ...   */
/*   ... snd_buf_right ---> rcv_buf_left       rcv_buf_right <--- snd_buf_left ...   */
/*                        |                                  |                       */
/*              halo-communication                 halo-communication                */

  MPI_Init(&argc, &argv);
  MPI_Comm_rank(MPI_COMM_WORLD, &my_rank);
  MPI_Comm_size(MPI_COMM_WORLD, &size);
  right = (my_rank+1)      % size;
  left  = (my_rank-1+size) % size;

  MPI_Info_create(&info_noncontig);
  MPI_Info_set(info_noncontig, "alloc_shared_noncontig", "true");
  MPI_Win_allocate_shared((MPI_Aint)(max_length*sizeof(float)), sizeof(float), info_noncontig, MPI_COMM_WORLD, &rcv_buf_left,  &win_rcv_buf_left );
  MPI_Win_allocate_shared((MPI_Aint)(max_length*sizeof(float)), sizeof(float), info_noncontig, MPI_COMM_WORLD, &rcv_buf_right, &win_rcv_buf_right);

/*... shared memory access to the rcv_buf_left, of the RIGHT neighbor process */
  MPI_Win_shared_query(win_rcv_buf_left,  right, &buf_size, &disp_unit, &rcv_buf_left_neighbor );

/*... shared memory access to the rcv_buf_right, of the LEFT neighbor process */
  MPI_Win_shared_query(win_rcv_buf_right, left,  &buf_size, &disp_unit, &rcv_buf_right_neighbor);


  if (my_rank == 0) printf("    message size      transfertime  duplex bandwidth per process and neighbor\n");

  length = start_length;

  for (j = 1; j <= number_package_sizes; j++)
  { 
    
    for (i = 0; i <= number_of_messages; i++)
    {
      if(i==1) start = MPI_Wtime();

      test_value = j*1000000 + i*10000 + my_rank*10 ; mid = (length-1)/number_of_messages*i;

      snd_buf_left[0]=test_value+1  ; snd_buf_left[mid]=test_value+2  ; snd_buf_left[length-1]=test_value+3;
      snd_buf_right[0]=test_value+6 ; snd_buf_right[mid]=test_value+7 ; snd_buf_right[length-1]=test_value+8;

/*    MPI_Win_fence(MPI_MODE_NOSTORE + MPI_MODE_NOPRECEDE, win_rcv_buf_left ); */
/*    MPI_Win_fence(MPI_MODE_NOSTORE + MPI_MODE_NOPRECEDE, win_rcv_buf_right); */
/*      ... instead of above, work-around for a bug with shared memory windows in some libraries: */
      MPI_Win_fence(MPI_MODE_NOSTORE, win_rcv_buf_left );
      MPI_Win_fence(MPI_MODE_NOSTORE, win_rcv_buf_right);

/*    MPI_Put(snd_buf_left,  length, MPI_FLOAT, left,  (MPI_Aint)0, length, MPI_FLOAT, win_rcv_buf_right); */
/*    MPI_Put(snd_buf_right, length, MPI_FLOAT, right, (MPI_Aint)0, length, MPI_FLOAT, win_rcv_buf_left ); */
/*      ... is substited by: */
      for(k=0; k<length; k++) rcv_buf_right_neighbor[k]  = snd_buf_left [k];
      for(k=0; k<length; k++) rcv_buf_left_neighbor [k] = snd_buf_right[k];

/*    MPI_Win_fence(MPI_MODE_NOSTORE + MPI_MODE_NOPUT + MPI_MODE_NOSUCCEED, win_rcv_buf_left ); */
/*    MPI_Win_fence(MPI_MODE_NOSTORE + MPI_MODE_NOPUT + MPI_MODE_NOSUCCEED, win_rcv_buf_right); */
/*      ... instead of above, work-around for a bug with shared memory windows in some libraries: */
      MPI_Win_fence(MPI_MODE_NOSTORE + MPI_MODE_NOPUT, win_rcv_buf_left );
      MPI_Win_fence(MPI_MODE_NOSTORE + MPI_MODE_NOPUT, win_rcv_buf_right);

/*    ...snd_buf_... is used to store the values that were stored in snd_buf_... in the neighbor process */
      test_value = j*1000000 + i*10000 + left*10  ; mid = (length-1)/number_of_messages*i;
      snd_buf_right[0]=test_value+6 ; snd_buf_right[mid]=test_value+7 ; snd_buf_right[length-1]=test_value+8;
      test_value = j*1000000 + i*10000 + right*10 ; mid = (length-1)/number_of_messages*i;
      snd_buf_left[0]=test_value+1  ; snd_buf_left[mid]=test_value+2  ; snd_buf_left[length-1]=test_value+3;
      if ((rcv_buf_left[0] != snd_buf_right[0]) || (rcv_buf_left[mid] != snd_buf_right[mid]) || 
                                                   (rcv_buf_left[length-1] != snd_buf_right[length-1])) {
         printf("%d: j=%d, i=%d --> snd_buf_right[0,%d,%d]=(%f,%f,%f)\n",
                    my_rank, j, i, mid, length-1, snd_buf_right[0], snd_buf_right[mid], snd_buf_right[length-1]);
         printf("%d:     is not identical to rcv_buf_left[0,%d,%d]=(%f,%f,%f)\n",
                    my_rank,       mid, length-1, rcv_buf_left[0],  rcv_buf_left[mid],  rcv_buf_left[length-1]);
      }
      if ((rcv_buf_right[0] != snd_buf_left[0]) || (rcv_buf_right[mid] != snd_buf_left[mid]) ||
                                                   (rcv_buf_right[length-1] != snd_buf_left[length-1])) {
         printf("%d: j=%d, i=%d --> snd_buf_left[0,%d,%d]=(%f,%f,%f)\n",
                    my_rank, j, i, mid, length-1, snd_buf_left[0],  snd_buf_left[mid],  snd_buf_left[length-1]);
         printf("%d:     is not identical to rcv_buf_right[0,%d,%d]=(%f,%f,%f)\n",
                    my_rank,       mid, length-1, rcv_buf_right[0], rcv_buf_right[mid], rcv_buf_right[length-1]);
      }

    }
    finish = MPI_Wtime();

    if (my_rank == 0) 
    {
//.........这里部分代码省略.........

int main(int argc, char** argv)
{
    char filename[256];
    int i, j, rank, nprocs, err, nerrs=0, expected;
    int ncid, cmode, varid[2], dimid[2], req[4], st[4], *buf;
    int *buf0, *buf1, *buf2;
    size_t len;
    MPI_Offset start[2], count[2];
    MPI_Info info;

    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &nprocs);

    /* this program is intended to run on one process */
    if (rank) goto fn_exit;

    /* get command-line arguments */
    if (argc > 2) {
        if (!rank) printf("Usage: %s [filename]\n",argv[0]);
        MPI_Finalize();
        return 1;
    }
    if (argc == 2) snprintf(filename, 256, "%s", argv[1]);
    else           strcpy(filename, "testfile.nc");

    if (rank == 0) {
        char *cmd_str = (char*)malloc(strlen(argv[0]) + 256);
        sprintf(cmd_str, "*** TESTING C   %s for writing interleaved fileviews ", basename(argv[0]));
        printf("%-66s ------ ", cmd_str);
        free(cmd_str);
    }

    MPI_Info_create(&info);
    MPI_Info_set(info, "romio_cb_write", "disable");
    MPI_Info_set(info, "ind_wr_buffer_size", "8");
    /* these 2 hints are required to cause a core dump if r1758 fix is not
     * presented */

    /* create a new file for writing ----------------------------------------*/
    cmode = NC_CLOBBER | NC_64BIT_DATA;
    err = ncmpi_create(MPI_COMM_SELF, filename, cmode, info, &ncid); CHECK_ERR

    MPI_Info_free(&info);

    /* define dimensions Y and X */
    err = ncmpi_def_dim(ncid, "Y", NY, &dimid[0]); CHECK_ERR
    err = ncmpi_def_dim(ncid, "X", NX, &dimid[1]); CHECK_ERR

    /* define 2D variables of integer type */
    err = ncmpi_def_var(ncid, "var0", NC_INT, 2, dimid, &varid[0]); CHECK_ERR
    err = ncmpi_def_var(ncid, "var1", NC_INT, 2, dimid, &varid[1]); CHECK_ERR

    /* enable fill mode */
    err = ncmpi_set_fill(ncid, NC_FILL, NULL); CHECK_ERR

    /* do not forget to exit define mode */
    err = ncmpi_enddef(ncid); CHECK_ERR

    /* now we are in data mode */
    buf = (int*) malloc(NY*NX * sizeof(int));

    /* fill the entire variable var0 with -1s */
    for (i=0; i<NY*NX; i++) buf[i] = -1;
    err = ncmpi_put_var_int_all(ncid, varid[0], buf); CHECK_ERR

    /* write 8 x 2 elements so this only interleaves the next two
     * iput requests */
    start[0] = 0; start[1] = 3;
    count[0] = 8; count[1] = 2;
    len = (size_t)(count[0] * count[1]);
    buf0 = (int*) malloc(len * sizeof(int));
    for (i=0; i<len; i++) buf0[i] = 50+i;
    err = ncmpi_iput_vara_int(ncid, varid[0], start, count, buf0, &req[0]);
    CHECK_ERR

    /* write 1 x 3 elements */
    start[0] = 1; start[1] = 8;
    count[0] = 1; count[1] = 5;
    len = (size_t)(count[0] * count[1]);
    buf1 = (int*) malloc(len * sizeof(int));
    for (i=0; i<len; i++) buf1[i] = 60+i;
    err = ncmpi_iput_vara_int(ncid, varid[0], start, count, buf1, &req[1]);
    CHECK_ERR

    /* write 1 x 3 elements */
    start[0] = 3; start[1] = 7;
    count[0] = 1; count[1] = 5;
    len = (size_t)(count[0] * count[1]);
    buf2 = (int*) malloc(len * sizeof(int));
    for (i=0; i<len; i++) buf2[i] = 70+i;
    err = ncmpi_iput_vara_int(ncid, varid[0], start, count, buf2, &req[2]);
    CHECK_ERR

    err = ncmpi_wait_all(ncid, 3, req, st); CHECK_ERR
    free(buf0); free(buf1); free(buf2);

    /* fill the entire variable var1 with -1s */
    for (i=0; i<NY*NX; i++) buf[i] = -1;
    err = ncmpi_put_var_int_all(ncid, varid[1], buf); CHECK_ERR
//.........这里部分代码省略.........

int test_file(char *filename, int mynod, int nprocs, char * cb_hosts, const char *msg, int verbose)
{
    MPI_Datatype typevec, newtype, t[3];
    int *buf, i, b[3], errcode, errors=0;
    MPI_File fh;
    MPI_Aint d[3];
    MPI_Status status;
    int SIZE = (STARTING_SIZE/nprocs)*nprocs;
    MPI_Info info;

    if (mynod==0 && verbose) fprintf(stderr, "%s\n", msg);

    buf = (int *) malloc(SIZE*sizeof(int));
    if (buf == NULL) {
	    perror("test_file");
	    MPI_Abort(MPI_COMM_WORLD, -1);
    }


    if (cb_hosts != NULL ) {
	    MPI_Info_create(&info);
	    MPI_Info_set(info, "cb_config_list", cb_hosts);
    } else {
	    info = MPI_INFO_NULL;
    }

    MPI_Type_vector(SIZE/nprocs, 1, nprocs, MPI_INT, &typevec);

    b[0] = b[1] = b[2] = 1;
    d[0] = 0;
    d[1] = mynod*sizeof(int);
    d[2] = SIZE*sizeof(int);
    t[0] = MPI_LB;
    t[1] = typevec;
    t[2] = MPI_UB;

    MPI_Type_struct(3, b, d, t, &newtype);
    MPI_Type_commit(&newtype);
    MPI_Type_free(&typevec);

    if (!mynod) {
	if(verbose) fprintf(stderr, "\ntesting noncontiguous in memory, noncontiguous in file using collective I/O\n");
	MPI_File_delete(filename, info);
    }
    MPI_Barrier(MPI_COMM_WORLD);

    errcode = MPI_File_open(MPI_COMM_WORLD, filename,
		    MPI_MODE_CREATE | MPI_MODE_RDWR, info, &fh);
    if (errcode != MPI_SUCCESS) {
	    handle_error(errcode, "MPI_File_open");
    }

    MPI_File_set_view(fh, 0, MPI_INT, newtype, "native", info);

    for (i=0; i<SIZE; i++) buf[i] = SEEDER(mynod,i,SIZE);
    errcode = MPI_File_write_all(fh, buf, 1, newtype, &status);
    if (errcode != MPI_SUCCESS) {
	    handle_error(errcode, "nc mem - nc file: MPI_File_write_all");
    }

    MPI_Barrier(MPI_COMM_WORLD);

    for (i=0; i<SIZE; i++) buf[i] = -1;

    errcode = MPI_File_read_at_all(fh, 0, buf, 1, newtype, &status);
    if (errcode != MPI_SUCCESS) {
	    handle_error(errcode, "nc mem - nc file: MPI_File_read_at_all");
    }

    /* the verification for N compute nodes is tricky. Say we have 3
     * processors.
     * process 0 sees: 0 -1 -1 3 -1 -1 ...
     * process 1 sees: -1 34 -1 -1 37 -1 ...
     * process 2 sees: -1 -1 68 -1 -1 71 ... */

    /* verify those leading -1s exist if they should */
    for (i=0; i<mynod; i++ ) {
	    if ( buf[i] != -1 ) {
		    if(verbose) fprintf(stderr, "Process %d: buf is %d, should be -1\n", mynod, buf[i]);
		    errors++;
	    }
    }
    /* now the modulo games are hairy.  processor 0 sees real data in the 0th,
     * 3rd, 6th... elements of the buffer (assuming nprocs==3 ).  proc 1 sees
     * the data in 1st, 4th, 7th..., and proc 2 sees it in 2nd, 5th, 8th */

    for(/* 'i' set in above loop */; i<SIZE; i++) {
	    if ( ((i-mynod)%nprocs) && buf[i] != -1)  {
		    if(verbose) fprintf(stderr, "Process %d: buf %d is %d, should be -1\n",
				    mynod, i, buf[i]);
		    errors++;
	    }
	    if ( !((i-mynod)%nprocs) && buf[i] != SEEDER(mynod,i,SIZE) ) {
		    if(verbose) fprintf(stderr, "Process %d: buf %d is %d, should be %d\n",
				    mynod, i, buf[i], SEEDER(mynod,i,SIZE));
		    errors++;
	    }
    }
    MPI_File_close(&fh);

//.........这里部分代码省略.........

void do_collective_read()
{
  MPI_Info info;
  MPI_Datatype contig;
  MPI_Comm sub_read_comm;
  MPI_File fh;
  char coll_path[PATH_LEN];
  int sub_comm_size, sub_rank, sub_comm_color;
  int disp;
  int rc;
  int *buf;

  ptimes[0].start = MPI_Wtime();
  ptimes[1].start = MPI_Wtime();
  sub_comm_color = get_sub_collective_io_comm(&sub_read_comm);  

  /* Construct a datatype for distributing the input data across all
   * processes. */
  MPI_Type_contiguous(data_size / sizeof(int), MPI_INT, &contig);
  MPI_Type_commit(&contig);
  
  /* Set the stripe_count and stripe_size, that is, the striping_factor                                                                                                                                    
   * and striping_unit. Both keys and values for MPI_Info_set must be                                                                                                                                      
   * in the form of ascii strings. */
  MPI_Info_create(&info);
  //  MPI_Info_set(info, "striping_factor", striping_factor);
  //  MPI_Info_set(info, "striping_unit", striping_unit);
  MPI_Info_set(info, "romio_cb_read", "enable");                                                                                                                                                       
  //  MPI_Info_set(info, "romio_cb_read", "disable");

  /* Get path to the target file of the communicator */
  MPI_Comm_size(sub_read_comm, &sub_comm_size);
  get_coll_io_path(coll_path, sub_comm_color);

  /* Delete the output file if it exists so that striping can be set                                                                                                                                          * on the output file. */
  //  rc = MPI_File_delete(coll_path, info);

  /* Create read data*/
  MPI_Comm_rank(sub_read_comm, &sub_rank);
  buf = create_io_data(-1);
  ptimes[1].end = MPI_Wtime();

  MPI_Barrier(MPI_COMM_WORLD);

  /* Open the file */
  ptimes[2].start = MPI_Wtime();
  rc = MPI_File_open(sub_read_comm, coll_path, 
		     MPI_MODE_RDONLY, 
		     info, &fh);
  if (rc != MPI_SUCCESS) {
    gio_err("MPI_File_open failed: %s  (%s:%s:%d)", coll_path, __FILE__, __func__, __LINE__);
  }
  ptimes[2].end   = MPI_Wtime();

  /* Set the file view for the output file. In this example, we will                                                                                                                                          * use the same contiguous datatype as we used for reading the data                                                                                                                                          * into local memory. A better example would be to read out just                                                                                                                                            * part of the data, say 4 contiguous elements followed by a gap of                                                                                                                                          * 4 elements, and repeated. */
  ptimes[3].start = MPI_Wtime();
#ifdef GIO_LARGE_FILE
  int i;
  for (i = 0; i < sub_rank; i++) {
    MPI_File_seek(fh, data_size, MPI_SEEK_CUR);
  }
#else  
  disp = sub_rank * data_size;
  MPI_File_set_view(fh, disp, contig, contig, "native", info);
#endif
  if (rc != MPI_SUCCESS) {
    gio_err("MPI_File_set_view failed  (%s:%s:%d)", __FILE__, __func__, __LINE__);
  }
  ptimes[3].end = MPI_Wtime();

  /* MPI Collective Read */
  ptimes[4].start = MPI_Wtime();
  rc = MPI_File_read_all(fh, buf, 1, contig, MPI_STATUS_IGNORE);
  if (rc != MPI_SUCCESS) {
    gio_err("MPI_File_set_view failed  (%s:%s:%d)", __FILE__, __func__, __LINE__);
  }
  ptimes[4].end = MPI_Wtime();

  validate_io_data(buf, sub_rank);

  /*Free data*/
  free_io_data(buf);

  /* Close Files */
  ptimes[5].start = MPI_Wtime();
  MPI_File_close(&fh);
  ptimes[5].end = MPI_Wtime();
  ptimes[0].end = MPI_Wtime();

  print_results();

  return;
}

void ADIOI_PIOFS_Open(ADIO_File fd, int *error_code)
{
    int amode, perm, old_mask, err;
    piofs_fstat_t piofs_fstat;
    char *value;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_PIOFS_OPEN";
#endif

    if (fd->perm == ADIO_PERM_NULL) {
	old_mask = umask(022);
	umask(old_mask);
	perm = old_mask ^ 0666;
    }
    else perm = fd->perm;

    amode = 0;
    if (fd->access_mode & ADIO_CREATE)
	amode = amode | O_CREAT;
    if (fd->access_mode & ADIO_RDONLY)
	amode = amode | O_RDONLY;
    if (fd->access_mode & ADIO_WRONLY)
	amode = amode | O_WRONLY;
    if (fd->access_mode & ADIO_RDWR)
	amode = amode | O_RDWR;
    if (fd->access_mode & ADIO_EXCL)
	amode = amode | O_EXCL;

#ifdef PROFILE
    MPE_Log_event(1, 0, "start open");
#endif
    fd->fd_sys = open(fd->filename, amode, perm);
#ifdef PROFILE
    MPE_Log_event(2, 0, "end open");
#endif

    llseek(fd->fd_sys, 0, SEEK_SET);
/* required to initiate use of 64-bit offset */

    if (fd->fd_sys != -1) {
	value = (char *) ADIOI_Malloc((MPI_MAX_INFO_VAL+1)*sizeof(char));

        /* get file striping information and set it in info */
	err = piofsioctl(fd->fd_sys, PIOFS_FSTAT, &piofs_fstat);

	if (!err) {
	    sprintf(value, "%d", piofs_fstat.st_bsu);
	    MPI_Info_set(fd->info, "striping_unit", value);

	    sprintf(value, "%d", piofs_fstat.st_cells);
	    MPI_Info_set(fd->info, "striping_factor", value);

	    sprintf(value, "%d", piofs_fstat.st_base_node);
	    MPI_Info_set(fd->info, "start_iodevice", value);
	}
	ADIOI_Free(value);

	if (fd->access_mode & ADIO_APPEND)
	    fd->fp_ind = fd->fp_sys_posn = llseek(fd->fd_sys, 0, SEEK_END);
    }

#ifdef PRINT_ERR_MSG
    *error_code = (fd->fd_sys == -1) ? MPI_ERR_UNKNOWN : MPI_SUCCESS;
#else
    if (fd->fd_sys == -1) {
	*error_code = MPIR_Err_setmsg(MPI_ERR_IO, MPIR_ADIO_ERROR,
			      myname, "I/O Error", "%s", strerror(errno));
	ADIOI_Error(ADIO_FILE_NULL, *error_code, myname);	    
    }
    else *error_code = MPI_SUCCESS;
#endif
}

/* This test spawns two child jobs and has them open a port and connect to
 * each other.
 * The two children repeatedly connect, accept, and disconnect from each other.
 */
int main(int argc, char *argv[])
{
    int error;
    int rank, size;
    int numprocs = 3;
    char *argv1[2] = { (char *) "connector", NULL };
    char *argv2[2] = { (char *) "acceptor", NULL };
    MPI_Comm comm_connector, comm_acceptor, comm_parent, comm;
    char port[MPI_MAX_PORT_NAME] = { 0 };
    MPI_Status status;
    MPI_Info spawn_path = MPI_INFO_NULL;
    int i, num_loops = 100;
    int data;
    int verbose = 0;
    int can_spawn, errs = 0;

    if (getenv("MPITEST_VERBOSE")) {
        verbose = 1;
    }

    IF_VERBOSE(("init.\n"));
    error = MPI_Init(&argc, &argv);
    check_error(error, "MPI_Init");

    errs += MTestSpawnPossible(&can_spawn);
    if (!can_spawn) {
        if (errs)
            printf(" Found %d errors\n", errs);
        else
            printf(" No Errors\n");
        fflush(stdout);
    } else {
        IF_VERBOSE(("size.\n"));
        error = MPI_Comm_size(MPI_COMM_WORLD, &size);
        check_error(error, "MPI_Comm_size");

        IF_VERBOSE(("rank.\n"));
        error = MPI_Comm_rank(MPI_COMM_WORLD, &rank);
        check_error(error, "MPI_Comm_rank");

        if (argc == 1) {
            /* Make sure that the current directory is in the path.
             * Not all implementations may honor or understand this, but
             * it is highly recommended as it gives users a clean way
             * to specify the location of the executable without
             * specifying a particular directory format (e.g., this
             * should work with both Windows and Unix implementations) */
            MPI_Info_create(&spawn_path);
            MPI_Info_set(spawn_path, (char *) "path", (char *) ".");

            IF_VERBOSE(("spawn connector.\n"));
            error = MPI_Comm_spawn((char *) "disconnect_reconnect2",
                                   argv1, numprocs, spawn_path, 0,
                                   MPI_COMM_WORLD, &comm_connector, MPI_ERRCODES_IGNORE);
            check_error(error, "MPI_Comm_spawn");

            IF_VERBOSE(("spawn acceptor.\n"));
            error = MPI_Comm_spawn((char *) "disconnect_reconnect2",
                                   argv2, numprocs, spawn_path, 0,
                                   MPI_COMM_WORLD, &comm_acceptor, MPI_ERRCODES_IGNORE);
            check_error(error, "MPI_Comm_spawn");
            MPI_Info_free(&spawn_path);

            if (rank == 0) {
                IF_VERBOSE(("recv port.\n"));
                error = MPI_Recv(port, MPI_MAX_PORT_NAME, MPI_CHAR, 0, 0, comm_acceptor, &status);
                check_error(error, "MPI_Recv");

                IF_VERBOSE(("send port.\n"));
                error = MPI_Send(port, MPI_MAX_PORT_NAME, MPI_CHAR, 0, 0, comm_connector);
                check_error(error, "MPI_Send");
            }

            IF_VERBOSE(("barrier acceptor.\n"));
            error = MPI_Barrier(comm_acceptor);
            check_error(error, "MPI_Barrier");

            IF_VERBOSE(("barrier connector.\n"));
            error = MPI_Barrier(comm_connector);
            check_error(error, "MPI_Barrier");

            error = MPI_Comm_free(&comm_acceptor);
            check_error(error, "MPI_Comm_free");
            error = MPI_Comm_free(&comm_connector);
            check_error(error, "MPI_Comm_free");

            if (rank == 0) {
                printf(" No Errors\n");
                fflush(stdout);
            }
        } else if ((argc == 2) && (strcmp(argv[1], "acceptor") == 0)) {
            IF_VERBOSE(("get_parent.\n"));
            error = MPI_Comm_get_parent(&comm_parent);
            check_error(error, "MPI_Comm_get_parent");
            if (comm_parent == MPI_COMM_NULL) {
                printf("acceptor's parent is NULL.\n");
//.........这里部分代码省略.........

int main(int argc, char ** argv) {
	int nprocs, mynod, errcode;
	options my_options = {NULL, 0, 0};
	MPI_File fh;
	MPI_Status status;
	MPI_Info  info;

	MPI_Init(&argc, &argv);
	MPI_Comm_size(MPI_COMM_WORLD, &nprocs);
	MPI_Comm_rank(MPI_COMM_WORLD, &mynod);

	parse_args(argc, argv, mynod, &my_options);

	if (my_options.do_aggregation) {
		MPI_Info_create(&info);
		MPI_Info_set(info, "romio_no_indep_rw", "true");
		MPI_Info_set(info, "cb_config_list", "leela.mcs.anl.gov:1");
	} else {
		info = MPI_INFO_NULL;
	}

	/* create the file w/o EXCL: this must not fail */
	errcode = MPI_File_open(MPI_COMM_WORLD, my_options.fname,
			MPI_MODE_CREATE|MPI_MODE_RDWR, info, &fh);
	if (errcode != MPI_SUCCESS) {
		handle_error(errcode, "MPI_File_open");
	}

	errcode = MPI_File_close(&fh);
	if (errcode != MPI_SUCCESS) {
		handle_error(errcode, "MPI_File_close");
	}

	/* now try to open w/ CREAT|EXCL: this must fail */
	errcode = MPI_File_open(MPI_COMM_WORLD, my_options.fname,
			MPI_MODE_CREATE|MPI_MODE_EXCL|MPI_MODE_RDWR, info, &fh);
	if (errcode == MPI_SUCCESS) {
		handle_error(errcode, "MPI_File_open: expected an error: got");
	}

	/* ignore the error: File_delete is not aggregator-aware */
	MPI_File_delete(my_options.fname, info);

	/* this must succeed: the file no longer exists */
	errcode = MPI_File_open(MPI_COMM_WORLD, my_options.fname,
			MPI_MODE_CREATE|MPI_MODE_EXCL|MPI_MODE_RDWR, info, &fh);
	if (errcode != MPI_SUCCESS) {
		handle_error(errcode, "MPI_File_open");
	}

	errcode = MPI_File_close(&fh);
	if (errcode != MPI_SUCCESS) {
		handle_error(errcode, "MPI_File_close");
	}

	if (mynod == 0) {
		printf(" No Errors\n");
	}

	MPI_Finalize();
	return 0;
}

int main(int argc, char **argv)
{
    int i, len, nkeys, flag, mynod, default_striping_factor, nprocs;
    MPI_File fh;
    MPI_Info info, info_used;
    char *filename, key[MPI_MAX_INFO_KEY], value[MPI_MAX_INFO_VAL];

    MPI_Init(&argc,&argv);

    MPI_Comm_rank(MPI_COMM_WORLD, &mynod);
    MPI_Comm_size(MPI_COMM_WORLD, &nprocs);

/* process 0 takes the file name as a command-line argument and 
   broadcasts it to other processes */
    if (!mynod) {
	i = 1;
	while ((i < argc) && strcmp("-fname", *argv)) {
	    i++;
	    argv++;
	}
	if (i >= argc) {
	    printf("\n*#  Usage: file_info -fname filename\n\n");
	    MPI_Abort(MPI_COMM_WORLD, 1);
	}
	argv++;
	len = strlen(*argv);
	filename = (char *) malloc(len+1);
	strcpy(filename, *argv);
	MPI_Bcast(&len, 1, MPI_INT, 0, MPI_COMM_WORLD);
	MPI_Bcast(filename, len+1, MPI_CHAR, 0, MPI_COMM_WORLD);
    }
    else {
	MPI_Bcast(&len, 1, MPI_INT, 0, MPI_COMM_WORLD);
	filename = (char *) malloc(len+1);
	MPI_Bcast(filename, len+1, MPI_CHAR, 0, MPI_COMM_WORLD);
    }

/* open the file with MPI_INFO_NULL */
    MPI_File_open(MPI_COMM_WORLD, filename, MPI_MODE_CREATE | MPI_MODE_RDWR, 
                  MPI_INFO_NULL, &fh);

/* check the default values set by ROMIO */
    MPI_File_get_info(fh, &info_used);
    MPI_Info_get_nkeys(info_used, &nkeys);

    for (i=0; i<nkeys; i++) {
	MPI_Info_get_nthkey(info_used, i, key);
	MPI_Info_get(info_used, key, MPI_MAX_INFO_VAL-1, value, &flag);
	if (!mynod) 
	    printf("Process %d, Default:  key = %s, value = %s\n", mynod, 
                key, value);
	if (!strcmp("striping_factor", key))
	  default_striping_factor = atoi(value);
    }

    MPI_File_close(&fh);

/* delete the file */
    if (!mynod) MPI_File_delete(filename, MPI_INFO_NULL);
    MPI_Barrier(MPI_COMM_WORLD);

/* set new info values. */

    MPI_Info_create(&info);

/* The following four hints are accepted on all machines. They can
   be specified at file-open time or later (any number of times). */

    /* buffer size for collective I/O */
    MPI_Info_set(info, "cb_buffer_size", "8388608");

    /* number of processes that actually perform I/O in collective I/O */
    sprintf(value, "%d", nprocs/2);
    MPI_Info_set(info, "cb_nodes", value);

    /* buffer size for data sieving in independent reads */
    MPI_Info_set(info, "ind_rd_buffer_size", "2097152");

    /* buffer size for data sieving in independent writes */
    MPI_Info_set(info, "ind_wr_buffer_size", "1048576");


/* The following three hints related to file striping are accepted only 
   on Intel PFS and IBM PIOFS file systems and are ignored elsewhere. 
   They can be specified only at file-creation time; if specified later 
   they will be ignored. */

    /* number of I/O devices across which the file will be striped.
       accepted only if 0 < value < default_striping_factor; 
       ignored otherwise */
    sprintf(value, "%d", default_striping_factor-1);
    MPI_Info_set(info, "striping_factor", value);

    /* the striping unit in bytes */
    MPI_Info_set(info, "striping_unit", "131072");

    /* the I/O device number from which to start striping the file.
       accepted only if 0 <= value < default_striping_factor; 
       ignored otherwise */
    sprintf(value, "%d", default_striping_factor-2);
//.........这里部分代码省略.........

/*
The test write a NP * NP matrix M, NP is the number of process:
put_vara:
Process N write N copy of it's rank to row N ([N, 0...WIDTH]) using different APIs on different variable
final result should be:
0 0 0 0 ...
1 1 1 1 ...
2 2 2 2 ...
.
.
.
*/
int simpletest(char* fname, int enable_log) {
    int buffer[MAXPROCESSES];
    MPI_Offset start[2], count[2];

    int i, j, ret, errlen;
    int NProc, MyRank, NP;      // Total process; Rank
    int fid;        // Data set ID
    int did[2];     // IDs of dimension
    int vid;        // IDs for variables
    int dims[2];
    char tmp[1024], tmp2[1024];
    MPI_Info Info;
    MPI_Comm_size(MPI_COMM_WORLD, &NP);
    MPI_Comm_rank(MPI_COMM_WORLD, &MyRank);

    if (NP == 1) {    // Act if there is WIDTH processes for easy debugging. Most debugger supports only single processes.
        NProc = SINGLEPROCNP;
        MyRank = SINGLEPROCRANK;
    }
    else{
        NProc = NP;
    }
    if (MyRank < MAXPROCESSES) {
        // Ensure each process have a independent buffer directory

        MPI_Info_create(&Info);
        if (enable_log) {
            MPI_Info_set(Info, "pnetcdf_log", "enable");
        }
        // Create new cdf file
        ret = ncmpi_create(MPI_COMM_WORLD, fname, NC_CLOBBER, Info, &fid);
        if (ret != NC_NOERR) {
            printf("Error create file\n");
            goto ERROR;
        }
        ret = ncmpi_set_fill(fid, NC_FILL, NULL);
        if (ret != NC_NOERR) {
            printf("Error set fill\n");
            goto ERROR;
        }
        ret = ncmpi_def_dim(fid, "X", NProc, did);  // X
        if (ret != NC_NOERR) {
            printf("Error def dim X\n");
            goto ERROR;
        }
        ret = ncmpi_def_dim(fid, "Y", NProc, did + 1);    // Y
        if (ret != NC_NOERR) {
            printf("Error def dim Y\n");
            goto ERROR;
        }
        ret = ncmpi_def_var(fid, "M", NC_INT, 2, did, vid);
        if (ret != NC_NOERR) {
            printf("Error def var M\n");
            goto ERROR;
        }
        ret = ncmpi_enddef(fid);
        if (ret != NC_NOERR) {
            printf("Error enddef\n");
            goto ERROR;
        }
        // Indep mode
        ret = ncmpi_begin_indep_data(fid);
        if (ret != NC_NOERR) {
            printf("Error begin indep\n");
            goto ERROR;
        }
        // We all write rank from now on
        for (i = 0; i < NProc; i++) {
            buffer[i] = MyRank;
        }

        // put_vara
        count[0] = 1;
        count[1] = NProc;
        start[0] = MyRank;
        start[1] = 0;
        ret = ncmpi_put_vara_int(fid, vid, start, count, buffer);
        if (ret != NC_NOERR) {
            MPI_Error_string(ret, tmp, &errlen);
            printf("Error put_varn: %d\n%s\n", errlen, tmp);
            goto ERROR;
        }
        // Collective mode
        ncmpi_end_indep_data(fid);
        if (ret != NC_NOERR) {
            printf("Error end indep");
            goto ERROR;
        }
//.........这里部分代码