LoadManager.hpp

Go to the documentation of this file.

 
#pragma once
#include <vector>
#include <functional>
#include <thread>
#include <mutex>
#include <boost/date_time.hpp>
#include "lib/Log.hpp"
#include "lib/Exception.hpp"
 
#ifndef DISABLE_MPI
#include "mpi.h"
#endif
#ifndef DISABLE_OMP
#include "omp.h"
#endif
 
#define HMP_CHUNK_PROPERTY_STACK 0 
#define HMP_CHUNK_PROPERTY_BEGIN 1 
#define HMP_CHUNK_PROPERTY_END 2 
 
#ifndef DISABLE_MPI
#define HMP_MPI_ENABLED 
#define HMP_ENABLE_IF_MPI(X) X 
#define HMP_DISABLE_IF_MPI(X) 
#define HMP_APPEND_IF_MPI(X) ,X 
#else
#define HMP_ENABLE_IF_MPI(X) 
#define HMP_DISABLE_IF_MPI(X) X 
#define HMP_APPEND_IF_MPI(X) 
#endif
 
namespace HMP
{
    class LoadManager;
    typedef int StackIdentifier; 
 
    class LoadManager
    {
    protected:
        struct DataStackBase
        {
            enum struct StackType
            {
                Explicit, 
                Implicit,
                Slave,
                Passive
            };
 
            enum struct MessageTag
            {
                Chunk, 
                ChunkReturn, 
            };
 
            virtual ~DataStackBase() {};
 
            virtual void applyCalculator(const int n) {};
 
            #ifdef HMP_MPI_ENABLED
 
            virtual void send(const int offset, const int count, const int serverRank, const MPI_Comm communicator) const {};
 
            virtual void receive(const int offset, const int count, const int rank, const MPI_Comm communicator, MPI_Request &request) {};
 
            virtual void broadcast(const int serverRank, const MPI_Comm communicator) {};
            #endif
 
            StackType type; 
            StackIdentifier master; 
            int size; 
            int typeMultiplicity; 
            int recommendedChunkSizeMultiple; 
            int recommendedChunksPerRank; 
            bool autoBroadcast; 
        };
 
        template <class StackT> struct DataStack : public DataStackBase
        {
            friend class LoadManager;
        protected:
            DataStack() {}
 
            void applyCalculator(const int i) override
            {
                if (type == StackType::Implicit) implicitCalculator(i);
                else if (type == StackType::Explicit) data[i] = explicitCalculator(i);
            }
 
            #ifdef HMP_MPI_ENABLED
 
            void send(const int offset, const int count, const int serverRank, const MPI_Comm communicator) const override
            {
                MPI_Send(static_cast<void *>(data + typeMultiplicity * offset), typeMultiplicity *count * sizeof(StackT), MPI_BYTE, serverRank, static_cast<int>(MessageTag::ChunkReturn), communicator);
            }
 
            void receive(const int offset, const int count, const int rank, const MPI_Comm communicator, MPI_Request &request) override
            {
                MPI_Irecv(static_cast<void *>(data + typeMultiplicity * offset), typeMultiplicity *count * sizeof(StackT), MPI_BYTE, rank, static_cast<int>(MessageTag::ChunkReturn), communicator, &request);
            }
 
            void broadcast(const int serverRank, const MPI_Comm communicator) override
            {
                MPI_Bcast(static_cast<void *>(data), typeMultiplicity * size * sizeof(StackT), MPI_BYTE, serverRank, communicator);
            }
            #endif
 
            std::function<StackT(int)> explicitCalculator; 
            std::function<void(int)> implicitCalculator; 
            StackT *data; 
        };
 
        struct Chunk
        {
        public:
            Chunk()
            {
                properties[HMP_CHUNK_PROPERTY_STACK] = -1;
                properties[HMP_CHUNK_PROPERTY_BEGIN] = 0;
                properties[HMP_CHUNK_PROPERTY_END] = 0;
            };
 
            Chunk(const int stackId, const int begin, const int end)
            {
                properties[HMP_CHUNK_PROPERTY_STACK] = stackId;
                properties[HMP_CHUNK_PROPERTY_BEGIN] = begin;
                properties[HMP_CHUNK_PROPERTY_END] = end;
            }
 
