CollectiveCommunication.h

#ifndef COLLECTIVECOMMUNICATION_H_
#define COLLECTIVECOMMUNICATION_H_
 
#include <mpi.h>
 
#include "Simulation.h"
#include "utils/Logger.h"
#include "CollectiveCommBase.h"
#include "CollectiveCommunicationInterface.h"
#include "utils/mardyn_assert.h"
 
 
/* Enable agglomerated reduce operations. This will store all values in one array and apply a
 * user defined reduce operation so that the MPI reduce operation is only called once. */
#define ENABLE_AGGLOMERATED_REDUCE 1
 
class CollectiveCommunication: public CollectiveCommBase, public CollectiveCommunicationInterface {
public:
    CollectiveCommunication() {
        _communicator = MPI_COMM_NULL;
        _agglomeratedType = MPI_DATATYPE_NULL;
    }
 
    virtual ~~CollectiveCommunication() {
        //mardyn_assert(_agglomeratedType == MPI_DATATYPE_NULL);
    }
 
    virtual void init(MPI_Comm communicator, int numValues, int key = 0) override {
        CollectiveCommBase::init(numValues);
 
        _communicator = communicator;
        _types.reserve(numValues);
    }
 
    // documentation in base class
    virtual void finalize() override {
        CollectiveCommBase::finalize();
        _types.clear();
 
        mardyn_assert(_agglomeratedType == MPI_DATATYPE_NULL);
    }
 
    // documentation in base class
    void appendInt(int intValue) override {
        CollectiveCommBase::appendInt(intValue);
        _types.push_back(MPI_INT);
    }
 
    // documentation in base class
    void appendUnsLong(unsigned long unsLongValue) override {
        CollectiveCommBase::appendUnsLong(unsLongValue);
        _types.push_back(MPI_UNSIGNED_LONG);
    }
 
    // documentation in base class
    void appendFloat(float floatValue) override {
        CollectiveCommBase::appendFloat(floatValue);
        _types.push_back(MPI_FLOAT);
    }
 
    // documentation in base class
    void appendDouble(double doubleValue) override {
        CollectiveCommBase::appendDouble(doubleValue);
        _types.push_back(MPI_DOUBLE);
    }
 
    // documentation in base class
    void appendLongDouble(long double longDoubleValue) override {
        CollectiveCommBase::appendLongDouble(longDoubleValue);
        _types.push_back(MPI_LONG_DOUBLE);
    }
 
    MPI_Comm getTopology() override {
        return _communicator;
    }
 
    // Getters don't need to be overridden, see parent class
 
    // documentation in base class
    void broadcast(int root = 0) override {
        setMPIType();
        valType * startOfValues = _values.data();
        MPI_CHECK(
                MPI_Bcast(startOfValues, 1, _agglomeratedType, root,
                        _communicator));
        MPI_CHECK(MPI_Type_free(&_agglomeratedType));
    }
 
    // documentation in base class
    void allreduceSum() override {
        allreduceCustom(ReduceType::SUM);
    }
 
    void allreduceCustom(ReduceType type) override {
        Log::global_log->debug() << "CollectiveCommunication: custom Allreduce" << std::endl;
#if ENABLE_AGGLOMERATED_REDUCE
        setMPIType();
        MPI_Op agglomeratedTypeAddOperator;
        const int commutative = 1;
        valType * startOfValues = _values.data();
 
        switch (type) {
        case ReduceType::SUM:
            MPI_CHECK(
                    MPI_Op_create(
                            (MPI_User_function * ) CollectiveCommunication::add,
                            commutative, &agglomeratedTypeAddOperator));
            break;
        case ReduceType::MAX:
            MPI_CHECK(
                    MPI_Op_create(
                            (MPI_User_function * ) CollectiveCommunication::max,
                            commutative, &agglomeratedTypeAddOperator));
            break;
        case ReduceType::MIN:
            MPI_CHECK(
                    MPI_Op_create(
                            (MPI_User_function * ) CollectiveCommunication::min,
                            commutative, &agglomeratedTypeAddOperator));
            break;
        default:
            Log::global_log->error()<<"invalid reducetype, aborting." << std::endl;
            Simulation::exit(1);
        }
 
        MPI_CHECK(
                MPI_Allreduce(MPI_IN_PLACE, startOfValues, 1, _agglomeratedType, agglomeratedTypeAddOperator, _communicator));
        MPI_CHECK(MPI_Op_free(&agglomeratedTypeAddOperator));
        MPI_CHECK(MPI_Type_free(&_agglomeratedType));
#else
        for (unsigned int i = 0; i < _types.size(); i++) {
            MPI_Op op = MPI_NO_OP;
            switch(type) {
            case ReduceType::SUM:
                op = MPI_SUM;
                break;
            case ReduceType::MIN:
                op = MPI_MIN;
                break;
            case ReduceType::MAX:
                op = MPI_MAX;
                break;
            default:
                Log::global_log->error()<<"invalid reducetype, aborting." << std::endl;
                Simulation::exit(1);
            }
            MPI_CHECK(MPI_Allreduce( MPI_IN_PLACE, &_values[i], 1, _types[i], op, _communicator ));
        }
#endif
    }
 
