refactors qk implementation and splits files

src/sysc/CMakeLists.txt

@@ -51,7 +51,10 @@ set(LIB_SOURCES
 )
 
 if(DEFINED  SC_VERSION_MAJOR AND SC_VERSION_MAJOR GREATER 2)
-    list(APPEND LIB_SOURCES tlm/scc/quantum_keeper.cpp)
+    list(APPEND LIB_SOURCES
+        tlm/scc/qk/global_time_keeper.cpp
+        tlm/scc/qk/sc_time_syncronizer.cpp
+    )
 endif()
 if(ENABLE_PYTHON4SC)
     find_package(pybind11 QUIET)

src/sysc/tlm/scc/qk/global_time_keeper.cpp

@@ -0,0 +1,94 @@
+#include "global_time_keeper.h"
+#include <tlm_utils/tlm_quantumkeeper.h>
+
+namespace tlm {
+namespace scc {
+namespace qk {
+///////////////////////////////////////////////////////////////////////////////
+//
+///////////////////////////////////////////////////////////////////////////////
+global_time_keeper::global_time_keeper() = default;
+
+global_time_keeper::~global_time_keeper() {
+    // shutting down time keeper thread
+    stop_it.store(true, std::memory_order_acq_rel);
+    update.notify_all();
+}
+// this function will be called from the systemc thread
+void global_time_keeper::start() {
+    std::thread t{&global_time_keeper::sync_local_times, this};
+    t.detach();
+    started = true;
+}
+// this function will be called from the systemc thread
+size_t global_time_keeper::get_channel_index() {
+    if(started)
+        throw std::runtime_error("global_time_keeper already started");
+    client_coms_channels.emplace_back(client_coms_channels.size());
+    update_it = true;
+    return client_coms_channels.size() - 1;
+}
+
+// runs in a background thread
+void global_time_keeper::sync_local_times() {
+    std::unique_lock<std::mutex> lock{upd_mtx};
+    while(true) {
+        update.wait_for(lock, std::chrono::milliseconds(1),
+                        [this]() -> bool { return update_it.load(std::memory_order_relaxed) || stop_it.load(std::memory_order_relaxed); });
+        if(update_it.exchange(false)) {
+#ifdef DEBUG_MT_SCHEDULING
+            SCCTRACEALL("global_time_keeper::sync_local_times") << "update loop";
+#endif
+            uint64_t min_local_time = std::numeric_limits<uint64_t>::max();
+            bool tail = false;
+            while(auto res = sc_coms_channel.client2time_keeper.front()) {
+                sc_coms_channel.thread_local_time = res->time_tick;
+                sc_coms_channel.client2time_keeper.pop();
+            }
+            for(size_t i = 0; i < client_coms_channels.size(); ++i) {
+                auto& client_coms_channel = client_coms_channels[i];
+                bool has_task = false;
+                bool has_entries = false;
+                while(auto res = client_coms_channel.client2time_keeper.front()) {
+                    has_entries = true;
+                    client_coms_channel.thread_local_time = res->time_tick;
+                    if(res->task.valid()) {
+#ifdef DEBUG_MT_SCHEDULING
+                        SCCTRACEALL("global_time_keeper::sync_local_times")
+                            << "forwarding task of client " << client_coms_channel.my_id << " with timestamp t=" << res->time_tick;
+#endif
+                        pending_tasks.emplace(client_coms_channel.my_id, res->time_tick, std::move(res->task));
+                        has_task = true;
+                    }
+                    client_coms_channel.client2time_keeper.pop();
+                }
+                if(has_entries)
+                    client_coms_channel.waiting4sc = has_task;
+                if(!client_coms_channel.waiting4sc)
+                    min_local_time = std::min(client_coms_channel.thread_local_time, min_local_time);
+#ifdef DEBUG_MT_SCHEDULING
+                if(has_entries) {
+                    SCCTRACEALL("global_time_keeper::sync_local_times")
+                        << "thread_local_time[" << i << "]=" << sc_core::sc_time::from_value(client_coms_channel.thread_local_time)
+                        << (client_coms_channel.waiting4sc ? " (waiting)" : " (running)");
+                }
+#endif
+            }
+            window_min_time = min_local_time;
+            sc_coms_channel.time_keeper2client.store(min_local_time);
+            auto window_max_time = min_local_time + std::max(tlm::tlm_global_quantum::instance().get().value(), sc_time_step.value());
+            for(auto& client_coms_channel : client_coms_channels) {
+                client_coms_channel.time_keeper2client.store(window_max_time);
+            }
+#ifdef DEBUG_MT_SCHEDULING
+            SCCTRACEALL("global_time_keeper::sync_local_times") << "window_min_time=" << window_min_time;
+#endif
+        } else if(stop_it.load()) {
+            break;
+        }
+    }
+}
+
+} // namespace qk
+} // namespace scc
+} // namespace tlm

