/src/logging-log4cxx/src/main/cpp/threadutility.cpp

Source
/*
 * 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 <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;
  bool                      terminated{ false };
  int                       retryCount{ 2 };
  Period                    maxDelay{ 0 };

  void doPeriodicTasks();

  void setTerminated()
  {
    std::lock_guard<std::mutex> lock(interrupt_mutex);
    terminated = true;
  }

  void 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)
{
  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} );
  if (!m_priv->thread.joinable())
  {
    m_priv->terminated = false;
    m_priv->thread = createThread(LOG4CXX_STR("log4cxx"), std::bind(&priv_data::doPeriodicTasks, m_priv.get()));
  }
  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()
{
  {
    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)
  {
    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)
  {
    auto currentTime = std::chrono::system_clock::now();
    TimePoint nextOperationTime = currentTime + this->maxDelay;
    {
      std::lock_guard<std::recursive_mutex> lock(this->job_mutex);
      for (auto& item : this->jobs)
      {
        if (this->terminated)
          return;
        if (item.removed)
          ;
        else if (item.nextRun <= currentTime)
        {
          try
          {
            item.f();
            item.nextRun = std::chrono::system_clock::now() + item.delay;
            if (item.nextRun < nextOperationTime)
              nextOperationTime = item.nextRun;
            item.errorCount = 0;
          }
          catch (std::exception& ex)
          {
            LogLog::warn(item.name, ex);
            ++item.errorCount;
          }
          catch (...)
          {
            LogLog::warn(item.name + LOG4CXX_STR(" threw an exception"));
            ++item.errorCount;
          }
        }
        else if (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;
      this->jobs.erase(pItem);
      if (this->jobs.empty())
        return;
    }

    std::unique_lock<std::mutex> lock(this->interrupt_mutex);
    this->interrupt.wait_until(lock, nextOperationTime);
  }
}

} //namespace helpers
} //namespace log4cxx

Coverage Report

Created: 2025-10-27 06:48

Line	Count	Source
1		/*
2		* Licensed to the Apache Software Foundation (ASF) under one or more
3		* contributor license agreements. See the NOTICE file distributed with
4		* this work for additional information regarding copyright ownership.
5		* The ASF licenses this file to You under the Apache License, Version 2.0
6		* (the "License"); you may not use this file except in compliance with
7		* the License. You may obtain a copy of the License at
8		*
9		* http://www.apache.org/licenses/LICENSE-2.0
10		*
11		* Unless required by applicable law or agreed to in writing, software
12		* distributed under the License is distributed on an "AS IS" BASIS,
13		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14		* See the License for the specific language governing permissions and
15		* limitations under the License.
16		*/
17
18		#include "log4cxx/helpers/threadutility.h"
19		#if !defined(LOG4CXX)
20		#define LOG4CXX 1
21		#endif
22		#include "log4cxx/private/log4cxx_private.h"
23		#include "log4cxx/helpers/loglog.h"
24		#include "log4cxx/helpers/transcoder.h"
25
26		#include <signal.h>
27		#include <mutex>
28		#include <list>
29		#include <condition_variable>
30		#include <algorithm>
31
32		#ifdef _WIN32
33		#include <windows.h>
34		#include <processthreadsapi.h>
35		#endif
36
37		#if LOG4CXX_EVENTS_AT_EXIT
38		#include <log4cxx/private/atexitregistry.h>
39		#endif
40		#if !defined(LOG4CXX)
41		#define LOG4CXX 1
42		#endif
43		#include <log4cxx/helpers/aprinitializer.h>
44
45		namespace LOG4CXX_NS
46		{
47		namespace helpers
48		{
49
50		struct ThreadUtility::priv_data
51		{
52		priv_data()
53		#if LOG4CXX_EVENTS_AT_EXIT
54		: atExitRegistryRaii{ [this]{ stopThread(); } }
55		#endif
56	0	{
57	0	}
58
59		~priv_data()
60	0	{ stopThread(); }
61
62		ThreadStartPre start_pre{nullptr};
63		ThreadStarted started{nullptr};
64		ThreadStartPost start_post{nullptr};
65
66		using TimePoint = std::chrono::time_point<std::chrono::system_clock>;
67		struct NamedPeriodicFunction
68		{
69		LogString name;
70		Period delay;
71		TimePoint nextRun;
72		std::function<void()> f;
73		int errorCount;
74		bool removed;
75		};
76		using JobStore = std::list<NamedPeriodicFunction>;
77		JobStore jobs;
78		std::recursive_mutex job_mutex;
79		std::thread thread;
80		std::condition_variable interrupt;
81		std::mutex interrupt_mutex;
82		bool terminated{ false };
83		int retryCount{ 2 };
84		Period maxDelay{ 0 };
85
86		void doPeriodicTasks();
87
88		void setTerminated()
89	0	{
90	0	std::lock_guard<std::mutex> lock(interrupt_mutex);
91	0	terminated = true;
92	0	}
93
94		void stopThread()
95	0	{
96	0	setTerminated();
97	0	interrupt.notify_all();
98	0	if (thread.joinable())
99	0	thread.join();
100	0	}
101
102		#if LOG4CXX_EVENTS_AT_EXIT
103		helpers::AtExitRegistry::Raii atExitRegistryRaii;
104		#endif
105		};
106
107		#if LOG4CXX_HAS_PTHREAD_SIGMASK
108		static thread_local sigset_t old_mask;
109		static thread_local bool sigmask_valid;
110		#endif
111
112		ThreadUtility::ThreadUtility()
113	0	: m_priv( std::make_unique<priv_data>() )
114	0	{
115		// Block signals by default.
116	0	configureFuncs( std::bind( &ThreadUtility::preThreadBlockSignals, this ),
117	0	nullptr,
118	0	std::bind( &ThreadUtility::postThreadUnblockSignals, this ) );
119	0	}
120
121	0	ThreadUtility::~ThreadUtility() {}
122
123		auto ThreadUtility::instancePtr() -> ManagerPtr
124	0	{
125	0	auto result = APRInitializer::getOrAddUnique<Manager>
126	0	( []() -> ObjectPtr
127	0	{ return std::make_shared<Manager>(); }
128	0	);
129	0	return result;
130	0	}
131
132		ThreadUtility* ThreadUtility::instance()
133	0	{
134	0	return &instancePtr()->value();
135	0	}
136
137		void ThreadUtility::configure( ThreadConfigurationType type )
138	0	{
139	0	auto utility = instance();
140
141	0	if ( type == ThreadConfigurationType::NoConfiguration )
142	0	{
143	0	utility->configureFuncs( nullptr, nullptr, nullptr );
144	0	}
145	0	else if ( type == ThreadConfigurationType::NameThreadOnly )
146	0	{
147	0	utility->configureFuncs( nullptr,
148	0	std::bind( &ThreadUtility::threadStartedNameThread, utility,
149	0	std::placeholders::_1,
150	0	std::placeholders::_2,
151	0	std::placeholders::_3 ),
152	0	nullptr );
153	0	}
154	0	else if ( type == ThreadConfigurationType::BlockSignalsOnly )
155	0	{
156	0	utility->configureFuncs( std::bind( &ThreadUtility::preThreadBlockSignals, utility ),
157	0	nullptr,
158	0	std::bind( &ThreadUtility::postThreadUnblockSignals, utility ) );
159	0	}
160	0	else if ( type == ThreadConfigurationType::BlockSignalsAndNameThread )
161	0	{
162	0	utility->configureFuncs( std::bind( &ThreadUtility::preThreadBlockSignals, utility ),
163	0	std::bind( &ThreadUtility::threadStartedNameThread, utility,
164	0	std::placeholders::_1,
165	0	std::placeholders::_2,
166	0	std::placeholders::_3 ),
167	0	std::bind( &ThreadUtility::postThreadUnblockSignals, utility ) );
168	0	}
169	0	}
170
171		void ThreadUtility::configureFuncs( ThreadStartPre pre_start,
172		ThreadStarted started,
173		ThreadStartPost post_start )
174	0	{
175	0	m_priv->start_pre = pre_start;
176	0	m_priv->started = started;
177	0	m_priv->start_post = post_start;
178	0	}
179
180		void ThreadUtility::preThreadBlockSignals()
181	0	{
182	0	#if LOG4CXX_HAS_PTHREAD_SIGMASK
183	0	sigset_t set;
184	0	sigfillset(&set);
185
186	0	if ( pthread_sigmask(SIG_SETMASK, &set, &old_mask) < 0 )
187	0	{
188	0	LOGLOG_ERROR( LOG4CXX_STR("Unable to set thread sigmask") );
189	0	sigmask_valid = false;
190	0	}
191	0	else
192	0	{
193	0	sigmask_valid = true;
194	0	}
195
196	0	#endif /* LOG4CXX_HAS_PTHREAD_SIGMASK */
197	0	}
198
199		void ThreadUtility::threadStartedNameThread(LogString threadName,
200		std::thread::id /threadId/,
201		std::thread::native_handle_type nativeHandle)
202	0	{
203	0	#if LOG4CXX_HAS_PTHREAD_SETNAME && !(defined(_WIN32) && defined(_LIBCPP_VERSION))
204	0	LOG4CXX_ENCODE_CHAR(sthreadName, threadName);
205	0	if (pthread_setname_np(static_cast<pthread_t>(nativeHandle), sthreadName.c_str()) < 0) {
206	0	LOGLOG_ERROR(LOG4CXX_STR("unable to set thread name"));
207	0	}
208		#elif defined(_WIN32)
209		typedef HRESULT (WINAPI *TSetThreadDescription)(HANDLE, PCWSTR);
210		static struct initialiser
211		{
212		HMODULE hKernelBase;
213		TSetThreadDescription SetThreadDescription;
214		initialiser()
215		: hKernelBase(GetModuleHandleA("KernelBase.dll"))
216		, SetThreadDescription(nullptr)
217		{
218		if (hKernelBase)
219		SetThreadDescription = reinterpret_cast<TSetThreadDescription>(GetProcAddress(hKernelBase, "SetThreadDescription"));
220		}
221		} win32Func;
222		if (win32Func.SetThreadDescription)
223		{
224		LOG4CXX_ENCODE_WCHAR(wthreadName, threadName);
225		if(FAILED(win32Func.SetThreadDescription(static_cast<HANDLE>(nativeHandle), wthreadName.c_str())))
226		LOGLOG_ERROR( LOG4CXX_STR("unable to set thread name") );
227		}
228		#endif
229	0	}
230
231		void ThreadUtility::postThreadUnblockSignals()
232	0	{
233	0	#if LOG4CXX_HAS_PTHREAD_SIGMASK
234
235		// Only restore the signal mask if we were able to set it in the first place.
236	0	if ( sigmask_valid )
237	0	{
238	0	if ( pthread_sigmask(SIG_SETMASK, &old_mask, nullptr) < 0 )
239	0	{
240	0	LOGLOG_ERROR( LOG4CXX_STR("Unable to set thread sigmask") );
241	0	}
242	0	}
243
244	0	#endif /* LOG4CXX_HAS_PTHREAD_SIGMASK */
245	0	}
246
247
248		ThreadStartPre ThreadUtility::preStartFunction()
249	0	{
250	0	return m_priv->start_pre;
251	0	}
252
253		ThreadStarted ThreadUtility::threadStartedFunction()
254	0	{
255	0	return m_priv->started;
256	0	}
257
258		ThreadStartPost ThreadUtility::postStartFunction()
259	0	{
260	0	return m_priv->start_post;
261	0	}
262
263		/**
264		* Add a periodic task
265		*/
266		void ThreadUtility::addPeriodicTask(const LogString& name, std::function<void()> f, const Period& delay)
267	0	{
268	0	std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);
269	0	if (m_priv->maxDelay < delay)
270	0	m_priv->maxDelay = delay;
271	0	auto currentTime = std::chrono::system_clock::now();
272	0	m_priv->jobs.push_back( priv_data::NamedPeriodicFunction{name, delay, currentTime + delay, f, 0, false} );
273	0	if (!m_priv->thread.joinable())
274	0	{
275	0	m_priv->terminated = false;
276	0	m_priv->thread = createThread(LOG4CXX_STR("log4cxx"), std::bind(&priv_data::doPeriodicTasks, m_priv.get()));
277	0	}
278	0	else
279	0	m_priv->interrupt.notify_one();
280	0	}
281
282		/**
283		* Is this already running a \c taskName periodic task?
284		*/
285		bool ThreadUtility::hasPeriodicTask(const LogString& name)
286	0	{
287	0	std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);
288	0	auto pItem = std::find_if(m_priv->jobs.begin(), m_priv->jobs.end()
289	0	, [&name](const priv_data::NamedPeriodicFunction& item)
290	0	{ return !item.removed && name == item.name; }
291	0	);
292	0	return m_priv->jobs.end() != pItem;
293	0	}
294
295		/**
296		* Remove all periodic tasks and stop the processing thread
297		*/
298		void ThreadUtility::removeAllPeriodicTasks()
299	0	{
300	0	{
301	0	std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);
302	0	while (!m_priv->jobs.empty())
303	0	m_priv->jobs.pop_back();
304	0	}
305	0	m_priv->stopThread();
306	0	}
307
308		/**
309		* Remove the \c taskName periodic task
310		*/
311		void ThreadUtility::removePeriodicTask(const LogString& name)
312	0	{
313	0	std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);
314	0	auto pItem = std::find_if(m_priv->jobs.begin(), m_priv->jobs.end()
315	0	, [&name](const priv_data::NamedPeriodicFunction& item)
316	0	{ return !item.removed && name == item.name; }
317	0	);
318	0	if (m_priv->jobs.end() != pItem)
319	0	{
320	0	pItem->removed = true;
321	0	m_priv->interrupt.notify_one();
322	0	}
323	0	}
324
325		/**
326		* Remove any periodic task matching \c namePrefix
327		*/
328		void ThreadUtility::removePeriodicTasksMatching(const LogString& namePrefix)
329	0	{
330	0	while (1)
331	0	{
332	0	std::lock_guard<std::recursive_mutex> lock(m_priv->job_mutex);
333	0	auto pItem = std::find_if(m_priv->jobs.begin(), m_priv->jobs.end()
334	0	, [&namePrefix](const priv_data::NamedPeriodicFunction& item)
335	0	{ return !item.removed && namePrefix.size() <= item.name.size() && item.name.substr(0, namePrefix.size()) == namePrefix; }
336	0	);
337	0	if (m_priv->jobs.end() == pItem)
338	0	break;
339	0	pItem->removed = true;
340	0	}
341	0	m_priv->interrupt.notify_one();
342	0	}
343
344		// Run ready tasks
345		void ThreadUtility::priv_data::doPeriodicTasks()
346	0	{
347	0	while (!this->terminated)
348	0	{
349	0	auto currentTime = std::chrono::system_clock::now();
350	0	TimePoint nextOperationTime = currentTime + this->maxDelay;
351	0	{
352	0	std::lock_guard<std::recursive_mutex> lock(this->job_mutex);
353	0	for (auto& item : this->jobs)
354	0	{
355	0	if (this->terminated)
356	0	return;
357	0	if (item.removed)
358	0	;
359	0	else if (item.nextRun <= currentTime)
360	0	{
361	0	try
362	0	{
363	0	item.f();
364	0	item.nextRun = std::chrono::system_clock::now() + item.delay;
365	0	if (item.nextRun < nextOperationTime)
366	0	nextOperationTime = item.nextRun;
367	0	item.errorCount = 0;
368	0	}
369	0	catch (std::exception& ex)
370	0	{
371	0	LogLog::warn(item.name, ex);
372	0	++item.errorCount;
373	0	}
374	0	catch (...)
375	0	{
376	0	LogLog::warn(item.name + LOG4CXX_STR(" threw an exception"));
377	0	++item.errorCount;
378	0	}
379	0	}
380	0	else if (item.nextRun < nextOperationTime)
381	0	nextOperationTime = item.nextRun;
382	0	}
383	0	}
384		// Delete removed and faulty tasks
385	0	while (1)
386	0	{
387	0	std::lock_guard<std::recursive_mutex> lock(this->job_mutex);
388	0	auto pItem = std::find_if(this->jobs.begin(), this->jobs.end()
389	0	, [this](const NamedPeriodicFunction& item)
390	0	{ return item.removed \|\| this->retryCount < item.errorCount; }
391	0	);
392	0	if (this->jobs.end() == pItem)
393	0	break;
394	0	this->jobs.erase(pItem);
395	0	if (this->jobs.empty())
396	0	return;
397	0	}
398
399	0	std::unique_lock<std::mutex> lock(this->interrupt_mutex);
400	0	this->interrupt.wait_until(lock, nextOperationTime);
401	0	}
402	0	}
403
404		} //namespace helpers
405		} //namespace log4cxx