Multithreaded DBus using glib quirkyness

[Richard Reich]

--=-My7Y0U+vjg6b3ZZAWx6+
Content-Type: text/plain
Content-Transfer-Encoding: 7bit

Hello,

Being new to dbus, I hope I'm not making some mistakes here... 

I'm experiencing some "weirdness" when I try to use threads with dbus. 
I'm using the glib extensions in my glibmm application.  I ran across
one bug in the header that mitr on irc.freenode.net/#dbus was able to
help me with, basically needing G_BEGIN_DECLS/G_END_DECLS in the header
dbus-glib.h.

After getting past this hurdle I've begun testing if dbus can accept the
load I'm going to be putting on it.  Max of 7500 msgs/sec, one server
serving up those messages and any number of clients, currently 10.

So I coded up a quick c++ glibmm app that created a few threads that
listened for the broadcast messages sent by dbus-send.

If I have 1 thread or just the main thread listening all is well.  If I
have more I get random "Illegal Operations" or segfaults.  I tried using
dbus_gthread_init(), but this caused my app to not recieve anything from
the dbus system.

So my question is, am I doing anything wrong?

Attached you should find the c++ file used to demonstrate this.

To send a message I use, from the web site, 
dbus-send --dest='org.freedesktop.ExampleService'     \
            org.freedesktop.ExampleMessage            \
            int32:47 string:'hello world' double:65.32

Though for some reason dbus-monitor nor my app get any of the data.


I'm using gcc 3.2.2

to compile the file I use this line.
g++ main.cpp `pkg-config dbus-glib-1 dbus-1 gthread-2.0 glibmm-2.0
--cflags --libs`


Thanks
Richard Reich

--=-My7Y0U+vjg6b3ZZAWx6+
Content-Disposition: attachment; filename=main.cpp
Content-Type: text/x-c++; name=main.cpp; charset=ANSI_X3.4-1968
Content-Transfer-Encoding: 7bit

#include <unistd.h>
#include <sys/time.h>

#include <iostream>

#include <glibmm.h>

#define DBUS_API_SUBJECT_TO_CHANGE 1

#include <dbus/dbus.h>
#include <dbus/dbus-glib.h>


using namespace std;
using namespace SigC;
using namespace Glib;

// copied from dbus-monitor.c
static DBusHandlerResult
handler_func (DBusMessageHandler *handler,
 	      DBusConnection     *connection,
	      DBusMessage        *message,
	      void               *user_data)
{
	printf("Hello World\n");
	cout << "Got it! PID: " << getpid() << endl;
//*
  DBusMessageIter iter;

  printf ("message name=%s; sender=%s\n", dbus_message_get_name (message),
          dbus_message_get_sender (message));

  dbus_message_iter_init (message, &iter);

  do
    {
      int type = dbus_message_iter_get_arg_type (&iter);
      char *str;
      dbus_uint32_t uint32;
      dbus_int32_t int32;
      double d;
      unsigned char byte;

      if (type == DBUS_TYPE_INVALID)
	break;

      switch (type)
	{
	case DBUS_TYPE_STRING:
	  str = dbus_message_iter_get_string (&iter);
	  printf ("string:%s\n", str);
	  break;

	case DBUS_TYPE_INT32:
	  int32 = dbus_message_iter_get_int32 (&iter);
	  printf ("int32:%d\n", int32);
	  break;

	case DBUS_TYPE_UINT32:
	  uint32 = dbus_message_iter_get_uint32 (&iter);
	  printf ("int32:%u\n", uint32);
	  break;

	case DBUS_TYPE_DOUBLE:
	  d = dbus_message_iter_get_double (&iter);
	  printf ("double:%f\n", d);
	  break;

	case DBUS_TYPE_BYTE:
	  byte = dbus_message_iter_get_byte (&iter);
	  printf ("byte:%d\n", byte);
	  break;

	default:
	  printf ("(unknown arg type %d)\n", type);
	  break;
	}
    } while (dbus_message_iter_next (&iter));

// */
  return DBUS_HANDLER_RESULT_ALLOW_MORE_HANDLERS;
}

void my_thread ( void )
{
	cout << "New thread " << getpid() << endl;
	Glib::RefPtr<MainContext> cont = MainContext::create();
	Glib::RefPtr<MainLoop> loop = MainLoop::create(cont);

	// Copied from dbus-monitor.c
///////////////////////////////////////////////////////////////////////////////
	DBusConnection *connection;
	DBusError error;
	DBusBusType type = DBUS_BUS_SYSTEM;
	DBusMessageHandler *handler;

	dbus_error_init (&error);
	connection = dbus_bus_get (type, &error);
	if (connection == NULL)
	{
	  fprintf (stderr, "Failed to open connection to %s message bus: %s\n",
		   (type == DBUS_BUS_SYSTEM) ? "system" : "session",
			   error.message);
	  dbus_error_free (&error);
	  exit (1);
	}

	dbus_connection_setup_with_g_main (connection, cont->gobj());

	handler = dbus_message_handler_new (handler_func, NULL, NULL);
	dbus_connection_add_filter (connection, handler);
///////////////////////////////////////////////////////////////////////////////

	loop->run();
	cout << "Ending" << endl;
}

int main(int argc, char *argv[])
{
	Glib::thread_init();
	// Comment this out and messages begin to be recieved.
	dbus_gthread_init();

	Glib::Dispatcher sync;

	Glib::RefPtr<MainLoop> main = MainLoop::create();

	// simulate 10 clients listening
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);
	Glib::Thread::create(slot(my_thread), false);

///////////////////////////////////////////////////////////////////////////////
/*
	DBusConnection *connection;
	DBusError error;
	DBusBusType type = DBUS_BUS_SYSTEM;
	DBusMessageHandler *handler;

	dbus_error_init (&error);
	connection = dbus_bus_get (type, &error);
	if (connection == NULL)
	{
	  fprintf (stderr, "Failed to open connection to %s message bus: %s\n",
		   (type == DBUS_BUS_SYSTEM) ? "system" : "session",
			   error.message);
	  dbus_error_free (&error);
	  exit (1);
	}

	dbus_connection_setup_with_g_main (connection, NULL);

	handler = dbus_message_handler_new (handler_func, NULL, NULL);
	dbus_connection_add_filter (connection, handler);
// */
///////////////////////////////////////////////////////////////////////////////

	main->run();

	return 0;
}


--=-My7Y0U+vjg6b3ZZAWx6+--