From 3c87a95a596b540b59de9beb396ac8b7f2ff574c Mon Sep 17 00:00:00 2001
From: Joshua Harlow <harlowja@yahoo-inc.com>
Date: Wed, 26 Nov 2014 11:40:02 -0800
Subject: [PATCH 3/5] Have the timeout decrement inside the wait() method

Currently it appears the timeout provided to the wait()
method will be reused X number of times (where X may be
indeterminate depending on reconnections, loss of messages
and so-on) so instead of reusing it which can potentially
result in a infinite number of calls have a new object be
used that will cause the subsequent timeouts used elsewhere
in the wait function to actually decay correctly.

Closes-bug: #1379394

Change-Id: I12c4ea1eef6b857d12246db0483adaf7c87e740c
(cherry picked from commit f1c7e78a56794f016ad63c28b036dff12e259953)
---
 oslo/messaging/_drivers/amqpdriver.py | 37 +++++++++++++++++++++++++++++------
 1 file changed, 31 insertions(+), 6 deletions(-)

diff --git a/oslo/messaging/_drivers/amqpdriver.py b/oslo/messaging/_drivers/amqpdriver.py
index f9f1b06..2d717b4 100644
--- a/oslo/messaging/_drivers/amqpdriver.py
+++ b/oslo/messaging/_drivers/amqpdriver.py
@@ -30,6 +30,30 @@ from oslo.messaging._drivers import common as rpc_common
 LOG = logging.getLogger(__name__)
 
 
+class _DecayingTimer(object):
+    def __init__(self, duration=None):
+        self._duration = duration
+        self._ends_at = None
+
+    def start(self):
+        if self._duration is not None:
+            self._ends_at = time.time() + max(0, self._duration)
+        return self
+
+    def check_return(self, msg_id):
+        if self._duration is None:
+            return None
+        if self._ends_at is None:
+            raise RuntimeError("Can not check/return a timeout from a timer"
+                               " that has not been started")
+        left = self._ends_at - time.time()
+        if left <= 0:
+            raise messaging.MessagingTimeout('Timed out waiting for a '
+                                             'reply to message ID %s'
+                                             % msg_id)
+        return left
+
+
 class AMQPIncomingMessage(base.IncomingMessage):
 
     def __init__(self, listener, ctxt, message, unique_id, msg_id, reply_q):
@@ -216,7 +240,7 @@ class ReplyWaiter(object):
             result = data['result']
         return result, ending
 
-    def _poll_connection(self, msg_id, timeout):
+    def _poll_connection(self, msg_id, timer):
         while True:
             while self.incoming:
                 message_data = self.incoming.pop(0)
@@ -228,14 +252,14 @@ class ReplyWaiter(object):
                 self.waiters.put(incoming_msg_id, message_data)
 
             try:
-                self.conn.consume(limit=1, timeout=timeout)
+                self.conn.consume(limit=1, timeout=timer.check_return(msg_id))
             except rpc_common.Timeout:
                 raise messaging.MessagingTimeout('Timed out waiting for a '
                                                  'reply to message ID %s'
                                                  % msg_id)
 
-    def _poll_queue(self, msg_id, timeout):
-        message = self.waiters.get(msg_id, timeout)
+    def _poll_queue(self, msg_id, timer):
+        message = self.waiters.get(msg_id, timeout=timer.check_return(msg_id))
         if message is self.waiters.WAKE_UP:
             return None, None, True  # lock was released
 
@@ -264,6 +288,7 @@ class ReplyWaiter(object):
         # have the first thread take responsibility for passing replies not
         # intended for itself to the appropriate thread.
         #
+        timer = _DecayingTimer(duration=timeout).start()
         final_reply = None
         while True:
             if self.conn_lock.acquire(False):
@@ -282,7 +307,7 @@ class ReplyWaiter(object):
 
                     # Now actually poll the connection
                     while True:
-                        reply, ending = self._poll_connection(msg_id, timeout)
+                        reply, ending = self._poll_connection(msg_id, timer)
                         if not ending:
                             final_reply = reply
                         else:
@@ -295,7 +320,7 @@ class ReplyWaiter(object):
                     self.waiters.wake_all(msg_id)
             else:
                 # We're going to wait for the first thread to pass us our reply
-                reply, ending, trylock = self._poll_queue(msg_id, timeout)
+                reply, ending, trylock = self._poll_queue(msg_id, timer)
                 if trylock:
                     # The first thread got its reply, let's try and take over
                     # the responsibility for polling
-- 
2.3.0