    virtual void allreduceSumAllowPrevious() override{
        allreduceSum();
    }
 
    // documentation in base class
    void scanSum() override {
    #if ENABLE_AGGLOMERATED_REDUCE
            setMPIType();
            MPI_Op agglomeratedTypeAddOperator;
            const int commutative = 1;
            valType * startOfValues = _values.data();
            MPI_CHECK(
                    MPI_Op_create(
                            (MPI_User_function * ) CollectiveCommunication::add,
                            commutative, &agglomeratedTypeAddOperator));
            MPI_CHECK(
                    MPI_Scan(MPI_IN_PLACE, startOfValues, 1, _agglomeratedType, agglomeratedTypeAddOperator, _communicator));
            MPI_CHECK(MPI_Op_free(&agglomeratedTypeAddOperator));
            MPI_CHECK(MPI_Type_free(&_agglomeratedType));
    #else
            for(unsigned int i = 0; i < _types.size(); i++ ) {
                MPI_CHECK( MPI_Scan( MPI_IN_PLACE, _values.data(), 1, _types[i], MPI_SUM, _communicator ) );
            }
    #endif
        }
 
    virtual size_t getTotalSize() override{
        return CollectiveCommBase::getTotalSize() + _types.capacity() * sizeof(MPI_Datatype);
    }
 
 
protected:
    void setMPIType() {
        int numblocks = _values.size();
        std::vector<int> blocklengths(numblocks);
        std::vector<MPI_Aint> disps(numblocks);
        int disp = 0;
        for (int i = 0; i < numblocks; i++) {
            blocklengths[i] = 1;
            disps[i] = disp;
            disp += sizeof(valType);
        }
        MPI_Datatype * startOfTypes = &(_types[0]);
#if MPI_VERSION >= 2 && MPI_SUBVERSION >= 0
        MPI_CHECK(
                MPI_Type_create_struct(numblocks, blocklengths.data(), disps.data(),
                        startOfTypes, &_agglomeratedType));
#else
        MPI_CHECK( MPI_Type_struct(numblocks, blocklengths.data(), disps.data(), startOfTypes, &_agglomeratedType) );
#endif
        MPI_CHECK(MPI_Type_commit(&_agglomeratedType));
    }
 
    static void add(valType *invec, valType *inoutvec, int */*len*/,
            MPI_Datatype *dtype) {
        int numints;
        int numaddr;
        int numtypes;
        int combiner;
 
        MPI_CHECK(
                MPI_Type_get_envelope(*dtype, &numints, &numaddr, &numtypes,
                        &combiner));
 
        std::vector<int> arrayInts(numints);
        std::vector<MPI_Aint> arrayAddr(numaddr);
        std::vector<MPI_Datatype> arrayTypes(numtypes);
 
        MPI_CHECK(
                MPI_Type_get_contents(*dtype, numints, numaddr, numtypes,
                        arrayInts.data(), arrayAddr.data(), arrayTypes.data()));
 
        for (int i = 0; i < numtypes; i++) {
            if (arrayTypes[i] == MPI_INT) {
                inoutvec[i].v_int += invec[i].v_int;
            }
            else if (arrayTypes[i] == MPI_UNSIGNED_LONG) {
                inoutvec[i].v_unsLong += invec[i].v_unsLong;
            }
            else if (arrayTypes[i] == MPI_FLOAT) {
                inoutvec[i].v_float += invec[i].v_float;
            }
            else if (arrayTypes[i] == MPI_DOUBLE) {
                inoutvec[i].v_double += invec[i].v_double;
            }
            else if (arrayTypes[i] == MPI_LONG_DOUBLE) {
                inoutvec[i].v_longDouble += invec[i].v_longDouble;
            }
        }
    }
 
