* Serialized multithreaded access to shared data (experimental)

* Refactored event loop code


git-svn-id: http://dev.openwengo.org/svn/openwengo/wengophone-ng/branches/wengophone-dbus-api/libs/dbus@12397 30a43799-04e7-0310-8b2b-ea0d24f86d0e

examples/echo/echo-client.cpp

@@ -22,29 +22,31 @@ void EchoClient::Echoed( const DBus::Variant& value )
 	cout << "!";
 }
 
-/* NOTE: using many threads is likely to trigger a race condition in the event loop
+/*
- * in fact access to the list of timeouts and watches inside the default dispatcher
+ * For some strange reason, libdbus frequently dies with an OOM
- * implementation (DBus::BusDispatcher) is NOT serialized
- *
- * (yes, I'm well aware of how much this sucks)
  */
-static const int THREADS = 1;
+
+static const int THREADS = 16;
 
 static bool spin = true;
 
-void* greeter_thread( void* )
+void* greeter_thread( void* arg )
 {
-	DBus::Connection conn = DBus::Connection::SessionBus();
+	DBus::Connection* conn = reinterpret_cast<DBus::Connection*>(arg);
 
-	EchoClient client(conn, ECHO_SERVER_PATH, ECHO_SERVER_NAME);
+	EchoClient client(*conn, ECHO_SERVER_PATH, ECHO_SERVER_NAME);
 
-	while(spin)
+	char idstr[16];
+
+	snprintf(idstr, sizeof(idstr), "%lu", pthread_self());
+
+	for(int i = 0; i < 100 && spin; ++i)
 	{
-		client.Hello("client");
+		cout << client.Hello(idstr) << endl;
-
-		cout << "*";
 	}
 
+	cout << idstr << " done " << endl;
+
 	return NULL;
 }
 
@@ -70,15 +72,19 @@ int main()
 
 	DBus::default_dispatcher = &dispatcher;
 
+	DBus::Connection conn = DBus::Connection::SessionBus();
+
 	pthread_t threads[THREADS];
 
 	for(int i = 0; i < THREADS; ++i)
 	{
-		pthread_create(threads+i, NULL, greeter_thread, NULL);
+		pthread_create(threads+i, NULL, greeter_thread, &conn);
 	}
 
 	dispatcher.enter();
 
