xt_redist.c

Go to the documentation of this file.

/*
 * Keywords:
 * Maintainer: Jörg Behrens <behrens@dkrz.de>
 *             Moritz Hanke <hanke@dkrz.de>
 *             Thomas Jahns <jahns@dkrz.de>
 * URL: https://doc.redmine.dkrz.de/yaxt/html/
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are  permitted provided that the following conditions are
 * met:
 *
 * Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 *
 * Redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the following disclaimer in the
 * documentation and/or other materials provided with the distribution.
 *
 * Neither the name of the DKRZ GmbH nor the names of its contributors
 * may be used to endorse or promote products derived from this software
 * without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
 * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <limits.h>
#include <stdbool.h>
#include <stdlib.h>

#include "core/core.h"
#include "xt/xt_core.h"
#include "xt/xt_redist.h"
#include "xt/xt_mpi.h"
#include "core/ppm_xfuncs.h"
#include "xt_redist_internal.h"

Xt_redist xt_redist_copy(Xt_redist redist) {

  return redist->vtable->copy(redist);
}

void xt_redist_delete(Xt_redist redist) {

  redist->vtable->delete(redist);
}

void xt_redist_s_exchange(Xt_redist redist, int num_arrays,
                          const void **src_data, void **dst_data) {

  redist->vtable->s_exchange(redist, num_arrays, src_data, dst_data);
}

void xt_redist_s_exchange1(Xt_redist redist, const void *src_data, void *dst_data) {

  redist->vtable->s_exchange1(redist, src_data, dst_data);
}

MPI_Datatype xt_redist_get_send_MPI_Datatype(Xt_redist redist, int rank) {

  return redist->vtable->get_send_MPI_Datatype(redist, rank);
}

MPI_Datatype xt_redist_get_recv_MPI_Datatype(Xt_redist redist, int rank) {

  return redist->vtable->get_recv_MPI_Datatype(redist, rank);
}

MPI_Comm xt_redist_get_MPI_Comm(Xt_redist redist) {

  return redist->vtable->get_MPI_Comm(redist);
}

int xt_redist_get_msg_ranks(Xt_redist redist, enum xt_msg_direction direction,
                            int **ranks)
{
  return redist->vtable->get_msg_ranks(redist, direction, ranks);
}


void
xt_redist_msgs_strided_copy(size_t n,
                            struct Xt_redist_msg *restrict src,
                            size_t src_stride,
                            struct Xt_redist_msg *restrict dst,
                            size_t dst_stride,
                            MPI_Comm comm) {


  unsigned char *restrict src_store = (unsigned char *)src,
    *restrict dst_store = (unsigned char *)dst;
  for (size_t i = 0; i < n; ++i) {
    struct Xt_redist_msg *restrict src_msg
      = (struct Xt_redist_msg *)(void *)(src_store + i * src_stride),
      *dst_msg = (struct Xt_redist_msg *)(void *)(dst_store + i * dst_stride);
    dst_msg->rank = src_msg->rank;
    xt_mpi_call(MPI_Type_dup(src_msg->datatype, &(dst_msg->datatype)), comm);
  }
}

void xt_redist_msgs_strided_destruct(size_t n, struct Xt_redist_msg *msgs,
                                     MPI_Comm comm, size_t ofs_stride) {

  unsigned char *restrict msgs_store = (unsigned char *)msgs;
  for (size_t i = 0; i < n; ++i) {
    MPI_Datatype *dt
      = &(((struct Xt_redist_msg *)(void *)(msgs_store + i * ofs_stride))->datatype);
    if (*dt != MPI_DATATYPE_NULL)
      xt_mpi_call(MPI_Type_free(dt), comm);
  }
}

void
xt_redist_check_comms(Xt_redist *redists, int num_redists, MPI_Comm comm) {
  int result;

  for (int i = 0; i < num_redists; ++i) {

    xt_mpi_call(MPI_Comm_compare(xt_redist_get_MPI_Comm(redists[i]),
                                 comm, &result), comm);

    if ((result != MPI_IDENT) && (result != MPI_CONGRUENT))
      Xt_abort(comm, "ERROR: MPI communicators do not match; cannot build "
               "redist_collection_static\n", __FILE__, __LINE__);
  }
}

size_t
xt_ranks_uniq_count(size_t num_rank_sets,
                    size_t *restrict num_ranks,
                    const int *ranks[num_rank_sets])
{
  size_t rank_pos[num_rank_sets];
  for (size_t j = 0; j < num_rank_sets; ++j)
    rank_pos[j] = 0;
  bool ranks_left;
  size_t num_messages = 0;
  do {
    int min_rank = INT_MAX;
    /* find minimal rank in list, guaranteed to be smaller than comm_size */
    for (size_t j = 0; j < num_rank_sets; ++j)
      if (rank_pos[j] < num_ranks[j] && ranks[j][rank_pos[j]] < min_rank)
        min_rank = ranks[j][rank_pos[j]];
    ranks_left = false;
    /* increment list index for all redists matching minimal rank and
     * see if any ranks are left */
    for (size_t j = 0; j < num_rank_sets; ++j) {
      rank_pos[j]
        += (rank_pos[j] < num_ranks[j] && ranks[j][rank_pos[j]] == min_rank);
      ranks_left |= (rank_pos[j] < num_ranks[j]);
    }
    ++num_messages;
  } while (ranks_left);
  return num_messages;
}

