我在线程中调用dbus方法时遇到了segfaults。这是我的场景:我有一个程序Service1,它公开了一个方法测试。第二个程序Service2公开方法公开。由于这个方法进行了一些严肃的数值计算,所以我将一些参数从expose传递给一个正在运行的线程读取器。反过来,这个线程在Service1结束其工作时调用Service1的方法测试。我在上一次dbus调用中得到了segfaults。在
代码:
# Service1.py
class Service1(Object):
def __init__(self, bus):
name = BusName('com.example.Service1', bus)
path = '/'
super(Service1, self).__init__(name, path)
@method(dbus_interface='com.example.Service1',
in_signature='s', out_signature='s')
def test(self, test):
print 'test being called'
return test
dbus_loop = DBusGMainLoop()
dsession = SessionBus(mainloop=dbus_loop)
loop = gobject.MainLoop()
gobject.threads_init()
im = Service1(dsession)
loop.run()
# Service2.py
dbus_loop = DBusGMainLoop()
dsession = SessionBus(mainloop=dbus_loop)
class Service2(Object):
def __init__(self, bus):
name = BusName('com.example.Service2', bus)
super(Service2, self).__init__(name, '/')
self.queue = Queue()
self.db = bus.get_object('com.example.Service1', '/')
self.dbi = dbus.Interface(self.db, dbus_interface='com.example.Service1')
@method(dbus_interface='com.example.Service2',
in_signature='', out_signature='')
def expose(self):
print 'calling expose'
self.queue.put(('params',))
def reader(self):
while True:
val = self.queue.get()
dd = self.dbi.test('test')
print dd
self.queue.task_done()
gobject.threads_init()
loop = gobject.MainLoop()
im = Service2(dsession)
reader = threading.Thread(target=im.reader)
reader.start()
loop.run()
要进行测试,请运行Service1.py、Service2.py和更高版本以下代码段:
^{pr2}$Service2.py在运行此代码几次后应该崩溃。但为什么呢?在
gobject.threads_init()
还不够,需要调用dbus.mainloop.glib.threads_init()
使dbus glib线程安全。在在
Service1.py
中,尝试在将dbus_loop
分配给DBusGMainLoop()
之前调用gobject.threads_init()
。在相关问题 更多 >
编程相关推荐