tec_worker.hpp

Go to the documentation of this file.

// Time-stamp: <Last changed 2026-02-25 16:21:42 by magnolia>
/*----------------------------------------------------------------------
------------------------------------------------------------------------
Copyright (c) 2020-2026 The Emacs Cat (https://github.com/olddeuteronomy/tec).
 
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at
 
     http://www.apache.org/licenses/LICENSE-2.0
 
   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
------------------------------------------------------------------------
----------------------------------------------------------------------*/
#pragma once
 
#include <atomic>
#include <cmath>
#include <memory>
#include <mutex>
#include <typeindex>
#include <unordered_map>
#include <functional>
#include <thread>
 
#include "tec/tec_def.hpp" // IWYU pragma: keep
#include "tec/tec_trace.hpp"
#include "tec/tec_status.hpp"
#include "tec/tec_queue.hpp"
#include "tec/tec_message.hpp"
#include "tec/tec_daemon.hpp"
 
 
namespace tec {
 
template <typename TParams>
class Worker : public Daemon {
public:
    using Params = TParams; 
    using id_t = std::thread::id; 
    using Lock = std::lock_guard<std::mutex>; 
 
    using CallbackFunc = std::function<void(Worker<Params>*, const Message&)>;
 
protected:
    Params params_; 
 
private:
    struct Slot {
        Worker<Params>* worker; 
        CallbackFunc callback; 
 
        explicit Slot(Worker<Params>* w, CallbackFunc cb)
            : worker{w}
            , callback{cb}
        {}
 
        ~Slot() = default;
    };
 
    std::unordered_map<std::type_index, std::unique_ptr<Slot>> slots_; 
    std::mutex mtx_slots_; 
    Signal sig_running_; 
    Signal sig_inited_; 
    Signal sig_terminated_; 
    SafeQueue<Message> mq_; 
    Status status_; 
    std::atomic_bool flag_running_; 
    std::atomic_bool flag_exited_; 
    std::mutex mtx_thread_proc_; 
    std::thread thread_; 
    id_t thread_id_; 
 
public:
    explicit Worker(const Params& params)
        : Daemon()
        , params_{params}
        , flag_running_{false}
        , flag_exited_{false}
        , thread_id_{0}
    {}
 
    virtual ~Worker() {
        if (thread_.joinable()) {
            terminate();
        }
    }
 
    id_t id() const { return thread_id_; }
 
    constexpr const Params& params() const { return params_; }
 
    const Signal& sig_terminated() const override { return sig_terminated_; }
 
    void send(Message&& msg) override {
        TEC_ENTER("Worker::send");
        TEC_TRACE("Message [{}] sent.", name(msg));
        mq_.enqueue(std::move(msg));
    }
 
    Status make_request(Request&& req, Reply&& rep) override {
        Signal ready;
        Status status;
        Payload payload{
            &ready,
            &status,
            std::move(req),
            std::move(rep)};
        send({&payload});
        ready.wait();
        return status;
    }
 
    template <typename Derived, typename T>
    void register_callback(Derived* worker, void (Derived::*callback)(const Message& msg)) {
        Lock lk{mtx_slots_};
        TEC_ENTER("Worker::register_callback");
        //
        // Ensure Derived is actually derived from Worker<Params>.
        //
        static_assert(std::is_base_of_v<Worker<Params>, Derived>,
                      "Derived must inherit from tec::Worker");
        std::type_index ndx = std::type_index(typeid(T));
        //
        // Remove existing handler.
        //
        if (auto slot = slots_.find(ndx); slot != slots_.end()) {
            slots_.erase(ndx);
        }
        //
        // Set the slot.
        //
        slots_[ndx] = std::make_unique<Slot>(
            worker,
            [callback](Worker<Params>* worker, const Message& msg) {
                // Safely downcast to Derived.
                auto derived = dynamic_cast<Derived*>(worker);
                (derived->*callback)(msg);
            });
        TEC_TRACE("Callback {} registered.", typeid(T).name());
    }
 
protected:
    virtual void dispatch(const Message& msg) {
        auto ndx = std::type_index(msg.type());
        if (auto slot = slots_.find(ndx); slot != slots_.end()) {
            slot->second->callback(slot->second->worker, std::cref(msg));
        }
    }
 
private:
 
