Make MutexQueue use jsemaphore for signaling

2 files changed, 141 insertions, 72 deletions

diff --git a/src/util/container.h b/src/util/container.h
index e83c3cd37..6d836a4d5 100644
--- a/src/util/container.h
+++ b/src/util/container.h
@@ -24,7 +24,7 @@ with this program; if not, write to the Free Software Foundation, Inc.,
 #include "../exceptions.h"
 #include "../jthread/jmutex.h"
 #include "../jthread/jmutexautolock.h"
-#include "../porting.h" // For sleep_ms
+#include "../jthread/jsemaphore.h"
 #include <list>
 #include <vector>
 #include <map>
@@ -201,6 +201,12 @@ public:
 		++m_list_size;
 	}
 	
+	void push_front(T t)
+	{
+		m_list.push_front(t);
+		++m_list_size;
+	}
+
 	T pop_front()
 	{
 		if(m_list.empty())
@@ -247,86 +253,141 @@ template<typename T>
 class MutexedQueue
 {
 public:
+	template<typename Key, typename U, typename Caller, typename CallerData>
+	friend class RequestQueue;
+
 	MutexedQueue()
 	{
 	}
 	bool empty()
 	{
 		JMutexAutoLock lock(m_mutex);
-		return m_list.empty();
+		return (m_size.GetValue() == 0);
 	}
 	void push_back(T t)
 	{
 		JMutexAutoLock lock(m_mutex);
 		m_list.push_back(t);
+		m_size.Post();
 	}
-	T pop_front(u32 wait_time_max_ms=0)
+
+	/* this version of pop_front returns a empty element of T on timeout.
+	 * Make sure default constructor of T creates a recognizable "empty" element
+	 */
+	T pop_frontNoEx(u32 wait_time_max_ms)
 	{
-		u32 wait_time_ms = 0;
+		if (m_size.Wait(wait_time_max_ms))
+		{
+			JMutexAutoLock lock(m_mutex);
 
-		for(;;)
+			typename std::list<T>::iterator begin = m_list.begin();
+			T t = *begin;
+			m_list.erase(begin);
+			return t;
+		}
+		else
 		{
-			{
-				JMutexAutoLock lock(m_mutex);
-
-				if(!m_list.empty())
-				{
-					typename std::list<T>::iterator begin = m_list.begin();
-					T t = *begin;
-					m_list.erase(begin);
-					return t;
-				}
-
-				if(wait_time_ms >= wait_time_max_ms)
-					throw ItemNotFoundException("MutexedQueue: queue is empty");
-			}
-
-			// Wait a while before trying again
-			sleep_ms(10);
-			wait_time_ms += 10;
+			return T();
 		}
 	}
+
+	T pop_front(u32 wait_time_max_ms)
+	{
+		if (m_size.Wait(wait_time_max_ms))
+		{
+			JMutexAutoLock lock(m_mutex);
+
+			typename std::list<T>::iterator begin = m_list.begin();
+			T t = *begin;
+			m_list.erase(begin);
+			return t;
+		}
+		else
+		{
+			throw ItemNotFoundException("MutexedQueue: queue is empty");
+		}
+	}
+
+	T pop_frontNoEx()
+	{
+		m_size.Wait();
+
+		JMutexAutoLock lock(m_mutex);
+
+		typename std::list<T>::iterator begin = m_list.begin();
+		T t = *begin;
+		m_list.erase(begin);
+		return t;
+	}
+
 	T pop_back(u32 wait_time_max_ms=0)
 	{
-		u32 wait_time_ms = 0;
+		if (m_size.Wait(wait_time_max_ms))
+		{
+			JMutexAutoLock lock(m_mutex);
+
+			typename std::list<T>::iterator last = m_list.end();
+			last--;
+			T t = *last;
+			m_list.erase(last);
+			return t;
+		}
+		else
+		{
+			throw ItemNotFoundException("MutexedQueue: queue is empty");
+		}
+	}
+
+	/* this version of pop_back returns a empty element of T on timeout.
+	 * Make sure default constructor of T creates a recognizable "empty" element
+	 */
+	T pop_backNoEx(u32 wait_time_max_ms=0)
+	{
+		if (m_size.Wait(wait_time_max_ms))
+		{
+			JMutexAutoLock lock(m_mutex);
 
-		for(;;)
+			typename std::list<T>::iterator last = m_list.end();
+			last--;
+			T t = *last;
+			m_list.erase(last);
+			return t;
+		}
+		else
 		{
-			{
-				JMutexAutoLock lock(m_mutex);
-
-				if(!m_list.empty())
-				{
-					typename std::list<T>::iterator last = m_list.end();
-					last--;
-					T t = *last;
-					m_list.erase(last);
-					return t;
-				}
-
-				if(wait_time_ms >= wait_time_max_ms)
-					throw ItemNotFoundException("MutexedQueue: queue is empty");
-			}
-
-			// Wait a while before trying again
-			sleep_ms(10);
-			wait_time_ms += 10;
+			return T();
 		}
 	}
 
+	T pop_backNoEx()
+	{
+		m_size.Wait();
+
+		JMutexAutoLock lock(m_mutex);
+
+		typename std::list<T>::iterator last = m_list.end();
+		last--;
+		T t = *last;
+		m_list.erase(last);
+		return t;
+	}
+
+protected:
 	JMutex & getMutex()
 	{
 		return m_mutex;
 	}
 
+	// NEVER EVER modify the >>list<< you got by using this function!
+	// You may only modify it's content
 	std::list<T> & getList()
 	{
 		return m_list;
 	}
 
-protected:
 	JMutex m_mutex;
 	std::list<T> m_list;
+	JSemaphore m_size;
 };
 
 #endif
diff --git a/src/util/thread.h b/src/util/thread.h
index bb8e03317..8b3c33621 100644
--- a/src/util/thread.h
+++ b/src/util/thread.h
@@ -24,6 +24,7 @@ with this program; if not, write to the Free Software Foundation, Inc.,
 #include "../jthread/jthread.h"
 #include "../jthread/jmutex.h"
 #include "../jthread/jmutexautolock.h"
+#include "porting.h"
 
 template<typename T>
 class MutexedVariable
@@ -123,36 +124,38 @@ public:
 	void add(Key key, Caller caller, CallerData callerdata,
 			ResultQueue<Key, T, Caller, CallerData> *dest)
 	{
-		JMutexAutoLock lock(m_queue.getMutex());
-		
-		/*
-			If the caller is already on the list, only update CallerData
-		*/
-		for(typename std::list< GetRequest<Key, T, Caller, CallerData> >::iterator
-				i = m_queue.getList().begin();
-				i != m_queue.getList().end(); ++i)
 		{
-			GetRequest<Key, T, Caller, CallerData> &request = *i;
-
-			if(request.key == key)
+			JMutexAutoLock lock(m_queue.getMutex());
+
+			/*
+				If the caller is already on the list, only update CallerData
+			*/
+			for(typename std::list< GetRequest<Key, T, Caller, CallerData> >::iterator
+					i = m_queue.getList().begin();
+					i != m_queue.getList().end(); ++i)
 			{
-				for(typename std::list< CallerInfo<Caller, CallerData, Key, T> >::iterator
-						i = request.callers.begin();
-						i != request.callers.end(); ++i)
+				GetRequest<Key, T, Caller, CallerData> &request = *i;
+
+				if(request.key == key)
 				{
-					CallerInfo<Caller, CallerData, Key, T> &ca = *i;
-					if(ca.caller == caller)
+					for(typename std::list< CallerInfo<Caller, CallerData, Key, T> >::iterator
+							i = request.callers.begin();
+							i != request.callers.end(); ++i)
 					{
-						ca.data = callerdata;
-						return;
+						CallerInfo<Caller, CallerData, Key, T> &ca = *i;
+						if(ca.caller == caller)
+						{
+							ca.data = callerdata;
+							return;
+						}
 					}
+					CallerInfo<Caller, CallerData, Key, T> ca;
+					ca.caller = caller;
+					ca.data = callerdata;
+					ca.dest = dest;
+					request.callers.push_back(ca);
+					return;
 				}
-				CallerInfo<Caller, CallerData, Key, T> ca;
-				ca.caller = caller;
-				ca.data = callerdata;
-				ca.dest = dest;
-				request.callers.push_back(ca);
-				return;
 			}
 		}
 
@@ -168,12 +171,17 @@ public:
 		ca.dest = dest;
 		request.callers.push_back(ca);
 		
-		m_queue.getList().push_back(request);
+		m_queue.push_back(request);
+	}
+
+	GetRequest<Key, T, Caller, CallerData> pop(unsigned int timeout_ms)
+	{
+		return m_queue.pop_front(timeout_ms);
 	}
 
-	GetRequest<Key, T, Caller, CallerData> pop(bool wait_if_empty=false)
+	GetRequest<Key, T, Caller, CallerData> pop()
 	{
-		return m_queue.pop_front(wait_if_empty);
+		return m_queue.pop_frontNoEx();
 	}
 
 	void pushResult(GetRequest<Key, T, Caller, CallerData> req,