/home/brandes/workspace/LAMA/src/lama/Communicator.hpp

Go to the documentation of this file.

#ifndef LAMA_COMMUNICATOR_HPP_
#define LAMA_COMMUNICATOR_HPP_

// for dll_import
#include <lama/config.hpp>

// base classes
#include <lama/NonCopyable.hpp>
#include <lama/Printable.hpp>

// others
#include <lama/LAMATypes.hpp>
#include <lama/HostReadAccess.hpp>
#include <lama/HostWriteAccess.hpp>
#include <lama/CommunicationPlan.hpp>

#include <lama/exception/LAMAAssert.hpp>

// logging
#include <logging/logging.hpp>

// boost
#include <boost/shared_ptr.hpp>

#include <memory>
#include <vector>
#include <cmath>

namespace lama
{

// Forward declaration of all classes that are used in the interface

class CommunicationPlan;

template<typename T> class LAMAArray;

class SyncToken;

class Distribution;

class Halo;

class LAMA_DLL_IMPORTEXPORT Communicator : public Printable, private NonCopyable
{

public:

    enum ThreadSafetyLevel { Funneled = 1,    
                             Serialized = 2,  
                             Multiple = 3     
                           };

    virtual ~Communicator();

    void factorize2(const double sizeX, const double sizeY, PartitionId procgrid[2]) const;

    void factorize3(const double sizeX, const double sizeY, const double sizeZ, PartitionId procgrid[3]) const;

    void getGrid2Rank( PartitionId pos[2], const PartitionId procgrid[2] ) const;

    void getGrid3Rank( PartitionId pos[3], const PartitionId procgrid[3] ) const;

    bool operator==(const Communicator& other) const;

    bool operator!=(const Communicator& other) const;

    virtual bool isEqual(const Communicator& other) const = 0;

    virtual ThreadSafetyLevel getThreadSafetyLevel() const =0;

    virtual PartitionId getSize() const = 0;

    virtual PartitionId getRank() const = 0;

    inline PartitionId getNeighbor(int pos) const;

    virtual void all2all( int* recvValues, const int* sendValues ) const = 0;

    virtual void exchangeByPlan(
        int* const recvData,
        const CommunicationPlan& recvPlan,
        const int* const sendData,
        const CommunicationPlan& sendPlan ) const = 0;

    virtual void exchangeByPlan(
        float* const recvData,
        const CommunicationPlan& recvPlan,
        const float* const sendData,
        const CommunicationPlan& sendPlan ) const = 0;

    virtual void exchangeByPlan(
        double* const recvData,
        const CommunicationPlan& recvPlan,
        const double* const sendData,
        const CommunicationPlan& sendPlan ) const = 0;

    virtual std::auto_ptr<SyncToken> exchangeByPlanAsync(
        int* const recvData,
        const CommunicationPlan& recvPlan,
        const int* const sendData,
        const CommunicationPlan& sendPlan ) const = 0;

    virtual std::auto_ptr<SyncToken> exchangeByPlanAsync(
        float* const recvData,
        const CommunicationPlan& recvPlan,
        const float* const sendData,
        const CommunicationPlan& sendPlan ) const = 0;

    virtual std::auto_ptr<SyncToken> exchangeByPlanAsync(
        double* const recvData,
        const CommunicationPlan& recvPlan,
        const double* const sendData,
        const CommunicationPlan& sendPlan ) const = 0;

    template<typename T> 
    void exchangeByPlan( 
        LAMAArray<T>& recvArray,
        const CommunicationPlan& recvPlan,
        const LAMAArray<T>& sendArray,
        const CommunicationPlan& sendPlan ) const;

    template<typename T> 
    std::auto_ptr<SyncToken> exchangeByPlanAsync( 
        LAMAArray<T>& recvArray,
        const CommunicationPlan& recvPlan,
        const LAMAArray<T>& sendArray,
        const CommunicationPlan& sendPlan ) const;
                      
    template<typename T>
    void updateHalo( LAMAArray<T>& haloValues, 
                     const LAMAArray<T>& localValues, 
                     const Halo& halo ) const;

