/*

 * 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/helpers/threadutility.h"

#if !defined(LOG4CXX)

  #define LOG4CXX 1

#endif

#include "log4cxx/private/log4cxx_private.h"

#include "log4cxx/helpers/loglog.h"

#include "log4cxx/helpers/transcoder.h"

#include <atomic>

#include <signal.h>

#include <mutex>

#include <list>

#include <condition_variable>

#include <algorithm>

#ifdef _WIN32

  #include <windows.h>

  #include <processthreadsapi.h>

#endif

#if LOG4CXX_EVENTS_AT_EXIT

#include <log4cxx/private/atexitregistry.h>

#endif

#if !defined(LOG4CXX)

  #define LOG4CXX 1

#endif

#include <log4cxx/helpers/aprinitializer.h>

namespace LOG4CXX_NS

{

namespace helpers

{

struct ThreadUtility::priv_data

{

  priv_data()

#if LOG4CXX_EVENTS_AT_EXIT

    : atExitRegistryRaii{ [this]{ stopThread(); } }

#endif

  {

  }

  ~priv_data()

  { stopThread(); }

  ThreadStartPre  start_pre{nullptr};

  ThreadStarted   started{nullptr};

  ThreadStartPost start_post{nullptr};

  using TimePoint = std::chrono::time_point<std::chrono::system_clock>;

  struct NamedPeriodicFunction

  {

    LogString             name;

    Period                delay;

    TimePoint             nextRun;

    std::function<void()> f;

    int                   errorCount;

    bool                  removed;

  };

  using JobStore = std::list<NamedPeriodicFunction>;

  JobStore                  jobs;

  std::recursive_mutex      job_mutex;

  std::thread               thread;

  std::condition_variable   interrupt;

  std::mutex                interrupt_mutex;

  std::atomic<bool>         terminated{ false };

  int                       retryCount{ 2 };

  Period                    maxDelay{ 0 };

  std::atomic<bool>         threadIsActive{ false };

  LoggerPtr                 log;

  void doPeriodicTasks();

  void setTerminated()

  {

    std::lock_guard<std::recursive_mutex> lock(job_mutex);

    terminated.store(true);

  }

  void stopThread()

  {

    LOGLOG_DEBUG(log, "stopThread");

    setTerminated();

    interrupt.notify_all();

    if (thread.joinable())

      thread.join();

  }

#if LOG4CXX_EVENTS_AT_EXIT

  helpers::AtExitRegistry::Raii atExitRegistryRaii;

#endif

};

#if LOG4CXX_HAS_PTHREAD_SIGMASK

  static thread_local sigset_t old_mask;

  static thread_local bool sigmask_valid;

#endif

ThreadUtility::ThreadUtility()

  : m_priv( std::make_unique<priv_data>() )

{

  // Block signals by default.

  configureFuncs( std::bind( &ThreadUtility::preThreadBlockSignals, this ),

    nullptr,

    std::bind( &ThreadUtility::postThreadUnblockSignals, this ) );

}

ThreadUtility::~ThreadUtility() {}

auto ThreadUtility::instancePtr() -> ManagerPtr

{

  auto result = APRInitializer::getOrAddUnique<Manager>

    ( []() -> ObjectPtr

      { return std::make_shared<Manager>(); }

    );

  return result;

}

ThreadUtility* ThreadUtility::instance()

{

  return &instancePtr()->value();

}

void ThreadUtility::configure( ThreadConfigurationType type )

{

  auto utility = instance();

  if ( type == ThreadConfigurationType::NoConfiguration )

  {

    utility->configureFuncs( nullptr, nullptr, nullptr );

  }

  else if ( type == ThreadConfigurationType::NameThreadOnly )

  {

    utility->configureFuncs( nullptr,

      std::bind( &ThreadUtility::threadStartedNameThread, utility,

        std::placeholders::_1,

        std::placeholders::_2,

        std::placeholders::_3 ),

      nullptr );

  }

  else if ( type == ThreadConfigurationType::BlockSignalsOnly )

  {

    utility->configureFuncs( std::bind( &ThreadUtility::preThreadBlockSignals, utility ),

      nullptr,

      std::bind( &ThreadUtility::postThreadUnblockSignals, utility ) );

  }

  else if ( type == ThreadConfigurationType::BlockSignalsAndNameThread )

  {

    utility->configureFuncs( std::bind( &ThreadUtility::preThreadBlockSignals, utility ),

      std::bind( &ThreadUtility::threadStartedNameThread, utility,

        std::placeholders::_1,

        std::placeholders::_2,

        std::placeholders::_3 ),

      std::bind( &ThreadUtility::postThreadUnblockSignals, utility ) );

  }

}

void ThreadUtility::configureFuncs( ThreadStartPre pre_start,

  ThreadStarted started,

  ThreadStartPost post_start )

{

  m_priv->start_pre = pre_start;

  m_priv->started = started;

  m_priv->start_post = post_start;

}

void ThreadUtility::preThreadBlockSignals()

{

#if LOG4CXX_HAS_PTHREAD_SIGMASK

  sigset_t set;

  sigfillset(&set);

  if ( pthread_sigmask(SIG_SETMASK, &set, &old_mask) < 0 )

  {

    LOGLOG_ERROR( LOG4CXX_STR("Unable to set thread sigmask") );

    sigmask_valid = false;

  }

  else

  {

    sigmask_valid = true;

  }

#endif /* LOG4CXX_HAS_PTHREAD_SIGMASK */

}

