How does this differ from asyncio.Queue in terms of backpressure or cancellation semantics?
replies(1):
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
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.