Tpetra_Distributor.hpp

// @HEADER
// ***********************************************************************
//
//          Tpetra: Templated Linear Algebra Services Package
//                 Copyright (2008) Sandia Corporation
//
// Under the terms of Contract DE-AC04-94AL85000 with Sandia Corporation,
// the U.S. Government retains certain rights in this software.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//
// 2. 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.
//
// 3. Neither the name of the Corporation nor the names of the
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY SANDIA CORPORATION "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 SANDIA CORPORATION OR THE
// 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.
//
// ************************************************************************
// @HEADER

#ifndef TPETRA_DISTRIBUTOR_HPP
#define TPETRA_DISTRIBUTOR_HPP

#include "Tpetra_Details_DistributorActor.hpp"
#include "Tpetra_Details_DistributorPlan.hpp"

#include "Tpetra_Util.hpp"
#include "Teuchos_as.hpp"
#include "Teuchos_Describable.hpp"
#include "Teuchos_ParameterListAcceptorDefaultBase.hpp"
#include "Teuchos_VerboseObject.hpp"
#include "Tpetra_Details_Behavior.hpp"

#include "KokkosCompat_View.hpp"
#include "Kokkos_Core.hpp"
#include "Kokkos_TeuchosCommAdapters.hpp"
#include <memory>
#include <sstream>
#include <type_traits>

namespace Tpetra {

  Teuchos::Array<std::string> distributorSendTypes ();