    template<typename T>
    std::auto_ptr<SyncToken> updateHaloAsync( LAMAArray<T>& haloValues,
                                              const LAMAArray<T>& localValues,
                                              const Halo& halo ) const;

    template<typename T>
    void shift( LAMAArray<T>& recv, const LAMAArray<T>& send, const int direction ) const;

    template<typename T>
    std::auto_ptr<SyncToken> shiftAsync( LAMAArray<T>& recvArray, const LAMAArray<T>& sendArray,
                                         const int direction ) const;

    void computeOwners(
        const std::vector<IndexType>& requiredIndexes,
        const Distribution& distribution,
        std::vector<PartitionId>& owners) const;

    virtual void bcast (double val[], const IndexType n, const PartitionId root) const = 0;
    virtual void bcast (float val[], const IndexType n, const PartitionId root) const = 0;
    virtual void bcast (int val[], const IndexType n, const PartitionId root) const = 0;

    virtual void scatter (double myvals[], const IndexType n, const PartitionId root, const double allvals[]) const = 0;
    virtual void scatter (float myvals[], const IndexType n, const PartitionId root, const float allvals[]) const = 0;
    virtual void scatter (int myvals[], const IndexType n, const PartitionId root, const int allvals[]) const = 0;

    virtual void scatter (double myvals[], const IndexType n, const PartitionId root, 
                          const double allvals[], const IndexType sizes[])  const = 0;
    virtual void scatter (float myvals[], const IndexType n, const PartitionId root, 
                          const float allvals[], const IndexType sizes[])  const = 0;
    virtual void scatter (int myvals[], const IndexType n, const PartitionId root, 
                          const int allvals[], const IndexType sizes[])  const = 0;

    virtual void gather (double allvals[], const IndexType n, const PartitionId root, const double myvals[]) const = 0;
    virtual void gather (float allvals[], const IndexType n, const PartitionId root, const float myvals[]) const = 0;
    virtual void gather (int allvals[], const IndexType n, const PartitionId root, const int myvals[]) const = 0;

    virtual void gather (double allvals[], const IndexType n, const PartitionId root,
                          const double myvals[], const IndexType sizes[])  const = 0;
    virtual void gather (float allvals[], const IndexType n, const PartitionId root,
                          const float myvals[], const IndexType sizes[])  const = 0;
    virtual void gather (int allvals[], const IndexType n, const PartitionId root,
                          const int myvals[], const IndexType sizes[])  const = 0;

    virtual IndexType shift( double newVals[], const IndexType newSize, 
                             const double oldVals[], const IndexType oldSize, 
                             const int direction ) const = 0;

    virtual IndexType shift( float newVals[], const IndexType newSize, 
                             const float oldVals[], const IndexType oldSize,
                             const int direction ) const = 0;

    virtual IndexType shift( int newVals[], const IndexType newSize, 
                             const int oldVals[], const IndexType oldSize,
                             const int direction ) const = 0;

    virtual std::auto_ptr<SyncToken> shiftAsync( double newVals[], const double oldVals[], 
                                                 const IndexType size, const int direction ) const;

    virtual std::auto_ptr<SyncToken> shiftAsync( float newVals[], const float oldVals[], 
                                                 const IndexType size, const int direction ) const;

    virtual std::auto_ptr<SyncToken> shiftAsync( int newVals[], const int oldVals[], 
                                                 const IndexType size, const int direction ) const;

    virtual float  sum(const float value) const = 0;
    virtual double sum(const double value) const = 0;
    virtual int    sum(const int value) const = 0;
    virtual size_t sum(const size_t value) const = 0;

    virtual float min(const float value) const = 0;
    virtual float max(const float value) const = 0;

    virtual double min(const double value) const = 0;
    virtual double max(const double value) const = 0;

    virtual int min(const int value) const = 0;
    virtual int max(const int value) const = 0;

    virtual void maxloc( double& val, int& location, const PartitionId root ) const = 0;
    virtual void maxloc( float&  val, int& location, const PartitionId root ) const = 0;
    virtual void maxloc( int&    val, int& location, const PartitionId root ) const = 0;

