quantum_keeper.h

#ifndef __SCC_TLM_QUANTUMKEEPER_H__
#define __SCC_TLM_QUANTUMKEEPER_H__
 
#include "rigtorp/SPSCQueue.h"
#include "sysc/kernel/sc_wait.h"
#include <atomic>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <future>
#include <limits>
#include <mutex>
#include <nonstd/optional.hpp>
#include <scc/async_thread.h>
#include <scc/peq.h>
#include <scc/report.h>
#include <scc/utilities.h>
#include <string>
#include <sysc/kernel/sc_object.h>
#include <sysc/kernel/sc_process.h>
#include <sysc/kernel/sc_simcontext.h>
#include <sysc/kernel/sc_spawn.h>
#include <sysc/kernel/sc_spawn_options.h>
#include <sysc/kernel/sc_time.h>
#include <systemc>
#include <thread>
#include <tlm>
#include <tlm_utils/tlm_quantumkeeper.h>
 
#if SC_VERSION_MAJOR < 3
#error "Multithreaded quantum keeper is only supported with SystemC 3.0 and newer"
#else
// #define DEBUG_MT_SCHDULING
 
namespace tlm {
namespace scc {
#ifdef __cpp_lib_hardware_interference_size
static constexpr size_t kCacheLineSize = std::hardware_destructive_interference_size;
#else
static constexpr size_t kCacheLineSize = 64;
#endif
//
struct quantumkeeper : public tlm_utils::tlm_quantumkeeper {
    using base = tlm_utils::tlm_quantumkeeper;
 
    quantumkeeper() {}
 
    virtual ~quantumkeeper() {}
 
    inline void check_and_sync(sc_core::sc_time core_inc) {
        // devirtualized inc()
        m_local_time += core_inc;
        if((sc_core::sc_time_stamp() + m_local_time) >= m_next_sync_point) {
            // devirtualized sync()
            ::sc_core::wait(m_local_time);
            m_local_time = sc_core::SC_ZERO_TIME;
            m_next_sync_point = sc_core::sc_time_stamp() + compute_local_quantum();
        }
    }
    sc_core::sc_time get_local_absolute_time() const { return base::get_current_time(); }
    inline void execute_on_sysc(std::function<void(void)> fct) { execute_on_sysc(fct, sc_core::sc_time_stamp()); }
    inline void execute_on_sysc(std::function<void(void)>& fct, sc_core::sc_time when) {
        if(when > sc_core::SC_ZERO_TIME)
            wait(when);
        fct();
    }
};
//
using callback_fct = sc_core::sc_time(void);
using callback_task = std::packaged_task<callback_fct>;
 
struct comms_entry {
    uint64_t time_tick;
    callback_task task;
};
//

struct thread_comms_channel {
    thread_comms_channel(uint64_t my_id)
    : my_id(my_id)
    , client2time_keeper(7)
    , time_keeper2client(0) {}
    thread_comms_channel(thread_comms_channel const& o)
    : my_id(o.my_id)
    , client2time_keeper(o.client2time_keeper.capacity())
    , time_keeper2client(0)
    , thread_local_time(o.thread_local_time) {
        assert(o.client2time_keeper.size() == 0);
    };
 
    rigtorp::SPSCQueue<comms_entry> client2time_keeper;
    std::atomic<uint64_t> time_keeper2client;
 
    const uint64_t my_id;
    uint64_t thread_local_time;
};
//
struct sc_time_syncronizer;
struct global_time_keeper {
    friend class sc_time_syncronizer;
    const sc_core::sc_time sc_time_step = 1_ms;
    static global_time_keeper& get() {
        static global_time_keeper keeper;
        return keeper;
    }
    inline uint64_t get_min_time_ticks() { return window_min_time; }
    inline uint64_t get_max_time_ticks(size_t idx) { return client_coms_channels[idx].time_keeper2client.load(); }
    inline void update_time_ticks(size_t idx, uint64_t tick) {
        client_coms_channels[idx].client2time_keeper.push(std::move(comms_entry{tick, std::move(callback_task())}));
        update_it.store(true);
        std::unique_lock<std::mutex> lk(upd_mtx);
        update.notify_all();
    }
    inline void schedule_task(size_t idx, std::packaged_task<sc_core::sc_time(void)>&& task, uint64_t when) {
        client_coms_channels[idx].client2time_keeper.push(std::move(comms_entry{when, std::move(task)}));
        update_it.store(true);
        std::unique_lock<std::mutex> lk(upd_mtx);
        update.notify_all();
    }
    inline uint64_t get_max_sc_time_ticks() { return sc_coms_channel.time_keeper2client.load(); }
    inline void update_sc_time_ticks(uint64_t tick) {
        sc_coms_channel.client2time_keeper.push(std::move(comms_entry{tick, std::move(callback_task())}));
#ifdef DEBUG_MT_SCHDULING
        SCCTRACEALL("global_time_keeper::update_sc_time_ticks") << "sc_time=" << sc_core::sc_time::from_value(tick);
#endif
        update_it.store(true);
        std::unique_lock<std::mutex> lk(upd_mtx);
        update.notify_all();
    }
 
protected:
    global_time_keeper();
 
    global_time_keeper(global_time_keeper const&) = delete;
 
    global_time_keeper(global_time_keeper&&) = delete;
 
    ~global_time_keeper();
 
    void start();
 
    size_t get_channel_index();
 
    void sync_local_times();
 
