/* Optimization of Scatter(v): for small chunk sizes n and np processes, sending np*n 
   individual messages to each process can take considerably longer than sending the 
   same data in fewer messages. This is always true for the short protocol, and up to a 
   certain message size this holds true for the eager protocol, too. Therefore, we do a 
   broadcast-style binominal-tree-wise distribtion of the data - this involves 
   sending more data in total, but reduces the number of sequential messages from 
   np down to ld(np). The actual reduction of the latency depends on the latency ratio
   of the original (smaller sized) and the new (bigger sized) messages. */
int MPID_SMI_Scatter (void *sendbuf, int sendcnt, struct MPIR_DATATYPE *sendtype, 
					  void *recvbuf, int recvcnt, struct MPIR_DATATYPE *recvtype, int root, 
					  struct MPIR_COMMUNICATOR *comm )
{
    MPI_Status  status;
    MPI_Request req;
    MPI_Aint    send_size, recv_size, send_extent, recv_extent, count, displs;
    int         rank, rel_rank, size, mask, is_leaf, dst, src, msglen;
    int         mpi_errno = MPI_SUCCESS;
    static char myname[] = "MPID_SMI_SCATTER";
    char       *rbuf, *tbuf;
	MPI_Datatype ttype; 
	
    size = comm->np;
    rank = comm->local_rank;

    /* Check for invalid arguments */
#ifndef MPIR_NO_ERROR_CHECKING
    if (root >= size)
		mpi_errno = MPIR_Err_setmsg(MPI_ERR_ROOT, MPIR_ERR_ROOT_TOOBIG,
									myname,(char *)0,(char *)0,root,size);
    if (root < 0) 
		mpi_errno = MPIR_Err_setmsg(MPI_ERR_ROOT,MPIR_ERR_DEFAULT,myname,
									(char *)0,(char *)0,root);
    if (mpi_errno)
		return MPIR_ERROR(comm, mpi_errno, myname );
#endif

    /* Get size & extent of send and recv types. We could use MPI_Type_extent/_size, but 
       this also only returns this value. */
    send_size = sendtype->size;
    recv_size = recvtype->size;
    send_extent = sendtype->extent;
	recv_extent = recvtype->extent;

    /* Check for length of data to send - this optimization is only effective up 
       to a certain (system-dependant) message length. Use standard scatter()-function
       for message sizes above. */
    msglen = (rank == root) ? sendcnt*send_size : recvcnt*recv_size;
    if (msglen > MPID_SMI_cfg.COLL_SCATTER_MAX)
		return MPID_SMI_Scatter_seq (sendbuf, sendcnt, sendtype, recvbuf, recvcnt, recvtype, root, comm);
	
    /* Switch communicators to the hidden collective */
    comm = comm->comm_coll;
    rel_rank = (rank >= root) ? rank - root : rank - root + size;
#define SCATTER_DEBUG 0
#define ABS_RANK(r) (((r) + root) % size)
    is_leaf = (rel_rank != 0);
    /* If real rank of root is not 0, we have a problem: the second half of the
       vector being sent is not contiguous, but wraps around the end of the vector.
       We can only handle this by copying the data into a temporary buffer which makes
       this buffer fit well with the 'relative ranks'. Of course, this does have a negative
       impact on performance! */
    if (rel_rank == 0 && root != 0) {
		ALLOCATE (tbuf, char *, size*sendcnt*send_extent);
		MEMCPY(tbuf, (char *)sendbuf + rank*sendcnt*send_extent, (size - rank)*sendcnt*send_extent);
		MEMCPY(tbuf + (size - rank)*sendcnt*send_extent, sendbuf, rank*sendcnt*send_extent);
    } else 

/*@
    MPI_Info_delete - Deletes a (key,value) pair from info

Input Parameters:
+ info - info object (handle)
- key - key (string)

.N fortran
@*/
int MPI_Info_delete(MPI_Info info, char *key)
{
    MPI_Info prev, curr;
    int done;
    static char myname[] = "MPI_INFO_DELETE";
    int mpi_errno;

    if ((info <= (MPI_Info) 0) || (info->cookie != MPIR_INFO_COOKIE)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO, MPIR_ERR_DEFAULT, myname, 
				     (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!key) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_DEFAULT, 
				     myname, (char *)0, (char *)0);
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (strlen(key) > MPI_MAX_INFO_KEY) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_KEY_TOOLONG,
				     myname, (char *)0, (char *)0, 
				     strlen(key), MPI_MAX_INFO_KEY );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!strlen(key)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_KEY_EMPTY,
				     myname, (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    prev = info;
    curr = info->next;
    done = 0;

    while (curr) {
	if (!strcmp(curr->key, key)) {
	    FREE(curr->key);   
	    FREE(curr->value);
	    prev->next = curr->next;
	    FREE(curr);
	    done = 1;
	    break;
	}
	prev = curr;
	curr = curr->next;
    }

    if (!done) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_NOKEY, MPIR_ERR_DEFAULT, 
				     myname, (char *)0, (char *)0, key );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    return MPI_SUCCESS;
}

