about summary refs log tree commit diff
path: root/library/proc_macro/src
diff options
context:
space:
mode:
authorNika Layzell <nika@thelayzells.com>2022-06-18 14:15:03 -0400
committerNika Layzell <nika@thelayzells.com>2022-07-29 17:38:12 -0400
commit6d1650fe45e675cd976a5401067606de325d8ae8 (patch)
treec27afa70251e54a698cdb61da88f07acf278df93 /library/proc_macro/src
parent2f847b81a0d8633f200f2c2269c1c43fe9e7def3 (diff)
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.
Diffstat (limited to 'library/proc_macro/src')
-rw-r--r--library/proc_macro/src/bridge/server.rs136
1 files changed, 64 insertions, 72 deletions
diff --git a/library/proc_macro/src/bridge/server.rs b/library/proc_macro/src/bridge/server.rs
index d46e325951d..e068ec60b6a 100644
--- a/library/proc_macro/src/bridge/server.rs
+++ b/library/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,12 +201,18 @@ 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,
@@ -177,15 +220,12 @@ impl ExecutionStrategy for CrossThread1 {
         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<