standalone_executor.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <kcenon/logger/core/thread_integration_detector.h>
#include <kcenon/logger/logger_export.h>
 
#if LOGGER_HAS_IEXECUTOR
 
#if __has_include(<kcenon/common/interfaces/executor_interface.h>)
#include <kcenon/common/interfaces/executor_interface.h>
#include <kcenon/common/patterns/result.h>
#elif __has_include(<common/interfaces/executor_interface.h>)
#include <common/interfaces/executor_interface.h>
#include <common/patterns/result.h>
#endif
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
 
namespace kcenon::logger::integration {
 
class function_job : public common::interfaces::IJob {
public:
    explicit function_job(std::function<void()> func, std::string name = "logger_task")
        : func_(std::move(func)), name_(std::move(name)) {}
 
    common::VoidResult execute() override {
        try {
            func_();
            return common::ok();
        } catch (const std::exception& ex) {
            return common::VoidResult(common::error_info{
                -1, ex.what(), "standalone_executor"
            });
        } catch (...) {
            return common::VoidResult(common::error_info{
                -1, "Unknown exception during job execution", "standalone_executor"
            });
        }
    }
 
    std::string get_name() const override { return name_; }
 
private:
    std::function<void()> func_;
    std::string name_;
};
 
class LOGGER_SYSTEM_API standalone_executor : public common::interfaces::IExecutor {
public:
    explicit standalone_executor(std::size_t queue_size = 8192,
                                  std::string name = "standalone_executor");
 
    ~standalone_executor() override;
 
    // Non-copyable, non-movable
    standalone_executor(const standalone_executor&) = delete;
    standalone_executor& operator=(const standalone_executor&) = delete;
    standalone_executor(standalone_executor&&) = delete;
    standalone_executor& operator=(standalone_executor&&) = delete;
 
    void start();
 
    // ===== IExecutor Interface Implementation =====
 
    common::Result<std::future<void>> execute(
        std::unique_ptr<common::interfaces::IJob>&& job) override;
 
    common::Result<std::future<void>> execute_delayed(
        std::unique_ptr<common::interfaces::IJob>&& job,
        std::chrono::milliseconds delay) override;
 
    size_t worker_count() const override;
 
    bool is_running() const override;
 
    size_t pending_tasks() const override;
 
    void shutdown(bool wait_for_completion = true) override;
 
    // ===== Additional Methods =====
 
    [[nodiscard]] std::string get_name() const noexcept { return name_; }
 
    [[nodiscard]] std::uint64_t dropped_count() const noexcept;
 
private:
    struct pending_task {
        std::unique_ptr<common::interfaces::IJob> job;
        std::shared_ptr<std::promise<void>> completion_promise;
        std::chrono::steady_clock::time_point execute_after;
    };
 
    void worker_loop();
 
    void drain_queue();
 
    bool enqueue_task(pending_task&& task);
 
private:
    const std::string name_;
    const std::size_t queue_size_;
 
    std::queue<pending_task> queue_;
    mutable std::mutex queue_mutex_;
    std::condition_variable queue_cv_;
 
    std::thread worker_thread_;
    std::atomic<bool> running_{false};
    std::atomic<bool> stop_requested_{false};
    std::atomic<std::uint64_t> dropped_count_{0};
};
 
class LOGGER_SYSTEM_API standalone_executor_factory {
public:
    static std::shared_ptr<common::interfaces::IExecutor> create(
        std::size_t queue_size = 8192,
        const std::string& name = "standalone_executor");
};
 
} // namespace kcenon::logger::integration
 
#endif // LOGGER_HAS_IEXECUTOR