Add multicore client rackscale integration test

Author: Erika Hunhoff

kernel/src/arch/x86_64/mod.rs

@@ -414,7 +414,8 @@ fn _start(argc: isize, _argv: *const *const u8) -> isize {
         .get()
         .map_or(false, |c| c.mode == crate::cmdline::Mode::Client)
     {
-        let _ = spin::lazy::Lazy::force(&rackscale::client::RPC_CLIENT);
+        // Force client instantiation
+        let _ = rackscale::client::RPC_CLIENT.clone();
     }
 
     // Bring up the rest of the system (needs topology, APIC, and global memory)

kernel/src/arch/x86_64/rackscale/client.rs

@@ -2,11 +2,13 @@
 // SPDX-License-Identifier: Apache-2.0 OR MIT
 
 use alloc::boxed::Box;
+use alloc::sync::Arc;
 use hashbrown::HashMap;
 use lazy_static::lazy_static;
 use spin::{Lazy, Mutex};
 
 use rpc::api::{RPCClient, RPCHandler, RegistrationHandler};
+use rpc::client::Client;
 use rpc::rpc::{ClientId, RPCError, RPCHeader};
 
 use crate::cmdline::Transport;
@@ -17,31 +19,31 @@ use crate::transport::shmem::SHMEM_REGION;
 /// A handle to an RPC client
 ///
 /// This is used to send requests to a remote control-plane.
-#[thread_local]
-pub(crate) static RPC_CLIENT: Lazy<Mutex<Box<dyn RPCClient>>> = Lazy::new(|| {
+lazy_static! {
+pub(crate) static ref RPC_CLIENT: Arc<Mutex<Box<Client>>> =
     // Create network stack and instantiate RPC Client
-    return if crate::CMDLINE
+    if crate::CMDLINE
         .get()
         .map_or(false, |c| c.transport == Transport::Ethernet)
     {
         // To support alloc_phys, client needs shared memory to be mapped
         lazy_static::initialize(&SHMEM_REGION);
 
-        Mutex::new(
+        Arc::new(Mutex::new(
             crate::transport::ethernet::init_ethernet_rpc(
                 smoltcp::wire::IpAddress::v4(172, 31, 0, 11),
                 6970,
                 true,
             )
             .expect("Failed to initialize ethernet RPC"),
-        )
+        ))
     } else {
         // Default is Shmem, even if transport unspecified
-        Mutex::new(
+        Arc::new(Mutex::new(
             crate::transport::shmem::init_shmem_rpc(true).expect("Failed to initialize shmem RPC"),
-        )
+        ))
     };
-});
+}
 
 // Mapping between local frame IDs and remote memory address space ID (node id, currently).
 lazy_static! {

kernel/src/arch/x86_64/rackscale/registration.rs

@@ -63,11 +63,10 @@ pub(crate) fn initialize_client(
     if send_client_data {
         let (affinity_shmem_offset, affinity_shmem_size) = get_affinity_shmem();
 
-        // TODO: calculate cores correctly
         let req = ClientRegistrationRequest {
             affinity_shmem_offset,
             affinity_shmem_size,
-            num_cores: 4,
+            num_cores: atopology::MACHINE_TOPOLOGY.num_threads() as u64,
         };
         let mut req_data = [0u8; core::mem::size_of::<ClientRegistrationRequest>()];
         unsafe { encode(&req, &mut (&mut req_data).as_mut()) }.unwrap();

kernel/tests/s06_rackscale_tests.rs

@@ -10,13 +10,14 @@
 //! * `s06_*`: Rackscale (distributed) tests
 
 use rexpect::errors::*;
+use rexpect::process::signal::SIGTERM;
 use rexpect::process::wait::WaitStatus;
 
-use testutils::builder::BuildArgs;
+use testutils::builder::{BuildArgs, Machine};
 use testutils::helpers::{
     setup_network, setup_shmem, spawn_dcm, spawn_nrk, SHMEM_PATH, SHMEM_SIZE,
 };
-use testutils::runner_args::{check_for_successful_exit, RunnerArgs};
+use testutils::runner_args::{check_for_successful_exit, wait_for_sigterm, RunnerArgs};
 
 #[cfg(not(feature = "baremetal"))]
 #[test]
@@ -479,3 +480,117 @@ fn s06_rackscale_shmem_multiinstance() {
 
     let _ignore = remove_file(SHMEM_PATH);
 }
+
+#[cfg(not(feature = "baremetal"))]
+#[test]
+fn s06_rackscale_shmem_userspace_multicore_test() {
+    rackscale_userspace_multicore_test(true);
+}
+
+#[cfg(not(feature = "baremetal"))]
+#[test]
+fn s06_rackscale_ethernet_userspace_multicore_test() {
+    rackscale_userspace_multicore_test(false);
+}
+
+#[cfg(not(feature = "baremetal"))]
+fn rackscale_userspace_multicore_test(is_shmem: bool) {
+    use std::fs::remove_file;
+    use std::sync::Arc;
+    use std::thread::sleep;
+    use std::time::Duration;
+
+    // Setup ivshmem file
+    setup_shmem(SHMEM_PATH, SHMEM_SIZE);
+
+    setup_network(2);
+    let timeout = 30_000;
+
+    let machine = Machine::determine();
+    let client_num_cores: usize = machine.max_cores() - 1;
+
+    // Create build for both controller and client
+    let build = Arc::new(
+        BuildArgs::default()
+            .module("init")
+            .user_feature("test-scheduler-smp")
+            .kernel_feature("shmem")
+            .kernel_feature("ethernet")
+            .kernel_feature("rackscale")
+            .release()
+            .build(),
+    );
+
+    // Run DCM and controller in separate thread
+    let build1 = build.clone();
+    let controller = std::thread::spawn(move || {
+        let controller_cmd = if is_shmem {
+            "mode=controller transport=shmem"
+        } else {
+            "mode=controller transport=ethernet"
+        };
+        let cmdline_controller = RunnerArgs::new_with_build("userspace-smp", &build1)
+            .timeout(timeout)
+            .cmd(controller_cmd)
+            .shmem_size(SHMEM_SIZE as usize)
+            .shmem_path(SHMEM_PATH)
+            .tap("tap0")
+            .no_network_setup()
+            .workers(2)
+            .use_vmxnet3();
+
+        let mut output = String::new();
+        let mut qemu_run = || -> Result<WaitStatus> {
+            let mut dcm = spawn_dcm(1, timeout)?;
+            let mut p = spawn_nrk(&cmdline_controller)?;
+
+            //output += p.exp_string("Finished sending requests!")?.as_str();
+            output += p.exp_eof()?.as_str();
+
+            dcm.send_control('c')?;
+            p.process.exit()
+        };
+
+        let _ignore = qemu_run();
+    });
+
+    // Run client in separate thead. Wait a bit to make sure DCM and controller started
+    let build2 = build.clone();
+    let _client = std::thread::spawn(move || {
+        sleep(Duration::from_millis(5_000));
+        let client_cmd = if is_shmem {
+            "mode=client transport=shmem"
+        } else {
+            "mode=client transport=ethernet"
+        };
+        let cmdline_client = RunnerArgs::new_with_build("userspace-smp", &build2)
+            .timeout(timeout)
+            .cmd(client_cmd)
+            .shmem_size(SHMEM_SIZE as usize)
+            .shmem_path(SHMEM_PATH)
+            .tap("tap2")
+            .no_network_setup()
+            .workers(2)
+            .cores(client_num_cores)
+            .memory(4096)
+            .use_vmxnet3();
+
+        let mut output = String::new();
+        let mut qemu_run = || -> Result<WaitStatus> {
+            let mut p = spawn_nrk(&cmdline_client)?;
+
+            for _i in 0..client_num_cores {
+                let r = p.exp_regex(r#"init: Hello from core (\d+)"#)?;
+                output += r.0.as_str();
+                output += r.1.as_str();
+            }
+            p.process.kill(SIGTERM)
+        };
+
+        wait_for_sigterm(&cmdline_client, qemu_run(), output);
+    });
+
+    controller.join().unwrap();
+
+    let _ignore = remove_file(SHMEM_PATH);
+}

lib/rpc/tests/integration-test.rs

@@ -74,3 +74,87 @@ fn test_client_server_shmem_transport() {
         .unwrap();
     assert_eq!(send_data, recv_data);
 }
+
+#[test]
+fn test_client_shmem_multithread() {
+    use spin::Mutex;
+    use std::alloc::{alloc, Layout};
+    use std::sync::Arc;
+    use std::thread;
+
+    use rpc::api::{RPCClient, RPCHandler, RPCServer, RegistrationHandler};
+    use rpc::client::Client;
+    use rpc::rpc::{ClientId, RPCError, RPCHeader};
+    use rpc::server::Server;
+    use rpc::transport::shmem::allocator::ShmemAllocator;
+    use rpc::transport::shmem::{Queue, Receiver, Sender};
+    use rpc::transport::ShmemTransport;
+
+    let alloc_size = 128 * 1024 * 1024;
+    let alloc = (unsafe { alloc(Layout::from_size_align(alloc_size, 1).expect("Layout failed")) }
+        as *mut u8) as u64;
+
+    let allocator = ShmemAllocator::new(alloc, alloc_size as u64);
+    // Create transport
+    let server_to_client_queue = Arc::new(Queue::with_capacity_in(true, 32, &allocator).unwrap());
+    let client_to_server_queue = Arc::new(Queue::with_capacity_in(true, 32, &allocator).unwrap());
+
+    let server_sender = Sender::with_shared_queue(server_to_client_queue.clone());
+    let server_receiver = Receiver::with_shared_queue(client_to_server_queue.clone());
+    let server_transport = ShmemTransport::new(server_receiver, server_sender);
+
+    thread::spawn(move || {
+        // Create a server
+        let rpc_server_transport = Box::new(server_transport);
+        let mut server = Server::new(rpc_server_transport);
+
+        // Register an echo RPC
+        fn echo_rpc_handler(_hdr: &mut RPCHeader, _payload: &mut [u8]) -> Result<(), RPCError> {
+            Ok(())
+        }
+        const ECHO_HANDLER: RPCHandler = echo_rpc_handler;
+        server.register(1, &ECHO_HANDLER).unwrap();
+
+        // Accept a client
+        fn register_client(
+            _hdr: &mut RPCHeader,
+            _payload: &mut [u8],
+        ) -> Result<ClientId, RPCError> {
+            Ok(0)
+        }
+        pub const CLIENT_REGISTRAR: RegistrationHandler = register_client;
+        server.add_client(&CLIENT_REGISTRAR).unwrap();
+
+        // Run the server
+        server.run_server().unwrap();
+    });
+
+    // Create a client
+    let client_sender = Sender::with_shared_queue(client_to_server_queue.clone());
+    let client_receiver = Receiver::with_shared_queue(server_to_client_queue.clone());
+    let client_transport = ShmemTransport::new(client_receiver, client_sender);
+    let rpc_client_transport = Box::new(client_transport);
+    let client = Arc::new(Mutex::new(Client::new(rpc_client_transport)));
+
+    {
+        // Connect to server
+        let my_client = Arc::clone(&client);
+        my_client.lock().connect(&[]).unwrap();
+    }
+
+    for _ in 1..10 {
+        let my_client = Arc::clone(&client);
+        thread::spawn(move || {
+            // Setup for RPCs
+            let send_data = [1u8; 40];
+            let mut recv_data = [0u8; 40];
+
+            // Test echo
+            my_client
+                .lock()
+                .call(0, 1, &[&send_data], &mut [&mut recv_data])
+                .unwrap();
+            assert_eq!(send_data, recv_data);
+        });
+    }
+}