MFXWorkerThread.h

Go to the documentation of this file.

/****************************************************************************/
// Eclipse SUMO, Simulation of Urban MObility; see https://eclipse.dev/sumo
// Copyright (C) 2004-2024 German Aerospace Center (DLR) and others.
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0/
// This Source Code may also be made available under the following Secondary
// Licenses when the conditions for such availability set forth in the Eclipse
// Public License 2.0 are satisfied: GNU General Public License, version 2
// or later which is available at
// https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
// SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
/****************************************************************************/
// A thread class together with a pool and a task for parallelized computation
/****************************************************************************/
#pragma once
#include <config.h>
 
// #define WORKLOAD_PROFILING
// at which interval report maximum workload of the threads, needs WORKLOAD_PROFILING
// undefine to use summary report only
#define WORKLOAD_INTERVAL 100
 
#include <list>
#include <vector>
#include "fxheader.h"
#ifdef WORKLOAD_PROFILING
#include <chrono>
#include <utils/common/MsgHandler.h>
#include <utils/common/ToString.h>
#endif
#include <utils/common/UtilExceptions.h>
 
 
// ===========================================================================
// class definitions
// ===========================================================================
class MFXWorkerThread : public FXThread {
 
public:
    class Task {
    public:
        virtual ~Task() {};
 
        virtual void run(MFXWorkerThread* context) = 0;
 
        void setIndex(const int newIndex) {
            myIndex = newIndex;
        }
    private:
        int myIndex;
    };
 
    class Pool {
    public:
        Pool(int numThreads = 0) : myPoolMutex(true), myRunningIndex(0), myException(nullptr)
#ifdef WORKLOAD_PROFILING
            , myNumBatches(0), myTotalMaxLoad(0.), myTotalSpread(0.)
#endif
        {
#ifdef WORKLOAD_PROFILING
            long long int timeDiff = 0;
            for (int i = 0; i < 100; i++) {
                const auto begin = std::chrono::high_resolution_clock::now();
                const auto end = std::chrono::high_resolution_clock::now();
                timeDiff += std::chrono::duration_cast<std::chrono::nanoseconds>(end - begin).count();
            }
            //std::cout << ("Average cost of a timing call (in ns): " + toString(timeDiff / 100.)) << std::endl;
#endif
            while (numThreads > 0) {
                new MFXWorkerThread(*this);
                numThreads--;
            }
        }
 
        virtual ~Pool() {
            clear();
        }
 
        void clear() {
            for (MFXWorkerThread* const worker : myWorkers) {
                delete worker;
            }
            myWorkers.clear();
        }
 
        void addWorker(MFXWorkerThread* const w) {
            myWorkers.push_back(w);
        }
 
        void add(Task* const t, int index = -1) {
            if (index < 0) {
                index = myRunningIndex % myWorkers.size();
            }
#ifdef WORKLOAD_PROFILING
            if (myRunningIndex == 0) {
                for (MFXWorkerThread* const worker : myWorkers) {
                    worker->startProfile();
                }
                myProfileStart = std::chrono::high_resolution_clock::now();
            }
#endif
            t->setIndex(myRunningIndex++);
            myWorkers[index]->add(t);
        }
 
        void addFinished(std::list<Task*>& tasks) {
            myMutex.lock();
            myFinishedTasks.splice(myFinishedTasks.end(), tasks);
            myCondition.signal();
            myMutex.unlock();
        }
 
        void setException(ProcessError& e) {
            myMutex.lock();
            if (myException == nullptr) {
                myException = new ProcessError(e);
            }
            myMutex.unlock();
        }
 
        void waitAll(const bool deleteFinished = true) {
            myMutex.lock();
            while ((int)myFinishedTasks.size() < myRunningIndex) {
                myCondition.wait(myMutex);
            }
#ifdef WORKLOAD_PROFILING
            if (myRunningIndex > 0) {
                const auto end = std::chrono::high_resolution_clock::now();
                const long long int elapsed = std::chrono::duration_cast<std::chrono::microseconds>(end - myProfileStart).count();
                double minLoad = std::numeric_limits<double>::max();
                double maxLoad = 0.;
                for (MFXWorkerThread* const worker : myWorkers) {
                    const double load = worker->endProfile(elapsed);
                    minLoad = MIN2(minLoad, load);
                    maxLoad = MAX2(maxLoad, load);
                }
#ifdef WORKLOAD_INTERVAL
                myTotalMaxLoad += maxLoad;
                myTotalSpread += maxLoad / minLoad;
                myNumBatches++;
                if (myNumBatches % WORKLOAD_INTERVAL == 0) {
                    WRITE_MESSAGE(toString(myFinishedTasks.size()) + " tasks, average maximum load: " + toString(myTotalMaxLoad / WORKLOAD_INTERVAL) + ", average spread: " + toString(myTotalSpread / WORKLOAD_INTERVAL));
                    myTotalMaxLoad = 0.;
                    myTotalSpread = 0.;
                }
#endif
            }
#endif
            if (deleteFinished) {
                for (Task* task : myFinishedTasks) {
                    delete task;
                }
            }
            ProcessError* toRaise = myException;
            myException = nullptr;
            myFinishedTasks.clear();
            myRunningIndex = 0;
            myMutex.unlock();
            if (toRaise != nullptr) {
                ProcessError err = *toRaise;
                delete toRaise;
                throw err;
            }
        }
 
