From e11ba36eedec2e493ad5897453e0261bcc1f937a Mon Sep 17 00:00:00 2001 From: Tobias Stocker Date: Fri, 3 Jul 2026 13:46:59 -0600 Subject: [PATCH 01/10] Fixed bug where is_prefetching is set wrong --- dispatcher/src/queue.rs | 38 +++++++++++++++++++------------------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/dispatcher/src/queue.rs b/dispatcher/src/queue.rs index fb1de73b..681b70f9 100644 --- a/dispatcher/src/queue.rs +++ b/dispatcher/src/queue.rs @@ -176,31 +176,31 @@ impl Future for IoWaitFuture<'_> { let local_cores = *self.work_queue.system_info.num_local_cores_watcher.borrow(); let active_fetch_count = lock_guard.fetching_in_progress; let idle_compute_cores = lock_guard.compute_waker_list.len(); - let mut is_prefetching = false; - let result = lock_guard + let extracted = lock_guard .io_queue .extract_if(|queue_element| match &queue_element.work { - WorkToDo::FunctionArguments { .. } => { - is_prefetching = true; - policy::should_io_take( - &queue_element.policy_data, - compute_pending, - active_fetch_count, - local_cores, - idle_compute_cores, - ) - } - _ => { - is_prefetching = false; - true - } + WorkToDo::FunctionArguments { .. } => policy::should_io_take( + &queue_element.policy_data, + compute_pending, + active_fetch_count, + local_cores, + idle_compute_cores, + ), + _ => true, }) - .next() - .map(|queue_element| (queue_element.work, queue_element.debt)); - // If the task is a prefetching task increase the counter accordingly + .next(); + // If the extracted task is a prefetching task increase the counter accordingly + let is_prefetching = matches!( + &extracted, + Some(IoQueueElement { + work: WorkToDo::FunctionArguments { .. }, + .. + }) + ); if is_prefetching { lock_guard.fetching_in_progress += 1; } + let result = extracted.map(|queue_element| (queue_element.work, queue_element.debt)); if let Some(mut result_tupple) = result { if let WorkToDo::FunctionArguments { function_id: _, From aa39bf39d3e3c69cf4360d45f8e41c0fa1dd3dab Mon Sep 17 00:00:00 2001 From: Tobias Date: Tue, 7 Jul 2026 04:56:32 -0600 Subject: [PATCH 02/10] Fixed data locality bug, added note to reqwest for another potential bug --- dispatcher/src/queue/policy/data_locality.rs | 2 +- .../src/function_driver/system_driver/reqwest.rs | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) 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/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 { From 1dd45eac689a5f13531aabd71ff2ab33fcf95510 Mon Sep 17 00:00:00 2001 From: Tobias Date: Tue, 7 Jul 2026 07:44:31 -0600 Subject: [PATCH 03/10] Allowing tilde in paths --- server/src/config.rs | 36 +++++++++++++++++++++++++++++------- 1 file changed, 29 insertions(+), 7 deletions(-) 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| { From 9ee2553d4951be6231111865d4b51215271b0133 Mon Sep 17 00:00:00 2001 From: Tobias Date: Wed, 8 Jul 2026 02:45:13 -0600 Subject: [PATCH 04/10] Moved input size record logging to happen after resolving input sets --- dispatcher/src/dispatcher.rs | 10 ---------- dispatcher/src/queue.rs | 12 +++++++++++- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/dispatcher/src/dispatcher.rs b/dispatcher/src/dispatcher.rs index 039c2537..8304da82 100644 --- a/dispatcher/src/dispatcher.rs +++ b/dispatcher/src/dispatcher.rs @@ -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); } From ca80bc86ea317ab31ba2be4a20d2c1ccf9abcade Mon Sep 17 00:00:00 2001 From: Tobias Date: Wed, 8 Jul 2026 06:40:41 -0600 Subject: [PATCH 05/10] Fixed new input size logging for multinode case --- dandelion_commons/src/records.rs | 10 ++++++++++ dispatcher/src/dispatcher.rs | 2 +- multinode/proto/multinode.proto | 2 ++ multinode/src/util.rs | 6 ++++++ 4 files changed, 19 insertions(+), 1 deletion(-) diff --git a/dandelion_commons/src/records.rs b/dandelion_commons/src/records.rs index 15ebd942..c80fe138 100644 --- a/dandelion_commons/src/records.rs +++ b/dandelion_commons/src/records.rs @@ -191,6 +191,16 @@ 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 8304da82..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 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/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. From 3a2638aac4ff47a5bba79ae1c51ff92f71b53223 Mon Sep 17 00:00:00 2001 From: Tobias Date: Wed, 8 Jul 2026 07:08:06 -0600 Subject: [PATCH 06/10] Fixed another caching issue --- multinode/src/client.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 96b1ce79..a51058f4 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, ); } From f2aad2ef937b224a19687eb6ec8113e3be7cc9d4 Mon Sep 17 00:00:00 2001 From: Tobias Date: Thu, 9 Jul 2026 01:07:43 -0600 Subject: [PATCH 07/10] Fixed formatting issue --- dandelion_commons/src/records.rs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/dandelion_commons/src/records.rs b/dandelion_commons/src/records.rs index c80fe138..688e8fa0 100644 --- a/dandelion_commons/src/records.rs +++ b/dandelion_commons/src/records.rs @@ -193,12 +193,7 @@ impl Recorder { #[cfg(feature = "timestamp")] pub fn get_input(&self) -> (u64, u64) { - unsafe { - ( - *self.inner.input_items.get(), - *self.inner.input_size.get(), - ) - } + unsafe { (*self.inner.input_items.get(), *self.inner.input_size.get()) } } #[cfg(feature = "timestamp")] From b24935b27f5ff1248dfe4baaff5df600c06cbc66 Mon Sep 17 00:00:00 2001 From: Tobias Date: Thu, 9 Jul 2026 06:19:31 -0600 Subject: [PATCH 08/10] Multinode caching fix --- multinode/src/client.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/multinode/src/client.rs b/multinode/src/client.rs index a51058f4..940c2b2e 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -904,7 +904,7 @@ async fn remote_queue_client_logic( invocation_id, function_arc, inputs, - !caching, + caching, recorder, ) } From b2f212d17649f0e9c13e0f81599ea31679ebd975 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Thu, 9 Jul 2026 15:27:43 +0200 Subject: [PATCH 09/10] Reset TSC for each function --- .../compute_driver/kvm/x86_64.rs | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) 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) } } From 8ead91f9af816c1358793db5cef0936f30a50649 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Thu, 9 Jul 2026 15:48:47 +0200 Subject: [PATCH 10/10] Fix tests for multinode client --- multinode/src/client/test.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) 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 }