int MPIR_GetErrorMessage( int errcode, char *dummy, const char **errmsg )
{
    const char *msg;

    _CheckForDebug();

    if (DebugFlag) {
	PRINTF( "GetErrorMessage for code %d\n", errcode );
	_PrintErrCode( errcode );
    }
    /* Check for valid message */
    if (errcode && (errcode & MPIR_ERR_CLASS_MASK) == 0) {
	/* Bogus error code */
	if (DebugFlag) {
	    PRINTF( "Bogus error code %d (class is 0)\n", errcode );
	}
	/* Convert it to an "invalid error code" message */
	errcode = MPIR_Err_setmsg( MPI_ERR_ARG, MPIR_ERR_ERRORCODE, 
				   (char *)0, (char *)0, (char *)0, errcode );
    }
    msg = MPIR_Err_map_code_to_string( errcode );
    if (!msg || !msg[0]) {
	/* pick up the default string */
	if (DebugFlag) {
	    PRINTF( "Map_code for %d returned null or blank\n", errcode );
	}
	msg = MPIR_Get_error_string( errcode );
    }
    *errmsg = msg;

    /* Old prototype has int return */
    return 0;
}

void ADIOI_PFS_Close(ADIO_File fd, int *error_code)
{
    int err;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_PFS_CLOSE";
#endif

#ifdef PROFILE
    MPE_Log_event(9, 0, "start close");
#endif
    err = close(fd->fd_sys);
#ifdef PROFILE
    MPE_Log_event(10, 0, "end close");
#endif
#ifdef PRINT_ERR_MSG
    *error_code = (err == 0) ? MPI_SUCCESS : MPI_ERR_UNKNOWN;
#else
    if (err == -1) {
	*error_code = MPIR_Err_setmsg(MPI_ERR_IO, MPIR_ADIO_ERROR,
			      myname, "I/O Error", "%s", strerror(errno));
	ADIOI_Error(fd, *error_code, myname);	    
    }
    else *error_code = MPI_SUCCESS;
#endif
}

