Skip to content

Commit a9bd768

Browse files
hunhoffehunhoffe
committed
Finished global core id impl
1 parent c994460 commit a9bd768

7 files changed

Lines changed: 162 additions & 67 deletions

File tree

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,7 @@ lazy_static! {
6565
};
6666
}
6767

68+
// List of hwthreads of all the clients in the rack
6869
lazy_static! {
6970
pub(crate) static ref HWTHREADS: Arc<Mutex<Vec<CpuThread>>> = {
7071
let mut hwthreads = Vec::try_with_capacity(get_num_clients() as usize)
@@ -73,6 +74,19 @@ lazy_static! {
7374
};
7475
}
7576

77+
// Keep track of which hwthreads have been allocated. Index corresponds to gtid of hwthread
78+
lazy_static! {
79+
pub(crate) static ref HWTHREADS_BUSY: Arc<Mutex<Vec<Option<bool>>>> = {
80+
// Assume each client has about 8 cores, for now
81+
let mut hwthreads_busy = Vec::try_with_capacity(get_num_clients() as usize * 8)
82+
.expect("Failed to create vector for rack cpu threads");
83+
for i in 0..(get_num_clients() as usize * 30) {
84+
hwthreads_busy.push(None);
85+
}
86+
Arc::new(Mutex::new(hwthreads_busy))
87+
};
88+
}
89+
7690
// Keep track of unfulfilled core assignments
7791
lazy_static! {
7892
pub(crate) static ref UNFULFILLED_CORE_ASSIGNMENTS: Arc<Mutex<Vec<Box<VecDeque<RequestCoreReq>>>>> = {

‎kernel/src/arch/x86_64/rackscale/processops/request_core.rs‎

Lines changed: 50 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -9,23 +9,27 @@ use core2::io::Write;
99
use rpc::rpc::*;
1010
use rpc::RPCClient;
1111

12-
use super::super::dcm::resource_alloc::dcm_resource_alloc;
13-
use super::super::kernelrpc::*;
14-
use crate::arch::rackscale::controller::get_local_pid;
15-
use crate::arch::rackscale::controller::UNFULFILLED_CORE_ASSIGNMENTS;
1612
use crate::error::KError;
1713
use crate::fs::cnrfs::MlnrKernelNode;
1814
use crate::fs::{cnrfs, NrLock};
1915
use crate::memory::VAddr;
2016
use crate::nr;
2117
use crate::nr::KernelNode;
2218

19+
use super::super::client::{get_local_client_id, get_num_clients};
20+
use super::super::controller::{
21+
get_local_pid, HWTHREADS, HWTHREADS_BUSY, UNFULFILLED_CORE_ASSIGNMENTS,
22+
};
23+
use super::super::dcm::resource_alloc::dcm_resource_alloc;
24+
use super::super::kernelrpc::*;
25+
use super::super::systemops::{gtid_to_local, local_to_gtid};
26+
2327
#[derive(Debug, Clone, Copy)]
2428
pub(crate) struct RequestCoreReq {
2529
pub core_id: u64,
2630
pub entry_point: u64,
2731
}
28-
unsafe_abomonate!(RequestCoreReq: core_id, entry_point);
32+
unsafe_abomonate!(RequestCoreReq: entry_point);
2933

3034
#[derive(Debug)]
3135
pub(crate) struct RequestCoreWorkRes {
@@ -66,10 +70,6 @@ pub(crate) fn rpc_request_core(
6670
return Err(RPCError::ExtraData);
6771
}
6872
info!("RequestCore() {:?}", res);
69-
70-
// TODO: could optimize for local case and call local function here
71-
// for now, will handle all the same (i.e., client ask for work from controller)
72-
7373
return res.ret;
7474
} else {
7575
return Err(RPCError::MalformedResponse);
@@ -95,24 +95,48 @@ pub(crate) fn handle_request_core(hdr: &mut RPCHeader, payload: &mut [u8]) -> Re
9595
}
9696
};
9797

98-
let node = dcm_resource_alloc(local_pid, true);
99-
100-
// Add request to be handled later.
101-
// TODO: handle local differently? For now, for simplicity, handle all the same
102-
// TODO: check capacity of core assignments?
103-
log::info!("Logged unfulfilled core assignment for {:?}", node);
104-
{
105-
let mut core_request_vec = UNFULFILLED_CORE_ASSIGNMENTS.lock();
106-
let mut deque = core_request_vec
107-
.get_mut(node as usize)
108-
.expect("failed to fetch core assignment deque for node");
109-
deque.push_back(*core_req);
98+
let client_id = dcm_resource_alloc(local_pid, true);
99+
100+
// controller chooses a core id - right now, sequentially for cores on the client_id.
101+
let num_clients = get_num_clients();
102+
let mut rack_hwthreads_busy = HWTHREADS_BUSY.lock();
103+
let mut index = 0;
104+
loop {
105+
//
106+
match rack_hwthreads_busy[local_to_gtid(index, client_id)] {
107+
// thread is busy, keep looking
108+
Some(true) => index += 1,
109+
// found an empty thread! set to busy and break
110+
Some(false) => {
111+
rack_hwthreads_busy[index] = Some(true);
112+
break;
113+
}
114+
// Ran out of threads for client; DCM should not have allowed this to happen
115+
None => panic!(
116+
"Should never happen - no empty hwthreads found for client {:?}",
117+
client_id
118+
),
119+
}
110120
}
111121

122+
let rack_hwthreads = HWTHREADS.lock();
123+
let gtid = rack_hwthreads[index].id;
124+
log::info!("Chose thread id {:?} for request", gtid);
125+
112126
// Construct and return result
113127
let res = KernelRpcRes {
114-
ret: convert_return(Ok((node, 0))),
128+
ret: convert_return(Ok((gtid as u64, 0))),
115129
};
130+
131+
// can handle request locally if same node otherwise must queue for remote node to handle
132+
if client_id != hdr.client_id {
133+
log::info!("Logged unfulfilled core assignment for {:?}", client_id);
134+
let mut core_request_vec = UNFULFILLED_CORE_ASSIGNMENTS.lock();
135+
let mut deque = core_request_vec
136+
.get_mut(client_id as usize)
137+
.expect("failed to fetch core assignment deque for client_id");
138+
deque.push_back(*core_req);
139+
}
116140
construct_ret(hdr, payload, res)
117141
}
118142

@@ -143,7 +167,10 @@ pub(crate) fn request_core_work(rpc_client: &mut dyn RPCClient) -> () {
143167
// But not sure if we can do that here because 1) client syscalls are ferried to the controller
144168
// and 2) how do you run it in the context of the (correct) remote process?
145169
// for now, copied & modified code from original syscall impl
146-
let gtid: usize = core_request.core_id.try_into().unwrap();
170+
let gtid: usize = gtid_to_local(
171+
core_request.core_id.try_into().unwrap(),
172+
get_local_client_id(),
173+
);
147174
let mut affinity = None;
148175
for thread in atopology::MACHINE_TOPOLOGY.threads() {
149176
if thread.id == gtid {

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

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,8 @@ use rpc::rpc::{ClientId, RPCError, RPCHeader};
1414
use rpc::RPCClient;
1515

1616
use super::dcm::node_registration::dcm_register_node;
17-
use crate::arch::rackscale::controller::{HWTHREADS, SHMEM_MANAGERS};
17+
use crate::arch::rackscale::client::get_num_clients;
18+
use crate::arch::rackscale::controller::{HWTHREADS, HWTHREADS_BUSY, SHMEM_MANAGERS};
1819
use crate::arch::rackscale::systemops::{local_to_gtid, local_to_node_id, local_to_package_id};
1920
use crate::error::KResult;
2021
use crate::memory::LARGE_PAGE_SIZE;
@@ -126,7 +127,8 @@ pub(crate) fn register_client(
126127
if remaining.len() == 0 {
127128
// Register client resources with DCM, DCM doesn't care about pids, so
128129
// send w/ dummy pid
129-
let client_id = dcm_register_node(0, req.num_cores, memslices);
130+
// TODO: register with one less core, assume init process uses that 1 core
131+
let client_id = dcm_register_node(0, req.num_cores - 1, memslices);
130132
info!("Registered client DCM, assigned client_id={:?}", client_id);
131133

132134
// Create shmem memory manager
@@ -145,8 +147,23 @@ pub(crate) fn register_client(
145147

146148
// Record information about the hardware threads
147149
info!("hwthreads: {:?}", hwthreads);
150+
148151
let mut rack_threads = HWTHREADS.lock();
152+
let mut rack_threads_busy = HWTHREADS_BUSY.lock();
153+
154+
// Make sure there's enough room to store data on whether core is busy or no
155+
let num_clients = get_num_clients() as usize;
156+
if rack_threads_busy.capacity() < hwthreads.len() * num_clients + client_id as usize
157+
{
158+
rack_threads_busy
159+
.resize_with(hwthreads.len() * num_clients + client_id as usize, || None);
160+
}
161+
149162
for hwthread in hwthreads {
163+
// set all threads to not busy
164+
rack_threads_busy[local_to_gtid(hwthread.id, client_id)] = Some(false);
165+
166+
// add thread to global state with global values made globally unique
150167
rack_threads.push(CpuThread {
151168
// these are global values to make sure no conflicts across rack
152169
id: local_to_gtid(hwthread.id, client_id),
@@ -157,6 +174,8 @@ pub(crate) fn register_client(
157174
thread_id: hwthread.thread_id,
158175
});
159176
}
177+
// Let's assume init process is running on hwthread 0 on the client so set that to busy
178+
rack_threads_busy[local_to_gtid(0, client_id)] = Some(true);
160179

161180
Ok(client_id)
162181
} else {

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

Lines changed: 25 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ use alloc::boxed::Box;
22
use alloc::string::String;
33

44
use kpi::io::{FileFlags, FileModes};
5+
use rpc::rpc::ClientId;
56

67
use crate::arch::process::{current_pid, Ring3Process};
78
use crate::error::KResult;
@@ -12,7 +13,7 @@ use crate::process::{KernArcBuffer, UserSlice};
1213
use crate::syscalls::{FsDispatch, ProcessDispatch, SystemCallDispatch, SystemDispatch};
1314

1415
use super::super::syscall::{Arch86SystemCall, Arch86SystemDispatch, Arch86VSpaceDispatch};
15-
use super::client::RPC_CLIENT;
16+
use super::client::{get_local_client_id, RPC_CLIENT};
1617
use super::fileops::close::rpc_close;
1718
use super::fileops::delete::rpc_delete;
1819
use super::fileops::getinfo::rpc_getinfo;
@@ -25,6 +26,7 @@ use super::processops::print::rpc_log;
2526
use super::processops::release_physical::rpc_release_physical;
2627
use super::processops::request_core::rpc_request_core;
2728
use super::systemops::get_hardware_threads::rpc_get_hardware_threads;
29+
use super::systemops::{gtid_to_local, is_gtid_local, local_to_gtid};
2830

2931
pub(crate) struct Arch86LwkSystemCall {
3032
pub(crate) local: Arch86SystemCall,
@@ -46,7 +48,14 @@ impl SystemDispatch<u64> for Arch86LwkSystemCall {
4648
}
4749

4850
fn get_core_id(&self) -> KResult<(u64, u64)> {
49-
self.local.get_core_id()
51+
// map local core ID to rackscale global core ID - since mapping is deterministic on number of
52+
// clients we can do this without making an RPC call
53+
self.local.get_core_id().and_then(|(core_id, n)| {
54+
Ok((
55+
local_to_gtid(core_id as usize, get_local_client_id()) as u64,
56+
n,
57+
))
58+
})
5059
}
5160
}
5261

@@ -158,7 +167,20 @@ impl ProcessDispatch<u64> for Arch86LwkSystemCall {
158167
fn request_core(&self, core_id: u64, entry_point: u64) -> KResult<(u64, u64)> {
159168
let mut client = RPC_CLIENT.lock();
160169
let pid = crate::arch::process::current_pid()?;
161-
rpc_request_core(&mut **client, pid, core_id, entry_point).map_err(|e| e.into())
170+
let ret = rpc_request_core(&mut **client, pid, core_id, entry_point).map_err(|e| e.into());
171+
172+
// request core locally if that's what was assigned this request
173+
let client_id = get_local_client_id();
174+
if let Ok((gtid, n)) = ret {
175+
if is_gtid_local(gtid as usize, client_id) {
176+
self.local
177+
.request_core(gtid_to_local(gtid as usize, client_id) as u64, entry_point)
178+
} else {
179+
ret
180+
}
181+
} else {
182+
ret
183+
}
162184
}
163185

164186
fn allocate_physical(&self, page_size: u64, affinity: u64) -> KResult<(u64, u64)> {

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

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -44,17 +44,15 @@ pub(crate) fn rpc_get_hardware_threads(
4444

4545
if let Ok((data_len, n)) = res.ret {
4646
if data_len as usize <= remaining.len() && data_len <= vaddr_buf_len {
47-
log::info!("There's a match! Writing into usesprace now");
4847
let mut user_slice =
4948
UserSlice::new(pid, UVAddr::try_from(vaddr_buf)?, data_len as usize)?;
5049
NrProcess::<Ring3Process>::write_to_userspace(
5150
&mut user_slice,
5251
&remaining[..data_len as usize],
5352
)?;
54-
log::info!("Returning value...");
5553
Ok((data_len, n))
5654
} else {
57-
log::info!(
55+
log::debug!(
5856
"Bad payload data: data_len: {:?} remaining.len(): {:?} vaddr_buf_len: {:?}",
5957
data_len,
6058
remaining.len(),
@@ -91,7 +89,7 @@ pub(crate) fn handle_get_hardware_threads(
9189
let additional_data = end - start;
9290
unsafe { encode(&*rack_threads, &mut &mut payload[start..end]) }
9391
.expect("Failed to encode hardware thread vector");
94-
log::info!(
92+
log::trace!(
9593
"Sending back {:?} bytes of data ({:?} hwthreads)",
9694
additional_data,
9795
rack_threads.len()

‎kernel/tests/s06_rackscale_tests.rs‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ fn s06_rackscale_phys_alloc_test() {
7979
.tap("tap2")
8080
.no_network_setup()
8181
.workers(2)
82+
.nobuild()
8283
.use_vmxnet3();
8384

8485
let mut output = String::new();
@@ -185,6 +186,7 @@ fn rackscale_fs_test(is_shmem: bool) {
185186
.tap("tap2")
186187
.no_network_setup()
187188
.workers(2)
189+
.nobuild()
188190
.use_vmxnet3();
189191

190192
let mut output = String::new();
@@ -264,6 +266,7 @@ fn s06_rackscale_shmem_fs_prop_test() {
264266
.tap("tap2")
265267
.no_network_setup()
266268
.workers(2)
269+
.nobuild()
267270
.use_vmxnet3();
268271

269272
let mut output = String::new();
@@ -349,6 +352,7 @@ fn s06_rackscale_shmem_multiinstance() {
349352
.tap(&tap)
350353
.no_network_setup()
351354
.workers(clients + 1)
355+
.nobuild()
352356
.use_vmxnet3();
353357

354358
let mut output = String::new();
@@ -434,7 +438,6 @@ fn rackscale_userspace_multicore_test(is_shmem: bool) {
434438
let mut dcm = spawn_dcm(1, timeout)?;
435439
let mut p = spawn_nrk(&cmdline_controller)?;
436440

437-
//output += p.exp_string("Finished sending requests!")?.as_str();
438441
output += p.exp_eof()?.as_str();
439442

440443
dcm.send_control('c')?;
@@ -463,13 +466,14 @@ fn rackscale_userspace_multicore_test(is_shmem: bool) {
463466
.workers(2)
464467
.cores(client_num_cores)
465468
.memory(4096)
469+
.nobuild()
466470
.use_vmxnet3();
467471

468472
let mut output = String::new();
469473
let mut qemu_run = || -> Result<WaitStatus> {
470474
let mut p = spawn_nrk(&cmdline_client)?;
471475

472-
for _i in 0..client_num_cores {
476+
for _i in 0..(client_num_cores - 1) {
473477
let r = p.exp_regex(r#"init: Hello from core (\d+)"#)?;
474478
output += r.0.as_str();
475479
output += r.1.as_str();
@@ -529,7 +533,7 @@ fn s06_rackscale_shmem_request_core_remote_test() {
529533
let mut qemu_run = || -> Result<WaitStatus> {
530534
let mut dcm = spawn_dcm(1, timeout)?;
531535
let mut p = spawn_nrk(&cmdline_controller)?;
532-
536+
output += p.exp_string("handle_request_core_work()")?.as_str();
533537
output += p.exp_eof()?.as_str();
534538

535539
dcm.send_control('c')?;
@@ -551,7 +555,7 @@ fn s06_rackscale_shmem_request_core_remote_test() {
551555
.tap("tap2")
552556
.no_network_setup()
553557
.workers(3)
554-
.cores(2)
558+
.cores(1)
555559
.memory(4096)
556560
.nobuild() // Use single build for all for consistency
557561
.use_vmxnet3();
@@ -562,6 +566,7 @@ fn s06_rackscale_shmem_request_core_remote_test() {
562566
output += p
563567
.exp_string("Client finished processing core work request")?
564568
.as_str();
569+
output += p.exp_string("vibrio::upcalls: Got a new core")?.as_str();
565570
p.process.exit()
566571
};
567572

@@ -580,19 +585,20 @@ fn s06_rackscale_shmem_request_core_remote_test() {
580585
.tap("tap4")
581586
.no_network_setup()
582587
.workers(3)
588+
.cores(2)
589+
.memory(4096)
583590
.nobuild() // Use build from previous client for consistency
584591
.use_vmxnet3();
585592

586593
let mut output = String::new();
587594
let mut qemu_run = || -> Result<WaitStatus> {
588595
let mut p = spawn_nrk(&cmdline_client)?;
589-
p.exp_string("request_core_remote_test OK")?;
590-
output = p.exp_eof()?;
591-
output += p.exp_eof()?.as_str();
596+
output += p.exp_string("Spawned core on CoreToken")?.as_str();
597+
output += p.exp_string("request_core_remote_test OK")?.as_str();
592598
p.process.exit()
593599
};
594600

595-
check_for_successful_exit(&cmdline_client, qemu_run(), output);
601+
let _ignore = qemu_run();
596602
});
597603

598604
controller.join().unwrap();

0 commit comments

Comments
 (0)