queue: Use a thread-local Pika connection.

According to the documentation: “Pika does not have any notion of
threading in the code. If you want to use Pika with threading, make
sure you have a Pika connection per thread, created in that thread. It
is not safe to share one Pika connection across threads, with one
exception: you may call the connection method add_callback_threadsafe
from another thread to schedule a callback within an active pika
connection.”

https://pika.readthedocs.io/en/stable/faq.html

This also means that synchronous Django code running in Tornado will
use its own synchronous SimpleQueueClient rather than sharing the
asynchronous TornadoQueueClient, which is unfortunate but necessary as
they’re about to be on different threads.

Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg
2022-04-15 17:30:58 -07:00
committed by Alex Vandiver
parent c9faefd50e
commit c263bfdb41
3 changed files with 24 additions and 34 deletions

View File

@@ -21,7 +21,7 @@ from zerver.tornado.event_queue import (
from zerver.tornado.sharding import notify_tornado_queue_name
if settings.USING_RABBITMQ:
from zerver.lib.queue import TornadoQueueClient, get_queue_client
from zerver.lib.queue import TornadoQueueClient, set_queue_client
class Command(BaseCommand):
@@ -67,8 +67,8 @@ class Command(BaseCommand):
print(f"Tornado server (re)started on port {port}")
if settings.USING_RABBITMQ:
queue_client = get_queue_client()
assert isinstance(queue_client, TornadoQueueClient)
queue_client = TornadoQueueClient()
set_queue_client(queue_client)
# Process notifications received via RabbitMQ
queue_name = notify_tornado_queue_name(port)
queue_client.start_json_consumer(
@@ -88,7 +88,8 @@ class Command(BaseCommand):
logging_data["port"] = str(port)
setup_event_queue(http_server, port)
add_client_gc_hook(missedmessage_hook)
setup_tornado_rabbitmq()
if settings.USING_RABBITMQ:
setup_tornado_rabbitmq(queue_client)
instance = ioloop.IOLoop.instance()