void ADIOI_NFS_Set_shared_fp(ADIO_File fd, ADIO_Offset offset, int *error_code)
{
    int err;
    MPI_Comm dupcommself;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_NFS_SET_SHARED_FP";
#endif

    if (fd->shared_fp_fd == ADIO_FILE_NULL) {
	MPI_Comm_dup(MPI_COMM_SELF, &dupcommself);
	fd->shared_fp_fd = ADIO_Open(MPI_COMM_SELF, dupcommself, fd->shared_fp_fname, 
             fd->file_system, ADIO_CREATE | ADIO_RDWR | ADIO_DELETE_ON_CLOSE, 
             0, MPI_BYTE, MPI_BYTE, M_ASYNC, MPI_INFO_NULL, 
             ADIO_PERM_NULL, error_code);
    }

    if (*error_code != MPI_SUCCESS) return;

    ADIOI_WRITE_LOCK(fd->shared_fp_fd, 0, SEEK_SET, sizeof(ADIO_Offset));
    lseek(fd->shared_fp_fd->fd_sys, 0, SEEK_SET);
    err = write(fd->shared_fp_fd->fd_sys, &offset, sizeof(ADIO_Offset));
    ADIOI_UNLOCK(fd->shared_fp_fd, 0, SEEK_SET, sizeof(ADIO_Offset));

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

/*@
    MPI_File_get_position - Returns the current position of the 
                individual file pointer in etype units relative to
                the current view

Input Parameters:
. fh - file handle (handle)

Output Parameters:
. offset - offset of individual file pointer (nonnegative integer)

.N fortran
@*/
int MPI_File_get_position(MPI_File fh, MPI_Offset *offset)
{
#ifndef PRINT_ERR_MSG
    int error_code;
    static char myname[] = "MPI_FILE_GET_POSITION";
#endif

#ifdef PRINT_ERR_MSG
    if ((fh <= (MPI_File) 0) || (fh->cookie != ADIOI_FILE_COOKIE)) {
	FPRINTF(stderr, "MPI_File_get_position: Invalid file handle\n");
	MPI_Abort(MPI_COMM_WORLD, 1);
    }
#else
    ADIOI_TEST_FILE_HANDLE(fh, myname);
#endif

    if (fh->access_mode & MPI_MODE_SEQUENTIAL) {
#ifdef PRINT_ERR_MSG
	FPRINTF(stderr, "MPI_File_get_position: Can't use this function because file was opened with MPI_MODE_SEQUENTIAL\n");
	MPI_Abort(MPI_COMM_WORLD, 1);
#else
	error_code = MPIR_Err_setmsg(MPI_ERR_UNSUPPORTED_OPERATION, MPIR_ERR_AMODE_SEQ,
				     myname, (char *) 0, (char *) 0);
	return ADIOI_Error(fh, error_code, myname);
#endif
    }

    ADIOI_Get_position(fh, offset);

    return MPI_SUCCESS;
}

/*@
    MPI_Info_free - Frees an info object

Input Parameters:
. info - info object (handle)

.N fortran
@*/
int MPI_Info_free(MPI_Info *info)
{
    MPI_Info curr, next;
    int mpi_errno;
    static char myname[] = "MPI_INFO_FREE";

    if ((*info <= (MPI_Info) 0) || ((*info)->cookie != MPIR_INFO_COOKIE)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO, MPIR_ERR_DEFAULT, myname, 
				     (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    curr = (*info)->next;
    FREE(*info);
    *info = MPI_INFO_NULL;

    while (curr) {
	next = curr->next;
	FREE(curr->key);
	FREE(curr->value);
	FREE(curr);
	curr = next;
    }

    return MPI_SUCCESS;
}

/*@
    MPI_File_write_at_all_end - Complete a split collective write using explict offset

Input Parameters:
. fh - file handle (handle)
. buf - initial address of buffer (choice)

Output Parameters:
. status - status object (Status)

.N fortran
@*/
int MPI_File_write_at_all_end(MPI_File fh, void *buf, MPI_Status *status)
{
#ifndef PRINT_ERR_MSG
    int error_code;
    static char myname[] = "MPI_FILE_WRITE_AT_ALL_END";
#endif

#ifdef PRINT_ERR_MSG
    if ((fh <= (MPI_File) 0) || (fh->cookie != ADIOI_FILE_COOKIE)) {
	FPRINTF(stderr, "MPI_File_write_at_all_end: Invalid file handle\n");
	MPI_Abort(MPI_COMM_WORLD, 1);
    }
#else
    ADIOI_TEST_FILE_HANDLE(fh, myname);
#endif

    if (!(fh->split_coll_count)) {
#ifdef PRINT_ERR_MSG
        FPRINTF(stderr, "MPI_File_write_at_all_end: Does not match a previous MPI_File_write_at_all_begin\n");
        MPI_Abort(MPI_COMM_WORLD, 1);
#else
	error_code = MPIR_Err_setmsg(MPI_ERR_IO, MPIR_ERR_NO_SPLIT_COLL,
                              myname, (char *) 0, (char *) 0);
	return ADIOI_Error(fh, error_code, myname);
#endif
    }

    fh->split_coll_count = 0;

    return MPI_SUCCESS;
}

/*@
    MPI_File_get_type_extent - Returns the extent of datatype in the file

Input Parameters:
. fh - file handle (handle)
. datatype - datatype (handle)

Output Parameters:
. extent - extent of the datatype (nonnegative integer)

.N fortran
@*/
int MPI_File_get_type_extent(MPI_File fh, MPI_Datatype datatype, 
                             MPI_Aint *extent)
{
#ifndef PRINT_ERR_MSG
    int error_code;
    static char myname[] = "MPI_FILE_GET_TYPE_EXTENT";
#endif

#ifdef PRINT_ERR_MSG
    if ((fh <= (MPI_File) 0) || (fh->cookie != ADIOI_FILE_COOKIE)) {
	FPRINTF(stderr, "MPI_File_get_type_extent: Invalid file handle\n");
	MPI_Abort(MPI_COMM_WORLD, 1);
    }
#else
    ADIOI_TEST_FILE_HANDLE(fh, myname);
#endif

    if (datatype == MPI_DATATYPE_NULL) {
#ifdef PRINT_ERR_MSG
        FPRINTF(stderr, "MPI_File_get_type_extent: Invalid datatype\n");
        MPI_Abort(MPI_COMM_WORLD, 1);
#else
	error_code = MPIR_Err_setmsg(MPI_ERR_TYPE, MPIR_ERR_TYPE_NULL,
				     myname, (char *) 0, (char *) 0);
	return ADIOI_Error(fh, error_code, myname);	    
#endif
    }

    return MPI_Type_extent(datatype, extent);
}

void ADIOI_PIOFS_Close(ADIO_File fd, int *error_code)
{
    int err;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_PIOFS_CLOSE";
#endif

#ifdef PROFILE
    MPE_Log_event(9, 0, "start close");
#endif
    err = close(fd->fd_sys);
#ifdef PROFILE
    MPE_Log_event(10, 0, "end close");
#endif
    if (err == -1) {
#ifdef MPICH2
	*error_code = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, myname, __LINE__, MPI_ERR_IO, "**io",
	    "**io %s", strerror(errno));
#elif defined(PRINT_ERR_MSG)
	*error_code =  MPI_ERR_UNKNOWN;
#else /* MPICH-1 */
	*error_code = MPIR_Err_setmsg(MPI_ERR_IO, MPIR_ADIO_ERROR,
			      myname, "I/O Error", "%s", strerror(errno));
	ADIOI_Error(fd, *error_code, myname);	    
#endif
    }
    else *error_code = MPI_SUCCESS;
}

void ADIOI_SCI_IreadContig(ADIO_File fd, void *buf, int count, 
                MPI_Datatype datatype, int file_ptr_type,
                ADIO_Offset offset, ADIO_Request *request, int *error_code)  
{
    int len, typesize;
#ifdef NO_AIO
    ADIO_Status status;
#else
    int err=-1;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_SCI_IREADCONTIG";
#endif
#endif

    (*request) = ADIOI_Malloc_request();
    (*request)->optype = ADIOI_READ;
    (*request)->fd = fd;
    (*request)->datatype = datatype;

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

#ifdef NO_AIO
    /* HP, FreeBSD, Linux */
    /* no support for nonblocking I/O. Use blocking I/O. */

    ADIOI_SCI_ReadContig(fd, buf, len, MPI_BYTE, file_ptr_type, offset, 
			 &status, error_code);  
    (*request)->queued = 0;
#ifdef HAVE_STATUS_SET_BYTES
    if (*error_code == MPI_SUCCESS) {
	MPI_Get_elements(&status, MPI_BYTE, &len);
	(*request)->nbytes = len;
    }
#endif

#else
    if (file_ptr_type == ADIO_INDIVIDUAL) offset = fd->fp_ind;
    err = ADIOI_SCI_aio(fd, buf, len, offset, 0, &((*request)->handle));
    if (file_ptr_type == ADIO_INDIVIDUAL) fd->fp_ind += len;

    (*request)->queued = 1;
    ADIOI_Add_req_to_list(request);

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

    fd->fp_sys_posn = -1;   /* set it to null. */
    fd->async_count++;
}

/*@
    MPI_Info_get_valuelen - Retrieves the length of the value associated with a key

Input Parameters:
+ info - info object (handle)
- key - key (string)

Output Parameters:
+ valuelen - length of value argument (integer)
- flag - true if key defined, false if not (boolean)

.N fortran
@*/
EXPORT_MPI_API int MPI_Info_get_valuelen(MPI_Info info, char *key, int *valuelen, int *flag)
{
    MPI_Info curr;
    int mpi_errno;
    static char myname[] = "MPI_INFO_GET_VALUELEN";

    if ((info <= (MPI_Info) 0) || (info->cookie != MPIR_INFO_COOKIE)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO, MPIR_ERR_DEFAULT, myname, 
				     (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!key) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_DEFAULT, 
				     myname, (char *)0, (char *)0);
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (strlen(key) > MPI_MAX_INFO_KEY) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_KEY_TOOLONG,
				     myname, (char *)0, (char *)0, 
				     strlen(key), MPI_MAX_INFO_KEY );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!strlen(key)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_KEY_EMPTY,
				     myname, (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }


    curr = info->next;
    *flag = 0;

    while (curr) {
	if (!strcmp(curr->key, key)) {
	    *valuelen = strlen(curr->value);
	    *flag = 1;
	    break;
	}
	curr = curr->next;
    }

    return MPI_SUCCESS;
}

void ADIOI_HFS_ReadContig(ADIO_File fd, void *buf, int count,
                     MPI_Datatype datatype, int file_ptr_type,
		     ADIO_Offset offset, ADIO_Status *status, int *error_code)
{
    MPI_Count err=-1, datatype_size, len;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_HFS_READCONTIG";
#endif

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

#ifdef SPPUX
    fd->fp_sys_posn = -1; /* set it to null, since we are using pread */

    if (file_ptr_type == ADIO_EXPLICIT_OFFSET)
	err = pread64(fd->fd_sys, buf, len, offset);
    else {    /* read from curr. location of ind. file pointer */
	err = pread64(fd->fd_sys, buf, len, fd->fp_ind);
	fd->fp_ind += err;
    }
#endif

#ifdef HPUX
    if (file_ptr_type == ADIO_EXPLICIT_OFFSET) {
	if (fd->fp_sys_posn != offset)
	    lseek64(fd->fd_sys, offset, SEEK_SET);
	err = read(fd->fd_sys, buf, len);
	fd->fp_sys_posn = offset + err;
	/* individual file pointer not updated */
    }
    else {  /* read from curr. location of ind. file pointer */
	if (fd->fp_sys_posn != fd->fp_ind)
	    lseek64(fd->fd_sys, fd->fp_ind, SEEK_SET);
	err = read(fd->fd_sys, buf, len);
	fd->fp_ind += err;
	fd->fp_sys_posn = fd->fp_ind;
    }
#endif

#ifdef HAVE_STATUS_SET_BYTES
    if (err != -1) MPIR_Status_set_bytes(status, datatype, err);
#endif

	if (err == -1 ) {
#ifdef MPICH
	    *error_code = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, myname, __LINE__, MPI_ERR_IO, "**io",
		"**io %s", strerror(errno));
#elif defined(PRINT_ERR_MSG)
	    *error_code = (err == -1) ? MPI_ERR_UNKNOWN : MPI_SUCCESS;
#else /* MPICH-1 */
	*error_code = MPIR_Err_setmsg(MPI_ERR_IO, MPIR_ADIO_ERROR,
			      myname, "I/O Error", "%s", strerror(errno));
	ADIOI_Error(fd, *error_code, myname);
#endif
    }
    else *error_code = MPI_SUCCESS;
}

