v2.1 plans

[frostyplanet]

• Implement a context both for threading(blocking) and async
• Change dependency to crossbeam-queue instead of crossbeam-channel
• Make the underlying queue implementation replaceable (optional with other lockless queue controlled by cargo feature) (lfqueue maybe?)
• Implement bounded channel size=0 (RFC: cancellation with guarantee #11)
• Increase speed under high contention
  • blocking context bounded size=1 (reference to crossbeam-channel)
  • async context bounded size=1 (comparing with kanal/flume)
  • mpsc tx (8, 16) > rx (1) (might need to improve more)
• Optimise for system with limited resources (VPS) RFC: lower CPU usage? #9
• Selection API for blocking context
• Allow mixed async blocking sender context (vice versa for receiver)

Have been looking into the code of crossbeam-channel. The implementation of Array and List flavor is identical to ArrayQueue and SegQueue of crossbeam-queue. But I don't know why crossbeam-channel does not depend on crossbeam-queue to save the coding.

[zach-schoenberger]

I noticed the same thing. the main difference being the baked in support for the channel select api as well as the blocking mechanism for send/receive. i think if you were to re-use the same blocking setup (context) using crossbeam-queue you would get the same result as their channel. which would let you implement an async version of the context.

[zach-schoenberger]

Agreed the Context in crossbeam is very heavy. It could be a lot lighter and it would be an interesting exercise to try and create the a sync context and async context to work with the queue instead of the channel. Although the cpu usage concern IMO seems a bit of a red herring. I've looked a couple different queue implementations and they all follow the same pattern: spin, spin yield, park. the only difference being the time spent in each of these. so if you want benchmarks to look good, then you configure more towards spins and higher cpu. the one thing that most lib's don't expose is how to configure these. that would be a novel concept.

[frostyplanet]

I have pushed some refactoring in branch "dev", made a few changes to LockedWaker to support thread context.
, thread context with the following missing:

• thread context blocking -> blocking channel, and not yet benchmark vs crossbeam
• send_timeout & read_timeout in thread context.
• interface for selection in thread context.
  It would be nice if someone could help me implement these functions.

I've use WakerCell to replace ArraryQueue(1), cargo bench --bench inner show individual operations speeding up, but it makes concurrent send/recv to MPSC channel slower.

commit b540c0a236686c49f20562fb87f24c1c1583dade (HEAD -> dev, origin/dev)
Author: plan <frostyplanet@gmail.com>
Date:   Wed Jul 2 15:01:03 2025 +0800

    benchmark: Add --bench inner for SpmcCell

commit 9cba9c22ceb989019bfc4efda21f97e810fa0a31
Author: plan <frostyplanet@gmail.com>
Date:   Mon Jun 30 01:38:53 2025 +0800

    impl WakerCell to replace ArrayQueue(size=1)

About the performance degradation in async context benchmark, I suspect the main ArrayQueue in MPSC mode has more conflicts when accessed faster, which may lead to retrying inside push/pop. I have to add more retry cycles and backoff inside poll_send() & poll_item, just to make the benchmarks catch previous scores.

I am currently focusing on async context optimise, might do some profiling to evaluate the cause. If no better ideas, I will try to implement direct-copy logic from SendFuture to RecvFuture (learned from kanal), which can bypass the queue.

[frostyplanet]

I have added API and benchmark for channels under blocking context. Without optimization maybe -30% performance compared to crossbeam. Still need to search for the right backoff pattern.

[frostyplanet]

Added timeout for blocking context. I've gained more experience through optimization.
Currently, the speed of blocking context is even better than crossbeam-channel. But when the channel is almost full, congestion makes things ugly. Still need to improve the algorithm.

[zach-schoenberger]

i really like that you avoided the issue thats in kanal where the receive future needs to be polled to completion in order to ensure no messages are dropped. i like this approach a lot more.

[zach-schoenberger]

What's the reasoning for using the pattern MAsyncRx/AsyncRx and similar? From what I can see its just for tracking the counts on Clone/Drop. But if you never Clone the M* version then you don't really incur much cost. Also as best I can tell there's not much reason to keep the sync/async channels separate. Would it be worth considering combining these all down to just Rx/Tx that can perform both sync/async operations?

[frostyplanet]

I think the API design is more from an engineering aspect than a technical one. I disagreed with Flume / Kanal style.

The Rust way recommends making things clear with types and marker trait. Previously, I managed a team of 20-30 people to develop distributed storage, with code of about 200-300 thousands lines.

1. Mixed blocking / async types lead to unspotable hang.

It was a painful experience writing Python or Go in the old days. Python is a blocking context runtime, adding async through gevent monkey patching. It's hard to pinpoint blocking calls that hide in async workflows (Either make the program slow or deadlock in runtime).
With GO, everything is natively async, calling c blocking API through bridging also suffers a big performance cost

With Rust, we have natively async code and blocking code at the same time, and use channels to communicate with each part. I prefer to be clear about which context the Tx/Rx is for. When cooperating with teammates, things like calling blocking recv() in async code by mistake will never happen.

2. Single or multi producer/consumer ?

The reason for providing extra API about SPSC, MPSC, we do not always require MPMC. I can optimise the underlying structure by removing the guarantee of concurrent send/recv.
The difference between AsyncRx and MAsyncRx, they share common code, but AsyncRx works with RegistrySingle, and MAsyncRx works with RegistryMulti.

Performance can be the same in 2.0.0, but in v2.1 branch, I will fully optimise the hot path, might need to introduce congestion control logic to the Tx side in MPSC (there might be locks). People will notice the difference in speed. I will also backport some commits from v2.1 to v2.0, if the changes are simple and safe enough.

Crossbeam, Flume, Kanal is all MPMC, the send/recv is only with immutable self, while tokio::mpsc recv() requires a mutable self.
I do not like mutable API, which brings inconvenience.

So lately (in v2.0.4), AsyncTx/AsyncRx/Tx/Rx is ?Sync, the M version has both Clone and Sync. Thus, preventing people from using MPSC like MPMC (likely leads to hang because RegistrySingle has no locks). And my old code calling Crossfire does not need to change

[frostyplanet]

Current state dev branch commit dd67664
For async context, the code is not much different than v2.0, it's in no doubt faster than kanal. So I did not put much effort to async lately.
I am currently working on blocking context. I feal the cost of thread park/unpark is indeed higher than async context switch.

I have Backoff loop in blocking context similar to crossbeam (looping 5~6 times before putting to sleep). Have similar speed to crossbeam. The overall cost of RegistryMulti (with VecDeque) is a bit smaller than crossbeam's SyncWaker (with Vec).
I discover due to this reason, mpsc 8 tx 1rx, is more likely to be congested than crossbeam. Too many unsuccessful try_send will lead to performance drop (<2M/s). even crossbeam pattern works for 8tx, it will be worse with 16 tx ( <600K/s). I might still have some ideas to lower the cost in high contention, but experiments will take time.

I think direct copy is better in blocking than async (since async future will cancel, blocking context will not, and timeout need to call send_timeout specificly). I might resolve to direct copy method after pursuing the congestion control algorithm.

A snapshot of benchmark result (commit dd67664)

mpsc bounded_sync (size 100)	1x1	2x1	4x1	8x1	16x1
crossfire(mpmc)	33.674
crossfire(mpsc)	36.116，31.461	30.79	23.626	- congest	-congest
crossfire(spsc)	35.67
crossbeam	17.817	18.881	19.659	19.553	16.912
kanal	6.5858	5.1853	7.092	3.2021	2.2863
flume	7.9591	5.4153	2.0428	0.71437	0.3531

mpmc bounded_sync (size 100)		2x2	4x4	8x8	16x16
crossfire::mpmc		39.584	25.154	21.036	16.231
crossbeam		14.199	15.507	13.926	8.9083
kanal		6.069	7.1058	6.5273	6.2235
flume		4.4146	2.568	1.8669	1.4746

mpsc bounded_async (size 100)	1x1	2x1	4x1	8x1	16x1
crossfire::mpmc (Me/s)	27.895	24.688	19.967	17.413	16.818
crossfire::mpsc (Me/s)	28.694	28.668	23.792	19.746	18.692
crossfire::spsc (Me/s)	29.642
kanal	8.4339	8.3596	7.7418	7.6994	7.8384
flume (Me/s)	5.3625	4.3943	2.9271	1.9059	1.4388

mpmc bounded_async (size 100)		2x2	4x4	8x8	16x16
crossfire::mpmc (Me/s)		22.081	17.493	17.062	17.003
kanal		16.45	4.3938	5.9732	11.752
flume (Me/s)		4.1362	2.685	2.3369	2.2937

[frostyplanet]

@zach-schoenberger Are you interested in participating v2.1? If you have time to spare, we can discuss the details

[frostyplanet]

Solve the congestion problem with a simple fix

commit d269a3b0fd71599d652bf3062bd8111dd9be4b71 (HEAD -> dev, origin/dev)
Author: plan <frostyplanet@gmail.com>
Date:   Sat Jul 12 17:42:34 2025 +0800

    blocking_tx: yield to other thread after send ok
    
    Sender is too fast, yield to slow down and give consumer a chance
    
    crossfire_bounded_100_blocking_n_1
    mpsc/4
        31.807 Melem/s
    mpsc/8
        22.653 Melem/s
    mpsc/16
        15.818 Melem/s

[zach-schoenberger]

absolutely!

[frostyplanet]

I have created discord channel: https://discord.gg/YqZrBRPFxX

[frostyplanet]

Just create a workflow named "cron-dev" to run tests in "dev" branch. I've noticed "cron-dev" takes 49m vs "cron-master" 33m. To locate the time cause, I have to isolate the time span in each case. I'm going to the log wrapper after fixing this issue NaturalIO/captains-log#6

My suspicion: something to do with the spinning in backoff, or there are bad cases not discovered through benchmarks.

[frostyplanet]

The main difference is tests about bound size=1 async->async. It looks like not much difference in my poll(), maybe the underlying operation to the queue is shorter without going to crossbeam-channel. The less chance of getting a message. I think it might be related to runtime behavior.
I have resolved to adaptive backoff according to the bound size, for bound size=1 special case, backoff 6 times.
It's a temporary solution until we have a better idea.

 commit 7d154c680a55d58f43fa414bdaa6d122b1fa6c50 (HEAD -> dev, origin/dev)
Author: plan <frostyplanet@gmail.com>
Date:   Thu Jul 17 02:37:18 2025 +0800

    async: detect tokio runtime for backoff settings

For test_pressure_bounded_async_multi_1
dev (2 try send/recv) 227.72s
dev (1 try send/recv + is_full/is_empty) 219
dev (adaptive backoff) 113.25s , test in SINGLE_THREAD_RUNTIME=1 89.92s

[frostyplanet]

Found the cause of test running slow, it was the lock inside Clone that hinder test setup, currently cron_dev only needs 35min to finish, compared with cron_master 47min.