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 f7bbf9b6 Improve async appender performance (#360)
f7bbf9b6 is described below
commit f7bbf9b6d3361f247f8ece759c432f0401c382d8
Author: Stephen Webb <[email protected]>
AuthorDate: Mon Mar 11 13:02:11 2024 +1100
Improve async appender performance (#360)
* A read-only dispatch thread improves performance by 30%
* Use a ring buffer to pass events to the dispatch thread
* compare_exchange_weak is said to yield better performance on some
platforms
---
src/main/cpp/asyncappender.cpp | 76 ++++++++++++++++++++++++----------
src/test/cpp/asyncappendertestcase.cpp | 61 ++++++++++++++++++++++++++-
2 files changed, 115 insertions(+), 22 deletions(-)
diff --git a/src/main/cpp/asyncappender.cpp b/src/main/cpp/asyncappender.cpp
index 5c540cff..332ba533 100644
--- a/src/main/cpp/asyncappender.cpp
+++ b/src/main/cpp/asyncappender.cpp
@@ -26,6 +26,7 @@
#include <log4cxx/helpers/threadutility.h>
#include <log4cxx/private/appenderskeleton_priv.h>
#include <thread>
+#include <atomic>
#include <condition_variable>
#if LOG4CXX_EVENTS_AT_EXIT
@@ -96,10 +97,13 @@ typedef std::map<LogString, DiscardSummary> DiscardMap;
}
#endif
+static const int CACHE_LINE_SIZE = 128;
+
struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkeletonPrivate
{
AsyncAppenderPriv() :
AppenderSkeletonPrivate(),
+ buffer(DEFAULT_BUFFER_SIZE),
bufferSize(DEFAULT_BUFFER_SIZE),
appenders(pool),
dispatcher(),
@@ -108,6 +112,9 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
#if LOG4CXX_EVENTS_AT_EXIT
, atExitRegistryRaii([this]{atExitActivated();})
#endif
+ , eventCount(0)
+ , dispatchedCount(0)
+ , commitCount(0)
{
}
@@ -140,7 +147,7 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
DiscardMap discardMap;
/**
- * Buffer size.
+ * The maximum number of undispatched events.
*/
int bufferSize;
@@ -167,6 +174,21 @@ struct AsyncAppender::AsyncAppenderPriv : public
AppenderSkeleton::AppenderSkele
#if LOG4CXX_EVENTS_AT_EXIT
helpers::AtExitRegistry::Raii atExitRegistryRaii;
#endif
+
+ /**
+ * Used to calculate the buffer position at which to store the next
event.
+ */
+ alignas(CACHE_LINE_SIZE) std::atomic<size_t> eventCount;
+
+ /**
+ * Used to calculate the buffer position from which to extract the next
event.
+ */
+ alignas(CACHE_LINE_SIZE) std::atomic<size_t> dispatchedCount;
+
+ /**
+ * Used to communicate to the dispatch thread when an event is
committed in buffer.
+ */
+ alignas(CACHE_LINE_SIZE) std::atomic<size_t> commitCount;
};
@@ -233,31 +255,38 @@ void AsyncAppender::append(const spi::LoggingEventPtr&
event, Pool& p)
// Get a copy of this thread's MDC.
event->getMDCCopy();
- std::unique_lock<std::mutex> lock(priv->bufferMutex);
if (!priv->dispatcher.joinable())
{
- priv->dispatcher = ThreadUtility::instance()->createThread(
LOG4CXX_STR("AsyncAppender"), &AsyncAppender::dispatch, this );
- priv->buffer.reserve(priv->bufferSize);
+ std::unique_lock<std::mutex> lock(priv->bufferMutex);
+ if (!priv->dispatcher.joinable())
+ priv->dispatcher =
ThreadUtility::instance()->createThread( LOG4CXX_STR("AsyncAppender"),
&AsyncAppender::dispatch, this );
}
while (true)
{
- size_t previousSize = priv->buffer.size();
-
- if (previousSize < (size_t)priv->bufferSize)
+ auto pendingCount = priv->eventCount - priv->dispatchedCount;
+ if (0 <= pendingCount && pendingCount < priv->bufferSize)
{
- priv->buffer.push_back(event);
-
- if (previousSize == 0)
+ // 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)
+ ;
+ // Write to the ring buffer
+ priv->buffer[index] = event;
+ // Notify the dispatch thread that an event has been
added
+ auto savedEventCount = oldEventCount;
+ while
(!priv->commitCount.compare_exchange_weak(oldEventCount, oldEventCount + 1,
std::memory_order_release))
{
- priv->bufferNotEmpty.notify_all();
+ oldEventCount = savedEventCount;
}
-
+ priv->bufferNotEmpty.notify_all();
break;
}
-
//
- // Following code is only reachable if buffer is full
+ // Following code is only reachable if buffer is full or
eventCount has overflowed
//
+ std::unique_lock<std::mutex> lock(priv->bufferMutex);
//
// if blocking and thread is not already interrupted
// and not the dispatcher then
@@ -270,7 +299,7 @@ void AsyncAppender::append(const spi::LoggingEventPtr&
event, Pool& p)
{
priv->bufferNotFull.wait(lock, [this]()
{
- return priv->buffer.empty();
+ return priv->eventCount - priv->dispatchedCount
< priv->bufferSize;
});
discard = false;
}
@@ -374,6 +403,7 @@ void AsyncAppender::setBufferSize(int size)
std::lock_guard<std::mutex> lock(priv->bufferMutex);
priv->bufferSize = (size < 1) ? 1 : size;
+ priv->buffer.resize(priv->bufferSize);
priv->bufferNotFull.notify_all();
}
@@ -455,26 +485,30 @@ void AsyncAppender::dispatch()
while (isActive)
{
+ Pool p;
+ LoggingEventList events;
+ events.reserve(priv->bufferSize);
//
// process events after lock on buffer is released.
//
- Pool p;
- LoggingEventList events;
{
std::unique_lock<std::mutex> lock(priv->bufferMutex);
priv->bufferNotEmpty.wait(lock, [this]() -> bool
- { return 0 < priv->buffer.size() ||
priv->closed; }
+ { return priv->dispatchedCount !=
priv->commitCount || priv->closed; }
);
isActive = !priv->closed;
- events = std::move(priv->buffer);
+ while (events.size() < priv->bufferSize &&
priv->dispatchedCount != priv->commitCount)
+ {
+ auto index = priv->dispatchedCount %
priv->buffer.size();
+ events.push_back(priv->buffer[index]);
+ ++priv->dispatchedCount;
+ }
for (auto discardItem : priv->discardMap)
{
events.push_back(discardItem.second.createEvent(p));
}
- priv->buffer.clear();
- priv->buffer.reserve(priv->bufferSize);
priv->discardMap.clear();
priv->bufferNotFull.notify_all();
}
diff --git a/src/test/cpp/asyncappendertestcase.cpp
b/src/test/cpp/asyncappendertestcase.cpp
index 57610200..8a6fccc5 100644
--- a/src/test/cpp/asyncappendertestcase.cpp
+++ b/src/test/cpp/asyncappendertestcase.cpp
@@ -124,6 +124,7 @@ class AsyncAppenderTestCase : public
AppenderSkeletonTestCase
LOGUNIT_TEST(closeTest);
LOGUNIT_TEST(test2);
LOGUNIT_TEST(testEventFlush);
+ LOGUNIT_TEST(testMultiThread);
LOGUNIT_TEST(testBadAppender);
LOGUNIT_TEST(testBufferOverflowBehavior);
#if LOG4CXX_HAS_DOMCONFIGURATOR
@@ -215,13 +216,71 @@ class AsyncAppenderTestCase : public
AppenderSkeletonTestCase
}
asyncAppender->close();
- root->debug(LOG4CXX_TEST_STR("m2"));
+ root->debug(LOG4CXX_STR("m2"));
const std::vector<spi::LoggingEventPtr>& v =
vectorAppender->getVector();
LOGUNIT_ASSERT_EQUAL(LEN, v.size());
+ Pool p;
+ for (size_t i = 0; i < LEN; i++)
+ {
+ LogString m(LOG4CXX_STR("message"));
+ StringHelper::toString(i, p, m);
+ LOGUNIT_ASSERT(v[i]->getMessage() == m);
+ }
LOGUNIT_ASSERT_EQUAL(true, vectorAppender->isClosed());
}
+
+ // this test checks all messages are delivered from multiple
threads
+ void testMultiThread()
+ {
+ size_t LEN = 2000; // Larger than default buffer size
(128)
+ int threadCount = 6;
+ auto root = Logger::getRootLogger();
+ auto vectorAppender =
std::make_shared<VectorAppender>();
+ auto asyncAppender = std::make_shared<AsyncAppender>();
+
asyncAppender->setName(LOG4CXX_STR("async-testMultiThread"));
+ asyncAppender->addAppender(vectorAppender);
+ root->addAppender(asyncAppender);
+
+ std::vector<std::thread> threads;
+ for ( int x = 0; x < threadCount; x++ )
+ {
+ std::thread thr([root, LEN]()
+ {
+ for (size_t i = 0; i < LEN; i++)
+ {
+ LOG4CXX_DEBUG(root, "message"
<< i);
+ }
+ });
+ threads.push_back( std::move(thr) );
+ }
+
+ for ( auto& thr : threads )
+ {
+ if ( thr.joinable() )
+ {
+ thr.join();
+ }
+ }
+ asyncAppender->close();
+
+ const std::vector<spi::LoggingEventPtr>& v =
vectorAppender->getVector();
+ LOGUNIT_ASSERT_EQUAL(LEN*threadCount, v.size());
+ std::vector<int> count(LEN, 0);
+ for (auto m : v)
+ {
+ auto i =
StringHelper::toInt(m->getMessage().substr(7));
+ LOGUNIT_ASSERT(0 <= i);
+ LOGUNIT_ASSERT(i < LEN);
+ ++count[i];
+ }
+ for (size_t i = 0; i < LEN; i++)
+ {
+ LOGUNIT_ASSERT_EQUAL(count[i], threadCount);
+ }
+ }
+
/**
* Checks that async will switch a bad appender to another
appender.
*/
