manticore.hpp

Go to the documentation of this file.

#ifndef MANTICORE_MANTICORE_HPP
#define MANTICORE_MANTICORE_HPP
 
#include <mutex>
#include <condition_variable>
#include <functional>
#include <string>
#include <stdexcept>
 
namespace manticore {
 
class Executor {
    std::mutex run_lock;
    std::condition_variable cv;
 
    size_t nthreads;
    size_t ncomplete;
    std::string fallback_error;
    std::string error_message;
 
    enum class Status : char { FREE, PRIMED, FINISHED };
    Status status;
    std::function<void()> fun;
 
    bool initialized = false;
    bool done() const { 
        return ncomplete == nthreads;
    }
 
public:
    void initialize(size_t n, std::string e) {
        nthreads = n;
        ncomplete = 0;
        fallback_error = std::move(e);
        error_message.clear();
        status = Status::FREE;
        initialized = true;
    }
 
    void initialize(size_t n) {
        initialize(n, "failed main thread execution");
    }
 
    void finish_thread(bool notify = true) {
        // Lock everything before bumping ncomplete to avoid problems with
        // simultaneous writes. We use a lock rather than an atomic variable to
        // ensure that the change propagates correctly to the main thread when
        // it calls done() after notification, otherwise the order of events
        // across multiple threads is not guaranteed.
        {
            std::lock_guard lck(run_lock);
            ++ncomplete;
        }
 
        if (notify) {
            cv.notify_all(); // possibly trigger loop exit in listen(). 
        }
    }
 
public:
    template<class Function_>
    void run(Function_ f) {
        if (!initialized) {
            f();
            return;
        }
 
        // Waiting until the main thread executor is free,
        // and then assigning it a task.
        std::unique_lock lk(run_lock);
        cv.wait(lk, [&]{ return status == Status::FREE; });
 
        fun = std::move(f);
        status = Status::PRIMED;
 
        // Notifying the main thread that there is a task. Only the
        // main thread waits on PRIMED, so other works should not proceed.
        lk.unlock();
        cv.notify_all();
 
        lk.lock();
        cv.wait(lk, [&]{ return status == Status::FINISHED; });
 
        // Making a copy of any error message so we can use it for throwing
        // after the unlock. Also clearing the error message for the next thread.
        auto errcopy = error_message;
        error_message.clear();
 
        // Unblocking other worker threads, if any are waiting.
        status = Status::FREE;
        lk.unlock();
        cv.notify_all();
 
        if (!errcopy.empty()) {
            throw std::runtime_error(errcopy);
        }
    }
 
public:
    void listen() {
        while (1) {
            std::unique_lock lk(run_lock);
 
            cv.wait(lk, [&]{ return status == Status::PRIMED || done(); });
            if (done()) {
                break;
            }
 
            try {
                fun();
            } catch (std::exception& x) {
                // No throw, we need to make sure we notify the worker of (failed) completion.
                error_message = x.what();
            } catch (...) {
                // For everything else.
                error_message = fallback_error;
            }
 
            status = Status::FINISHED;
 
            // Unlock before notifying, see example in https://en.cppreference.com/w/cpp/thread/condition_variable
            lk.unlock();
            cv.notify_all();
        }
 
        initialized = false;
    }
};
 
}
 
#endif