From 3245ac513554d9f77e150ca9de07e5b064f238ec Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Wed, 1 Jul 2026 11:27:00 +0200 Subject: [PATCH 01/10] fix typo --- server/src/main.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/src/main.rs b/server/src/main.rs index a9609e28..b120a3f8 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -146,7 +146,7 @@ fn main() -> () { println!("core allocation:"); println!("minimum system cores {}", min_sys_cores); - println!("maximum sysmte cores {}", max_sys_cores); + println!("maximum system cores {}", max_sys_cores); println!("compute cores: {:?}", compute_cores); // set up dispatcher configuration basics From b35ab26ed647c7d4191007ac431803f10e3881eb Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Wed, 1 Jul 2026 12:00:18 +0200 Subject: [PATCH 02/10] prevent-scheduling-on-timeout --- machine_interface/src/function_driver/thread_utils.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/machine_interface/src/function_driver/thread_utils.rs b/machine_interface/src/function_driver/thread_utils.rs index d817ad1c..2b48eaba 100644 --- a/machine_interface/src/function_driver/thread_utils.rs +++ b/machine_interface/src/function_driver/thread_utils.rs @@ -148,6 +148,11 @@ fn run_thread(core_id: u8, mut queue: impl EngineWorkQueue) { 'engine: loop { // TODO catch unwind so we can always return an error or shut down gracefully let (args, debt) = waker::manual_pull(&mut queue); + + if !debt.is_alive() { + continue 'engine; + } + match args { WorkToDo::FunctionArguments { function_id: _, @@ -233,6 +238,10 @@ fn run_thread(core_id: u8, mut queue: impl EngineWorkQueue) { recorder.record(RecordPoint::EngineStart); + if !debt.is_alive() { + continue 'engine; + } + let result = engine_state.run( function.config.clone(), function_context, From 9ef79fba192df8217ff1a266798808f6639d5493 Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Wed, 1 Jul 2026 12:02:47 +0200 Subject: [PATCH 03/10] move above recorder --- machine_interface/src/function_driver/thread_utils.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/machine_interface/src/function_driver/thread_utils.rs b/machine_interface/src/function_driver/thread_utils.rs index 2b48eaba..c72d25d5 100644 --- a/machine_interface/src/function_driver/thread_utils.rs +++ b/machine_interface/src/function_driver/thread_utils.rs @@ -235,13 +235,14 @@ fn run_thread(core_id: u8, mut queue: impl EngineWorkQueue) { function_context.content.push(None); } } - - recorder.record(RecordPoint::EngineStart); - + if !debt.is_alive() { continue 'engine; } + recorder.record(RecordPoint::EngineStart); + + let result = engine_state.run( function.config.clone(), function_context, From 258bd0dc79cef7749f26d13ef99f9287ed3dc15a Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Wed, 1 Jul 2026 12:05:58 +0200 Subject: [PATCH 04/10] fix formatting for CI --- machine_interface/src/function_driver/thread_utils.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/machine_interface/src/function_driver/thread_utils.rs b/machine_interface/src/function_driver/thread_utils.rs index c72d25d5..ecfb5422 100644 --- a/machine_interface/src/function_driver/thread_utils.rs +++ b/machine_interface/src/function_driver/thread_utils.rs @@ -235,14 +235,13 @@ fn run_thread(core_id: u8, mut queue: impl EngineWorkQueue) { function_context.content.push(None); } } - + if !debt.is_alive() { continue 'engine; } recorder.record(RecordPoint::EngineStart); - let result = engine_state.run( function.config.clone(), function_context, From 088045cde054b4c59ca0723c86ff2a8f3a9a9978 Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Wed, 1 Jul 2026 13:02:14 +0200 Subject: [PATCH 05/10] prevent offloading work that is not alive --- multinode/src/client.rs | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 483783e2..e7bf8a76 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -246,8 +246,10 @@ 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().filter_map(|(work, debt)| { + if !debt.is_alive() { + return None; + } // there is some work so send it out // find the local function id to use let promise_id = if let Some(free_id) = free_debt_ids.pop() { @@ -302,12 +304,12 @@ async fn remote_queue_server_logic( work, ), ); - Invocation { + Some(Invocation { metadata_sets, function_id, invocation_id: promise_id, caching, - } + }) })); } if invocations.is_empty() { @@ -399,6 +401,9 @@ async fn remote_queue_server_logic( } } QueueOption::TryOffload(work, debt) => { + if !debt.is_alive() { + continue; + } // 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).await; From 8092c206c1231e8e8c56c221f8871ddd8d987992 Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Wed, 1 Jul 2026 16:41:43 +0200 Subject: [PATCH 06/10] yield in dispatcher composition loop to make sure term signals work --- dispatcher/src/dispatcher.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/dispatcher/src/dispatcher.rs b/dispatcher/src/dispatcher.rs index 039c2537..35de8fad 100644 --- a/dispatcher/src/dispatcher.rs +++ b/dispatcher/src/dispatcher.rs @@ -28,6 +28,7 @@ use machine_interface::{ memory_domain::{MemoryDomain, MemoryResource}, }; use std::{collections::BTreeMap, sync::Arc}; +use tokio::task::yield_now; // 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 @@ -410,6 +411,7 @@ impl Dispatcher { awaited_sets.len(), non_ready_functions.len() ); + yield_now().await; } recorder.add_children(recorders); From eee8003c96cb24cde93bc50a067707cad80708e0 Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Sat, 11 Jul 2026 22:10:55 +0200 Subject: [PATCH 07/10] add queue option CancelInvocation to multi node setup --- machine_interface/src/promise.rs | 30 ++++---- multinode/proto/multinode.proto | 3 +- multinode/src/client.rs | 115 ++++++++++++++++++++++++++++--- multinode/src/client/test.rs | 2 + 4 files changed, 125 insertions(+), 25 deletions(-) diff --git a/machine_interface/src/promise.rs b/machine_interface/src/promise.rs index dee0264c..25e45677 100644 --- a/machine_interface/src/promise.rs +++ b/machine_interface/src/promise.rs @@ -9,7 +9,7 @@ use core::{ task::{Poll, Waker}, }; use dandelion_commons::{err_dandelion, DandelionError, DandelionResult, PromiseError}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; static WAKER_INDEX: u8 = 0b0000_0001; static DEBT_ALIVE: u8 = 0b0001_0000; @@ -17,11 +17,13 @@ static PROMISE_ALIVE: u8 = 0b0010_0000; static CONTENT_SET: u8 = 0b0100_0000; static ALIVE: u8 = DEBT_ALIVE | PROMISE_ALIVE; +type AbortHandle = Box; + // TODO replace 2 wakers with a futures::task::AtomicWaker struct PromiseData { /// Abort handle, only to be called once, as long as this value ///non null that means the function has not been aborted or terminated on it's own - abort_handle: AtomicPtr ()>, + abort_handle: Mutex>, /// Points to raw box of the results the engine has put in there results: Cell>, /// TODO replace Option with only waker when Waker::noop stabilizes, @@ -118,7 +120,7 @@ impl PromiseBuffer { let data_ptr = self.internal.get_promise_data()?; let data = unsafe { &mut (&mut *data_ptr).data }; let default = ManuallyDrop::new(PromiseData { - abort_handle: AtomicPtr::new(ptr::null_mut()), + abort_handle: Mutex::new(None), results: Cell::new(err_dandelion!(DandelionError::PromiseError( PromiseError::Default ))), @@ -151,9 +153,9 @@ impl Promise { } fn abort_internal(&mut self) { let data = unsafe { &(&*self.data).data }; - let abort_handle = data.abort_handle.swap(ptr::null_mut(), Ordering::SeqCst); - if !abort_handle.is_null() { - unsafe { (*abort_handle)() } + let abort_handle = data.abort_handle.lock().unwrap().take(); + if let Some(abort_handle) = abort_handle { + abort_handle() } } } @@ -217,8 +219,8 @@ impl Debt { pub fn fulfill(self, results: DandelionResult) { let data = unsafe { &(&*self.data).data }; - // make sure we are not aborted by this promise anymore - data.abort_handle.store(ptr::null_mut(), Ordering::SeqCst); + // prevent a later abort from invoking the callback + data.abort_handle.lock().unwrap().take(); // write a result data.results.set(results); let flags = data.flags.fetch_or(CONTENT_SET, Ordering::SeqCst); @@ -227,18 +229,20 @@ impl Debt { waker.wake(); } } - pub fn install_abort_handle(&self, handle: fn()) { + pub fn install_abort_handle(&self, handle: F) + where + F: FnOnce() + Send + 'static, + { let data = unsafe { &(&*self.data).data }; - data.abort_handle - .store(handle as *mut fn(), Ordering::SeqCst); + *data.abort_handle.lock().unwrap() = Some(Box::new(handle)); } } impl Drop for Debt { fn drop(&mut self) { let data = unsafe { &(&*self.data).data }; - // make sure we can't get aborted by this handle anymore - data.abort_handle.store(ptr::null_mut(), Ordering::SeqCst); + // prevent a later abort from invoking the callback + data.abort_handle.lock().unwrap().take(); // if promise is still alive, there is still a promise waiting for a result let flags = data.flags.load(Ordering::SeqCst); if flags & PROMISE_ALIVE == 1 { diff --git a/multinode/proto/multinode.proto b/multinode/proto/multinode.proto index cb346365..a23460dd 100644 --- a/multinode/proto/multinode.proto +++ b/multinode/proto/multinode.proto @@ -124,6 +124,7 @@ message QueueMessage { RepeatedInvocations invocations = 2; // Message containing the work for one function, can be refused Invocation try_offload = 3; - // TODO add message to cancel invocation + // Message indicating remote invocation should not be scheduled + uint32 cancel_invocation = 4; } } diff --git a/multinode/src/client.rs b/multinode/src/client.rs index e7bf8a76..57e01706 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -29,7 +29,7 @@ use machine_interface::{ }; use prost::bytes::{Bytes, BytesMut}; use std::{ - collections::{BTreeMap, BinaryHeap}, + collections::{BTreeMap, BTreeSet, BinaryHeap}, sync::Arc, time::{Duration, Instant, SystemTime}, }; @@ -145,6 +145,7 @@ enum QueueOption { Message(remote_message::RemoteMessage, Option), WorkAvailable, TryOffload(WorkToDo, machine_interface::promise::Debt), + CancelRemote(u32), } async fn remote_queue_sever_notification(receiver: Arc, sender: mpsc::Sender) { @@ -202,6 +203,7 @@ async fn remote_queue_server_try_offload( /// The protocol logic handling for the remote queue server async fn remote_queue_server_logic( mut message_receiver: mpsc::Receiver, + local_sender: mpsc::Sender, message_sender: mpsc::Sender, queue: WorkQueue, export_registry: ExportRegistry, @@ -213,6 +215,7 @@ async fn remote_queue_server_logic( let mut invocations_running = 0; let mut debt_map = BTreeMap::new(); + let mut cancelled_debt_ids = BTreeSet::new(); let mut free_debt_ids = BinaryHeap::new(); let mut max_debt_id = 0; @@ -294,6 +297,18 @@ async fn remote_queue_server_logic( ); let caching = caching; let function_id = function_id.to_string(); + let cancel_sender = local_sender.clone(); + debt.install_abort_handle(move || { + if cancel_sender + .try_send(QueueOption::CancelRemote(promise_id)) + .is_err() + { + warn!( + "Failed to enqueue cancellation for remote invocation {}", + promise_id + ); + } + }); debt_map.insert( promise_id, ( @@ -339,10 +354,22 @@ 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 Some((debt, mut recorder, start_epoch, remote_data_references, work)) = + debt_map.remove(&invocation_id) + else { + if cancelled_debt_ids.remove(&invocation_id) { + // Do not reuse this ID: the cancellation message can arrive at the + // worker after an earlier result, where it would otherwise remain + // queued and incorrectly cancel a later invocation with the same ID. + trace!("Received cancellation acknowledgement for remote invocation {}", invocation_id); + } else { + warn!( + "Received response for unknown remote invocation {}", + invocation_id + ); + } + continue; + }; free_debt_ids.push(invocation_id); drop(remote_data_references); // remote did not do work, was a try offload request, reenqueu the work @@ -449,6 +476,18 @@ async fn remote_queue_server_logic( }); let caching = caching; let function_id = function_id.to_string(); + let cancel_sender = local_sender.clone(); + debt.install_abort_handle(move || { + if cancel_sender + .try_send(QueueOption::CancelRemote(promise_id)) + .is_err() + { + warn!( + "Failed to enqueue cancellation for remote invocation {}", + promise_id + ); + } + }); debt_map.insert( promise_id, ( @@ -478,6 +517,21 @@ async fn remote_queue_server_logic( .unwrap(); } } + QueueOption::CancelRemote(invocation_id) => { + if let Some((_debt, _recorder, _start_epoch, remote_data_references, _work)) = + debt_map.remove(&invocation_id) + { + drop(remote_data_references); + cancelled_debt_ids.insert(invocation_id); + if message_sender + .send(queue_message::QueueMessage::CancelInvocation(invocation_id)) + .await + .is_err() + { + break; + } + } + } } } warn!("Arrived at end of remtote_queue_server_logic, which should stay in the loop forever"); @@ -537,11 +591,12 @@ pub async fn remote_queue_server( queue.add_remote_channel(node_id, offload_sender); spawn(remote_queue_server_try_offload( offload_receiver, - queue_option_sender, + queue_option_sender.clone(), )); remote_queue_server_logic( queue_option_receiver, + queue_option_sender.clone(), queue_message_sender, queue, export_registry, @@ -557,7 +612,7 @@ pub enum PollingOption { QueueStateChanged(usize), LocalCoreCountChanged(usize), // Results(RemoteMessage, Option<(Vec>, u64)>), - Results(remote_message::RemoteMessage), + Results(u32, remote_message::RemoteMessage), } async fn remote_queue_client_receiver( @@ -646,6 +701,7 @@ async fn dispatcher_call( }; sender .send(PollingOption::Results( + invocation_id, remote_message::RemoteMessage::Response(Response { invocation_id, response: Some(response_message), @@ -676,6 +732,7 @@ async fn remote_queue_client_logic( // TODO: think about the general issue of state synchronization between the dispatcher and multinode client adding things, // and the engines and server taking things. let mut work_from_remote = 0; + let mut cancelled_remote_invocations = BTreeSet::new(); while let Some(current_future) = receiver.recv().await { match current_future { @@ -715,6 +772,13 @@ async fn remote_queue_client_logic( remote_had_work = true; } } + queue_message::QueueMessage::CancelInvocation(invocation_id) => { + trace!( + "Queue Client received cancellation for invocation {}", + invocation_id + ); + cancelled_remote_invocations.insert(invocation_id); + } // TODO for try offload decide when to refuse work queue_message::QueueMessage::TryOffload(invocation) => { trace!("Queue Client recieved try offload"); @@ -733,6 +797,17 @@ async fn remote_queue_client_logic( metadata_sets, caching, } = invocation; + if cancelled_remote_invocations.remove(&invocation_id) { + trace!("Skipping canceled try-offload invocation {}", invocation_id); + message_sender + .send(remote_message::RemoteMessage::Response(Response { + invocation_id, + response: None, + })) + .await + .unwrap(); + continue; + } let function_arc = Arc::new(function_id); let recorder = Recorder::new(function_arc.clone(), start_instance); let inputs = @@ -752,7 +827,6 @@ async fn remote_queue_client_logic( // 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 = std::time::SystemTime::elapsed(&std::time::SystemTime::UNIX_EPOCH) @@ -764,6 +838,18 @@ async fn remote_queue_client_logic( metadata_sets, caching, } = invocation; + if cancelled_remote_invocations.remove(&invocation_id) { + trace!("Skipping canceled invocation {}", invocation_id); + message_sender + .send(remote_message::RemoteMessage::Response(Response { + invocation_id, + response: None, + })) + .await + .unwrap(); + continue; + } + work_from_remote += 1; let function_arc = Arc::new(function_id); let recorder = Recorder::new(function_arc.clone(), start_instance); let inputs = proto_data_sets_to_composition_sets( @@ -787,10 +873,17 @@ async fn remote_queue_client_logic( // TODO: recover from message reception failure panic!("Receiving remote queue message faied with: {}", error); } - PollingOption::Results(results) => { - trace!("Queue Client sending out result"); + PollingOption::Results(invocation_id, results) => { work_from_remote -= 1; - message_sender.send(results).await.unwrap(); + if cancelled_remote_invocations.remove(&invocation_id) { + trace!( + "Suppressing result for canceled remote invocation {}", + invocation_id + ); + } else { + trace!("Queue Client sending out result"); + message_sender.send(results).await.unwrap(); + } let occupancy = std::cmp::max(queue_state, work_from_remote); if remote_had_work && occupancy < num_local_cores { let engines: Vec<_> = EngineType::iter() diff --git a/multinode/src/client/test.rs b/multinode/src/client/test.rs index df4197d7..086e7685 100644 --- a/multinode/src/client/test.rs +++ b/multinode/src/client/test.rs @@ -78,6 +78,7 @@ fn test_remote_queue_server() { let mut context = Context::from_waker(Waker::noop()); let mut server_future = Box::pin(remote_queue_server_logic( queue_option_receiver, + queue_option_sender.clone(), queue_message_sender, work_queue.clone(), ExportRegistry::new(1), @@ -454,6 +455,7 @@ fn test_remote_queue_client() { // send back a result, mark the queue as changed and poll to get it processed poll_option_sender .try_send(crate::client::PollingOption::Results( + invocation_id_3, remote_message::RemoteMessage::Response(Response { invocation_id: invocation_id_3, response: Some(response::Response::ErrorMsg( From 9d2e6722025b9dfb3b5d2758bb00f849c1b8792a Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Sat, 11 Jul 2026 22:29:45 +0200 Subject: [PATCH 08/10] revert wrong merge changes --- dispatcher/src/dispatcher.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/dispatcher/src/dispatcher.rs b/dispatcher/src/dispatcher.rs index 3ecf4019..d59bc2dc 100644 --- a/dispatcher/src/dispatcher.rs +++ b/dispatcher/src/dispatcher.rs @@ -31,7 +31,6 @@ use std::{ collections::BTreeMap, sync::{atomic::AtomicUsize, Arc}, }; -use tokio::task::yield_now; // 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 @@ -434,7 +433,6 @@ impl Dispatcher { awaited_sets.len(), non_ready_functions.len() ); - yield_now().await; } recorder.add_children(recorders); From 299feac7eb085d007ca306d2b01f4810bb3cd7e7 Mon Sep 17 00:00:00 2001 From: Lanzelot Moll Date: Thu, 16 Jul 2026 19:08:30 +0200 Subject: [PATCH 09/10] recycle debt ids --- multinode/src/client.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/multinode/src/client.rs b/multinode/src/client.rs index 441936a2..d4567759 100644 --- a/multinode/src/client.rs +++ b/multinode/src/client.rs @@ -415,8 +415,9 @@ async fn remote_queue_server_logic( )) = debt_map.remove(&invocation_id) else { if cancelled_debt_ids.remove(&invocation_id) { - // Do not reuse this ID: a delayed cancellation can otherwise - // cancel a later invocation with the same ID. + // The acknowledgement guarantees that the remote will not send a + // delayed response that could be mistaken for a later invocation. + free_debt_ids.push(invocation_id); trace!( "Received cancellation acknowledgement for remote invocation {}", invocation_id From 8bc544b2235f463aca4d883a932413b11612fbc7 Mon Sep 17 00:00:00 2001 From: Tom Kuchler Date: Mon, 20 Jul 2026 17:00:52 +0200 Subject: [PATCH 10/10] Update promise --- machine_interface/src/promise.rs | 136 ++++++++++++------------------- 1 file changed, 54 insertions(+), 82 deletions(-) diff --git a/machine_interface/src/promise.rs b/machine_interface/src/promise.rs index 25e45677..2f921507 100644 --- a/machine_interface/src/promise.rs +++ b/machine_interface/src/promise.rs @@ -6,29 +6,29 @@ use core::{ pin::Pin, ptr, sync::atomic::{AtomicPtr, AtomicU8, Ordering}, - task::{Poll, Waker}, + task::Poll, }; use dandelion_commons::{err_dandelion, DandelionError, DandelionResult, PromiseError}; -use std::sync::{Arc, Mutex}; +use futures::task::AtomicWaker; +use std::{ + mem::MaybeUninit, + sync::{atomic::Ordering::Acquire, Arc}, +}; -static WAKER_INDEX: u8 = 0b0000_0001; -static DEBT_ALIVE: u8 = 0b0001_0000; -static PROMISE_ALIVE: u8 = 0b0010_0000; -static CONTENT_SET: u8 = 0b0100_0000; -static ALIVE: u8 = DEBT_ALIVE | PROMISE_ALIVE; +// debt sets content on drop so this is both the alive flag and the content lock +static DEBT_ALIVE: u8 = 0b0000_0001; +static PROMISE_ALIVE: u8 = 0b0000_0010; +static ABORT_SET: u8 = 0b0000_0100; type AbortHandle = Box; -// TODO replace 2 wakers with a futures::task::AtomicWaker struct PromiseData { /// Abort handle, only to be called once, as long as this value ///non null that means the function has not been aborted or terminated on it's own - abort_handle: Mutex>, + abort_handle: MaybeUninit, /// Points to raw box of the results the engine has put in there results: Cell>, - /// TODO replace Option with only waker when Waker::noop stabilizes, - /// and we can use it as default and use clone_from() on the cells - wakers: [Cell>; 2], + waker: AtomicWaker, flags: AtomicU8, } @@ -84,22 +84,18 @@ impl PromiseBufferInternal { return Ok(current); } - fn drop_promise_data(&self, data_ptr: *mut DataWrapper, drop_origin: u8) { - let data = unsafe { &(&*data_ptr).data }; - let previous_flags = data.flags.fetch_and(!drop_origin, Ordering::SeqCst); - if ((previous_flags & !drop_origin) & ALIVE) == 0 { - // drop data in union so we can reuse - unsafe { ManuallyDrop::::drop(&mut (*data_ptr).data) }; - // reinsert at head - let mut head = self.head.load(Ordering::Acquire); + fn drop_promise_data(&self, data_ptr: *mut DataWrapper) { + // drop data in union so we can reuse + unsafe { ManuallyDrop::::drop(&mut (*data_ptr).data) }; + // reinsert at head + let mut head = self.head.load(Ordering::Acquire); + unsafe { (*data_ptr).next = head }; + while let Err(current_head) = + self.head + .compare_exchange(head, data_ptr, Ordering::AcqRel, Ordering::Acquire) + { + head = current_head; unsafe { (*data_ptr).next = head }; - while let Err(current_head) = - self.head - .compare_exchange(head, data_ptr, Ordering::AcqRel, Ordering::Acquire) - { - head = current_head; - unsafe { (*data_ptr).next = head }; - } } } } @@ -120,11 +116,11 @@ impl PromiseBuffer { let data_ptr = self.internal.get_promise_data()?; let data = unsafe { &mut (&mut *data_ptr).data }; let default = ManuallyDrop::new(PromiseData { - abort_handle: Mutex::new(None), + abort_handle: MaybeUninit::uninit(), results: Cell::new(err_dandelion!(DandelionError::PromiseError( - PromiseError::Default + PromiseError::DroppedDebt ))), - wakers: [Cell::new(None), Cell::new(None)], + waker: AtomicWaker::new(), flags: AtomicU8::new(DEBT_ALIVE | PROMISE_ALIVE), }); *data = default; @@ -153,9 +149,11 @@ impl Promise { } fn abort_internal(&mut self) { let data = unsafe { &(&*self.data).data }; - let abort_handle = data.abort_handle.lock().unwrap().take(); - if let Some(abort_handle) = abort_handle { - abort_handle() + // check there is an abort handle and the debt has not already been dropped + let flags = data.flags.load(Acquire); + if flags & (ABORT_SET | DEBT_ALIVE) == (ABORT_SET | DEBT_ALIVE) { + let abort_hanlder = unsafe { data.abort_handle.assume_init_read() }; + abort_hanlder(); } } } @@ -166,29 +164,14 @@ impl futures::future::Future for Promise { // handle this by returning pending again fn poll(self: Pin<&mut Self>, cx: &mut core::task::Context<'_>) -> Poll { let data = unsafe { &(&*self.data).data }; - let flags = data.flags.load(Ordering::SeqCst); // update the waker - let waker_index = (flags & WAKER_INDEX) ^ WAKER_INDEX; - data.wakers[usize::from(waker_index)].set(Some(cx.waker().clone())); - - // set waker to next one, to ensure the fulfill never reads a waker wrie in progress, - // because of 2 polls after each other where the fulfill has the index of the second - // waker being written. This is automatically avoided, by checking for changes in the - // debt flags. - // want to take content if there is some and flip index - let new_flags = (flags ^ WAKER_INDEX) & !CONTENT_SET; - let current_flags = - match data - .flags - .compare_exchange(flags, new_flags, Ordering::SeqCst, Ordering::SeqCst) - { - Ok(changed_flags) | Err(changed_flags) => changed_flags, - }; + data.waker.register(cx.waker()); + let flags = data.flags.load(Ordering::Acquire); // the only changes the debt ever does is set the content or get dropped, which also sets content // if there was an error it could only have been seting the content. - if current_flags & CONTENT_SET != 0 { + if flags & DEBT_ALIVE == 0 { return Poll::Ready(data.results.replace(err_dandelion!( DandelionError::PromiseError(PromiseError::TakenPromise,) ))); @@ -201,7 +184,11 @@ impl futures::future::Future for Promise { impl Drop for Promise { fn drop(&mut self) { self.abort_internal(); - self.origin.drop_promise_data(self.data, PROMISE_ALIVE); + let data = unsafe { &(&*self.data).data }; + let previous_flags = data.flags.fetch_and(!PROMISE_ALIVE, Ordering::SeqCst); + if (previous_flags & DEBT_ALIVE) == 0 { + self.origin.drop_promise_data(self.data); + } } } @@ -219,48 +206,33 @@ impl Debt { pub fn fulfill(self, results: DandelionResult) { let data = unsafe { &(&*self.data).data }; - // prevent a later abort from invoking the callback - data.abort_handle.lock().unwrap().take(); - // write a result + // write a result, the flag will be set and the waker called when the drop is executed data.results.set(results); - let flags = data.flags.fetch_or(CONTENT_SET, Ordering::SeqCst); - let waker_index = flags & WAKER_INDEX; - if let Some(waker) = data.wakers[usize::from(waker_index)].take() { - waker.wake(); - } } + + // The installer needs to be able to deal with the abort arriving after the debt has been dropped. + // This can happen due to race conditions when dropping / fulfilling the debt and calls to the abort. pub fn install_abort_handle(&self, handle: F) where F: FnOnce() + Send + 'static, { - let data = unsafe { &(&*self.data).data }; - *data.abort_handle.lock().unwrap() = Some(Box::new(handle)); + let data = unsafe { &mut (&mut *self.data).data }; + data.abort_handle.write(Box::new(handle)); + data.flags.fetch_or(ABORT_SET, Ordering::AcqRel); } } impl Drop for Debt { fn drop(&mut self) { let data = unsafe { &(&*self.data).data }; - // prevent a later abort from invoking the callback - data.abort_handle.lock().unwrap().take(); - // if promise is still alive, there is still a promise waiting for a result - let flags = data.flags.load(Ordering::SeqCst); - if flags & PROMISE_ALIVE == 1 { - // always pay your debts - if flags & CONTENT_SET == 0 { - data.results - .set(err_dandelion!(DandelionError::PromiseError( - PromiseError::DroppedDebt - ))); - data.flags.fetch_or(CONTENT_SET, Ordering::SeqCst); - } - let waker_index = data.flags.load(Ordering::SeqCst) & WAKER_INDEX; - if let Some(waker) = data.wakers[usize::from(waker_index)].take() { - waker.wake(); - } + // mark debt as dropped + let previous_flags = data.flags.fetch_and(!DEBT_ALIVE, Ordering::SeqCst); + + // call waker, if the promise has been dropped the waker should be able to deal with a delayed signal + data.waker.wake(); + + if previous_flags & PROMISE_ALIVE == 0 { + self.origin.as_ref().drop_promise_data(self.data); } - self.origin - .as_ref() - .drop_promise_data(self.data, DEBT_ALIVE); } }