Improve congestion control

[Mallets]

This PR introduces:

1. congestion detection in the transmission pipeline per priority queue
2. early dropping in case congestion is detected bypassing the wait_before_drop timeout

Early dropping drastically helps in the case of having a slow subscriber in the system, e.g.:

                       |-> | fast subscriber |
| pub | -> | router | -|
                       |-> | slow subscriber |

That means a slow subscriber will have a minima impact on the system when CongestionControl::Drop is used.
No change in behaviour is introduced in the case of using CongestionControl::Block.

Validation steps

Set CongestionControl::Block in the z_pub_thr example.

Modify the subscriber callback to block for some time in the z_sub example.

while let Ok(sample) = subscriber.recv_async().await {
    tokio::time::sleep(Duration::from_millis(100)).await;
}

Main

Run the router:

./target/release/zenohd

Run throughput publisher:

./target/release/examples/z_pub_thr 8 -m client

Run slow subscriber:

./target/release/examples/z_sub -m client -k test/thr

Run throughput subscriber:

./target/release/examples/z_sub_thr -m client -n 500 -s 1000
Press CTRL-C to quit...
774.8841943707797 msg/s
779.9427331103915 msg/s
775.5408884676764 msg/s
781.3848084140717 msg/s
782.5382842182961 msg/s
781.2933373648069 msg/s
781.5737730415275 msg/s

This PR

Run the router:

./target/release/zenohd

Run throughput publisher:

./target/release/examples/z_pub_thr 8 -m client

Run slow subscriber:

./target/release/examples/z_sub -m client -k test/thr

Run throughput subscriber:

./target/release/examples/z_sub_thr -m client -n 5000000 -s 1000
Press CTRL-C to quit...
3053345.844926824 msg/s
3126748.454285023 msg/s
3077751.70238141 msg/s
3115776.8953388007 msg/s
3103958.315081412 msg/s
3127926.9733049045 msg/s
3104865.110821202 msg/s

[yellowhatter]

Discussional: in order to reduce contention around this atomic, I'd recommend switching to a set of separate AtomicBool and make 64-byte padding (maybe indirect by putting them into StageIn etc). The cost of operation for multi-(thread,core,priority) setup should be smaller.

[Mallets]

I thought about that but increasing the number of atomics also increases the amount of data to be kept in cache. Having one atomic per priority will significantly put more burden on moving data across different cache level I assume, thus reducing performance in any case. I don't have an experimental comparison but I'm not expecting a lot of contention in general.

[yellowhatter]

There is also some systematic throughput regression on my Linux laptop:

Ref:

This PR:

[yellowhatter]

The bench is using CC::Block, so I suppose we are having smth with data structure sizes, not with atomics...

[Mallets]

Interesting, on my laptop I've actually gained ~500K msg/s with 8 bytes payload with this PR...

[yellowhatter]

Interesting, on my laptop I've actually gained ~500K msg/s with 8 bytes payload with this PR...

I will retest it!

[Mallets]

I have to say that the gain in performance I observe on my laptop is not justified either... Nothing on this PR should provide better performance on the nominal case.

[yellowhatter]

I have to say that the gain in performance I observe on my laptop is not justified either... Nothing on this PR should provide better performance on the nominal case.

So, after reboot I've got results equal to the reference!

[yellowhatter]

A really great thing for slow subscriber problem!