When to use threads with SelectConnection

[James McKinney]

I had been using BlockingConnection, but this adapter makes it difficult/impossible to survive RabbitMQ restarts. (I've tried catching ConnectionClosedByBroker, but this exception seems to be raised from outside any code I write – I can't catch it. If there is a consistent way to recover from this event, please let me know.)

I am therefore switching to SelectConnection, which offers a callback for this event.

When using BlockingConnection, I had been using a ThreadPoolExecutor, and the on_message callback would submit work to the pool, since some tasks take longer than a reasonable heartbeat. Also, this enables for concurrent processing of messages, allowing me to get better performance with a higher prefetch count.

I've read the documentation/issues/discussions, but it's unclear to me whether I need to use threads with SelectConnection for either (1) heartbeat or (2) concurrency, or both.

Could you clarify?

[James McKinney]

Okay, I can confirm that threads are definitely needed for concurrency in SelectConnection (the IO loop blocks on all adapters).

New issue: I'm working out how to wait for the threads to terminate, before closing the channel and connection. For example, in the basic.cancel-ok callback, I run:

self.executor.shutdown(cancel_futures=True)

self.channel.close()

ThreadPoolExecutor.shutdown() has wait=True by default, so it blocks until all threads terminate. At the end of most of my threads, I use connection.ioloop.add_callback_threadsafe to call channel.basic_ack() in the callback. From debugging, I can see that:

1. add_callback_threadsafe() returns, in the thread

2. the thread terminates

3. executor.shutdown() returns, in the basic.cancel-ok callback

4. channel.close() is called, in the basic.cancel-ok callback

5. the thread-safe callback runs, but the channel is now closed

Is there any way to ensure that the callback runs, before the channel is closed?

--
You received this message because you are subscribed to the Google Groups "Pika" group.
To unsubscribe from this group and stop receiving emails from it, send an email to pika-python...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/9f1f3111-4ed9-4ed4-860e-1ad250287025n%40googlegroups.com.

--

James McKinney

Head of Technology

+1-514-247-0223 | @mckinneyjames | timezone: EST

💙💛 Learn how we are supporting Ukraine's Reconstruction here #URC2023 💙💛

www.open-contracting.org | follow us @opencontracting

[Nough You]

isn't that why try: except: finally: exists? try finally. 

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVe9cHbJp7NP42ewgw4FYtfTb7tPpLg51bvdgWHMWBmdqg%40mail.gmail.com.

[Gavin M. Roy]

You should never use threads. The whole idea of using an async client is to dispatch your code on the IOLoop, not work around it.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVe9cHbJp7NP42ewgw4FYtfTb7tPpLg51bvdgWHMWBmdqg%40mail.gmail.com.

[James McKinney]

> The whole idea of using an async client is to dispatch your code on the IOLoop, not work around it.

Maybe I'm missing something, but to test, I wrote a simple consumer callback that sleeps for 5 seconds and logs messages:

def consumer_callback(channel, method, properties, body):

    logger.info("Sleep")
    time.sleep(5)
    logger.info("Wake!")

I then created 10 messages in RabbitMQ. Using `basic_consume` on the SelectConnection, the log messages appear as:

Sleep

Wake!

Sleep

Wake!

etc.

with five seconds elapsing between "Sleep" and "Wake!", so about 50 seconds total.

If instead I use a consumer callback that submits the messages to a thread pool (whose callback is otherwise identical), I see:

Sleep

Sleep

...

Wake!

Wake!

...

With 5 seconds between the "Sleep" messages and the "Wake!" messages, so 5 seconds in total.

If there's a way to get the same concurrency without manually creating threads, I would love to know.

> isn't that why try: except: finally: exists? try finally. 

To answer my question, I need to do:

self.executor.shutdown(cancel_futures=True)
self.connection.ioloop.call_later(0, self.channel.close)

