1  /*
  2  * Copyright (c) 2013-2024, The PurpleI2P Project
  3  *
  4  * This file is part of Purple i2pd project and licensed under BSD3
  5  *
  6  * See full license text in LICENSE file at top of project tree
  7  */
  8  
  9  #ifndef QUEUE_H__
 10  #define QUEUE_H__
 11  
 12  #include <list>
 13  #include <mutex>
 14  #include <thread>
 15  #include <condition_variable>
 16  #include <functional>
 17  #include <utility>
 18  
 19  namespace i2p
 20  {
 21  namespace util
 22  {
 23  	template<typename Element>
 24  	class Queue
 25  	{
 26  		public:
 27  
 28  			void Put (Element e)
 29  			{
 30  				std::unique_lock<std::mutex> l(m_QueueMutex);
 31  				m_Queue.push_back (std::move(e));
 32  				m_NonEmpty.notify_one ();
 33  			}
 34  
 35  			void Put (std::list<Element>& list)
 36  			{
 37  				if (!list.empty ())
 38  				{
 39  					std::unique_lock<std::mutex> l(m_QueueMutex);
 40  					m_Queue.splice (m_Queue.end (), list); 
 41  					m_NonEmpty.notify_one ();
 42  				}	
 43  			}		
 44  		
 45  			Element GetNext ()
 46  			{
 47  				std::unique_lock<std::mutex> l(m_QueueMutex);
 48  				auto el = GetNonThreadSafe ();
 49  				if (!el)
 50  				{
 51  					m_NonEmpty.wait (l);
 52  					el = GetNonThreadSafe ();
 53  				}
 54  				return el;
 55  			}
 56  
 57  			Element GetNextWithTimeout (int usec)
 58  			{
 59  				std::unique_lock<std::mutex> l(m_QueueMutex);
 60  				auto el = GetNonThreadSafe ();
 61  				if (!el)
 62  				{
 63  					m_NonEmpty.wait_for (l, std::chrono::milliseconds (usec));
 64  					el = GetNonThreadSafe ();
 65  				}
 66  				return el;
 67  			}
 68  
 69  			void Wait ()
 70  			{
 71  				std::unique_lock<std::mutex> l(m_QueueMutex);
 72  				m_NonEmpty.wait (l);
 73  			}
 74  
 75  			bool Wait (int sec, int usec)
 76  			{
 77  				std::unique_lock<std::mutex> l(m_QueueMutex);
 78  				return m_NonEmpty.wait_for (l, std::chrono::seconds (sec) + std::chrono::milliseconds (usec)) != std::cv_status::timeout;
 79  			}
 80  
 81  			bool IsEmpty ()
 82  			{
 83  				std::unique_lock<std::mutex> l(m_QueueMutex);
 84  				return m_Queue.empty ();
 85  			}
 86  
 87  			int GetSize () const
 88  			{
 89  				std::unique_lock<std::mutex> l(m_QueueMutex);
 90  				return m_Queue.size ();
 91  			}
 92  
 93  			void WakeUp () { m_NonEmpty.notify_all (); };
 94  
 95  			Element Get ()
 96  			{
 97  				std::unique_lock<std::mutex> l(m_QueueMutex);
 98  				return GetNonThreadSafe ();
 99  			}
100  
101  			Element Peek ()
102  			{
103  				std::unique_lock<std::mutex> l(m_QueueMutex);
104  				return GetNonThreadSafe (true);
105  			}
106  
107  			void GetWholeQueue (std::list<Element>& queue)
108  			{
109  				if (!queue.empty ())
110  				{	
111  					std::list<Element> newQueue;
112  					queue.swap (newQueue);
113  				}	
114  				{
115  					std::unique_lock<std::mutex> l(m_QueueMutex);
116  					m_Queue.swap (queue);
117  				}
118  			}		
119  
120  		private:
121  		
122  			Element GetNonThreadSafe (bool peek = false)
123  			{
124  				if (!m_Queue.empty ())
125  				{
126  					auto el = m_Queue.front ();
127  					if (!peek)
128  						m_Queue.pop_front ();
129  					return el;
130  				}
131  				return nullptr;
132  			}
133  
134  		private:
135  
136  			std::list<Element> m_Queue;
137  			mutable std::mutex m_QueueMutex;
138  			std::condition_variable m_NonEmpty;
139  	};
140  }
141  }
142  
143  #endif