tx_commit() semantics with async connections (e.g Pika SelectConnection)
I've been drilling into an issue for a few days where my Pika channel is abruptly closed. I'm pretty sure I've found either:
- A large gap in my understanding
- A misuse of the Rabbit/Pika APIs
- A serious bug in Pika.
My original understanding about transactions is that calling tx_select() is a synchronous operation, and that when it returns, whatever actions you've sent to the broker are committed.
However, in stepping through the Pika tx_commit() code, it seems like it all it does is add the Tx.Commit message to an outgoing buffer, and that the tx_commit() call returns without anything actually sent to the broker. (I've set breakpoints in the debugger at strategic points to verify this.)
In my message handler, I'm writing a reply to the incoming message, using the same channel. What I'm seeing (verified via WireShark) is that the Tx.Commit message is followed by some number of Basic.Publish, Content-Header, and Content-Body records corresponding to my reply message writes. This violates the AMQP standard, which says that nothing should be sent to the broker after sending a Tx.Commit and before receiving a Tx.Commit-Ok reply. (If it helps understanding, my incoming message queue has a few messages in it already, so I see a series of incoming messages immediately after starting.)
I see that the tx_commit() method takes a callback method, but if I were to simply block in my message handler, waiting for the callback to be invoked, I'd deadlock because Pika's buffer processing code doesn't get a chance to run.
My code looks like this:
def echo_on_connected(connection):
"""Called when we are fully connected to RabbitMQ"""
# Open a channel
connection.channel(echo_on_channel_open)
# Step #3
def echo_on_channel_open(channel):
channel.queue_declare(queue=echo_queue_name, durable=True, arguments={"x-ha-policy": "all"})
channel.tx_select()
channel.basic_consume(echo_handle_delivery, queue=echo_queue_name)
def echo_handle_delivery(channel, method, header, body):
print "Received message %s" % (body)
# Write a reply message on the same channel
channel.basic_publish(exchange='',
routing_key='reply_queue',
body="replying to %s" % (body),
properties=BasicProperties(delivery_mode=2))
# Ack the incoming message, then force the write to be flushed.
channel.basic_ack(delivery_tag=method.delivery_tag)
channel.tx_commit()
def echo_thread():
connection_parameters = ConnectionParameters(host=broker_host)
echo_connection = SelectConnection(connection_parameters, echo_on_connected)
echo_connection.ioloop.start()
<div>
<p>I've been drilling into an issue for a few days where my Pika channel is abruptly closed. I'm pretty sure I've found either:<br></p>
<ul>
<li>A large gap in my understanding</li>
<li>A misuse of the Rabbit/Pika APIs</li>
<li>A serious bug in Pika.</li>
</ul>
<p>My original understanding about transactions is that calling tx_select() is a synchronous operation, and that when it returns, whatever actions you've sent to the broker are committed.</p>
<p>However, in stepping through the Pika tx_commit() code, it seems like it all it does is add the Tx.Commit message to an outgoing buffer, and that the tx_commit() call returns without anything actually sent to the broker. (I've set breakpoints in the debugger at strategic points to verify this.)</p>
<p>In my message handler, I'm writing a reply to the incoming message, using the same channel. What I'm seeing (verified via WireShark) is that the Tx.Commit message is followed by some number of Basic.Publish, Content-Header, and Content-Body records corresponding to my reply message writes. This violates the AMQP standard, which says that nothing should be sent to the broker after sending a Tx.Commit and before receiving a Tx.Commit-Ok reply. (If it helps understanding, my incoming message queue has a few messages in it already, so I see a series of incoming messages immediately after starting.)<br></p>
<p>I see that the tx_commit() method takes a callback method, but if I were to simply block in my message handler, waiting for the callback to be invoked, I'd deadlock because Pika's buffer processing code doesn't get a chance to run.<br></p>So given the above, is it me? Is it Pika?<br><br>My code looks like this:<br><br><span>def echo_on_connected(connection):</span><br><span> """Called when we are fully connected to RabbitMQ"""</span><br><span> # Open a channel</span><br><span> connection.channel(echo_on_channel_open)</span><br><br><span># Step #3</span><br><span>def echo_on_channel_open(channel):</span><br><span> channel.queue_declare(queue=echo_queue_name, durable=True, arguments={"x-ha-policy": "all"})</span><br><span> channel.tx_select()</span><br><span> channel.basic_consume(echo_handle_delivery, queue=echo_queue_name)</span><br><br><span>def echo_handle_delivery(channel, method, header, body):</span><br><span> print "Received message %s" % (body)</span><br><br><span> # Write a reply message on the same channel</span><br><span> channel.basic_publish(exchange='',</span><br><span> routing_key='reply_queue',</span><br><span> body="replying to %s" % (body),</span><br><span> properties=BasicProperties(delivery_mode=2))</span><br><br><span> # Ack the incoming message, then force the write to be flushed.</span><br><span> channel.basic_ack(delivery_tag=method.delivery_tag)</span><br><span> channel.tx_commit()</span><br><br><span>def echo_thread():</span><br><span> connection_parameters = ConnectionParameters(host=broker_host)</span><br><span> echo_connection = SelectConnection(connection_parameters, echo_on_connected)</span><br><span> echo_connection.ioloop.start()</span><br>
</div>
RSS Feed