using add_callback_threadsafe lost acknowledgement when stopping program

[Stephane-Thales]

I'm using pika to process RabitMQ message by small batch, and using a thread for each batch. At the end of the function in the thread, I send acknowledgement of the messages through add_callback_threadsafe to the channel.

In parallèle I'm catching SIGINT signals to stop the program properly, by waiting with thread.join() that all threads finish before stopping the channel consume and closing the connection.

But once the CtrlC is sent to generate the SIGINT, event if the program wait for all threads to finish, the acknowledgement will not be processed.

==> is there a way to force the channel/connection to process the waiting add_callback_threadsafe before closing the connection ?

( initial post here)

schema of the issue

git repo for the error

(note : you need to change the connection info to RabbitMQ : host, port, login, password line 107 to 111)

Versions :

• Python 3.7.13
• pika 1.3.1
• RabbitMQ 3.6.10

[lukebakken]

I have forked your test project - https://github.com/lukebakken/pika-1402

Seems like there are some errors. I'll fix them in my fork

https://github.com/lukebakken/pika-1402/tree/lukebakken/pika-1402

DEBUG main : queue test_ctrlC purged
DEBUG main : test messages created in test_ctrlC
 [*] Waiting for messages. To exit press CTRL+C
DEBUG block_process : start of block_process function
Exception in thread Thread-2:
Traceback (most recent call last):
  File "/home/lbakken/.asdf/installs/python/3.11.0/lib/python3.11/threading.py", line 1038, in _bootstrap_inner
    self.run()
  File "/home/lbakken/.asdf/installs/python/3.11.0/lib/python3.11/threading.py", line 1394, in run
    self.function(*self.args, **self.kwargs)
  File "/home/lbakken/development/lukebakken/pika-1402/test_pika_blockthread.py", line 43, in block_process
    if event and event.isAlive():
                 ^^^^^^^^^^^^^
AttributeError: 'Timer' object has no attribute 'isAlive'. Did you mean: 'is_alive'?

[lukebakken]

@Stephane-Thales I'm pretty sure I have reproduced the issue. I think I can simplify the code to make it a bit easier to repeat. Stay tuned!

[Stephane-Thales]

Hi @lukebakken , if you have been able to reproduce the issue, can this thread be converted from Q&A to Issue ? Or is there a workaround to get pika process the add_callback_threadsafe outside the .start_consuming() ?

[lukebakken]

The next thing I was going to try is to run the Pika connection in its own thread, so that the main thread (which catches the signal) isn't the one running the Pika connection.

I'll eventually create an issue when I know exactly what the bug is.

[Stephane-Thales]

I've done a first test but i'm not sure I've done it properly, and I fail to do the stop_consuming.

Main thread A : signal that will launch the stop consume() within cleanClose()
1st thread B : start_consume()
other threads C : process and send the add_callback_threadsafe for ack

C ==> B work
but A ==> B don't work, error on the Ctrl+C with pika.exceptions.StreamLostError: Stream connection lost

I'll let you try, please keep me posted.
thanks

[lukebakken]

@Stephane-Thales please see the latest version of my code here:

https://github.com/lukebakken/pika-1402/blob/lukebakken/pika-1402/test_pika_blockthread.py

Note that I did the following

• Simplified it a bit by removing the "batch processing via Queue" code since it wasn't related to the current issue
• Moved the Pika connection to its own thread
• Instead of using a consume callback, move to a generator-style for loop which allows for checking if exiting is requested. This could also be accomplished via SelectConnection and a timer.

The code works correctly when CTRL-C is pressed. My guess is that the original issue stems from the fact that BlockingConnection is built on top of SelectConnection and running it in the same thread as that which receives the signal doesn't work quite right. People should be using SelectConnection anyway when their Pika program is more complex than just a simple publisher / consumer.

I'll file a bug but fixing it will be low priority. Thanks for reporting this and let me know if my solution works in your final product.

[lukebakken]

Oh interesting, I'm navigating Pika's code and it appears stop_consuming() without a consumer tag basically does what cancel does 🙄 I'm going to change the test program to save and pass the consumer tag.

[lukebakken]

If you haven't figured this out already, I didn't write Pika, I just maintain it 😉

[Stephane-Thales]

Thanks for the reactive answers !

[lukebakken]

Check out the latest version of my code.

[Stephane-Thales]

Thanks, so far all test look working, I've added back the queue/batch part, and that work too, I will test with more data, then convert with my specific code and continue the test.
Thank-you for your help.