From 9b7ced5c35ac288a2b4ddbe38fd3e208f90b3809 Mon Sep 17 00:00:00 2001 From: blkart Date: Mon, 25 Apr 2016 17:09:11 +0800 Subject: [PATCH] Don't hold the connection when reply fail This change moves the reply retry code to upper layer to be able to release the connection while we wait between two retries. In the worse scenario, a client waits for more than 30 replies and died/restart, the server tries to send this 30 replies to this th lient and can wait too 60s per replies. During this replies for other clients are just stuck. This change fixes that. Related-bug: #1477914 Closes-bug: #1521958 Related upstream commit: daddb82788918296f8b34d6cdeb40d01620fb183 Signed-off-by: blkart --- oslo/messaging/_drivers/amqpdriver.py | 58 ++++++++++++++++++++++---- oslo/messaging/_drivers/impl_rabbit.py | 41 ++++-------------- 2 files changed, 56 insertions(+), 43 deletions(-) diff --git a/oslo/messaging/_drivers/amqpdriver.py b/oslo/messaging/_drivers/amqpdriver.py index 42c1c4aaf..7c489243c 100644 --- a/oslo/messaging/_drivers/amqpdriver.py +++ b/oslo/messaging/_drivers/amqpdriver.py @@ -17,6 +17,7 @@ import logging import threading +import time import uuid import cachetools @@ -60,16 +61,19 @@ def _send_reply(self, conn, reply=None, failure=None, msg['ending'] = True rpc_amqp._add_unique_id(msg) + unique_id = msg[rpc_amqp.UNIQUE_ID] # If a reply_q exists, add the msg_id to the reply and pass the # reply_q to direct_send() to use it as the response queue. # Otherwise use the msg_id for backward compatibility. if self.reply_q: msg['_msg_id'] = self.msg_id - try: - conn.direct_send(self.reply_q, rpc_common.serialize_msg(msg)) - except rpc_amqp.AMQPDestinationNotFound: - self._obsolete_reply_queues.add(self.reply_q, self.msg_id) + LOG.debug("sending reply msg_id: %(msg_id)s " + "reply queue: %(reply_q)s" % { + 'msg_id': self.msg_id, + 'unique_id': unique_id, + 'reply_q': self.reply_q}) + conn.direct_send(self.reply_q, rpc_common.serialize_msg(msg)) else: # TODO(sileht): look at which version of oslo-incubator rpc # send need this, but I guess this is older than icehouse @@ -83,16 +87,52 @@ def reply(self, reply=None, failure=None, log_failure=True): # because reply should not be expected by caller side return - # NOTE(sileht): return without hold the a connection if possible + # NOTE(sileht): return without using a connection if possible if (self.reply_q and not self._obsolete_reply_queues.reply_q_valid(self.reply_q, self.msg_id)): return - with self.listener.driver._get_connection( - rpc_amqp.PURPOSE_SEND) as conn: - self._send_reply(conn, reply, failure, log_failure=log_failure) - self._send_reply(conn, ending=True) + duration = 60 + amqp_notfound_count = 0 + last_amqp_notfound_time = 0 + timer = rpc_common.DecayingTimer(duration=duration) + timer.start() + + while True: + try: + with self.listener.driver._get_connection( + rpc_amqp.PURPOSE_SEND) as conn: + self._send_reply(conn, reply, failure, + log_failure=log_failure) + self._send_reply(conn, ending=True) + return + except rpc_amqp.AMQPDestinationNotFound: + if timer.check_return() > 0: + current_time = time.time() + amqp_notfound_count += 1 + msg = _("The reply %(msg_id)s cannot be sent, " + "reply queue %(reply_q)s doesn't exist " + "after retried %(times)s times, keep retrying...") % { + 'msg_id': self.msg_id, + 'reply_q': self.reply_q, + 'times': amqp_notfound_count} + if amqp_notfound_count == 1 or\ + current_time - last_amqp_notfound_time >= 20: + LOG.info(msg) + last_amqp_notfound_time = current_time + time.sleep(0.25) + continue + else: + self._obsolete_reply_queues.add(self.reply_q, self.msg_id) + msg = _("The reply %(msg_id)s cannot be sent, " + "reply queue %(reply_q)s doesn't exist after " + "%(duration)s sec abandoning...") % { + 'msg_id': self.msg_id, + 'reply_q': self.reply_q, + 'duration': duration} + LOG.error(msg) + return def acknowledge(self): self.listener.msg_id_cache.add(self.unique_id) diff --git a/oslo/messaging/_drivers/impl_rabbit.py b/oslo/messaging/_drivers/impl_rabbit.py index 0073a28a9..ce4ed8273 100644 --- a/oslo/messaging/_drivers/impl_rabbit.py +++ b/oslo/messaging/_drivers/impl_rabbit.py @@ -1044,18 +1044,10 @@ def declare_fanout_consumer(self, topic, callback): def direct_send(self, msg_id, msg): """Send a 'direct' message.""" - duration = 60 - amqp_notfound_count = 0 - last_amqp_notfound_time = 0 - timer = rpc_common.DecayingTimer(duration) - timer.start() - # NOTE(sileht): retry at least 60sec, after we have a good change - # that the caller is really dead too... - - while True: - try: - self.publisher_send(DirectPublisher, msg_id, msg) - except self.connection.channel_errors as exc: + try: + self.publisher_send(DirectPublisher, msg_id, msg) + except self.connection.channel_errors as exc: + if exc.code == 404: # NOTE(noelbk/sileht): # If rabbit dies, the consumer can be disconnected before the # publisher sends, and if the consumer hasn't declared the @@ -1064,28 +1056,9 @@ def direct_send(self, msg_id, msg): # So we set passive=True to the publisher exchange and catch # the 404 kombu ChannelError and retry until the exchange # appears - if exc.code == 404 and timer.check_return() > 0: - current_time = time.time() - amqp_notfound_count += 1 - msg = _("The exchange to reply to %s doesn't " - "exist for %s times, retrying...") % ( - msg_id, amqp_notfound_count) - if amqp_notfound_count == 1 or\ - current_time - last_amqp_notfound_time >= 20: - LOG.info(msg) - last_amqp_notfound_time = current_time - time.sleep(0.25) - continue - elif exc.code == 404: - msg = _("The exchange to reply to " - "%(routing_key)s still doesn't exist after " - "%(duration)s sec abandonning...") % { - 'duration': duration, - 'routing_key': msg_id} - LOG.error(msg) - raise rpc_amqp.AMQPDestinationNotFound(msg) - raise - return + raise rpc_amqp.AMQPDestinationNotFound( + "exchange %s doesn't exits" % msg_id) + raise def topic_send(self, exchange_name, topic, msg, timeout=None, retry=None): """Send a 'topic' message."""