            bool isVoid() const
            {
                return (properties[HMP_CHUNK_PROPERTY_STACK] == -1);
            }
 
            bool operator==(const Chunk &rhs) const
            {
                return (properties[HMP_CHUNK_PROPERTY_STACK] == rhs.properties[HMP_CHUNK_PROPERTY_STACK]) && (properties[HMP_CHUNK_PROPERTY_BEGIN] == rhs.properties[HMP_CHUNK_PROPERTY_BEGIN]) && (properties[HMP_CHUNK_PROPERTY_END] == rhs.properties[HMP_CHUNK_PROPERTY_END]);
            }
 
            bool operator!=(const Chunk &rhs) const
            {
                return !this->operator==(rhs);
            }
 
            int properties[3]; 
        };
 
    public:
        virtual ~LoadManager()
        {
            HMP_ENABLE_IF_MPI(MPI_Comm_free(&_communicator));
 
            while (_stacks.size() > 0)
            {
                delete _stacks.back();
                _stacks.pop_back();
            }
        }
 
        template <class StackT> StackIdentifier addMasterStackExplicit(StackT *const data, const int size, const std::function<StackT(int)> &calculator, const int recommendedChunkSizeMultiple = 1, const int recommendedChunksPerRank = 10, const bool autoBroadcast = false)
        {
            DataStack<StackT> *ds = new DataStack<StackT>();
            ds->type = DataStackBase::StackType::Explicit;
            ds->master = -1;
            ds->size = size;
            ds->typeMultiplicity = 1;
            ds->recommendedChunkSizeMultiple = recommendedChunkSizeMultiple;
            ds->recommendedChunksPerRank = recommendedChunksPerRank;
            ds->autoBroadcast = autoBroadcast;
            ds->explicitCalculator = calculator;
            ds->data = data;
            return _registerStack(ds);
        }
 
        template <class StackT> StackIdentifier addMasterStackImplicit(StackT *const data, const int size, const std::function<void(int)> &calculator, const int typeMultiplicity = 1, const int recommendedChunkSizeMultiple = 1, const int recommendedChunksPerRank = 10, const bool autoBroadcast = false)
        {
            DataStack<StackT> *ds = new DataStack<StackT>();
            ds->type = DataStackBase::StackType::Implicit;
            ds->master = -1;
            ds->size = size;
            ds->typeMultiplicity = typeMultiplicity;
            ds->recommendedChunkSizeMultiple = recommendedChunkSizeMultiple;
            ds->recommendedChunksPerRank = recommendedChunksPerRank;
            ds->autoBroadcast = autoBroadcast;
            ds->implicitCalculator = calculator;
            ds->data = data;
            return _registerStack(ds);
        }
 
        template <class StackT> StackIdentifier addSlaveStack(StackT *const data, const int size, const StackIdentifier master, const int typeMultiplicity = 1)
        {
            DataStack<StackT> *ds = new DataStack<StackT>();
            ds->type = DataStackBase::StackType::Slave;
            ds->master = master;
            ds->size = size;
            ds->typeMultiplicity = typeMultiplicity;
            ds->autoBroadcast = false;
            ds->data = data;
            return _registerStack(ds);
        }
 
        template <class StackT> StackIdentifier addPassiveStack(StackT *const data, const int size)
        {
            DataStack<StackT> *ds = new DataStack<StackT>();
            ds->type = DataStackBase::StackType::Passive;
            ds->master = -1;
            ds->size = size;
            ds->typeMultiplicity = 1;
            ds->autoBroadcast = false;
            ds->data = data;
            return _registerStack(ds);
        }
 
        virtual void calculate(const StackIdentifier *stackIds, const int size) = 0;
 
        void calculate(const std::initializer_list<StackIdentifier> &stackIds)
        {
            calculate(stackIds.begin(), int(stackIds.size()));
        }
 
        void calculate(const StackIdentifier stackId)
        {
            calculate({ stackId });
        }
 
        void calculateAll()
        {
            std::vector<StackIdentifier> all(_stacks.size());
            for (StackIdentifier i = 0; i < StackIdentifier(_stacks.size()); ++i) all[i] = i;
            calculate(all.data(), int(all.size()));
        }
 
