Bugfixes and minor improvements

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 {

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,

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);
         }
 

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,

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)
     }
 }

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 {

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 {

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,
                             )
                         }

multinode/src/client/test.rs

@@ -283,15 +283,15 @@ 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,
                     duration,
                     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
         }

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.

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,18 +270,21 @@ 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),
                 },
             };
         }
 
+        // 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| {