proc_macro: use crossbeam channels for the proc_macro cross-thread bridge

This is done by having the crossbeam dependency inserted into the
proc_macro server code from the server side, to avoid adding a
dependency to proc_macro.

In addition, this introduces a -Z command-line option which will switch
rustc to run proc-macros using this cross-thread executor. With the
changes to the bridge in #98186, #98187, #98188 and #98189, the
performance of the executor should be much closer to same-thread
execution.

In local testing, the crossbeam executor was substantially more
performant than either of the two existing CrossThread strategies, so
they have been removed to keep things simple.

Author: mystor

proc_macro/src/bridge/server.rs

@@ -2,6 +2,8 @@
 
 use super::*;
 
+use std::marker::PhantomData;
+
 // FIXME(eddyb) generate the definition of `HandleStore` in `server.rs`.
 use super::client::HandleStore;
 
@@ -143,6 +145,41 @@ pub trait ExecutionStrategy {
     ) -> Buffer;
 }
 
+pub struct MaybeCrossThread<P> {
+    cross_thread: bool,
+    marker: PhantomData<P>,
+}
+
+impl<P> MaybeCrossThread<P> {
+    pub const fn new(cross_thread: bool) -> Self {
+        MaybeCrossThread { cross_thread, marker: PhantomData }
+    }
+}
+
+impl<P> ExecutionStrategy for MaybeCrossThread<P>
+where
+    P: MessagePipe<Buffer> + Send + 'static,
+{
+    fn run_bridge_and_client(
+        &self,
+        dispatcher: &mut impl DispatcherTrait,
+        input: Buffer,
+        run_client: extern "C" fn(BridgeConfig<'_>) -> Buffer,
+        force_show_panics: bool,
+    ) -> Buffer {
+        if self.cross_thread {
+            <CrossThread<P>>::new().run_bridge_and_client(
+                dispatcher,
+                input,
+                run_client,
+                force_show_panics,
+            )
+        } else {
+            SameThread.run_bridge_and_client(dispatcher, input, run_client, force_show_panics)
+        }
+    }
+}
+
 pub struct SameThread;
 
 impl ExecutionStrategy for SameThread {
@@ -164,28 +201,31 @@ impl ExecutionStrategy for SameThread {
     }
 }
 
-// NOTE(eddyb) Two implementations are provided, the second one is a bit
-// faster but neither is anywhere near as fast as same-thread execution.
+pub struct CrossThread<P>(PhantomData<P>);
 
-pub struct CrossThread1;
+impl<P> CrossThread<P> {
+    pub const fn new() -> Self {
+        CrossThread(PhantomData)
+    }
+}
 
-impl ExecutionStrategy for CrossThread1 {
+impl<P> ExecutionStrategy for CrossThread<P>
+where
+    P: MessagePipe<Buffer> + Send + 'static,
+{
     fn run_bridge_and_client(
         &self,
         dispatcher: &mut impl DispatcherTrait,
         input: Buffer,
         run_client: extern "C" fn(BridgeConfig<'_>) -> Buffer,
         force_show_panics: bool,
     ) -> Buffer {
-        use std::sync::mpsc::channel;
-
-        let (req_tx, req_rx) = channel();
-        let (res_tx, res_rx) = channel();
+        let (mut server, mut client) = P::new();
 
         let join_handle = thread::spawn(move || {
-            let mut dispatch = |buf| {
-                req_tx.send(buf).unwrap();
-                res_rx.recv().unwrap()
+            let mut dispatch = |b: Buffer| -> Buffer {
+                client.send(b);
+                client.recv().expect("server died while client waiting for reply")
             };
 
             run_client(BridgeConfig {
@@ -196,75 +236,27 @@ impl ExecutionStrategy for CrossThread1 {
             })
         });
 
-        for b in req_rx {
-            res_tx.send(dispatcher.dispatch(b)).unwrap();
+        while let Some(b) = server.recv() {
+            server.send(dispatcher.dispatch(b));
         }
 
         join_handle.join().unwrap()
     }
 }
 
-pub struct CrossThread2;
+/// A message pipe used for communicating between server and client threads.
+pub trait MessagePipe<T>: Sized {
+    /// Create a new pair of endpoints for the message pipe.
+    fn new() -> (Self, Self);
 
-impl ExecutionStrategy for CrossThread2 {
-    fn run_bridge_and_client(
-        &self,
-        dispatcher: &mut impl DispatcherTrait,
-        input: Buffer,
-        run_client: extern "C" fn(BridgeConfig<'_>) -> Buffer,
-        force_show_panics: bool,
-    ) -> Buffer {
-        use std::sync::{Arc, Mutex};
-
-        enum State<T> {
-            Req(T),
-            Res(T),
-        }
-
-        let mut state = Arc::new(Mutex::new(State::Res(Buffer::new())));
-
-        let server_thread = thread::current();
-        let state2 = state.clone();
-        let join_handle = thread::spawn(move || {
-            let mut dispatch = |b| {
-                *state2.lock().unwrap() = State::Req(b);
-                server_thread.unpark();
-                loop {
-                    thread::park();
-                    if let State::Res(b) = &mut *state2.lock().unwrap() {
-                        break b.take();
-                    }
-                }
-            };
-
-            let r = run_client(BridgeConfig {
-                input,
-                dispatch: (&mut dispatch).into(),
-                force_show_panics,
-                _marker: marker::PhantomData,
-            });
-
-            // Wake up the server so it can exit the dispatch loop.
-            drop(state2);
-            server_thread.unpark();
-
-            r
-        });
-
-        // Check whether `state2` was dropped, to know when to stop.
-        while Arc::get_mut(&mut state).is_none() {
-            thread::park();
-            let mut b = match &mut *state.lock().unwrap() {
-                State::Req(b) => b.take(),
-                _ => continue,
-            };
-            b = dispatcher.dispatch(b.take());
-            *state.lock().unwrap() = State::Res(b);
-            join_handle.thread().unpark();
-        }
+    /// Send a message to the other endpoint of this pipe.
+    fn send(&mut self, value: T);
 
-        join_handle.join().unwrap()
-    }
+    /// Receive a message from the other endpoint of this pipe.
+    ///
+    /// Returns `None` if the other end of the pipe has been destroyed, and no
+    /// message was received.
+    fn recv(&mut self) -> Option<T>;
 }
 
 fn run_server<