        void broadcast(const StackIdentifier *stackIds, const int size)
        {
            #ifdef HMP_MPI_ENABLED
            for (int i = 0; i < size; ++i)
            {
                for (StackIdentifier s = 0; s < StackIdentifier(_stacks.size()); ++s)
                {
                    if (s == stackIds[i] || _stacks[s]->master == stackIds[i]) _stacks[s]->broadcast(_serverRank, _communicator);
                }
            }
            #endif
        }
 
        void broadcast(const std::initializer_list<StackIdentifier> &stackIds)
        {
            broadcast(stackIds.begin(), int(stackIds.size()));
        }
 
        void broadcast(const StackIdentifier stackId)
        {
            broadcast({ stackId });
        }
 
        void broadcastAll()
        {
            std::vector<StackIdentifier> all(_stacks.size());
            for (StackIdentifier i = 0; i < StackIdentifier(_stacks.size()); ++i) all[i] = i;
            broadcast(all.data(), int(all.size()));
        }
 
        virtual void printRuntimeStatistics() const {}
 
    protected:
        LoadManager(const int serverRank HMP_APPEND_IF_MPI(const MPI_Comm communicator)) 
        {
            HMP_ENABLE_IF_MPI(MPI_Comm_dup(communicator, &_communicator));
            HMP_ENABLE_IF_MPI(MPI_Comm_size(_communicator, &_commSize));
            HMP_DISABLE_IF_MPI(_commSize = 1);
            HMP_ENABLE_IF_MPI(MPI_Comm_rank(_communicator, &_rank));
            HMP_DISABLE_IF_MPI(_rank = 0);
 
            if (serverRank >= _commSize) throw Exception(Exception::Type::MpiError, "Server rank must not exceed communicator size.");
            else _serverRank = serverRank;
        }
 
        virtual StackIdentifier _registerStack(DataStackBase *stack)
        {
            //If stack is a slave stack, check that the type multiplicity matches the type multiplicity of the master
            if (stack->type == DataStackBase::StackType::Slave && stack->typeMultiplicity != _stacks[stack->master]->typeMultiplicity) throw Exception(Exception::Type::ArgumentError, "Stack type multiplicity of slave stack does not match type multiplicity of associated master stack.");
 
            _stacks.push_back(stack);
            return StackIdentifier(_stacks.size() - 1);
        }
 
        void _calculateChunk(const Chunk &chunk)
        {
            #ifndef DISABLE_OMP
            #pragma omp parallel for schedule(guided)
            #endif
            for (int i = chunk.properties[HMP_CHUNK_PROPERTY_BEGIN]; i < chunk.properties[HMP_CHUNK_PROPERTY_END]; ++i) _stacks[chunk.properties[HMP_CHUNK_PROPERTY_STACK]]->applyCalculator(i);
        }
 
        std::vector<DataStackBase *> _stacks; 
        int _serverRank; 
        int _rank; 
        int _commSize; 
        HMP_ENABLE_IF_MPI(MPI_Comm _communicator); 
    };
 