src/sysc/tlm/scc/qk/global_time_keeper.h

@@ -0,0 +1,125 @@
+#ifndef __SCC_TLM_QK_GLOBAL_TIME_KEEPER_H__
+#define __SCC_TLM_QK_GLOBAL_TIME_KEEPER_H__
+
+#include "types.h"
+#include <deque>
+
+namespace tlm {
+namespace scc {
+namespace qk {
+///////////////////////////////////////////////////////////////////////////////
+//
+///////////////////////////////////////////////////////////////////////////////
+struct sc_time_syncronizer;
+/**
+ * @brief the global time keeper, a singleton
+ * @details The global time keeper receives actual time stamps of all client threads
+ * and calculates the max time they are alloed to advance beyond systemc. All calculations are done
+ * based on ticks which is the minimal time step a simulator can do. The SystemC thread is special
+ * in that it is time wise always the last/slowest thread.
+ *
+ */
+struct global_time_keeper {
+    friend class sc_time_syncronizer;
+    /**
+     * @brief the maximum timestep the simulator is allowed to do if there are no future events
+     *
+     */
+    const sc_core::sc_time sc_time_step = 1_ms;
+    /**
+     * @brief the singleton getter
+     *
+     * @return global_time_keeper&
+     */
+    static global_time_keeper& get() {
+        static global_time_keeper keeper;
+        return keeper;
+    }
+    /**
+     * @brief Get the minimum time ticks of all client threads (not SystemC)
+     *
+     * @return uint64_t the absolute number of ticks
+     */
+    inline uint64_t get_min_time_ticks() { return window_min_time; }
+    /**
+     * @brief Get the maximum time ticks a client thread is allowed to advance
+     *
+     * @param idx the id of the client thread, to be obtained using get_channel_index()
+     * @return uint64_t the absolute number of maximum ticks, if no new information it returns -1
+     */
+    inline uint64_t get_max_time_ticks(size_t idx) { return client_coms_channels[idx].time_keeper2client.load(); }
+    /**
+     * @brief updates the global time keeper with the local time ticks of a clinet thread
+     *
+     * @param idx the id of the client thread, to be obtained using get_channel_index()
+     * @param tick the absolute number of actual ticks
+     */
+    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();
+    }
+    /**
+     * @brief updates the global time keeper with the local time ticks of a clinet thread and schedules a task at this new time point
+     *
+     * @param idx the id of the client thread, to be obtained using get_channel_index()
+     * @param tick  the absolute number of actual ticks
+     * @param task the task to be executed in the SystemC kernel. It shall return the time it used for execution
+     */
+    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();
+    }
+    /**
+     * @brief Get the maximum sc time ticks a client thread is allowed to advance
+     *
+     * @return uint64_t  the absolute number of ticks
+     */
+    inline uint64_t get_max_sc_time_ticks() { return sc_coms_channel.time_keeper2client.load(); }
+    /**
+     * @brief updates the global time keeper with the local time ticks of the SystemC thread
+     *
+     * @param tick the absolute number of actual SystemC ticks
+     */
+    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_SCHEDULING
+        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_uint64_t window_min_time;
+    std::mutex upd_mtx;
+    std::condition_variable update;
+    thread_comms_channel sc_coms_channel{-1ULL};
+    std::deque<thread_comms_channel> client_coms_channels;
+    rigtorp::SPSCQueue<std::tuple<size_t, uint64_t, callback_task>> pending_tasks{1024};
+    bool started = false;
+};
+} // namespace qk
+} // namespace scc
+} // namespace tlm
+#endif // __SCC_TLM_QK_GLOBAL_TIME_KEEPER_H__

src/sysc/tlm/scc/qk/sc_time_syncronizer.cpp

