Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add fn force_push to ArrayQueue #789

Merged
merged 1 commit into from Mar 9, 2022
Merged

Add fn force_push to ArrayQueue #789

merged 1 commit into from Mar 9, 2022

Conversation

brunocodutra
Copy link
Contributor

@brunocodutra brunocodutra commented Feb 15, 2022

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
Copy link
Contributor Author

brunocodutra commented Feb 16, 2022

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
Copy link
Contributor Author

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

@brunocodutra
Copy link
Contributor Author

brunocodutra commented Feb 17, 2022

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 brunocodutra changed the title Implement RingQueue as a MPMC circular buffer Add fn push_or_swap to ArrayQueue Feb 17, 2022
brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Feb 20, 2022
brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Feb 20, 2022
brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Feb 20, 2022
brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Feb 24, 2022
@brunocodutra
Copy link
Contributor Author

Anything missing, that I could work on?

brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Feb 26, 2022
brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Feb 26, 2022
Copy link
Member

@taiki-e taiki-e left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

crossbeam-queue/src/array_queue.rs Outdated Show resolved Hide resolved
crossbeam-queue/src/array_queue.rs Outdated Show resolved Hide resolved
force_push makes it possible for ArrayQueue to be used as a ring-buffer.
.is_ok()
{
// Move the tail.
self.tail.store(new_tail, Ordering::SeqCst);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this needs to be CAS, as the tail may have changed.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's actually not necessary. Up until the point where a thread attempts to update the head, it has only read state that depends on tail, then moves the head only if it lags by exactly one lap. Immediately after the head is updated, tail has not changed, so all other threads that are concurrently attempting to push will invariably fail the CAS on head, because it no longer lags the tail by exactly one lap. This means that until the thread that moved the head updates the tail, no other thread attempting to push can make progress and update the tail. As for threads attempting to pop, they never move the tail either, so we don't need to worry about the tail updating underneath us here.

Does that make sense?

@taiki-e taiki-e changed the title Add fn push_or_swap to ArrayQueue Add fn force_push to ArrayQueue Mar 4, 2022
taiki-e
taiki-e approved these changes Mar 9, 2022
Copy link
Member

@taiki-e taiki-e left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bors r+

@bors
Copy link
Contributor

bors bot commented Mar 9, 2022

Build succeeded:

@bors bors bot merged commit 0081fcc into crossbeam-rs:master Mar 9, 2022
22 checks passed
bors bot added a commit that referenced this pull request Mar 15, 2022
800: Prepare for the next release r=taiki-e a=taiki-e

- crossbeam-channel 0.5.2 -> 0.5.3
  - Fix panic on very large timeout. (#798)
- crossbeam-epoch 0.9.7 -> 0.9.8
  - Make `Atomic::null()` const function at 1.61+. (#797)
- crossbeam-queue 0.3.4 -> 0.3.5
  - Add `ArrayQueue::force_push`. (#789)
- crossbeam-utils 0.8.7 -> 0.8.8
  - Fix a bug when unstable `loom` support is enabled. (#787)


Co-authored-by: Taiki Endo <te316e89@gmail.com>
@taiki-e
Copy link
Member

taiki-e commented Mar 15, 2022

Published in crossbeam-queue 0.3.5.

brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Apr 12, 2022
this is possible now that crossbeam-rs/crossbeam#789 has been released with v0.3.5.
brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Apr 12, 2022
this is possible now that crossbeam-rs/crossbeam#789 has been released with v0.3.5.
brunocodutra added a commit to brunocodutra/ring-channel that referenced this pull request Apr 14, 2022
this is possible now that crossbeam-rs/crossbeam#789 has been released with v0.3.5.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Development

Successfully merging this pull request may close these issues.

ArrayQueue: is it possible to push and pop and the same time?
2 participants