    class LoadManagerMaster : public LoadManager
    {
        friend LoadManager *newLoadManager(const int serverRank HMP_APPEND_IF_MPI(const MPI_Comm communicator));
    public:
        virtual void calculate(const StackIdentifier *stackIds, const int size) override
        {
            //reset calculation runtime statistics
            for (int i = 0; i < _commSize; ++i)
            {
                for (StackIdentifier s = 0; s < StackIdentifier(_stacks.size()); ++s)
                {
                    _currentCalculationComputeTimeBuffer[i * _stacks.size() + s] = 0.0f;
                    _currentCalculationWorkDone[i][s] = 0;
                    _currentCalculationTime[i][s] = 0.0f;
 
                }
            }
            boost::posix_time::ptime tic = boost::posix_time::microsec_clock::local_time();
 
            //init chunk spawner
            _initChunkSpawner(stackIds, size);
 
            //spawn local worker
            std::thread *t = new std::thread([&]() { this->_runLocalClient(); });
 
            #ifdef HMP_MPI_ENABLED
            //issue initial chunks
            for (int i = 0; i < _commSize; ++i) if (i != _serverRank) _issueChunk(i);
 
            //collect results and issue consecutive chunks
            int receiveFlag;
            MPI_Status receiveStatus;
            for (;;)
            {
                begin:
                for (int rank = 0; rank < _commSize; ++rank)
                {
                    if (rank == _serverRank || _pendingRequests[rank] == MPI_REQUEST_NULL) continue;
                    MPI_Test(_pendingRequests + rank, &receiveFlag, &receiveStatus);
                    if (receiveFlag == 1)
                    {
                        _despawnChunk(rank);
                        _issueChunk(receiveStatus.MPI_SOURCE);
                    }
                }
                for (int rank = 0; rank < _commSize; ++rank) if (_pendingRequests[rank] != MPI_REQUEST_NULL) goto begin;
                break;
            }
            #endif
 
            //join local worker
            t->join();
 
            //broadcast result
            for (int s = 0; s < size; ++s)
            {
                if (_stacks[stackIds[s]]->autoBroadcast) broadcast(stackIds[s]);
            }
 
            //gather runtime statistics
            HMP_ENABLE_IF_MPI(MPI_Gather(MPI_IN_PLACE, int(_stacks.size()), MPI_FLOAT, _currentCalculationComputeTimeBuffer.data(), int(_stacks.size()), MPI_FLOAT, _serverRank, _communicator));
 
            for (int i = 0; i < _commSize; ++i)
            {
                for (StackIdentifier s = 0; s < StackIdentifier(_stacks.size()); ++s)
                {
                    _totalComputeTime[i][s] += _currentCalculationComputeTimeBuffer[i * _stacks.size() + s];
                }
            }
 
            boost::posix_time::ptime toc = boost::posix_time::microsec_clock::local_time();
            _totalCalculationTime += float((toc - tic).total_milliseconds());
        }
 
        void printRuntimeStatistics() const override
        {
            Log::log << Log::LogLevel::Debug << "LoadManager total time spent in calculate() calls is " << _totalCalculationTime << "ms" << Log::endl;
            for (int i = 0; i < _commSize; ++i)
            {
                float timePerRank = 0.0;
                for (StackIdentifier j = 0; j < StackIdentifier(_stacks.size()); ++j) timePerRank += _totalComputeTime[i][j];
                Log::log << Log::LogLevel::Debug << "LoadManager (rank " << i << ") active computing time was " << std::setiosflags(std::ios::fixed) << std::setprecision(0) << timePerRank << "ms (Efficiency: " << std::setprecision(1) << 100.0f * timePerRank / _totalCalculationTime << "%)" << Log::endl;
 
                for (StackIdentifier j = 0; j < StackIdentifier(_stacks.size()); ++j) Log::log << Log::LogLevel::Debug << "\t active computing time on stack " << j << " was " << _totalComputeTime[i][j] << "ms" << Log::endl;
            }
        }
 
    protected:
        LoadManagerMaster(const int serverRank HMP_APPEND_IF_MPI(const MPI_Comm communicator)) : LoadManager(serverRank HMP_APPEND_IF_MPI(communicator))
        {
            _totalCalculationTime = 0.0f;
            _totalComputeTime = new std::vector<float>[_commSize];
            _currentCalculationWorkDone = new std::vector<int>[_commSize];
            _currentCalculationTime = new std::vector<float>[_commSize];
            _currentCalculationChunkSpawntime = new boost::posix_time::ptime[_commSize];
            _currentCalculationChunkSpawned = new Chunk[_commSize];
 
            #ifdef HMP_MPI_ENABLED
            _pendingRequests = new MPI_Request[_commSize];
            for (int i = 0; i < _commSize; ++i) _pendingRequests[i] = MPI_REQUEST_NULL;
            #endif
        }
 
        ~LoadManagerMaster()
        {
            delete[] _totalComputeTime;
            delete[] _currentCalculationWorkDone;
            delete[] _currentCalculationTime;
            delete[] _currentCalculationChunkSpawntime;
            delete[] _currentCalculationChunkSpawned;
            
            HMP_ENABLE_IF_MPI(delete[] _pendingRequests);
        }
 
