Add std::sync::mpsc::Receiver::recv_deadline()
Essentially renames recv_max_until to recv_deadline (mostly copying recv_timeout
documentation). This function is useful to avoid the often unnecessary call to
Instant::now in recv_timeout (e.g. when the user already has a deadline). A
concrete example would be something along those lines:
```rust
use std::sync::mpsc::Receiver;
use std::time::{Duration, Instant};
/// Reads a batch of elements
///
/// Returns as soon as `max_size` elements have been received or `timeout` expires.
fn recv_batch_timeout<T>(receiver: &Receiver<T>, timeout: Duration, max_size: usize) -> Vec<T> {
recv_batch_deadline(receiver, Instant::now() + timeout, max_size)
}
/// Reads a batch of elements
///
/// Returns as soon as `max_size` elements have been received or `deadline` is reached.
fn recv_batch_deadline<T>(receiver: &Receiver<T>, deadline: Instant, max_size: usize) -> Vec<T> {
let mut result = Vec::new();
while let Ok(x) = receiver.recv_deadline(deadline) {
result.push(x);
if result.len() == max_size {
break;
}
}
result
}
```
Essentially renames recv_max_until to recv_deadline (mostly copying recv_timeout
documentation). This function is useful to avoid the often unnecessary call to
Instant::now in recv_timeout (e.g. when the user already has a deadline). A
concrete example would be something along those lines:
```rust
use std::sync::mpsc::Receiver;
use std::time::{Duration, Instant};
/// Reads a batch of elements
///
/// Returns as soon as `max_size` elements have been received or `timeout` expires.
fn recv_batch_timeout<T>(receiver: &Receiver<T>, timeout: Duration, max_size: usize) -> Vec<T> {
recv_batch_deadline(receiver, Instant::now() + timeout, max_size)
}
/// Reads a batch of elements
///
/// Returns as soon as `max_size` elements have been received or `deadline` is reached.
fn recv_batch_deadline<T>(receiver: &Receiver<T>, deadline: Instant, max_size: usize) -> Vec<T> {
let mut result = Vec::new();
while let Ok(x) = receiver.recv_deadline(deadline) {
result.push(x);
if result.len() == max_size {
break;
}
}
result
}
```
Improve performance of spsc_queue and stream.
This PR makes two main changes:
1. It switches the `spsc_queue` node caching strategy from keeping a shared
counter of the number of nodes in the cache to keeping a consumer only counter
of the number of node eligible to be cached.
2. It separates the consumer and producers fields of `spsc_queue` and `stream` into
a producer cache line and consumer cache line.
Overall, it speeds up `mpsc` in `spsc` mode by 2-10x.
Variance is higher than I'd like (that 2-10x speedup is on one benchmark), I believe this is due to the drop check in `send` (`fn stream::Queue::send:107`). I think this check can be combined with the sleep detection code into a version which only uses 1 shared variable, and only one atomic access per `send`, but I haven't looked through the select implementation enough to be sure.
The code currently assumes a cache line size of 64 bytes. I added a CacheAligned newtype in `mpsc` which I expect to reuse for `shared`. It doesn't really belong there, it would probably be best put in `core::sync::atomic`, but putting it in `core` would involve making it public, which I thought would require an RFC.
Benchmark runner is [here](3eca46279c/shootout), benchmarks [here](3eca46279c/queue_bench/src/lib.rs (L170-L293)).
Fixes#44512.
This commit makes two main changes.
1. It switches the spsc_queue node caching strategy from keeping a shared
counter of the number of nodes in the cache to keeping a consumer only counter
of the number of node eligible to be cached.
2. It separate the consumer and producers fields of spsc_queue and stream into
a producer cache line and consumer cache line.
Decompose Adjustment into smaller steps and remove the method map.
The method map held method callee information for:
* actual method calls (`x.f(...)`)
* overloaded unary, binary, indexing and call operators
* *every overloaded deref adjustment* (many can exist for each expression)
That last one was a historical ~~accident~~ hack, and part of the motivation for this PR, along with:
* a desire to compose adjustments more freely
* containing the autoderef logic better to avoid mutation within an inference snapshot
* not creating `TyFnDef` types which are incompatible with the original one
* i.e. we used to take a`TyFnDef`'s `for<'a> &'a T -> &'a U` signature and instantiate `'a` using a region inference variable, *then* package the resulting `&'b T -> &'b U` signature in another `TyFnDef`, while keeping *the same* `DefId` and `Substs`
* to fix#3548 by explicitly writing autorefs for the RHS of comparison operators
Individual commits tell their own story, of "atomic" changes avoiding breaking semantics.
Future work based on this PR could include:
* removing the signature from `TyFnDef`, now that it's always "canonical"
* some questions of variance remain, as subtyping *still* treats the signature differently
* moving part of the typeck logic for methods, autoderef and coercion into `rustc::traits`
* allowing LUB coercions (joining multiple expressions) to "stack up" many adjustments
* transitive coercions (e.g. reify or unsize after multiple steps of autoderef)
r? @nikomatsakis
Correct some stability versions
These were found by running tidy on stable versions of rust and finding
features stabilised with the wrong version numbers.
Closes#37915
This commit enhances documentation with several links and
fixes an error in the `sync_channel` documentation as well:
`send` doesn't panic when the senders are all disconnected
Add a method to the mpsc::Receiver for producing a non-blocking iterator
Currently, the `mpsc::Receiver` offers methods for receiving values in both blocking (`recv`) and non-blocking (`try_recv`) flavours. However only blocking iteration over values is supported. This PR adds a non-blocking iterator to complement the `try_recv` method, just as the blocking iterator complements the `recv` method.
Use-case
-------------
I predominantly use rust in my work on real-time systems and in particular real-time audio generation/processing. I use `mpsc::channel`s to communicate between threads in a purely non-blocking manner. I.e. I might send messages from the GUI thread to the audio thread to update the state of the dsp-graph, or from the audio thread to the GUI thread to display the RMS of each node. These are just a couple examples (I'm probably using 30+ channels across my various projects). I almost exclusively use the `mpsc::Receiver::try_recv` method to avoid blocking any of the real-time threads and causing unwanted glitching/stuttering. Now that I mention it, I can't think of a single time that I personally have used the `recv` method (though I can of course see why it would be useful, and perhaps the common case for many people).
As a result of this experience, I can't help but feel there is a large hole in the `Receiver` API.
| blocking | non-blocking |
|------------|--------------------|
| `recv` | `try_recv` |
| `iter` | 🙀 |
For the most part, I've been working around this using `while let Ok(v) = r.try_recv() { ... }`, however as nice as this is, it is clearly no match for the Iterator API.
As an example, in the majority of my channel use cases I only want to check for *n* number of messages before breaking from the loop so that I don't miss the audio IO callback or hog the GUI thread for too long when an unexpectedly large number of messages are sent. Currently, I have to write something like this:
```rust
let mut take = 100;
while let Ok(msg) = rx.try_recv() {
// Do stuff with msg
if take == 0 {
break;
}
take -= 1;
}
```
or wrap the `try_recv` call in a `Range<usize>`/`FilterMap` iterator combo.
On the other hand, this PR would allow for the following:
```rust
for msg in rx.try_iter().take(100) {
// Do stuff with msg
}
```
I imagine this might also be useful to game devs, embedded or anyone doing message passing across real-time threads.
