src/main/cpp/asyncappender.cpp

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <log4cxx/asyncappender.h>

#include <log4cxx/helpers/loglog.h>
#include <log4cxx/spi/loggingevent.h>
#include <log4cxx/helpers/stringhelper.h>
#include <log4cxx/helpers/optionconverter.h>
#include <log4cxx/helpers/threadutility.h>
#include <log4cxx/private/appenderskeleton_priv.h>
#include <thread>
#include <atomic>
#include <condition_variable>

#if LOG4CXX_EVENTS_AT_EXIT
#include <log4cxx/private/atexitregistry.h>
#endif

using namespace LOG4CXX_NS;
using namespace LOG4CXX_NS::helpers;
using namespace LOG4CXX_NS::spi;

namespace
{

/**
 * The default buffer size is set to 128 events.
*/
enum { DEFAULT_BUFFER_SIZE = 128 };

class DiscardSummary
{
	private:
		/**
		 * First event of the highest severity.
		*/
		LoggingEventPtr maxEvent;

		/**
		* Total count of messages discarded.
		*/
		int count;

		/**
		* Why created
		*/
		LogString reason;

	public:
		/**
		 * Create new instance.
		 *
		 * @param event must not be null.
		*/
		DiscardSummary(const LoggingEventPtr& event, const LogString& reason);

		/** Move values from \c src into a new instance.
		*/
		DiscardSummary(DiscardSummary&& src);
		/** Copy constructor.  */
		DiscardSummary(const DiscardSummary&) = delete;
		/** Assignment operator. */
		DiscardSummary& operator=(const DiscardSummary&) = delete;

		/**
		 * Add discarded event to summary.
		 *
		 * @param event event, may not be null.
		*/
		void add(const LoggingEventPtr& event);

		/**
		 * Create an event with a discard count and the message from \c maxEvent.
		 *
		 * @return the new event.
		 */
		LoggingEventPtr createEvent();


		/**
		* The number of messages discarded.
		*/
		int getCount() const { return count; }
};

typedef std::map<LogString, DiscardSummary> DiscardMap;

}

#ifdef __cpp_lib_hardware_interference_size
	using std::hardware_constructive_interference_size;
	using std::hardware_destructive_interference_size;
#else
	// 64 bytes on x86-64 │ L1_CACHE_BYTES │ L1_CACHE_SHIFT │ __cacheline_aligned │ ...
	constexpr std::size_t hardware_constructive_interference_size = 64;
	constexpr std::size_t hardware_destructive_interference_size = 64;
#endif

struct AsyncAppender::AsyncAppenderPriv : public AppenderSkeleton::AppenderSkeletonPrivate
{
	using BaseType = AppenderSkeleton::AppenderSkeletonPrivate;
	AsyncAppenderPriv()
		: AppenderSkeletonPrivate()
		, bufferSize(DEFAULT_BUFFER_SIZE)
		, blocking(true)
#if LOG4CXX_EVENTS_AT_EXIT
		, atExitRegistryRaii([this]{if (setClosed()) stopDispatcher();})
#endif
		, eventCount(0)
		, dispatchedCount(0)
		, commitCount(0)
		{ }

	~AsyncAppenderPriv()
	{
		if (setClosed())
			close();
	}

	/**
	 * Event buffer.
	*/
	struct EventData
	{
		LoggingEventPtr event;
		size_t pendingCount;
	};
	std::vector<EventData> buffer;

	/**
	 *  Mutex used to guard access to buffer and discardMap.
	 */
	std::mutex bufferMutex;

	std::condition_variable bufferNotFull;
	std::condition_variable bufferNotEmpty;

	/**
	  * Map of DiscardSummary objects keyed by logger name.
	*/
	DiscardMap discardMap;

	/**
	 * The maximum number of undispatched events.
	*/
	int bufferSize;

	/**
	 * Nested appenders.
	*/
	helpers::AppenderAttachableImpl appenders;

	/**
	 *  Dispatcher.
	 */
	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.
	*/
	std::atomic<bool> dispatcherActive{ false };

