<commit_before>#ifndef QUEUE_H__
#define QUEUE_H__

#include <queue>
#include <mutex>
#include <thread>
#include <condition_variable>

namespace i2p
{
namespace util
{
	template<typename Element>
	class Queue
	{	
		public:

			void Put (Element * e)
			{
				std::unique_lock<std::mutex>  l(m_QueueMutex);
				m_Queue.push (e);	
				m_NonEmpty.notify_one ();
			}

			Element * GetNext ()
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				Element * el = GetNonThreadSafe ();
				if (!el)
				{
					m_NonEmpty.wait (l);
					el = GetNonThreadSafe ();
				}	
				return el;
			}

			Element * GetNextWithTimeout (int usec)
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				Element * el = GetNonThreadSafe ();
				if (!el)
				{
					m_NonEmpty.wait_for (l, std::chrono::milliseconds (usec));
					el = GetNonThreadSafe ();
				}	
				return el;
			}

			bool Wait (int sec, int usec)
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return m_NonEmpty.wait_for (l, std::chrono::seconds (sec) + std::chrono::milliseconds (usec)) != std::cv_status::timeout;
			}

			bool IsEmpty () 
			{	
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return m_Queue.empty ();
			}
			
			void WakeUp () { m_NonEmpty.notify_one (); };

			Element * Get ()
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return GetNonThreadSafe ();
			}	

			Element * Peek ()
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return GetNonThreadSafe (true);
			}	
			
		private:

			Element * GetNonThreadSafe (bool peek = false)
			{
				if (!m_Queue.empty ())
				{
					Element * el = m_Queue.front ();
					if (!peek)
						m_Queue.pop ();
					return el;
				}				
				return nullptr;
			}	
			
		private:

			std::queue<Element *> m_Queue;
			std::mutex m_QueueMutex;
			std::condition_variable m_NonEmpty;
	};	

	template<class Msg>
	class MsgQueue: public Queue<Msg>
	{
		public:

			MsgQueue (): m_Thread (std::bind (&MsgQueue<Msg>::Run, this)) {};
<commit_msg>Should fix the segfault at exit<commit_after>#ifndef QUEUE_H__
#define QUEUE_H__

#include <queue>
#include <mutex>
#include <thread>
#include <condition_variable>

namespace i2p
{
namespace util
{
	template<typename Element>
	class Queue
	{	
		public:

			void Put (Element * e)
			{
				std::unique_lock<std::mutex>  l(m_QueueMutex);
				m_Queue.push (e);	
				m_NonEmpty.notify_one ();
			}

			Element * GetNext ()
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				Element * el = GetNonThreadSafe ();
				if (!el)
				{
					m_NonEmpty.wait (l);
					el = GetNonThreadSafe ();
				}	
				return el;
			}

			Element * GetNextWithTimeout (int usec)
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				Element * el = GetNonThreadSafe ();
				if (!el)
				{
					m_NonEmpty.wait_for (l, std::chrono::milliseconds (usec));
					el = GetNonThreadSafe ();
				}	
				return el;
			}

			bool Wait (int sec, int usec)
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return m_NonEmpty.wait_for (l, std::chrono::seconds (sec) + std::chrono::milliseconds (usec)) != std::cv_status::timeout;
			}

			bool IsEmpty () 
			{	
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return m_Queue.empty ();
			}
			
			void WakeUp () { m_NonEmpty.notify_one (); };

			Element * Get ()
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return GetNonThreadSafe ();
			}	

			Element * Peek ()
			{
				std::unique_lock<std::mutex> l(m_QueueMutex);
				return GetNonThreadSafe (true);
			}	
			
		private:

			Element * GetNonThreadSafe (bool peek = false)
			{
				if (!m_Queue.empty ())
				{
					Element * el = m_Queue.front ();
					if (!peek)
						m_Queue.pop ();
					return el;
				}				
				return nullptr;
			}	
			
		private:

			std::queue<Element *> m_Queue;
			std::mutex m_QueueMutex;
			std::condition_variable m_NonEmpty;
	};	

	template<class Msg>
	class MsgQueue: public Queue<Msg>
	{
		public:

			MsgQueue (): m_Thread (std::bind (&MsgQueue<Msg>::Run, this)) {};
			void Stop()
			{
					m_Thread.detach();
			}

{'fraction_non_alphanumeric': 0.14331603657029898, 'fraction_numerical': 0.0004941932295527551, 'mean_word_length': 2.9073359073359075, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2378000', 'n_tokens_mistral': 1890, 'n_tokens_neox': 1488, 'n_words': 286}