Arbitrary field ordering in invocation requests, single-shot composition invocations, more github test cases. (#86)

* Allow overwriting of loaded functions

* Change timestamp points

* Allowing for arbitrary field ordering in invocation request

* Added single-shot composition invocations

* Changed record export to json format

* Added two more github tests to check the timestamp and log_function_stdio features

* Remove unneeded comments

---------

Co-authored-by: Tom Kuchler <tom.j.kuchler@gmail.com>
Co-authored-by: Tobias Stocker <tstocker@r630-09.ethz.ch>

Author: 3 people

.github/workflows/cargo-tests.yaml

@@ -18,7 +18,21 @@ jobs:
         compute_driver: [ mmu ]
         system_driver: [ reqwest_io ]
         platform: [ubuntu-24.04, ubuntu-24.04-arm]
+        additional_features: [""]
 
+        # the following additional features should be platform agnostic, thus we only test them on one specification
+        include:
+        - platform: ubuntu-24.04
+          compute_driver: mmu
+          system_driver: reqwest_io
+          additional_features: "timestamp"
+        
+        - platform: ubuntu-24.04
+          compute_driver: mmu
+          system_driver: reqwest_io
+          additional_features: "log_function_stdio"
+
+    name: ${{matrix.platform}} - ${{matrix.compute_driver}}|${{matrix.system_driver}} - ${{ matrix.additional_features || 'core' }}
     runs-on: ${{matrix.platform}}
 
     steps:
@@ -27,7 +41,7 @@ jobs:
     # set up feature specific prerequisits
     - name: mmu worker build
       if: ${{ matrix.compute_driver == 'mmu' }}
-      run: cargo build -F ${{matrix.compute_driver}} --bin mmu_worker --target $(arch)-unknown-linux-gnu
+      run: cargo build -F ${{matrix.compute_driver}},${{matrix.additional_features}} --bin mmu_worker --target $(arch)-unknown-linux-gnu
     # run the actual tests
     - name: run tests
-      run: SINGLE_CORE_MODE=true cargo test -F ${{matrix.compute_driver}},${{matrix.system_driver}}
+      run: SINGLE_CORE_MODE=true cargo test -F ${{matrix.compute_driver}},${{matrix.system_driver}},${{matrix.additional_features}}

dandelion_commons/src/lib.rs

@@ -154,6 +154,8 @@ pub enum DispatcherError {
     CompositionCombine,
     /// dispatcher found mistake when trying to find waiting functions
     DependencyError,
+    /// dispatcher got invalid composition
+    InvalidComposition,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq)]

dandelion_commons/src/records.rs

@@ -5,29 +5,25 @@ use std::time::Instant;
 /// Maximum usize to expect when converting a record point to a usize
 /// By setting the last element to this explicitly, the compiler will throw an error,
 /// if there are more than this, because it enumerates from 0 and won't allow a number to be assigned twice.
-const LAST_RECORD_POINT: usize = 11;
+const LAST_RECORD_POINT: usize = 9;
 
 #[repr(usize)]
 #[derive(Clone, Copy, Debug, PartialEq)]
 pub enum RecordPoint {
-    /// Queue to load the function code + ctx
-    PrepareEnvQueue,
-    /// Enqueue parsing operation (async)
-    ParsingQueue,
+    /// Frontend has finished deserializing
+    DeserializationEnd,
+    /// When the request first enters the dispatcher
+    EnterDispatcher,
+    /// Queue to get the function executed on the engine (async)
+    ExecutionQueue,
     /// Start parsing (sync)
     ParsingStart,
     /// Finished Parsing (sync)
     ParsingEnd,
-    /// Dequeue from parsing (async)
-    ParsingDequeue,
     /// Start loading code + alloc ctx (sync)
     LoadStart,
     /// Start data transfer to the ctx (sync)
     TransferStart,
-    /// Queue to get an engine for execution
-    GetEngineQueue,
-    /// Queue to get the function executed on the engine (async)
-    ExecutionQueue,
     /// Start execution of the function on the engine (sync)
     EngineStart,
     /// End execution of the function on the engine (sync)
@@ -77,23 +73,27 @@ impl FunctionTimestamp {
 #[cfg(feature = "timestamp")]
 impl fmt::Display for FunctionTimestamp {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "function_id {}, time_points: ", self.function_id)?;
+        write!(
+            f,
+            "{{\"function_id\": {}, \"time_points\": [",
+            self.function_id
+        )?;
         // write own time points
         for index in 0..LAST_RECORD_POINT {
             let duration = unsafe { *self.time_points[index].get() };
             write!(f, "{},", duration.as_micros())?;
         }
         let duration = unsafe { *self.time_points[LAST_RECORD_POINT].get() };
-        write!(f, "{}, children: {{", duration.as_micros())?;
+        write!(f, "{}], \"children\": [", duration.as_micros())?;
         let child_guard = self.children.lock().unwrap();
         let num_children = child_guard.len();
         if num_children > 0 {
             for index in 0..num_children - 1 {
-                write!(f, "[{}],", child_guard[index])?;
+                write!(f, "{},", child_guard[index])?;
             }
-            write!(f, "[{}]", child_guard[num_children - 1])?;
+            write!(f, "{}", child_guard[num_children - 1])?;
         }
-        write!(f, "}}")?;
+        write!(f, "]}}")?;
         Ok(())
     }
 }

