/*

 * 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;

#if 15 < LOG4CXX_ABI_VERSION

namespace

{

#endif

/**

 * 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);

#if 15 < LOG4CXX_ABI_VERSION

    /** Copy constructor.  */

    DiscardSummary(const DiscardSummary&) = delete;

    /** Assignment operator. */

    DiscardSummary& operator=(const DiscardSummary&) = delete;

#else

    /**

     * Create new instance.

     *

     * @param event event, may not be null.

    */

    DiscardSummary(const LoggingEventPtr& event);

    /** Copy constructor.  */

    DiscardSummary(const DiscardSummary& src);

    /** Assignment operator. */

    DiscardSummary& operator=(const DiscardSummary& src);

#endif

    /**

     * 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(Pool& p);

#if LOG4CXX_ABI_VERSION <= 15

    static

    ::LOG4CXX_NS::spi::LoggingEventPtr createEvent(::LOG4CXX_NS::helpers::Pool& p,

      size_t discardedCount);

#endif

    /**

    * The number of messages discarded.

    */

    int getCount() const { return count; }

};

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

#if 15 < LOG4CXX_ABI_VERSION

}

#endif

#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()

    , buffer(DEFAULT_BUFFER_SIZE)

    , bufferSize(DEFAULT_BUFFER_SIZE)

    , dispatcher()

    , locationInfo(false)

    , 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;

  /**

   * Used to determine when to restart dispatch thread.

  */

  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)

  {

    // Restart dispatcher if it has stopped by an exception in an attached appender.

    if (!this->dispatcherActive && this->dispatcher.joinable())

      this->dispatcher.join();

    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

          );

      }

    }

  }

  void stopDispatcher()

  {

    bufferNotEmpty.notify_all();

    bufferNotFull.notify_all();

    if (dispatcher.joinable())

    {

      dispatcher.join();

    }

  }

  void close();

  /**

   * Should location info be included in dispatched messages.

  */

  bool locationInfo;

  /**

   * 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>())

{

}

Unexecuted instantiation: log4cxx::AsyncAppender::AsyncAppender()

Unexecuted instantiation: log4cxx::AsyncAppender::AsyncAppender()

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("LOCATIONINFO"), LOG4CXX_STR("locationinfo")))

  {

    setLocationInfo(OptionConverter::toBoolean(value, false));

  }

  if (StringHelper::equalsIgnoreCase(option, LOG4CXX_STR("BUFFERSIZE"), LOG4CXX_STR("buffersize")))

  {

    setBufferSize(OptionConverter::toInt(value, DEFAULT_BUFFER_SIZE));

  }

  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, Pool& pool1)

{

  doAppendImpl(event, pool1);

}

void AsyncAppender::append(const spi::LoggingEventPtr& event, Pool& p)

{

  if (priv->bufferSize <= 0)

  {

    priv->appenders.appendLoopOnAppenders(event, p);

    return;

  }

  // Get a copy of this thread's diagnostic context

  event->LoadDC();

  priv->checkDispatcher(getName());

  if (priv->dispatcher.get_id() == 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->bufferSize)

    {

      // 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)

        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->bufferSize;

      });

      --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();

}

void AsyncAppender::removeAppender(const AppenderPtr appender)

{

  priv->appenders.removeAppender(appender);

}

void AsyncAppender::removeAppender(const LogString& n)

{

  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);

}

bool AsyncAppender::getLocationInfo() const

{

  return priv->locationInfo;

}

void AsyncAppender::setLocationInfo(bool flag)

{

  priv->locationInfo = flag;

}

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->bufferMutex);

  if (priv->dispatcher.joinable())

  {

    throw RuntimeException(LOG4CXX_STR("AsyncAppender buffer size cannot be changed now"));

  }

  priv->bufferSize = (size < 1) ? 1 : size;

  priv->buffer.resize(priv->bufferSize);

  priv->bufferNotFull.notify_all();

}

int AsyncAppender::getBufferSize() const

{

  return priv->bufferSize;

}

void AsyncAppender::setBlocking(bool value)

{

  std::lock_guard<std::mutex> lock(priv->bufferMutex);

  priv->blocking = value;

  priv->bufferNotFull.notify_all();

}

bool AsyncAppender::getBlocking() const

{

  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))

{

}

#if LOG4CXX_ABI_VERSION <= 15

DiscardSummary::DiscardSummary(const LoggingEventPtr& event) :

  maxEvent(event), count(1)

{

}

DiscardSummary::DiscardSummary(const DiscardSummary& src) :

  maxEvent(src.maxEvent), count(src.count)

{

}

DiscardSummary& DiscardSummary::operator=(const DiscardSummary& src)

{

  maxEvent = src.maxEvent;

  count = src.count;

  return *this;

}

#endif

void DiscardSummary::add(const LoggingEventPtr& event)

{

  if (this->maxEvent->getLevel()->toInt() < event->getLevel()->toInt())

    this->maxEvent = event;

  ++this->count;

}

LoggingEventPtr DiscardSummary::createEvent(Pool& p)

{

  LogString msg(LOG4CXX_STR("Discarded "));

  StringHelper::toString(this->count, p, 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()

    );

}

#if LOG4CXX_ABI_VERSION <= 15

::LOG4CXX_NS::spi::LoggingEventPtr

DiscardSummary::createEvent(::LOG4CXX_NS::helpers::Pool& p,

  size_t discardedCount)

{

  LogString msg(LOG4CXX_STR("Discarded "));

  StringHelper::toString(discardedCount, p, msg);

  msg.append(LOG4CXX_STR(" messages due to a full event buffer"));

  return std::make_shared<LoggingEvent>(

        LOG4CXX_STR(""),

        LOG4CXX_NS::Level::getError(),

        msg,

        LocationInfo::getLocationUnavailable() );

}

#endif

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->bufferSize, 0);

  bool isActive = true;

  while (isActive)

  {

    Pool p;

    LoggingEventList events;

    events.reserve(this->bufferSize);

    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->bufferSize && 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(p));

        discardCount += discardItem.second.getCount();

      }

      this->discardMap.clear();

    }

    for (auto item : events)

    {

      try

      {

        this->appenders.appendLoopOnAppenders(item, p);

      }

      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 (1 < failureCount)

      break;

  }

  if (LogLog::isDebugEnabled())

  {

    Pool p;

    LogString msg(LOG4CXX_STR("[") + appenderName + LOG4CXX_STR("] AsyncAppender"));

#ifdef _DEBUG

    msg += LOG4CXX_STR(" iterationCount ");

    StringHelper::toString(iterationCount, p, msg);

    msg += LOG4CXX_STR(" waitCount ");

    StringHelper::toString(waitCount, p, msg);

    msg += LOG4CXX_STR(" producerBlockedCount ");

    StringHelper::toString(producerBlockedCount, p, msg);

    msg += LOG4CXX_STR(" commitCount ");

    StringHelper::toString(this->commitCount, p, msg);

#endif

    msg += LOG4CXX_STR(" dispatchedCount ");

    StringHelper::toString(this->dispatchedCount, p, msg);

    msg += LOG4CXX_STR(" discardCount ");

    StringHelper::toString(discardCount, p, msg);

    msg += LOG4CXX_STR(" pendingCountHistogram");

    for (auto item : pendingCountHistogram)

    {

      msg += logchar(' ');

      StringHelper::toString(item, p, msg);

    }

    LogLog::debug(msg);

  }

  this->dispatcherActive = false;

  if (0 < this->blockedCount) // Restart this dispatcher?

    this->bufferNotFull.notify_all();

}