    virtual void swap( double val[], const IndexType n, const PartitionId partner ) const = 0;
    virtual void swap( float  val[], const IndexType n, const PartitionId partner ) const = 0;
    virtual void swap( int    val[], const IndexType n, const PartitionId partner ) const = 0;

    virtual void gather(std::vector<float>& values, float value) const = 0;

    virtual void synchronize() const = 0;

    virtual void writeAt(std::ostream& stream) const;

    const std::string& getType() const { return mCommunicatorType; }

protected:

    // Default constructor can only be called by base classes.

    Communicator( const std::string& type );

    std::string mCommunicatorType;

    LAMA_LOG_DECL_STATIC_LOGGER(logger);

    static void getUserProcArray( PartitionId userProcArray[3] );

    template<typename T>
    IndexType shift0( T newVals[], const IndexType newSize,
                      const T oldVals[], const IndexType oldSize ) const;

    template<typename T>
    std::auto_ptr<SyncToken> defaultShiftAsync( T newVals[], const T oldVals[], 
                                                const IndexType size, const int direction ) const;

    virtual ContextPtr getCommunicationContext() const = 0;

};

typedef boost::shared_ptr<const Communicator> CommunicatorPtr;

/* -------------------------------------------------------------------------- */

PartitionId Communicator::getNeighbor(int pos) const
{
    PartitionId size = getSize();
    PartitionId rank = getRank();

    LAMA_ASSERT( std::abs(pos) <= size,
            "neighbor pos "<<pos<<" out of range ("<<size<<")" );

    return (size + rank+pos)%size;
}

/* -------------------------------------------------------------------------- */

template<typename T>
void Communicator::exchangeByPlan( LAMAArray<T>& recvArray,
                                   const CommunicationPlan& recvPlan,
                                   const LAMAArray<T>& sendArray,
                                   const CommunicationPlan& sendPlan ) const
{
    LAMA_ASSERT_ERROR( sendArray.size() == sendPlan.totalQuantity(), 
                  "Send array has size " << sendArray.size() 
                  << ", but send plan requires " << sendPlan.totalQuantity() << " entries" );
                  
    IndexType recvSize = recvPlan.totalQuantity();

    ContextPtr comCtx = getCommunicationContext();

    WriteAccess<T> recvData( recvArray, comCtx );
    ReadAccess<T> sendData( sendArray, comCtx );

    recvData.clear();
    recvData.resize( recvSize );

    exchangeByPlan( recvData.get(), recvPlan, sendData.get(), sendPlan );
}

/* -------------------------------------------------------------------------- */

template<typename T>
std::auto_ptr<SyncToken> Communicator::exchangeByPlanAsync( LAMAArray<T>& recvArray,
                                                            const CommunicationPlan& recvPlan,
                                                            const LAMAArray<T>& sendArray,
                                                            const CommunicationPlan& sendPlan ) const
{
    LAMA_ASSERT_ERROR( sendArray.size() == sendPlan.totalQuantity(), 
                  "Send array has size " << sendArray.size() 
                  << ", but send plan requires " << sendPlan.totalQuantity() << " entries" );
                  
    IndexType recvSize = recvPlan.totalQuantity();

    // allocate accesses, SyncToken will take ownership

    std::auto_ptr<HostWriteAccess<T> > recvData( new HostWriteAccess<T>( recvArray ) );
    std::auto_ptr<HostReadAccess<T> > sendData( new HostReadAccess<T>( sendArray ) );

    recvData->clear();
    recvData->resize( recvSize );

    std::auto_ptr<SyncToken> token = exchangeByPlanAsync( recvData->get(), recvPlan, sendData->get(), sendPlan );

    // Add the read and write access to the sync token to get it freed after successful wait

    token->pushAccess( std::auto_ptr<BaseAccess> ( recvData.release() ) );
    token->pushAccess( std::auto_ptr<BaseAccess> ( sendData.release() ) );

    return token;
}

}

#endif // LAMA_COMMUNICATOR_HPP_