openmpi/ompi/mca/io/ompio/io_ompio_coll_array.c

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*
 * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
 *                         University Research and Technology
 *                         Corporation.  All rights reserved.
 * Copyright (c) 2004-2007 The University of Tennessee and The University
 *                         of Tennessee Research Foundation.  All rights
 *                         reserved.
 * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
 *                         University of Stuttgart.  All rights reserved.
 * Copyright (c) 2004-2005 The Regents of the University of California.
 *                         All rights reserved.
 * Copyright (c) 2008-2011 University of Houston. All rights reserved.
 * $COPYRIGHT$
 *
 * Additional copyrights may follow
 *
 * $HEADER$
 */

#include "ompi_config.h"

#include "ompi/runtime/params.h"
#include "ompi/communicator/communicator.h"
#include "ompi/mca/pml/pml.h"
#include "opal/datatype/opal_datatype.h"
#include "ompi/datatype/ompi_datatype.h"
#include "ompi/request/request.h"

#include <math.h>
#include "io_ompio.h"


int ompi_io_ompio_allgatherv_array (void *sbuf,
                                    int scount,
                                    ompi_datatype_t *sdtype,
                                    void *rbuf,
                                    int *rcounts,
                                    int *disps,
                                    ompi_datatype_t *rdtype,
                                    int root_index,
                                    int *procs_in_group,
                                    int procs_per_group,
                                    ompi_communicator_t *comm)
{
    int err = OMPI_SUCCESS;
    OPAL_PTRDIFF_TYPE extent, lb;
    int i, rank, j;
    char *send_buf = NULL;
    struct ompi_datatype_t *newtype, *send_type;

    rank = ompi_comm_rank (comm);
    for (j = 0; j < procs_per_group; j++) {
        if (procs_in_group[j] == rank) {
            break;
        }
    }

    if (MPI_IN_PLACE == sbuf) {
        err = opal_datatype_get_extent (&rdtype->super, &lb, &extent);
        if (OMPI_SUCCESS != err) {
            return OMPI_ERROR;
        }
        send_type = rdtype;
        send_buf = (char*)rbuf;

        for (i = 0; i < j; i++) {
            send_buf += (rcounts[i] * extent);
        }
    }
    else {
        send_buf = (char*)sbuf;
        send_type = sdtype;
    }

    err = ompi_io_ompio_gatherv_array (send_buf,
                                       rcounts[j],
                                       send_type,
                                       rbuf,
                                       rcounts,
                                       disps,
                                       rdtype,
                                       root_index,
                                       procs_in_group,
                                       procs_per_group,
                                       comm);
    if (OMPI_SUCCESS != err) {
        return err;
    }

    err = ompi_datatype_create_indexed (procs_per_group,
                                        rcounts,
                                        disps,
                                        rdtype,
                                        &newtype);
    if (MPI_SUCCESS != err) {
        return err;
    }
    err = ompi_datatype_commit (&newtype);
    if(MPI_SUCCESS != err) {
        return err;
    }

    ompi_io_ompio_bcast_array (rbuf,
                               1,
                               newtype,
                               root_index,
                               procs_in_group,
                               procs_per_group,
                               comm);

    ompi_datatype_destroy (&newtype);

    return OMPI_SUCCESS;
}

int ompi_io_ompio_gatherv_array (void *sbuf,
                                 int scount,
                                 ompi_datatype_t *sdtype,
                                 void *rbuf,
                                 int *rcounts,
                                 int *disps,
                                 ompi_datatype_t *rdtype,
                                 int root_index,
                                 int *procs_in_group,
                                 int procs_per_group,
                                 struct ompi_communicator_t *comm)
{
    int i, rank;
    int err = OMPI_SUCCESS;
    char *ptmp;
    OPAL_PTRDIFF_TYPE extent, lb;
    ompi_request_t **reqs=NULL;

    rank = ompi_comm_rank (comm);

    if (procs_in_group[root_index] != rank)  {
        if (scount > 0) {
            return MCA_PML_CALL(send(sbuf,
                                     scount,
                                     sdtype,
                                     procs_in_group[root_index],
                                     OMPIO_TAG_GATHERV,
                                     MCA_PML_BASE_SEND_STANDARD,
                                     comm));
        }
        return err;
    }

    /* writer processes, loop receiving data from proceses
       belonging to each corresponding root */