/*@

MPI_Attr_delete - Deletes attribute value associated with a key

Input Parameters:
+ comm - communicator to which attribute is attached (handle) 
- keyval - The key value of the deleted attribute (integer) 

.N fortran

.N Errors
.N MPI_ERR_COMM
.N MPI_ERR_PERM_KEY
@*/
EXPORT_MPI_API int MPI_Attr_delete ( MPI_Comm comm, int keyval )
{
  MPIR_HBT_node *attr;
  MPIR_Attr_key *attr_key;
  int            mpi_errno    = MPI_SUCCESS;
  struct MPIR_COMMUNICATOR *comm_ptr;
  static char myname[] = "MPI_ATTR_DELETE";

  TR_PUSH(myname);

  comm_ptr = MPIR_GET_COMM_PTR(comm);
  MPIR_TEST_MPI_COMM(comm,comm_ptr,comm_ptr,myname);
  
  if ( ( (keyval == MPI_KEYVAL_INVALID) && (mpi_errno = MPI_ERR_OTHER) ) )
	return MPIR_ERROR(comm_ptr, mpi_errno, myname);

  attr_key = MPIR_GET_KEYVAL_PTR( keyval );
  MPIR_TEST_MPI_KEYVAL(keyval,attr_key,comm_ptr,myname);

  if (comm == MPI_COMM_WORLD && attr_key->permanent) 
	return MPIR_ERROR( comm_ptr, 
	     MPIR_ERRCLASS_TO_CODE(MPI_ERR_ARG,MPIR_ERR_PERM_KEY),myname );

  MPIR_HBT_lookup(comm_ptr->attr_cache, keyval, &attr);
  if (attr != (MPIR_HBT_node *)0) {
	if ( attr_key->delete_fn.c_delete_fn ) {
	    if (attr_key->FortranCalling) {
		MPI_Aint  invall = (MPI_Aint)attr->value;
		int inval = (int)invall;
		(*attr_key->delete_fn.f77_delete_fn)(comm, 
					   &keyval, &inval,
					   attr_key->extra_state, &mpi_errno );
		attr->value = (void *)(MPI_Aint)inval;
	    }
	    else
		mpi_errno = (*attr_key->delete_fn.c_delete_fn)(comm, 
					    keyval, attr->value,
					    attr_key->extra_state );
	    if (mpi_errno) 
		return MPIR_ERROR( comm_ptr, mpi_errno, myname );
	    }
	MPIR_HBT_delete(comm_ptr->attr_cache, keyval, &attr);
	/* We will now have one less reference to keyval */
	MPIR_REF_DECR(attr_key);
	if ( attr != (MPIR_HBT_node *)0 ) 
	  (void) MPIR_HBT_free_node ( attr );
	}
  else {
      mpi_errno = MPIR_Err_setmsg( MPI_ERR_ARG, MPIR_ERR_NOKEY, myname,
				   "Key not in communicator", 
				   "Key %d not in communicator", keyval );
      return MPIR_ERROR( comm_ptr, mpi_errno, myname );
      /* "Error in MPI_ATTR_DELETE: key not in communicator" ); */
  }

  TR_POP;
  return(mpi_errno);
}