    template <typename Params>
    struct details {
        static void thread_proc(Worker<Params>& wt) {
            TEC_ENTER("Worker::thread_proc");
            // Signal termination on exit.
            Signal::OnExit on_terminating{&wt.sig_terminated_};
            //
            // Obtains thread ID and waits for the running signal.
            //
            wt.thread_id_ = std::this_thread::get_id();
            TEC_TRACE("thread {} created.", wt.id());
            wt.sig_running_.wait();
            TEC_TRACE("`sig_running' received.");
            //
            // Exit immediately if flagged.
            //
            if (wt.flag_exited_) {
                return;
            }
            //
            // Initialize the worker.
            //
            TEC_TRACE("on_init() called ...");
            wt.status_ = wt.on_init();
            TEC_TRACE("on_init() returned {}.", wt.status_);
            //
            // Signal initialization complete.
            //
            wt.sig_inited_.set();
            TEC_TRACE("`sig_inited' signalled.");
 
            if (wt.status_) {
                //
                // Process messages if initialized successfully.
                //
                TEC_TRACE("entering message loop.");
                bool stop = false;
                do {
                    auto msg = wt.mq_.dequeue();
                    TEC_TRACE("received Message [{}].", name(msg));
                    if (is_null(msg)) {
                        stop = true;
                        wt.flag_exited_ = true;
                    } else {
                        wt.dispatch(std::cref(msg));
                    }
                } while (!stop);
                TEC_TRACE("leaving message loop, {} message(s) left in queue...", wt.mq_.size());
            }
            //
            // Finalize if it was initialized successfully.
            //
            if (wt.status_) {
                TEC_TRACE("on_exit() called ...");
                wt.status_ = wt.on_exit();
                TEC_TRACE("on_exit() returned {}.", wt.status_);
            }
        }
    };
 
protected:
    virtual Status on_init() { return {}; }
 
    virtual Status on_exit() { return {}; }
 
protected:
    virtual Status create_thread() {
        thread_ = std::thread(details<Params>::thread_proc, std::ref(*this));
        return {};
    }
 
public:
    Status run() override {
        Lock lk{mtx_thread_proc_};
        TEC_ENTER("Worker::run");
        // Create the Daemon's thread in suspended state.
        auto status = create_thread();
        if (!status) {
            return status;
        }
        //
        // Resume the thread.
        //
        flag_running_ = true;
        sig_running_.set();
        TEC_TRACE("`sig_running' signalled.");
        //
        // Wait for thread initialization completed.
        //
        TEC_TRACE("waiting for `sig_inited' signalled ...");
        sig_inited_.wait();
        // Return a result of thread initialization.
        return status_;
    }
 
    Status terminate() override {
        Lock lk{mtx_thread_proc_};
        TEC_ENTER("Worker::terminate");
        if (!thread_.joinable()) {
            TEC_TRACE("WARNING: no thread exists!");
            return status_;
        }
        if (!flag_running_) {
            TEC_TRACE("Exiting the suspended thread...");
            flag_exited_ = true;
            sig_running_.set();
        }
        //
        // Send the null message on normal exiting.
        //
        if (status_ && !flag_exited_) {
            send(nullmsg());
            TEC_TRACE("QUIT sent.");
        }
        //
        // Wait for the thread to finish its execution.
        //
        TEC_TRACE("waiting for thread {} to finish ...", id());
        thread_.join();
        TEC_TRACE("thread {} finished OK.", id());
 
        return status_;
    }
 
}; // class Worker
 
} // namespace tec