  class Distributor :
    public Teuchos::Describable,
    public Teuchos::ParameterListAcceptorDefaultBase {
  public:


    explicit Distributor (const Teuchos::RCP<const Teuchos::Comm<int> >& comm);

    Distributor (const Teuchos::RCP<const Teuchos::Comm<int> >& comm,
                 const Teuchos::RCP<Teuchos::FancyOStream>& out);

    Distributor (const Teuchos::RCP<const Teuchos::Comm<int> >& comm,
                 const Teuchos::RCP<Teuchos::ParameterList>& plist);

    Distributor (const Teuchos::RCP<const Teuchos::Comm<int> >& comm,
                 const Teuchos::RCP<Teuchos::FancyOStream>& out,
                 const Teuchos::RCP<Teuchos::ParameterList>& plist);

    Distributor (const Distributor& distributor);

    virtual ~Distributor () = default;

    void swap (Distributor& rhs);



    void setParameterList (const Teuchos::RCP<Teuchos::ParameterList>& plist);

    Teuchos::RCP<const Teuchos::ParameterList> getValidParameters () const;



    size_t createFromSends (const Teuchos::ArrayView<const int>& exportProcIDs);

    template <class Ordinal>
    void
    createFromRecvs (const Teuchos::ArrayView<const Ordinal>& remoteIDs,
                     const Teuchos::ArrayView<const int>& remoteProcIDs,
                     Teuchos::Array<Ordinal>& exportIDs,
                     Teuchos::Array<int>& exportProcIDs);

    void
    createFromSendsAndRecvs (const Teuchos::ArrayView<const int>& exportProcIDs,
                             const Teuchos::ArrayView<const int>& remoteProcIDs);



    size_t getNumReceives() const;

    size_t getNumSends() const;

    bool hasSelfMessage() const;

    size_t getMaxSendLength() const;

    size_t getTotalReceiveLength() const;

    Teuchos::ArrayView<const int> getProcsFrom() const;

    Teuchos::ArrayView<const int> getProcsTo() const;

    Teuchos::ArrayView<const size_t> getLengthsFrom() const;

    Teuchos::ArrayView<const size_t> getLengthsTo() const;

    Details::EDistributorHowInitialized howInitialized() const {
      return plan_.howInitialized();
    }



    Teuchos::RCP<Distributor> getReverse(bool create=true) const;



    void doWaits ();

    void doReverseWaits ();

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doPostsAndWaits (
      const ExpView &exports,
      size_t numPackets,
      const ImpView &imports);

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doPostsAndWaits (const ExpView &exports,
                     const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
                     const ImpView &imports,
                     const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID);

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doPosts (const ExpView &exports,
             size_t numPackets,
             const ImpView &imports);

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doPosts (const ExpView &exports,
             const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
             const ImpView &imports,
             const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID);

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doReversePostsAndWaits (const ExpView &exports,
                            size_t numPackets,
                            const ImpView &imports);

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doReversePostsAndWaits (const ExpView &exports,
                            const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
                            const ImpView &imports,
                            const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID);

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doReversePosts (const ExpView &exports,
                    size_t numPackets,
                    const ImpView &imports);

    template <class ExpView, class ImpView>
    typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
    doReversePosts (const ExpView &exports,
                    const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
                    const ImpView &imports,
                    const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID);



    std::string description() const;

    void
    describe (Teuchos::FancyOStream& out,
              const Teuchos::EVerbosityLevel verbLevel =
                Teuchos::Describable::verbLevel_default) const;

    const Details::DistributorPlan& getPlan() const { return plan_; }
  private:
    Details::DistributorPlan plan_;
    Details::DistributorActor actor_;



    static bool getVerbose();

    std::unique_ptr<std::string>
    createPrefix(const char methodName[]) const;

    bool verbose_ = getVerbose();

    mutable Teuchos::RCP<Distributor> reverseDistributor_;

    template <class Ordinal>
    void computeSends (const Teuchos::ArrayView<const Ordinal> &remoteGIDs,
                       const Teuchos::ArrayView<const int> &remoteProcIDs,
                       Teuchos::Array<Ordinal> &exportGIDs,
                       Teuchos::Array<int> &exportProcIDs);

    void createReverseDistributor() const;


    std::string
    localDescribeToString (const Teuchos::EVerbosityLevel vl) const;
  }; // class Distributor

  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doPostsAndWaits (const ExpView& exports,
                   size_t numPackets,
                   const ImpView& imports)
  {
    actor_.doPostsAndWaits(plan_, exports, numPackets, imports);
  }

  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doPostsAndWaits(const ExpView& exports,
                  const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
                  const ImpView& imports,
                  const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID)
  {
    actor_.doPostsAndWaits(plan_, exports, numExportPacketsPerLID, imports, numImportPacketsPerLID);
  }


  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doPosts (const ExpView &exports,
           size_t numPackets,
           const ImpView &imports)
  {
    actor_.doPosts(plan_, exports, numPackets, imports);
  }

  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doPosts (const ExpView &exports,
           const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
           const ImpView &imports,
           const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID)
  {
    actor_.doPosts(plan_, exports, numExportPacketsPerLID, imports, numImportPacketsPerLID);
  }

  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doReversePostsAndWaits (const ExpView& exports,
                          size_t numPackets,
                          const ImpView& imports)
  {
    doReversePosts (exports, numPackets, imports);
    doReverseWaits ();
  }

  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doReversePostsAndWaits (const ExpView& exports,
                          const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
                          const ImpView& imports,
                          const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID)
  {
    doReversePosts (exports, numExportPacketsPerLID, imports,
                    numImportPacketsPerLID);
    doReverseWaits ();
  }

  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doReversePosts (const ExpView &exports,
                  size_t numPackets,
                  const  ImpView &imports)
  {
    // FIXME (mfh 29 Mar 2012) WHY?
    TEUCHOS_TEST_FOR_EXCEPTION(
      ! plan_.getIndicesTo().is_null(), std::runtime_error,
      "Tpetra::Distributor::doReversePosts(3 args): Can only do "
      "reverse communication when original data are blocked by process.");
    if (reverseDistributor_.is_null ()) {
      createReverseDistributor ();
    }
    reverseDistributor_->doPosts (exports, numPackets, imports);
  }

