Add get_hardware_threads syscall rpc

Author: hunhoffe

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

@@ -66,12 +66,9 @@ lazy_static! {
 }
 
 lazy_static! {
-    pub(crate) static ref HWTHREADS: Arc<Mutex<Vec<Vec<CpuThread>>>> = {
+    pub(crate) static ref HWTHREADS: Arc<Mutex<Vec<(ClientId, CpuThread)>>> = {
         let mut hwthreads = Vec::try_with_capacity(get_num_clients() as usize)
             .expect("Failed to create vector for rack cpu threads");
-        for i in 0..get_num_clients() {
-            hwthreads.push(Vec::new());
-        }
         Arc::new(Mutex::new(hwthreads))
     };
 }
@@ -229,6 +226,13 @@ fn register_rpcs(server: &mut Box<dyn RPCServer>) {
             &REQUEST_CORE_WORK_HANDLER,
         )
         .unwrap();
+
+    server
+        .register(
+            KernelRpc::GetHardwareThreads as RPCType,
+            &GET_HARDWARE_THREADS_HANDLER,
+        )
+        .unwrap();
 }
 
 // Lookup the local pid corresponding to a remote pid

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

@@ -44,6 +44,7 @@ pub(crate) enum KernelRpc {
     FileRename = 10,
     /// Create a directory.
     MkDir = 11,
+
     /// Log (print) message of a process.
     Log = 12,
     /// Allocate physical memory for a process.
@@ -52,8 +53,12 @@ pub(crate) enum KernelRpc {
     ReleasePhysical = 14,
     /// Allocate a core for a process
     RequestCore = 15,
+
     /// Request work (e.g., request cores) - used by client to ask controller for tasks
     RequestWork = 16,
+
+    /// Get the hardware threads for the rack
+    GetHardwareThreads = 17,
 }
 
 impl TryFrom<RPCType> for KernelRpc {
@@ -79,6 +84,7 @@ impl TryFrom<RPCType> for KernelRpc {
             15 => Ok(KernelRpc::ReleasePhysical),
             15 => Ok(KernelRpc::RequestCore),
             16 => Ok(KernelRpc::RequestWork),
+            17 => Ok(KernelRpc::GetHardwareThreads),
             _ => Err(KError::InvalidRpcType),
         }
     }

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

@@ -14,6 +14,7 @@ pub(crate) mod kernelrpc;
 pub(crate) mod processops;
 pub(crate) mod registration;
 pub(crate) mod syscalls;
+pub(crate) mod systemops;
 
 pub(crate) use self::kernelrpc::KernelRpc;
 
@@ -40,6 +41,10 @@ pub(crate) const RELEASE_PHYSICAL_HANDLER: RPCHandler =
     processops::release_physical::handle_release_physical;
 pub(crate) const LOG_HANDLER: RPCHandler = processops::print::handle_log;
 
+// Re-export handlers: system operations
+pub(crate) const GET_HARDWARE_THREADS_HANDLER: RPCHandler =
+    systemops::get_hardware_threads::handle_get_hardware_threads;
+
 // Client polls for work
 pub(crate) const REQUEST_CORE_WORK_HANDLER: RPCHandler =
     processops::request_core::handle_request_core_work;

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

@@ -145,23 +145,9 @@ pub(crate) fn register_client(
                 // Record information about the hardware threads
                 info!("hwthreads: {:?}", hwthreads);
                 let mut rack_threads = HWTHREADS.lock();
-                let mut gtid = 0;
-                for i in 0..node_id as usize {
-                    gtid += rack_threads[i].len();
-                }
-                info!("Starting client gtid at: {:?}", gtid);
                 for hwthread in hwthreads {
-                    // Create new, globally unique global thread id (gtid)
-                    rack_threads[node_id as usize].push(CpuThread {
-                        id: gtid,
-                        node_id: hwthread.node_id,
-                        package_id: hwthread.package_id,
-                        core_id: hwthread.core_id,
-                        thread_id: hwthread.thread_id,
-                    });
-                    gtid += 1;
+                    rack_threads.push((node_id, *hwthread));
                 }
-                info!("rack_threads: {:?}", rack_threads);
 
                 Ok(node_id)
             } else {

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

@@ -24,6 +24,7 @@ use super::processops::allocate_physical::rpc_allocate_physical;
 use super::processops::print::rpc_log;
 use super::processops::release_physical::rpc_release_physical;
 use super::processops::request_core::rpc_request_core;
+use super::systemops::get_hardware_threads::rpc_get_hardware_threads;
 
 pub(crate) struct Arch86LwkSystemCall {
     pub(crate) local: Arch86SystemCall,
@@ -35,7 +36,9 @@ impl Arch86VSpaceDispatch for Arch86LwkSystemCall {}
 
 impl SystemDispatch<u64> for Arch86LwkSystemCall {
     fn get_hardware_threads(&self, vaddr_buf: u64, vaddr_buf_len: u64) -> KResult<(u64, u64)> {
-        self.local.get_hardware_threads(vaddr_buf, vaddr_buf_len)
+        let pid = crate::arch::process::current_pid()?;
+        let mut client = RPC_CLIENT.lock();
+        rpc_get_hardware_threads(&mut **client, pid, vaddr_buf, vaddr_buf_len).map_err(|e| e.into())
     }
 
     fn get_stats(&self) -> KResult<(u64, u64)> {

kernel/src/arch/x86_64/rackscale/systemops/get_hardware_threads.rs

@@ -0,0 +1,114 @@
+// Copyright © 2021 University of Colorado. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+
+use alloc::vec::Vec;
+
+use abomonation::{decode, encode, unsafe_abomonate, Abomonation};
+use core2::io::Result as IOResult;
+use core2::io::Write;
+use fallible_collections::FallibleVecGlobal;
+
+use kpi::system::CpuThread;
+use rpc::rpc::*;
+use rpc::RPCClient;
+
+use super::super::kernelrpc::*;
+use crate::arch::process::Ring3Process;
+use crate::arch::rackscale::controller::{get_local_pid, HWTHREADS};
+use crate::nrproc::NrProcess;
+use crate::process::{UVAddr, UserSlice};
+
+pub(crate) fn rpc_get_hardware_threads(
+    rpc_client: &mut dyn RPCClient,
+    pid: usize,
+    vaddr_buf: u64,
+    vaddr_buf_len: u64,
+) -> Result<(u64, u64), RPCError> {
+    // Setup result
+    // TODO: make dynamic, for now, size copied from kpi implementation
+    let mut res_data = [0u8; core::mem::size_of::<KernelRpcRes>() + 5 * 4096];
+
+    // Call GetHardwareThreads() RPC
+    rpc_client
+        .call(
+            pid,
+            KernelRpc::GetHardwareThreads as RPCType,
+            &[&[]],
+            &mut [&mut res_data],
+        )
+        .unwrap();
+
+    // Decode and return result
+    if let Some((res, remaining)) = unsafe { decode::<KernelRpcRes>(&mut res_data) } {
+        log::info!("GetHardwareThreads() {:?}", res);
+
+        if let Ok((data_len, n)) = res.ret {
+            if data_len as usize <= remaining.len() && data_len <= vaddr_buf_len {
+                log::info!("There's a match! Writing into usesprace now");
+                let mut user_slice =
+                    UserSlice::new(pid, UVAddr::try_from(vaddr_buf)?, data_len as usize)?;
+                NrProcess::<Ring3Process>::write_to_userspace(
+                    &mut user_slice,
+                    &remaining[..data_len as usize],
+                )?;
+                log::info!("Returning value...");
+                Ok((data_len, n))
+            } else {
+                log::info!(
+                    "Bad payload data: data_len: {:?} remaining.len(): {:?} vaddr_buf_len: {:?}",
+                    data_len,
+                    remaining.len(),
+                    vaddr_buf_len
+                );
+                Err(RPCError::MalformedResponse)
+            }
+        } else {
+            res.ret
+        }
+    } else {
+        Err(RPCError::MalformedResponse)
+    }
+}
+
+// RPC Handler function for get_hardware_threads() RPCs in the controller
+pub(crate) fn handle_get_hardware_threads(
+    hdr: &mut RPCHeader,
+    payload: &mut [u8],
+) -> Result<(), RPCError> {
+    // Lookup local pid
+    let local_pid = { get_local_pid(hdr.client_id, hdr.pid) };
+    if local_pid.is_err() {
+        return construct_error_ret(hdr, payload, RPCError::NoFileDescForPid);
+    }
+    let local_pid = local_pid.unwrap();
+
+    let rack_threads = HWTHREADS.lock();
+
+    // calculate total number of threads
+    let mut hwthreads =
+        Vec::try_with_capacity(rack_threads.len()).expect("failed to allocate space for hwthreads");
+    for i in 0..rack_threads.len() {
+        hwthreads.push(rack_threads[i].1);
+    }
+    log::info!(
+        "Found {:?} hardware threads: {:?}",
+        hwthreads.len(),
+        hwthreads
+    );
+
+    // Encode hwthread information into payload buffer
+    let start = KernelRpcRes_SIZE as usize;
+    let end = start
+        + hwthreads.len() * core::mem::size_of::<CpuThread>()
+        + core::mem::size_of::<Vec<CpuThread>>();
+    let additional_data = end - start;
+    unsafe { encode(&hwthreads, &mut &mut payload[start..end]) }
+        .expect("Failed to encode hardware thread vector");
+    log::info!("Sending back {:?} bytes of data", additional_data);
+
+    // Construct return
+    let res = KernelRpcRes {
+        ret: Ok((additional_data as u64, 0)),
+    };
+    construct_ret_extra_data(hdr, payload, res, additional_data as u64)
+}

kernel/src/arch/x86_64/rackscale/systemops/mod.rs

@@ -0,0 +1,4 @@
+// Copyright © 2022 University of Colorado. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+
+pub mod get_hardware_threads;

usr/init/src/init.rs

@@ -150,27 +150,46 @@ fn request_core_remote_test() {
     let s = &vibrio::upcalls::PROCESS_SCHEDULER;
 
     let threads = vibrio::syscalls::System::threads().expect("Can't get system topology");
+    info!("threads: {:?}", threads);
 
-    // Only ask for more cores on the machine with only 1 core
-    if threads.len() == 1 {
-        let HW_THREAD_ID = 1;
-        match vibrio::syscalls::Process::request_core(
-            HW_THREAD_ID,
-            VAddr::from(vibrio::upcalls::upcall_while_enabled as *const fn() as u64),
-        ) {
-            Ok(_) => {
-                info!("request_core_remote_test OK");
-            }
-            Err(e) => {
-                error!("Can't spawn on {:?}: {:?}", HW_THREAD_ID, e);
+    for thread in threads.iter() {
+        if thread.id != 0 {
+            let r = vibrio::syscalls::Process::request_core(
+                thread.id,
+                VAddr::from(vibrio::upcalls::upcall_while_enabled as *const fn() as u64),
+            );
+            match r {
+                Ok(ctoken) => {
+                    info!("Spawned core on {:?} <-> {}", ctoken, thread.id);
+                }
+                Err(_e) => {
+                    panic!("Failed to spawn to core {}", thread.id);
+                }
             }
         }
-    } else {
-        // Run scheduler on core 0
-        let scb: SchedulerControlBlock = SchedulerControlBlock::new(0);
-        loop {
-            s.run(&scb);
-        }
+    }
+
+    /*
+    for thread in threads {
+        s.spawn(
+            32 * 4096,
+            move |_| {
+                info!(
+                    "Hello from core {}",
+                    lineup::tls2::Environment::scheduler().core_id
+                );
+            },
+            ptr::null_mut(),
+            thread.id,
+            None,
+        );
+    }
+    */
+
+    // Run scheduler on core 0
+    let scb: SchedulerControlBlock = SchedulerControlBlock::new(0);
+    loop {
+        s.run(&scb);
     }
 }