    err = opal_datatype_get_extent (&rdtype->super, &lb, &extent);
    if (OMPI_SUCCESS != err) {
        return OMPI_ERROR;
    }
    reqs = (ompi_request_t **) malloc ( procs_per_group *sizeof(ompi_request_t *));
    if ( NULL == reqs ) {
	return OMPI_ERR_OUT_OF_RESOURCE;
    }
    for (i=0; i<procs_per_group; i++) {
        ptmp = ((char *) rbuf) + (extent * disps[i]);

        if (procs_in_group[i] == rank) {
            if (MPI_IN_PLACE != sbuf &&
                (0 < scount) &&
                (0 < rcounts[i])) {
                err = ompi_datatype_sndrcv (sbuf,
                                            scount,
                                            sdtype,
                                            ptmp,
                                            rcounts[i],
                                            rdtype);
            }
	    reqs[i] = MPI_REQUEST_NULL;
        }
        else {
            /* Only receive if there is something to receive */
            if (rcounts[i] > 0) {
                err = MCA_PML_CALL(irecv(ptmp,
                                         rcounts[i],
                                         rdtype,
                                         procs_in_group[i],
                                         OMPIO_TAG_GATHERV,
                                         comm,
                                         &reqs[i]));
            }
	    else {
		reqs[i] = MPI_REQUEST_NULL;
	    }
        }

        if (OMPI_SUCCESS != err) {
	    free ( reqs );
            return err;
        }
    }
    /* All done */
    err = ompi_request_wait_all ( procs_per_group, reqs, MPI_STATUSES_IGNORE );
    if ( NULL != reqs ) {
	free ( reqs );
    }
    return err;
}

int ompi_io_ompio_scatterv_array (void *sbuf,
                                  int *scounts,
                                  int *disps,
                                  ompi_datatype_t *sdtype,
                                  void *rbuf,
                                  int rcount,
                                  ompi_datatype_t *rdtype,
                                  int root_index,
                                  int *procs_in_group,
                                  int procs_per_group,
                                  struct ompi_communicator_t *comm)
{
    int i, rank;
    int err = OMPI_SUCCESS;
    char *ptmp;
    OPAL_PTRDIFF_TYPE extent, lb;
    ompi_request_t ** reqs=NULL;

    rank = ompi_comm_rank (comm);

    if (procs_in_group[root_index] != rank) {
        if (rcount > 0) {
            err = MCA_PML_CALL(recv(rbuf,
                                    rcount,
                                    rdtype,
                                    procs_in_group[root_index],
                                    OMPIO_TAG_SCATTERV,
                                    comm,
                                    MPI_STATUS_IGNORE));
        }
        return err;
    }

    /* writer processes, loop sending data to proceses
       belonging to each corresponding root */

    err = opal_datatype_get_extent (&sdtype->super, &lb, &extent);
    if (OMPI_SUCCESS != err) {
        return OMPI_ERROR;
    }
    reqs = ( ompi_request_t **) malloc ( procs_per_group * sizeof ( ompi_request_t *));
    if (NULL == reqs ) {
	return OMPI_ERR_OUT_OF_RESOURCE;
    }

    for (i=0 ; i<procs_per_group ; ++i) {
        ptmp = ((char *) sbuf) + (extent * disps[i]);

        if (procs_in_group[i] == rank) {
            if (MPI_IN_PLACE != sbuf &&
                (0 < scounts[i]) &&
                (0 < rcount)) {
                err = ompi_datatype_sndrcv (ptmp,
                                            scounts[i],
                                            sdtype,
                                            rbuf,
                                            rcount,
                                            rdtype);
            }
	    reqs[i] = MPI_REQUEST_NULL;
        }
        else {
            /* Only receive if there is something to receive */
            if (scounts[i] > 0) {
                err = MCA_PML_CALL(isend(ptmp,
                                         scounts[i],
                                         sdtype,
                                         procs_in_group[i],
                                         OMPIO_TAG_SCATTERV,
                                         MCA_PML_BASE_SEND_STANDARD,
                                         comm,
				         &reqs[i]));
            }
	    else {
		reqs[i] = MPI_REQUEST_NULL;
            }
        }
        if (OMPI_SUCCESS != err) { 
            free ( reqs );	    
            return err;
        }
    }
    /* All done */
    err = ompi_request_wait_all ( procs_per_group, reqs, MPI_STATUSES_IGNORE );
    if ( NULL != reqs ) {
	free ( reqs );
    }
    return err;
}

int ompi_io_ompio_allgather_array (void *sbuf,
                                   int scount,
                                   ompi_datatype_t *sdtype,
                                   void *rbuf,
                                   int rcount,
                                   ompi_datatype_t *rdtype,
                                   int root_index,
                                   int *procs_in_group,
                                   int procs_per_group,
                                   ompi_communicator_t *comm)
{
    int err = OMPI_SUCCESS;
    int rank;
    OPAL_PTRDIFF_TYPE extent, lb;

    rank = ompi_comm_rank (comm);

    if (((void *) 1) == sbuf && 0 != rank) {
        err = opal_datatype_get_extent (&rdtype->super, &lb, &extent);
        if (OMPI_SUCCESS != err) {
            return OMPI_ERROR;
        }
        sbuf = ((char*) rbuf) + (rank * extent * rcount);
        sdtype = rdtype;
        scount = rcount;
    }

    /* Gather and broadcast. */
    err = ompi_io_ompio_gather_array (sbuf,
                                      scount,
                                      sdtype,
                                      rbuf,
                                      rcount,
                                      rdtype,
                                      root_index,
                                      procs_in_group,
                                      procs_per_group,
                                      comm);

    if (OMPI_SUCCESS == err) {
        err = ompi_io_ompio_bcast_array (rbuf,
                                         rcount * procs_per_group,
                                         rdtype,
                                         root_index,
                                         procs_in_group,
                                         procs_per_group,
                                         comm);
    }
    /* All done */

    return err;
}

