Add fn force_push to ArrayQueue

[brunocodutra]

This is an attempt to implement a straightforward MPMC ring-buffer and close #680.

This proposal adds a new method push_or_swap to ArrayQueue, that atomically swaps the oldest element when the queue is full, instead of returning Err back to the caller like push does. As such, push_or_swap never fails to insert the element into the queue.

I couldn't find any benchmarks I could run, (am I missing anything obvious?), however I did run benchmarks from ring-channel where I compared this implementation against an emulation of a ring-buffer, that keeps popping elements until pushing succeeds, i.e. something like the following:

while let Err(v) = q.push(value) {
   q.pop();
   value = v;
}

I got the results below on my machine, which show that push_or_swap fares much better when capacity is low and the probability that pushing fails is high (the baseline was set to the push-based implementation).

• Note 1: the relevant metric in the benchmarks below is the throughput, which is scaled by the "channel efficiency", defined as total_number_of_messages_received / total_number_of_messages_sent.
• Note 2: benchmark names are suffixed by SB/PxR/C, where S is the size of the element in bytes, P is the number of threads producing, R the number of threads consuming, and C is the capacity of the ring-buffer:
• Note 3: the source code for the benchmarks is available here.

Benchmarking mpmc/Block/32B/8x8/1: Warming up for 3.0000 s
Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 7.2s, enable flat sampling, or reduce sample count to 50.
mpmc/Block/32B/8x8/1    time:   [1.3466 ms 1.3982 ms 1.4507 ms]                                  
                        thrpt:  [1.4117 Melem/s 1.4647 Melem/s 1.5209 Melem/s]
                 change:
                        time:   [-33.037% -28.797% -24.494%] (p = 0.00 < 0.05)
                        thrpt:  [+32.440% +40.443% +49.337%]
                        Performance has improved.
Found 2 outliers among 100 measurements (2.00%)
  1 (1.00%) low mild
  1 (1.00%) high mild
mpmc/Block/32B/8x8/16   time:   [367.57 us 374.55 us 382.12 us]                                  
                        thrpt:  [5.3596 Melem/s 5.4679 Melem/s 5.5717 Melem/s]
                 change:
                        time:   [-2.1237% +0.3288% +2.6459%] (p = 0.79 > 0.05)
                        thrpt:  [-2.5777% -0.3277% +2.1698%]
                        No change in performance detected.
Found 3 outliers among 100 measurements (3.00%)
  3 (3.00%) high mild

Benchmarking mpsc/Block/32B/15x1/1: Warming up for 3.0000 s
Warning: Unable to complete 100 samples in 5.0s. You may wish to increase target time to 9.9s, enable flat sampling, or reduce sample count to 50.
mpsc/Block/32B/15x1/1   time:   [3.5292 ms 3.7286 ms 3.9535 ms]                                   
                        thrpt:  [971.28 Kelem/s 1.0299 Melem/s 1.0881 Melem/s]
                 change:
                        time:   [-51.773% -43.940% -34.318%] (p = 0.00 < 0.05)
                        thrpt:  [+52.248% +78.380% +107.35%]
                        Performance has improved.
Found 5 outliers among 100 measurements (5.00%)
  4 (4.00%) high mild
  1 (1.00%) high severe
mpsc/Block/32B/15x1/16  time:   [853.29 us 873.07 us 895.27 us]                                   
                        thrpt:  [4.2892 Melem/s 4.3983 Melem/s 4.5003 Melem/s]
                 change:
                        time:   [-8.3188% +0.1727% +9.3995%] (p = 0.97 > 0.05)
                        thrpt:  [-8.5919% -0.1724% +9.0736%]
                        No change in performance detected.
Found 5 outliers among 100 measurements (5.00%)
  3 (3.00%) high mild
  2 (2.00%) high severe

spmc/Block/32B/1x15/1   time:   [163.94 us 169.05 us 173.89 us]                                  
                        thrpt:  [1.4722 Melem/s 1.5144 Melem/s 1.5616 Melem/s]
                 change:
                        time:   [-6.0575% -1.4457% +3.5710%] (p = 0.55 > 0.05)
                        thrpt:  [-3.4479% +1.4669% +6.4481%]
                        No change in performance detected.
Found 7 outliers among 100 measurements (7.00%)
  6 (6.00%) low mild
  1 (1.00%) high mild
spmc/Block/32B/1x15/16  time:   [49.955 us 53.012 us 56.021 us]                                    
                        thrpt:  [4.5697 Melem/s 4.8291 Melem/s 5.1246 Melem/s]
                 change:
                        time:   [-9.8603% -3.6168% +3.6703%] (p = 0.31 > 0.05)
                        thrpt:  [-3.5403% +3.7526% +10.939%]
                        No change in performance detected.

spsc/Block/32B/1x1/1    time:   [92.707 us 98.294 us 103.02 us]                                 
                        thrpt:  [2.4851 Melem/s 2.6044 Melem/s 2.7614 Melem/s]
                 change:
                        time:   [-13.073% -5.2960% +2.5130%] (p = 0.21 > 0.05)
                        thrpt:  [-2.4514% +5.5922% +15.039%]
                        No change in performance detected.
spsc/Block/32B/1x1/2    time:   [79.525 us 87.271 us 94.110 us]                                 
                        thrpt:  [2.7202 Melem/s 2.9334 Melem/s 3.2191 Melem/s]
                 change:
                        time:   [-18.141% -8.7754% +0.3419%] (p = 0.07 > 0.05)
                        thrpt:  [-0.3407% +9.6196% +22.162%]
                        No change in performance detected.

[brunocodutra]

Two test cases failed on CI, mpmc on several runs and drops once. The former looks like an issue with the test itself, I didn't account for the fact one of the threads may never successfully pop an element. The latter got me a bit worried. I'll have a look later today.

[brunocodutra]

Ok, so it turns out both test cases that failed had bugs in them, which are now hopefully fixed.

[brunocodutra]

Something just occurred to me, instead of a new queue type, this could simply be a second flavor of push on ArrayQueue, perhaps push_displacing as suggested on #680 push_or_swap.

[brunocodutra]

Anything missing, that I could work on?

[taiki-e]

Thanks for the PR! The implementation looks good to me.

[taiki-e]

bors r+

[bors]

Build succeeded:

• ci

[taiki-e]

Published in crossbeam-queue 0.3.5.