+	cout << "terminating" << endl;
+
 	for(int i = 0; i < THREADS; ++i)
 	{
 		pthread_join(threads[i], NULL);

include/dbus-c++/dbus.h

@@ -43,6 +43,7 @@
 #include "util.h"
 #include "dispatcher.h"
 #include "eventloop.h"
+#include "eventloop-integration.h"
 #include "introspection.h"
 
 #endif//__DBUSXX_DBUS_H

include/dbus-c++/dispatcher.h

@@ -31,6 +31,7 @@
 
 #include "api.h"
 #include "connection.h"
+#include "eventloop.h"
 
 namespace DBus {
 
@@ -117,6 +118,7 @@ public:
 
 private:
 
+	DefaultMutex _mutex_p;
 	Connection::PrivatePList _pending_queue;
 };
 

include/dbus-c++/eventloop-integration.h

@@ -0,0 +1,98 @@
+/*
+ *
+ *  D-Bus++ - C++ bindings for D-Bus
+ *
+ *  Copyright (C) 2005-2007  Paolo Durante <shackan@gmail.com>
+ *
+ *
+ *  This library is free software; you can redistribute it and/or
+ *  modify it under the terms of the GNU Lesser General Public
+ *  License as published by the Free Software Foundation; either
+ *  version 2.1 of the License, or (at your option) any later version.
+ *
+ *  This library is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ *  Lesser General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Lesser General Public
+ *  License along with this library; if not, write to the Free Software
+ *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ *
+ */
+
+
+#ifndef __DBUSXX_EVENTLOOP_INTEGRATION_H
+#define __DBUSXX_EVENTLOOP_INTEGRATION_H
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include "api.h"
+#include "dispatcher.h"
+#include "util.h"
+#include "eventloop.h"
+
+namespace DBus {
+
+/* 
+ * Glue between the event loop and the DBus library
+ */
+
+class BusDispatcher;
+
+class DXXAPI BusTimeout : public Timeout, public DefaultTimeout
+{
+	BusTimeout( Timeout::Internal*, BusDispatcher* );
+
+	void toggle();
+
+friend class BusDispatcher;
+};
+
+class DXXAPI BusWatch : public Watch, public DefaultWatch
+{
+	BusWatch( Watch::Internal*, BusDispatcher* );
+
+	void toggle();
+
+friend class BusDispatcher;
+};
+
+class DXXAPI BusDispatcher : public Dispatcher, public DefaultMainLoop
+{
+public:
+
+	BusDispatcher() : _running(false)
+	{}
+
+	~BusDispatcher()
+	{}
+
+	virtual void enter();
+
+	virtual void leave();
+
+	virtual void do_iteration();
+
+	virtual Timeout* add_timeout( Timeout::Internal* );
+
+	virtual void rem_timeout( Timeout* );
+
+	virtual Watch* add_watch( Watch::Internal* );
+
+	virtual void rem_watch( Watch* );
+
+	void watch_ready( DefaultWatch& );
+
+	void timeout_expired( DefaultTimeout& );
+
+private:
+
+	bool _running;
+};
+
+} /* namespace DBus */
+
+#endif//__DBUSXX_EVENTLOOP_INTEGRATION_H

include/dbus-c++/eventloop.h

@@ -29,23 +29,33 @@
 #include "config.h"
 #endif
 
+#ifdef HAVE_PTHREAD
+#include <pthread.h>
+#endif
+
 #include <list>
 
 #include "api.h"
-#include "dispatcher.h"
 #include "util.h"
 
 namespace DBus {
 
-class EepleMainLoop;
+/*
+ * these Default* classes implement a very simple event loop which
+ * is used here as the default main loop, if you want to hook
+ * a different one use the Bus* classes in eventloop-integration.h
+ * or the Glib::Bus* classes as a reference
+ */
 
-class DXXAPI EepleTimeout
+class DefaultMainLoop;
+
+class DXXAPI DefaultTimeout
 {
 public:
 
-	EepleTimeout( int interval, bool repeat, EepleMainLoop* );
+	DefaultTimeout( int interval, bool repeat, DefaultMainLoop* );
 
-	virtual ~EepleTimeout();
+	virtual ~DefaultTimeout();
 
 	bool enabled(){ return _enabled; }
 	void enabled(bool e){ _enabled = e; }
@@ -59,7 +69,7 @@ public:
 	void* data(){ return _data; }
 	void data(void* d){ _data = d; }
 
-	Slot<void, EepleTimeout&> expired;
+	Slot<void, DefaultTimeout&> expired;
 	
 private:
 
@@ -72,20 +82,20 @@ private:
 
 	void* _data;
 	
-	EepleMainLoop* _disp;
+	DefaultMainLoop* _disp;
 
-friend class EepleMainLoop;
+friend class DefaultMainLoop;
 };
 
-typedef std::list< EepleTimeout* > Timeouts;
+typedef std::list< DefaultTimeout* > DefaultTimeouts;
 
-class DXXAPI EepleWatch
+class DXXAPI DefaultWatch
 {
 public:
 
-	EepleWatch( int fd, int flags, EepleMainLoop* );
+	DefaultWatch( int fd, int flags, DefaultMainLoop* );
 
-	virtual ~EepleWatch();
+	virtual ~DefaultWatch();
 
 	bool enabled(){ return _enabled; }
 	void enabled(bool e){ _enabled = e; }
@@ -100,7 +110,7 @@ public:
 	void* data(){ return _data; }
 	void data(void* d){ _data = d; }
 
-	Slot<void, EepleWatch&> ready;
+	Slot<void, DefaultWatch&> ready;
 
 private:
 
@@ -112,89 +122,58 @@ private:
 
 	void* _data;
 
-	EepleMainLoop* _disp;
+	DefaultMainLoop* _disp;
 
-friend class EepleMainLoop;
+friend class DefaultMainLoop;
 };
 
-typedef std::list< EepleWatch* > Watches;
+typedef std::list< DefaultWatch* > DefaultWatches;
 
-class DXXAPI EepleMainLoop
+class DXXAPI DefaultMutex
 {
 public:
 
-	EepleMainLoop();
+	DefaultMutex();
 
-	virtual ~EepleMainLoop();
+	~DefaultMutex();
+
+	void lock();
+
+	void unlock();
+
+private:
+
+#if defined HAVE_PTHREAD
+
+	pthread_mutex _mutex;
+
+#elif defined HAVE_WIN32
+
+//TODO: use a critical section
+
+#endif
+};
+
+class DXXAPI DefaultMainLoop
+{
+public:
+
+	DefaultMainLoop();
+
+	virtual ~DefaultMainLoop();
 
 	virtual void dispatch();
 
 private:
 
-	Timeouts _timeouts;
+	DefaultMutex _mutex_t;
-	Watches _watches;
+	DefaultTimeouts _timeouts;
 
-friend class EepleTimeout;
+	DefaultMutex _mutex_w;
-friend class EepleWatch;
+	DefaultWatches _watches;
-};
 
-/* the classes below are those you are going to implement if you
+friend class DefaultTimeout;
- * want to use another event loop (Qt, Glib, boost, whatever).
+friend class DefaultWatch;
- *
- * Don't forget to set 'default_dispatcher' accordingly!
- */
-
-class BusDispatcher;
-
-class DXXAPI BusTimeout : public Timeout, public EepleTimeout
-{
-	BusTimeout( Timeout::Internal*, BusDispatcher* );
-
-	void toggle();
-
-friend class BusDispatcher;
-};
-
-class DXXAPI BusWatch : public Watch, public EepleWatch
-{
-	BusWatch( Watch::Internal*, BusDispatcher* );
-
-	void toggle();
-
-friend class BusDispatcher;
-};
-
-class DXXAPI BusDispatcher : public Dispatcher, public EepleMainLoop
-{
-public:
-
-	BusDispatcher() : _running(false)
-	{}
-
-	~BusDispatcher()
-	{}
-
-	virtual void enter();
-
-	virtual void leave();
-
-	virtual void do_iteration();
-
-	virtual Timeout* add_timeout( Timeout::Internal* );
-
-	virtual void rem_timeout( Timeout* );
-
-	virtual Watch* add_watch( Watch::Internal* );
-
-	virtual void rem_watch( Watch* );
-
-	void watch_ready( EepleWatch& );
-
-	void timeout_expired( EepleTimeout& );
-
-private:
-
-	bool _running;
 };
 
 } /* namespace DBus */

src/Makefile.am

@@ -20,7 +20,6 @@ HEADER_FILES = \
 	$(HEADER_DIR)/interface.h \
 	$(HEADER_DIR)/message.h \
 	$(HEADER_DIR)/dispatcher.h \
-	$(HEADER_DIR)/eventloop.h \
 	$(HEADER_DIR)/object.h \
 	$(HEADER_DIR)/pendingcall.h \
 	$(HEADER_DIR)/server.h \
@@ -29,14 +28,16 @@ HEADER_FILES = \
 	$(HEADER_DIR)/refptr_impl.h \
 	$(HEADER_DIR)/introspection.h \
 	$(HEADER_DIR)/api.h \
+	$(HEADER_DIR)/eventloop.h \
+	$(HEADER_DIR)/eventloop-integration.h \
 	$(GLIB_H)
 
 lib_includedir=$(includedir)/dbus-c++-1/dbus-c++/
 lib_include_HEADERS = $(HEADER_FILES)
 
 lib_LTLIBRARIES = libdbus-c++-1.la
-libdbus_c___1_la_SOURCES = $(HEADER_FILES) interface.cpp object.cpp introspection.cpp debug.cpp eventloop.cpp types.cpp connection.cpp connection_p.h property.cpp dispatcher.cpp dispatcher_p.h pendingcall.cpp pendingcall_p.h error.cpp internalerror.h message.cpp message_p.h server.cpp server_p.h $(GLIB_CPP)
+libdbus_c___1_la_SOURCES = $(HEADER_FILES) interface.cpp object.cpp introspection.cpp debug.cpp types.cpp connection.cpp connection_p.h property.cpp dispatcher.cpp dispatcher_p.h pendingcall.cpp pendingcall_p.h error.cpp internalerror.h message.cpp message_p.h server.cpp server_p.h eventloop.cpp eventloop-integration.cpp $(GLIB_CPP)
-libdbus_c___1_la_LIBADD = $(dbus_LIBS) $(glib_LIBS)
+libdbus_c___1_la_LIBADD = $(dbus_LIBS) $(glib_LIBS) $(pthread_LIBS)
 
 MAINTAINERCLEANFILES = \
 	Makefile.in

src/debug.cpp

@@ -32,9 +32,7 @@ static void _debug_log_default(const char* format, ...)
 {
 #ifdef DEBUG
 
-	static int debug_env = -1;
+	static int debug_env = getenv("DBUSXX_VERBOSE") ? 1 : 0;
-
-	if(debug_env < 0) debug_env = getenv("DBUSXX_VERBOSE") ? 1 : 0;
 
 	if(debug_env)
 	{

src/dispatcher.cpp

@@ -143,11 +143,15 @@ void Dispatcher::Private::on_toggle_timeout( DBusTimeout* timeout, void* data )
 
 void Dispatcher::queue_connection( Connection::Private* cp )
 {
+	_mutex_p.lock();
 	_pending_queue.push_back(cp);
+	_mutex_p.unlock();
 }
 
 void Dispatcher::dispatch_pending()
 {
+	_mutex_p.lock();
+
 	while(_pending_queue.size() > 0)
 	{
 		Connection::PrivatePList::iterator i, j;
@@ -166,6 +170,7 @@ void Dispatcher::dispatch_pending()
 			i = j;
 		}
 	}
+	_mutex_p.unlock();
 }
 
 #ifdef DBUS_HAS_THREADS_INIT_DEFAULT

src/eventloop-integration.cpp

@@ -0,0 +1,163 @@
+/*
+ *
+ *  D-Bus++ - C++ bindings for D-Bus
+ *
+ *  Copyright (C) 2005-2007  Paolo Durante <shackan@gmail.com>
+ *
+ *
+ *  This library is free software; you can redistribute it and/or
+ *  modify it under the terms of the GNU Lesser General Public
+ *  License as published by the Free Software Foundation; either
+ *  version 2.1 of the License, or (at your option) any later version.
+ *
+ *  This library is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ *  Lesser General Public License for more details.
+ *
+ *  You should have received a copy of the GNU Lesser General Public
+ *  License along with this library; if not, write to the Free Software
+ *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ *
+ */
+
+
+#include <dbus-c++/eventloop-integration.h>
+#include <dbus-c++/debug.h>
+
+#include <sys/poll.h>
+
+#include <dbus/dbus.h>
+
+using namespace DBus;
+
+BusTimeout::BusTimeout( Timeout::Internal* ti, BusDispatcher* bd )
+: Timeout(ti), DefaultTimeout(Timeout::interval(), true, bd)
+{
+	DefaultTimeout::enabled(Timeout::enabled());
+}
+
+void BusTimeout::toggle()
+{
+	debug_log("timeout %p toggled (%s)", this, Timeout::enabled() ? "on":"off");
+
+	DefaultTimeout::enabled(Timeout::enabled());
+}
+
+BusWatch::BusWatch( Watch::Internal* wi, BusDispatcher* bd )
+: Watch(wi), DefaultWatch(Watch::descriptor(), 0, bd)
+{
+	int flags = POLLHUP | POLLERR;
+
+	if(Watch::flags() & DBUS_WATCH_READABLE)
+		flags |= POLLIN;
+	if(Watch::flags() & DBUS_WATCH_WRITABLE)
+		flags |= POLLOUT;
+
+	DefaultWatch::flags(flags);
+	DefaultWatch::enabled(Watch::enabled());
+}
+
+void BusWatch::toggle()
+{
+	debug_log("watch %p toggled (%s)", this, Watch::enabled() ? "on":"off");
+
+	DefaultWatch::enabled(Watch::enabled());
+}
+
+void BusDispatcher::enter()
+{
+	debug_log("entering dispatcher %p", this);
+
+	_running = true;
+
+	while(_running)
+	{
+		do_iteration();
+	}
+
+	debug_log("leaving dispatcher %p", this);
+}
+
+void BusDispatcher::leave()
+{
+	_running = false;
+}
+
+void BusDispatcher::do_iteration()
+{
+	dispatch_pending();
+	dispatch();
+}
+
+Timeout* BusDispatcher::add_timeout( Timeout::Internal* ti )
+{
+	BusTimeout* bt = new BusTimeout(ti, this);
+
+	bt->expired = new Callback<BusDispatcher, void, DefaultTimeout&>(this, &BusDispatcher::timeout_expired);
+	bt->data(bt);
+
+	debug_log("added timeout %p (%s)", bt, ((Timeout*)bt)->enabled() ? "on":"off");
+
+	return bt;
+}
+
+void BusDispatcher::rem_timeout( Timeout* t )
+{
+	debug_log("removed timeout %p", t);
+
+	delete t;
+}
+
+Watch* BusDispatcher::add_watch( Watch::Internal* wi )
+{
+	BusWatch* bw = new BusWatch(wi, this);
+
+	bw->ready = new Callback<BusDispatcher, void, DefaultWatch&>(this, &BusDispatcher::watch_ready);
+	bw->data(bw);
+
+	debug_log("added watch %p (%s) fd=%d flags=%d",
+		bw, ((Watch*)bw)->enabled() ? "on":"off", ((Watch*)bw)->descriptor(), ((Watch*)bw)->flags()
+	);
+
+	return bw;
+}
+
+void BusDispatcher::rem_watch( Watch* w )
+{
+	debug_log("removed watch %p", w);
+
+	delete w;
+}
+
+void BusDispatcher::timeout_expired( DefaultTimeout& et )
+{
+	debug_log("timeout %p expired", &et);
+
+	BusTimeout* timeout = reinterpret_cast<BusTimeout*>(et.data());
+
+	timeout->handle();
+}
+
+void BusDispatcher::watch_ready( DefaultWatch& ew )
+{
+	BusWatch* watch = reinterpret_cast<BusWatch*>(ew.data());
+
+	debug_log("watch %p ready, flags=%d state=%d",
+		watch, ((Watch*)watch)->flags(), watch->state()
+	);
+
+	int flags = 0;
+
+	if(watch->state() & POLLIN)
+		flags |= DBUS_WATCH_READABLE;
+	if(watch->state() & POLLOUT)
+		flags |= DBUS_WATCH_WRITABLE;
+	if(watch->state() & POLLHUP)
+		flags |= DBUS_WATCH_HANGUP;
+	if(watch->state() & POLLERR)
+		flags |= DBUS_WATCH_ERROR;
+
+	watch->handle(flags);
+}
+

src/eventloop.cpp

@@ -37,7 +37,7 @@ static double millis( timeval tv )
 	return (tv.tv_sec*1000.0 + tv.tv_usec/1000.0);
 }
 	
-EepleTimeout::EepleTimeout( int interval, bool repeat, EepleMainLoop* ed )
+DefaultTimeout::DefaultTimeout( int interval, bool repeat, DefaultMainLoop* ed )
 : _enabled(true), _interval(interval), _repeat(repeat), _expiration(0), _data(0), _disp(ed)
 {
 	timeval now;
@@ -45,57 +45,113 @@ EepleTimeout::EepleTimeout( int interval, bool repeat, EepleMainLoop* ed )
 
 	_expiration = millis(now) + interval;
 
+	_disp->_mutex_t.lock();
 	_disp->_timeouts.push_back(this);
+	_disp->_mutex_t.unlock();
 }
 
-EepleTimeout::~EepleTimeout()
+DefaultTimeout::~DefaultTimeout()
 {
+	_disp->_mutex_t.lock();
 	_disp->_timeouts.remove(this);
+	_disp->_mutex_t.unlock();
 }
 
-EepleWatch::EepleWatch( int fd, int flags, EepleMainLoop* ed )
+DefaultWatch::DefaultWatch( int fd, int flags, DefaultMainLoop* ed )
 : _enabled(true), _fd(fd), _flags(flags), _state(0), _data(0), _disp(ed)
 {
+	_disp->_mutex_w.lock();
 	_disp->_watches.push_back(this);
+	_disp->_mutex_w.unlock();
 }
 
-EepleWatch::~EepleWatch()
+DefaultWatch::~DefaultWatch()
 {
+	_disp->_mutex_w.lock();
 	_disp->_watches.remove(this);
+	_disp->_mutex_w.unlock();
 }
 
-EepleMainLoop::EepleMainLoop()
+DefaultMutex::DefaultMutex()
+{
+#if defined HAVE_PTHREAD
+
+	pthread_mutex_init(&_mutex, NULL);
+
+#elif defined HAVE_WIN32
+#endif
+}
+
+DefaultMutex::~DefaultMutex()
+{
+#if defined HAVE_PTHREAD
+
+	pthread_mutex_destroy(&_mutex);
+
+#elif defined HAVE_WIN32
+#endif
+}
+
+void DefaultMutex::lock()
+{
+#if defined HAVE_PTHREAD
+
+	pthread_mutex_lock(&_mutex);
+
+#elif defined HAVE_WIN32
+#endif
+}
+
+void DefaultMutex::unlock()
+{
+#if defined HAVE_PTHREAD
+
+	pthread_mutex_unlock(&_mutex);
+
+#elif defined HAVE_WIN32
+#endif
+}
+
+DefaultMainLoop::DefaultMainLoop()
 {
 }
 
-EepleMainLoop::~EepleMainLoop()
+DefaultMainLoop::~DefaultMainLoop()
 {
-	Watches::iterator wi = _watches.begin();
+	_mutex_w.lock();
+
+	DefaultWatches::iterator wi = _watches.begin();
 	while(wi != _watches.end())
 	{
-		Watches::iterator wmp = wi;
+		DefaultWatches::iterator wmp = wi;
 		++wmp;
 		delete (*wi);
 		wi = wmp;
 	}
+	_mutex_w.unlock();
 
-	Timeouts::iterator ti = _timeouts.begin();
+	_mutex_t.lock();
+
+	DefaultTimeouts::iterator ti = _timeouts.begin();
 	while(ti != _timeouts.end())
 	{
-		Timeouts::iterator tmp = ti;
+		DefaultTimeouts::iterator tmp = ti;
 		++tmp;
 		delete (*ti);
 		ti = tmp;
 	}
+	_mutex_t.unlock();
 }
 
-void EepleMainLoop::dispatch()
+void DefaultMainLoop::dispatch()
 {
+	_mutex_w.lock();
+
 	int nfd = _watches.size();
 
 	pollfd fds[nfd];
 
-	Watches::iterator wi = _watches.begin();
+	DefaultWatches::iterator wi = _watches.begin();
 
 	for(nfd = 0; wi != _watches.end(); ++wi)
 	{
@@ -108,10 +164,13 @@ void EepleMainLoop::dispatch()
 			++nfd;
 		}
 	}
+	_mutex_w.unlock();
 
 	int wait_min = 10000;
 
-	Timeouts::iterator ti;
+	DefaultTimeouts::iterator ti;
+
+	_mutex_t.lock();
 
 	for(ti = _timeouts.begin(); ti != _timeouts.end(); ++ti)
 	{
@@ -119,8 +178,7 @@ void EepleMainLoop::dispatch()
 			wait_min = (*ti)->interval();
 	}
 
-	//int rfd = poll(fds, nfd, wait_min);
+	_mutex_t.unlock();
-	//if(rfd) debug_log("%d descriptors ready");
 
 	poll(fds, nfd, wait_min);
 
@@ -129,11 +187,13 @@ void EepleMainLoop::dispatch()
 
 	double now_millis = millis(now);
 
+	_mutex_t.lock();
+
 	ti = _timeouts.begin();
 
 	while(ti != _timeouts.end())
 	{
-		Timeouts::iterator tmp = ti;
+		DefaultTimeouts::iterator tmp = ti;
 		++tmp;
 
 		if((*ti)->enabled() && now_millis >= (*ti)->_expiration)
@@ -150,13 +210,17 @@ void EepleMainLoop::dispatch()
 		ti = tmp;
 	}
 
+	_mutex_t.unlock();
+
+	_mutex_w.lock();
+
 	for(int j = 0; j < nfd; ++j)
 	{
-		Watches::iterator wi;
+		DefaultWatches::iterator wi;
 
 		for(wi = _watches.begin(); wi != _watches.end();)
 		{
-			Watches::iterator tmp = wi;
+			DefaultWatches::iterator tmp = wi;
 			++tmp;
 
 			if((*wi)->enabled() && (*wi)->_fd == fds[j].fd)
@@ -174,138 +238,6 @@ void EepleMainLoop::dispatch()
 			wi = tmp;
 		}
 	}
-}
+	_mutex_w.unlock();
-
-/*
-*/
-
-BusTimeout::BusTimeout( Timeout::Internal* ti, BusDispatcher* bd )
-: Timeout(ti), EepleTimeout(Timeout::interval(), true, bd)
-{
-	EepleTimeout::enabled(Timeout::enabled());
-}
-
-void BusTimeout::toggle()
-{
-	debug_log("timeout %p toggled (%s)", this, Timeout::enabled() ? "on":"off");
-
-	EepleTimeout::enabled(Timeout::enabled());
-}
-
-BusWatch::BusWatch( Watch::Internal* wi, BusDispatcher* bd )
-: Watch(wi), EepleWatch(Watch::descriptor(), 0, bd)
-{
-	int flags = POLLHUP | POLLERR;
-
-	if(Watch::flags() & DBUS_WATCH_READABLE)
-		flags |= POLLIN;
-	if(Watch::flags() & DBUS_WATCH_WRITABLE)
-		flags |= POLLOUT;
-
-	EepleWatch::flags(flags);
-	EepleWatch::enabled(Watch::enabled());
-}
-
-void BusWatch::toggle()
-{
-	debug_log("watch %p toggled (%s)", this, Watch::enabled() ? "on":"off");
-
-	EepleWatch::enabled(Watch::enabled());
-}
-
-void BusDispatcher::enter()
-{
-	debug_log("entering dispatcher %p", this);
-
-	_running = true;
-
-	while(_running)
-	{
-		do_iteration();
-	}
-
-	debug_log("leaving dispatcher %p", this);
-}
-
-void BusDispatcher::leave()
-{
-	_running = false;
-}
-
-void BusDispatcher::do_iteration()
-{
-	dispatch_pending();
-	dispatch();
-}
-
-Timeout* BusDispatcher::add_timeout( Timeout::Internal* ti )
-{
-	BusTimeout* bt = new BusTimeout(ti, this);
-
-	bt->expired = new Callback<BusDispatcher, void, EepleTimeout&>(this, &BusDispatcher::timeout_expired);
-	bt->data(bt);
-
-	debug_log("added timeout %p (%s)", bt, ((Timeout*)bt)->enabled() ? "on":"off");
-
-	return bt;
-}
-
-void BusDispatcher::rem_timeout( Timeout* t )
-{
-	debug_log("removed timeout %p", t);
-
-	delete t;
-}
-
-Watch* BusDispatcher::add_watch( Watch::Internal* wi )
-{
-	BusWatch* bw = new BusWatch(wi, this);
-
-	bw->ready = new Callback<BusDispatcher, void, EepleWatch&>(this, &BusDispatcher::watch_ready);
-	bw->data(bw);
-
-	debug_log("added watch %p (%s) fd=%d flags=%d",
-		bw, ((Watch*)bw)->enabled() ? "on":"off", ((Watch*)bw)->descriptor(), ((Watch*)bw)->flags()
-	);
-
-	return bw;
-}
-
-void BusDispatcher::rem_watch( Watch* w )
-{
-	debug_log("removed watch %p", w);
-
-	delete w;
-}
-
-void BusDispatcher::timeout_expired( EepleTimeout& et )
-{
-	debug_log("timeout %p expired", &et);
-
-	BusTimeout* timeout = reinterpret_cast<BusTimeout*>(et.data());
-
-	timeout->handle();
-}
-
-void BusDispatcher::watch_ready( EepleWatch& ew )
-{
-	BusWatch* watch = reinterpret_cast<BusWatch*>(ew.data());
-
-	debug_log("watch %p ready, flags=%d state=%d",
-		watch, ((Watch*)watch)->flags(), watch->state()
-	);
-
-	int flags = 0;
-
-	if(watch->state() & POLLIN)
-		flags |= DBUS_WATCH_READABLE;
-	if(watch->state() & POLLOUT)
-		flags |= DBUS_WATCH_WRITABLE;
-	if(watch->state() & POLLHUP)
-		flags |= DBUS_WATCH_HANGUP;
-	if(watch->state() & POLLERR)
-		flags |= DBUS_WATCH_ERROR;
-
-	watch->handle(flags);
 }