MPI_Datatype
xt_create_compound_datatype(size_t num_redists,
                            const MPI_Aint displacements[num_redists],
                            const MPI_Datatype datatypes[num_redists],
                            const int block_lengths[num_redists],
                            MPI_Comm comm)
{
  size_t num_datatypes = 0;
  /* allocate more than max_auto_dt datatype items from heap */
  enum { max_auto_dt = 8 };
  for (size_t i = 0; i < num_redists; ++i)
    num_datatypes += (datatypes[i] != MPI_DATATYPE_NULL);
  MPI_Datatype *datatypes_, dt_auto[max_auto_dt];
  MPI_Aint *displacements_, disp_auto[max_auto_dt];
  int *block_lengths_, bl_auto[max_auto_dt];

  if (num_datatypes != num_redists) {
    if (num_datatypes > max_auto_dt) {
      datatypes_ = xmalloc(num_datatypes * sizeof(*datatypes_));
      displacements_ = xmalloc(num_datatypes * sizeof(*displacements_));
      block_lengths_ = xmalloc(num_datatypes * sizeof(*block_lengths_));
    } else {
      datatypes_ = dt_auto;
      displacements_ = disp_auto;
      block_lengths_ = bl_auto;
    }
    num_datatypes = 0;

    for (size_t i = 0; i < num_redists; ++i) {
      if (datatypes[i] != MPI_DATATYPE_NULL) {

        datatypes_[num_datatypes] = datatypes[i];
        displacements_[num_datatypes] = displacements[i];
        block_lengths_[num_datatypes] = block_lengths[i];
        ++num_datatypes;
      }
    }
  } else {
    datatypes_ = (MPI_Datatype *)datatypes;
    displacements_ = (MPI_Aint *)displacements;
    block_lengths_ = (int *)block_lengths;
  }
  MPI_Datatype datatype;
  if (num_datatypes > 1)
    xt_mpi_call(MPI_Type_create_struct((int)num_datatypes, block_lengths_,
                                       displacements_, datatypes_, &datatype),
                comm);
  else if (displacements_[0] == 0)
    xt_mpi_call(MPI_Type_dup(datatypes_[0], &datatype), comm);
  else
    xt_mpi_call(MPI_Type_create_hindexed(1, (int [1]){1}, displacements_,
                                         datatypes_[0], &datatype), comm);

  xt_mpi_call(MPI_Type_commit(&datatype), comm);

  if (num_datatypes != num_redists && num_datatypes > max_auto_dt) {
    free(datatypes_);
    free(displacements_);
  }

  return datatype;
}

/*
 * Local Variables:
 * c-basic-offset: 2
 * coding: utf-8
 * indent-tabs-mode: nil
 * show-trailing-whitespace: t
 * require-trailing-newline: t
 * End:
 */