Running mainloop within a thread

[Tom Vaughan]

Hi,

Please forgive me if this has been covered already. My attempts to
search the archives turned up empty.

I want to have a unittest module that starts a dbus test service, runs
some tests, and then stops the dbus test service. This means, I
assume, I have to run the mainloop within a thread. For example:

class TestDaemon(dbus.service.Object, threading.Thread):

        def run(self):
                self.__loop = gobject.MainLoop()
                self.__loop.run()

        [snip]

        def __init__(self):
                gobject.threads_init()

                dbus.mainloop.glib.threads_init()
                dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

                # these are assigned so they do not go out-of-scope
and get garbage collected
                self.__sessionbus = dbus.SessionBus()
                self.__busname =
dbus.service.BusName(resconn.namespace, self.__sessionbus)

                dbus.service.Object.__init__(self, self.__sessionbus, '/Test')

                threading.Thread.__init__(self)
                self.setDaemon(True)

class TestCase(unittest.TestCase):

        def setUp(self):
                daemon = TestDaemon()
                daemon.start()

        [snip]

This works, sort of. Everything is fine until I run a test that
transfers a large amount of data over dbus. I run the same test 10
times. The first 1 or 2 iterations are OK, and then the dbus test
services appears to die. If I don't start the dbus test service within
the unittest module, but rather in a separate script, everything is
hunky dory.

Any clues what I any doing wrong? Thanks. Full source code is below.

-Tom

import sys
import threading
import time
import unittest

import numpy

import gobject

import dbus
import dbus.service
import dbus.mainloop.glib

import resconn
import wrapper.restest as wrapper

class TestDaemon(dbus.service.Object, threading.Thread):

        def run(self):
                self.__loop = gobject.MainLoop()
                self.__loop.run()

        @resconn.benchmark
        def tolist(self, length, rawdata):
                return [rawdata[i] for i in range(0, length)]

        @resconn.benchmark
        @dbus.service.method(resconn.namespace, in_signature='d',
out_signature='d')
        def DBUSTest0(self, value):
                return value

        @resconn.benchmark
        @dbus.service.method(resconn.namespace, in_signature='d',
out_signature='d')
        def DBUSTest1(self, value):
                test1 = wrapper.cds_restest_dbus_test1(value)
                return test1

        @resconn.benchmark
        @dbus.service.method(resconn.namespace, in_signature='d',
out_signature='(uad)')
        def DBUSTest2(self, value):
                test2 = wrapper.cds_restest_dbus_test2(value)
                ndoubles = test2.ndoubles
                doubles = self.tolist(ndoubles,
wrapper.doubles.frompointer(test2.doubles))
                return [ndoubles, doubles]

        @resconn.benchmark
        @dbus.service.method(resconn.namespace, in_signature='ud',
out_signature='(ad)')
        def DBUSTest3(self, length, value):
                test3 = wrapper.cds_restest_dbus_test3(length, value)
                return test3

        @resconn.benchmark
        @dbus.service.method(resconn.namespace, in_signature='',
out_signature='')
        def Exit(self):
                self.__loop.quit()

        def __init__(self):
                gobject.threads_init()

                dbus.mainloop.glib.threads_init()
                dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

                # these are assigned so they do not go out-of-scope
and get garbage collected
                self.__sessionbus = dbus.SessionBus()
                self.__busname =
dbus.service.BusName(resconn.namespace, self.__sessionbus)

                dbus.service.Object.__init__(self, self.__sessionbus, '/Test')

                threading.Thread.__init__(self)
                self.setDaemon(True)

class TestClient(resconn.Object):

        def __new__(self):
                self.__object =
dbus.SessionBus().get_object(resconn.namespace, '/Test')
                return dbus.Interface(self.__object, resconn.namespace)

class TestCase(unittest.TestCase):

        def setUp(self):
                daemon = TestDaemon()
                daemon.run()

                try:
                        self.client = TestClient()
                except Exception, detail:
                        print detail

        def tearDown(self):
                self.client.Exit()

        @resconn.benchmark
        def testDBUS(self):

                @resconn.benchmark
                def testDBUS0():
                        value = 12.34
                        t1 = time.time()
                        rc1 = self.client.DBUSTest0(value)
                        t2 = time.time()
                        print 'D-BUS Test 0...',
                        if value != rc1:
                                print 'FAILED! value = %.04f rc =
%.04f' % (value, rc1)
                                passed = False
                        else:
                                print 'PASSED!'
                                passed = True

                        return passed, t2 - t1

                @resconn.benchmark
                def testDBUS1():
                        value = 12.34
                        t1 = time.time()
                        rc1 = self.client.DBUSTest1(value)
                        t2 = time.time()
                        print 'D-BUS Test 1...',
                        if value != rc1:
                                print 'FAILED! value = %.04f rc =
%.04f' % (value, rc1)
                                passed = False
                        else:
                                print 'PASSED!'
                                passed = True

                        return passed, t2 - t1

                @resconn.benchmark
                def testDBUS2():
                        value = 12.34
                        t1 = time.time()
                        rc1 = self.client.DBUSTest2(value)
                        t2 = time.time()
                        print 'D-BUS Test 2...',
                        print 'PASSED!'
                        passed = True

                        return passed, t2 - t1

                def test(f):
                        times = []
                        niters = 10
                        for i in range(0, niters):
                                passed, elapsed = f()
                                times.append(elapsed)

                        print '%s: niters = %d' % (f.__name__, niters)
                        print 'min = %.04f s max = %.04f s mean =
%.04f s std = %.04f s' % \
                          (min(times), max(times), numpy.mean(times),
numpy.std(times))

                test(testDBUS0)
                test(testDBUS1)
                test(testDBUS2)

def testsuite():
        __testsuite = unittest.TestSuite()
        __testsuite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestCase))
        return __testsuite

# $Id:$
#
# Local Variables:
# indent-tabs-mode: nil
# python-continuation-offset: 2
# python-indent: 8
# End:
# vim: ai et si sw=8 ts=8


-- 
Website: www.software6.net
E-mail/Google Talk: tom (at) software6 (dot) net
Mobile: +1 (310) 704-0787