@@ -0,0 +1,109 @@
+#include "sc_time_syncronizer.h"
+#include "global_time_keeper.h"
+#include <systemc>
+#include <tlm_utils/tlm_quantumkeeper.h>
+
+namespace tlm {
+namespace scc {
+namespace qk {
+sc_time_syncronizer::sc_time_syncronizer(global_time_keeper& gtk)
+: gtk(gtk) {
+    sc_core::sc_register_stage_callback(*this, sc_core::SC_PRE_TIMESTEP | sc_core::SC_POST_END_OF_ELABORATION);
+    sc_core::sc_spawn_options opt;
+    opt.spawn_method();
+    sc_core::sc_spawn([this]() { method_callback(); }, nullptr, &opt);
+}
+
+void sc_time_syncronizer::method_callback() {
+    auto res = this->gtk.get_max_sc_time_ticks();
+    if(res != std::numeric_limits<uint64_t>::max()) {
+        sc_max_time.store(res, std::memory_order_seq_cst);
+    }
+    auto& pending_tasks = this->gtk.pending_tasks;
+    if(pending_tasks.size()) {
+#ifdef DEBUG_MT_SCHEDULING
+        SCCTRACEALL(__PRETTY_FUNCTION__) << "updating pending tasks";
+#endif
+        while(pending_tasks.size()) {
+            auto res = pending_tasks.front();
+            auto idx = std::get<0>(*res);
+            auto t = std::get<1>(*res);
+            auto& peq = sc_task_que[idx];
+            if(t > sc_core::sc_time_stamp().value()) {
+                auto d = sc_core::sc_time::from_value(t) - sc_core::sc_time_stamp();
+#ifdef DEBUG_MT_SCHEDULING
+                SCCDEBUG(__PRETTY_FUNCTION__) << "scheduling task from client " << idx << " with t=" << t << " with delay " << d;
+#endif
+                peq.notify(std::move(std::get<2>(*res)), d);
+            } else {
+#ifdef DEBUG_MT_SCHEDULING
+                SCCDEBUG(__PRETTY_FUNCTION__) << "scheduling task from client " << idx << " with t=" << t << " now";
+#endif
+                peq.notify(std::move(std::get<2>(*res)), sc_core::SC_ZERO_TIME);
+            }
+            pending_tasks.pop();
+        }
+        sc_core::next_trigger(sc_core::SC_ZERO_TIME);
+    } else if(sc_core::sc_get_curr_simcontext()->pending_activity_at_current_time()) {
+#ifdef DEBUG_MT_SCHEDULING
+        SCCTRACEALL(__PRETTY_FUNCTION__) << "yield to next delta cycle";
+#endif
+        sc_core::next_trigger(sc_core::SC_ZERO_TIME);
+    } else {
+        auto time_to_next_evt = sc_core::sc_time_to_pending_activity(sc_core::sc_get_curr_simcontext());
+        if(!sc_is_free_running) {
+            auto min_time = sc_core::sc_time::from_value(gtk.get_max_sc_time_ticks());
+            auto abs_time_to_next_evt = sc_core::sc_time_stamp() + time_to_next_evt;
+            if(min_time < abs_time_to_next_evt) {
+                if(min_time > sc_core::sc_time_stamp()) {
+                    sc_core::next_trigger(min_time - sc_core::sc_time_stamp());
+                } else {
+                    // slow down systemc execution to be the slower than client threads
+                    std::this_thread::yield();
+                    // std::this_thread::sleep_for(std::chrono::microseconds{1});
+                    sc_core::next_trigger(sc_core::SC_ZERO_TIME); // play it again, Sam
+                }
+            } else {
+#ifdef DEBUG_MT_SCHEDULING
+                SCCTRACEALL(__PRETTY_FUNCTION__) << "advancing SC time lockstepped to " << abs_time_to_next_evt;
+#endif
+                sc_core::next_trigger(time_to_next_evt);
+            }
+        } else {
+            // all threads are blocked by SystemC, so we can run freely but to avoid starving the kernel we only proceed to the
+            // time of the slowest client thread as this one is waiting to be served
+            auto next_time_point = sc_core::sc_time_stamp() + time_to_next_evt;
+            if(next_time_point.value() == std::numeric_limits<uint64_t>::max()) {
+                time_to_next_evt = tlm_utils::tlm_quantumkeeper::get_global_quantum();
+                if(time_to_next_evt == sc_core::SC_ZERO_TIME)
+                    time_to_next_evt = 1_us;
+            }
+#ifdef DEBUG_MT_SCHEDULING
+            SCCTRACEALL(__PRETTY_FUNCTION__) << "advancing SC time free running to " << next_time_point;
+#endif
+            sc_core::next_trigger(time_to_next_evt);
+        }
+    }
+}
+
+void sc_time_syncronizer::stage_callback(const sc_core::sc_stage& stage) {
+    switch(stage) {
+    case sc_core::SC_PRE_TIMESTEP: {
+        sc_core::sc_time next_time;
+        if(sc_core::sc_get_curr_simcontext()->next_time(next_time)) {
+            gtk.update_sc_time_ticks(next_time.value());
+        }
+#ifdef DEBUG_MT_SCHEDULING
+        SCCTRACEALL(__PRETTY_FUNCTION__) << "advancing SystemC kernel time to " << next_time << ", get_min_time()=" << get_min_time();
+#endif
+    } break;
+    case sc_core::SC_POST_END_OF_ELABORATION:
+        gtk.start();
+        break;
+    default:
+        break;
+    }
+}
+} // namespace qk
+} // namespace scc
+} // namespace tlm