Initial commit

cameradar_standalone/include/spdlog/details/async_log_helper.h

@@ -0,0 +1,313 @@
+//
+// Copyright(c) 2015 Gabi Melman.
+// Distributed under the MIT License (http://opensource.org/licenses/MIT)
+//
+
+// async log helper :
+// Process logs asynchronously using a back thread.
+//
+// If the internal queue of log messages reaches its max size,
+// then the client call will block until there is more room.
+//
+// If the back thread throws during logging, a spdlog::spdlog_ex exception
+// will be thrown in client's thread when tries to log the next message
+
+#pragma once
+
+#include <chrono>
+#include <thread>
+#include <functional>
+
+#include "../common.h"
+#include "../sinks/sink.h"
+#include "./mpmc_bounded_q.h"
+#include "./log_msg.h"
+#include "./format.h"
+#include "./os.h"
+
+namespace spdlog {
+namespace details {
+
+class async_log_helper {
+    // Async msg to move to/from the queue
+    // Movable only. should never be copied
+    enum class async_msg_type { log, flush, terminate };
+    struct async_msg {
+        std::string logger_name;
+        level::level_enum level;
+        log_clock::time_point time;
+        size_t thread_id;
+        std::string txt;
+        async_msg_type msg_type;
+
+        async_msg() = default;
+        ~async_msg() = default;
+
+        async_msg(async_msg&& other) SPDLOG_NOEXCEPT : logger_name(std::move(other.logger_name)),
+                                                       level(std::move(other.level)),
+                                                       time(std::move(other.time)),
+                                                       txt(std::move(other.txt)),
+                                                       msg_type(std::move(other.msg_type)) {}
+
+        async_msg(async_msg_type m_type) : msg_type(m_type){};
+
+        async_msg& operator=(async_msg&& other) SPDLOG_NOEXCEPT {
+            logger_name = std::move(other.logger_name);
+            level = other.level;
+            time = std::move(other.time);
+            thread_id = other.thread_id;
+            txt = std::move(other.txt);
+            msg_type = other.msg_type;
+            return *this;
+        }
+
+        // never copy or assign. should only be moved..
+        async_msg(const async_msg&) = delete;
+        async_msg& operator=(async_msg& other) = delete;
+
+        // construct from log_msg
+        async_msg(const details::log_msg& m)
+        : logger_name(m.logger_name)
+        , level(m.level)
+        , time(m.time)
+        , thread_id(m.thread_id)
+        , txt(m.raw.data(), m.raw.size())
+        , msg_type(async_msg_type::log) {}
+
+        // copy into log_msg
+        void
+        fill_log_msg(log_msg& msg) {
+            msg.clear();
+            msg.logger_name = logger_name;
+            msg.level = level;
+            msg.time = time;
+            msg.thread_id = thread_id;
+            msg.raw << txt;
+        }
+    };
+
+public:
+    using item_type = async_msg;
+    using q_type = details::mpmc_bounded_queue<item_type>;
+
+    using clock = std::chrono::steady_clock;
+
+    async_log_helper(
+        formatter_ptr formatter,
+        const std::vector<sink_ptr>& sinks,
+        size_t queue_size,
+        const async_overflow_policy overflow_policy = async_overflow_policy::block_retry,
+        const std::function<void()>& worker_warmup_cb = nullptr,
+        const std::chrono::milliseconds& flush_interval_ms = std::chrono::milliseconds::zero());
+
+    void log(const details::log_msg& msg);
+
+    // stop logging and join the back thread
+    ~async_log_helper();
+
+    void set_formatter(formatter_ptr);
+
+    void flush();
+
+private:
+    formatter_ptr _formatter;
+    std::vector<std::shared_ptr<sinks::sink>> _sinks;
+
+    // queue of messages to log
+    q_type _q;
+
+    bool _flush_requested;
+
+    bool _terminate_requested;
+
+    // last exception thrown from the worker thread
+    std::shared_ptr<spdlog_ex> _last_workerthread_ex;
+
+    // overflow policy
+    const async_overflow_policy _overflow_policy;
+
+    // worker thread warmup callback - one can set thread priority, affinity, etc
+    const std::function<void()> _worker_warmup_cb;
+
+    // auto periodic sink flush parameter
+    const std::chrono::milliseconds _flush_interval_ms;
+
+    // worker thread
+    std::thread _worker_thread;
+
+    void push_msg(async_msg&& new_msg);
+    // throw last worker thread exception or if worker thread is not active
+
+    void throw_if_bad_worker();
+
+    // worker thread main loop
+    void worker_loop();
+
+    // pop next message from the queue and process it. will set the last_pop to the pop time
+    // return false if termination of the queue is required
+    bool process_next_msg(log_clock::time_point& last_pop, log_clock::time_point& last_flush);
+
+    void handle_flush_interval(log_clock::time_point& now, log_clock::time_point& last_flush);
+
+    // sleep,yield or return immediatly using the time passed since last message as a hint
+    static void sleep_or_yield(const spdlog::log_clock::time_point& now,
+                               const log_clock::time_point& last_op_time);
+};
+}
+}
+
+///////////////////////////////////////////////////////////////////////////////
+// async_sink class implementation
+///////////////////////////////////////////////////////////////////////////////
+inline spdlog::details::async_log_helper::async_log_helper(
+    formatter_ptr formatter,
+    const std::vector<sink_ptr>& sinks,
+    size_t queue_size,
+    const async_overflow_policy overflow_policy,
+    const std::function<void()>& worker_warmup_cb,
+    const std::chrono::milliseconds& flush_interval_ms)
+: _formatter(formatter)
+, _sinks(sinks)
+, _q(queue_size)
+, _flush_requested(false)
+, _terminate_requested(false)
+, _overflow_policy(overflow_policy)
+, _worker_warmup_cb(worker_warmup_cb)
+, _flush_interval_ms(flush_interval_ms)
+, _worker_thread(&async_log_helper::worker_loop, this) {}
+
+// Send to the worker thread termination message(level=off)
+// and wait for it to finish gracefully
+inline spdlog::details::async_log_helper::~async_log_helper() {
+    try {
+        push_msg(async_msg(async_msg_type::terminate));
+        _worker_thread.join();
+    } catch (...) // don't crash in destructor
+    {}
+}
+
+// Try to push and block until succeeded
+inline void
+spdlog::details::async_log_helper::log(const details::log_msg& msg) {
+    push_msg(async_msg(msg));
+}
+
+// Try to push and block until succeeded
+inline void
+spdlog::details::async_log_helper::push_msg(details::async_log_helper::async_msg&& new_msg) {
+    throw_if_bad_worker();
+    if (!_q.enqueue(std::move(new_msg)) &&
+        _overflow_policy != async_overflow_policy::discard_log_msg) {
+        auto last_op_time = details::os::now();
+        auto now = last_op_time;
+        do {
+            now = details::os::now();
+            sleep_or_yield(now, last_op_time);
+        } while (!_q.enqueue(std::move(new_msg)));
+    }
+}
+
+inline void
+spdlog::details::async_log_helper::flush() {
+    push_msg(async_msg(async_msg_type::flush));
+}
+
+inline void
+spdlog::details::async_log_helper::worker_loop() {
+    try {
+        if (_worker_warmup_cb) _worker_warmup_cb();
+        auto last_pop = details::os::now();
+        auto last_flush = last_pop;
+        while (process_next_msg(last_pop, last_flush))
+            ;
+    } catch (const std::exception& ex) {
+        _last_workerthread_ex = std::make_shared<spdlog_ex>(
+            std::string("async_logger worker thread exception: ") + ex.what());
+    } catch (...) {
+        _last_workerthread_ex = std::make_shared<spdlog_ex>("async_logger worker thread exception");
+    }
+}
+
+// process next message in the queue
+// return true if this thread should still be active (no msg with level::off was received)
+inline bool
+spdlog::details::async_log_helper::process_next_msg(log_clock::time_point& last_pop,
+                                                    log_clock::time_point& last_flush) {
+    async_msg incoming_async_msg;
+    log_msg incoming_log_msg;
+
+    if (_q.dequeue(incoming_async_msg)) {
+        last_pop = details::os::now();
+        switch (incoming_async_msg.msg_type) {
+        case async_msg_type::flush: _flush_requested = true; break;
+
+        case async_msg_type::terminate:
+            _flush_requested = true;
+            _terminate_requested = true;
+            break;
+
+        default:
+            incoming_async_msg.fill_log_msg(incoming_log_msg);
+            _formatter->format(incoming_log_msg);
+            for (auto& s : _sinks) s->log(incoming_log_msg);
+        }
+        return true;
+    }
+
+    // Handle empty queue..
+    // This is the only place where the queue can terminate or flush to avoid losing messages
+    // already in the queue
+    else {
+        auto now = details::os::now();
+        handle_flush_interval(now, last_flush);
+        sleep_or_yield(now, last_pop);
+        return !_terminate_requested;
+    }
+}
+
+inline void
+spdlog::details::async_log_helper::handle_flush_interval(log_clock::time_point& now,
+                                                         log_clock::time_point& last_flush) {
+    auto should_flush =
+        _flush_requested || (_flush_interval_ms != std::chrono::milliseconds::zero() &&
+                             now - last_flush >= _flush_interval_ms);
+    if (should_flush) {
+        for (auto& s : _sinks) s->flush();
+        now = last_flush = details::os::now();
+        _flush_requested = false;
+    }
+}
+inline void
+spdlog::details::async_log_helper::set_formatter(formatter_ptr msg_formatter) {
+    _formatter = msg_formatter;
+}
+
+// sleep,yield or return immediatly using the time passed since last message as a hint
+inline void
+spdlog::details::async_log_helper::sleep_or_yield(
+    const spdlog::log_clock::time_point& now, const spdlog::log_clock::time_point& last_op_time) {
+    using std::chrono::milliseconds;
+    using namespace std::this_thread;
+
+    auto time_since_op = now - last_op_time;
+
+    // spin upto 1 ms
+    if (time_since_op <= milliseconds(1)) return;
+
+    // yield upto 10ms
+    if (time_since_op <= milliseconds(10)) return yield();
+
+    // sleep for half of duration since last op
+    if (time_since_op <= milliseconds(100)) return sleep_for(time_since_op / 2);
+
+    return sleep_for(milliseconds(100));
+}
+
+// throw if the worker thread threw an exception or not active
+inline void
+spdlog::details::async_log_helper::throw_if_bad_worker() {
+    if (_last_workerthread_ex) {
+        auto ex = std::move(_last_workerthread_ex);
+        throw * ex;
+    }
+}