int ompi_io_ompio_gather_array (void *sbuf,
                                int scount,
                                ompi_datatype_t *sdtype,
                                void *rbuf,
                                int rcount,
                                ompi_datatype_t *rdtype,
                                int root_index,
                                int *procs_in_group,
                                int procs_per_group,
                                struct ompi_communicator_t *comm)
{
    int i;
    int rank;
    char *ptmp;
    OPAL_PTRDIFF_TYPE incr;
    OPAL_PTRDIFF_TYPE extent, lb;
    int err = OMPI_SUCCESS;
    ompi_request_t ** reqs=NULL;
    
    rank = ompi_comm_rank (comm);

    /* Everyone but the writers sends data and returns. */
    if (procs_in_group[root_index] != rank) {
        err = MCA_PML_CALL(send(sbuf,
                                scount,
                                sdtype,
                                procs_in_group[root_index],
                                OMPIO_TAG_GATHER,
                                MCA_PML_BASE_SEND_STANDARD,
                                comm));
        return err;
    }

    /* writers, loop receiving the data. */
    opal_datatype_get_extent (&rdtype->super, &lb, &extent);
    incr = extent * rcount;

    reqs = ( ompi_request_t **) malloc ( procs_per_group * sizeof ( ompi_request_t *));
    if (NULL == reqs ) {
	return OMPI_ERR_OUT_OF_RESOURCE;
    }

    for (i = 0, ptmp = (char *) rbuf;
         i < procs_per_group;
         ++i, ptmp += incr) {
        if (procs_in_group[i] == rank) {
            if (MPI_IN_PLACE != sbuf) {
                err = ompi_datatype_sndrcv (sbuf,
                                            scount,
                                            sdtype ,
                                            ptmp,
                                            rcount,
                                            rdtype);
            }
            else {
                err = OMPI_SUCCESS;
            }
	    reqs[i] = MPI_REQUEST_NULL;
        }
        else {
            err = MCA_PML_CALL(irecv(ptmp,
                                     rcount,
                                     rdtype,
                                     procs_in_group[i],
                                     OMPIO_TAG_GATHER,
                                     comm,
                                     &reqs[i]));
            /*
            for (k=0 ; k<4 ; k++)
                printf ("RECV %p  %d \n",
                        ((struct iovec *)ptmp)[k].iov_base,
                        ((struct iovec *)ptmp)[k].iov_len);
            */
        }

        if (OMPI_SUCCESS != err) {
	    free ( reqs );
            return err;
        }
    }

    /* All done */
    err = ompi_request_wait_all ( procs_per_group, reqs, MPI_STATUSES_IGNORE );
    if ( NULL != reqs ) {
	free ( reqs );
    }

    return err;
}

int ompi_io_ompio_bcast_array (void *buff,
                               int count,
                               ompi_datatype_t *datatype,
                               int root_index,
                               int *procs_in_group,
                               int procs_per_group,
                               ompi_communicator_t *comm)
{
    int i, rank;
    int err = OMPI_SUCCESS;
    ompi_request_t ** reqs=NULL;

    rank = ompi_comm_rank (comm);

    /* Non-writers receive the data. */
    if (procs_in_group[root_index] != rank) {
        err = MCA_PML_CALL(recv(buff,
                                count,
                                datatype,
                                procs_in_group[root_index],
                                OMPIO_TAG_BCAST,
                                comm,
                                MPI_STATUS_IGNORE));
        return err;
    }

    /* Writers sends data to all others. */
    reqs = ( ompi_request_t **) malloc ( procs_per_group * sizeof ( ompi_request_t *));
    if (NULL == reqs ) {
	return OMPI_ERR_OUT_OF_RESOURCE;
    }

    for (i=0 ; i<procs_per_group ; i++) {
        if (procs_in_group[i] == rank) {
	    reqs[i] = MPI_REQUEST_NULL;
            continue;
        }

        err = MCA_PML_CALL(isend(buff,
                                 count,
                                 datatype,
                                 procs_in_group[i],
                                 OMPIO_TAG_BCAST,
                                 MCA_PML_BASE_SEND_STANDARD,
                                 comm,
			         &reqs[i]));
        if (OMPI_SUCCESS != err) {
	    free ( reqs );
            return err;
        }
    }
    err = ompi_request_wait_all ( procs_per_group, reqs, MPI_STATUSES_IGNORE );
    if ( NULL != reqs ) {
	free ( reqs );
    }

    return err;
}