|
| 1 | +use anyhow::Result; |
| 2 | +use std::sync::Arc; |
| 3 | +use tokio::sync::{mpsc, oneshot}; |
| 4 | + |
| 5 | +pub trait Task: Send + 'static { |
| 6 | + type Output: Send + 'static; |
| 7 | + fn run(&self) -> Result<Self::Output>; |
| 8 | +} |
| 9 | + |
| 10 | +trait TaskTrait: Send { |
| 11 | + fn run_boxed(self: Box<Self>); |
| 12 | +} |
| 13 | + |
| 14 | +impl<T: Task> TaskTrait for T { |
| 15 | + fn run_boxed(self: Box<Self>) { |
| 16 | + match self.run() { |
| 17 | + Ok(_) => { /* Task succeeded, do nothing */ } |
| 18 | + Err(e) => { |
| 19 | + // Log the error if the task failed. |
| 20 | + // Consider adding a proper logging mechanism later. |
| 21 | + eprintln!("Task failed: {}", e); |
| 22 | + } |
| 23 | + } |
| 24 | + } |
| 25 | +} |
| 26 | + |
| 27 | +#[derive(Clone)] |
| 28 | +pub struct Queue { |
| 29 | + inner: Arc<QueueInner>, |
| 30 | +} |
| 31 | + |
| 32 | +struct QueueInner { |
| 33 | + sender: mpsc::Sender<Box<dyn TaskTrait>>, |
| 34 | + shutdown_sender: Option<oneshot::Sender<()>>, |
| 35 | +} |
| 36 | + |
| 37 | +impl Queue { |
| 38 | + pub fn new() -> Self { |
| 39 | + let (sender, mut receiver) = mpsc::channel::<Box<dyn TaskTrait>>(32); // Channel for tasks |
| 40 | + let (shutdown_tx, mut shutdown_rx) = oneshot::channel(); |
| 41 | + |
| 42 | + tokio::spawn(async move { |
| 43 | + loop { |
| 44 | + tokio::select! { |
| 45 | + Some(task) = receiver.recv() => { |
| 46 | + task.run_boxed(); |
| 47 | + } |
| 48 | + _ = &mut shutdown_rx => { |
| 49 | + // Drain the channel before shutting down? Optional. |
| 50 | + // For now, just break. |
| 51 | + break; |
| 52 | + } |
| 53 | + else => break, |
| 54 | + } |
| 55 | + } |
| 56 | + }); |
| 57 | + |
| 58 | + Self { |
| 59 | + inner: Arc::new(QueueInner { |
| 60 | + sender, |
| 61 | + shutdown_sender: Some(shutdown_tx), |
| 62 | + }), |
| 63 | + } |
| 64 | + } |
| 65 | + |
| 66 | + /// Submits a task to the queue asynchronously, waiting if the channel is full. |
| 67 | + /// The task is executed in the background, and its result is ignored. |
| 68 | + pub async fn submit<T>(&self, task: T) -> Result<()> |
| 69 | + where |
| 70 | + T: Task + 'static, |
| 71 | + { |
| 72 | + self.inner |
| 73 | + .sender |
| 74 | + .send(Box::new(task)) |
| 75 | + .await |
| 76 | + .map_err(|e| anyhow::anyhow!("Failed to submit task: {}", e)) |
| 77 | + } |
| 78 | +} |
| 79 | + |
| 80 | +impl Default for Queue { |
| 81 | + fn default() -> Self { |
| 82 | + Self::new() |
| 83 | + } |
| 84 | +} |
| 85 | + |
| 86 | +impl Drop for QueueInner { |
| 87 | + fn drop(&mut self) { |
| 88 | + if let Some(sender) = self.shutdown_sender.take() { |
| 89 | + sender.send(()).ok(); |
| 90 | + } |
| 91 | + } |
| 92 | +} |
| 93 | + |
| 94 | +#[cfg(test)] |
| 95 | +mod tests { |
| 96 | + use super::*; |
| 97 | + use anyhow::anyhow; |
| 98 | + use std::time::Duration; |
| 99 | + use tokio::time::sleep; |
| 100 | + |
| 101 | + struct TestTask(i32); |
| 102 | + impl Task for TestTask { |
| 103 | + type Output = i32; |
| 104 | + fn run(&self) -> Result<Self::Output> { |
| 105 | + std::thread::sleep(Duration::from_millis(10)); |
| 106 | + Ok(self.0 * 2) |
| 107 | + } |
| 108 | + } |
| 109 | + |
| 110 | + struct ErrorTask; |
| 111 | + impl Task for ErrorTask { |
| 112 | + type Output = (); |
| 113 | + fn run(&self) -> Result<Self::Output> { |
| 114 | + Err(anyhow!("Task failed intentionally")) |
| 115 | + } |
| 116 | + } |
| 117 | + |
| 118 | + #[tokio::test] |
| 119 | + async fn test_submit_and_process() { |
| 120 | + let queue = Queue::new(); |
| 121 | + // Submit a few tasks |
| 122 | + for i in 0..5 { |
| 123 | + queue.submit(TestTask(i)).await.unwrap(); |
| 124 | + } |
| 125 | + // Submit a task that will fail |
| 126 | + queue.submit(ErrorTask).await.unwrap(); |
| 127 | + |
| 128 | + // Allow some time for tasks to be processed by the background worker. |
| 129 | + // In a real scenario, you might not wait like this, but for testing, |
| 130 | + // we need to ensure the background task has a chance to run. |
| 131 | + sleep(Duration::from_millis(100)).await; |
| 132 | + |
| 133 | + // We can't directly assert results here, but we can check the queue still works. |
| 134 | + queue.submit(TestTask(10)).await.unwrap(); |
| 135 | + sleep(Duration::from_millis(50)).await; // Allow time for the last task |
| 136 | + } |
| 137 | + |
| 138 | + #[tokio::test] |
| 139 | + async fn test_channel_backpressure_submit() { |
| 140 | + let queue = Queue::new(); |
| 141 | + |
| 142 | + // Fill the channel (channel size is 32) using submit |
| 143 | + let mut tasks = Vec::new(); |
| 144 | + for i in 0..32 { |
| 145 | + let queue_clone = queue.clone(); |
| 146 | + // Spawn tasks to submit concurrently, as submit waits |
| 147 | + tasks.push(tokio::spawn(async move { |
| 148 | + queue_clone |
| 149 | + .submit(TestTask(i)) |
| 150 | + .await |
| 151 | + .expect("Submit should succeed"); |
| 152 | + })); |
| 153 | + } |
| 154 | + // Wait for all initial submissions to likely be sent (though not necessarily processed) |
| 155 | + for task in tasks { |
| 156 | + task.await.unwrap(); |
| 157 | + } |
| 158 | + |
| 159 | + // Try submitting one more task. This should wait until a slot is free. |
| 160 | + // We'll use a timeout to ensure it doesn't block forever if something is wrong. |
| 161 | + let submit_task = queue.submit(TestTask(33)); |
| 162 | + match tokio::time::timeout(Duration::from_millis(200), submit_task).await { |
| 163 | + Ok(Ok(_)) => { /* Successfully submitted after waiting */ } |
| 164 | + Ok(Err(e)) => panic!("Submit failed unexpectedly: {}", e), |
| 165 | + Err(_) => panic!("Submit timed out, likely blocked due to backpressure not resolving"), |
| 166 | + } |
| 167 | + |
| 168 | + // Allow time for processing |
| 169 | + sleep(Duration::from_millis(100)).await; |
| 170 | + } |
| 171 | + |
| 172 | + #[tokio::test] |
| 173 | + async fn test_shutdown() { |
| 174 | + let queue = Queue::new(); |
| 175 | + queue.submit(TestTask(1)).await.unwrap(); |
| 176 | + queue.submit(TestTask(2)).await.unwrap(); |
| 177 | + // Queue is dropped here, triggering shutdown |
| 178 | + drop(queue); |
| 179 | + |
| 180 | + // Allow time for shutdown signal to be processed and potentially |
| 181 | + // for the background task to finish ongoing work (though not guaranteed here). |
| 182 | + sleep(Duration::from_millis(100)).await; |
| 183 | + // No direct assertion, just checking it doesn't panic/hang. |
| 184 | + } |
| 185 | + |
| 186 | + #[tokio::test] |
| 187 | + async fn test_queue_cloning() { |
| 188 | + let queue1 = Queue::new(); |
| 189 | + let queue2 = queue1.clone(); |
| 190 | + |
| 191 | + // Submit tasks via both clones |
| 192 | + let task1 = queue1.submit(TestTask(10)); |
| 193 | + let task2 = queue2.submit(TestTask(20)); |
| 194 | + |
| 195 | + // Wait for submissions to complete |
| 196 | + tokio::try_join!(task1, task2).unwrap(); |
| 197 | + |
| 198 | + // Allow time for processing |
| 199 | + sleep(Duration::from_millis(100)).await; |
| 200 | + } |
| 201 | + |
| 202 | + #[tokio::test] |
| 203 | + async fn test_error_task_does_not_stop_queue() { |
| 204 | + let queue = Queue::new(); |
| 205 | + |
| 206 | + queue.submit(TestTask(1)).await.unwrap(); |
| 207 | + queue.submit(ErrorTask).await.unwrap(); // Submit the failing task |
| 208 | + queue.submit(TestTask(2)).await.unwrap(); |
| 209 | + |
| 210 | + // Allow time for tasks to process |
| 211 | + sleep(Duration::from_millis(100)).await; |
| 212 | + |
| 213 | + // Submit another task to ensure the queue is still running after the error |
| 214 | + queue.submit(TestTask(3)).await.unwrap(); |
| 215 | + sleep(Duration::from_millis(50)).await; |
| 216 | + // If we reach here without panic, the queue continued after the error. |
| 217 | + // We expect an error message "Task failed: Task failed intentionally" |
| 218 | + // to be printed to stderr during the test run. |
| 219 | + } |
| 220 | +} |
0 commit comments