1 files changed, 399 insertions, 0 deletions
diff --git a/geom_matching/wasserstein/include/spdlog/details/async_log_helper.h b/geom_matching/wasserstein/include/spdlog/details/async_log_helper.h
new file mode 100644
index 0000000..6145dfa
--- /dev/null
+++ b/geom_matching/wasserstein/include/spdlog/details/async_log_helper.h
@@ -0,0 +1,399 @@
+//
+// 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.
+//
+
+#pragma once
+
+#include "spdlog/common.h"
+#include "spdlog/sinks/sink.h"
+#include "spdlog/details/mpmc_bounded_q.h"
+#include "spdlog/details/log_msg.h"
+#include "spdlog/details/os.h"
+#include "spdlog/formatter.h"
+
+#include <chrono>
+#include <exception>
+#include <functional>
+#include <memory>
+#include <string>
+#include <thread>
+#include <utility>
+#include <vector>
+
+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;
+        size_t msg_id;
+
+        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)),
+                    thread_id(other.thread_id),
+                    txt(std::move(other.txt)),
+                    msg_type(std::move(other.msg_type)),
+                    msg_id(other.msg_id)
+        {}
+
+        async_msg(async_msg_type m_type):
+            level(level::info),
+            thread_id(0),
+            msg_type(m_type),
+            msg_id(0)
+        {}
+
+        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;
+            msg_id = other.msg_id;
+            return *this;
+        }
+
+        // never copy or assign. should only be moved..
+        async_msg(const async_msg&) = delete;
+        async_msg& operator=(const async_msg& other) = delete;
+
+        // construct from log_msg
+        async_msg(const details::log_msg& m):
+            level(m.level),
+            time(m.time),
+            thread_id(m.thread_id),
+            txt(m.raw.data(), m.raw.size()),
+            msg_type(async_msg_type::log),
+            msg_id(m.msg_id)
+        {
+#ifndef SPDLOG_NO_NAME
+            logger_name = *m.logger_name;
+#endif
+        }
+
+
+        // copy into log_msg
+        void fill_log_msg(log_msg &msg)
+        {
+            msg.logger_name = &logger_name;
+            msg.level = level;
+            msg.time = time;
+            msg.thread_id = thread_id;
+            msg.raw << txt;
+            msg.msg_id = msg_id;
+        }
+    };
+
+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 log_err_handler err_handler,
+                     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(),
+                     const std::function<void()>& worker_teardown_cb = nullptr);
+
+    void log(const details::log_msg& msg);
+
+    // stop logging and join the back thread
+    ~async_log_helper();
+
+    void set_formatter(formatter_ptr);
+
+    void flush(bool wait_for_q);
+
+    void set_error_handler(spdlog::log_err_handler err_handler);
+
+private:
+    formatter_ptr _formatter;
+    std::vector<std::shared_ptr<sinks::sink>> _sinks;
+
+    // queue of messages to log
+    q_type _q;
+
+    log_err_handler _err_handler;
+
+    bool _flush_requested;
+
+    bool _terminate_requested;
+
+
+    // 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 teardown callback
+    const std::function<void()> _worker_teardown_cb;
+
+    // worker thread
+    std::thread _worker_thread;
+
+    void push_msg(async_msg&& new_msg);
+
+    // 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 immediately 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);
+
+    // wait until the queue is empty
+    void wait_empty_q();
+
+};
+}
+}
+
+///////////////////////////////////////////////////////////////////////////////
+// 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,
+    log_err_handler err_handler,
+    const async_overflow_policy overflow_policy,
+    const std::function<void()>& worker_warmup_cb,
+    const std::chrono::milliseconds& flush_interval_ms,
+    const std::function<void()>& worker_teardown_cb):
+    _formatter(formatter),
+    _sinks(sinks),
+    _q(queue_size),
+    _err_handler(err_handler),
+    _flush_requested(false),
+    _terminate_requested(false),
+    _overflow_policy(overflow_policy),
+    _worker_warmup_cb(worker_warmup_cb),
+    _flush_interval_ms(flush_interval_ms),
+    _worker_teardown_cb(worker_teardown_cb),
+    _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 (if the policy is not to discard when the queue is full)
+inline void spdlog::details::async_log_helper::log(const details::log_msg& msg)
+{
+    push_msg(async_msg(msg));
+}
+
+inline void spdlog::details::async_log_helper::push_msg(details::async_log_helper::async_msg&& new_msg)
+{
+    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)));
+    }
+}
+
+// optionally wait for the queue be empty and request flush from the sinks
+inline void spdlog::details::async_log_helper::flush(bool wait_for_q)
+{
+    push_msg(async_msg(async_msg_type::flush));
+    if (wait_for_q)
+        wait_empty_q(); //return only make after the above flush message was processed
+}
+
+inline void spdlog::details::async_log_helper::worker_loop()
+{
+    if (_worker_warmup_cb) _worker_warmup_cb();
+    auto last_pop = details::os::now();
+    auto last_flush = last_pop;
+    auto active = true;
+    while (active)
+    {
+        try
+        {
+            active = process_next_msg(last_pop, last_flush);
+        }
+        catch (const std::exception &ex)
+        {
+            _err_handler(ex.what());
+        }
+        catch (...)
+        {
+            _err_handler("Unknown exception");
+        }
+    }
+    if (_worker_teardown_cb) _worker_teardown_cb();
+
+
+}
+
+// process next message in the queue
+// return true if this thread should still be active (while no terminate msg 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;
+
+    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:
+            log_msg incoming_log_msg;
+            incoming_async_msg.fill_log_msg(incoming_log_msg);
+            _formatter->format(incoming_log_msg);
+            for (auto &s : _sinks)
+            {
+                if (s->should_log(incoming_log_msg.level))
+                {
+                    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;
+    }
+}
+
+// flush all sinks if _flush_interval_ms has expired
+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;
+}
+
+
+// spin, yield or sleep. use 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 namespace std::this_thread;
+    using std::chrono::milliseconds;
+    using std::chrono::microseconds;
+
+    auto time_since_op = now - last_op_time;
+
+    // spin upto 50 micros
+    if (time_since_op <= microseconds(50))
+        return;
+
+    // yield upto 150 micros
+    if (time_since_op <= microseconds(100))
+        return std::this_thread::yield();
+
+    // sleep for 20 ms upto 200 ms
+    if (time_since_op <= milliseconds(200))
+        return sleep_for(milliseconds(20));
+
+    // sleep for 200 ms
+    return sleep_for(milliseconds(200));
+}
+
+// wait for the queue to be empty
+inline void spdlog::details::async_log_helper::wait_empty_q()
+{
+    auto last_op = details::os::now();
+    while (_q.approx_size() > 0)
+    {
+        sleep_or_yield(details::os::now(), last_op);
+    }
+}
+
+inline void spdlog::details::async_log_helper::set_error_handler(spdlog::log_err_handler err_handler)
+{
+    _err_handler = err_handler;
+}
+
+
+