  template <class ExpView, class ImpView>
  typename std::enable_if<(Kokkos::is_view<ExpView>::value && Kokkos::is_view<ImpView>::value)>::type
  Distributor::
  doReversePosts (const ExpView &exports,
                  const Teuchos::ArrayView<const size_t>& numExportPacketsPerLID,
                  const ImpView &imports,
                  const Teuchos::ArrayView<const size_t>& numImportPacketsPerLID)
  {
    // FIXME (mfh 29 Mar 2012) WHY?
    TEUCHOS_TEST_FOR_EXCEPTION(
      ! plan_.getIndicesTo().is_null(), std::runtime_error,
      "Tpetra::Distributor::doReversePosts(3 args): Can only do "
      "reverse communication when original data are blocked by process.");
    if (reverseDistributor_.is_null ()) {
      createReverseDistributor ();
    }
    reverseDistributor_->doPosts (exports, numExportPacketsPerLID,
                                  imports, numImportPacketsPerLID);
  }

  template <class OrdinalType>
  void Distributor::
  computeSends(const Teuchos::ArrayView<const OrdinalType>& importGIDs,
               const Teuchos::ArrayView<const int>& importProcIDs,
               Teuchos::Array<OrdinalType>& exportGIDs,
               Teuchos::Array<int>& exportProcIDs)
  {
    // NOTE (mfh 19 Apr 2012): There was a note on this code saying:
    // "assumes that size_t >= Ordinal".  The code certainly does
    // assume that sizeof(size_t) >= sizeof(OrdinalType) as well as
    // sizeof(size_t) >= sizeof(int).  This is because it casts the
    // OrdinalType elements of importGIDs (along with their
    // corresponding process IDs, as int) to size_t, and does a
    // doPostsAndWaits<size_t>() to send the packed data.
    using Teuchos::ArrayView;
    using std::endl;
    using size_type = typename ArrayView<const OrdinalType>::size_type;
    const char errPrefix[] = "Tpetra::Distributor::computeSends: ";
    const char suffix[] =
      "  Please report this bug to the Tpetra developers.";

    const int myRank = plan_.getComm()->getRank ();

    TEUCHOS_TEST_FOR_EXCEPTION
      (importGIDs.size () != importProcIDs.size (),
       std::invalid_argument, errPrefix << "On Process " << myRank
       << ": importProcIDs.size()=" << importProcIDs.size()
       << " != importGIDs.size()=" << importGIDs.size() << ".");

    const size_type numImports = importProcIDs.size();
    Kokkos::View<size_t*, Kokkos::HostSpace> importObjs("importObjs", 2*numImports);
    // Pack pairs (importGIDs[i], my process ID) to send into importObjs.
    for (size_type i = 0; i < numImports; ++i) {
      importObjs[2*i]   = static_cast<size_t>(importGIDs[i]);
      importObjs[2*i+1] = static_cast<size_t>(myRank);
    }
    //
    // Use a temporary Distributor to send the (importGIDs[i], myRank)
    // pairs to importProcIDs[i].
    //
    Distributor tempPlan(plan_.getComm());
    // mfh 20 Mar 2014: An extra-cautious cast from unsigned to
    // signed, in order to forestall any possible causes for Bug 6069.
    const size_t numExportsAsSizeT =
      tempPlan.createFromSends(importProcIDs);
    const size_type numExports =
      static_cast<size_type>(numExportsAsSizeT);
    TEUCHOS_TEST_FOR_EXCEPTION
      (numExports < 0, std::logic_error, errPrefix <<
       "tempPlan.createFromSends() returned numExports="
       << numExportsAsSizeT << " as a size_t, which overflows to "
       << numExports << " when cast to " <<
       Teuchos::TypeNameTraits<size_type>::name () << "." << suffix);
    TEUCHOS_TEST_FOR_EXCEPTION
      (size_type(tempPlan.getTotalReceiveLength()) != numExports,
       std::logic_error, errPrefix << "tempPlan.getTotalReceiveLength()="
       << tempPlan.getTotalReceiveLength () << " != numExports="
       << numExports  << "." << suffix);

    if (numExports > 0) {
      exportGIDs.resize(numExports);
      exportProcIDs.resize(numExports);
    }

    // exportObjs: Packed receive buffer.  (exportObjs[2*i],
    // exportObjs[2*i+1]) will give the (GID, PID) pair for export i,
    // after tempPlan.doPostsAndWaits(...) finishes below.
    //
    // FIXME (mfh 19 Mar 2014) This only works if OrdinalType fits in
    // size_t.  This issue might come up, for example, on a 32-bit
    // machine using 64-bit global indices.  I will add a check here
    // for that case.
    static_assert(sizeof(size_t) >= sizeof(OrdinalType),
      "Tpetra::Distributor::computeSends: "
      "sizeof(size_t) < sizeof(OrdinalType).");

    TEUCHOS_TEST_FOR_EXCEPTION
      (tempPlan.getTotalReceiveLength () < size_t(numExports),
       std::logic_error,
       errPrefix << "tempPlan.getTotalReceiveLength()="
       << tempPlan.getTotalReceiveLength() << " < numExports="
       << numExports << "." << suffix);

    Kokkos::View<size_t*, Kokkos::HostSpace> exportObjs("exportObjs", tempPlan.getTotalReceiveLength() * 2);
    tempPlan.doPostsAndWaits(importObjs, 2, exportObjs);

    // Unpack received (GID, PID) pairs into exportIDs resp. exportProcIDs.
    for (size_type i = 0; i < numExports; ++i) {
      exportGIDs[i] = static_cast<OrdinalType> (exportObjs[2*i]);
      exportProcIDs[i] = static_cast<int> (exportObjs[2*i+1]);
    }
  }

