Attachment 847690 Details for Bug 1050213 – reproducing failure mode

[patch] reproducing failure mode

reproducer.diff.txt (text/plain), 3.39 KB, created by Russell Bryant on 2014-01-09 15:28:04 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Russell Bryant

Created: 2014-01-09 15:28:04 UTC

Size: 3.39 KB

patch

obsolete

>diff --git a/nova/openstack/common/rpc/amqp.py b/nova/openstack/common/rpc/amqp.py
>index bde9e43..2e01b4d 100644
>--- a/nova/openstack/common/rpc/amqp.py
>+++ b/nova/openstack/common/rpc/amqp.py
>@@ -173,8 +173,8 @@ class ConnectionContext(rpc_common.Connection):
>                                            exchange_name,
>                                            ack_on_error)
> 
>-    def consume_in_thread(self):
>-        self.connection.consume_in_thread()
>+    def consume_in_thread(self, **kwargs):
>+        self.connection.consume_in_thread(**kwargs)
> 
>     def __getattr__(self, key):
>         """Proxy all other calls to the Connection instance."""
>@@ -193,7 +193,7 @@ class ReplyProxy(ConnectionContext):
>         self._reply_q = 'reply_' + uuid.uuid4().hex
>         super(ReplyProxy, self).__init__(conf, connection_pool, pooled=False)
>         self.declare_direct_consumer(self._reply_q, self._process_data)
>-        self.consume_in_thread()
>+        self.consume_in_thread(replies=True)
> 
>     def _process_data(self, message_data):
>         msg_id = message_data.pop('_msg_id', None)
>diff --git a/nova/openstack/common/rpc/impl_qpid.py b/nova/openstack/common/rpc/impl_qpid.py
>index f5af17d..f1b80ea 100644
>--- a/nova/openstack/common/rpc/impl_qpid.py
>+++ b/nova/openstack/common/rpc/impl_qpid.py
>@@ -447,6 +447,9 @@ class NotifyPublisher(Publisher):
>                                               node_opts)
> 
> 
>+hacked = False
>+
>+
> class Connection(object):
>     """Connection object."""
> 
>@@ -592,9 +595,9 @@ class Connection(object):
> 
>         return self.ensure(_connect_error, _declare_consumer)
> 
>-    def iterconsume(self, limit=None, timeout=None):
>+    def iterconsume(self, limit=None, timeout=None, replies=False):
>         """Return an iterator that will consume from all queues/consumers."""
>-
>+        start = time.time()
>         def _error_callback(exc):
>             if isinstance(exc, qpid_exceptions.Empty):
>                 LOG.debug(_('Timed out waiting for RPC response: %s') %
>@@ -605,6 +608,11 @@ class Connection(object):
>                               str(exc))
> 
>         def _consume():
>+            if not hacked and time.time() - start > 30:
>+                LOG.error('forcing failure')
>+                global hacked
>+                hacked = True
>+                raise Exception('forced failure')
>             nxt_receiver = self.session.next_receiver(timeout=timeout)
>             try:
>                 self._lookup_consumer(nxt_receiver).consume()
>@@ -692,21 +700,21 @@ class Connection(object):
>         """Send a notify message on a topic."""
>         self.publisher_send(NotifyPublisher, topic, msg)
> 
>-    def consume(self, limit=None):
>+    def consume(self, limit=None, replies=False):
>         """Consume from all queues/consumers."""
>-        it = self.iterconsume(limit=limit)
>+        it = self.iterconsume(limit=limit, replies=replies)
>         while True:
>             try:
>                 it.next()
>             except StopIteration:
>                 return
> 
>-    def consume_in_thread(self):
>+    def consume_in_thread(self, replies=False):
>         """Consumer from all queues/consumers in a greenthread."""
>         @excutils.forever_retry_uncaught_exceptions
>         def _consumer_thread():
>             try:
>-                self.consume()
>+                self.consume(replies=replies)
>             except greenlet.GreenletExit:
>                 return
>         if self.consumer_thread is None:

Actions: View | Diff

Attachments on bug 1050213: 847344 | 847690