        bool isFull() const {
            return myRunningIndex - (int)myFinishedTasks.size() >= size();
        }
 
        int size() const {
            return (int)myWorkers.size();
        }
 
        void lock() {
            myPoolMutex.lock();
        }
 
        void unlock() {
            myPoolMutex.unlock();
        }
 
        const std::vector<MFXWorkerThread*>& getWorkers() {
            return myWorkers;
        }
    private:
        std::vector<MFXWorkerThread*> myWorkers;
        FXMutex myMutex;
        FXMutex myPoolMutex;
        FXCondition myCondition;
        std::list<Task*> myFinishedTasks;
        int myRunningIndex;
        ProcessError* myException;
#ifdef WORKLOAD_PROFILING
        int myNumBatches;
        double myTotalMaxLoad;
        double myTotalSpread;
        std::chrono::high_resolution_clock::time_point myProfileStart;
#endif
    };
 
public:
    MFXWorkerThread(Pool& pool): FXThread(), myPool(pool), myStopped(false)
#ifdef WORKLOAD_PROFILING
        , myCounter(0), myBusyTime(0), myTotalBusyTime(0), myTotalTime(0)
#endif
    {
        pool.addWorker(this);
        start();
    }
 
    virtual ~MFXWorkerThread() {
        stop();
#ifdef WORKLOAD_PROFILING
        const double load = 100. * myTotalBusyTime / myTotalTime;
        WRITE_MESSAGE("Thread " + toString((long long int)this) + " ran " + toString(myCounter) +
                      " tasks and had a load of " + toString(load) + "% (" + toString(myTotalBusyTime) +
                      "us / " + toString(myTotalTime) + "us), " + toString(myTotalBusyTime / (double)myCounter) + " per task.");
#endif
    }
 
    void add(Task* t) {
        myMutex.lock();
        myTasks.push_back(t);
        myCondition.signal();
        myMutex.unlock();
    }
 
    FXint run() {
        while (!myStopped) {
            myMutex.lock();
            while (!myStopped && myTasks.empty()) {
                myCondition.wait(myMutex);
            }
            if (myStopped) {
                myMutex.unlock();
                break;
            }
            myCurrentTasks.splice(myCurrentTasks.end(), myTasks);
            myMutex.unlock();
            try {
                for (Task* const t : myCurrentTasks) {
#ifdef WORKLOAD_PROFILING
                    const auto before = std::chrono::high_resolution_clock::now();
#endif
                    t->run(this);
#ifdef WORKLOAD_PROFILING
                    const auto after = std::chrono::high_resolution_clock::now();
                    myBusyTime += std::chrono::duration_cast<std::chrono::microseconds>(after - before).count();
                    myCounter++;
#endif
                }
            } catch (ProcessError& e) {
                myPool.setException(e);
            }
            myPool.addFinished(myCurrentTasks);
        }
        return 0;
    }
 
    void stop() {
        myMutex.lock();
        myStopped = true;
        myCondition.signal();
        myMutex.unlock();
        join();
    }
 
#ifdef WORKLOAD_PROFILING
    void startProfile() {
        myBusyTime = 0;
    }
 
    double endProfile(const long long int time) {
        myTotalTime += time;
        myTotalBusyTime += myBusyTime;
        return time == 0 ? 100. : 100. * myBusyTime / time;
    }
#endif
 
private:
    Pool& myPool;
    FXMutex myMutex;
    FXCondition myCondition;
    std::list<Task*> myTasks;
    std::list<Task*> myCurrentTasks;
    bool myStopped;
#ifdef WORKLOAD_PROFILING
    int myCounter;
    long long int myBusyTime;
    long long int myTotalBusyTime;
    long long int myTotalTime;
#endif
};