CollectiveCommunicationNonBlocking.h

/*
 * CollectiveCommunicationNonBlocking.h
 *
 *  Created on: May 2, 2016
 *      Author: seckler
 */
#pragma once
 
#include "CollectiveCommBaseInterface.h"
#include "CollectiveCommunicationSingleNonBlocking.h"
#include <map>
#include <utils/Logger.h>
#include "CollectiveCommunicationInterface.h"
#include "Simulation.h"
 
#if MPI_VERSION >= 3
 
using Log::global_log;
class CollectiveCommunicationNonBlocking: public CollectiveCommunicationInterface {
public:
    CollectiveCommunicationNonBlocking() :
            _currentKey(-1), _comms() {
    }
 
    ~CollectiveCommunicationNonBlocking() override = default;
 
    void init(MPI_Comm communicator, int numValues, int key = 0) override {
        if (_currentKey != -1) {
            global_log->error() << "CollectiveCommunicationNonBlocking: previous communication with key " << _currentKey
                    << " not yet finalized" << std::endl;
            Simulation::exit(234);
        }
 
        _currentKey = key;
 
        // add the key, if it is not yet existent:
        if (_comms.count(_currentKey) == 1) {
            // this happens, if the key is already existent.
            global_log->debug() << "CollectiveCommunicationNonBlocking: key " << _currentKey
                    << " already existent. Reusing information." << std::endl;
        } else {
            global_log->debug() << "CollectiveCommunicationNonBlocking: key " << _currentKey
                                << " not existent. Cannot reuse information." << std::endl;
            // Creates the CollectiveCommunicationSingleNonBlocking object
            auto [_, inserted] = _comms.try_emplace(_currentKey);
            if (not inserted) {
                global_log->error() << "CollectiveCommunicationNonBlocking: key " << _currentKey
                                    << " could not be inserted. Aborting!" << std::endl;
                Simulation::exit(498789);
            }
        }
        _comms.at(_currentKey).init(communicator, numValues, _currentKey);
    }
 
    void finalize() override {
        _comms.at(_currentKey).finalize();
        if (_currentKey == 0) {
            global_log->debug() << "CollectiveCommunicationNonBlocking: finalizing with key " << _currentKey
                    << ", thus the entry is removed." << std::endl;
            _comms.erase(_currentKey);
        }
        _currentKey = -1;
    }
 
    void appendInt(int intValue) override {
        _comms.at(_currentKey).appendInt(intValue);
    }
 
    void appendUnsLong(unsigned long unsLongValue) override {
        _comms.at(_currentKey).appendUnsLong(unsLongValue);
    }
 
    void appendFloat(float floatValue) override {
        _comms.at(_currentKey).appendFloat(floatValue);
    }
 
    void appendDouble(double doubleValue) override {
        _comms.at(_currentKey).appendDouble(doubleValue);
    }
 
    void appendLongDouble(long double longDoubleValue) override {
        _comms.at(_currentKey).appendLongDouble(longDoubleValue);
    }
 
    MPI_Comm getTopology() override {
        return _comms.at(_currentKey).getTopology();
    }
 
    int getInt() override {
        return _comms.at(_currentKey).getInt();
    }
 
    unsigned long getUnsLong() override {
        return _comms.at(_currentKey).getUnsLong();
    }
 
    float getFloat() override {
        return _comms.at(_currentKey).getFloat();
    }
 
    double getDouble() override {
        return _comms.at(_currentKey).getDouble();
    }
 
    long double getLongDouble() override {
        return _comms.at(_currentKey).getLongDouble();
    }
 
    void broadcast(int root = 0) override {
        _comms.at(_currentKey).broadcast(root);
    }
 
    void allreduceSum() override {
        Log::global_log->debug() << "CollectiveCommunicationNonBlocking: normal Allreduce" << std::endl;
        _comms.at(_currentKey).allreduceSum();
    }
 
    virtual void allreduceSumAllowPrevious() override {
        Log::global_log->debug() << "CollectiveCommunicationNonBlocking: nonblocking Allreduce with id "<< _currentKey << std::endl;
        mardyn_assert(_currentKey > 0); // _currentKey has to be positive non-zero and should be unique for allreduceSumAllowPrevious
        _comms.at(_currentKey).allreduceSumAllowPrevious();
    }
 
    void allreduceCustom(ReduceType type) override{
        _comms.at(_currentKey).allreduceCustom(type);
    }
 
    void scanSum() override {
        _comms.at(_currentKey).scanSum();
    }
 
    /*
     void waitSpecific(int key) {
     _comms[key].waitAndUpdateData();
     }*/
 
    size_t getTotalSize() override {
        size_t tmp = 0;
        for (auto& comm : _comms){
            tmp += comm.second.getTotalSize();
        }
        return tmp;
    }
private:
    int _currentKey;
    std::map<int, CollectiveCommunicationSingleNonBlocking> _comms;
};
 
#endif // MPI_VERSION >= 3