	/**
	 * Used to determine whether to restart dispatch thread.
	*/
	int dispatcherStartCount{ 0 };

	/**
	 *  The function the dispatcher executes.
	 */
	void dispatch(const LogString& appenderName);

	/**
	 * Start dispatcher if not already running.
	 */
	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)
		{
			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();
		}
	}

	void stopDispatcher()
	{
		bufferNotEmpty.notify_all();
		bufferNotFull.notify_all();

		// 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())
		{
			stoppedDispatcher.join();
		}
	}

	void close();

	/**
	 * Does appender block when buffer is full.
	*/
	bool blocking;

#if LOG4CXX_EVENTS_AT_EXIT
	helpers::AtExitRegistry::Raii atExitRegistryRaii;
#endif

	/**
	 * Used to calculate the buffer position at which to store the next event.
	*/
	alignas(hardware_constructive_interference_size) std::atomic<size_t> eventCount;

	/**
	 * Used to calculate the buffer position from which to extract the next event.
	*/
	alignas(hardware_constructive_interference_size) std::atomic<size_t> dispatchedCount;

	/**
	 * Used to communicate to the dispatch thread when an event is committed in buffer.
	*/
	alignas(hardware_constructive_interference_size) std::atomic<size_t> commitCount;

	bool isClosed()
	{
		std::lock_guard<std::mutex> lock(this->bufferMutex);
		return this->closed;
	}

	bool setClosed()
	{
		std::lock_guard<std::mutex> lock(this->bufferMutex);
		return BaseType::setClosed();
	}

	/**
	 * Used to ensure the dispatch thread does not wait when a logging thread is waiting.
	*/
	alignas(hardware_constructive_interference_size) int blockedCount{0};
};


IMPLEMENT_LOG4CXX_OBJECT(AsyncAppender)

#define priv static_cast<AsyncAppenderPriv*>(m_priv.get())

AsyncAppender::AsyncAppender()
	: AppenderSkeleton(std::make_unique<AsyncAppenderPriv>())
{
}

AsyncAppender::~AsyncAppender()
{
}

void AsyncAppender::addAppender(const AppenderPtr& newAppender)
{
	priv->appenders.addAppender(newAppender);
}


void AsyncAppender::setOption(const LogString& option,
	const LogString& value)
{
	if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("BUFFERSIZE"), LOG4CXX_STR("buffersize")))
	{
		setBufferSize(OptionConverter::toInt(value, DEFAULT_BUFFER_SIZE));
	}

	else if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("BLOCKING"), LOG4CXX_STR("blocking")))
	{
		setBlocking(OptionConverter::toBoolean(value, true));
	}
	else
	{
		AppenderSkeleton::setOption(option, value);
	}
}


void AsyncAppender::doAppend( const spi::LoggingEventPtr& event )
{
	doAppendImpl( event );
}

