From a047a353438d1e9449f94850b29d79bb2dd8f9b1 Mon Sep 17 00:00:00 2001 From: Eric Windisch Date: Wed, 19 Jun 2013 14:35:11 -0400 Subject: Make ZeroMQ based RPC consumer threads more robust Consumers could potentially raise an uncaught exception and exit prematurely. Related to bug 1189711 Change-Id: I9cb32d8530f3a1dd623e193476e02b990b40b466 --- openstack/common/rpc/impl_zmq.py | 1 + tests/unit/rpc/test_zmq.py | 21 +++++++++++++++++++++ 2 files changed, 22 insertions(+) diff --git a/openstack/common/rpc/impl_zmq.py b/openstack/common/rpc/impl_zmq.py index 33d5cb6..a5ddccf 100644 --- a/openstack/common/rpc/impl_zmq.py +++ b/openstack/common/rpc/impl_zmq.py @@ -383,6 +383,7 @@ class ZmqBaseReactor(ConsumerBase): LOG.info(_("In reactor registered")) def consume_in_thread(self): + @excutils.forever_retry_uncaught_exceptions def _consume(sock): LOG.info(_("Consuming socket")) while True: diff --git a/tests/unit/rpc/test_zmq.py b/tests/unit/rpc/test_zmq.py index 5f6e9b0..55e3eb1 100644 --- a/tests/unit/rpc/test_zmq.py +++ b/tests/unit/rpc/test_zmq.py @@ -114,6 +114,27 @@ class _RpcZmqBaseTestCase(common.BaseRpcTestCase): finally: self.topic_nested = tmp_topic + def test_service_consume_thread_unexpected_exceptions(self): + def my_TopicConsumer_consume(myself, *args, **kwargs): + self.consume_calls += 1 + # see if it can sustain three failures + if self.consume_calls < 3: + raise Exception('unexpected exception') + else: + self.orig_ZmqReactor_consume(myself, *args, **kwargs) + + self.consume_calls = 0 + self.orig_ZmqReactor_consume = impl_zmq.ZmqReactor.consume + self.stubs.Set(impl_zmq.ZmqReactor, 'consume', + my_TopicConsumer_consume) + + value = 42 + result = self.rpc.call(FLAGS, self.context, self.topic, + {"method": "echo", + "args": {"value": value}}) + self.assertEqual(value, result) + self.stubs.UnsetAll() + class RpcZmqBaseTopicTestCase(_RpcZmqBaseTestCase): """Base topic RPC ZMQ test case. -- cgit