void ThreadUtility::threadStartedNameThread(LogString threadName,

  std::thread::id /*threadId*/,

  std::thread::native_handle_type nativeHandle)

{

#if LOG4CXX_HAS_PTHREAD_SETNAME && !(defined(_WIN32) && defined(_LIBCPP_VERSION))

  LOG4CXX_ENCODE_CHAR(sthreadName, threadName);

  if (pthread_setname_np(static_cast<pthread_t>(nativeHandle), sthreadName.c_str()) < 0) {

    LOGLOG_ERROR(LOG4CXX_STR("unable to set thread name"));

  }

#elif defined(_WIN32)

  typedef HRESULT (WINAPI *TSetThreadDescription)(HANDLE, PCWSTR);

  static struct initialiser

  {

    HMODULE hKernelBase;

    TSetThreadDescription SetThreadDescription;

    initialiser()

      : hKernelBase(GetModuleHandleA("KernelBase.dll"))

      , SetThreadDescription(nullptr)

    {

      if (hKernelBase)

        SetThreadDescription = reinterpret_cast<TSetThreadDescription>(GetProcAddress(hKernelBase, "SetThreadDescription"));

    }

  } win32Func;

  if (win32Func.SetThreadDescription)

  {

    LOG4CXX_ENCODE_WCHAR(wthreadName, threadName);

    if(FAILED(win32Func.SetThreadDescription(static_cast<HANDLE>(nativeHandle), wthreadName.c_str())))

      LOGLOG_ERROR( LOG4CXX_STR("unable to set thread name") );

  }

#endif

}

void ThreadUtility::postThreadUnblockSignals()

{

#if LOG4CXX_HAS_PTHREAD_SIGMASK

  // Only restore the signal mask if we were able to set it in the first place.

  if ( sigmask_valid )

  {

    if ( pthread_sigmask(SIG_SETMASK, &old_mask, nullptr) < 0 )

    {

      LOGLOG_ERROR( LOG4CXX_STR("Unable to set thread sigmask") );

    }

  }

#endif /* LOG4CXX_HAS_PTHREAD_SIGMASK */

}

ThreadStartPre ThreadUtility::preStartFunction()

{

  return m_priv->start_pre;

}

ThreadStarted ThreadUtility::threadStartedFunction()

{

  return m_priv->started;

}

ThreadStartPost ThreadUtility::postStartFunction()

{

  return m_priv->start_post;

}

/**

 * Add a periodic task

 */

void ThreadUtility::addPeriodicTask(const LogString& name, std::function<void()> f, const Period& delay)

{

  if (!m_priv->log)

    m_priv->log = Logger::getLogger("ThreadUtility");

  LOGLOG_DEBUG(m_priv->log, LOG4CXX_STR("addPeriodicTask: ") << name);

  std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);

  if (m_priv->maxDelay < delay)

    m_priv->maxDelay = delay;

  auto currentTime = std::chrono::system_clock::now();

  m_priv->jobs.push_back( priv_data::NamedPeriodicFunction{name, delay, currentTime + delay, f, 0, false} );

  // Restart thread if it has stopped.

  if (!m_priv->threadIsActive.load() && m_priv->thread.joinable())

    m_priv->thread.join();

  if (!m_priv->thread.joinable())

  {

    m_priv->terminated.store(false);

    m_priv->threadIsActive.store(true);

    m_priv->thread = createThread(LOG4CXX_STR("log4cxx"), [this]()

      {

        LOGLOG_DEBUG(m_priv->log, "doPeriodicTasks: " << "started");

        m_priv->doPeriodicTasks();

        LOGLOG_DEBUG(m_priv->log, "doPeriodicTasks: " << "stopped");

        m_priv->threadIsActive.store(false);

      });

  }

  else

    m_priv->interrupt.notify_one();

}

/**

 * Is this already running a \c taskName periodic task?

 */

bool ThreadUtility::hasPeriodicTask(const LogString& name)

{

  std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);

  auto pItem = std::find_if(m_priv->jobs.begin(), m_priv->jobs.end()

    , [&name](const priv_data::NamedPeriodicFunction& item)

    { return !item.removed && name == item.name; }

    );

  return m_priv->jobs.end() != pItem;

}

/**

 * Remove all periodic tasks and stop the processing thread

 */