void AsyncAppender::append( const spi::LoggingEventPtr& event )
{
	if (priv->bufferSize <= 0)
	{
		priv->appenders.appendLoopOnAppenders(event);
		return;
	}

	// Get a copy of this thread's diagnostic context
	event->LoadDC();

	priv->checkDispatcher(getName());

	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();
		auto iter = priv->discardMap.find(loggerName);
		if (priv->discardMap.end() == iter)
			priv->discardMap.emplace(loggerName, DiscardSummary{ event, LOG4CXX_STR("from an attached appender") });
		else
			iter->second.add(event);
	}
	else while (true)
	{
		auto pendingCount = priv->eventCount - priv->dispatchedCount;
		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->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};
			// Notify the dispatch thread that an event has been added
			auto failureCount = 0;
			auto savedEventCount = oldEventCount;
			while (!priv->commitCount.compare_exchange_weak(oldEventCount, oldEventCount + 1, std::memory_order_release))
			{
				oldEventCount = savedEventCount;
				if (2 < ++failureCount) // Did the scheduler suspend a thread between claiming a slot and advancing commitCount?
					std::this_thread::yield(); // Wait a bit
			}
			priv->bufferNotEmpty.notify_all();
			break;
		}
		//
		//   Following code is only reachable if buffer is full or eventCount has overflowed
		//
		std::unique_lock<std::mutex> lock(priv->bufferMutex);
		priv->bufferNotEmpty.notify_all();
		//
		//   if blocking and thread is not already interrupted
		//      and not the dispatcher then
		//      wait for a buffer notification
		bool discard = true;

		if (priv->blocking
			&& !priv->closed)
		{
			++priv->blockedCount;
			priv->bufferNotFull.wait(lock, [this]()
			{
				priv->checkDispatcher(getName());
				return priv->eventCount - priv->dispatchedCount < priv->buffer.size();
			});
			--priv->blockedCount;
			discard = false;
		}

		//
		//   if blocking is false or thread has been interrupted
		//   add event to discard map.
		//
		if (discard)
		{
			LogString loggerName = event->getLoggerName();
			DiscardMap::iterator iter = priv->discardMap.find(loggerName);

			if (iter == priv->discardMap.end())
			{
				priv->discardMap.emplace(loggerName, DiscardSummary{ event, LOG4CXX_STR("due to a full event buffer") });
			}
			else
			{
				iter->second.add(event);
			}

			break;
		}
	}
}

void AsyncAppender::close()
{
	if (priv->setClosed())
		priv->close();
}

void AsyncAppender::AsyncAppenderPriv::close()
{
	this->stopDispatcher();
	for (auto item : this->appenders.getAllAppenders())
	{
		item->close();
	}
}

AppenderList AsyncAppender::getAllAppenders() const
{
	return priv->appenders.getAllAppenders();
}

AppenderPtr AsyncAppender::getAppender(const LogString& n) const
{
	return priv->appenders.getAppender(n);
}

bool AsyncAppender::isAttached(const AppenderPtr& appender) const
{
	return priv->appenders.isAttached(appender);
}

bool AsyncAppender::requiresLayout() const
{
	return false;
}

void AsyncAppender::removeAllAppenders()
{
	priv->appenders.removeAllAppenders();
}

bool AsyncAppender::removeAppender(const AppenderPtr& appender)
{
	return priv->appenders.removeAppender(appender);
}

bool AsyncAppender::removeAppender(const LogString& n)
{
	return priv->appenders.removeAppender(n);
}

bool AsyncAppender::replaceAppender(const AppenderPtr& oldAppender, const AppenderPtr& newAppender)
{
	return priv->appenders.replaceAppender(oldAppender, newAppender);
}

void AsyncAppender::replaceAppenders( const AppenderList& newList)
{
	priv->appenders.replaceAppenders(newList);
}


void AsyncAppender::setBufferSize(int size)
{
	if (size < 0)
	{
		throw IllegalArgumentException(LOG4CXX_STR("size argument must be non-negative"));
	}

	std::lock_guard<std::mutex> lock(priv->dispatcherMutex);
	priv->bufferSize = (size < 1) ? 1 : size;
}

int AsyncAppender::getBufferSize() const
{
	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)
{
	std::lock_guard<std::mutex> lock(priv->bufferMutex);
	priv->blocking = value;
	priv->bufferNotFull.notify_all();
}

bool AsyncAppender::getBlocking() const
{
	std::lock_guard<std::mutex> lock(priv->bufferMutex);
	return priv->blocking;
}

DiscardSummary::DiscardSummary(const LoggingEventPtr& event, const LogString& reasonArg)
	: maxEvent(event)
	, count(1)
	, reason(reasonArg)
{
}

DiscardSummary::DiscardSummary(DiscardSummary&& other)
	: maxEvent(std::move(other.maxEvent))
	, count(other.count)
	, reason(std::move(other.reason))
{
}


void DiscardSummary::add(const LoggingEventPtr& event)
{
	if (this->maxEvent->getLevel()->toInt() < event->getLevel()->toInt())
		this->maxEvent = event;
	++this->count;
}