        virtual StackIdentifier _registerStack(DataStackBase *stack) override
        {
            StackIdentifier identifier = LoadManager::_registerStack(stack);
 
            for (int i = 0; i < _commSize; ++i)
            {
                _totalComputeTime[i].push_back(0.0f);
                _currentCalculationWorkDone[i].push_back(0);
                _currentCalculationTime[i].push_back(0.0f);
            }
 
            _currentCalculationComputeTimeBuffer.resize(_commSize * _stacks.size());
            _currentCalculationStackMask.resize(_stacks.size());
            _currentCalculationStackProgress.resize(_stacks.size());
 
            return identifier;
        }
 
        void _runLocalClient()
        {
            for (;;)
            {
                //get chunk
                Chunk c = _spawnChunk(_serverRank);
                if (c.isVoid()) break;
 
                boost::posix_time::ptime tic = boost::posix_time::microsec_clock::local_time();
                _calculateChunk(c);
                boost::posix_time::ptime toc = boost::posix_time::microsec_clock::local_time();
                _currentCalculationComputeTimeBuffer[c.properties[HMP_CHUNK_PROPERTY_STACK]] += (toc - tic).total_milliseconds();
 
                _despawnChunk(_serverRank);
            }
        }
 
        void _initChunkSpawner(const StackIdentifier *stackIds, const int size)
        {
            //reset calculation statistics
            for (int i = 0; i < _commSize; ++i)
            {
                for (StackIdentifier s = 0; s < StackIdentifier(_stacks.size()); ++s)
                {
                    _currentCalculationWorkDone[i][s] = 0;
                    _currentCalculationTime[i][s] = 0.0f;
                }
            }
            for (StackIdentifier s = 0; s < StackIdentifier(_stacks.size()); ++s)
            {
                _currentCalculationStackMask[s] = false;
                _currentCalculationStackProgress[s] = 0;
            }
 
            //enable selected stacks
            for (int i = 0; i < size; ++i) _currentCalculationStackMask[stackIds[i]] = true;
        }
 
        void _initChunkSpawner(const std::initializer_list<StackIdentifier> &stackIds)
        {
            _initChunkSpawner(stackIds.begin(), int(stackIds.size()));
        }
 
        Chunk _spawnChunk(const int rank)
        {
            std::lock_guard<std::mutex> lock(_currentCalculationChunkSpawnerLock);
 
            Chunk c;
            for (StackIdentifier s = 0; s < StackIdentifier(_stacks.size()); ++s)
            {
                //skip inactive and completed stacks
                if (!_currentCalculationStackMask[s] || _currentCalculationStackProgress[s] >= _stacks[s]->size) continue;
 
                //determine max chunk size
                int maximumWorkShare = int(_stacks[s]->size / (_stacks[s]->recommendedChunksPerRank * _commSize));
                maximumWorkShare = (int(maximumWorkShare / _stacks[s]->recommendedChunkSizeMultiple) + 1) * _stacks[s]->recommendedChunkSizeMultiple;
                if (maximumWorkShare < 1) maximumWorkShare = 1;
 
                //determine dynamic chunk size according to compute power
                float myComputePower = _currentCalculationWorkDone[rank][s] / _currentCalculationTime[rank][s];
                float totalComputePower = 0;
                for (int i = 0; i < _commSize; ++i) 
                    for (StackIdentifier j = 0; j < StackIdentifier(_stacks.size()); ++j) totalComputePower += _currentCalculationWorkDone[i][j] / _currentCalculationTime[i][j];
                
                int newCurrentCalculationProgress;
                if (std::isfinite(myComputePower) && std::isfinite(totalComputePower))
                {
                    //factor in amount of remaining work
                    int remainingWork = int(_stacks[s]->size - _currentCalculationStackProgress[s]);
                    int myWorkShare = int((myComputePower / totalComputePower) * remainingWork);
                    myWorkShare = (int(myWorkShare / _stacks[s]->recommendedChunkSizeMultiple) + 1) * _stacks[s]->recommendedChunkSizeMultiple;
 
                    //clip to min/max chunk size
                    int minumumWorkTime = 100;
                    int minimumWorkShare = int(myComputePower * minumumWorkTime);
                    if (minimumWorkShare < 1) minimumWorkShare = 1;
                    if (myWorkShare < minimumWorkShare) myWorkShare = minimumWorkShare;
                    if (myWorkShare > maximumWorkShare) myWorkShare = maximumWorkShare;
                    newCurrentCalculationProgress = _currentCalculationStackProgress[s] + myWorkShare;
                }
                else newCurrentCalculationProgress = _currentCalculationStackProgress[s] + maximumWorkShare;
 
                //clip chunk to stack size
                if (newCurrentCalculationProgress >= _stacks[s]->size) newCurrentCalculationProgress = _stacks[s]->size;
 
                //write chunk parameters
                _currentCalculationWorkDone[rank][s] += newCurrentCalculationProgress - _currentCalculationStackProgress[s];
                c.properties[HMP_CHUNK_PROPERTY_STACK] = s;
                c.properties[HMP_CHUNK_PROPERTY_BEGIN] = decltype(c.properties[HMP_CHUNK_PROPERTY_BEGIN])(_currentCalculationStackProgress[s]);
                c.properties[HMP_CHUNK_PROPERTY_END] = decltype(c.properties[HMP_CHUNK_PROPERTY_END])(newCurrentCalculationProgress);
                Log::log << Log::LogLevel::Debug << "LoadManager spawned chunk (stack " << s << ", from " << _currentCalculationStackProgress[s] << ", to " << newCurrentCalculationProgress << ", rank  " << rank << ")" << Log::endl;
 
                //return chunk
                _currentCalculationStackProgress[s] = newCurrentCalculationProgress;
                break;
            }
            _currentCalculationChunkSpawntime[rank] = boost::posix_time::microsec_clock::local_time();
            _currentCalculationChunkSpawned[rank] = c;
            return c;
        }
 