  template <class OrdinalType>
  void Distributor::
  createFromRecvs (const Teuchos::ArrayView<const OrdinalType> &remoteGIDs,
                   const Teuchos::ArrayView<const int> &remoteProcIDs,
                   Teuchos::Array<OrdinalType> &exportGIDs,
                   Teuchos::Array<int> &exportProcIDs)
  {
    using std::endl;
    const char errPrefix[] = "Tpetra::Distributor::createFromRecvs: ";
    const int myRank = plan_.getComm()->getRank();

    std::unique_ptr<std::string> prefix;
    if (verbose_) {
      prefix = createPrefix("createFromRecvs");
      std::ostringstream os;
      os << *prefix << "Start" << endl;
      std::cerr << os.str();
    }

    const bool debug = Details::Behavior::debug("Distributor");
    if (debug) {
      using Teuchos::outArg;
      using Teuchos::REDUCE_MAX;
      using Teuchos::reduceAll;
      // In debug mode, first test locally, then do an all-reduce to
      // make sure that all processes passed.
      const int errProc =
        (remoteGIDs.size () != remoteProcIDs.size ()) ? myRank : -1;
      int maxErrProc = -1;
      reduceAll(*plan_.getComm(), REDUCE_MAX, errProc, outArg(maxErrProc));
      TEUCHOS_TEST_FOR_EXCEPTION
        (maxErrProc != -1, std::runtime_error, errPrefix << "Lists "
         "of remote IDs and remote process IDs must have the same "
         "size on all participating processes.  Maximum process ID "
         "with error: " << maxErrProc << ".");
    }
    else { // in non-debug mode, just test locally
      // NOTE (mfh 13 Feb 2020) This needs to throw std::runtime_error
      // in order to make an existing Distributor unit test pass.
      TEUCHOS_TEST_FOR_EXCEPTION
        (remoteGIDs.size() != remoteProcIDs.size(), std::runtime_error,
         errPrefix << "On Process " << myRank << ": "
         "remoteGIDs.size()=" << remoteGIDs.size() <<
         " != remoteProcIDs.size()=" << remoteProcIDs.size() << ".");
    }

    computeSends(remoteGIDs, remoteProcIDs, exportGIDs, exportProcIDs);

    plan_.createFromRecvs(remoteProcIDs);

    if (verbose_) {
      std::ostringstream os;
      os << *prefix << "Done" << endl;
      std::cerr << os.str();
    }
  }

} // namespace Tpetra

#endif // TPETRA_DISTRIBUTOR_HPP