ACP: Export MPMC APIs

[obeis]

Proposal

Problem statement

The standard library currently provides no concurrent queue that permits multiple consumers. Given that we no have scoped threads, a multi-consumer concurrent queue is the last missing piece to be able to implement basic parallelism via "fill a queue with work to be done, then have N workers do the work".

The standard library already contains an implementation of an mpmc queue, ever since crossbeam's queue was ported over as the underlying implementation for our standard mpsc queue. However, so far this extra power is currently not exposed to users. If we're anyway spending the maintenance effort on such a queue, I think we should let our users benefit as well. :)

Motivating examples or use cases

For instance, the formatting in bootstrap is currently using a pretty complicated "poor man's async" scheme to run mutliple instances of rustfmt concurrently when formatting many files. However it anyway limits this to 2*available_parallelism many workers, so with an MPMC queue, a much simpler implementation with one thread per worker would be possible. In our pretty similar code for ./miri fmt we didn't bother with the manual async so formatting is just unnecessarily sequential.

The ui_test crate just imports crossbeam-channel for a similar situation (walking the file system and then processing things in parallel); that dependency could be entirely avoided if there was an MPMC queue in std.

Solution sketch

Shared usage:

#![feature(mpmc_channel)]

use std::thread;
use std::sync::mpmc::channel;

// Create a shared channel that can be sent along from many threads
// where tx is the sending half (tx for transmission), and rx is the receiving
// half (rx for receiving).
let (tx, rx) = channel();
for i in 0..10 {
    let tx = tx.clone();
    thread::spawn(move || {
        tx.send(i).unwrap();
    });
}

for _ in 0..5 {
    let rx1 = rx.clone();
    let rx2 = rx.clone();
    thread::spawn(move || {
        let j = rx1.recv().unwrap();
        assert!(0 <= j && j < 10);
    });
    thread::spawn(move || {
        let j = rx2.recv().unwrap();
        assert!(0 <= j && j < 10);
    });
}

Also, we will provide iterator functionality similar to mpsc (IntoIter, Iter, TryIter).

• The new Receiver type will implement the Clone, Send, and Sync traits.

• What do we do with the mpsc module? I think we can deprecate the mpsc module after stabilizing mpmc.

Alternatives

We could do nothing, and ask people to depend on crossbeam when they need an mpmc queue.

Links and related work

Go's native channels are MPMC. (They also allow receiving on multiple channels at once, but that is very complicated to implement and not part of this proposal. It seems orthogonal to the single- vs multiple-consumer question: our MPSC queues don't allow a receiver to receive on multiple queues at once, and neither will our MPMC queues.)

What happens now?

This issue contains an API change proposal (or ACP) and is part of the libs-api team feature lifecycle. Once this issue is filed, the libs-api team will review open proposals as capability becomes available. Current response times do not have a clear estimate, but may be up to several months.

Possible responses

The libs team may respond in various different ways. First, the team will consider the problem (this doesn't require any concrete solution or alternatives to have been proposed):

• We think this problem seems worth solving, and the standard library might be the right place to solve it.
• We think that this probably doesn't belong in the standard library.

Second, if there's a concrete solution:

• We think this specific solution looks roughly right, approved, you or someone else should implement this. (Further review will still happen on the subsequent implementation PR.)
• We're not sure this is the right solution, and the alternatives or other materials don't give us enough information to be sure about that. Here are some questions we have that aren't answered, or rough ideas about alternatives we'd want to see discussed.

[the8472]

This was already approved in #384 and doesn't need another ACP.

For instance, the formatting in bootstrap is currently using a pretty complicated "poor man's async" scheme to run mutliple instances of rustfmt concurrently when formatting many files.

That's a bit of red herring I think. The code there is about spawning processes and awaiting their completion. That's not the same as an in-process worker pool.

[obeis]

we need a more detailed ACP because there several questions

https://github.com/rust-lang/libs-team/issues/384#issuecomment-2135775934

[the8472]

Hah, I forgot about that part.

[Amanieu]

https://github.com/rust-lang/rust/pull/126839 has been merged so I don't think this is needed any more?

[obeis]

Completed by rust-lang/rust#126839

[obeis]

@rustbot labels +ACP-accepted

[rustbot]

Error: Label ACP-accepted can only be set by Rust team members

Please file an issue on GitHub at triagebot if there's a problem with this bot, or reach out on #t-infra on Zulip.