    std::atomic<bool> stop_it{false};
    std::atomic<bool> update_it{false};
    std::atomic_bool all_threads_blocked{true};
    std::atomic_uint64_t window_min_time;
    std::mutex upd_mtx;
    std::condition_variable update;
    thread_comms_channel sc_coms_channel{0};
    std::deque<thread_comms_channel> client_coms_channels;
    rigtorp::SPSCQueue<std::tuple<size_t, uint64_t, callback_task>> pending_tasks{1024};
    bool started = false;
};
//

struct sc_time_syncronizer : sc_core::sc_object, sc_core::sc_stage_callback_if, sc_core::sc_process_host {
    ~sc_time_syncronizer() = default;
    static sc_time_syncronizer& get() {
        static sc_time_syncronizer keeper(global_time_keeper::get());
        return keeper;
    }
    size_t get_channel_index() {
        auto res = gtk.get_channel_index();
        sc_task_que.emplace_back();
        blocked_channels.emplace_back(true);
        auto& sctq = sc_task_que[sc_task_que.size() - 1];
        sc_core::sc_spawn(
            [this, &sctq]() {
                wait(sc_core::SC_ZERO_TIME);
                while(true) {
                    sctq.get()();
                }
            },
            sc_core::sc_gen_unique_name("peq_cb", false));
 
        return res;
    }
    inline sc_core::sc_time get_sc_kernel_time() { return sc_core::sc_time::from_value(sc_max_time.load(std::memory_order_seq_cst)); }
 
    inline void notify_channel_blocked(size_t idx, bool is_blocked) {
        assert(idx < blocked_channels.size());
        blocked_channels[idx] = is_blocked;
        sc_is_free_running = !std::any_of(std::begin(blocked_channels), std::end(blocked_channels), [](bool b) { return !b; });
    }
 
private:
    sc_time_syncronizer(global_time_keeper& gtk);
    void method_callback();
    void stage_callback(const sc_core::sc_stage& stage) override;
    sc_core::sc_time get_min_time() { return sc_core::sc_time::from_value(gtk.get_min_time_ticks()); }
    global_time_keeper& gtk;
    sc_core::sc_vector<::scc::peq<callback_task>> sc_task_que;
    std::atomic_int64_t sc_max_time{0};
    sc_core::sc_process_handle method_handle;
    std::vector<bool> blocked_channels;
    bool sc_is_free_running{true};
};
//

struct quantumkeeper_mt {
    quantumkeeper_mt()
    : gtk_idx(sc_time_syncronizer::get().get_channel_index()) {
        sc_core::sc_spawn_options opt;
        opt.spawn_method();
        opt.set_sensitivity(&keep_alive);
    }
    virtual ~quantumkeeper_mt() = default;
 
    void run_thread(std::function<sc_core::sc_time()> f) {
        sc_time_syncronizer::get().notify_channel_blocked(gtk_idx, false);
        thread_executor.start(f);
        wait(thread_executor.thread_finish_event());
    }
    inline void inc(sc_core::sc_time const& core_inc) {
        local_absolute_time += core_inc;
        auto res = global_time_keeper::get().get_max_time_ticks(gtk_idx);
        if(res != std::numeric_limits<uint64_t>::max())
            local_time_ticks_limit = res;
    }
    inline void check_and_sync(sc_core::sc_time const& core_inc) {
        inc(core_inc);
        global_time_keeper::get().update_time_ticks(gtk_idx, local_absolute_time.value());
        // wait until the SystemC thread advanced to the same time then we are minus the global quantum
        while(local_absolute_time.value() > local_time_ticks_limit) {
            std::this_thread::yield(); // should do the same than __builtin_ia32_pause() or _mm_pause() on MSVC
            auto res = global_time_keeper::get().get_max_time_ticks(gtk_idx);
            if(res != std::numeric_limits<uint64_t>::max())
                local_time_ticks_limit = res;
        }
    }
    inline void execute_on_sysc(std::function<void(void)> fct) { execute_on_sysc(fct, local_absolute_time); }
    inline void execute_on_sysc(std::function<void(void)>& fct, sc_core::sc_time when) {
        callback_task task([this, &fct]() {
            sc_time_syncronizer::get().notify_channel_blocked(gtk_idx, true);
            auto t0 = sc_core::sc_time_stamp();
            fct();
            sc_time_syncronizer::get().notify_channel_blocked(gtk_idx, false);
            return t0 - sc_core::sc_time_stamp();
        });
        auto ret = task.get_future();
        global_time_keeper::get().schedule_task(gtk_idx, std::move(task), when.value());
        auto duration = ret.get();
        check_and_sync(duration);
    }
    sc_core::sc_time get_local_time() const { return local_absolute_time - get_current_sc_time(); }
    sc_core::sc_time get_local_absolute_time() const { return local_absolute_time; }
    sc_core::sc_time get_current_sc_time() const { return sc_time_syncronizer::get().get_sc_kernel_time(); }
    void reset() { reset(get_current_sc_time()); }
    void reset(sc_core::sc_time abs_time) { local_absolute_time = sc_core::sc_time_stamp(); }
 
    void unblock_thread() { global_time_keeper::get().update_time_ticks(gtk_idx, local_absolute_time.value()); }
 
protected:
    size_t gtk_idx;
    ::scc::async_thread thread_executor;
    sc_core::sc_event keep_alive;
    sc_core::sc_time local_absolute_time;
    uint64_t local_time_ticks_limit{0};
};
} // namespace scc
} // namespace tlm
#endif
#endif