runtime/rpc: use Arc to reduce copy

We do busy polling for some time before doing await, for small
allocations we could avoid the context switching and reduce the latency.

default.nix

@@ -15,7 +15,7 @@ let
       version = "0.1.0";
 
       src = ./src;
-      cargoSha256 = "1r8yjzixkknivawi286iyjjhaf5q8ll3a53q54dc9m9dhx20358b";
+      cargoSha256 = "1q5s193qcqjcvln7zphminfdg8by3dr05zq242965bmh0q46246v";
 
       nativeBuildInputs = [
         pkgs.gnumake

src/Cargo.lock

@@ -187,7 +187,7 @@ dependencies = [
 [[package]]
 name = "libasync"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#157439bc88cbb18bb40009428acf1fdee800e32e"
+source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#a116142f6346193f272528e68c09af5eda771c0d"
 dependencies = [
  "embedded-hal",
  "libcortex_a9",
@@ -199,7 +199,7 @@ dependencies = [
 [[package]]
 name = "libboard_zynq"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#157439bc88cbb18bb40009428acf1fdee800e32e"
+source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#a116142f6346193f272528e68c09af5eda771c0d"
 dependencies = [
  "bit_field",
  "embedded-hal",
@@ -233,7 +233,7 @@ dependencies = [
 [[package]]
 name = "libcortex_a9"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#157439bc88cbb18bb40009428acf1fdee800e32e"
+source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#a116142f6346193f272528e68c09af5eda771c0d"
 dependencies = [
  "bit_field",
  "libregister",
@@ -249,7 +249,7 @@ checksum = "c7d73b3f436185384286bd8098d17ec07c9a7d2388a6599f824d8502b529702a"
 [[package]]
 name = "libregister"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#157439bc88cbb18bb40009428acf1fdee800e32e"
+source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#a116142f6346193f272528e68c09af5eda771c0d"
 dependencies = [
  "bit_field",
  "vcell",
@@ -259,7 +259,7 @@ dependencies = [
 [[package]]
 name = "libsupport_zynq"
 version = "0.0.0"
-source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#157439bc88cbb18bb40009428acf1fdee800e32e"
+source = "git+https://git.m-labs.hk/M-Labs/zynq-rs.git#a116142f6346193f272528e68c09af5eda771c0d"
 dependencies = [
  "compiler_builtins",
  "libboard_zynq",

src/runtime/src/comms.rs

@@ -17,7 +17,7 @@ use libboard_zynq::{
     },
     timer::GlobalTimer,
 };
-use libcortex_a9::{semaphore::Semaphore, mutex::Mutex};
+use libcortex_a9::{semaphore::Semaphore, mutex::Mutex, sync_channel::{Sender, Receiver}};
 use futures::{select_biased, future::FutureExt};
 use libasync::{smoltcp::{Sockets, TcpStream}, task};
 use libconfig::{Config, net_settings};
@@ -124,6 +124,35 @@ async fn read_string(stream: &TcpStream, max_length: usize) -> Result<String> {
     Ok(String::from_utf8(bytes).map_err(|err| Error::Utf8Error(err.utf8_error()))?)
 }
 
+const RETRY_LIMIT: usize = 100;
+
+async fn fast_send(sender: &mut Sender<'_, kernel::Message>, content: kernel::Message) {
+    let mut content = content;
+    for _ in 0..RETRY_LIMIT {
+        match sender.try_send(content) {
+            Ok(()) => {
+                return
+            },
+            Err(v) => {
+                content = v;
+            }
+        }
+    }
+    sender.async_send(content).await;
+}
+
+async fn fast_recv(receiver: &mut Receiver<'_, kernel::Message>) -> kernel::Message {
+    for _ in 0..RETRY_LIMIT {
+        match receiver.try_recv() {
+            Ok(v) => {
+                return v;
+            },
+            Err(()) => ()
+        }
+    }
+    receiver.async_recv().await
+}
+
 async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kernel::Control>>) -> Result<()> {
     control.borrow_mut().tx.async_send(kernel::Message::StartRequest).await;
     loop {
@@ -143,7 +172,7 @@ async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kern
                     match host_request {
                         Request::RPCReply => {
                             let tag = read_bytes(stream, 512).await?;
-                            let slot = match control.borrow_mut().rx.async_recv().await {
+                            let slot = match fast_recv(&mut control.borrow_mut().rx).await {
                                 kernel::Message::RpcRecvRequest(slot) => slot,
                                 other => panic!("expected root value slot from core1, not {:?}", other),
                             };
@@ -156,8 +185,8 @@ async fn handle_run_kernel(stream: Option<&TcpStream>, control: &Rc<RefCell<kern
                                         0 as *mut ()
                                     } else {
                                         let mut control = control.borrow_mut();
-                                        control.tx.async_send(kernel::Message::RpcRecvReply(Ok(size))).await;
+                                        fast_send(&mut control.tx, kernel::Message::RpcRecvReply(Ok(size))).await;
-                                        match control.rx.async_recv().await {
+                                        match fast_recv(&mut control.rx).await {
                                             kernel::Message::RpcRecvRequest(slot) => slot,
                                             other => panic!("expected nested value slot from kernel CPU, not {:?}", other),
                                         }

src/runtime/src/kernel/mod.rs

@@ -1,5 +1,5 @@
 use core::ptr;
-use alloc::{vec::Vec, string::String};
+use alloc::{vec::Vec, string::String, sync::Arc};
 
 use libcortex_a9::{mutex::Mutex, sync_channel, semaphore::Semaphore};
 use crate::eh_artiq;
@@ -32,7 +32,7 @@ pub enum Message {
     StartRequest,
     KernelFinished,
     KernelException(&'static eh_artiq::Exception<'static>, &'static [usize]),
-    RpcSend { is_async: bool, data: Vec<u8> },
+    RpcSend { is_async: bool, data: Arc<Vec<u8>> },
     RpcRecvRequest(*mut ()),
     RpcRecvReply(Result<usize, RPCException>),
 

src/runtime/src/kernel/rpc.rs

@@ -1,6 +1,6 @@
 //! Kernel-side RPC API
 
-use alloc::vec::Vec;
+use alloc::{vec::Vec, sync::Arc};
 use cslice::{CSlice, AsCSlice};
 
 use crate::eh_artiq;
@@ -14,7 +14,7 @@ fn rpc_send_common(is_async: bool, service: u32, tag: &CSlice<u8>, data: *const
     let core1_tx = unsafe { KERNEL_CHANNEL_1TO0.as_mut().unwrap() };
     let mut buffer = Vec::<u8>::new();
     send_args(&mut buffer, service, tag.as_ref(), data).expect("RPC encoding failed");
-    core1_tx.send(Message::RpcSend { is_async, data: buffer });
+    core1_tx.send(Message::RpcSend { is_async, data: Arc::new(buffer) });
 }
 
 pub extern fn rpc_send(service: u32, tag: &CSlice<u8>, data: *const *const ()) {