This will ensure the (pending) thread-safe callback is run before the (newly-scheduled) channel.close.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAPnH5kpN%3DaAVHY-sUTfi1c8zNxC%2B0ZsVHsF6WVFa%3DVOZtZNxAQ%40mail.gmail.com.

[James McKinney]

I forgot to mention: In my test I set prefetch_count to 10, so it gets all the messages at once.

[Gavin M. Roy]

time.sleep will block the IOLoop and is generally a bad idea in async code. Depending on your needs, you might want to consider using the AsyncIO connection adapter, then you can use asyncio.sleep: https://docs.python.org/3/library/asyncio-task.html#asyncio.sleep

On Sun, Jul 2, 2023 at 11:12 AM James McKinney <jmck...@open-contracting.org> wrote:

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVe9LZj7ftfmHh1-F2yrF7GSb1mjcKDj%2B4QmmxZ9G1pAuw%40mail.gmail.com.

[James McKinney]

Aha. I just used sleep here to try to simulate a CPU-heavy computation. However, if I simulate that with something like:

for i in range(100000000):

    pass

I see that the messages are indeed processed concurrently.

Thanks!

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAPnH5kqTJps8nDqs8THhusxDOxvTqFz-emKODJYdtwFu35QArg%40mail.gmail.com.

[Gavin M. Roy]

If you're thinking of doing compute heavy work on an IOLoop you should consider using the ThreadPoolExecutor for that processing with the IOLoop staying in the foreground. See https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVc7h6ik%2Bd79hV2oYj5Ti0C_fYCRksMiZ8Xt97J7CnuSNQ%40mail.gmail.com.

[James McKinney]

At first you said threads shouldn't be used, but they are fine to use for compute heavy work?

If I'm using the SelectConnection, would I use the ThreadPoolExecutor from concurrent.futures instead?

Or, do you recommend AsyncioConnection over SelectConnection, anyway?

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAPnH5kott9BMDSogwE%2B9jT5sikQ_3kLQu2xAw6H4dCG%3Ddy95BQ%40mail.gmail.com.

[Gavin M. Roy]

Sorry I was not clear.

1. Threads should not be intermixed in the actual pika client usage. It's very easy to get threading wrong and have incorrect protocol behaviors.

2. I would absolutely use asyncio over the IOLoop in the SelectConnection. There's a whole ecosystem around asyncio in Python3 you benefit from. If asyncio existed when I wrote SelectConnection, I wouldn't have written SelectConnection or it's IOLoop.

3. The idea behind the threaded executor is you let the asyncio loop manage the background execution of CPU intensive tasks in a way that doesn't interrupt or change the context of IOLoop and the things running on it. 

More context: amqp is a bidirectional RPC protocol. To do it right, the client library should be able to respond in real time to RPCs from the server to the client. It also expects strict sequencing of RPC frames. Introducing threads that are aware of pika connections and channels introduces the ability to intersperse RPCs frames in a way that breaks the protocol. Instead if you receive a message for example, then dispatch the work via a thread pool executor managed by the IOLoop, the client usage stays non-threaded, while compute heave work is processed "in the background" and returns from the function when complete as if it were just a normal execution of a function on the ioloop.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVfiOrAeANgtMnP347Sn8%3DE-NNWTmd67r%2BqFpDFXaMdDXg%40mail.gmail.com.

[James McKinney]

Thank you for the explanation!

In case anyone is interested, I wrote a package for my Pika wrappers (to avoid repetitive boilerplate across my projects). https://yapw.readthedocs.io/en/latest/

James

On Sun, Jul 2, 2023 at 12:11 PM Gavin M. Roy <gavi...@gmail.com> wrote:

Sorry I was not clear.

1. Threads should not be intermixed in the actual pika client usage. It's very easy to get threading wrong and have incorrect protocol behaviors.

2. I would absolutely use asyncio over the IOLoop in the SelectConnection. There's a whole ecosystem around asyncio in Python3 you benefit from. If asyncio existed when I wrote SelectConnection, I wouldn't have written SelectConnection or it's IOLoop.