/*@
    MPI_Info_get_nthkey - Returns the nth defined key in info

Input Parameters:
+ info - info object (handle)
- n - key number (integer)

Output Parameters:
. keys - key (string)

.N fortran
@*/
EXPORT_MPI_API int MPI_Info_get_nthkey(MPI_Info info, int n, char *key)
{
    MPI_Info curr;
    int nkeys, i;
    int mpi_errno;
    static char myname[] = "MPI_INFO_GET_NTHKEY";

    if ((info <= (MPI_Info) 0) || (info->cookie != MPIR_INFO_COOKIE)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO, MPIR_ERR_DEFAULT, myname, 
				     (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!key) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_DEFAULT, 
				     myname, (char *)0, (char *)0);
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    curr = info->next;
    nkeys = 0;
    while (curr) {
	curr = curr->next;
	nkeys++;
    }

    if ((n < 0) || (n >= nkeys)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_ARG, MPIR_ERR_INFO_NKEY, myname,
				     "n is an invalid number",
				     "n = %d is an invalid number", n );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    curr = info->next;
    i = 0;
    while (i < n) {
	curr = curr->next;
	i++;
    }
    strcpy(key, curr->key);

    return MPI_SUCCESS;
}

/*@
    MPI_Type_indexed - Creates an indexed datatype

Input Parameters:
+ count - number of blocks -- also number of entries in indices and blocklens
. blocklens - number of elements in each block (array of nonnegative integers) 
. indices - displacement of each block in multiples of old_type (array of 
  integers)
- old_type - old datatype (handle) 

Output Parameter:
. newtype - new datatype (handle) 

.N fortran

The indices are displacements, and are based on a zero origin.  A common error
is to do something like to following
.vb
    integer a(100)
    integer blens(10), indices(10)
    do i=1,10
         blens(i)   = 1
10       indices(i) = 1 + (i-1)*10
    call MPI_TYPE_INDEXED(10,blens,indices,MPI_INTEGER,newtype,ierr)
    call MPI_TYPE_COMMIT(newtype,ierr)
    call MPI_SEND(a,1,newtype,...)
.ve
expecting this to send 'a(1),a(11),...' because the indices have values 
'1,11,...'.   Because these are `displacements` from the beginning of 'a',
it actually sends 'a(1+1),a(1+11),...'.

If you wish to consider the displacements as indices into a Fortran array,
consider declaring the Fortran array with a zero origin
.vb
    integer a(0:99)
.ve

.N Errors
.N MPI_ERR_COUNT
.N MPI_ERR_TYPE
.N MPI_ERR_ARG
.N MPI_ERR_EXHAUSTED
@*/
EXPORT_MPI_API int MPI_Type_indexed( 
	int count, 
	int blocklens[], 
	int indices[], 
	MPI_Datatype old_type, 
	MPI_Datatype *newtype )
{
  MPI_Aint      *hindices;
  int           i, mpi_errno = MPI_SUCCESS;
  int           total_count;
  struct MPIR_DATATYPE *old_dtype_ptr;
  static char myname[] = "MPI_TYPE_INDEXED";
  MPIR_ERROR_DECL;

  TR_PUSH(myname);
  /* Check for bad arguments */
  old_dtype_ptr   = MPIR_GET_DTYPE_PTR(old_type);
  MPIR_TEST_DTYPE(old_type,old_dtype_ptr,MPIR_COMM_WORLD,myname);
  if ( 
   ( (count    <  0)                 && (mpi_errno = MPI_ERR_COUNT) ) ||
   ( (old_dtype_ptr->dte_type == MPIR_UB) && (mpi_errno = MPI_ERR_TYPE) )  ||
   ( (old_dtype_ptr->dte_type == MPIR_LB) && (mpi_errno = MPI_ERR_TYPE) ) )
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno,myname);
	
  /* Are we making a null datatype? */
  total_count = 0;
  for (i=0; i<count; i++) {
      total_count += blocklens[i];
      if (blocklens[i] < 0) {
	  mpi_errno = MPIR_Err_setmsg( MPI_ERR_ARG, MPIR_ERR_ARG_ARRAY_VAL,
				       myname, (char *)0, (char *)0,
				       "blocklens", i, blocklens[i] );
	  return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno,myname);
      }
  }
  if (total_count == 0) {
      return MPI_Type_contiguous( 0, MPI_INT, newtype );
      }

  /* Generate a call to MPI_Type_hindexed instead.  This means allocating
     a temporary displacement array, multiplying all displacements
     by extent(old_type), and using that */
  MPIR_ALLOC(hindices,(MPI_Aint *)MALLOC(count*sizeof(MPI_Aint)),
	     MPIR_COMM_WORLD,MPI_ERR_EXHAUSTED,myname);
  for (i=0; i<count; i++) {
      hindices[i] = (MPI_Aint)indices[i] * old_dtype_ptr->extent;
  }
  MPIR_ERROR_PUSH(MPIR_COMM_WORLD);
  mpi_errno = MPI_Type_hindexed( count, blocklens, hindices, old_type, 
				 newtype );
  MPIR_ERROR_POP(MPIR_COMM_WORLD);
  FREE(hindices);
  TR_POP;
  MPIR_RETURN(MPIR_COMM_WORLD,mpi_errno, myname);
}