dispatcher/src/dispatcher.rs

@@ -6,7 +6,7 @@ use crate::{
 use core::pin::Pin;
 use dandelion_commons::{
     records::{RecordPoint, Recorder},
-    DandelionResult, FunctionId,
+    DandelionError, DandelionResult, DispatcherError, FunctionId,
 };
 use futures::{
     future::{join_all, ready, Either},
@@ -102,14 +102,13 @@ impl Dispatcher {
 
     pub async fn queue_function_by_name(
         &self,
-        function_name: String,
+        function_id: Arc<String>,
         inputs: Vec<DispatcherInput>,
         caching: bool,
-        start_time: std::time::Instant,
+        mut recorder: Recorder,
     ) -> DandelionResult<(Vec<Option<CompositionSet>>, Recorder)> {
-        debug!("Queuing function {}", function_name);
-        let function_id = Arc::new(function_name);
-        let recorder = Recorder::new(function_id.clone(), start_time);
+        debug!("Queuing function {}", function_id);
+        recorder.record(RecordPoint::EnterDispatcher);
 
         let mut input_vec = Vec::with_capacity(inputs.len());
         input_vec.resize(inputs.len(), None);
@@ -130,6 +129,53 @@ impl Dispatcher {
         return Ok((results, recorder));
     }
 
+    pub async fn queue_unregistered_composition(
+        &self,
+        composition_desc: String,
+        inputs: Vec<DispatcherInput>,
+        caching: bool,
+        recorder: Recorder,
+    ) -> DandelionResult<(Vec<Option<CompositionSet>>, Recorder)> {
+        debug!("Parsing single use composition");
+        let composition_meta_pairs = self
+            .function_registry
+            .parse_compositions(&composition_desc.as_str())?;
+        if composition_meta_pairs.len() != 1 {
+            debug!(
+                "Expected exactly one composition got {}",
+                composition_meta_pairs.len()
+            );
+            return Err(DandelionError::Dispatcher(
+                DispatcherError::InvalidComposition,
+            ));
+        }
+
+        debug!(
+            "Queuing single use composition {}",
+            composition_meta_pairs[0].0
+        );
+        let mut input_vec = Vec::with_capacity(inputs.len());
+        input_vec.resize(inputs.len(), None);
+        for (index, input) in inputs.into_iter().enumerate() {
+            match input {
+                DispatcherInput::None => (),
+                DispatcherInput::Set(set) => {
+                    input_vec[index] = Some(set);
+                }
+            }
+        }
+        let results = self
+            .queue_composition(
+                composition_meta_pairs[0].1.clone(),
+                input_vec,
+                caching,
+                recorder.get_sub_recorder(),
+            )
+            .await?;
+
+        return Ok((results, recorder));
+    }
+
     /// Queue a composition for execution.
     /// Returns a Vec of sets with the index corresponding to the set index in the composition definition
     ///
@@ -432,7 +478,7 @@ impl Dispatcher {
     ) -> Pin<
         Box<dyn Future<Output = DandelionResult<Vec<Option<CompositionSet>>>> + 'dispatcher + Send>,
     > {
-        trace!("queueing function with id: {}", function_id);
+        debug!("Queueing function with id: {}", function_id);
         Box::pin(async move {
             // find an engine capable of running the function
             // TODO: think about more distinctions, that allow pushing chains of functions which can be executed by single engine,
@@ -447,7 +493,6 @@ impl Dispatcher {
 
                     let metadata = func_info.metadata;
                     // run on engine
-                    recorder.record(RecordPoint::GetEngineQueue);
                     trace!(
                         "Running function {} with input sets {:?} and output sets {:?} and alternatives: {:?}",
                         function_id,
@@ -473,12 +518,12 @@ impl Dispatcher {
                     recorder.record(RecordPoint::FutureReturn);
 
                     #[cfg(feature = "log_function_stdio")]
-                    for opt in result.content.iter() {
+                    for opt in context.content.iter() {
                         if opt.as_ref().is_some_and(|s| s.ident == "stdio") {
                             for itm in opt.as_ref().unwrap().buffers.iter() {
                                 if itm.ident == "stderr" && itm.data.size > 0 {
                                     let mut stderr_output: Vec<u8> = vec![0; itm.data.size];
-                                    result.context.read(itm.data.offset, &mut stderr_output)?;
+                                    context.context.read(itm.data.offset, &mut stderr_output)?;
                                     warn!(
                                         "Function result contains stderr output:\n{}",
                                         std::str::from_utf8(stderr_output.as_slice())
@@ -487,7 +532,7 @@ impl Dispatcher {
                                 }
                                 if itm.ident == "stdout" && itm.data.size > 0 {
                                     let mut stdout_output: Vec<u8> = vec![0; itm.data.size];
-                                    result.context.read(itm.data.offset, &mut stdout_output)?;
+                                    context.context.read(itm.data.offset, &mut stdout_output)?;
                                     debug!(
                                         "Function output:\n{}",
                                         std::str::from_utf8(stdout_output.as_slice())

dispatcher/src/function_registry.rs

@@ -550,6 +550,19 @@ impl FunctionRegistry {
         Ok(())
     }
 
+    /// Parses the compositions without inserting it into the registry.
+    pub fn parse_compositions(
+        &self,
+        composition_desc: &str,
+    ) -> DandelionResult<Vec<(FunctionId, Composition, Metadata)>> {
+        // TODO: might want to return the parsing issue back to the user in a better way
+        let module = dparser::parse(composition_desc).map_err(|parse_error| {
+            print_errors(composition_desc, parse_error);
+            DandelionError::Composition(CompositionError::ParsingError)
+        })?;
+        self.composition_from_module(module)
+    }
+
     /// Checks if a function identifier is registered in the function registry.
     pub fn exists_id(&self, function_id: &FunctionId) -> bool {
         let lock_guard = self

dispatcher/src/remote.rs

@@ -0,0 +1,102 @@
+use std::thread::{self, JoinHandle};
+
+use dandelion_commons::DandelionResult;
+use log::error;
+use machine_interface::machine_config::EngineType;
+use multinode::client::{create_client, RemoteClient};
+use tokio::runtime::Builder;
+
+use crate::queue::{get_engine_flag, WorkQueue};
+
+static RETRY_TIMEOUT: u64 = 1000;
+
+enum Status {
+    INACTIVE,
+    ACTIVE,
+    STOPPED,
+    FINISHED,
+}
+
+pub struct RemoteNode {
+    client: RemoteClient,
+    work_queue: WorkQueue,
+    engines: Vec<(EngineType, u32)>,
+    status: Status,
+    polling_thread: Option<JoinHandle<()>>,
+}
+
+impl RemoteNode {
+    pub async fn new(
+        local_host: String,
+        local_port: u16,
+        remote_host: String,
+        remote_port: u16,
+        work_queue: WorkQueue,
+        engines: Vec<(EngineType, u32)>,
+    ) -> RemoteNode {
+        let client = create_client(
+            local_host,
+            local_port,
+            remote_host,
+            remote_port,
+            RETRY_TIMEOUT,
+        )
+        .await;
+        RemoteNode {
+            client,
+            work_queue,
+            engines,
+            status: Status::INACTIVE,
+            polling_thread: None,
+        }
+    }
+
+    fn start_polling(&mut self) {
+        let queue = self.work_queue.clone();
+        self.polling_thread = Some(thread::spawn(move || {
+            let rt = Builder::new_current_thread()
+                .enable_io()
+                .build()
+                .expect("Failed to create polling thread!");
+            rt.block_on(async {
+                loop {
+                    let (work, dept) = queue.get_work(0);
+                }
+            })
+        }));
+    }
+
+    async fn stop_polling(&mut self) {
+        self.status = Status::STOPPED;
+        if let Some(handle) = self.polling_thread.take() {
+            let res = handle.join();
+            if let Err(err) = res {
+                error!("Polling thread exited with error: {:?}", err);
+            }
+        }
+    }
+
+    pub fn add_remote(
+        &mut self,
+        remote_host: String,
+        remote_port: u16,
+        engines: Vec<(EngineType, u32)>,
+    ) -> DandelionResult<()> {
+        // TODO: add node to list of remotes
+
+        if self.polling_thread.is_none() {
+            self.start_polling();
+        }
+
+        Ok(())
+    }
+
+    pub fn remove_remote(&mut self, remote_host: String, remote_port: u16) -> DandelionResult<()> {
+        // TODO: remove remote from list of remotes
+
+        // TODO: wait for all jobs to complete
+        // TODO: if list becomes empty stop the polling thread
+
+        Ok(())
+    }
+}

machine_interface/Cargo.toml

@@ -17,6 +17,7 @@ backend_debug = []
 reqwest_io = ["std", "dep:reqwest", "dep:http", "dep:bytes", "dep:memcache"]
 bytes_context = ["std", "dep:bytes"]
 timestamp = ["dandelion_commons/timestamp"]
+vendored-ssl = ["dep:openssl"]
 test_export = []
 
 [build-dependencies]
@@ -58,7 +59,7 @@ memcache = { version = "0.18.0", optional = true }
 bytes = { version = "1.6", optional = true}
 # To make sure we can compile everywhere. 
 # This sets it for the other dependencies we have that use it.
-openssl = { version = "0.10", features = ["vendored"] }
+openssl = { version = "0.10", features = ["vendored"], optional = true }
 
 # disable benchmarks in library, to not run all unit tests on every benchmark
 # also needs to be disabled for criterion flags to work that are not available for tests