void ThreadUtility::removeAllPeriodicTasks()

{

  LOGLOG_DEBUG(m_priv->log, "removeAllPeriodicTasks");

  {

    std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);

    while (!m_priv->jobs.empty())

      m_priv->jobs.pop_back();

  }

  m_priv->stopThread();

}

/**

 * Remove the \c taskName periodic task

 */

void ThreadUtility::removePeriodicTask(const LogString& name)

{

  std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);

  auto pItem = std::find_if(m_priv->jobs.begin(), m_priv->jobs.end()

    , [&name](const priv_data::NamedPeriodicFunction& item)

    { return !item.removed && name == item.name; }

    );

  if (m_priv->jobs.end() != pItem)

  {

    LOGLOG_DEBUG(m_priv->log, LOG4CXX_STR("removePeriodicTask: ") << name);

    pItem->removed = true;

    m_priv->interrupt.notify_one();

  }

}

/**

 * Remove any periodic task matching \c namePrefix

 */

void ThreadUtility::removePeriodicTasksMatching(const LogString& namePrefix)

{

  while (1)

  {

    std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);

    auto pItem = std::find_if(m_priv->jobs.begin(), m_priv->jobs.end()

      , [&namePrefix](const priv_data::NamedPeriodicFunction& item)

      { return !item.removed && namePrefix.size() <= item.name.size() && item.name.substr(0, namePrefix.size()) == namePrefix; }

      );

    if (m_priv->jobs.end() == pItem)

      break;

    pItem->removed = true;

  }

  m_priv->interrupt.notify_one();

}

// Run ready tasks

void ThreadUtility::priv_data::doPeriodicTasks()

{

  while (!this->terminated.load())

  {

    auto currentTime = std::chrono::system_clock::now();

    TimePoint nextOperationTime = currentTime + this->maxDelay;

    // Run each due task with job_mutex released, so a long-running callback

    // (e.g. a reconnect blocked on network I/O) does not stall removePeriodicTask()

    while (true)

    {

      std::function<void()> taskCallback;

      LogString taskName;

      Period taskDelay;

      bool foundTask = false;

      // Take a copy of the next due task while holding the lock

      {

        if (this->terminated.load())

          return;

        std::lock_guard<std::recursive_mutex> lock(this->job_mutex);

        auto pItem = std::find_if(this->jobs.begin(), this->jobs.end()

          , [currentTime](const NamedPeriodicFunction& item)

          { return !item.removed && item.nextRun <= currentTime; }

          );

        if (pItem != this->jobs.end())

        {

          taskCallback = pItem->f;

          taskName = pItem->name;

          taskDelay = pItem->delay;

          foundTask = true;

        }

      }

      // No more tasks are due, leave the loop

      if (!foundTask)

      {

        break;

      }

      // Execute the callback outside the lock

      bool success = false;

      try

      {

        taskCallback();

        success = true;

      }

      catch (std::exception& ex)

      {

        LogLog::warn(taskName, ex);

      }

      catch (...)

      {

        LogLog::warn(taskName + LOG4CXX_STR(" threw an exception"));

      }

      // Re-find the task (it may have been removed while running) and reschedule it

      {

        std::lock_guard<std::recursive_mutex> lock(this->job_mutex);

        auto pItem = std::find_if(this->jobs.begin(), this->jobs.end()

          , [&taskName](const NamedPeriodicFunction& item)

          { return !item.removed && taskName == item.name; }

          );

        if (pItem != this->jobs.end())

        {

          // Always push nextRun out, so a failing task waits before the next retry

          pItem->nextRun = std::chrono::system_clock::now() + taskDelay;

          if (success)

            pItem->errorCount = 0;

          else

            ++pItem->errorCount;

        }

      }

    }

    // Update nextOperationTime under the lock

    {

      std::lock_guard<std::recursive_mutex> lock(this->job_mutex);

      for (const auto& item : this->jobs)

      {

        if (!item.removed && item.nextRun < nextOperationTime)

        {

          nextOperationTime = item.nextRun;

        }

      }

    }

    // Delete removed and faulty tasks

    while (1)

    {

      std::lock_guard<std::recursive_mutex> lock(this->job_mutex);

      auto pItem = std::find_if(this->jobs.begin(), this->jobs.end()

        , [this](const NamedPeriodicFunction& item)

        { return item.removed || this->retryCount < item.errorCount; }

        );

      if (this->jobs.end() == pItem)

        break;

      LOGLOG_DEBUG(this->log, LOG4CXX_STR("doPeriodicTasks: erase ") << pItem->name);

      this->jobs.erase(pItem);

      if (this->jobs.empty())

        return;

    }

    // Wait until the next task is due or an add/remove/shutdown wakes us

    std::unique_lock<std::mutex> lock(this->interrupt_mutex);

    this->interrupt.wait_until(lock, nextOperationTime);

  }

}

} //namespace helpers

} //namespace log4cxx