void ADIOI_NTFS_WriteContig(ADIO_File fd, void *buf, int count, 
                   MPI_Datatype datatype, int file_ptr_type,
	           ADIO_Offset offset, ADIO_Status *status, int *error_code)
{
	//int rank;
	DWORD dwTemp;
	DWORD dwNumWritten = 0;
    int err=-1, datatype_size, len;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_NTFS_WRITECONTIG";
#endif

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

    if (file_ptr_type == ADIO_EXPLICIT_OFFSET) {
	if (fd->fp_sys_posn != offset)
	{
		dwTemp = DWORDHIGH(offset);
		SetFilePointer(fd->fd_sys, DWORDLOW(offset), &dwTemp, FILE_BEGIN);
	}
	err = WriteFile(fd->fd_sys, buf, len, &dwNumWritten, NULL);
	//MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	//printf("[%d]W(%d,%d)\n", rank, DWORDLOW(offset), dwNumWritten);
	fd->fp_sys_posn = offset + dwNumWritten;
	/* individual file pointer not updated */        
    }
    else { /* write from curr. location of ind. file pointer */
	if (fd->fp_sys_posn != fd->fp_ind)
	{
		dwTemp = DWORDHIGH(fd->fp_ind);
		SetFilePointer(fd->fd_sys, DWORDLOW(fd->fp_ind), &dwTemp, FILE_BEGIN);
	}
	err = WriteFile(fd->fd_sys, buf, len, &dwNumWritten, NULL);
	//MPI_Comm_rank(MPI_COMM_WORLD, &rank);
	//printf("[%d]w(%d,%d)\n", rank, DWORDLOW(offset), dwNumWritten);
	fd->fp_ind = fd->fp_ind + dwNumWritten;
	fd->fp_sys_posn = fd->fp_ind;
    }

#ifdef HAVE_STATUS_SET_BYTES
    if (err != FALSE) MPIR_Status_set_bytes(status, datatype, dwNumWritten);
#endif

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

void ADIOI_HFS_Open(ADIO_File fd, int *error_code)
{
    int perm, old_mask, amode;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_HFS_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;

    fd->fd_sys = open64(fd->filename, amode, perm);
    fd->fd_direct = -1;

    if ((fd->fd_sys != -1) && (fd->access_mode & ADIO_APPEND)) {
	fd->fp_ind = lseek64(fd->fd_sys, 0, SEEK_END);
#ifdef HPUX
	fd->fp_sys_posn = fd->fp_ind;
#endif
    }

#ifdef SPPUX
	fd->fp_sys_posn = -1;  /* set it to null bec. we use pread, pwrite*/
#endif

	if (fd->fd_sys == -1 ) {
#ifdef MPICH2
	    *error_code = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, myname, __LINE__, MPI_ERR_IO, "**io", 
		"**io %s", strerror(errno));
#elif defined(PRINT_ERR_MSG)
	    *error_code = MPI_ERR_UNKNOWN;
#else /* MPICH-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);	    
#endif
    }
    else *error_code = MPI_SUCCESS;
}

