This is an automated email from the ASF dual-hosted git repository.
swebb2066 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/logging-log4cxx.git
The following commit(s) were added to refs/heads/master by this push:
new f5f0c240 Prevent potential data race when restarting the AsyncAppender
dispatcher (#752)
f5f0c240 is described below
commit f5f0c24019b7274daad2508cab3de98d00359f1b
Author: Stephen Webb <[email protected]>
AuthorDate: Thu Sep 3 11:16:38 2026 +1000
Prevent potential data race when restarting the AsyncAppender dispatcher
(#752)
* Limit use of 'bufferSize' property to initialization
* Keep dispatcher thread alive even when an attached appender repeatedly
throws an exception
---
src/main/cpp/asyncappender.cpp | 101 ++++++++++++++++++++-------------
src/test/cpp/asyncappendertestcase.cpp | 75 ++++++++++++++++++++++++
2 files changed, 138 insertions(+), 38 deletions(-)
diff --git a/src/main/cpp/asyncappender.cpp b/src/main/cpp/asyncappender.cpp
index ef12808d..d2ec3d36 100644
--- a/src/main/cpp/asyncappender.cpp
+++ b/src/main/cpp/asyncappender.cpp
@@ -139,10 +139,7 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
using BaseType = AppenderSkeleton::AppenderSkeletonPrivate;
AsyncAppenderPriv()
: AppenderSkeletonPrivate()
- , buffer(DEFAULT_BUFFER_SIZE)
, bufferSize(DEFAULT_BUFFER_SIZE)
- , dispatcher()
- , locationInfo(false)
, blocking(true)
#if LOG4CXX_EVENTS_AT_EXIT
, atExitRegistryRaii([this]{if (setClosed()) stopDispatcher();})
@@ -196,10 +193,23 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
*/
std::thread dispatcher;
+ /**
+ * Serializes join()/joinable()/move-assignment on \c dispatcher:
+ * concurrent use of those operations on the same std::thread object
+ * is a data race with undefined behaviour.
+ */
+ std::mutex dispatcherMutex;
+
+ /**
+ * The dispatcher's thread id (written while holding \c dispatcherMutex,
+ * readable by logging threads without touching the thread object).
+ */
+ std::atomic<std::thread::id> dispatcherId{ std::thread::id() };
+
/**
* Used to determine when to restart dispatch thread.
*/
- bool dispatcherActive{ false };
+ std::atomic<bool> dispatcherActive{ false };
/**
* Used to determine whether to restart dispatch thread.
@@ -216,24 +226,33 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
*/
void checkDispatcher(const LogString& appenderName)
{
+ if (this->dispatcherActive) // Fast path: no lock while the
dispatcher is running
+ return;
+
+ // A stopped (or not yet started) dispatcher may be observed by
several
+ // logging threads concurrently; serialize all
join()/joinable()/
+ // move-assignment on the thread object.
+ std::lock_guard<std::mutex> lock(this->dispatcherMutex);
+
// Restart dispatcher if it has stopped by an exception in an
attached appender.
if (!this->dispatcherActive && this->dispatcher.joinable())
+ {
this->dispatcher.join();
+ this->dispatcherId = std::thread::id();
+ }
if (!this->dispatcher.joinable() && this->dispatcherStartCount
<= 1)
{
- std::lock_guard<std::recursive_mutex> lock(this->mutex);
- if (!this->dispatcher.joinable())
- {
- ++this->dispatcherStartCount;
- this->dispatcherActive = true;
- this->dispatcher =
ThreadUtility::instance()->createThread
- ( LOG4CXX_STR("AsyncAppender")
- ,
&AsyncAppender::AsyncAppenderPriv::dispatch
- , this
- , appenderName
- );
- }
+ this->buffer.resize(this->bufferSize);
+ ++this->dispatcherStartCount;
+ this->dispatcherActive = true;
+ this->dispatcher =
ThreadUtility::instance()->createThread
+ ( LOG4CXX_STR("AsyncAppender")
+ , &AsyncAppender::AsyncAppenderPriv::dispatch
+ , this
+ , appenderName
+ );
+ this->dispatcherId = this->dispatcher.get_id();
}
}
@@ -242,9 +261,18 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
bufferNotEmpty.notify_all();
bufferNotFull.notify_all();
- if (dispatcher.joinable())
+ // Move the thread object out under the lock but join outside
it:
+ // the exiting dispatcher may still need bufferMutex, and a
thread
+ // blocked on dispatcherMutex may be holding bufferMutex.
+ std::thread stoppedDispatcher;
+ {
+ std::lock_guard<std::mutex> lock(dispatcherMutex);
+ stoppedDispatcher = std::move(dispatcher);
+ dispatcherId = std::thread::id();
+ }
+ if (stoppedDispatcher.joinable())
{
- dispatcher.join();
+ stoppedDispatcher.join();
}
}
@@ -253,7 +281,7 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
/**
* Should location info be included in dispatched messages.
*/
- bool locationInfo;
+ bool locationInfo{ true };
/**
* Does appender block when buffer is full.
@@ -359,7 +387,7 @@ void AsyncAppender::append(
LOG4CXX_APPEND_FORMAL_PARAMETERS )
priv->checkDispatcher(getName());
- if (priv->dispatcher.get_id() == std::this_thread::get_id()) // From an
appender attached to this?
+ if (priv->dispatcherId.load() == std::this_thread::get_id()) // From an
appender attached to this?
{
std::unique_lock<std::mutex> lock(priv->bufferMutex);
auto loggerName = event->getLoggerName();
@@ -372,13 +400,13 @@ void AsyncAppender::append(
LOG4CXX_APPEND_FORMAL_PARAMETERS )
else while (true)
{
auto pendingCount = priv->eventCount - priv->dispatchedCount;
- if (0 <= pendingCount && pendingCount < priv->bufferSize)
+ if (0 <= pendingCount && pendingCount < priv->buffer.size())
{
// Claim a slot in the ring buffer
auto oldEventCount = priv->eventCount++;
auto index = oldEventCount % priv->buffer.size();
// Wait for a free slot
- while (priv->bufferSize <= oldEventCount -
priv->dispatchedCount)
+ while (priv->buffer.size() <= oldEventCount -
priv->dispatchedCount)
std::this_thread::yield(); // Allow the
dispatch thread to free a slot
// Write to the ring buffer
priv->buffer[index] =
AsyncAppenderPriv::EventData{event, pendingCount};
@@ -412,7 +440,7 @@ void AsyncAppender::append(
LOG4CXX_APPEND_FORMAL_PARAMETERS )
priv->bufferNotFull.wait(lock, [this]()
{
priv->checkDispatcher(getName());
- return priv->eventCount - priv->dispatchedCount
< priv->bufferSize;
+ return priv->eventCount - priv->dispatchedCount
< priv->buffer.size();
});
--priv->blockedCount;
discard = false;
@@ -511,7 +539,6 @@ void AsyncAppender::setLocationInfo(bool flag)
priv->locationInfo = flag;
}
-
void AsyncAppender::setBufferSize(int size)
{
if (size < 0)
@@ -519,20 +546,14 @@ void AsyncAppender::setBufferSize(int size)
throw IllegalArgumentException(LOG4CXX_STR("size argument must
be non-negative"));
}
- std::lock_guard<std::mutex> lock(priv->bufferMutex);
- if (priv->dispatcher.joinable())
- {
- throw RuntimeException(LOG4CXX_STR("AsyncAppender buffer size
cannot be changed now"));
- }
+ std::lock_guard<std::mutex> lock(priv->dispatcherMutex);
priv->bufferSize = (size < 1) ? 1 : size;
- priv->buffer.resize(priv->bufferSize);
- priv->bufferNotFull.notify_all();
}
int AsyncAppender::getBufferSize() const
{
- std::lock_guard<std::mutex> lock(priv->bufferMutex);
- return priv->bufferSize;
+ std::lock_guard<std::mutex> lock(priv->dispatcherMutex);
+ return priv->buffer.empty() ? priv->bufferSize :
static_cast<int>(priv->buffer.size());
}
void AsyncAppender::setBlocking(bool value)
@@ -628,13 +649,13 @@ void AsyncAppender::AsyncAppenderPriv::dispatch(const
LogString& appenderName)
size_t waitCount = 0;
size_t producerBlockedCount = 0;
int failureCount = 0;
- std::vector<size_t> pendingCountHistogram(this->bufferSize, 0);
+ std::vector<size_t> pendingCountHistogram(this->buffer.size(), 0);
bool isActive = true;
while (isActive)
{
LoggingEventList events;
- events.reserve(this->bufferSize);
+ events.reserve(this->buffer.size());
for (int count = 0; count < 2 && this->dispatchedCount ==
this->commitCount; ++count)
std::this_thread::yield(); // Wait a bit
if (this->dispatchedCount == this->commitCount)
@@ -647,7 +668,7 @@ void AsyncAppender::AsyncAppenderPriv::dispatch(const
LogString& appenderName)
}
isActive = !this->isClosed();
- while (events.size() < this->bufferSize &&
this->dispatchedCount != this->commitCount)
+ while (events.size() < this->buffer.size() &&
this->dispatchedCount != this->commitCount)
{
auto index = this->dispatchedCount %
this->buffer.size();
const auto& data = this->buffer[index];
@@ -668,6 +689,12 @@ void AsyncAppender::AsyncAppenderPriv::dispatch(const
LogString& appenderName)
this->discardMap.clear();
}
+ // A fault in an attached appender must not permanently disable
this
+ // dispatch thread: producers using the default Blocking=true
would
+ // hang forever once the ring buffer fills. Reset the failure
budget
+ // for each batch so a transient fault (e.g. a temporarily full
disk)
+ // only limits retries within the current batch.
+ failureCount = 0;
for (auto item : events)
{
try
@@ -690,8 +717,6 @@ void AsyncAppender::AsyncAppenderPriv::dispatch(const
LogString& appenderName)
}
}
++iterationCount;
- if (1 < failureCount)
- break;
}
if (LogLog::isDebugEnabled())
{
diff --git a/src/test/cpp/asyncappendertestcase.cpp
b/src/test/cpp/asyncappendertestcase.cpp
index 265eced6..a2db6133 100644
--- a/src/test/cpp/asyncappendertestcase.cpp
+++ b/src/test/cpp/asyncappendertestcase.cpp
@@ -38,6 +38,7 @@
#include <log4cxx/file.h>
#include <ostream>
#include <thread>
+#include <atomic>
#include <fstream>
#if LOG4CXX_ASYNC_BUFFER_SUPPORTS_FMT
@@ -124,6 +125,36 @@ class BlockableVectorAppender : public VectorAppender
};
LOG4CXX_PTR_DEF(BlockableVectorAppender);
+/**
+ * Vector appender that throws on the first \c failureCount calls to append.
+ */
+class TransientlyFailingVectorAppender : public VectorAppender
+{
+ private:
+ std::atomic<int> failuresRemaining;
+ public:
+ TransientlyFailingVectorAppender(int failureCount)
+ : failuresRemaining(failureCount)
+ {
+ }
+
+ void append( LOG4CXX_APPEND_FORMAL_PARAMETERS ) override
+ {
+ if (0 < failuresRemaining--)
+ {
+ throw RuntimeException(LOG4CXX_STR("Intentional
transient exception"));
+ }
+ VectorAppender::append( LOG4CXX_APPEND_PARAMETERS );
+ }
+
+ /** Goes negative once an append call has succeeded. */
+ int remainingFailures() const
+ {
+ return failuresRemaining;
+ }
+};
+LOG4CXX_PTR_DEF(TransientlyFailingVectorAppender);
+
/**
* An appender that adds logging events
*/
@@ -162,6 +193,7 @@ class AsyncAppenderTestCase : public
AppenderSkeletonTestCase
LOGUNIT_TEST(testEventFlush);
LOGUNIT_TEST(testMultiThread);
LOGUNIT_TEST(testBadAppender);
+ LOGUNIT_TEST(testDispatcherRecoversFromAppenderExceptions);
LOGUNIT_TEST(testBufferOverflowBehavior);
LOGUNIT_TEST(testLoggingAppender);
#if LOG4CXX_HAS_DOMCONFIGURATOR
@@ -426,6 +458,49 @@ class AsyncAppenderTestCase : public
AppenderSkeletonTestCase
LOGUNIT_ASSERT(0 < v.size());
}
+ /**
+ * Checks the dispatch thread survives exceptions thrown by an
+ * attached appender. Regression test: two exceptions at any two
+ * points in the dispatch thread's lifetime used to stop
dispatching
+ * permanently, after which producers using the default
Blocking=true
+ * hung forever once the buffer filled.
+ */
+ void testDispatcherRecoversFromAppenderExceptions()
+ {
+ // Configure Log4cxx
+ AsyncAppenderPtr async;
+ auto r = LogManager::getLoggerRepository();
+ r->ensureIsConfigured([r, &async]()
+ {
+ async = std::make_shared<AsyncAppender>();
+
async->setName(LOG4CXX_STR("async-testDispatcherRecovers"));
+ async->activateOptions();
+ r->getRootLogger()->addAppender(async);
+ r->setConfigured(true);
+ });
+ LOGUNIT_ASSERT(async);
+ // The old code tolerated only 4 exceptions over the
appender lifetime
+ // (2 per dispatch thread incarnation, one restart
allowed)
+ auto failingAppender =
std::make_shared<TransientlyFailingVectorAppender>(4);
+
failingAppender->setName(LOG4CXX_STR("async-transientlyFailingVector"));
+ async->addAppender(failingAppender);
+
+ // Log messages until one has been delivered (bounded
wait)
+ auto root = r->getRootLogger();
+ for (int i = 0; 0 <=
failingAppender->remainingFailures() && i < 100; i++)
+ {
+ LOG4CXX_INFO_ASYNC(root, "message" << i);
+ std::this_thread::sleep_for(
std::chrono::milliseconds( 10 ) );
+ }
+ LOG4CXX_INFO_ASYNC(root, "final message");
+ async->close();
+
+ // Check dispatching recovered once the transient fault
cleared
+ auto& v = failingAppender->getVector();
+ LOGUNIT_ASSERT(!v.empty());
+ LOGUNIT_ASSERT(v.back()->getRenderedMessage() ==
LOG4CXX_STR("final message"));
+ }
+
/**
* Tests behavior when the the async buffer overflows.
*/
