Skip to content

Commit

Permalink
Revert "Exchange promise for condition_variable when flushing (fixes #…
Browse files Browse the repository at this point in the history
…3221) (#3228)"

This reverts commit 16e0d2e.
  • Loading branch information
gabime committed Nov 1, 2024
1 parent 48ce4a9 commit 1437cb2
Show file tree
Hide file tree
Showing 3 changed files with 44 additions and 65 deletions.
38 changes: 18 additions & 20 deletions include/spdlog/async_logger-inl.h
Original file line number Diff line number Diff line change
Expand Up @@ -32,30 +32,28 @@ SPDLOG_INLINE spdlog::async_logger::async_logger(std::string logger_name,

// send the log message to the thread pool
SPDLOG_INLINE void spdlog::async_logger::sink_it_(const details::log_msg &msg){
SPDLOG_TRY {
if (auto pool_ptr = thread_pool_.lock()){
pool_ptr->post_log(shared_from_this(), msg, overflow_policy_);
}
else {
throw_spdlog_ex("async log: thread pool doesn't exist anymore");
}
}
SPDLOG_LOGGER_CATCH(msg.source)
SPDLOG_TRY{if (auto pool_ptr = thread_pool_.lock()){
pool_ptr->post_log(shared_from_this(), msg, overflow_policy_);
}
else {
throw_spdlog_ex("async log: thread pool doesn't exist anymore");
}
}
SPDLOG_LOGGER_CATCH(msg.source)
}

// send flush request to the thread pool
SPDLOG_INLINE void spdlog::async_logger::flush_() {
SPDLOG_TRY {
auto pool_ptr = thread_pool_.lock();
if (!pool_ptr) {
throw_spdlog_ex("async flush: thread pool doesn't exist anymore");
}
SPDLOG_INLINE void spdlog::async_logger::flush_(){SPDLOG_TRY{auto pool_ptr = thread_pool_.lock();
if (!pool_ptr) {
throw_spdlog_ex("async flush: thread pool doesn't exist anymore");
}

// Wait for the flush operation to complete.
// This might throw exception if the flush message get dropped because of overflow.
pool_ptr->post_and_wait_for_flush(shared_from_this(), overflow_policy_);
}
SPDLOG_LOGGER_CATCH(source_loc())
std::future<void> future = pool_ptr->post_flush(shared_from_this(), overflow_policy_);
// Wait for the flush operation to complete.
// This might throw exception if the flush message get dropped because of overflow.
future.get();
}
SPDLOG_LOGGER_CATCH(source_loc())
}

//
Expand Down
29 changes: 7 additions & 22 deletions include/spdlog/details/thread_pool-inl.h
Original file line number Diff line number Diff line change
Expand Up @@ -62,25 +62,13 @@ void SPDLOG_INLINE thread_pool::post_log(async_logger_ptr &&worker_ptr,
post_async_msg_(std::move(async_m), overflow_policy);
}

void SPDLOG_INLINE thread_pool::post_and_wait_for_flush(async_logger_ptr &&worker_ptr,
std::future<void> SPDLOG_INLINE thread_pool::post_flush(async_logger_ptr &&worker_ptr,
async_overflow_policy overflow_policy) {
std::mutex m;
std::unique_lock<std::mutex> l(m);
std::condition_variable cv;
std::atomic<async_msg_flush> cv_flag{async_msg_flush::not_synced};
post_async_msg_(async_msg(std::move(worker_ptr), async_msg_type::flush, [&cv, &cv_flag](async_msg_flush flushed) {
cv_flag.store(flushed, std::memory_order_relaxed);
cv.notify_all();
}), overflow_policy);
while(cv_flag.load(std::memory_order_relaxed) == async_msg_flush::not_synced) {
cv.wait_for(l, std::chrono::milliseconds(100), [&cv_flag]() {
return cv_flag.load(std::memory_order_relaxed) != async_msg_flush::not_synced;
});
}

if(cv_flag.load(std::memory_order_relaxed) == async_msg_flush::synced_not_flushed) {
throw spdlog_ex("Request for flushing got dropped.");
}
std::promise<void> promise;
std::future<void> future = promise.get_future();
post_async_msg_(async_msg(std::move(worker_ptr), async_msg_type::flush, std::move(promise)),
overflow_policy);
return future;
}

size_t SPDLOG_INLINE thread_pool::overrun_counter() { return q_.overrun_counter(); }
Expand Down Expand Up @@ -124,10 +112,7 @@ bool SPDLOG_INLINE thread_pool::process_next_msg_() {
}
case async_msg_type::flush: {
incoming_async_msg.worker_ptr->backend_flush_();
if(incoming_async_msg.flush_callback) {
incoming_async_msg.flush_callback(async_msg_flush::synced_flushed);
incoming_async_msg.flush_callback = nullptr;
}
incoming_async_msg.flush_promise.set_value();
return true;
}

Expand Down
42 changes: 19 additions & 23 deletions include/spdlog/details/thread_pool.h
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

#include <chrono>
#include <functional>
#include <future>
#include <memory>
#include <thread>
#include <vector>
Expand All @@ -22,60 +23,55 @@ using async_logger_ptr = std::shared_ptr<spdlog::async_logger>;

enum class async_msg_type { log, flush, terminate };

enum class async_msg_flush { not_synced, synced_flushed, synced_not_flushed };

// Async msg to move to/from the queue
// Movable only. should never be copied
struct async_msg : log_msg_buffer {
async_msg_type msg_type{async_msg_type::log};
async_logger_ptr worker_ptr;
std::function<void(async_msg_flush)> flush_callback;
std::promise<void> flush_promise;

async_msg() = default;
~async_msg() {
if (flush_callback) {
flush_callback(async_msg_flush::synced_not_flushed);
flush_callback = nullptr;
}
}
~async_msg() = default;

// should only be moved in or out of the queue..
async_msg(const async_msg &) = delete;

async_msg(async_msg &&other) SPDLOG_NOEXCEPT
// support for vs2013 move
#if defined(_MSC_VER) && _MSC_VER <= 1800
async_msg(async_msg &&other)
: log_msg_buffer(std::move(other)),
msg_type(other.msg_type),
worker_ptr(std::move(other.worker_ptr)),
flush_callback(std::move(other.flush_callback)) {
other.flush_callback = nullptr;
}
async_msg &operator=(async_msg &&other) SPDLOG_NOEXCEPT {
*static_cast<log_msg_buffer *>(this) = static_cast<log_msg_buffer&&>(other);
worker_ptr(std::move(other.worker_ptr)) {}

async_msg &operator=(async_msg &&other) {
*static_cast<log_msg_buffer *>(this) = std::move(other);
msg_type = other.msg_type;
worker_ptr = std::move(other.worker_ptr);
flush_callback = std::move(other.flush_callback);
other.flush_callback = nullptr;
return *this;
}
#else // (_MSC_VER) && _MSC_VER <= 1800
async_msg(async_msg &&) = default;
async_msg &operator=(async_msg &&) = default;
#endif

// construct from log_msg with given type
async_msg(async_logger_ptr &&worker, async_msg_type the_type, const details::log_msg &m)
: log_msg_buffer{m},
msg_type{the_type},
worker_ptr{std::move(worker)},
flush_callback{} {}
flush_promise{} {}

async_msg(async_logger_ptr &&worker, async_msg_type the_type)
: log_msg_buffer{},
msg_type{the_type},
worker_ptr{std::move(worker)},
flush_callback{} {}
flush_promise{} {}

async_msg(async_logger_ptr &&worker, async_msg_type the_type, std::function<void(async_msg_flush)> &&callback)
async_msg(async_logger_ptr &&worker, async_msg_type the_type, std::promise<void> &&promise)
: log_msg_buffer{},
msg_type{the_type},
worker_ptr{std::move(worker)},
flush_callback{std::move(callback)} {}
flush_promise{std::move(promise)} {}

explicit async_msg(async_msg_type the_type)
: async_msg{nullptr, the_type} {}
Expand All @@ -102,7 +98,7 @@ class SPDLOG_API thread_pool {
void post_log(async_logger_ptr &&worker_ptr,
const details::log_msg &msg,
async_overflow_policy overflow_policy);
void post_and_wait_for_flush(async_logger_ptr &&worker_ptr,
std::future<void> post_flush(async_logger_ptr &&worker_ptr,
async_overflow_policy overflow_policy);
size_t overrun_counter();
void reset_overrun_counter();
Expand Down

0 comments on commit 1437cb2

Please sign in to comment.