void ADIOI_SFS_Fcntl(ADIO_File fd, int flag, ADIO_Fcntl_t *fcntl_struct, int *error_code)
{
    int  i, ntimes, len;
    ADIO_Offset curr_fsize, alloc_size, size, done;
    ADIO_Status status;
    char *buf;
#ifndef PRINT_ERR_MSG
    static char myname[] = "ADIOI_SFS_FCNTL";
#endif

    switch(flag) {
    case ADIO_FCNTL_GET_FSIZE:
        /* On SFS, I find that a write from one process, which changes
           the file size, does not automatically make the new file size 
           visible to other processes. Therefore, a sync-barrier-sync is 
           needed. (Other processes are able to read the data written 
           though; only file size is returned incorrectly.) */

	fsync(fd->fd_sys);
	MPI_Barrier(fd->comm);
	fsync(fd->fd_sys);
	
	fcntl_struct->fsize = llseek(fd->fd_sys, 0, SEEK_END);
	if (fd->fp_sys_posn != -1) 
	     llseek(fd->fd_sys, fd->fp_sys_posn, SEEK_SET);
	if (fcntl_struct->fsize == -1) {
#ifdef MPICH2
	    *error_code = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, myname, __LINE__, MPI_ERR_IO, "**io",
		"**io %s", strerror(errno));
#elif defined(PRINT_ERR_MSG)
	    *error_code = MPI_ERR_UNKNOWN;
#else /* MPICH-1 */
	    *error_code = MPIR_Err_setmsg(MPI_ERR_IO, MPIR_ADIO_ERROR,
			      myname, "I/O Error", "%s", strerror(errno));
	    ADIOI_Error(fd, *error_code, myname);	    
#endif
	}
	else *error_code = MPI_SUCCESS;
	break;

    case ADIO_FCNTL_SET_DISKSPACE:
	ADIOI_GEN_Prealloc(fd, fcntl_struct->diskspace, error_code);
	break;

    case ADIO_FCNTL_SET_ATOMICITY:
	fd->atomicity = (fcntl_struct->atomicity == 0) ? 0 : 1;
	*error_code = MPI_SUCCESS;
	break;

    default:
	FPRINTF(stderr, "Unknown flag passed to ADIOI_SFS_Fcntl\n");
	MPI_Abort(MPI_COMM_WORLD, 1);
    }
}

