Implement per-thread event queues.

components/compat/compatcomponent.cpp

@@ -154,7 +154,7 @@ void CompatComponent::CommandPipeThread(const String& commandPath)
 				line[strlen(line) - 1] = '\0';
 
 			String command = line;
-			Event::Post(boost::bind(&CompatComponent::ProcessCommand, this, command));
+			Application::GetEQ().Post(boost::bind(&CompatComponent::ProcessCommand, this, command));
 		}
 
 		fclose(fp);

lib/base/Makefile.am

@@ -20,8 +20,8 @@ libbase_la_SOURCES =  \
 	dynamicobject.h \
 	dynamictype.cpp \
 	dynamictype.h \
-	event.cpp \
-	event.h \
+	eventqueue.cpp \
+	eventqueue.h \
 	exception.cpp \
 	exception.h \
 	fifo.cpp \

lib/base/application.cpp

@@ -32,6 +32,7 @@ String Application::m_PkgLibDir;
 String Application::m_PkgDataDir;
 int Application::m_ArgC;
 char **Application::m_ArgV;
+EventQueue Application::m_EQ;
 
 /**
  * Constructor for the Application class.
@@ -123,7 +124,7 @@ bool Application::ProcessEvents(void)
 	if (m_ShuttingDown)
 		return false;
 
-	Event::ProcessEvents(boost::posix_time::milliseconds(sleep * 1000));
+	GetEQ().ProcessEvents(boost::posix_time::milliseconds(sleep * 1000));
 
 	DynamicObject::FlushTx();
 
@@ -138,6 +139,8 @@ void Application::RunEventLoop(void) const
 {
 	boost::mutex::scoped_lock lock(m_Mutex);
 
+	GetEQ().SetOwner(boost::this_thread::get_id());
+
 #ifdef _DEBUG
 	double nextProfile = 0;
 #endif /* _DEBUG */
@@ -189,7 +192,7 @@ void Application::TimeWatchThreadProc(void)
 			 * causes the event loop to wake up thereby
 			 * solving the problem that timed_wait()
 			 * uses an absolute timestamp for the timeout */