LoggingEventPtr DiscardSummary::createEvent()
{
	LogString msg(LOG4CXX_STR("Discarded "));
	StringHelper::toString(this->count, msg);
	msg.append(LOG4CXX_STR(" messages ") + this->reason + LOG4CXX_STR(" including: "));
	msg.append(this->maxEvent->getRenderedMessage());
	return std::make_shared<LoggingEvent>
		( this->maxEvent->getLoggerName()
		, this->maxEvent->getLevel()
		, msg
		, LocationInfo::getLocationUnavailable()
		);
}

void AsyncAppender::AsyncAppenderPriv::dispatch(const LogString& appenderName)
{
	size_t discardCount = 0;
	size_t iterationCount = 0;
	size_t waitCount = 0;
	size_t producerBlockedCount = 0;
	int failureCount = 0;
	std::vector<size_t> pendingCountHistogram(this->buffer.size(), 0);
	bool isActive = true;

	while (isActive)
	{
		LoggingEventList events;
		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)
		{
			++waitCount;
			std::unique_lock<std::mutex> lock(this->bufferMutex);
			this->bufferNotEmpty.wait(lock, [this]() -> bool
				{ return 0 < this->blockedCount || this->dispatchedCount != this->commitCount || this->closed; }
			);
		}
		isActive = !this->isClosed();

		while (events.size() < this->buffer.size() && this->dispatchedCount != this->commitCount)
		{
			auto index = this->dispatchedCount % this->buffer.size();
			const auto& data = this->buffer[index];
			events.push_back(data.event);
			if (data.pendingCount < pendingCountHistogram.size())
				++pendingCountHistogram[data.pendingCount];
			++this->dispatchedCount;
		}
		this->bufferNotFull.notify_all();
		{
			std::lock_guard<std::mutex> lock(this->bufferMutex);
			producerBlockedCount += this->blockedCount;
			for (auto& discardItem : this->discardMap)
			{
				events.push_back(discardItem.second.createEvent());
				discardCount += discardItem.second.getCount();
			}
			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
			{
				this->appenders.appendLoopOnAppenders(item);
			}
			catch (std::exception& ex)
			{
				if (1 < ++failureCount)
					break;
				if (!this->isClosed())
					this->errorHandler->error(LOG4CXX_STR("[") + appenderName + LOG4CXX_STR("] AsyncAppender"), ex, spi::ErrorCode::WRITE_FAILURE, item);
			}
			catch (...)
			{
				if (1 < ++failureCount)
					break;
				if (!this->isClosed())
					this->errorHandler->error(LOG4CXX_STR("[") + appenderName + LOG4CXX_STR("] AsyncAppender unknown exception thrown"));
			}
		}
		++iterationCount;
	}
	if (LogLog::isDebugEnabled())
	{
		LogString msg(LOG4CXX_STR("[") + appenderName + LOG4CXX_STR("] AsyncAppender"));
#ifdef _DEBUG
		msg += LOG4CXX_STR(" iterationCount ");
		StringHelper::toString(iterationCount, msg);
		msg += LOG4CXX_STR(" waitCount ");
		StringHelper::toString(waitCount, msg);
		msg += LOG4CXX_STR(" producerBlockedCount ");
		StringHelper::toString(producerBlockedCount, msg);
		msg += LOG4CXX_STR(" commitCount ");
		StringHelper::toString(this->commitCount, msg);
#endif
		msg += LOG4CXX_STR(" dispatchedCount ");
		StringHelper::toString(this->dispatchedCount, msg);
		msg += LOG4CXX_STR(" discardCount ");
		StringHelper::toString(discardCount, msg);
		msg += LOG4CXX_STR(" pendingCountHistogram");
		for (auto item : pendingCountHistogram)
		{
			msg += logchar(' ');
			StringHelper::toString(item, msg);
		}
		LogLog::debug(msg);
	}
	this->dispatcherActive = false;
	if (0 < this->blockedCount) // Restart this dispatcher?
		this->bufferNotFull.notify_all();
}