3. The idea behind the threaded executor is you let the asyncio loop manage the background execution of CPU intensive tasks in a way that doesn't interrupt or change the context of IOLoop and the things running on it. 

More context: amqp is a bidirectional RPC protocol. To do it right, the client library should be able to respond in real time to RPCs from the server to the client. It also expects strict sequencing of RPC frames. Introducing threads that are aware of pika connections and channels introduces the ability to intersperse RPCs frames in a way that breaks the protocol. Instead if you receive a message for example, then dispatch the work via a thread pool executor managed by the IOLoop, the client usage stays non-threaded, while compute heave work is processed "in the background" and returns from the function when complete as if it were just a normal execution of a function on the ioloop.

On Sun, Jul 2, 2023 at 12:01 PM James McKinney <jmckinney@open-contracting.org> wrote:

At first you said threads shouldn't be used, but they are fine to use for compute heavy work?

If I'm using the SelectConnection, would I use the ThreadPoolExecutor from concurrent.futures instead?

Or, do you recommend AsyncioConnection over SelectConnection, anyway?

On Sun, Jul 2, 2023 at 11:39 AM Gavin M. Roy <gavi...@gmail.com> wrote:

If you're thinking of doing compute heavy work on an IOLoop you should consider using the ThreadPoolExecutor for that processing with the IOLoop staying in the foreground. See https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/9f1f3111-4ed9-4ed4-860e-1ad250287025n%40googlegroups.com.

--

James McKinney

Head of Technology

+1-514-247-0223 | @mckinneyjames | timezone: EST

💙💛 Learn how we are supporting Ukraine's Reconstruction here #URC2023 💙💛

www.open-contracting.org | follow us @opencontracting

--
You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVe9cHbJp7NP42ewgw4FYtfTb7tPpLg51bvdgWHMWBmdqg%40mail.gmail.com.

--
You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAPnH5kpN%3DaAVHY-sUTfi1c8zNxC%2B0ZsVHsF6WVFa%3DVOZtZNxAQ%40mail.gmail.com.

--

James McKinney

Head of Technology

+1-514-247-0223 | @mckinneyjames | timezone: EST

💙💛 Learn how we are supporting Ukraine's Reconstruction here #URC2023 💙💛

www.open-contracting.org | follow us @opencontracting

--
You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVe9LZj7ftfmHh1-F2yrF7GSb1mjcKDj%2B4QmmxZ9G1pAuw%40mail.gmail.com.

--
You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAPnH5kqTJps8nDqs8THhusxDOxvTqFz-emKODJYdtwFu35QArg%40mail.gmail.com.

--

James McKinney

Head of Technology

+1-514-247-0223 | @mckinneyjames | timezone: EST

💙💛 Learn how we are supporting Ukraine's Reconstruction here #URC2023 💙💛

www.open-contracting.org | follow us @opencontracting

--
You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVc7h6ik%2Bd79hV2oYj5Ti0C_fYCRksMiZ8Xt97J7CnuSNQ%40mail.gmail.com.

--
You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAPnH5kott9BMDSogwE%2B9jT5sikQ_3kLQu2xAw6H4dCG%3Ddy95BQ%40mail.gmail.com.

--

James McKinney

Head of Technology

+1-514-247-0223 | @mckinneyjames | timezone: EST

💙💛 Learn how we are supporting Ukraine's Reconstruction here #URC2023 💙💛

www.open-contracting.org | follow us @opencontracting

--
You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAJNoxVfiOrAeANgtMnP347Sn8%3DE-NNWTmd67r%2BqFpDFXaMdDXg%40mail.gmail.com.

--

You received this message because you are subscribed to the Google Groups "Pika" group.

To unsubscribe from this group and stop receiving emails from it, send an email to pika-python+unsubscribe@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/pika-python/CAPnH5kpW9kMV%2BOy2mq%3DWMeFJG39Nq_3oJ1p%2B_0hTOq1jHEvNog%40mail.gmail.com.