        void _issueChunk(const int rank)
        {
            #ifdef HMP_MPI_ENABLED
            Chunk c = _spawnChunk(rank);
            MPI_Send(&c.properties, 3, MPI_INT, rank, static_cast<int>(DataStackBase::MessageTag::Chunk), _communicator);
 
            if (!c.isVoid())
            {
                for (StackIdentifier i = 0; i < StackIdentifier(_stacks.size()); ++i)
                {
                    //we may expect to receive data from additional slave stacks; due to MPI's non-overtaking policy, it is sufficient to only store the most recent request per rank
                    if (i == c.properties[HMP_CHUNK_PROPERTY_STACK] || _stacks[i]->master == c.properties[HMP_CHUNK_PROPERTY_STACK]) _stacks[i]->receive(c.properties[HMP_CHUNK_PROPERTY_BEGIN], c.properties[HMP_CHUNK_PROPERTY_END] - c.properties[HMP_CHUNK_PROPERTY_BEGIN], rank, _communicator, _pendingRequests[rank]);
                }
            }
            else _pendingRequests[rank] = MPI_REQUEST_NULL;
            #endif
        }
 
        void _despawnChunk(const int rank)
        {
            float chunktime = float((boost::posix_time::microsec_clock::local_time() - _currentCalculationChunkSpawntime[rank]).total_milliseconds());
            std::lock_guard<std::mutex> lock(_currentCalculationChunkSpawnerLock);
            _currentCalculationTime[rank][_currentCalculationChunkSpawned[rank].properties[HMP_CHUNK_PROPERTY_STACK]] += chunktime;
        }
 
        float _totalCalculationTime; 
        std::vector<float> *_totalComputeTime; 
        std::vector<int> *_currentCalculationWorkDone; 
        std::vector<float> *_currentCalculationTime; 
        boost::posix_time::ptime *_currentCalculationChunkSpawntime; 
        Chunk *_currentCalculationChunkSpawned; 
        std::vector<float> _currentCalculationComputeTimeBuffer; 
        std::vector<bool> _currentCalculationStackMask; 
        std::vector<int> _currentCalculationStackProgress; 
        std::mutex _currentCalculationChunkSpawnerLock; 
 
        HMP_ENABLE_IF_MPI(MPI_Request *_pendingRequests); 
    };
 
