spdlog/details/thread_pool.h

0001 // Copyright(c) 2015-present, Gabi Melman & spdlog contributors.
0002 // Distributed under the MIT License (http://opensource.org/licenses/MIT)
0003
0004 #pragma once
0005
0006 #include <spdlog/details/log_msg_buffer.h>
0007 #include <spdlog/details/mpmc_blocking_q.h>
0008 #include <spdlog/details/os.h>
0009
0010 #include <chrono>
0011 #include <functional>
0012 #include <future>
0013 #include <memory>
0014 #include <thread>
0015 #include <vector>
0016
0017 namespace spdlog {
0018 class async_logger;
0019
0020 namespace details {
0021
0022 using async_logger_ptr = std::shared_ptr<spdlog::async_logger>;
0023
0024 enum class async_msg_type { log, flush, terminate };
0025
0026 // Async msg to move to/from the queue
0027 // Movable only. should never be copied
0028 struct async_msg : log_msg_buffer {
0029     async_msg_type msg_type{async_msg_type::log};
0030     async_logger_ptr worker_ptr;
0031     std::promise<void> flush_promise;
0032
0033     async_msg() = default;
0034     ~async_msg() = default;
0035
0036     // should only be moved in or out of the queue..
0037     async_msg(const async_msg &) = delete;
0038
0039 // support for vs2013 move
0040 #if defined(_MSC_VER) && _MSC_VER <= 1800
0041     async_msg(async_msg &&other)
0042         : log_msg_buffer(std::move(other)),
0043           msg_type(other.msg_type),
0044           worker_ptr(std::move(other.worker_ptr)) {}
0045
0046     async_msg &operator=(async_msg &&other) {
0047         *static_cast<log_msg_buffer *>(this) = std::move(other);
0048         msg_type = other.msg_type;
0049         worker_ptr = std::move(other.worker_ptr);
0050         return *this;
0051     }
0052 #else  // (_MSC_VER) && _MSC_VER <= 1800
0053     async_msg(async_msg &&) = default;
0054     async_msg &operator=(async_msg &&) = default;
0055 #endif
0056
0057     // construct from log_msg with given type
0058     async_msg(async_logger_ptr &&worker, async_msg_type the_type, const details::log_msg &m)
0059         : log_msg_buffer{m},
0060           msg_type{the_type},
0061           worker_ptr{std::move(worker)},
0062           flush_promise{} {}
0063
0064     async_msg(async_logger_ptr &&worker, async_msg_type the_type)
0065         : log_msg_buffer{},
0066           msg_type{the_type},
0067           worker_ptr{std::move(worker)},
0068           flush_promise{} {}
0069
0070     async_msg(async_logger_ptr &&worker, async_msg_type the_type, std::promise<void> &&promise)
0071         : log_msg_buffer{},
0072           msg_type{the_type},
0073           worker_ptr{std::move(worker)},
0074           flush_promise{std::move(promise)} {}
0075
0076     explicit async_msg(async_msg_type the_type)
0077         : async_msg{nullptr, the_type} {}
0078 };
0079
0080 class SPDLOG_API thread_pool {
0081 public:
0082     using item_type = async_msg;
0083     using q_type = details::mpmc_blocking_queue<item_type>;
0084
0085     thread_pool(size_t q_max_items,
0086                 size_t threads_n,
0087                 std::function<void()> on_thread_start,
0088                 std::function<void()> on_thread_stop);
0089     thread_pool(size_t q_max_items, size_t threads_n, std::function<void()> on_thread_start);
0090     thread_pool(size_t q_max_items, size_t threads_n);
0091
0092     // message all threads to terminate gracefully and join them
0093     ~thread_pool();
0094
0095     thread_pool(const thread_pool &) = delete;
0096     thread_pool &operator=(thread_pool &&) = delete;
0097
0098     void post_log(async_logger_ptr &&worker_ptr,
0099                   const details::log_msg &msg,
0100                   async_overflow_policy overflow_policy);
0101     std::future<void> post_flush(async_logger_ptr &&worker_ptr,
0102                                  async_overflow_policy overflow_policy);
0103     size_t overrun_counter();
0104     void reset_overrun_counter();
0105     size_t discard_counter();
0106     void reset_discard_counter();
0107     size_t queue_size();
0108
0109 private:
0110     q_type q_;
0111
0112     std::vector<std::thread> threads_;
0113
0114     void post_async_msg_(async_msg &&new_msg, async_overflow_policy overflow_policy);
0115     void worker_loop_();
0116
0117     // process next message in the queue
0118     // return true if this thread should still be active (while no terminate msg
0119     // was received)
0120     bool process_next_msg_();
0121 };
0122
0123 }  // namespace details
0124 }  // namespace spdlog
0125
0126 #ifdef SPDLOG_HEADER_ONLY
0127     #include "thread_pool-inl.h"
0128 #endif