diff --git a/src/event.cpp b/src/event.cpp index 16f3f451bb..1805986388 100644 --- a/src/event.cpp +++ b/src/event.cpp @@ -1,5 +1,5 @@ /* - * Copyright Andrey Semashev 2007 - 2021. + * Copyright Andrey Semashev 2007 - 2026. * Distributed under the Boost Software License, Version 1.0. * (See accompanying file LICENSE_1_0.txt or copy at * http://www.boost.org/LICENSE_1_0.txt) @@ -71,11 +71,16 @@ BOOST_LOG_API void atomic_based_event::wait() //! Sets the object to a signalled state BOOST_LOG_API void atomic_based_event::set_signalled() { - if (m_state.load(boost::memory_order_relaxed) != 0u) - { - boost::atomic_thread_fence(boost::memory_order_release); - } - else if (m_state.exchange(1u, boost::memory_order_release) == 0u) + // This must be an unconditional read-modify-write, not a load with an + // "already signalled, skip the notify" fast path. The event is paired with + // external predicates (unbounded_fifo_queue's interruption flag and queue + // contents) that the caller publishes before this call, so a load-only fast + // path could observe a stale non-zero m_state (StoreLoad reordering on x86) + // or skip the notify while a concurrent wait() consumes the prior signal -- + // either way the waiter re-parks with the wakeup lost. The exchange is a + // full barrier, ordered with wait()'s exchange in m_state's modification + // order, so a notify is guaranteed whenever the prior signal was consumed. + if (m_state.exchange(1u, boost::memory_order_acq_rel) == 0u) { m_state.notify_one(); } @@ -185,11 +190,12 @@ BOOST_LOG_API void winapi_based_event::wait() //! Sets the object to a signalled state BOOST_LOG_API void winapi_based_event::set_signalled() { - if (m_state.load(boost::memory_order_relaxed) != 0u) - { - boost::atomic_thread_fence(boost::memory_order_release); - } - else if (m_state.exchange(1u, boost::memory_order_release) == 0u) + // Unconditional read-modify-write, for the same reason as + // atomic_based_event::set_signalled (see the comment there). Both sub-paths + // are affected: the auto-reset event's sticky signal does not rescue the + // kernel-event path, because a load-only fast path would skip SetEvent + // entirely rather than leave a pending kernel signal. + if (m_state.exchange(1u, boost::memory_order_acq_rel) == 0u) { if (!m_event) { diff --git a/test/run/sink_async_frontend_flush.cpp b/test/run/sink_async_frontend_flush.cpp new file mode 100644 index 0000000000..d1c3c3541e --- /dev/null +++ b/test/run/sink_async_frontend_flush.cpp @@ -0,0 +1,99 @@ +/* + * Copyright 2026 - The MathWorks, Inc. + * Distributed under the Boost Software License, Version 1.0. + * (See accompanying file LICENSE_1_0.txt or copy at + * http://www.boost.org/LICENSE_1_0.txt) + */ +/*! + * \file sink_async_frontend_flush.cpp + * \author Conor Burgess + * \date 08.07.2026 + * + * \brief This file contains a test for asynchronous_sink flushing. + * + * One thread repeatedly flushes the logging core while another logs, for a few + * seconds. This stresses the handshake between asynchronous_sink::flush() (via + * unbounded_fifo_queue::interrupt_dequeue) and the sink feeding thread parked + * in the internal event. A lost-wakeup regression in that event made flush() + * hang forever on this configuration. A single run rarely trips it, so the + * test is meant to be looped; if the wakeup is lost the flush thread never + * joins and the test hangs, which is itself the failure signal. + */ + +#define BOOST_TEST_MODULE sink_async_frontend_flush + +#include + +#include + +#if !defined(BOOST_LOG_NO_THREADS) + +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include + +#include "test_barrier.hpp" + +namespace logging = boost::log; +namespace sinks = boost::log::sinks; +namespace src = boost::log::sources; + +// One thread flushes the core while another logs, for a few seconds, both +// started together via a barrier. If the flush/feed handshake loses a wakeup, +// flush() hangs and the flush thread never joins. +BOOST_AUTO_TEST_CASE(async_flush_no_lost_wakeup) +{ + typedef sinks::asynchronous_sink< sinks::text_ostream_backend > sink_t; + + boost::shared_ptr< std::ostream > strm(new std::ostringstream()); + boost::shared_ptr< sink_t > sink = boost::make_shared< sink_t >(); + sink->locked_backend()->add_stream(strm); + logging::core::get()->add_sink(sink); + + const int run_seconds = 5; + std::atomic< bool > stop(false); + test_barrier barrier(2u); + + std::thread flusher([&]() + { + barrier.arrive_and_wait(); + while (!stop.load(std::memory_order_relaxed)) + logging::core::get()->flush(); + }); + + barrier.arrive_and_wait(); + src::severity_logger< int > lg; + const std::chrono::steady_clock::time_point start = std::chrono::steady_clock::now(); + while (std::chrono::duration_cast< std::chrono::seconds >( + std::chrono::steady_clock::now() - start).count() < run_seconds) + { + BOOST_LOG_SEV(lg, 0) << "stress"; + std::this_thread::yield(); + } + + stop.store(true, std::memory_order_relaxed); + flusher.join(); + + logging::core::get()->remove_all_sinks(); + + BOOST_CHECK(true); +} + +#else // !defined(BOOST_LOG_NO_THREADS) + +BOOST_AUTO_TEST_CASE(async_flush_no_lost_wakeup) +{ + BOOST_CHECK(true); +} + +#endif // !defined(BOOST_LOG_NO_THREADS)