Show HN: Сulsans – Thread-safe async-aware queue for Python | Hacker News Reader

Show HN: Сulsans – Thread-safe async-aware queue for Python

github.com

2 pointsx42005e1f4 comments

In my previous post [0], I described how I came to create aiologic. Here, I want to do the same for a derivative library - Сulsans.

In October 2024, I started thinking about how to present aiologic queues. Andrew Svetlov's Janus library [1] had been around for quite some time and was much more popular, so I knew that comparisons with it would be inevitable. However, Janus seemed to be in a suspended state: there had been no major changes for three years, and almost all commits during that period were made by Dependabot. So I asked a relevant question [2].

During the discussion, I pointed out Janus' performance issues and stated that they could be solved by implementing queues on top of my primitives. But since Janus is a mature library, such a radical change could not be accepted. Therefore, as proof of concept, I implemented a new library - Culsans. That is how its story began.

Over time, both libraries underwent changes. Janus received significant performance improvements in 1.2.0, not least due to my PRs [3]. In 2.0.0, contrary to the above, backward compatibility was broken as a result of the implementation of shutdown methods. And Culsans became an independent library with its own features (which neither aiologic nor Janus have).

So, what is Culsans? It is a library that provides a way to communicate within a single process between different threads, different tasks (including from different event loops; asyncio, Curio, Trio, AnyIO - whatever you want), and even different greenlets (eventlet/gevent), all in a single instance. Its queues are fully compatible with the standard queues via Janus-like interfaces (as well as with Janus itself) and provide additional features such as dynamic maxsize. In short, I invite you to try out my library and see for yourself.

[0] https://news.ycombinator.com/item?id=46308839 [1] https://github.com/aio-libs/janus [2] https://github.com/aio-libs/janus/issues/679 [3] https://github.com/aio-libs/janus/pull/704

Open articleView on HN

How does this differ from asyncio.Queue in terms of backpressure or cancellation semantics?

If you use `culsans.Queue().async_q` as a direct replacement for `asyncio.Queue()`, then there is essentially no difference. The difference becomes apparent when you use additional features:

1. If checkpoints are enabled (by default when using Trio, or if you explicitly apply `aiologic.lowlevel.enable_checkpoints()`), then every call that is not explicitly non-blocking can be cancelled (even if no waiting is required). For comparison, `await queue.put(await queue.get())` for `queue = asyncio.Queue()` in an infinite loop will never yield back to the event loop (when 0 < size < maxsize is true), and as a result, no other asyncio tasks will ever continue their execution, and such a loop cannot be cancelled (see PEP 492).

2. With multithreading and corresponding race conditions, method calls are synchronized using the underlying lock (as in `queue.Queue`). This means that such synchronization can temporarily block the event loop, but this is rarely a bottleneck (the same is used in Janus). In general, this delays task cancellation and timeout handling if someone else is still holding the lock. If you need extremely fast and scalable queues, `aiologic.SimpleQueue` may be the best option (it does not use any form of internal state synchronization!).

I am not sure I understand your question well enough. `asyncio.Queue` works exclusively in cooperative multitasking (it is not thread-safe) with all the resulting simplifications. The principle of operation of Culsans queues under the same conditions is almost the same as that of any other queues capable of operating as purely asynchronous with cancellation support (perhaps you are referring to starting new threads or new tasks as an implementation detail? aiologic and Culsans do not use any of this). As soon as preemptive multitasking is introduced, the behavior may change somewhat - `culsans.Queue` relies on sync-only synchronization of the internal state, `aiologic.Queue` on async-aware synchronization (without blocking the event loop; still used because `heapq` functions are not thread-safe, and they are required for priority queues; but the wait queues are combined, which achieves fairness and solves python/cpython#90968), and `aiologic.SimpleQueue` does not synchronize the internal state at all due to the use of effectively atomic operations.

Thanks, that clarifies it. The checkpoint-based cancellation and the sync-vs-async locking model differences were exactly what I was trying to understand.

I would like to add that you can also read about some non-trivial details in the "Performance" section of the aiologic documentation [4]. What is described there for standard primitives also applies to Culsans queues (specifically, the mutex case; however, other documentation sections (such as "Why?", "Overview", and "Libraries") are also relevant to Culsans, since aiologic is used under the hood).

[4] https://aiologic.readthedocs.io/latest/performance.html

Reply on news.ycombinator.com