-			Event::Post(boost::bind(&Timer::AdjustTimers,
+			GetEQ().Post(boost::bind(&Timer::AdjustTimers,
 			    -timeDiff));
 		}
 
@@ -600,3 +603,13 @@ boost::mutex& Application::GetMutex(void)
 {
 	return m_Mutex;
 }
+
+/**
+ * Returns the main thread's event queue.
+ *
+ * @returns The event queue.
+ */
+EventQueue& Application::GetEQ(void)
+{
+	return m_EQ;
+}

lib/base/application.h

@@ -86,6 +86,8 @@ public:
 
 	static boost::mutex& GetMutex(void);
 
+	static EventQueue& GetEQ(void);
+
 protected:
 	void RunEventLoop(void) const;
 
@@ -104,6 +106,7 @@ private:
 	static String m_LocalStateDir; /**< The local state dir. */
 	static String m_PkgLibDir; /**< The package lib dir. */
 	static String m_PkgDataDir; /**< The package data dir. */
+	static EventQueue m_EQ; /**< The main thread's event queue. */
 
 #ifndef _WIN32
 	static void SigIntHandler(int signum);

lib/base/eventqueue.cpp

@@ -21,53 +21,53 @@
 
 using namespace icinga;
 
-vector<Event> Event::m_Events;
-condition_variable Event::m_EventAvailable;
-boost::mutex Event::m_Mutex;
-
-/**
- * Constructor for the Event class
- *
- * @param callback The callback function for the new event object.
- */
-Event::Event(const Event::Callback& callback)
-	: m_Callback(callback)
+EventQueue::EventQueue(void)
+	: m_Stopped(false)
 { }
 
+boost::thread::id EventQueue::GetOwner(void) const
+{
+	return m_Owner;
+}
+
+void EventQueue::SetOwner(boost::thread::id owner)
+{
+	m_Owner = owner;
+}
+
+void EventQueue::Stop(void)
+{
+	boost::mutex::scoped_lock lock(m_Mutex);
+	m_Stopped = true;
+	m_EventAvailable.notify_all();
+}
+
 /**
  * Waits for events using the specified timeout value and processes
  * them.
  *
  * @param timeout The wait timeout.
+ * @returns false if the queue has been stopped, true otherwise.
  */
-void Event::ProcessEvents(millisec timeout)
+bool EventQueue::ProcessEvents(millisec timeout)
 {
-	vector<Event> events;
-
-	assert(Application::IsMainThread());
-
-	Application::GetMutex().unlock();
+	vector<Callback> events;
 
 	{
 		boost::mutex::scoped_lock lock(m_Mutex);
 
-		while (m_Events.empty()) {
-			if (!m_EventAvailable.timed_wait(lock, timeout)) {
-				Application::GetMutex().lock();
-
-				return;
-			}
+		while (m_Events.empty() && !m_Stopped) {
+			if (!m_EventAvailable.timed_wait(lock, timeout))
+				return !m_Stopped;
 		}
 
 		events.swap(m_Events);
 	}
 
-	Application::GetMutex().lock();
-
-	BOOST_FOREACH(const Event& ev, events) {
+	BOOST_FOREACH(const Callback& ev, events) {
 		double st = Utility::GetTime();
 
-		ev.m_Callback();
+		ev();
 
 		double et = Utility::GetTime();
 
@@ -77,6 +77,8 @@ void Event::ProcessEvents(millisec timeout)
 			Logger::Write(LogWarning, "base", msgbuf.str());
 		}
 	}
+
+	return !m_Stopped;
 }
 
 /**
@@ -85,18 +87,16 @@ void Event::ProcessEvents(millisec timeout)
  *
  * @param callback The callback function for the event.
  */
-void Event::Post(const Event::Callback& callback)
+void EventQueue::Post(const EventQueue::Callback& callback)
 {
-	if (Application::IsMainThread()) {
+	if (boost::this_thread::get_id() == m_Owner) {
 		callback();
 		return;
 	}
 
-	Event ev(callback);
-
 	{
 		boost::mutex::scoped_lock lock(m_Mutex);
-		m_Events.push_back(ev);
+		m_Events.push_back(callback);
 		m_EventAvailable.notify_all();
 	}
 }

lib/base/eventqueue.h

@@ -17,35 +17,41 @@
  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.             *
  ******************************************************************************/
 
-#ifndef EVENT_H
-#define EVENT_H
+#ifndef EVENTQUEUE_H
+#define EVENTQUEUE_H
 
 namespace icinga
 {
 
 /**
- * A thread-safe event that can be posted to the main thread's event queue.
+ * An event queue.
  *
  * @ingroup base
  */
-class I2_BASE_API Event
+class I2_BASE_API EventQueue
 {
 public:
 	typedef function<void ()> Callback;
 
-	static void ProcessEvents(millisec timeout);
-	static void Post(const Callback& callback);
+	EventQueue(void);
+
+	bool ProcessEvents(millisec timeout = boost::posix_time::milliseconds(30000));
+	void Post(const Callback& callback);
+
+	void Stop(void);
+
+	boost::thread::id GetOwner(void) const;
+	void SetOwner(boost::thread::id owner);
 
 private:
-	Event(const Callback& callback);
+	boost::thread::id m_Owner;
 
-	Callback m_Callback;
-
-	static vector<Event> m_Events;
-	static condition_variable m_EventAvailable;
-	static boost::mutex m_Mutex;
+	boost::mutex m_Mutex;
+	bool m_Stopped;
+	vector<Callback> m_Events;
+	condition_variable m_EventAvailable;
 };
 
 }
 
-#endif /* EVENT_H */
+#endif /* EVENTQUEUE_H */

lib/base/i2-base.h

@@ -180,7 +180,7 @@ namespace tuples = boost::tuples;
 #include "utility.h"
 #include "object.h"
 #include "exception.h"
-#include "event.h"
+#include "eventqueue.h"
 #include "value.h"
 #include "convert.h"
 #include "dictionary.h"

lib/base/logger.cpp

@@ -81,7 +81,7 @@ void Logger::Write(LogSeverity severity, const String& facility,
 	entry.Facility = facility;
 	entry.Message = message;
 
-	Event::Post(boost::bind(&Logger::ForwardLogEntry, entry));
+	Application::GetEQ().Post(boost::bind(&Logger::ForwardLogEntry, entry));
 }
 
 /**

lib/base/process-unix.cpp

@@ -133,7 +133,7 @@ void Process::WorkerThreadProc(int taskFd)
 					if (fd >= 0)
 						tasks[fd] = task;
 				} catch (...) {
-					Event::Post(boost::bind(&Process::FinishException, task, boost::current_exception()));
+					Application::GetEQ().Post(boost::bind(&Process::FinishException, task, boost::current_exception()));
 				}
 			}
 
@@ -148,7 +148,7 @@ void Process::WorkerThreadProc(int taskFd)
 				prev = it;
 				tasks.erase(prev);
 
-				Event::Post(boost::bind(&Process::FinishResult, task, task->m_Result));
+				Application::GetEQ().Post(boost::bind(&Process::FinishResult, task, task->m_Result));
 			}
 		}
 	}

lib/base/scriptinterpreter.cpp

@@ -42,15 +42,7 @@ void ScriptInterpreter::Stop(void)
 {
 	assert(Application::IsMainThread());
 
-	{
-		boost::mutex::scoped_lock lock(m_Mutex);
-
-		if (m_Shutdown)
-			return;
-
-		m_Shutdown = true;
-		m_CallAvailable.notify_all();
-	}
+	m_EQ.Stop();
 
 	BOOST_FOREACH(const String& function, m_SubscribedFunctions) {
 		ScriptFunction::Unregister(function);
@@ -61,35 +53,17 @@ void ScriptInterpreter::Stop(void)
 
 void ScriptInterpreter::ThreadWorkerProc(void)
 {
-	boost::mutex::scoped_lock lock(m_Mutex);
+	m_EQ.SetOwner(boost::this_thread::get_id());
 
-	for (;;) {
-		while (m_Calls.empty() && !m_Shutdown)
-			m_CallAvailable.wait(lock);
-
-		if (m_Shutdown)
-			break;
-
-		ScriptCall call = m_Calls.front();
-		m_Calls.pop_front();
-
-		ProcessCall(call.Task, call.Function, call.Arguments);
-	}
+	while (m_EQ.ProcessEvents())
+		; /* empty loop */
 }
 
 void ScriptInterpreter::ScriptFunctionThunk(const ScriptTask::Ptr& task,
     const String& function, const vector<Value>& arguments)
 {
-	ScriptCall call;
-	call.Task = task;
-	call.Function = function;
-	call.Arguments = arguments;
-
-	{
-		boost::mutex::scoped_lock lock(m_Mutex);
-		m_Calls.push_back(call);
-		m_CallAvailable.notify_all();
-	}
+	m_EQ.Post(boost::bind(&ScriptInterpreter::ProcessCall, this,
+	    task, function, arguments));
 }
 
 void ScriptInterpreter::SubscribeFunction(const String& name)

lib/base/scriptinterpreter.h

@@ -23,13 +23,6 @@
 namespace icinga
 {
 
-struct ScriptCall
-{
-	ScriptTask::Ptr Task;
-	String Function;
-	vector<Value> Arguments;
-};
-
 /**
  * A script interpreter.
  *
@@ -56,13 +49,8 @@ protected:
 	void UnsubscribeFunction(const String& name);
 
 private:
-	boost::mutex m_Mutex;
-	bool m_Shutdown;
-	deque<ScriptCall> m_Calls;
-	condition_variable m_CallAvailable;
-
-	set<String> m_SubscribedFunctions; /* Not protected by the mutex. */
-
+	EventQueue m_EQ;
+	set<String> m_SubscribedFunctions;
 	boost::thread m_Thread;
 
 	void ThreadWorkerProc(void);

lib/base/socket.cpp

@@ -532,7 +532,7 @@ void Socket::HandleReadableClient(void)
 	}
 
 	if (new_data)
-		Event::Post(boost::bind(boost::ref(OnDataAvailable), GetSelf()));
+		Application::GetEQ().Post(boost::bind(boost::ref(OnDataAvailable), GetSelf()));
 }
 
 void Socket::HandleWritableServer(void)
@@ -557,7 +557,7 @@ void Socket::HandleReadableServer(void)
 
 	TcpSocket::Ptr client = boost::make_shared<TcpSocket>();
 	client->SetFD(fd);
-	Event::Post(boost::bind(boost::ref(OnNewClient), GetSelf(), client));
+	Application::GetEQ().Post(boost::bind(boost::ref(OnNewClient), GetSelf(), client));
 }
 
 /**

lib/base/stream.cpp

@@ -40,9 +40,9 @@ void Stream::SetConnected(bool connected)
 	m_Connected = connected;
 
 	if (m_Connected)
-		Event::Post(boost::bind(boost::ref(OnConnected), GetSelf()));
+		Application::GetEQ().Post(boost::bind(boost::ref(OnConnected), GetSelf()));
 	else
-		Event::Post(boost::bind(boost::ref(OnClosed), GetSelf()));
+		Application::GetEQ().Post(boost::bind(boost::ref(OnClosed), GetSelf()));
 }
 
 /**