From 46183095883a24ba10f378188735728b93a0d9b5 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Wed, 24 Jun 2026 13:52:09 +0200 Subject: [PATCH 1/8] Add composition ids into queueing --- dispatcher/src/dispatcher.rs | 79 ++++++--- dispatcher/src/queue.rs | 150 +++++++++++++----- dispatcher/src/queue/policy/base.rs | 21 ++- dispatcher/src/queue/policy/data_locality.rs | 33 +++- .../dispatcher_tests/combination_tests.rs | 2 +- .../tests/dispatcher_tests/function_tests.rs | 16 +- .../tests/dispatcher_tests/registry_tests.rs | 6 +- machine_interface/src/function_driver.rs | 4 +- .../function_driver/system_driver/reqwest.rs | 4 +- .../src/function_driver/test_queue.rs | 12 +- .../src/function_driver/thread_utils.rs | 5 +- multinode/src/client.rs | 64 +++++--- multinode/src/client/test.rs | 1 + 13 files changed, 281 insertions(+), 116 deletions(-) diff --git a/dispatcher/src/dispatcher.rs b/dispatcher/src/dispatcher.rs index 4cba4ee6..d59bc2dc 100644 --- a/dispatcher/src/dispatcher.rs +++ b/dispatcher/src/dispatcher.rs @@ -21,21 +21,27 @@ use machine_interface::memory_domain::ContextTrait; use machine_interface::{ composition::{ get_sharding, AnyShardingMode, Composition, CompositionSet, InputSetDescriptor, - JoinStrategy, LocalCompositionSet, RemoteData, ShardingMode, + JoinStrategy, LocalCompositionSet, ShardingMode, }, function_driver::{Metadata, WorkToDo}, machine_config::{get_available_domains, DomainType, EngineType, IntoEnumIterator}, memory_domain::{MemoryDomain, MemoryResource}, }; -use std::{collections::BTreeMap, sync::Arc}; +use std::{ + collections::BTreeMap, + sync::{atomic::AtomicUsize, Arc}, +}; // TODO also here and in registry replace Arc Box with static references from leaked boxes for things we expect to be there for // the entire execution time anyway +// TODO: the composition ids currently just go up from 0, there is no overflow handling. +// We are not expecting to overflow, but this should be handled for proper deployments. pub struct Dispatcher { function_registry: FunctionRegistry, work_queue: WorkQueue, domains: Vec>>, any_sharding_mode: AnyShardingMode, + composition_id: AtomicUsize, } impl Dispatcher { @@ -68,9 +74,15 @@ impl Dispatcher { work_queue, domains, any_sharding_mode, + composition_id: AtomicUsize::new(0), }); } + pub fn get_composition_id(&self) -> usize { + self.composition_id + .fetch_add(1, std::sync::atomic::Ordering::AcqRel) + } + pub fn insert_function( &self, function_name: String, @@ -105,9 +117,15 @@ impl Dispatcher { ) -> DandelionResult<(Vec>, Recorder)> { debug!("Queuing function {}", function_id); recorder.record(RecordPoint::EnterDispatcher); - + let composition_id = self.get_composition_id(); let results = self - .queue_function(function_id, inputs, caching, recorder.clone()) + .queue_function( + composition_id, + function_id, + inputs, + caching, + recorder.clone(), + ) .await?; // if any set is not local send them trough reference resolution @@ -119,9 +137,12 @@ impl Dispatcher { } }) { self.work_queue - .do_work(WorkToDo::SetsToResolve { - input_sets: results, - }) + .do_work( + WorkToDo::SetsToResolve { + input_sets: results, + }, + composition_id, + ) .await? .get_composition() } else { @@ -136,13 +157,6 @@ impl Dispatcher { Ok((local_results, recorder)) } - pub async fn delete_remote_data(&self, remote_data: RemoteData) -> DandelionResult<()> { - self.work_queue - .do_work(WorkToDo::RemoteToDelete { remote_data }) - .await - .map(|_| ()) - } - pub async fn queue_unregistered_composition( &self, composition_desc: String, @@ -151,6 +165,7 @@ impl Dispatcher { recorder: Recorder, ) -> DandelionResult<(Vec>, Recorder)> { debug!("Parsing single use composition"); + let composition_meta_pairs = self .function_registry .parse_compositions(&composition_desc.as_str())?; @@ -164,12 +179,14 @@ impl Dispatcher { )); } + let composition_id = self.get_composition_id(); debug!( - "Queuing single use composition {}", - composition_meta_pairs[0].0 + "Queuing single use composition {} with id {}", + composition_meta_pairs[0].0, composition_id, ); let results = self .queue_composition( + composition_id, composition_meta_pairs[0].1.clone(), inputs, caching, @@ -186,9 +203,12 @@ impl Dispatcher { } }) { self.work_queue - .do_work(WorkToDo::SetsToResolve { - input_sets: results, - }) + .do_work( + WorkToDo::SetsToResolve { + input_sets: results, + }, + composition_id, + ) .await? .get_composition() } else { @@ -211,6 +231,7 @@ impl Dispatcher { /// * `inputs` vec of input set options, where the index in the vec is the input set number pub async fn queue_composition( &self, + composition_id: usize, composition: Composition, inputs: Vec>, caching: bool, @@ -309,6 +330,7 @@ impl Dispatcher { // start all functions that are ready and insert their sets into the awaited ones for args in ready_functions.into_iter() { awaited_sets.push(Either::Right(self.queue_function_sharded( + composition_id, args.function_id, args.function_index, args.input_sets, @@ -389,6 +411,7 @@ impl Dispatcher { } if args.missing_sets.is_empty() { awaited_sets.push(Either::Right(self.queue_function_sharded( + composition_id, args.function_id, args.function_index, args.input_sets, @@ -423,6 +446,7 @@ impl Dispatcher { /// Also handles sharing of sets async fn queue_function_sharded<'context>( &self, + composition_id: usize, function_id: FunctionId, // index of the function within the composition function_index: usize, @@ -459,6 +483,7 @@ impl Dispatcher { .map(|ins| { let new_recorder = Recorder::new_from_parent(function_id.clone(), &recorder); let future_box = Box::pin(self.queue_function( + composition_id, function_id.clone(), ins, caching, @@ -474,7 +499,13 @@ impl Dispatcher { } else { let new_recorder = Recorder::new_from_parent(function_id.clone(), &recorder); let future_box = self - .queue_function(function_id, vec![], caching, new_recorder.clone()) + .queue_function( + composition_id, + function_id, + vec![], + caching, + new_recorder.clone(), + ) .await .and_then(|result| Ok(vec![result])); recorders = vec![new_recorder]; @@ -518,6 +549,7 @@ impl Dispatcher { /// the index describes which output set the composition belongs to. pub async fn queue_function<'dispatcher>( &'dispatcher self, + composition_id: usize, function_id: FunctionId, input_sets: Vec>, caching: bool, @@ -563,7 +595,11 @@ impl Dispatcher { recorder: recorder.clone(), }; - let sets = self.work_queue.do_work(args).await?.get_composition(); + let sets = self + .work_queue + .do_work(args, composition_id) + .await? + .get_composition(); recorder.record(RecordPoint::FutureReturn); #[cfg(feature = "log_function_stdio")] @@ -608,6 +644,7 @@ impl Dispatcher { } FunctionType::Composition(comp_info) => { self.queue_composition( + composition_id, (*comp_info.composition).clone(), input_sets, caching, diff --git a/dispatcher/src/queue.rs b/dispatcher/src/queue.rs index 4986887d..39372bdf 100644 --- a/dispatcher/src/queue.rs +++ b/dispatcher/src/queue.rs @@ -41,6 +41,8 @@ pub fn get_engine_flag(t: EngineType) -> u32 { } struct ComputeQueueElement { + /// ID of the composition this work belongs to. + composition_id: usize, /// Flags indicating which engines can run this task flags: u32, /// The WorkToDo content of the queue element @@ -50,6 +52,8 @@ struct ComputeQueueElement { } struct IoQueueElement { + /// ID of the composition this work belongs to. + composition_id: usize, /// Flags indicating which engines can run this task flags: u32, /// The WorkToDo content of the queue element @@ -96,7 +100,7 @@ pub struct WorkQueue { /// Tracks current system informations used by the any sharding policy. pub system_info: Arc, /// Channels for asking remote node to take work - remote_nodes: Arc>>>, + remote_nodes: Arc>>>, } struct ComputeWaitFuture<'queue> { @@ -164,7 +168,7 @@ impl<'list> IoWaitFuture<'list> { } impl Future for IoWaitFuture<'_> { - type Output = (WorkToDo, Debt); + type Output = (WorkToDo, Debt, Option); fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll { let mut lock_guard = self.work_queue.inner.lock().unwrap(); @@ -200,22 +204,23 @@ impl Future for IoWaitFuture<'_> { 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: _, - function_alternatives: _, - input_sets: _, - metadata: _, - caching: _, - recorder, - } = &mut result_tupple.0 - { - recorder.record(RecordPoint::IOQueueEnd); - } - + let result = extracted.map(|queue_element| { + ( + queue_element.work, + queue_element.debt, + queue_element.composition_id, + ) + }); + if let Some((mut work, debt, composition_id)) = result { + let composition_id_option = + if let WorkToDo::FunctionArguments { recorder, .. } = &mut work { + recorder.record(RecordPoint::IOQueueEnd); + Some(composition_id) + } else { + None + }; // Found some work, so core is not idle - Poll::Ready(result_tupple) + Poll::Ready((work, debt, composition_id_option)) } else { // Did not find any work, so need to add to waker queue lock_guard.io_waker_list.push_back(cx.waker().clone()); @@ -275,7 +280,14 @@ impl WorkQueue { /// Pushes the work and debt to the back of the queue and sets the flags accordingly. /// Returns an error if the queue is full. /// TODO: check or define here and other places, if the flags need to match fully, just checking that any flag is set would be enough - fn push_compute(&self, mut work: WorkToDo, debt: Debt, flags: u32, had_fetching: bool) { + fn push_compute( + &self, + mut work: WorkToDo, + debt: Debt, + flags: u32, + composition_id: usize, + had_fetching: bool, + ) { if let WorkToDo::FunctionArguments { function_id: _, function_alternatives: _, @@ -302,9 +314,33 @@ impl WorkQueue { if had_fetching { queue_guard.fetching_in_progress -= 1; } - queue_guard - .compute_queue - .push_back(ComputeQueueElement { flags, work, debt }); + + let new_element = ComputeQueueElement { + composition_id, + flags, + work, + debt, + }; + // if list empty or the back has the same or a smaller id just push to the back + if queue_guard.compute_queue.is_empty() + || queue_guard.compute_queue.back().unwrap().composition_id <= composition_id + { + queue_guard.compute_queue.push_back(new_element); + // check the front has same or bigger id so can just push front + } else if queue_guard.compute_queue.front().unwrap().composition_id >= composition_id { + queue_guard.compute_queue.push_front(new_element); + // find the place to insert + } else { + let index = queue_guard + .compute_queue + .iter() + .position(|elem| elem.composition_id > composition_id) + .unwrap(); + let mut tail = queue_guard.compute_queue.split_off(index); + queue_guard.compute_queue.push_back(new_element); + queue_guard.compute_queue.append(&mut tail); + } + // call first waker with matching flags if there are any if let Some(waker_to_call) = queue_guard .compute_waker_list @@ -318,7 +354,14 @@ impl WorkQueue { } } - fn push_io(&self, mut work: WorkToDo, debt: Debt, flags: u32, try_offload: bool) { + fn push_io( + &self, + mut work: WorkToDo, + debt: Debt, + flags: u32, + composition_id: usize, + try_offload: bool, + ) { if let WorkToDo::FunctionArguments { function_id: _, function_alternatives: _, @@ -333,7 +376,7 @@ impl WorkQueue { // Policy: compute element metadata and optionally offload to a remote node let Some((work, debt, policy_data)) = - policy::prepare_io_element(work, debt, try_offload, &self.remote_nodes) + policy::prepare_io_element(work, debt, try_offload, composition_id, &self.remote_nodes) else { // if prepare_io_element returned None it has already offloaded the work to a remote return; @@ -341,6 +384,7 @@ impl WorkQueue { let mut queue_guard = self.inner.lock().unwrap(); let new_element = IoQueueElement { + composition_id, flags, work, debt, @@ -363,7 +407,25 @@ impl WorkQueue { _ => true, }; - queue_guard.io_queue.push_back(new_element); + // if list empty or the back has the same or a smaller id just push to the back + if queue_guard.io_queue.is_empty() + || queue_guard.io_queue.back().unwrap().composition_id <= composition_id + { + queue_guard.io_queue.push_back(new_element); + // check the front has same or smaller id so can just push front + } else if queue_guard.io_queue.front().unwrap().composition_id >= composition_id { + queue_guard.io_queue.push_front(new_element); + // find the place to insert + } else { + let index = queue_guard + .io_queue + .iter() + .position(|elem| elem.composition_id > composition_id) + .unwrap(); + let mut tail = queue_guard.io_queue.split_off(index); + queue_guard.io_queue.push_back(new_element); + queue_guard.io_queue.append(&mut tail); + } // call first waker with matching flags if there are any // only call waker if we know the core wants to take this element @@ -374,18 +436,15 @@ impl WorkQueue { } } - fn push(&self, work: WorkToDo, debt: Debt, try_offload: bool) { + fn push(&self, work: WorkToDo, debt: Debt, composition_id: usize, try_offload: bool) { let (flags, local) = match &work { WorkToDo::Shutdown(engine_type) => (get_engine_flag(*engine_type), true), - WorkToDo::SetsToResolve { input_sets: _ } => (0, false), - WorkToDo::RemoteToDelete { remote_data: _ } => (0, false), + WorkToDo::SetsToResolve { .. } => (0, false), + WorkToDo::RemoteToDelete { .. } => (0, false), WorkToDo::FunctionArguments { - function_id: _, function_alternatives, input_sets, - metadata: _, - caching: _, - recorder: _, + .. } => { // check if all the sets are already fully locally available let local = input_sets.into_iter().all(|set_option| { @@ -416,17 +475,21 @@ impl WorkQueue { ); if local { - self.push_compute(work, debt, flags, false); + self.push_compute(work, debt, flags, composition_id, false); } else { - self.push_io(work, debt, flags, try_offload); + self.push_io(work, debt, flags, composition_id, try_offload); } } /// Inserts the work into the queue setting the flags according to the supported engines and /// awaits the future before returning the result. - pub async fn do_work(&self, work: WorkToDo) -> DandelionResult { + pub async fn do_work( + &self, + work: WorkToDo, + composition_id: usize, + ) -> DandelionResult { let (promise, debt) = self.promise_buffer.get_promise()?; - self.push(work, debt, true); + self.push(work, debt, composition_id, true); return promise.await; } @@ -437,7 +500,7 @@ impl WorkQueue { engine_flags: u32, node_id: u64, number_of_functions: usize, - ) -> Vec<(WorkToDo, Debt)> { + ) -> Vec<(WorkToDo, Debt, usize)> { let mut guard = self.inner.lock().unwrap(); let InnerQueue { compute_queue, @@ -460,7 +523,7 @@ impl WorkQueue { ComputeWaitFuture::new(engine_flags, &self).await } - pub async fn get_io_work(&self) -> (WorkToDo, Debt) { + pub async fn get_io_work(&self) -> (WorkToDo, Debt, Option) { // try to get work, if there is none, insert self into waker and try again IoWaitFuture::new(&self).await } @@ -505,7 +568,7 @@ impl WorkQueue { pub fn add_remote_channel( &self, node_id: u64, - channel: mpsc::UnboundedSender<(WorkToDo, Debt)>, + channel: mpsc::UnboundedSender<(WorkToDo, Debt, usize)>, ) { self.remote_nodes.lock().unwrap().insert(node_id, channel); } @@ -516,8 +579,8 @@ impl WorkQueue { } /// Put work back into queue after trying to offload without success. - pub fn reenqueue(&self, work: WorkToDo, debt: Debt) { - self.push(work, debt, false); + pub fn reenqueue(&self, work: WorkToDo, debt: Debt, composition_id: usize) { + self.push(work, debt, composition_id, false); } /// Increases the number of remote cores. @@ -588,11 +651,11 @@ impl EngineWorkQueue for EngineQueue { fn get_io_engine_args( &self, - ) -> impl Future { + ) -> impl Future)> { self.work_queue.get_io_work() } - fn requeu_engine_args(&self, work: WorkToDo, debt: Debt) { + fn requeu_engine_args(&self, work: WorkToDo, debt: Debt, composition_id: usize) { let flags = match &work { WorkToDo::FunctionArguments { function_id: _, @@ -607,7 +670,8 @@ impl EngineWorkQueue for EngineQueue { _ => panic!("Should not reenqueue non function arguments"), }; trace!("Reenqueue with flags: {}", flags); - self.work_queue.push_compute(work, debt, flags, true) + self.work_queue + .push_compute(work, debt, flags, composition_id, true) } fn remove_self_from_queue(&self) { diff --git a/dispatcher/src/queue/policy/base.rs b/dispatcher/src/queue/policy/base.rs index 6f4601a2..d074e2e0 100644 --- a/dispatcher/src/queue/policy/base.rs +++ b/dispatcher/src/queue/policy/base.rs @@ -17,7 +17,8 @@ pub fn prepare_io_element( work: WorkToDo, debt: Debt, _try_offload: bool, - _remote_nodes: &std::sync::Mutex>>, + _composition_id: usize, + _remote_nodes: &std::sync::Mutex>>, ) -> Option<(WorkToDo, Debt, IOElementData)> { Some((work, debt, IOElementData)) } @@ -41,7 +42,7 @@ pub fn get_work_for_remote( _node_id: u64, number_of_functions: usize, queue_state_decrease: &impl Fn(), -) -> Vec<(WorkToDo, Debt)> { +) -> Vec<(WorkToDo, Debt, usize)> { let mut functions = Vec::with_capacity(number_of_functions); // first check the queue with unresolved references, since those are easier to steal functions.extend( @@ -59,7 +60,13 @@ pub fn get_work_for_remote( false } }) - .map(|queue_element| (queue_element.work, queue_element.debt)) + .map(|queue_element| { + ( + queue_element.work, + queue_element.debt, + queue_element.composition_id, + ) + }) .take(number_of_functions), ); // did not find enough work in the io_queue so check compute queue @@ -82,7 +89,13 @@ pub fn get_work_for_remote( false } }) - .map(|queue_element| (queue_element.work, queue_element.debt)) + .map(|queue_element| { + ( + queue_element.work, + queue_element.debt, + queue_element.composition_id, + ) + }) .take(still_needed), ); } diff --git a/dispatcher/src/queue/policy/data_locality.rs b/dispatcher/src/queue/policy/data_locality.rs index 93e913ba..588762d7 100644 --- a/dispatcher/src/queue/policy/data_locality.rs +++ b/dispatcher/src/queue/policy/data_locality.rs @@ -19,7 +19,8 @@ pub fn prepare_io_element( work: WorkToDo, debt: Debt, try_offload: bool, - remote_nodes: &std::sync::Mutex>>, + composition_id: usize, + remote_nodes: &std::sync::Mutex>>, ) -> Option<(WorkToDo, Debt, IOElementData)> { let (remote_data, total_input_size) = if let WorkToDo::FunctionArguments { ref input_sets, .. } = work { @@ -56,9 +57,9 @@ pub fn prepare_io_element( if let Some(node_sender) = maybe_sender { // If the remote node has disconnected its receiver is gone, in which case // we recover the work and fall back to executing it locally. - match node_sender.send((work, debt)) { + match node_sender.send((work, debt, composition_id)) { Ok(()) => return None, - Err(mpsc::error::SendError((work, debt))) => { + Err(mpsc::error::SendError((work, debt, composition_id))) => { // since the sender does not work, remove it. remote_nodes.lock().unwrap().remove(&node_id); return Some(( @@ -110,7 +111,7 @@ pub fn get_work_for_remote( node_id: u64, number_of_functions: usize, queue_state_decrease: &impl Fn(), -) -> Vec<(WorkToDo, Debt)> { +) -> Vec<(WorkToDo, Debt, usize)> { let mut functions = Vec::with_capacity(number_of_functions); // go through all the input sets and find those with the most data already on the node asking for work functions.extend( @@ -135,7 +136,13 @@ pub fn get_work_for_remote( false } }) - .map(|queue_element| (queue_element.work, queue_element.debt)) + .map(|queue_element| { + ( + queue_element.work, + queue_element.debt, + queue_element.composition_id, + ) + }) .take(number_of_functions), ); @@ -157,7 +164,13 @@ pub fn get_work_for_remote( false } }) - .map(|queue_element| (queue_element.work, queue_element.debt)) + .map(|queue_element| { + ( + queue_element.work, + queue_element.debt, + queue_element.composition_id, + ) + }) .take(still_needed), ); } @@ -182,7 +195,13 @@ pub fn get_work_for_remote( false } }) - .map(|queue_element| (queue_element.work, queue_element.debt)) + .map(|queue_element| { + ( + queue_element.work, + queue_element.debt, + queue_element.composition_id, + ) + }) .take(still_needed), ); } diff --git a/dispatcher/tests/dispatcher_tests/combination_tests.rs b/dispatcher/tests/dispatcher_tests/combination_tests.rs index 2bf8b3c2..b205f22e 100644 --- a/dispatcher/tests/dispatcher_tests/combination_tests.rs +++ b/dispatcher/tests/dispatcher_tests/combination_tests.rs @@ -112,7 +112,7 @@ pub fn fetch_compute( let mut result_sets = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_composition(composition, inputs, false, recorder)) + .block_on(dispatcher.queue_composition(0, composition, inputs, false, recorder)) .unwrap(); assert_eq!(1, result_sets.len()); diff --git a/dispatcher/tests/dispatcher_tests/function_tests.rs b/dispatcher/tests/dispatcher_tests/function_tests.rs index dc6159c5..d40be785 100644 --- a/dispatcher/tests/dispatcher_tests/function_tests.rs +++ b/dispatcher/tests/dispatcher_tests/function_tests.rs @@ -34,7 +34,7 @@ pub fn single_domain_and_engine_basic( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_function(function_id, Vec::new(), false, recorder)); + .block_on(dispatcher.queue_function(0, function_id, Vec::new(), false, recorder)); match result { Ok(_) => (), Err(err) => panic!("Failed with: {:?}", err), @@ -80,7 +80,7 @@ pub fn single_domain_and_engine_matmul( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_function(function_id, inputs, false, recorder)); + .block_on(dispatcher.queue_function(0, function_id, inputs, false, recorder)); let out_sets = match result { Ok(context) => context, Err(err) => panic!("Failed with: {:?}", err), @@ -146,7 +146,7 @@ pub fn composition_single_matmul( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_composition(composition, inputs, false, recorder)); + .block_on(dispatcher.queue_composition(0, composition, inputs, false, recorder)); let out_contexts = match result { Ok(context) => context, Err(err) => panic!("Failed with: {:?}", err), @@ -171,7 +171,7 @@ fn composition_option_helper( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_composition(composition, inputs, false, recorder)); + .block_on(dispatcher.queue_composition(0, composition, inputs, false, recorder)); let out_contexts = match result { Ok(context) => context, Err(err) => panic!("Failed with: {:?}", err), @@ -355,7 +355,7 @@ pub fn composition_parallel_matmul( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_composition(composition, inputs, false, recorder)); + .block_on(dispatcher.queue_composition(0, composition, inputs, false, recorder)); let out_vec = match result { Ok(v) => v, Err(err) => panic!("Failed with: {:?}", err), @@ -435,7 +435,7 @@ pub fn composition_chain_matmul( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_composition(composition, inputs, false, recorder)); + .block_on(dispatcher.queue_composition(0, composition, inputs, false, recorder)); let out_contexts = match result { Ok(context) => context, Err(err) => panic!("Failed with: {:?}", err), @@ -642,7 +642,7 @@ pub fn composition_diamond_matmac( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_composition(composition, inputs, false, recorder)); + .block_on(dispatcher.queue_composition(0, composition, inputs, false, recorder)); let out_contexts = match result { Ok(context) => context, Err(err) => panic!("Failed with: {:?}", err), @@ -783,7 +783,7 @@ pub fn composition_chain_large_matmac( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_composition(composition, inputs, false, recorder)); + .block_on(dispatcher.queue_composition(0, composition, inputs, false, recorder)); let out_contexts = match result { Ok(context) => context, Err(err) => panic!("Failed with: {:?}", err), diff --git a/dispatcher/tests/dispatcher_tests/registry_tests.rs b/dispatcher/tests/dispatcher_tests/registry_tests.rs index 04643af0..ad7d344b 100644 --- a/dispatcher/tests/dispatcher_tests/registry_tests.rs +++ b/dispatcher/tests/dispatcher_tests/registry_tests.rs @@ -115,12 +115,13 @@ pub fn single_input_fixed( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_function(function_id.clone(), inputs, false, recorder)); + .block_on(dispatcher.queue_function(0, function_id.clone(), inputs, false, recorder)); recorder = Recorder::new(function_id.clone(), Instant::now()); let overwrite_result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() .block_on(dispatcher.queue_function( + 0, function_id.clone(), overwrite_inputs, false, @@ -235,12 +236,13 @@ pub fn multiple_input_fixed( let result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() - .block_on(dispatcher.queue_function(function_id.clone(), inputs, false, recorder)); + .block_on(dispatcher.queue_function(0, function_id.clone(), inputs, false, recorder)); recorder = Recorder::new(function_id.clone(), Instant::now()); let overwrite_result = tokio::runtime::Builder::new_current_thread() .build() .unwrap() .block_on(dispatcher.queue_function( + 0, function_id.clone(), overwrite_inputs, false, diff --git a/machine_interface/src/function_driver.rs b/machine_interface/src/function_driver.rs index d5af05df..91ffa253 100644 --- a/machine_interface/src/function_driver.rs +++ b/machine_interface/src/function_driver.rs @@ -76,10 +76,10 @@ pub trait EngineWorkQueue { /// Unstable: This is a temproary addition to the interface use with caution fn get_io_engine_args( &self, - ) -> impl std::future::Future + Send; + ) -> impl std::future::Future)> + Send; /// Function to return a Work to do to the queue after fetching all reference sets /// Unstable: This is a temproary addition to the interface use with caution - fn requeu_engine_args(&self, work: WorkToDo, debt: crate::promise::Debt); + fn requeu_engine_args(&self, work: WorkToDo, debt: crate::promise::Debt, composition_id: usize); fn remove_self_from_queue(&self); } diff --git a/machine_interface/src/function_driver/system_driver/reqwest.rs b/machine_interface/src/function_driver/system_driver/reqwest.rs index bce79793..ec25671e 100644 --- a/machine_interface/src/function_driver/system_driver/reqwest.rs +++ b/machine_interface/src/function_driver/system_driver/reqwest.rs @@ -630,7 +630,7 @@ async fn engine_loop(queue: impl EngineWorkQueue + Clone + Send + 'static) -> De loop { let ticket = semaphore.clone().acquire_owned().await.unwrap(); debug!("IO engine loop has ticket"); - let (args, debt) = queue.get_io_engine_args().await; + let (args, debt, id_option) = queue.get_io_engine_args().await; debug!("IO engine loop has work"); match args { WorkToDo::FunctionArguments { @@ -645,6 +645,7 @@ async fn engine_loop(queue: impl EngineWorkQueue + Clone + Send + 'static) -> De let queue_clone = queue.clone(); debug!("Resolving references for call to {}", function_id); recorder.record(RecordPoint::FetchingStart); + let composition_id = id_option.unwrap(); resolve_all_sets( client_clone, @@ -669,6 +670,7 @@ async fn engine_loop(queue: impl EngineWorkQueue + Clone + Send + 'static) -> De recorder, }, debt, + composition_id, ), }; }, diff --git a/machine_interface/src/function_driver/test_queue.rs b/machine_interface/src/function_driver/test_queue.rs index 18dd1f19..88b6c2a5 100644 --- a/machine_interface/src/function_driver/test_queue.rs +++ b/machine_interface/src/function_driver/test_queue.rs @@ -35,11 +35,17 @@ impl EngineWorkQueue for TestQueue { async fn get_compute_engine_args(&self) -> (WorkToDo, Debt) { self.reciever.lock().await.recv().await.unwrap() } - async fn get_io_engine_args(&self) -> (WorkToDo, Debt) { - self.reciever.lock().await.recv().await.unwrap() + async fn get_io_engine_args(&self) -> (WorkToDo, Debt, Option) { + let (work, debt) = self.reciever.lock().await.recv().await.unwrap(); + (work, debt, Some(0)) } - fn requeu_engine_args(&self, _work: WorkToDo, _debt: crate::promise::Debt) { + fn requeu_engine_args( + &self, + _work: WorkToDo, + _debt: crate::promise::Debt, + _composition_id: usize, + ) { panic!("should not requeue on the test queue"); } diff --git a/machine_interface/src/function_driver/thread_utils.rs b/machine_interface/src/function_driver/thread_utils.rs index d817ad1c..f9174601 100644 --- a/machine_interface/src/function_driver/thread_utils.rs +++ b/machine_interface/src/function_driver/thread_utils.rs @@ -122,7 +122,6 @@ mod waker { pub(super) fn manual_pull(queue: &mut impl EngineWorkQueue) -> (WorkToDo, Debt) { let mut queue_future = core::pin::pin!(queue.get_compute_engine_args()); - // let (sender, receiver) = channel::<()>(); let sender_box = Box::new(sender); let raw_waker = RawWaker::new(Box::into_raw(sender_box) as *const (), &WAKER_TABLE); @@ -253,10 +252,10 @@ fn run_thread(core_id: u8, mut queue: impl EngineWorkQueue) { }); debt.fulfill(results); } - WorkToDo::SetsToResolve { input_sets: _ } => { + WorkToDo::SetsToResolve { .. } => { panic!("Compute engine should never get sets to resolve") } - WorkToDo::RemoteToDelete { remote_data: _ } => { + WorkToDo::RemoteToDelete { .. } => { panic!("Compute engine should never get remote data to delete") } WorkToDo::Shutdown(_) => { diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 940c2b2e..efebff36 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -155,7 +155,7 @@ async fn receive_message( enum QueueOption { Message(remote_message::RemoteMessage, Option), WorkAvailable, - TryOffload(WorkToDo, machine_interface::promise::Debt), + TryOffload(WorkToDo, machine_interface::promise::Debt, usize), /// The connection to the remote node was lost, so the server logic should tear down. Disconnected, } @@ -220,14 +220,17 @@ async fn remote_queue_server_sender( /// Translating the messages from the queue for offlaoding into something the server logic understands async fn remote_queue_server_try_offload( - mut queue_receiver: mpsc::UnboundedReceiver<(WorkToDo, Debt)>, + mut queue_receiver: mpsc::UnboundedReceiver<(WorkToDo, Debt, usize)>, sender: mpsc::Sender, queue: WorkQueue, ) { - while let Some((work, debt)) = queue_receiver.recv().await { - if let Err(send_err) = sender.send(QueueOption::TryOffload(work, debt)).await { - if let SendError(QueueOption::TryOffload(w, d)) = send_err { - queue.reenqueue(w, d); + while let Some((work, debt, composition_id)) = queue_receiver.recv().await { + if let Err(send_err) = sender + .send(QueueOption::TryOffload(work, debt, composition_id)) + .await + { + if let SendError(QueueOption::TryOffload(w, d, id)) = send_err { + queue.reenqueue(w, d, id); } break; } @@ -235,8 +238,8 @@ async fn remote_queue_server_try_offload( // can't send anymore so make sure the channel does not have things added to it. queue_receiver.close(); // drain the channel, reqenque all the work - while let Some((work, debt)) = queue_receiver.recv().await { - queue.reenqueue(work, debt); + while let Some((work, debt, id)) = queue_receiver.recv().await { + queue.reenqueue(work, debt, id); } } @@ -287,7 +290,7 @@ async fn remote_queue_server_logic( // TODO: consider limiting the work we give to a node based on the know max capacity, // to limit potential stragglers if we know the node asked for more than it can handle (possibly because of race conditions) // do not give even more. - invocations.extend(work_found.into_iter().map(|(work, debt)| + invocations.extend(work_found.into_iter().map(|(work, debt, composition_id)| { // there is some work so send it out // find the local function id to use @@ -336,6 +339,7 @@ async fn remote_queue_server_logic( debt_map.insert( promise_id, ( + composition_id, debt, new_recorder, start_reference, @@ -384,10 +388,16 @@ async fn remote_queue_server_logic( response, } = response; // TODO: handle failure - let (debt, mut recorder, start_epoch, remote_data_references, work) = - debt_map.remove(&invocation_id).expect( - "Should always get back function response for a present debt", - ); + let ( + composition_id, + debt, + mut recorder, + start_epoch, + remote_data_references, + work, + ) = debt_map + .remove(&invocation_id) + .expect("Should always get back function response for a present debt"); free_debt_ids.push(invocation_id); drop(remote_data_references); // remote did not do work, was a try offload request, reenqueu the work @@ -417,7 +427,7 @@ async fn remote_queue_server_logic( debt.fulfill(result) } else { // did not get response so need to reenqueue the work - queue.reenqueue(work, debt); + queue.reenqueue(work, debt, composition_id); } } remote_message::RemoteMessage::NodeUpdate(node_update) => { @@ -445,10 +455,10 @@ async fn remote_queue_server_logic( } } } - QueueOption::TryOffload(work, debt) => { + QueueOption::TryOffload(work, debt, composition_id) => { // if this node already sent enough work for the remote to be at capacity don't send more if invocations_running >= remote_num_cores as usize { - queue.reenqueue(work, debt); + queue.reenqueue(work, debt, composition_id); continue; } invocations_running += 1; @@ -494,6 +504,7 @@ async fn remote_queue_server_logic( debt_map.insert( promise_id, ( + composition_id, debt, new_recorder, start_reference, @@ -549,15 +560,19 @@ async fn remote_queue_server_logic( () } // Reenqueue work that was tried to offload - QueueOption::TryOffload(work, debt) => { - queue.reenqueue(work, debt); + QueueOption::TryOffload(work, debt, composition_id) => { + queue.reenqueue(work, debt, composition_id); } } } // Recover any work that was offloaded but never completed, so it can be re-scheduled locally or on another node. - for (_promise_id, (debt, _recorder, _start_epoch, _remote_data_references, work)) in debt_map { - queue.reenqueue(work, debt); + for ( + _promise_id, + (composition_id, debt, _recorder, _start_epoch, _remote_data_references, work), + ) in debt_map + { + queue.reenqueue(work, debt, composition_id); } } @@ -753,8 +768,15 @@ async fn dispatcher_call( caching: bool, recorder: Recorder, ) { + let composition_id = dispatcher.get_composition_id(); let function_result = dispatcher - .queue_function(function_id, input_sets, caching, recorder.clone()) + .queue_function( + composition_id, + function_id, + input_sets, + caching, + recorder.clone(), + ) .await; let response_message = match function_result { Ok(sets) => { diff --git a/multinode/src/client/test.rs b/multinode/src/client/test.rs index 69acd5de..ea0cea4a 100644 --- a/multinode/src/client/test.rs +++ b/multinode/src/client/test.rs @@ -49,6 +49,7 @@ async fn mock_dispatcher(work_queue: WorkQueue, engine_type: machine_config::Eng caching: false, recorder: Recorder::new(function_id, Instant::now()), }, + 0, ) .await; let error = if let Err(error) = result { From 98fd5a9d5954396657eb90de7aa87c291a728392 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Tue, 30 Jun 2026 14:51:29 +0200 Subject: [PATCH 2/8] Deduplicate work request sending --- multinode/src/client.rs | 126 +++++++++------------------------------- 1 file changed, 26 insertions(+), 100 deletions(-) diff --git a/multinode/src/client.rs b/multinode/src/client.rs index efebff36..5ad24efc 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -832,38 +832,13 @@ async fn remote_queue_client_logic( queue_message::QueueMessage::NoWork(true) => { trace!("Queue Client recieved NoWork(true)"); debug_assert!(data_option.is_none()); - - remote_had_work = false + remote_had_work = false; } // remote signals it may have work so can ask for it, if we have capacity queue_message::QueueMessage::NoWork(false) => { trace!("Queue Client recieved NoWork(false)"); debug_assert!(data_option.is_none()); - - // let current_jobs_queued = *local_queue_state.borrow(); - let occupancy = std::cmp::max(queue_state, work_from_remote); - if occupancy < num_local_cores { - let engines = EngineType::iter() - .map(|engine_type| proto::Engine { - engine_type: engine_type_dtop(engine_type) as i32, - engine_capacity: (num_local_cores - occupancy) as u32, - }) - .collect(); - if message_sender - .send(remote_message::RemoteMessage::WorkRequest( - RepeatedEngines { engines }, - )) - .await - .is_err() - { - break; - } - remote_had_work = false; - } else { - // Only set to true, if we did not send out a message already, - // otherwise avoid retriggering sending a message on state change, before we get answer. - remote_had_work = true; - } + remote_had_work = true; } // TODO for try offload decide when to refuse work queue_message::QueueMessage::TryOffload(invocation) => { @@ -872,7 +847,6 @@ async fn remote_queue_client_logic( work_from_remote += 1; // mark remote as having work, so we ask for more as idle cores change - remote_had_work = true; let start_instance = Instant::now(); let start_time = std::time::SystemTime::elapsed(&std::time::SystemTime::UNIX_EPOCH) @@ -896,12 +870,12 @@ async fn remote_queue_client_logic( caching, recorder, ); + remote_had_work = true; } queue_message::QueueMessage::Invocations(invocations) => { trace!("Queue Client recieved invocation"); // mark remote as having work, so we ask for more as idle cores change - remote_had_work = true; work_from_remote += invocations.invocations.len(); let start_instance = Instant::now(); let start_time = @@ -930,6 +904,7 @@ async fn remote_queue_client_logic( recorder, ) } + remote_had_work = true; } } } @@ -949,29 +924,6 @@ async fn remote_queue_client_logic( if message_sender.send(results).await.is_err() { break; } - let occupancy = std::cmp::max(queue_state, work_from_remote); - if remote_had_work && occupancy < num_local_cores { - let engines: Vec<_> = EngineType::iter() - .map(|engine_type| proto::Engine { - engine_type: engine_type_dtop(engine_type) as i32, - engine_capacity: (num_local_cores - occupancy) as u32, - }) - .collect(); - - trace!("Asking for more work"); - if message_sender - .send(remote_message::RemoteMessage::WorkRequest( - RepeatedEngines { engines }, - )) - .await - .is_err() - { - break; - } - trace!("Finished sending the message asking for more work"); - // set false, to avoid double sending if multiple cores become idle, but did not have a response in between - remote_had_work = false; - } } // getting a notification so should poll the queue PollingOption::QueueStateChanged(current_queue_state) => { @@ -981,30 +933,6 @@ async fn remote_queue_client_logic( remote_had_work, ); queue_state = current_queue_state; - // send message to get work if we have not asked already - let occupancy = std::cmp::max(queue_state, work_from_remote); - if remote_had_work && occupancy < num_local_cores { - let engines: Vec<_> = EngineType::iter() - .map(|engine_type| proto::Engine { - engine_type: engine_type_dtop(engine_type) as i32, - engine_capacity: (num_local_cores - occupancy) as u32, - }) - .collect(); - - trace!("Asking for more work"); - if message_sender - .send(remote_message::RemoteMessage::WorkRequest( - RepeatedEngines { engines }, - )) - .await - .is_err() - { - break; - } - trace!("Finished sending the message asking for more work"); - // set false, to avoid double sending if multiple cores become idle, but did not have a response in between - remote_had_work = false; - } } PollingOption::LocalCoreCountChanged(new_core_number) => { trace!("Sending new local core count: {}", new_core_number); @@ -1018,32 +946,30 @@ async fn remote_queue_client_logic( { break; } - // check if we now want to get more work - let occupancy = std::cmp::max(queue_state, work_from_remote); - if remote_had_work && occupancy < num_local_cores { - let engines: Vec<_> = EngineType::iter() - .map(|engine_type| proto::Engine { - engine_type: engine_type_dtop(engine_type) as i32, - engine_capacity: (num_local_cores - occupancy) as u32, - }) - .collect(); - - trace!("Asking for more work"); - if message_sender - .send(remote_message::RemoteMessage::WorkRequest( - RepeatedEngines { engines }, - )) - .await - .is_err() - { - break; - } - trace!("Finished sending the message asking for more work"); - // set false, to avoid double sending if multiple cores become idle, but did not have a response in between - remote_had_work = false; - } } } + let occupancy = std::cmp::max(queue_state, work_from_remote); + if remote_had_work && occupancy < num_local_cores { + trace!("Asking for more work"); + let engines = EngineType::iter() + .map(|engine_type| proto::Engine { + engine_type: engine_type_dtop(engine_type) as i32, + engine_capacity: (num_local_cores - occupancy) as u32, + }) + .collect(); + if message_sender + .send(remote_message::RemoteMessage::WorkRequest( + RepeatedEngines { engines }, + )) + .await + .is_err() + { + break; + } + trace!("Finished sending message asking for more work"); + // set false, to avoid double sending if multiple cores become idle, but did not have a response in between + remote_had_work = false; + } } // Reaching here means the connection was lost; the caller will attempt to reconnect. trace!("remote_queue_client_logic exited, connection to remote was lost"); From 6ec05f01b90085dddd441141c362a012767f2d27 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Tue, 30 Jun 2026 16:22:35 +0200 Subject: [PATCH 3/8] Add prefetching to multinode --- dispatcher/src/queue.rs | 21 ++- dispatcher/src/queue/policy/base.rs | 4 +- dispatcher/src/queue/policy/data_locality.rs | 6 +- multinode/src/client.rs | 18 +- multinode/src/client/test.rs | 181 +++++++++++-------- 5 files changed, 136 insertions(+), 94 deletions(-) diff --git a/dispatcher/src/queue.rs b/dispatcher/src/queue.rs index 39372bdf..98b25778 100644 --- a/dispatcher/src/queue.rs +++ b/dispatcher/src/queue.rs @@ -1,4 +1,5 @@ mod policy; +pub use policy::PREFETCH_PER_CORE; use dandelion_commons::{ err_dandelion, records::RecordPoint, DandelionError, DandelionResult, DispatcherError, @@ -79,7 +80,7 @@ struct InnerQueue { /// List of io engines ready to take more work io_waker_list: LinkedList, /// The number of functions for which we are currently prefetching - fetching_in_progress: usize, + prefetching_in_progress: usize, } const MAX_QUEUE: usize = 4096; @@ -178,7 +179,7 @@ impl Future for IoWaitFuture<'_> { // to find work where they can do their own fetching instead. let compute_pending = lock_guard.compute_queue.len(); let local_cores = *self.work_queue.system_info.num_local_cores_watcher.borrow(); - let active_fetch_count = lock_guard.fetching_in_progress; + let active_fetch_count = lock_guard.prefetching_in_progress; let idle_compute_cores = lock_guard.compute_waker_list.len(); let extracted = lock_guard .io_queue @@ -202,7 +203,7 @@ impl Future for IoWaitFuture<'_> { }) ); if is_prefetching { - lock_guard.fetching_in_progress += 1; + lock_guard.prefetching_in_progress += 1; } let result = extracted.map(|queue_element| { ( @@ -240,7 +241,7 @@ impl WorkQueue { io_queue: LinkedList::new(), compute_waker_list: LinkedList::new(), io_waker_list: LinkedList::new(), - fetching_in_progress: 0, + prefetching_in_progress: 0, })), promise_buffer: PromiseBuffer::init(MAX_QUEUE), queue_state_sender, @@ -255,14 +256,14 @@ impl WorkQueue { } } - pub fn queue_state_watcher(&self) -> watch::Receiver { - self.queue_state_receiver.clone() - } - pub fn queueing_notifier(&self) -> Arc { self.queuing_notifier.clone() } + pub fn queue_state_watcher(&self) -> watch::Receiver { + self.queue_state_receiver.clone() + } + fn queue_state_decrease(&self) { self.queue_state_sender.send_if_modified(|current_state| { *current_state = current_state.saturating_sub(1); @@ -312,7 +313,7 @@ impl WorkQueue { let mut queue_guard = self.inner.lock().unwrap(); if had_fetching { - queue_guard.fetching_in_progress -= 1; + queue_guard.prefetching_in_progress -= 1; } let new_element = ComputeQueueElement { @@ -394,7 +395,7 @@ impl WorkQueue { // if not, no reason to call waker let compute_length = queue_guard.compute_queue.len(); let local_cores = *self.system_info.num_local_cores_watcher.borrow(); - let already_fetching = queue_guard.fetching_in_progress; + let already_fetching = queue_guard.prefetching_in_progress; let idle_compute_cores = queue_guard.compute_waker_list.len(); let would_process = match &new_element.work { WorkToDo::FunctionArguments { .. } => policy::should_io_take( diff --git a/dispatcher/src/queue/policy/base.rs b/dispatcher/src/queue/policy/base.rs index d074e2e0..5739db26 100644 --- a/dispatcher/src/queue/policy/base.rs +++ b/dispatcher/src/queue/policy/base.rs @@ -5,7 +5,7 @@ use tokio::sync::mpsc; use super::super::{ComputeQueueElement, IoQueueElement}; -const LOCAL_WORK_PER_CORE: usize = 2; +pub const PREFETCH_PER_CORE: usize = 1; /// Empty struct since the base policy does not use additional information for scheduling decisions. pub struct IOElementData; @@ -31,7 +31,7 @@ pub fn should_io_take( local_cores: usize, _idle_compute_cores: usize, ) -> bool { - compute_pending + active_fetch_count < LOCAL_WORK_PER_CORE * local_cores + compute_pending + active_fetch_count < PREFETCH_PER_CORE * local_cores } /// Selects work items from the local queues to hand off to a remote node. diff --git a/dispatcher/src/queue/policy/data_locality.rs b/dispatcher/src/queue/policy/data_locality.rs index 588762d7..2c5a57e9 100644 --- a/dispatcher/src/queue/policy/data_locality.rs +++ b/dispatcher/src/queue/policy/data_locality.rs @@ -5,7 +5,7 @@ use tokio::sync::mpsc; use super::super::{ComputeQueueElement, IoQueueElement}; -const LOCAL_WORK_PER_CORE: usize = 2; +pub const PREFETCH_PER_CORE: usize = 1; pub struct IOElementData { pub remote_data: BTreeMap, @@ -59,7 +59,7 @@ pub fn prepare_io_element( // we recover the work and fall back to executing it locally. match node_sender.send((work, debt, composition_id)) { Ok(()) => return None, - Err(mpsc::error::SendError((work, debt, composition_id))) => { + Err(mpsc::error::SendError((work, debt, _))) => { // since the sender does not work, remove it. remote_nodes.lock().unwrap().remove(&node_id); return Some(( @@ -98,7 +98,7 @@ pub fn should_io_take( // additionally want to prevent fetching, if there is remote data and no local core is idle // 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 + || (compute_pending + active_fetch_count < PREFETCH_PER_CORE * local_cores && element_data.remote_data.is_empty()) } diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 5ad24efc..121f74ff 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -814,6 +814,7 @@ async fn remote_queue_client_logic( bool, Recorder, ), + prefetch_multiplier: usize, export_registry: ExportRegistry, mut num_local_cores: usize, ) { @@ -948,15 +949,23 @@ async fn remote_queue_client_logic( } } } - let occupancy = std::cmp::max(queue_state, work_from_remote); - if remote_had_work && occupancy < num_local_cores { - trace!("Asking for more work"); + if remote_had_work { + let occupancy = std::cmp::max(queue_state, work_from_remote); + // ask for work for the local idle cores + let engine_number = if occupancy < num_local_cores { + num_local_cores - occupancy + } else if occupancy < num_local_cores + num_local_cores * prefetch_multiplier { + num_local_cores + num_local_cores * prefetch_multiplier - occupancy + } else { + continue; + }; let engines = EngineType::iter() .map(|engine_type| proto::Engine { engine_type: engine_type_dtop(engine_type) as i32, - engine_capacity: (num_local_cores - occupancy) as u32, + engine_capacity: engine_number as u32, }) .collect(); + trace!("Asking for more work"); if message_sender .send(remote_message::RemoteMessage::WorkRequest( RepeatedEngines { engines }, @@ -1056,6 +1065,7 @@ pub async fn remote_queue_client( recorder, )); }, + dispatcher::queue::PREFETCH_PER_CORE, export_registry, local_core_count, ) diff --git a/multinode/src/client/test.rs b/multinode/src/client/test.rs index ea0cea4a..ed23bd7d 100644 --- a/multinode/src/client/test.rs +++ b/multinode/src/client/test.rs @@ -303,6 +303,7 @@ fn test_remote_queue_client() { poll_option_receiver, remote_message_sender, dispatcher_send, + 0, ExportRegistry::new(1), 0, )); @@ -319,7 +320,7 @@ fn test_remote_queue_client() { remote_message => panic!("Expected work request not {:?}", remote_message), } - // update local queue state + // update local queue state, expect message asking for work poll_option_sender .try_send(crate::client::PollingOption::QueueStateChanged(0)) .unwrap(); @@ -443,12 +444,11 @@ fn test_remote_queue_client() { } Poll::Pending | Poll::Ready(None) => panic!("Should receive work now"), }; - // notify that queue state has changed, should not trigger asking for more work + // Notify that queue state has changed, should not trigger asking for more work, + // since prefetch capacity is 0. poll_option_sender .try_send(crate::client::PollingOption::QueueStateChanged(3)) .unwrap(); - - // should not lead to asking for more work assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); assert!(dispatcher_receiver.is_empty()); @@ -494,7 +494,7 @@ fn test_remote_queue_client() { remote_message => panic!("Expected work request not {:?}", remote_message), } - // send back that there is none + // send back that there is some poll_option_sender .try_send(crate::client::PollingOption::Message( Ok(queue_message::QueueMessage::NoWork(true)), @@ -524,73 +524,104 @@ fn test_remote_queue_client() { } } -// #[test_log::test] -// fn test_combined() { -// // create socket connecting the two sides -// let (client_socket, server_socket) = tokio::io::duplex(4096); - -// // create variable needed for server side -// let work_queue = WorkQueue::init(); -// work_queue.add_local_cores(2); -// let engine_type = machine_config::EngineType::iter().next().unwrap(); - -// // create variable needed for client side -// let (dispatcher_sender, mut dispatcher_receiver) = mpsc::channel(1); - -// // spawn both on a new runtime -// let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap(); -// let (remote_data_deletion_sender, _remote_data_deletion_receiver) = mpsc::unbounded_channel(); -// runtime.spawn(remote_queue_server( -// client_socket, -// work_queue.clone(), -// ExportRegistry::new(1), -// remote_data_deletion_sender, -// )); -// runtime.spawn(remote_queue_client( -// server_socket, -// dispatcher_sender, -// ExportRegistry::new(2), -// work_queue.clone(), -// )); - -// // create one waiting engine -// let mut context = Context::from_waker(Waker::noop()); -// let mut engine_future = Box::pin(work_queue.get_compute_work(get_engine_flag(engine_type))); -// assert!(match engine_future.poll_unpin(&mut context) { -// Poll::Pending => true, -// _ => false, -// }); - -// // send work on the work queue -// let mut test_dispatcher_future = Box::pin(mock_dispatcher(work_queue.clone(), engine_type)); -// assert_eq!( -// Poll::Pending, -// test_dispatcher_future.poll_unpin(&mut context) -// ); - -// // should now have something on the dispatcher receiver -// let callback = match dispatcher_receiver.blocking_recv().unwrap() { -// DispatcherCommand::RemoteFunctionRequest { -// function_id, -// inputs, -// is_cold, -// recorder: _, -// callback, -// } => { -// assert_eq!("dummy_function", function_id.as_str()); -// assert_eq!(0, inputs.len()); -// assert!(is_cold); -// callback -// } -// _ => panic!("Received unexpeceted dispatcher command"), -// }; - -// // send back a result -// assert!(callback.send(err_dandelion!(EXPECTED_ERROR)).is_ok()); -// loop { -// match test_dispatcher_future.poll_unpin(&mut context) { -// Poll::Pending => (), -// Poll::Ready(()) => break, -// } -// } -// } +#[test_log::test] +fn test_remote_queue_client_prefetch() { + // constants used in the test + let expected_function_id = "dummy function".to_string(); + const INVOCATION_ID: u32 = 7; + + let (dispatcher_sender, mut dispatcher_receiver) = mpsc::channel(64); + let (poll_option_sender, poll_option_receiver) = mpsc::channel(64); + let (remote_message_sender, mut remote_message_receiver) = mpsc::channel(64); + + let dispatcher_send = + |registry, duration, invocation_id, function_id, inputs, is_cold, recorder| { + dispatcher_sender + .blocking_send(( + registry, + duration, + invocation_id, + function_id, + inputs, + is_cold, + recorder, + )) + .unwrap(); + }; + + let mut context = Context::from_waker(Waker::noop()); + let mut client_future = Box::pin(remote_queue_client_logic( + poll_option_receiver, + remote_message_sender, + dispatcher_send, + 1, + ExportRegistry::new(1), + 0, + )); + + // send message with the number of local cores and that the local queue state has changed + poll_option_sender + .try_send(crate::client::PollingOption::LocalCoreCountChanged(1)) + .unwrap(); + + assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); + match remote_message_receiver.try_recv().unwrap() { + remote_message::RemoteMessage::NodeUpdate(update) => assert_eq!(1, update.num_local_cores), + remote_message => panic!("Expected work request not {:?}", remote_message), + } + + // expect to ask for work for 1 core + assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); + match remote_message_receiver.try_recv().unwrap() { + remote_message::RemoteMessage::WorkRequest(engines) => { + assert!(engines.engines.len() > 0); + assert_eq!(1, engines.engines[0].engine_capacity); + } + remote_message => panic!("Expected work request not {:?}", remote_message), + } + + // send back work for 1 core + poll_option_sender + .try_send(crate::client::PollingOption::Message( + Ok(queue_message::QueueMessage::Invocations( + RepeatedInvocations { + invocations: vec![Invocation { + invocation_id: INVOCATION_ID, + function_id: expected_function_id.clone(), + metadata_sets: vec![], + caching: true, + }], + }, + )), + None, + )) + .unwrap(); + + // poll client and check dispatcher queue for the work that was received + assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); + let _invocation_id_1 = match dispatcher_receiver.poll_recv(&mut context) { + Poll::Ready(Some(( + _registry, + _duration, + invocation_id, + function_id, + _inputs, + is_cold, + _recorder, + ))) => { + assert!(!is_cold); + assert_eq!(expected_function_id, function_id.as_str()); + invocation_id + } + Poll::Pending | Poll::Ready(None) => panic!("Should receive work now"), + }; + + // expect it to ask for 1 more work given the prefetch capacity + match remote_message_receiver.try_recv().unwrap() { + remote_message::RemoteMessage::WorkRequest(engines) => { + assert!(engines.engines.len() > 0); + assert_eq!(1, engines.engines[0].engine_capacity); + } + remote_message => panic!("Expected work request not {:?}", remote_message), + } +} From 821baa7f2c41f7825f6d17ceb254ea17555b9cbb Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Wed, 1 Jul 2026 11:47:00 +0200 Subject: [PATCH 4/8] Fix prefetching according to number of idle cores --- dispatcher/src/queue.rs | 51 +++++++-------- dispatcher/src/queue/policy/base.rs | 3 - dispatcher/src/queue/policy/data_locality.rs | 4 -- multinode/src/client.rs | 64 ++++++++++-------- multinode/src/client/test.rs | 69 +++++++++++++------- 5 files changed, 107 insertions(+), 84 deletions(-) diff --git a/dispatcher/src/queue.rs b/dispatcher/src/queue.rs index 98b25778..ccbb7594 100644 --- a/dispatcher/src/queue.rs +++ b/dispatcher/src/queue.rs @@ -93,9 +93,8 @@ pub struct WorkQueue { /// Holds the two queues, first one for work ready to be run locally, second one for engines waiting for fitting work to arrive inner: Arc>, promise_buffer: PromiseBuffer, - /// Used to keep track of the total number of jobs currently in the queue - queue_state_sender: watch::Sender, - queue_state_receiver: watch::Receiver, + /// Notifier to indicate that the number of idle cores has changed + idle_notifier: Arc, /// Notifier to send out notification that queueing is happening queuing_notifier: Arc, /// Tracks current system informations used by the any sharding policy. @@ -105,20 +104,28 @@ pub struct WorkQueue { } struct ComputeWaitFuture<'queue> { + was_idle: bool, flags: u32, work_queue: &'queue WorkQueue, } impl<'list> ComputeWaitFuture<'list> { fn new(flags: u32, work_queue: &'list WorkQueue) -> ComputeWaitFuture<'list> { - Self { flags, work_queue } + Self { + was_idle: false, + flags, + work_queue, + } } } impl Future for ComputeWaitFuture<'_> { type Output = (WorkToDo, Debt); - fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll { + fn poll( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> Poll { let mut lock_guard = self.work_queue.inner.lock().unwrap(); // check if there is any work with the flags we are looking for let result = lock_guard @@ -131,7 +138,6 @@ impl Future for ComputeWaitFuture<'_> { if let Some(waker) = lock_guard.io_waker_list.pop_front() { waker.wake(); } - self.work_queue.queue_state_decrease(); if let WorkToDo::FunctionArguments { function_id: _, @@ -144,6 +150,9 @@ impl Future for ComputeWaitFuture<'_> { { recorder.record(RecordPoint::ComputeQueueEnd); } + if self.was_idle { + self.work_queue.idle_notifier.notify_waiters(); + } Poll::Ready(result_tupple) } else { // Did not find any work, so need to add to waker queue @@ -152,6 +161,11 @@ impl Future for ComputeWaitFuture<'_> { waker: cx.waker().clone(), }; lock_guard.compute_waker_list.push_back(waker_element); + // have newly become idle + if !self.was_idle { + self.work_queue.idle_notifier.notify_waiters(); + self.was_idle = true; + } // lock was ready once, need to set new one Poll::Pending } @@ -233,7 +247,6 @@ impl Future for IoWaitFuture<'_> { impl WorkQueue { /// Creates a new WorkQueue of given size. pub fn init() -> Self { - let (queue_state_sender, queue_state_receiver) = watch::channel(0); let (num_local_cores_sender, num_local_cores_watcher) = watch::channel(0); WorkQueue { inner: Arc::new(Mutex::new(InnerQueue { @@ -244,8 +257,7 @@ impl WorkQueue { prefetching_in_progress: 0, })), promise_buffer: PromiseBuffer::init(MAX_QUEUE), - queue_state_sender, - queue_state_receiver, + idle_notifier: Arc::new(Notify::new()), queuing_notifier: Arc::new(Notify::new()), system_info: Arc::new(SystemInfo { num_local_cores_watcher, @@ -260,22 +272,12 @@ impl WorkQueue { self.queuing_notifier.clone() } - pub fn queue_state_watcher(&self) -> watch::Receiver { - self.queue_state_receiver.clone() + pub fn idle_notifier(&self) -> Arc { + self.idle_notifier.clone() } - fn queue_state_decrease(&self) { - self.queue_state_sender.send_if_modified(|current_state| { - *current_state = current_state.saturating_sub(1); - *current_state < *self.system_info.num_local_cores_watcher.borrow() - }); - } - - fn queue_state_increase(&self) { - self.queue_state_sender.send_if_modified(|current_state| { - *current_state += 1; - *current_state < *self.system_info.num_local_cores_watcher.borrow() - }); + pub fn idle_cores(&self) -> usize { + self.inner.lock().unwrap().compute_waker_list.len() } /// Pushes the work and debt to the back of the queue and sets the flags accordingly. @@ -463,8 +465,6 @@ impl WorkQueue { for alternative in function_alternatives { flags |= get_engine_flag(alternative.engine); } - // only want to count the actual functions to execute - self.queue_state_increase(); // check if system flag is set, if so, put in system queue (flags, local) } @@ -514,7 +514,6 @@ impl WorkQueue { engine_flags, node_id, number_of_functions, - &|| self.queue_state_decrease(), ) } diff --git a/dispatcher/src/queue/policy/base.rs b/dispatcher/src/queue/policy/base.rs index 5739db26..e1cb87e8 100644 --- a/dispatcher/src/queue/policy/base.rs +++ b/dispatcher/src/queue/policy/base.rs @@ -41,7 +41,6 @@ pub fn get_work_for_remote( engine_flags: u32, _node_id: u64, number_of_functions: usize, - queue_state_decrease: &impl Fn(), ) -> Vec<(WorkToDo, Debt, usize)> { let mut functions = Vec::with_capacity(number_of_functions); // first check the queue with unresolved references, since those are easier to steal @@ -50,7 +49,6 @@ pub fn get_work_for_remote( .extract_if(|queue_element| { if let WorkToDo::FunctionArguments { recorder, .. } = &mut queue_element.work { if queue_element.flags & engine_flags != 0 { - queue_state_decrease(); recorder.record(RecordPoint::IOQueueEnd); true } else { @@ -78,7 +76,6 @@ pub fn get_work_for_remote( if let WorkToDo::FunctionArguments { recorder, .. } = &mut queue_element.work { if queue_element.flags & engine_flags != 0 { recorder.record(RecordPoint::ComputeQueueEnd); - queue_state_decrease(); // don't need to poke IO cores to do more prefetching, // since those queues are empty if we are taking from here true diff --git a/dispatcher/src/queue/policy/data_locality.rs b/dispatcher/src/queue/policy/data_locality.rs index 2c5a57e9..894aa670 100644 --- a/dispatcher/src/queue/policy/data_locality.rs +++ b/dispatcher/src/queue/policy/data_locality.rs @@ -110,7 +110,6 @@ pub fn get_work_for_remote( engine_flags: u32, node_id: u64, number_of_functions: usize, - queue_state_decrease: &impl Fn(), ) -> Vec<(WorkToDo, Debt, usize)> { let mut functions = Vec::with_capacity(number_of_functions); // go through all the input sets and find those with the most data already on the node asking for work @@ -127,7 +126,6 @@ pub fn get_work_for_remote( .unwrap_or(false); if queue_element.flags & engine_flags != 0 && node_has_most_data { recorder.record(RecordPoint::IOQueueEnd); - queue_state_decrease(); true } else { false @@ -155,7 +153,6 @@ pub fn get_work_for_remote( if let WorkToDo::FunctionArguments { recorder, .. } = &mut queue_element.work { if queue_element.flags & engine_flags != 0 { recorder.record(RecordPoint::IOQueueEnd); - queue_state_decrease(); true } else { false @@ -184,7 +181,6 @@ pub fn get_work_for_remote( if let WorkToDo::FunctionArguments { recorder, .. } = &mut queue_element.work { if queue_element.flags & engine_flags != 0 { recorder.record(RecordPoint::ComputeQueueEnd); - queue_state_decrease(); // Don't need to wake IO cores to do more prefetching, // since those queues must be empty if we are taking from here. true diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 121f74ff..462e23f0 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -669,7 +669,7 @@ pub async fn remote_queue_server( pub enum PollingOption { Message(DandelionResult, Option), - QueueStateChanged(usize), + IdleChanged, LocalCoreCountChanged(usize), // Results(RemoteMessage, Option<(Vec>, u64)>), Results(remote_message::RemoteMessage), @@ -723,18 +723,16 @@ async fn remote_queue_client_sender( // TODO: think about limiting number of notification, to make sure we are not adding // additional load when the queue is filled / emptied in big strides async fn remote_queue_client_queue_state( - mut receiver: watch::Receiver, + receiver: Arc, sender: mpsc::Sender, ) { + // Send a initial idle changed message, to make sure it always asks once. + // An idle machine otherwise would never start asking for work + sender.send(PollingOption::IdleChanged).await.unwrap(); loop { - receiver.changed().await.unwrap(); - trace!("received local queue state"); - let queue_state = *receiver.borrow_and_update(); - if sender - .send(PollingOption::QueueStateChanged(queue_state)) - .await - .is_err() - { + receiver.notified().await; + trace!("received capacity notification"); + if sender.send(PollingOption::IdleChanged).await.is_err() { break; } } @@ -814,12 +812,12 @@ async fn remote_queue_client_logic( bool, Recorder, ), + idle_cores: impl Fn() -> usize, prefetch_multiplier: usize, export_registry: ExportRegistry, mut num_local_cores: usize, ) { let mut remote_had_work = true; - let mut queue_state = 0; // This makes sure we don't overfetch from this node. // TODO: think about the general issue of state synchronization between the dispatcher and multinode client adding things, // and the engines and server taking things. @@ -927,13 +925,11 @@ async fn remote_queue_client_logic( } } // getting a notification so should poll the queue - PollingOption::QueueStateChanged(current_queue_state) => { + PollingOption::IdleChanged => { trace!( - "Queue Client checking updated queue state: {:?}, remote_had_work {}", - current_queue_state, + "Queue Client checking updated queue state, remote_had_work {}", remote_had_work, ); - queue_state = current_queue_state; } PollingOption::LocalCoreCountChanged(new_core_number) => { trace!("Sending new local core count: {}", new_core_number); @@ -950,12 +946,28 @@ async fn remote_queue_client_logic( } } if remote_had_work { - let occupancy = std::cmp::max(queue_state, work_from_remote); - // ask for work for the local idle cores - let engine_number = if occupancy < num_local_cores { - num_local_cores - occupancy - } else if occupancy < num_local_cores + num_local_cores * prefetch_multiplier { - num_local_cores + num_local_cores * prefetch_multiplier - occupancy + // ! This currently assumes all works in the queue comes from a single remote master. + // ! The assumption starts to break if there are things directly enqueued or come from multiple master. + // ! The reason we need this assumption is, because otherwise, the client cannot get a good bound + // ! of how full the the queue is, since the dispatcher may not have processed new invocations added by + // ! a message received to the work queue yet when we perform this check. + let idle = idle_cores(); + // Want to make sure we fill the prefetching buffer. + let max_prefetch = num_local_cores * prefetch_multiplier; + // The amount of work that is in the system from the remote, need to subtract the amount actually running on a compute core + // everything else is still in prefetching + let current_prefetch = work_from_remote.saturating_sub(num_local_cores - idle); + // If we have 0 prefetching, ask for as much as we have idle cores + let engine_number = if prefetch_multiplier == 0 { + if work_from_remote < num_local_cores { + num_local_cores - work_from_remote + } else { + continue; + } + // otherwise check how much prefetch capacity we have and check if we have already reached it + } else if max_prefetch > current_prefetch { + max_prefetch - current_prefetch + // if neither do not ask for more work } else { continue; }; @@ -1019,12 +1031,6 @@ pub async fn remote_queue_client( } trace!("Queue Client sent out initial message"); - // Create a second copy of the watcher to wait on asynchronously, marke changed to check once in the beginning, - // in case there are already idle cores - let mut local_queue_state = queue.queue_state_watcher(); - // needs to be marked changed, so a server with an empty queue and already stead number of local cores starts to poll - local_queue_state.mark_changed(); - // start sender loop let (remote_message_sender, remote_message_reciever) = mpsc::channel(64); let sender_handle = spawn(remote_queue_client_sender( @@ -1043,8 +1049,9 @@ pub async fn remote_queue_client( poll_option_sender.clone(), )); // spawn queue state loop + let local_idle_notifier = queue.idle_notifier(); let queue_state_handle = spawn(remote_queue_client_queue_state( - local_queue_state, + local_idle_notifier, poll_option_sender.clone(), )); @@ -1065,6 +1072,7 @@ pub async fn remote_queue_client( recorder, )); }, + || queue.idle_cores(), dispatcher::queue::PREFETCH_PER_CORE, export_registry, local_core_count, diff --git a/multinode/src/client/test.rs b/multinode/src/client/test.rs index ed23bd7d..60ed5daa 100644 --- a/multinode/src/client/test.rs +++ b/multinode/src/client/test.rs @@ -1,5 +1,8 @@ use std::{ - sync::{Arc, RwLock}, + sync::{ + atomic::{AtomicUsize, Ordering::SeqCst}, + Arc, RwLock, + }, task::Poll, time::Instant, }; @@ -283,6 +286,9 @@ fn test_remote_queue_client() { let (poll_option_sender, poll_option_receiver) = mpsc::channel(64); let (remote_message_sender, mut remote_message_receiver) = mpsc::channel(64); + let idle_cores = Arc::new(AtomicUsize::new(0)); + let idle_clone = idle_cores.clone(); + let dispatcher_send = |registry, duration, invocation_id, function_id, inputs, caching, recorder| { dispatcher_sender @@ -303,6 +309,7 @@ fn test_remote_queue_client() { poll_option_receiver, remote_message_sender, dispatcher_send, + || idle_clone.load(SeqCst), 0, ExportRegistry::new(1), 0, @@ -312,6 +319,7 @@ fn test_remote_queue_client() { poll_option_sender .try_send(crate::client::PollingOption::LocalCoreCountChanged(3)) .unwrap(); + idle_cores.store(3, SeqCst); // receive the message about the updated core count assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); @@ -320,11 +328,7 @@ fn test_remote_queue_client() { remote_message => panic!("Expected work request not {:?}", remote_message), } - // update local queue state, expect message asking for work - poll_option_sender - .try_send(crate::client::PollingOption::QueueStateChanged(0)) - .unwrap(); - assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); + // Should also send a work request for idle cores with the node update match remote_message_receiver.try_recv().unwrap() { remote_message::RemoteMessage::WorkRequest(engines) => { assert!(engines.engines.len() > 0); @@ -394,13 +398,7 @@ fn test_remote_queue_client() { Poll::Pending | Poll::Ready(None) => panic!("Should receive work now"), }; - // notify that queue state has changed - poll_option_sender - .try_send(crate::client::PollingOption::QueueStateChanged(2)) - .unwrap(); - - // should now send request for more work - assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); + // expect it to ask for 1 more immediately when receiveing the work and seeing there is still capacity to prefetch match remote_message_receiver.try_recv().unwrap() { remote_message::RemoteMessage::WorkRequest(engines) => { assert!(engines.engines.len() > 0); @@ -409,6 +407,19 @@ fn test_remote_queue_client() { remote_message => panic!("Expected work request not {:?}", remote_message), } + // notify that number of idle cores has changed + idle_cores.store(1, SeqCst); + poll_option_sender + .try_send(crate::client::PollingOption::IdleChanged) + .unwrap(); + + // should now send request for more work, since it should prefetch one + assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); + // should not get another message asking for work, since still waiting for the last + remote_message_receiver + .try_recv() + .expect_err("Should not have another message yet"); + // send work poll_option_sender .try_send(crate::client::PollingOption::Message( @@ -444,13 +455,17 @@ fn test_remote_queue_client() { } Poll::Pending | Poll::Ready(None) => panic!("Should receive work now"), }; - // Notify that queue state has changed, should not trigger asking for more work, - // since prefetch capacity is 0. + // expecting no more message for work, since prefetching is set to 0 + remote_message_receiver.try_recv().unwrap_err(); + + // Notify that no more cores are idle, still should not trigger more work + idle_cores.store(0, SeqCst); poll_option_sender - .try_send(crate::client::PollingOption::QueueStateChanged(3)) + .try_send(crate::client::PollingOption::IdleChanged) .unwrap(); assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); assert!(dispatcher_receiver.is_empty()); + remote_message_receiver.try_recv().unwrap_err(); // send back a result, mark the queue as changed and poll to get it processed poll_option_sender @@ -463,10 +478,7 @@ fn test_remote_queue_client() { }), )) .unwrap(); - - poll_option_sender - .try_send(crate::client::PollingOption::QueueStateChanged(2)) - .unwrap(); + idle_cores.store(1, SeqCst); assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); @@ -487,14 +499,13 @@ fn test_remote_queue_client() { remote_message => panic!("Expected reponse not {:?}", remote_message), } - // check that client send out a request for more. - // (since it got the result and the queue change, should be able to process both) + // check that client send out a request for more, since it should have capacity again match remote_message_receiver.try_recv().unwrap() { remote_message::RemoteMessage::WorkRequest(_) => (), remote_message => panic!("Expected work request not {:?}", remote_message), } - // send back that there is some + // send back that there is none poll_option_sender .try_send(crate::client::PollingOption::Message( Ok(queue_message::QueueMessage::NoWork(true)), @@ -534,6 +545,9 @@ fn test_remote_queue_client_prefetch() { let (poll_option_sender, poll_option_receiver) = mpsc::channel(64); let (remote_message_sender, mut remote_message_receiver) = mpsc::channel(64); + let idle_cores = Arc::new(AtomicUsize::new(0)); + let idle_clone = idle_cores.clone(); + let dispatcher_send = |registry, duration, invocation_id, function_id, inputs, is_cold, recorder| { dispatcher_sender @@ -554,6 +568,7 @@ fn test_remote_queue_client_prefetch() { poll_option_receiver, remote_message_sender, dispatcher_send, + || idle_clone.load(SeqCst), 1, ExportRegistry::new(1), 0, @@ -563,6 +578,7 @@ fn test_remote_queue_client_prefetch() { poll_option_sender .try_send(crate::client::PollingOption::LocalCoreCountChanged(1)) .unwrap(); + idle_cores.store(1, SeqCst); assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); match remote_message_receiver.try_recv().unwrap() { @@ -616,6 +632,13 @@ fn test_remote_queue_client_prefetch() { Poll::Pending | Poll::Ready(None) => panic!("Should receive work now"), }; + // notify the client that the number of idle cores went to 0, so should now prefetch one more + idle_cores.store(0, SeqCst); + poll_option_sender + .try_send(crate::client::PollingOption::IdleChanged) + .unwrap(); + assert_eq!(Poll::Pending, client_future.poll_unpin(&mut context)); + // expect it to ask for 1 more work given the prefetch capacity match remote_message_receiver.try_recv().unwrap() { remote_message::RemoteMessage::WorkRequest(engines) => { From 262df537380d9921a0a7df2474c1010e30130da5 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Wed, 1 Jul 2026 16:16:18 +0200 Subject: [PATCH 5/8] Fix overcounting prefetching --- dispatcher/src/queue.rs | 44 ++++++++++++++++++++++++++++++++--------- multinode/src/client.rs | 33 +++++++++++++++++++++++-------- 2 files changed, 60 insertions(+), 17 deletions(-) diff --git a/dispatcher/src/queue.rs b/dispatcher/src/queue.rs index ccbb7594..c489ec4d 100644 --- a/dispatcher/src/queue.rs +++ b/dispatcher/src/queue.rs @@ -127,6 +127,10 @@ impl Future for ComputeWaitFuture<'_> { cx: &mut std::task::Context<'_>, ) -> Poll { let mut lock_guard = self.work_queue.inner.lock().unwrap(); + trace!( + "Compute trying to take task with {} in queue", + lock_guard.compute_queue.len() + ); // check if there is any work with the flags we are looking for let result = lock_guard .compute_queue @@ -150,9 +154,7 @@ impl Future for ComputeWaitFuture<'_> { { recorder.record(RecordPoint::ComputeQueueEnd); } - if self.was_idle { - self.work_queue.idle_notifier.notify_waiters(); - } + self.work_queue.idle_notifier.notify_one(); Poll::Ready(result_tupple) } else { // Did not find any work, so need to add to waker queue @@ -163,10 +165,9 @@ impl Future for ComputeWaitFuture<'_> { lock_guard.compute_waker_list.push_back(waker_element); // have newly become idle if !self.was_idle { - self.work_queue.idle_notifier.notify_waiters(); + self.work_queue.idle_notifier.notify_one(); self.was_idle = true; } - // lock was ready once, need to set new one Poll::Pending } } @@ -187,6 +188,10 @@ impl Future for IoWaitFuture<'_> { fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll { let mut lock_guard = self.work_queue.inner.lock().unwrap(); + trace!( + "Io trying to take task with {} in queue", + lock_guard.io_queue.len() + ); // Always take work that is not FunctionArguments. // Take FunctionArguments only if the local queue is smaller than a certain threshold. // Do this to avoid overloading the IO cores with too much parallel fetching and to make it easier for remotes @@ -216,9 +221,6 @@ impl Future for IoWaitFuture<'_> { .. }) ); - if is_prefetching { - lock_guard.prefetching_in_progress += 1; - } let result = extracted.map(|queue_element| { ( queue_element.work, @@ -227,6 +229,9 @@ impl Future for IoWaitFuture<'_> { ) }); if let Some((mut work, debt, composition_id)) = result { + if is_prefetching { + lock_guard.prefetching_in_progress += 1; + } let composition_id_option = if let WorkToDo::FunctionArguments { recorder, .. } = &mut work { recorder.record(RecordPoint::IOQueueEnd); @@ -234,10 +239,26 @@ impl Future for IoWaitFuture<'_> { } else { None }; + trace!( + "Io did find task, {} in queue, idle_compute_cores {}, compute_pending {}, active_fetch_count {}, local_cores {}", + lock_guard.io_queue.len(), + idle_compute_cores, + compute_pending, + active_fetch_count, + local_cores, + ); // Found some work, so core is not idle Poll::Ready((work, debt, composition_id_option)) } else { // Did not find any work, so need to add to waker queue + trace!( + "Io did not find task, {} in queue, idle_compute_cores {}, compute_pending {}, active_fetch_count {}, local_cores {}", + lock_guard.io_queue.len(), + idle_compute_cores, + compute_pending, + active_fetch_count, + local_cores, + ); lock_guard.io_waker_list.push_back(cx.waker().clone()); Poll::Pending } @@ -491,7 +512,7 @@ impl WorkQueue { ) -> DandelionResult { let (promise, debt) = self.promise_buffer.get_promise()?; self.push(work, debt, composition_id, true); - return promise.await; + promise.await } /// Tries to acquire some work that matches the given flags starting from the head of the queue. @@ -508,6 +529,11 @@ impl WorkQueue { io_queue, .. } = &mut *guard; + trace!( + "Trying to get work for remote from io queue ({}), compute queue ({})", + io_queue.len(), + compute_queue.len() + ); policy::get_work_for_remote( io_queue, compute_queue, diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 462e23f0..c42fc94d 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -124,7 +124,6 @@ async fn receive_message( ) -> std::io::Result<(Bytes, Option)> { let packed_metadata = receiver.read_u64().await?; let (metadata_size, _) = unpack_metadata_size_and_flags(packed_metadata); - trace!("strart receiving: {}", metadata_size); // new buffer with size of message let mut metadata_buffer = BytesMut::with_capacity(metadata_size as usize); @@ -135,7 +134,6 @@ async fn receive_message( return Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof)); } } - trace!("finish receiving"); // Keep for when we want to send additional data long with requests // if (flags & ADDITIONAL_DATA_BUFFER) != 0 { @@ -254,6 +252,7 @@ async fn remote_queue_server_logic( mut remote_num_cores: u64, ) { let mut waiting_for_work = false; + // TODO remove, can also just look at the size of the dept map let mut invocations_running = 0; let mut debt_map = BTreeMap::new(); @@ -382,7 +381,10 @@ async fn remote_queue_server_logic( remote_message::RemoteMessage::Response(response) => { invocations_running -= 1; debug_assert!(data_option.is_none()); - trace!("Queue Server received response"); + trace!( + "Queue Server received response, outstanding: {}", + invocations_running + ); let Response { invocation_id, response, @@ -818,6 +820,7 @@ async fn remote_queue_client_logic( mut num_local_cores: usize, ) { let mut remote_had_work = true; + let mut invocation_request_in_flight = false; // This makes sure we don't overfetch from this node. // TODO: think about the general issue of state synchronization between the dispatcher and multinode client adding things, // and the engines and server taking things. @@ -831,6 +834,8 @@ async fn remote_queue_client_logic( queue_message::QueueMessage::NoWork(true) => { trace!("Queue Client recieved NoWork(true)"); debug_assert!(data_option.is_none()); + assert!(invocation_request_in_flight); + invocation_request_in_flight = false; remote_had_work = false; } // remote signals it may have work so can ask for it, if we have capacity @@ -872,7 +877,12 @@ async fn remote_queue_client_logic( remote_had_work = true; } queue_message::QueueMessage::Invocations(invocations) => { - trace!("Queue Client recieved invocation"); + trace!( + "Queue Client recieved {} invocations", + invocations.invocations.len() + ); + assert!(invocation_request_in_flight); + invocation_request_in_flight = false; // mark remote as having work, so we ask for more as idle cores change work_from_remote += invocations.invocations.len(); @@ -918,8 +928,10 @@ async fn remote_queue_client_logic( break; } PollingOption::Results(results) => { - trace!("Queue Client sending out result"); work_from_remote -= 1; + trace!("Queue Client sending out result, {} outstanding", { + work_from_remote + }); if message_sender.send(results).await.is_err() { break; } @@ -945,7 +957,12 @@ async fn remote_queue_client_logic( } } } - if remote_had_work { + trace!( + "remote had work: {}, invocation in flight: {}", + remote_had_work, + invocation_request_in_flight + ); + if remote_had_work && !invocation_request_in_flight { // ! This currently assumes all works in the queue comes from a single remote master. // ! The assumption starts to break if there are things directly enqueued or come from multiple master. // ! The reason we need this assumption is, because otherwise, the client cannot get a good bound @@ -977,7 +994,7 @@ async fn remote_queue_client_logic( engine_capacity: engine_number as u32, }) .collect(); - trace!("Asking for more work"); + trace!("Asking for more work for {} engines", engine_number); if message_sender .send(remote_message::RemoteMessage::WorkRequest( RepeatedEngines { engines }, @@ -989,7 +1006,7 @@ async fn remote_queue_client_logic( } trace!("Finished sending message asking for more work"); // set false, to avoid double sending if multiple cores become idle, but did not have a response in between - remote_had_work = false; + invocation_request_in_flight = true; } } // Reaching here means the connection was lost; the caller will attempt to reconnect. From 9c5ad94aebf729eebec34fe58a076190b8482e39 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Thu, 2 Jul 2026 10:55:58 +0200 Subject: [PATCH 6/8] Update dependencies --- Cargo.lock | 463 ++++++++++++++++------------------------------------- 1 file changed, 142 insertions(+), 321 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f37e77c6..8e9a9168 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -88,15 +88,15 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "ar_archive_writer" -version = "0.5.1" +version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7eb93bbb63b9c227414f6eb3a0adfddca591a8ce1e9b60661bb08969b87e340b" +checksum = "4087686b4b0a3427190bae57a1d9a478dbb2d40c5dc1bd6e2b6d797913bdd348" dependencies = [ "object", ] @@ -113,9 +113,9 @@ dependencies = [ [[package]] name = "assert_cmd" -version = "2.2.1" +version = "2.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39bae1d3fa576f7c6519514180a72559268dd7d1fe104070956cb687bc6673bd" +checksum = "2aa3a22042e45de04255c7bf3626e239f450200fd0493c1e382263544b20aea6" dependencies = [ "anstyle", "bstr", @@ -145,9 +145,9 @@ dependencies = [ [[package]] name = "autocfg" -version = "1.5.0" +version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "base64" @@ -163,15 +163,15 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a" [[package]] name = "bitflags" -version = "2.11.1" +version = "2.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" +checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" [[package]] name = "bitvec" -version = "1.0.1" +version = "1.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bc2832c24239b0141d5674bb9174f9d68a8b5b3f2753311927c172ca46f7e9c" +checksum = "ddcec3d12c579d40898fe0a9a358a803c23e9c52ca3c425707f81c9436211837" dependencies = [ "funty", "radium", @@ -204,20 +204,20 @@ dependencies = [ [[package]] name = "bstr" -version = "1.12.1" +version = "1.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63044e1ae8e69f3b5a92c736ca6269b8d12fa7efe39bf34ddb06d102cf0e2cab" +checksum = "5cee35f73844aa3014bb606320a6c1f010249dbdf43342fe54b5a4f6a8ed4b79" dependencies = [ "memchr", "regex-automata", - "serde", + "serde_core", ] [[package]] name = "bumpalo" -version = "3.20.2" +version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" [[package]] name = "byteorder" @@ -227,9 +227,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" [[package]] name = "bytes" -version = "1.11.1" +version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593" [[package]] name = "cast" @@ -239,9 +239,9 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "cc" -version = "1.2.62" +version = "1.2.65" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1dce859f0832a7d088c4f1119888ab94ef4b5d6795d1ce05afb7fe159d79f98" +checksum = "e228eec9be7c17ccb640b59b36a5cd805ea2a564a4c5e162c2f659fea30d3b96" dependencies = [ "find-msvc-tools", "shlex", @@ -515,9 +515,6 @@ name = "deranged" version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" -dependencies = [ - "powerfmt", -] [[package]] name = "difflib" @@ -541,9 +538,9 @@ dependencies = [ [[package]] name = "displaydoc" -version = "0.2.5" +version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" +checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", @@ -560,9 +557,9 @@ dependencies = [ [[package]] name = "either" -version = "1.15.0" +version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" [[package]] name = "encoding_rs" @@ -587,11 +584,12 @@ dependencies = [ [[package]] name = "env_filter" -version = "1.0.1" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32e90c2accc4b07a8456ea0debdc2e7587bdd890680d71173a15d4ae604f6eef" +checksum = "900d271a03799a1ee8d1ca9b19893b48ca674a9284fefcfb85f05e74ed314217" dependencies = [ "log", + "regex", ] [[package]] @@ -609,9 +607,9 @@ dependencies = [ [[package]] name = "env_logger" -version = "0.11.10" +version = "0.11.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0621c04f2196ac3f488dd583365b9c09be011a4ab8b9f37248ffcc8f6198b56a" +checksum = "de671bd27a75a797dc9ae289ba1e77276e75e2026408aab65185384e2d5cd3f6" dependencies = [ "anstream", "anstyle", @@ -812,22 +810,20 @@ dependencies = [ [[package]] name = "getrandom" -version = "0.4.2" +version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", - "wasip2", - "wasip3", ] [[package]] name = "h2" -version = "0.4.14" +version = "0.4.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "171fefbc92fe4a4de27e0698d6a5b392d6a0e333506bc49133760b3bcf948733" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" dependencies = [ "atomic-waker", "bytes", @@ -913,9 +909,9 @@ checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" [[package]] name = "http" -version = "1.4.0" +version = "1.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a" +checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" dependencies = [ "bytes", "itoa", @@ -964,9 +960,9 @@ checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424" [[package]] name = "hyper" -version = "1.9.0" +version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6299f016b246a94207e63da54dbe807655bf9e00044f73ded42c3ac5305fbcca" +checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498" dependencies = [ "atomic-waker", "bytes", @@ -1122,12 +1118,6 @@ dependencies = [ "zerovec", ] -[[package]] -name = "id-arena" -version = "2.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" - [[package]] name = "idna" version = "1.1.0" @@ -1167,8 +1157,6 @@ checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", "hashbrown 0.17.1", - "serde", - "serde_core", ] [[package]] @@ -1229,21 +1217,20 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "js-sys" -version = "0.3.98" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67df7112613f8bfd9150013a0314e196f4800d3201ae742489d999db2f979f08" +checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102" dependencies = [ "cfg-if", "futures-util", - "once_cell", "wasm-bindgen", ] [[package]] name = "kvm-bindings" -version = "0.14.0" +version = "0.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b3c06ff73c7ce03e780887ec2389d62d2a2a9ddf471ab05c2ff69207cd3f3b4" +checksum = "11cf0ca75d59e9d298647c59cf6c5286fa048120caa77972a7a504a0824d234f" dependencies = [ "vmm-sys-util", ] @@ -1254,7 +1241,7 @@ version = "0.24.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "333f77a20344a448f3f70664918135fddeb804e938f28a99d685bd92926e0b19" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", "kvm-bindings", "libc", "vmm-sys-util", @@ -1266,12 +1253,6 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" -[[package]] -name = "leb128fmt" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" - [[package]] name = "libc" version = "0.2.186" @@ -1301,9 +1282,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.29" +version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" dependencies = [ "serde_core", ] @@ -1363,9 +1344,9 @@ dependencies = [ [[package]] name = "memchr" -version = "2.8.0" +version = "2.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" [[package]] name = "mime" @@ -1375,9 +1356,9 @@ checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" [[package]] name = "mio" -version = "1.2.0" +version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" +checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", "wasi", @@ -1429,11 +1410,11 @@ dependencies = [ [[package]] name = "nix" -version = "0.31.2" +version = "0.31.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d6d0705320c1e6ba1d912b5e37cf18071b6c2e9b7fa8215a1e8a7651966f5d3" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", "cfg-if", "cfg_aliases", "libc", @@ -1450,9 +1431,9 @@ dependencies = [ [[package]] name = "num-conv" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c6673768db2d862beb9b39a78fdcb1a69439615d5794a1be50caa9bc92c81967" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" [[package]] name = "num-traits" @@ -1502,11 +1483,11 @@ checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" [[package]] name = "openssl" -version = "0.10.79" +version = "0.10.81" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf0b434746ee2832f4f0baf10137e1cabb18cbe6912c69e2e33263c45250f542" +checksum = "77823a27f0babb03091cb9ed9ef80af3b39dbc82f97e8fa530374b7dafd87a45" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", "cfg-if", "foreign-types", "libc", @@ -1533,18 +1514,18 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" [[package]] name = "openssl-src" -version = "300.6.0+3.6.2" +version = "300.6.1+3.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a8e8cbfd3a4a8c8f089147fd7aaa33cf8c7450c4d09f8f80698a0cf093abeff4" +checksum = "46eb8fb9fb3b61ce1c0f8a026c4c1a0714d3a9e138e7fbde78753ce2babc3846" dependencies = [ "cc", ] [[package]] name = "openssl-sys" -version = "0.9.115" +version = "0.9.117" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "158fe5b292746440aa6e7a7e690e55aeb72d41505e2804c23c6973ad0e9c9781" +checksum = "b47e7e6bb2c38cd930d25a23b40fa52e068c10e85f3e03a7f5ba5aaca5713695" dependencies = [ "cc", "libc", @@ -1711,9 +1692,9 @@ dependencies = [ [[package]] name = "prost" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" dependencies = [ "bytes", "prost-derive", @@ -1721,9 +1702,9 @@ dependencies = [ [[package]] name = "prost-build" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" +checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", "itertools 0.14.0", @@ -1740,9 +1721,9 @@ dependencies = [ [[package]] name = "prost-derive" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", "itertools 0.14.0", @@ -1753,9 +1734,9 @@ dependencies = [ [[package]] name = "prost-types" -version = "0.14.3" +version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8991c4cbdb8bc5b11f0b074ffe286c30e523de90fee5ba8132f1399f23cb3dd7" +checksum = "f94967dc7688f3054c7fac87473ffae4cc4c3904800e2d9f5b857246d8963b0a" dependencies = [ "prost", ] @@ -1772,9 +1753,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.45" +version = "1.0.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" dependencies = [ "proc-macro2", ] @@ -1893,14 +1874,14 @@ version = "0.5.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", ] [[package]] name = "regex" -version = "1.12.3" +version = "1.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e10754a14b9137dd7b1e3e5b0493cc9171fdd105e0ab477f51b72e7f3ac0e276" +checksum = "f1292b7759ae1cb9ec195452d1390a074f0cd8541ab7a5a8c31cd6db45d4a6ba" dependencies = [ "aho-corasick", "memchr", @@ -1921,9 +1902,9 @@ dependencies = [ [[package]] name = "regex-syntax" -version = "0.8.10" +version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" [[package]] name = "reqwest" @@ -1987,7 +1968,7 @@ version = "1.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", "errno", "libc", "linux-raw-sys", @@ -1996,9 +1977,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.40" +version = "0.23.41" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" dependencies = [ "once_cell", "rustls-pki-types", @@ -2009,9 +1990,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.14.1" +version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" +checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" dependencies = [ "zeroize", ] @@ -2048,15 +2029,6 @@ dependencies = [ "winapi-util", ] -[[package]] -name = "scc" -version = "2.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "46e6f046b7fef48e2660c57ed794263155d713de679057f2d0c169bfc6e756cc" -dependencies = [ - "sdd", -] - [[package]] name = "schannel" version = "0.1.29" @@ -2081,19 +2053,13 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" -[[package]] -name = "sdd" -version = "3.0.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "490dcfcbfef26be6800d11870ff2df8774fa6e86d047e3e8c8a76b25655e41ca" - [[package]] name = "security-framework" version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", "core-foundation 0.10.1", "core-foundation-sys", "libc", @@ -2110,12 +2076,6 @@ dependencies = [ "libc", ] -[[package]] -name = "semver" -version = "1.0.28" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd" - [[package]] name = "serde" version = "1.0.228" @@ -2158,9 +2118,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.149" +version = "1.0.150" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" dependencies = [ "indexmap 2.14.0", "itoa", @@ -2184,24 +2144,23 @@ dependencies = [ [[package]] name = "serial_test" -version = "3.4.0" +version = "3.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "911bd979bf1070a3f3aa7b691a3b3e9968f339ceeec89e08c280a8a22207a32f" +checksum = "699f4197115b8a7e7ff19c9a315a4bd6fffec26cc4626ef45ecaea389e081c6d" dependencies = [ "futures-executor", "futures-util", "log", "once_cell", "parking_lot", - "scc", "serial_test_derive", ] [[package]] name = "serial_test_derive" -version = "3.4.0" +version = "3.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0a7d91949b85b0d2fb687445e448b40d322b6b3e4af6b44a29b21d9a5f33e6d9" +checksum = "94e153fc76e1c6a068703d6d29c508a0b15c061c4b7e43da59cc097bc342673c" dependencies = [ "proc-macro2", "quote", @@ -2219,9 +2178,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "signal-hook-registry" @@ -2241,15 +2200,15 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "smallvec" -version = "1.15.1" +version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" [[package]] name = "socket2" -version = "0.6.3" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" +checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", "windows-sys 0.61.2", @@ -2309,9 +2268,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" [[package]] name = "syn" -version = "2.0.117" +version = "2.0.118" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99" +checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422" dependencies = [ "proc-macro2", "quote", @@ -2344,7 +2303,7 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", "core-foundation 0.9.4", "system-configuration-sys", ] @@ -2372,7 +2331,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.2", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -2395,20 +2354,20 @@ checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683" [[package]] name = "test-log" -version = "0.2.20" +version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f46bf474f0a4afebf92f076d54fd5e63423d9438b8c278a3d2ccb0f47f7cdb3" +checksum = "9b9c218384242b5c89b68303ab6f6fc53a312d923f0c14dc6bb860c6aeee40f1" dependencies = [ - "env_logger 0.11.10", + "env_logger 0.11.11", "test-log-macros", "tracing-subscriber", ] [[package]] name = "test-log-core" -version = "0.2.20" +version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37d4d41320b48bc4a211a9021678fcc0c99569b594ea31c93735b8e517102b4c" +checksum = "c26ef8b00e4d382e59f6a8ddb3cd790b3a5bb29f21a358a9a69ea2f29f13f27b" dependencies = [ "proc-macro2", "quote", @@ -2417,9 +2376,9 @@ dependencies = [ [[package]] name = "test-log-macros" -version = "0.2.20" +version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9beb9249a81e430dffd42400a49019bcf548444f1968ff23080a625de0d4d320" +checksum = "944ad38adcbb71eaa682c56bceeb079e4ca82b4b3edc2a0fde5cb297b77dac8d" dependencies = [ "syn", "test-log-core", @@ -2442,12 +2401,11 @@ dependencies = [ [[package]] name = "time" -version = "0.3.47" +version = "0.3.53" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" +checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50" dependencies = [ "deranged", - "itoa", "num-conv", "powerfmt", "serde_core", @@ -2457,15 +2415,15 @@ dependencies = [ [[package]] name = "time-core" -version = "0.1.8" +version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" +checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" [[package]] name = "time-macros" -version = "0.2.27" +version = "0.2.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215" +checksum = "c431b87111666e491a90baa837f914fb45cd5dc3c268591b0220ff5057f2085f" dependencies = [ "num-conv", "time-core", @@ -2569,11 +2527,11 @@ dependencies = [ [[package]] name = "tower-http" -version = "0.6.10" +version = "0.6.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68d6fdd9f81c2819c9a8b0e0cd91660e7746a8e6ea2ba7c6b2b057985f6bcb51" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" dependencies = [ - "bitflags 2.11.1", + "bitflags 2.13.0", "bytes", "futures-util", "http", @@ -2663,12 +2621,6 @@ version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af" -[[package]] -name = "unicode-xid" -version = "0.2.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" - [[package]] name = "untrusted" version = "0.9.0" @@ -2701,11 +2653,11 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "uuid" -version = "1.23.1" +version = "1.23.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76" +checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53" dependencies = [ - "getrandom 0.4.2", + "getrandom 0.4.3", "js-sys", "serde_core", "wasm-bindgen", @@ -2775,27 +2727,18 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" [[package]] name = "wasip2" -version = "1.0.3+wasi-0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" -dependencies = [ - "wit-bindgen 0.57.1", -] - -[[package]] -name = "wasip3" -version = "0.4.0+wasi-0.3.0-rc-2026-01-06" +version = "1.0.4+wasi-0.2.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" dependencies = [ - "wit-bindgen 0.51.0", + "wit-bindgen", ] [[package]] name = "wasm-bindgen" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49ace1d07c165b0864824eee619580c4689389afa9dc9ed3a4c75040d82e6790" +checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4" dependencies = [ "cfg-if", "once_cell", @@ -2806,9 +2749,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-futures" -version = "0.4.71" +version = "0.4.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96492d0d3ffba25305a7dc88720d250b1401d7edca02cc3bcd50633b424673b8" +checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d" dependencies = [ "js-sys", "wasm-bindgen", @@ -2816,9 +2759,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e68e6f4afd367a562002c05637acb8578ff2dea1943df76afb9e83d177c8578" +checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -2826,9 +2769,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d95a9ec35c64b2a7cb35d3fead40c4238d0940c86d107136999567a4703259f2" +checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e" dependencies = [ "bumpalo", "proc-macro2", @@ -2839,52 +2782,18 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.121" +version = "0.2.126" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4e0100b01e9f0d03189a92b96772a1fb998639d981193d7dbab487302513441" +checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24" dependencies = [ "unicode-ident", ] -[[package]] -name = "wasm-encoder" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "990065f2fe63003fe337b932cfb5e3b80e0b4d0f5ff650e6985b1048f62c8319" -dependencies = [ - "leb128fmt", - "wasmparser", -] - -[[package]] -name = "wasm-metadata" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" -dependencies = [ - "anyhow", - "indexmap 2.14.0", - "wasm-encoder", - "wasmparser", -] - -[[package]] -name = "wasmparser" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" -dependencies = [ - "bitflags 2.11.1", - "hashbrown 0.15.5", - "indexmap 2.14.0", - "semver", -] - [[package]] name = "web-sys" -version = "0.3.98" +version = "0.3.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b572dff8bcf38bad0fa19729c89bb5748b2b9b1d8be70cf90df697e3a8f32aa" +checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141" dependencies = [ "js-sys", "wasm-bindgen", @@ -3038,100 +2947,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" -[[package]] -name = "wit-bindgen" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" -dependencies = [ - "wit-bindgen-rust-macro", -] - [[package]] name = "wit-bindgen" version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" -[[package]] -name = "wit-bindgen-core" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc" -dependencies = [ - "anyhow", - "heck", - "wit-parser", -] - -[[package]] -name = "wit-bindgen-rust" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" -dependencies = [ - "anyhow", - "heck", - "indexmap 2.14.0", - "prettyplease", - "syn", - "wasm-metadata", - "wit-bindgen-core", - "wit-component", -] - -[[package]] -name = "wit-bindgen-rust-macro" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c0f9bfd77e6a48eccf51359e3ae77140a7f50b1e2ebfe62422d8afdaffab17a" -dependencies = [ - "anyhow", - "prettyplease", - "proc-macro2", - "quote", - "syn", - "wit-bindgen-core", - "wit-bindgen-rust", -] - -[[package]] -name = "wit-component" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" -dependencies = [ - "anyhow", - "bitflags 2.11.1", - "indexmap 2.14.0", - "log", - "serde", - "serde_derive", - "serde_json", - "wasm-encoder", - "wasm-metadata", - "wasmparser", - "wit-parser", -] - -[[package]] -name = "wit-parser" -version = "0.244.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" -dependencies = [ - "anyhow", - "id-arena", - "indexmap 2.14.0", - "log", - "semver", - "serde", - "serde_derive", - "serde_json", - "unicode-xid", - "wasmparser", -] - [[package]] name = "writeable" version = "0.6.3" @@ -3155,9 +2976,9 @@ checksum = "09041cd90cf85f7f8b2df60c646f853b7f535ce68f85244eb6731cf89fa498ec" [[package]] name = "yoke" -version = "0.8.2" +version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "abe8c5fda708d9ca3df187cae8bfb9ceda00dd96231bed36e445a1a48e66f9ca" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" dependencies = [ "stable_deref_trait", "yoke-derive", @@ -3178,18 +2999,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.48" +version = "0.8.52" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eed437bf9d6692032087e337407a86f04cd8d6a16a37199ed57949d415bd68e9" +checksum = "ce1022995ff5ff5d841ad7d994facc23098cd40152f2c1d11cd607c6f530653f" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.48" +version = "0.8.52" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70e3cd084b1788766f53af483dd21f93881ff30d7320490ec3ef7526d203bad4" +checksum = "1ae7f38b72ec2a254e2b87ef277cf2cd4fb97cbebf944faa6f33354da0867930" dependencies = [ "proc-macro2", "quote", @@ -3198,9 +3019,9 @@ dependencies = [ [[package]] name = "zerofrom" -version = "0.1.7" +version = "0.1.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69faa1f2a1ea75661980b013019ed6687ed0e83d069bc1114e2cc74c6c04c4df" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" dependencies = [ "zerofrom-derive", ] @@ -3219,9 +3040,9 @@ dependencies = [ [[package]] name = "zeroize" -version = "1.8.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" +checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" [[package]] name = "zerotrie" From eb48300c7d70a6644c442b64ee708cbebdc84a07 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Fri, 10 Jul 2026 11:18:45 +0200 Subject: [PATCH 7/8] Minor clean up --- dispatcher/src/queue.rs | 13 ++----------- multinode/src/client.rs | 9 ++------- 2 files changed, 4 insertions(+), 18 deletions(-) diff --git a/dispatcher/src/queue.rs b/dispatcher/src/queue.rs index c489ec4d..11b6997e 100644 --- a/dispatcher/src/queue.rs +++ b/dispatcher/src/queue.rs @@ -213,14 +213,6 @@ impl Future for IoWaitFuture<'_> { _ => true, }) .next(); - // If the extracted task is a prefetching task increase the counter accordingly - let is_prefetching = matches!( - &extracted, - Some(IoQueueElement { - work: WorkToDo::FunctionArguments { .. }, - .. - }) - ); let result = extracted.map(|queue_element| { ( queue_element.work, @@ -229,12 +221,11 @@ impl Future for IoWaitFuture<'_> { ) }); if let Some((mut work, debt, composition_id)) = result { - if is_prefetching { - lock_guard.prefetching_in_progress += 1; - } let composition_id_option = if let WorkToDo::FunctionArguments { recorder, .. } = &mut work { recorder.record(RecordPoint::IOQueueEnd); + // If the extracted task is a prefetching task increase the counter accordingly + lock_guard.prefetching_in_progress += 1; Some(composition_id) } else { None diff --git a/multinode/src/client.rs b/multinode/src/client.rs index c42fc94d..ce832115 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -252,8 +252,6 @@ async fn remote_queue_server_logic( mut remote_num_cores: u64, ) { let mut waiting_for_work = false; - // TODO remove, can also just look at the size of the dept map - let mut invocations_running = 0; let mut debt_map = BTreeMap::new(); let mut free_debt_ids = BinaryHeap::new(); @@ -366,7 +364,6 @@ async fn remote_queue_server_logic( break; } } else { - invocations_running += invocations.len(); if message_sender .send(queue_message::QueueMessage::Invocations( RepeatedInvocations { invocations }, @@ -379,11 +376,10 @@ async fn remote_queue_server_logic( } } remote_message::RemoteMessage::Response(response) => { - invocations_running -= 1; debug_assert!(data_option.is_none()); trace!( "Queue Server received response, outstanding: {}", - invocations_running + debt_map.len() ); let Response { invocation_id, @@ -459,11 +455,10 @@ async fn remote_queue_server_logic( } QueueOption::TryOffload(work, debt, composition_id) => { // if this node already sent enough work for the remote to be at capacity don't send more - if invocations_running >= remote_num_cores as usize { + if debt_map.len() >= remote_num_cores as usize { queue.reenqueue(work, debt, composition_id); continue; } - invocations_running += 1; // Ask remote if it can take the invocation, otherwise requeue it locally let promise_id = if let Some(free_id) = free_debt_ids.pop() { free_id From 0da22d36abfa938049a2774e0cb1fa8234b586e0 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Fri, 10 Jul 2026 11:27:37 +0200 Subject: [PATCH 8/8] Fix test --- multinode/src/client/test.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/multinode/src/client/test.rs b/multinode/src/client/test.rs index 60ed5daa..c0dbc6e5 100644 --- a/multinode/src/client/test.rs +++ b/multinode/src/client/test.rs @@ -549,7 +549,7 @@ fn test_remote_queue_client_prefetch() { let idle_clone = idle_cores.clone(); 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, @@ -557,7 +557,7 @@ fn test_remote_queue_client_prefetch() { invocation_id, function_id, inputs, - is_cold, + caching, recorder, )) .unwrap(); @@ -622,10 +622,10 @@ fn test_remote_queue_client_prefetch() { 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 }