/*@
    MPI_Info_get - Retrieves the value associated with a key

Input Parameters:
+ info - info object (handle)
. key - key (string)
- valuelen - length of value argument (integer)

Output Parameters:
+ value - value (string)
- flag - true if key defined, false if not (boolean)

.N fortran
@*/
EXPORT_MPI_API int MPI_Info_get(MPI_Info info, char *key, int valuelen, char *value, int *flag)
{
    MPI_Info curr;
    int mpi_errno;
    static char myname[] = "MPI_INFO_GET";

    if ((info <= (MPI_Info) 0) || (info->cookie != MPIR_INFO_COOKIE)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO, MPIR_ERR_DEFAULT, myname, 
				     (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!key) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_DEFAULT, 
				     myname, (char *)0, (char *)0);
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (strlen(key) > MPI_MAX_INFO_KEY) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_KEY_TOOLONG,
				     myname, (char *)0, (char *)0,strlen(key), 
				     MPI_MAX_INFO_KEY );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!strlen(key)) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_INFO_KEY, MPIR_ERR_KEY_EMPTY,
				     myname, (char *)0, (char *)0 );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (valuelen <= 0) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_ARG, MPIR_ERR_INFO_VALLEN, 
				     myname, (char *)0, (char *)0, valuelen );
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    if (!value) {
	mpi_errno = MPIR_Err_setmsg( MPI_ERR_ARG, MPIR_ERR_INFO_VAL_INVALID,
				     myname, 
				     "Value is an invalid address", (char *)0);
	return MPIR_ERROR( MPIR_COMM_WORLD, mpi_errno, myname );
    }

    curr = info->next;
    *flag = 0;

    while (curr) {
	if (!strcmp(curr->key, key)) {
	    strncpy(value, curr->value, valuelen);
	    value[valuelen] = '\0';
	    *flag = 1;
	    break;
	}
	curr = curr->next;
    }

    return MPI_SUCCESS;
}