    static void max(valType *invec, valType *inoutvec, int */*len*/,
            MPI_Datatype *dtype) {
        int numints;
        int numaddr;
        int numtypes;
        int combiner;
 
        MPI_CHECK(
                MPI_Type_get_envelope(*dtype, &numints, &numaddr, &numtypes,
                        &combiner));
 
        std::vector<int> arrayInts(numints);
        std::vector<MPI_Aint> arrayAddr(numaddr);
        std::vector<MPI_Datatype> arrayTypes(numtypes);
 
        MPI_CHECK(
                MPI_Type_get_contents(*dtype, numints, numaddr, numtypes,
                        arrayInts.data(), arrayAddr.data(), arrayTypes.data()));
 
        for (int i = 0; i < numtypes; i++) {
            if (arrayTypes[i] == MPI_INT) {
                inoutvec[i].v_int = std::max(inoutvec[i].v_int, invec[i].v_int);
            }
            else if (arrayTypes[i] == MPI_UNSIGNED_LONG) {
                inoutvec[i].v_unsLong = std::max(inoutvec[i].v_unsLong, invec[i].v_unsLong);
            }
            else if (arrayTypes[i] == MPI_FLOAT) {
                inoutvec[i].v_float = std::max(inoutvec[i].v_float, invec[i].v_float);
            }
            else if (arrayTypes[i] == MPI_DOUBLE) {
                inoutvec[i].v_double = std::max(inoutvec[i].v_double, invec[i].v_double);
            }
            else if (arrayTypes[i] == MPI_LONG_DOUBLE) {
                inoutvec[i].v_longDouble = std::max(inoutvec[i].v_longDouble, invec[i].v_longDouble);
            }
        }
    }
 
    static void min(valType *invec, valType *inoutvec, int */*len*/,
            MPI_Datatype *dtype) {
        int numints;
        int numaddr;
        int numtypes;
        int combiner;
 
        MPI_CHECK(
                MPI_Type_get_envelope(*dtype, &numints, &numaddr, &numtypes,
                        &combiner));
 
        std::vector<int> arrayInts(numints);
        std::vector<MPI_Aint> arrayAddr(numaddr);
        std::vector<MPI_Datatype> arrayTypes(numtypes);
 
        MPI_CHECK(
                MPI_Type_get_contents(*dtype, numints, numaddr, numtypes,
                        arrayInts.data(), arrayAddr.data(), arrayTypes.data()));
 
        for (int i = 0; i < numtypes; i++) {
            if (arrayTypes[i] == MPI_INT) {
                inoutvec[i].v_int = std::min(inoutvec[i].v_int, invec[i].v_int);
            }
            else if (arrayTypes[i] == MPI_UNSIGNED_LONG) {
                inoutvec[i].v_unsLong = std::min(inoutvec[i].v_unsLong, invec[i].v_unsLong);
            }
            else if (arrayTypes[i] == MPI_FLOAT) {
                inoutvec[i].v_float = std::min(inoutvec[i].v_float, invec[i].v_float);
            }
            else if (arrayTypes[i] == MPI_DOUBLE) {
                inoutvec[i].v_double = std::min(inoutvec[i].v_double, invec[i].v_double);
            }
            else if (arrayTypes[i] == MPI_LONG_DOUBLE) {
                inoutvec[i].v_longDouble = std::min(inoutvec[i].v_longDouble, invec[i].v_longDouble);
            }
        }
    }
 
    std::vector<MPI_Datatype> _types;
 
    MPI_Datatype _agglomeratedType;
 
    MPI_Comm _communicator;
 
};
 
#endif /* COLLECTIVECOMMUNICATION_H_ */