    class LoadManagerSlave : public LoadManager
    {
        friend LoadManager *newLoadManager(const int serverRank HMP_APPEND_IF_MPI(const MPI_Comm communicator));
    public:
        virtual void calculate(const StackIdentifier *stackIds, const int size) override
        {
            #ifdef HMP_MPI_ENABLED
            memset(_currentCalculationComputeTimeBuffer.data(), 0, _currentCalculationComputeTimeBuffer.size() * sizeof(float));
 
            Chunk c;
            for (;;)
            {
                //receive chunk
                _waitChunk(c);
                if (c.isVoid()) break;
 
                //compute chunk
                boost::posix_time::ptime tic = boost::posix_time::microsec_clock::local_time();
                _calculateChunk(c);
                boost::posix_time::ptime toc = boost::posix_time::microsec_clock::local_time();
                _currentCalculationComputeTimeBuffer[c.properties[HMP_CHUNK_PROPERTY_STACK]] += (toc - tic).total_milliseconds();
 
                //return chunk
                _returnChunk(c);
            }
 
            //broadcast result
            for (int s = 0; s < size; ++s)
            {
                if (_stacks[stackIds[s]]->autoBroadcast) broadcast(stackIds[s]);
            }
 
            //gather compute time statistics
            MPI_Gather(_currentCalculationComputeTimeBuffer.data(), int(_currentCalculationComputeTimeBuffer.size()), MPI_FLOAT, nullptr, int(_currentCalculationComputeTimeBuffer.size()), MPI_FLOAT, _serverRank, _communicator);
            #endif
        }
 
    protected:
        LoadManagerSlave(const int serverRank HMP_APPEND_IF_MPI(const MPI_Comm communicator)) : LoadManager(serverRank HMP_APPEND_IF_MPI(communicator)) {}
 
        virtual StackIdentifier _registerStack(DataStackBase *stack) override
        {
            StackIdentifier identifier = LoadManager::_registerStack(stack);
 
            _currentCalculationComputeTimeBuffer.resize(_stacks.size());
 
            return identifier;
        }
 
        void _waitChunk(Chunk &chunk) const
        {
            #ifdef HMP_MPI_ENABLED
            MPI_Status status;
            MPI_Recv(&chunk.properties, 3, MPI_INT, _serverRank, static_cast<int>(DataStackBase::MessageTag::Chunk), _communicator, &status);
            #endif
        }
 
        void _returnChunk(const Chunk &chunk) const
        {
            #ifdef HMP_MPI_ENABLED
            for (int i = 0; i < int(_stacks.size()); ++i)
            {
                if (i == chunk.properties[HMP_CHUNK_PROPERTY_STACK] || _stacks[i]->master == chunk.properties[HMP_CHUNK_PROPERTY_STACK]) _stacks[i]->send(chunk.properties[HMP_CHUNK_PROPERTY_BEGIN], chunk.properties[HMP_CHUNK_PROPERTY_END] - chunk.properties[HMP_CHUNK_PROPERTY_BEGIN], _serverRank, _communicator);
            }
            #endif
        }
 
        std::vector<float> _currentCalculationComputeTimeBuffer; 
    };
 
    inline LoadManager *newLoadManager(const int serverRank = 0 HMP_APPEND_IF_MPI(const MPI_Comm communicator = MPI_COMM_WORLD))
    {
        int myRank;
        HMP_ENABLE_IF_MPI(MPI_Comm_rank(communicator, &myRank));
        HMP_DISABLE_IF_MPI(myRank = 0);
 
        if (myRank == serverRank) return new LoadManagerMaster(serverRank HMP_APPEND_IF_MPI(communicator));
        else return new LoadManagerSlave(serverRank HMP_APPEND_IF_MPI(communicator));
    };
 
} //namespace HMP
 
#undef HMP_CHUNK_PROPERTY_STACK
#undef HMP_CHUNK_PROPERTY_BEGIN
#undef HMP_CHUNK_PROPERTY_END
 
#undef HMP_MPI_ENABLED
#undef HMP_ENABLE_IF_MPI
#undef HMP_DISABLE_IF_MPI
#undef HMP_APPEND_IF_MPI