tirbofish/dropbear · commit
3a5b98541ab65424afc34ccfea9ab9552f57afbe
futures work well now. i created a new crate so other people can use it.
Signature present but could not be verified.
Unverified
@@ -6,7 +6,7 @@ package.repository = "https://github.com/4tkbytes/dropbear-engine" package.readme = "README.md" resolver = "3" -members = [ "dropbear-engine", "eucalyptus-core", "eucalyptus-editor"] +members = [ "dropbear-engine", "dropbear_future-queue", "eucalyptus-core", "eucalyptus-editor"] # members = ["dropbear-engine", "eucalyptus-core", "eucalyptus-editor", "redback-runtime"] @@ -9,6 +9,7 @@ repository.workspace = true readme.workspace = true [dependencies] +dropbear_future-queue = { path = "../dropbear_future-queue" } anyhow.workspace = true app_dirs2.workspace = true bytemuck.workspace = true @@ -32,11 +33,9 @@ parking_lot.workspace = true lazy_static.workspace = true gltf.workspace = true rayon.workspace = true -tokio.workspace = true backtrace.workspace = true os_info.workspace = true rustc_version_runtime.workspace = true -ahash = "0.8.12" [target.'cfg(not(target_os = "android"))'.dependencies] rfd.workspace = true @@ -1,448 +0,0 @@ -//! Polling and async features in the dropbear engine where scenes are single threaded. - -use std::any::Any; -use std::collections::VecDeque; -use std::pin::Pin; -use std::sync::Arc; -use ahash::{HashMap, HashMapExt}; -use parking_lot::Mutex; -use tokio::sync::oneshot; -use std::future::Future; - -/// A type used for a future. -/// -/// It must include a [`Send`] trait to be usable for the [`FutureQueue`] -pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + Sync>>; -pub type AnyResult = Arc<dyn Any + Send + Sync>; -pub type ResultSender = oneshot::Sender<AnyResult>; -pub type ResultReceiver = oneshot::Receiver<AnyResult>; -pub type FutureStorage = Arc<Mutex<VecDeque<(u64, BoxFuture<()>)>>>; - -/// A status showing the future, used by the [`ResultReceiver`] and [`ResultSender`] -#[derive(Clone)] -pub enum FutureStatus { - NotPolled, - CurrentlyPolling, - Completed(AnyResult), -} - -/// A handle to the future task -#[derive(Default, Clone)] -pub struct FutureHandle { - pub id: u64, -} - -/// Internal storage per handle — separate from FutureHandle -struct HandleEntry { - receiver: ResultReceiver, - status: FutureStatus, -} - -/// A queue used for futures. -pub struct FutureQueue { - /// The queue for the futures. - queued: FutureStorage, - /// A place to store all handle data - handle_registry: Arc<Mutex<HashMap<u64, HandleEntry>>>, - /// Next id to be processed - next_id: Arc<Mutex<u64>>, -} - -impl FutureQueue { - /// Creates a new [`Arc<FutureQueue>`]. - pub fn new() -> Self { - Self { - queued: Arc::new(Mutex::new(VecDeque::new())), - handle_registry: Arc::new(Mutex::new(HashMap::new())), - next_id: Arc::new(Mutex::new(0)), - } - } - - /// Pushes a future to the FutureQueue. It will sit and wait - /// to be processed until [`FutureQueue::poll`] is called. - /// - /// This creates a new hash by using the [`ahash`] crate. The type is not required - /// to implement [`std::hash::Hash`]. - pub fn push<F, T>(&self, future: F) -> FutureHandle - where - F: Future<Output = T> + Send + Sync + 'static, - T: Send + Sync + 'static, - { - let mut next_id = self.next_id.lock(); - let id = *next_id; - *next_id += 1; - - let (sender, receiver) = oneshot::channel(); - - let entry = HandleEntry { - receiver, - status: FutureStatus::NotPolled, - }; - - self.handle_registry.lock().insert(id, entry); - - let registry_clone = self.handle_registry.clone(); - let wrapped_future = Box::pin(async move { - let result = future.await; - let boxed_result = Arc::new(result) as AnyResult; - - let _ = sender.send(boxed_result.clone()); - - let mut registry = registry_clone.lock(); - if let Some(entry) = registry.get_mut(&id) { - entry.status = FutureStatus::Completed(boxed_result); - } - }); - - self.queued.lock().push_back((id, wrapped_future)); - - FutureHandle { id } // 👈 Simple handle - } - - /// Polls all the futures in the future queue and resolves the handles. - /// - /// This function spawns a new async thread for each item inside the thread and - /// sends updates to the Handle's receiver. - pub fn poll(&self) { // 👈 Removed unused generics <T, F> - let mut queue = self.queued.lock(); - let mut futures_to_spawn = Vec::new(); - - while let Some((id, future)) = queue.pop_front() { - // Update status to CurrentlyPolling - if let Some(entry) = self.handle_registry.lock().get_mut(&id) { - entry.status = FutureStatus::CurrentlyPolling; - } - - futures_to_spawn.push(future); - } - - for future in futures_to_spawn { - tokio::spawn(future); - } - } - - /// Exchanges the future for the result. - /// - /// When the handle is not successful, it will return nothing. When the handle is successful, - /// it will return the result and drop the handle, removing the usage of it. - pub fn exchange(&self, handle: &FutureHandle) -> Option<AnyResult> { - let mut registry = self.handle_registry.lock(); - if let Some(entry) = registry.get_mut(&handle.id) { - match &entry.status { - FutureStatus::Completed(result) => { - return Some(result.clone()); // Clone the Arc - } - _ => { - return match entry.receiver.try_recv() { - Ok(result) => { - entry.status = FutureStatus::Completed(result.clone()); - Some(result) - } - Err(_) => None, - } - } - } - } - None - } - - /// Exchanges the handle and safely downcasts it into a specific type. - pub fn exchange_as<T: Any + Send + Sync + 'static>(&self, handle: &FutureHandle) -> Option<Arc<T>> { - self.exchange(handle)? - .downcast() - .ok() - } - - /// Retrieve a handle by u64 ID - pub fn get_handle(&self, id: u64) -> Option<FutureHandle> { - let registry = self.handle_registry.lock(); - if registry.contains_key(&id) { - Some(FutureHandle { id }) - } else { - None - } - } - - /// Get status of a handle - pub fn get_status(&self, id: u64) -> Option<FutureStatus> { - let registry = self.handle_registry.lock(); - registry.get(&id).map(|entry| entry.status.clone()) - } - - /// Cleans up any completed handles and removes them from the registry. - /// - /// You can do this manually, however this is typically done at the end of the frame. - pub fn cleanup(&self) { - let mut registry = self.handle_registry.lock(); - let completed_ids: Vec<u64> = registry - .iter() - .filter_map(|(&id, entry)| { - matches!(entry.status, FutureStatus::Completed(_)).then_some(id) - }) - .collect(); - - for id in completed_ids { - registry.remove(&id); - } - } -} - -impl Default for FutureQueue { - fn default() -> Self { - Self::new() - } -} - -mod tests { - use std::sync::Arc; - use std::time::Duration; - use tokio::sync::watch; - use tokio::time::sleep; - use crate::future::{FutureHandle, FutureQueue, FutureStatus}; - - #[tokio::test] - async fn test_basic_future_completion() { - let queue = Arc::new(FutureQueue::new()); - - // Push a simple future - let handle = queue.push(async { - sleep(Duration::from_millis(10)).await; - 42i32 - }); - - // Poll to start it - queue.poll(); - - // Wait for completion - tokio::time::sleep(Duration::from_millis(50)).await; - - // Exchange result - let result = queue.exchange_as::<i32>(&handle).unwrap(); - assert_eq!(*result, 42); - } - - #[tokio::test] - async fn test_multiple_futures() { - let queue = Arc::new(FutureQueue::new()); - - let handles: Vec<FutureHandle> = (0..5) - .map(|i| { - queue.push(async move { - sleep(Duration::from_millis(10 + i * 5)).await; - i * 10 - }) - }) - .collect(); - - queue.poll(); - - // Wait for all to complete - tokio::time::sleep(Duration::from_millis(10000)).await; - - // Check all results - for (i, handle) in handles.iter().enumerate() { - let result = queue.exchange_as::<i32>(handle).unwrap(); - assert_eq!(*result, (i * 10) as i32); - } - } - - #[tokio::test] - async fn test_status_tracking() { - let queue = Arc::new(FutureQueue::new()); - - let handle = queue.push(async { - sleep(Duration::from_millis(50)).await; - "done".to_string() - }); - - // Before polling - assert!(matches!(queue.get_status(handle.id), Some(FutureStatus::NotPolled))); - - queue.poll(); - - // After polling, before completion - assert!(matches!(queue.get_status(handle.id), Some(FutureStatus::CurrentlyPolling))); - - // Wait for completion - tokio::time::sleep(Duration::from_millis(100)).await; - - // Should be completed - assert!(matches!(queue.get_status(handle.id), Some(FutureStatus::Completed(_)))); - - // Exchange should still work - let result = queue.exchange_as::<String>(&handle).unwrap(); - assert_eq!(*result, "done"); - } - - #[tokio::test] - async fn test_exchange_before_completion_returns_none() { - let queue = Arc::new(FutureQueue::new()); - - let handle = queue.push(async { - sleep(Duration::from_millis(100)).await; - true - }); - - queue.poll(); - - // Try to exchange immediately — should return None - assert!(queue.exchange_as::<bool>(&handle).is_none()); - - // Wait and try again - tokio::time::sleep(Duration::from_millis(150)).await; - assert!(queue.exchange_as::<bool>(&handle).is_some()); - } - - #[tokio::test] - async fn test_cleanup_removes_completed_handles() { - let queue = Arc::new(FutureQueue::new()); - - let handle = queue.push(async { - sleep(Duration::from_millis(10)).await; - 123i32 - }); - - queue.poll(); - tokio::time::sleep(Duration::from_millis(50)).await; - - // Should be in registry before cleanup - assert!(queue.get_handle(handle.id).is_some()); - - queue.cleanup(); - - // Should be removed after cleanup - assert!(queue.get_handle(handle.id).is_none()); - } - - #[tokio::test] - async fn test_progress_channel_integration() { - let queue = Arc::new(FutureQueue::new()); - - let (progress_tx, mut progress_rx) = watch::channel(0.0f32); - - let handle = queue.push(async move { - progress_tx.send(0.25).unwrap(); - sleep(Duration::from_millis(20)).await; - - progress_tx.send(0.75).unwrap(); - sleep(Duration::from_millis(20)).await; - - progress_tx.send(1.0).unwrap(); - "final_result".to_string() - }); - - queue.poll(); - - // Check progress updates - tokio::time::sleep(Duration::from_millis(10)).await; - assert_eq!(*progress_rx.borrow_and_update(), 0.25); - - tokio::time::sleep(Duration::from_millis(30)).await; - assert_eq!(*progress_rx.borrow_and_update(), 0.75); - - tokio::time::sleep(Duration::from_millis(30)).await; - assert_eq!(*progress_rx.borrow_and_update(), 1.0); - - // Check final result - let result = queue.exchange_as::<String>(&handle).unwrap(); - assert_eq!(*result, "final_result"); - } - - #[tokio::test] - async fn test_error_handling() { - let queue = Arc::new(FutureQueue::new()); - - let handle = queue.push(async { - sleep(Duration::from_millis(10)).await; - Result::<i32, &'static str>::Err("something went wrong") - }); - - queue.poll(); - tokio::time::sleep(Duration::from_millis(50)).await; - - let result = queue.exchange_as::<Result<i32, &'static str>>(&handle).unwrap(); - assert!(result.is_err()); - assert_eq!(result.unwrap_err(), "something went wrong"); - } - - #[tokio::test] - async fn test_get_handle_returns_correct_handle() { - let queue = Arc::new(FutureQueue::new()); - - let handle = queue.push(async { - sleep(Duration::from_millis(10)).await; - 999i32 - }); - - let retrieved_handle = queue.get_handle(handle.id).unwrap(); - assert_eq!(retrieved_handle.id, handle.id); - - // Invalid ID should return None - assert!(queue.get_handle(999999).is_none()); - } - - #[tokio::test] - async fn test_exchange_by_id() { - let queue = Arc::new(FutureQueue::new()); - - let handle = queue.push(async { - sleep(Duration::from_millis(10)).await; - "test_string".to_string() - }); - - queue.poll(); - tokio::time::sleep(Duration::from_millis(50)).await; - - let result = queue.exchange_as::<String>(&handle).unwrap(); - assert_eq!(*result, "test_string"); - } - - #[tokio::test] - async fn test_concurrent_futures() { - let queue = Arc::new(FutureQueue::new()); - - // Push 10 concurrent futures - let handles: Vec<FutureHandle> = (0..10) - .map(|i| { - queue.push(async move { - // Simulate variable work - let delay = Duration::from_millis(10 + (i * 5) as u64); - sleep(delay).await; - i - }) - }) - .collect(); - - queue.poll(); - - // Wait for all to complete - tokio::time::sleep(Duration::from_millis(10000)).await; - - // Verify all results - for (i, handle) in handles.iter().enumerate() { - let result = queue.exchange_as::<usize>(handle).unwrap(); - assert_eq!(*result, i); - } - } - - #[tokio::test] - async fn test_downcast_failure_returns_none() { - let queue = Arc::new(FutureQueue::new()); - - let handle = queue.push(async { - sleep(Duration::from_millis(10)).await; - 42i32 - }); - - queue.poll(); - tokio::time::sleep(Duration::from_millis(50)).await; - - // Try to downcast to wrong type - let result = queue.exchange_as::<String>(&handle); - assert!(result.is_none()); - - // But correct type works - let result = queue.exchange_as::<i32>(&handle).unwrap(); - assert_eq!(*result, 42); - } -} @@ -12,7 +12,6 @@ pub mod resources; pub mod scene; pub mod procedural; pub mod utils; -pub mod future; use app_dirs2::{AppDataType, AppInfo}; use chrono::Local; @@ -31,6 +30,8 @@ use std::{ sync::Arc, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, }; +use std::cell::RefCell; +use std::rc::Rc; use bytemuck::Contiguous; use wgpu::{ BindGroupLayout, Device, Instance, Queue, Surface, SurfaceConfiguration, SurfaceError, @@ -51,7 +52,8 @@ pub use gilrs; use log::LevelFilter; pub use wgpu; pub use winit; -use crate::future::FutureQueue; +pub use dropbear_future_queue as future; +use dropbear_future_queue::{FutureQueue, Throwable}; /// The backend information, such as the device, queue, config, surface, renderer, window and more. pub struct State { @@ -66,14 +68,14 @@ pub struct State { pub instance: Instance, pub viewport_texture: Texture, pub texture_id: Arc<TextureId>, - pub future_queue: Arc<FutureQueue>, + pub future_queue: Throwable<FutureQueue>, pub window: Arc<Window>, } impl State { /// Asynchronously initialised the state and sets up the backend and surface for wgpu to render to. - pub async fn new(window: Arc<Window>, future_queue: Arc<FutureQueue>) -> anyhow::Result<Self> { + pub async fn new(window: Arc<Window>, future_queue: Throwable<FutureQueue>) -> anyhow::Result<Self> { let size = window.inner_size(); // create backend @@ -376,13 +378,13 @@ pub struct App { /// A queue that polls through futures for asynchronous functions /// /// Winit doesn't use async, so this is the next best alternative. - future_queue: Arc<FutureQueue>, + future_queue: Throwable<FutureQueue>, } impl App { /// Creates a new instance of the application. It only sets the default for the struct + the /// window config. - fn new(config: WindowConfiguration, future_queue: Option<Arc<FutureQueue>>) -> Self { + fn new(config: WindowConfiguration, future_queue: Option<Throwable<FutureQueue>>) -> Self { let result = Self { state: None, config: config.clone(), @@ -393,7 +395,7 @@ impl App { target_fps: config.max_fps, // default settings for now gilrs: GilrsBuilder::new().build().unwrap(), - future_queue: future_queue.unwrap_or_else(|| Arc::new(FutureQueue::new())), + future_queue: future_queue.unwrap_or_else(|| Rc::new(RefCell::new(FutureQueue::new()))), }; log::debug!("Created new instance of app"); result @@ -430,7 +432,7 @@ impl App { /// /// It takes an input of a scene manager and an input manager, and expects you to return back the changed /// managers. - pub async fn run<F>(config: WindowConfiguration, app_name: &str, future_queue: Option<Arc<FutureQueue>>, setup: F) -> anyhow::Result<()> + pub async fn run<F>(config: WindowConfiguration, app_name: &str, future_queue: Option<Throwable<FutureQueue>>, setup: F) -> anyhow::Result<()> where F: FnOnce(scene::Manager, input::Manager) -> (scene::Manager, input::Manager), { @@ -553,7 +555,7 @@ impl App { /// /// # Parameters /// * config - [`WindowConfiguration`]: The configuration/settings of the window. -/// * queue - [`Option<Arc<FutureQueue>>`]: An optional value for a [`FutureQueue`] +/// * queue - [`Option<Throwable<FutureQueue>>`]: An optional value for a [`FutureQueue`] /// * setup - [`FnOnce`]: A function that sets up all the scenes. It shouldn't be loaded /// but instead be set as an [`Arc<Mutex<T>>>`]. macro_rules! run_app { @@ -614,7 +616,7 @@ impl ApplicationHandler for App { } WindowEvent::Resized(size) => state.resize(size.width, size.height), WindowEvent::RedrawRequested => { - self.future_queue.poll(); + self.future_queue.borrow_mut().poll(); let frame_start = Instant::now(); @@ -646,7 +648,7 @@ impl ApplicationHandler for App { } state.window.request_redraw(); - self.future_queue.cleanup(); + self.future_queue.borrow_mut().cleanup(); } WindowEvent::KeyboardInput { event: @@ -0,0 +1,17 @@ +[package] +name = "dropbear_future-queue" + +version = "0.1.0" +edition.workspace = true +license-file.workspace = true +repository.workspace = true +readme = "README.md" +description = "A queue for polling futures in a single threaded context" + +[dependencies] +ahash = "0.8.12" +parking_lot = "0.12.4" +tokio = { version = "1", features = ["rt", "sync", "time", "rt-multi-thread"] } + +[dev-dependencies] +tokio = { version = "1", features = ["rt", "sync", "time", "rt-multi-thread"] } @@ -0,0 +1,43 @@ +# dropbear-futurequeue + +A helper queue for polling futures in single threaded systems such as in winit. + +# Example + +```rust +use tokio::runtime::Runtime; +use tokio::time::sleep; +use dropbear_future_queue::FutureQueue; + +fn main() { + // requires a tokio thread + let rt = Runtime::new().unwrap(); + let _guard = rt.enter(); + + // create new queue + let queue = FutureQueue::new(); + + // create a new handle to keep for reference + let handle = queue.push(async move { + sleep(1000).await; + 67 + 41 + }); + + // assume this is the event loop + loop { + // executes all the futures in the database + queue.poll(); + + println!("Current status of compututation: {:?}", queue.get_status(&handle)); + + // check if it is ready to be taken + if let Some(result) = queue.exchange_as::<i32>(&handle) { + println!("67 + 41 = {}", result); + break; + } + + // cleans up any ids not needed anymore. + queue.cleanup() + } +} +``` @@ -0,0 +1,323 @@ +//! Enabling multithreading for functions and apps that are purely single threaded. +//! +//! This was originally a module in my [dropbear](https://github.com/4tkbytes/dropbear) game engine, +//! however I thought there were barely any libraries that had future queuing. It takes inspiration +//! from Unity and how they handle their events. +//! +//! # Example +//! ```rust +//! use tokio::runtime::Runtime; +//! use tokio::time::sleep; +//! use dropbear_future_queue::FutureQueue; +//! +//! // requires a tokio thread +//! let rt = Runtime::new().unwrap(); +//! let _guard = rt.enter(); +//! +//! // create new queue +//! let queue = FutureQueue::new(); +//! +//! // create a new handle to keep for reference +//! let handle = queue.push(async move { +//! sleep(1000).await; +//! 67 + 41 +//! }); +//! +//! // assume this is the event loop +//! loop { +//! // executes all the futures in the database +//! queue.poll(); +//! +//! println!("Current status of compututation: {:?}", queue.get_status(&handle)); +//! +//! // check if it is ready to be taken +//! if let Some(result) = queue.exchange_as::<i32>(&handle) { +//! println!("67 + 41 = {}", result); +//! break; +//! } +//! +//! // cleans up any ids not needed anymore. +//! queue.cleanup() +//! } +//! ``` + +use std::any::Any; +use std::cell::RefCell; +use std::collections::VecDeque; +use std::pin::Pin; +use std::sync::Arc; +use ahash::{HashMap, HashMapExt}; +use parking_lot::Mutex; +use tokio::sync::oneshot; +use std::future::Future; +use std::rc::Rc; + +/// A type used for a future. +/// +/// It must include a [`Send`] trait to be usable for the [`FutureQueue`] +pub type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send>>; +/// A clonable generic result. It can/preferred to be downcasted to your specific type. +pub type AnyResult = Arc<dyn Any + Send + Sync>; +/// Internal function: A result receiver +type ResultReceiver = oneshot::Receiver<AnyResult>; +/// A type for storing the queue. It uses a [`VecDeque`] to store [`FutureHandle`]'s and [`BoxFuture`] +pub type FutureStorage = Arc<Mutex<VecDeque<(FutureHandle, BoxFuture<()>)>>>; +/// A type recommended to be used by [`FutureQueue`] to allow being thrown around in your app +pub type Throwable<T> = Rc<RefCell<T>>; + +/// A status showing the future, used by the [`ResultReceiver`] and [`ResultSender`] +#[derive(Clone)] +pub enum FutureStatus { + NotPolled, + CurrentlyPolling, + Completed(AnyResult), +} + +/// A handle to the future task +#[derive(Default, Copy, Clone, Eq, Hash, PartialEq, Debug)] +pub struct FutureHandle { + pub id: u64, +} + +/// Internal storage per handle — separate from FutureHandle +struct HandleEntry { + receiver: ResultReceiver, + status: FutureStatus, +} + +/// A queue for polling futures. It is stored in here until [`FutureQueue::poll`] is run. +pub struct FutureQueue { + /// The queue for the futures. + queued: FutureStorage, + /// A place to store all handle data + handle_registry: Arc<Mutex<HashMap<FutureHandle, HandleEntry>>>, + /// Next id to be processed + next_id: Arc<Mutex<u64>>, +} + +impl FutureQueue { + /// Creates a new [`Arc<FutureQueue>`]. + pub fn new() -> Self { + Self { + queued: Arc::new(Mutex::new(VecDeque::new())), + handle_registry: Arc::new(Mutex::new(HashMap::new())), + next_id: Arc::new(Mutex::new(0)), + } + } + + /// Pushes a future to the FutureQueue. It will sit and wait + /// to be processed until [`FutureQueue::poll`] is called. + pub fn push<F, T>(&self, future: F) -> FutureHandle + where + F: Future<Output = T> + Send + 'static, + T: Send + Sync + 'static, + { + let mut next_id = self.next_id.lock(); + let id = *next_id; + *next_id += 1; + + let id = FutureHandle { id }; + + let (sender, receiver) = oneshot::channel::<AnyResult>(); + + let entry = HandleEntry { + receiver, + status: FutureStatus::NotPolled, + }; + + self.handle_registry.lock().insert(id, entry); + + let registry_clone = self.handle_registry.clone(); + + let wrapped_future: Pin<Box<dyn Future<Output = ()> + Send>> = Box::pin(async move { + log("Starting future execution"); + let result = future.await; + let boxed_result: AnyResult = Arc::new(result); + log("Future completed, sending result"); + + let _ = sender.send(boxed_result.clone()); + + let mut registry = registry_clone.lock(); + if let Some(entry) = registry.get_mut(&id) { + entry.status = FutureStatus::Completed(boxed_result); + log("Updated status to completed"); + } + }); + + self.queued.lock().push_back((id, wrapped_future)); + + id + } + + /// Polls all the futures in the future queue and resolves the handles. + /// + /// This function spawns a new async thread for each item inside the thread and + /// sends updates to the Handle's receiver. + pub fn poll(&self) { + let mut queue = self.queued.lock(); + log("Locked queue for polling"); + + if queue.is_empty() { + log("Queue is empty, nothing to poll"); + return; + } + + let mut futures_to_spawn = Vec::new(); + + while let Some((id, future)) = queue.pop_front() { + log(format!("Processing future with id: {:?}", id)); + + { + let mut registry = self.handle_registry.lock(); + if let Some(entry) = registry.get_mut(&id) { + entry.status = FutureStatus::CurrentlyPolling; + log("Updated status to CurrentlyPolling"); + } + } + + futures_to_spawn.push(future); + } + + drop(queue); + + for future in futures_to_spawn { + log("Spawning future with tokio"); + tokio::spawn(future); + } + } + + /// Exchanges the future for the result. + /// + /// When the handle is not successful, it will return nothing. When the handle is successful, + /// it will return the result and drop the handle, removing the usage of it. + pub fn exchange(&self, handle: &FutureHandle) -> Option<AnyResult> { + let mut registry = self.handle_registry.lock(); + if let Some(entry) = registry.get_mut(handle) { + return match &entry.status { + FutureStatus::Completed(result) => { + log("FutureStatus::Completed - returning cached result"); + Some(result.clone()) + } + _ => { + log("Future not completed yet, checking receiver"); + match entry.receiver.try_recv() { + Ok(result) => { + log("Received result from channel"); + entry.status = FutureStatus::Completed(result.clone()); + Some(result) + } + Err(oneshot::error::TryRecvError::Empty) => { + log("Channel is empty - future still running"); + None + }, + Err(oneshot::error::TryRecvError::Closed) => { + log("Channel is closed - future may have panicked"); + None + }, + } + } + } + } else { + log("Handle not found in registry"); + } + None + } + + /// Exchanges the handle and safely downcasts it into a specific type. + pub fn exchange_as<T: Any + Send + Sync + 'static>(&self, handle: &FutureHandle) -> Option<Arc<T>> { + self.exchange(handle)? + .downcast() + .ok() + } + + /// Get status of a handle + pub fn get_status(&self, handle: &FutureHandle) -> Option<FutureStatus> { + let registry = self.handle_registry.lock(); + registry.get(handle).map(|entry| entry.status.clone()) + } + + /// Cleans up any completed handles and removes them from the registry. + /// + /// You can do this manually, however this is typically done at the end of the frame. + pub fn cleanup(&self) { + let mut registry = self.handle_registry.lock(); + let completed_ids: Vec<FutureHandle> = registry + .iter() + .filter_map(|(&id, entry)| { + matches!(entry.status, FutureStatus::Completed(_)).then_some(id) + }) + .collect(); + + for id in completed_ids { + registry.remove(&id); + } + } +} + + +/// Internal function for logging to a file for tests (when stdout is not available). +/// +/// Only logs if the [`LOG_TO_FILE`] constant is set to true. +#[cfg(test)] +fn log(msg: impl ToString) { + use std::io::Write; + + let mut file = std::fs::OpenOptions::new().append(true).create(true).open("test.log").unwrap(); + file.write_all(format!("{}\n", msg.to_string()).as_bytes()).unwrap(); +} + +#[cfg(not(test))] +fn log(_msg: impl ToString) { + +} + +impl Default for FutureQueue { + fn default() -> Self { + Self::new() + } +} + +#[test] +fn test_future_queue() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let _guard = rt.enter(); + + let queue = FutureQueue::new(); + log("Created new queue"); + + let handle = queue.push(async move { + log("Inside the pushed future - starting work"); + tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; + log("Inside the pushed future - work completed"); + 67 + 41 + }); + log("Created new handle"); + + queue.poll(); + log("Initial poll completed"); + + let mut attempts = 0; + let max_attempts = 100; + let elapsed = std::time::Instant::now(); + + loop { + let now = std::time::Instant::now(); + attempts += 1; + log(format!("Attempt {}: Checking for result", attempts)); + log(format!("Time since last attempt: {} ms", elapsed.elapsed().as_millis() - now.elapsed().as_millis())); + + if let Some(result) = queue.exchange(&handle) { + let result = result.downcast::<i32>().unwrap(); + log(format!("Success! 67 + 41 = {}", result)); + assert_eq!(*result, 108); + break; + } + + if attempts >= max_attempts { + log("Max attempts reached - test failed"); + panic!("Future never completed"); + } + } + + log("Test completed successfully"); +} @@ -0,0 +1,53 @@ +Created new queue +Created new handle +Locked queue for polling +Processing future with id: FutureHandle { id: 0 } +Updated status to CurrentlyPolling +Spawning future with tokio +Initial poll completed +Attempt 1: Checking for result +Created new queue +Created new handle +Locked queue for polling +Processing future with id: FutureHandle { id: 0 } +Updated status to CurrentlyPolling +Spawning future with tokio +Initial poll completed +Attempt 1: Checking for result +Time since last attempt: 0 +Future not completed yet, checking receiver +Channel is empty - future still running +Starting future execution +Inside the pushed future - starting work +Attempt 2: Checking for result +Time since last attempt: 2 +Future not completed yet, checking receiver +Channel is empty - future still running +Attempt 3: Checking for result +Time since last attempt: 4 +Future not completed yet, checking receiver +Channel is empty - future still running +Attempt 4: Checking for result +Time since last attempt: 6 +Future not completed yet, checking receiver +Channel is empty - future still running +Attempt 5: Checking for result +Time since last attempt: 8 +Future not completed yet, checking receiver +Channel is empty - future still running +Attempt 6: Checking for result +Time since last attempt: 10 +Future not completed yet, checking receiver +Channel is empty - future still running +Attempt 7: Checking for result +Time since last attempt: 12 +Future not completed yet, checking receiver +Channel is empty - future still running +Inside the pushed future - work completed +Future completed, sending result +Updated status to completed +Attempt 8: Checking for result +Time since last attempt: 14 +FutureStatus::Completed - returning cached result +Success! 67 + 41 = 108 +Test completed successfully