diff --git a/dandelion_commons/src/records.rs b/dandelion_commons/src/records.rs index 15ebd942..688e8fa0 100644 --- a/dandelion_commons/src/records.rs +++ b/dandelion_commons/src/records.rs @@ -191,6 +191,11 @@ impl Recorder { } } + #[cfg(feature = "timestamp")] + pub fn get_input(&self) -> (u64, u64) { + unsafe { (*self.inner.input_items.get(), *self.inner.input_size.get()) } + } + #[cfg(feature = "timestamp")] pub fn set_node_id(&mut self, node_id: u64) { unsafe { diff --git a/dispatcher/src/dispatcher.rs b/dispatcher/src/dispatcher.rs index 039c2537..4cba4ee6 100644 --- a/dispatcher/src/dispatcher.rs +++ b/dispatcher/src/dispatcher.rs @@ -361,7 +361,7 @@ impl Dispatcher { .collect(); awaited_sets.push(Either::Left(ready(Ok(( new_sets, - *function_index, + args.function_index, Vec::new(), ))))); None @@ -554,16 +554,6 @@ impl Dispatcher { metadata.output_sets, function_alternatives ); - #[cfg(feature = "timestamp")] - { - let (total_items, total_size) = - input_sets.iter().fold((0, 0), |(number, size), set| { - set.as_ref() - .map(|set| (number + set.len(), size + set.size())) - .unwrap_or((number, size)) - }); - recorder.record_input(total_items as u64, total_size as u64); - } let args = WorkToDo::FunctionArguments { function_id: function_id.clone(), function_alternatives, diff --git a/dispatcher/src/queue.rs b/dispatcher/src/queue.rs index 681b70f9..4986887d 100644 --- a/dispatcher/src/queue.rs +++ b/dispatcher/src/queue.rs @@ -279,12 +279,22 @@ impl WorkQueue { if let WorkToDo::FunctionArguments { function_id: _, function_alternatives: _, - input_sets: _, + input_sets: _input_sets, metadata: _, caching: _, recorder, } = &mut work { + #[cfg(feature = "timestamp")] + { + let (total_items, total_size) = + _input_sets.iter().fold((0, 0), |(number, size), set| { + set.as_ref() + .map(|set| (number + set.len(), size + set.size())) + .unwrap_or((number, size)) + }); + recorder.record_input(total_items as u64, total_size as u64); + } recorder.record(RecordPoint::ComputeQueueStart); } diff --git a/dispatcher/src/queue/policy/data_locality.rs b/dispatcher/src/queue/policy/data_locality.rs index 5df61bc2..93e913ba 100644 --- a/dispatcher/src/queue/policy/data_locality.rs +++ b/dispatcher/src/queue/policy/data_locality.rs @@ -98,7 +98,7 @@ pub fn should_io_take( // always take it if there are idle cores, only prefetch if it is prefetching via IO, not from other nodes idle_compute_cores > 0 || (compute_pending + active_fetch_count < LOCAL_WORK_PER_CORE * local_cores - && !element_data.remote_data.is_empty()) + && element_data.remote_data.is_empty()) } /// Selects work items from the local queues to hand off to a remote node, diff --git a/machine_interface/src/function_driver/compute_driver/kvm/x86_64.rs b/machine_interface/src/function_driver/compute_driver/kvm/x86_64.rs index ec9bf24a..a629839b 100644 --- a/machine_interface/src/function_driver/compute_driver/kvm/x86_64.rs +++ b/machine_interface/src/function_driver/compute_driver/kvm/x86_64.rs @@ -1,5 +1,5 @@ use dandelion_commons::{err_dandelion, DandelionError, DandelionResult}; -use kvm_bindings::{kvm_fpu, kvm_regs, kvm_segment, kvm_sregs, kvm_xcrs}; +use kvm_bindings::{kvm_fpu, kvm_msr_entry, kvm_regs, kvm_segment, kvm_sregs, kvm_xcrs, Msrs}; use kvm_ioctls::{VcpuFd, VmFd}; use log::{debug, trace}; use std::{os::raw::c_void, slice}; @@ -195,6 +195,26 @@ impl ResetState { ..Default::default() }) .unwrap(); + + // reset the cpu time step counter register + let msrs = Msrs::from_entries(&[ + // set the MSR_IA32_TSC + kvm_msr_entry { + index: 0x10, + data: 0, + ..Default::default() + }, + ]) + .unwrap(); + assert_eq!(1, vcpu.set_msrs(&msrs).unwrap()); + let msrs = Msrs::from_entries(&[kvm_msr_entry { + index: 0x3b, + data: 0, + ..Default::default() + }]) + .unwrap(); + assert_eq!(1, vcpu.set_msrs(&msrs).unwrap()); + Ok(page_fault_metadata) } } diff --git a/machine_interface/src/function_driver/system_driver/reqwest.rs b/machine_interface/src/function_driver/system_driver/reqwest.rs index d170d8b6..bce79793 100644 --- a/machine_interface/src/function_driver/system_driver/reqwest.rs +++ b/machine_interface/src/function_driver/system_driver/reqwest.rs @@ -654,6 +654,10 @@ async fn engine_loop(queue: impl EngineWorkQueue + Clone + Send + 'static) -> De move |sets_result| { recorder.record(RecordPoint::FetchingEnd); match sets_result { + // FIXME: this leaks the queue's `fetching_in_progress` count, since + // that is only decremented in `requeu_engine_args`. Repeated fetch + // failures permanently inflate the count and can starve the io queue + // of taking any more prefetch work. Err(err) => debt.fulfill(Err(err)), Ok(sets) => queue_clone.requeu_engine_args( WorkToDo::FunctionArguments { diff --git a/multinode/proto/multinode.proto b/multinode/proto/multinode.proto index cb346365..d74affd7 100644 --- a/multinode/proto/multinode.proto +++ b/multinode/proto/multinode.proto @@ -83,6 +83,8 @@ message Timestamps { uint64 transfer_start = 11; uint64 engine_start = 12; uint64 engine_end = 13; + uint64 input_items = 14; + uint64 input_size = 15; } message RepeatedMetadataSet { diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 96b1ce79..940c2b2e 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -871,7 +871,7 @@ async fn remote_queue_client_logic( invocation_id, function_arc, inputs, - !caching, + caching, recorder, ); } @@ -904,7 +904,7 @@ async fn remote_queue_client_logic( invocation_id, function_arc, inputs, - !caching, + caching, recorder, ) } diff --git a/multinode/src/client/test.rs b/multinode/src/client/test.rs index df4197d7..69acd5de 100644 --- a/multinode/src/client/test.rs +++ b/multinode/src/client/test.rs @@ -283,7 +283,7 @@ fn test_remote_queue_client() { let (remote_message_sender, mut remote_message_receiver) = mpsc::channel(64); let dispatcher_send = - |registry, duration, invocation_id, function_id, inputs, is_cold, recorder| { + |registry, duration, invocation_id, function_id, inputs, caching, recorder| { dispatcher_sender .blocking_send(( registry, @@ -291,7 +291,7 @@ fn test_remote_queue_client() { invocation_id, function_id, inputs, - is_cold, + caching, recorder, )) .unwrap(); @@ -365,10 +365,10 @@ fn test_remote_queue_client() { invocation_id, function_id, _inputs, - is_cold, + caching, _recorder, ))) => { - assert!(!is_cold); + assert!(caching); assert_eq!(expected_function_id, function_id.as_str()); invocation_id } @@ -382,10 +382,10 @@ fn test_remote_queue_client() { invocation_id, function_id, _inputs, - is_cold, + caching, _recorder, ))) => { - assert!(!is_cold); + assert!(caching); assert_eq!(expected_function_id, function_id.as_str()); invocation_id } @@ -433,10 +433,10 @@ fn test_remote_queue_client() { invocation_id, function_id, _inputs, - is_cold, + caching, _recorder, ))) => { - assert!(!is_cold); + assert!(caching); assert_eq!(expected_function_id, function_id.as_str()); invocation_id } diff --git a/multinode/src/util.rs b/multinode/src/util.rs index 205dd624..378484c6 100644 --- a/multinode/src/util.rs +++ b/multinode/src/util.rs @@ -52,8 +52,11 @@ pub(crate) fn recorder_dtop( { use dandelion_commons::records::RecordPoint; + let (input_items, input_size) = _recorder.get_input(); Some(proto::Timestamps { start_epoch: _start_epoch.as_micros() as u64, + input_items, + input_size, io_queue_start: _recorder .get_timestamp(RecordPoint::IOQueueStart) .as_micros() as u64, @@ -114,7 +117,10 @@ pub(crate) fn recorder_add_timestamps( transfer_start, engine_start, engine_end, + input_items, + input_size, } = remote_time; + _recorder.record_input(input_items, input_size); // The local RemoteTake and the local_reference were taken at approximately same time. // Both the local reference and the start_epoch are offsets from unix epoch start. // The every remote timestamp is an offset from the start_epoch. diff --git a/server/src/config.rs b/server/src/config.rs index 33ae774a..dab261a1 100644 --- a/server/src/config.rs +++ b/server/src/config.rs @@ -1,5 +1,5 @@ use core::fmt; -use std::{fs::File, path::Path, str::FromStr}; +use std::{env, fs::File, path::Path, str::FromStr}; use clap::Parser; use log::{error, warn}; @@ -14,6 +14,22 @@ const DEFAULT_MULTINODE_RECONNECT_INTERVAL: u64 = 1000; use machine_interface::composition::DEFAULT_AUTOSHARDING_OFFLOAD_CONST; use machine_interface::function_driver::system_driver::reqwest::DEFAULT_CONCURRENCY_LIMIT; +/// Expand a leading `~` (or `~/...`) in a path to the current user's home directory, +/// same as a shell would. Paths that don't start with `~` are returned unchanged. +fn expand_tilde(path: &str) -> String { + let Some(home) = env::var_os("HOME") else { + return path.to_string(); + }; + let home = home.to_string_lossy(); + if path == "~" { + home.into_owned() + } else if let Some(rest) = path.strip_prefix("~/") { + format!("{}/{}", home.trim_end_matches('/'), rest) + } else { + path.to_string() + } +} + #[derive(serde::Deserialize, Debug)] pub struct PreloadFunc { #[serde(rename = "name")] @@ -254,11 +270,9 @@ impl DandelionConfig { // if a config path is given -> read + parse it and merge into args config if !cli_config.config_path.is_empty() { - match File::open(Path::new(&cli_config.config_path)) { - Err(err) => warn!( - "Could not load config file {}: {}", - cli_config.config_path, err - ), + let config_path = expand_tilde(&cli_config.config_path); + match File::open(Path::new(&config_path)) { + Err(err) => warn!("Could not load config file {}: {}", config_path, err), Ok(config_file) => match serde_json::from_reader(config_file) { Ok(file_config) => cli_config.merge_serde_into_args(file_config), Err(err) => warn!("Could not load config file: {}", err), @@ -266,6 +280,11 @@ impl DandelionConfig { }; } + // expand `~` in any configuration parameters that are file system paths + cli_config.bin_preload_path = expand_tilde(&cli_config.bin_preload_path); + cli_config.folder_path = expand_tilde(&cli_config.folder_path); + cli_config.multinode_config = cli_config.multinode_config.as_deref().map(expand_tilde); + cli_config .total_cores .get_or_insert(num_cpus::get_physical()); @@ -324,7 +343,7 @@ impl DandelionConfig { Ok(f) => f, }; let PreloadFile { - functions, + mut functions, compositions, } = match serde_json::from_reader(reader) { Err(err) => { @@ -333,6 +352,9 @@ impl DandelionConfig { } Ok(json) => json, }; + for func in functions.iter_mut() { + func.bin_path = expand_tilde(&func.bin_path); + } // sanity checks if !functions.iter().all(|pf| {