eredu_runtime/cache/
worker.rs

1//! Backend-neutral bounded physical worker for cache backing-store tasks.
2
3use std::{
4    collections::HashMap,
5    panic::{catch_unwind, AssertUnwindSafe},
6    sync::{
7        atomic::{AtomicBool, Ordering},
8        mpsc, Arc, Condvar, Mutex,
9    },
10    thread::{self, JoinHandle},
11};
12
13use super::{
14    CacheIoAdmission, CacheIoCompletionDisposition, CacheIoExecutionState,
15    CacheIoExecutionStateError, CacheIoOperationKey, CacheIoPreparation, CacheIoStartDisposition,
16};
17
18enum CacheIoWorkerRequest<Task, Output> {
19    Operation {
20        key: CacheIoOperationKey,
21        task: Box<Task>,
22        completion: Arc<CacheIoCompletion<Output>>,
23    },
24    Stop,
25}
26
27#[derive(Debug, Clone)]
28enum CacheIoCompletionState<Output> {
29    Finished(Result<Output, String>),
30    Cancelled,
31}
32
33#[derive(Debug)]
34struct CacheIoCompletion<Output> {
35    state: Mutex<Option<CacheIoCompletionState<Output>>>,
36    ready: Condvar,
37    released: Mutex<bool>,
38    released_ready: Condvar,
39}
40
41impl<Output> Default for CacheIoCompletion<Output> {
42    fn default() -> Self {
43        Self {
44            state: Mutex::new(None),
45            ready: Condvar::new(),
46            released: Mutex::new(false),
47            released_ready: Condvar::new(),
48        }
49    }
50}
51
52impl<Output> CacheIoCompletion<Output> {
53    fn finish(&self, result: Result<Output, String>) {
54        if let Ok(mut state) = self.state.lock() {
55            if state.is_none() {
56                *state = Some(CacheIoCompletionState::Finished(result));
57                self.ready.notify_all();
58            }
59        }
60    }
61
62    fn cancel(&self) -> bool {
63        let Ok(mut state) = self.state.lock() else {
64            return false;
65        };
66        if state.is_some() {
67            return false;
68        }
69        *state = Some(CacheIoCompletionState::Cancelled);
70        self.ready.notify_all();
71        true
72    }
73
74    fn is_ready(&self) -> bool {
75        self.state.lock().map_or(true, |state| state.is_some())
76    }
77
78    fn release_task_resources(&self) {
79        if let Ok(mut released) = self.released.lock() {
80            *released = true;
81            self.released_ready.notify_all();
82        }
83    }
84
85    fn wait_for_task_resources(&self) -> Result<(), CacheIoWorkerError> {
86        let mut released = self
87            .released
88            .lock()
89            .map_err(|_| CacheIoWorkerError::Poisoned)?;
90        while !*released {
91            released = self
92                .released_ready
93                .wait(released)
94                .map_err(|_| CacheIoWorkerError::Poisoned)?;
95        }
96        Ok(())
97    }
98}
99
100impl<Output: Clone> CacheIoCompletion<Output> {
101    fn wait(&self, generation: u64) -> Result<Output, CacheIoWorkerError> {
102        let mut state = self
103            .state
104            .lock()
105            .map_err(|_| CacheIoWorkerError::Poisoned)?;
106        while state.is_none() {
107            state = self
108                .ready
109                .wait(state)
110                .map_err(|_| CacheIoWorkerError::Poisoned)?;
111        }
112        match state.as_ref().expect("completion state was awaited") {
113            CacheIoCompletionState::Finished(Ok(output)) => Ok(output.clone()),
114            CacheIoCompletionState::Finished(Err(error)) => {
115                Err(CacheIoWorkerError::OperationFailed(error.clone()))
116            }
117            CacheIoCompletionState::Cancelled => Err(CacheIoWorkerError::Cancelled { generation }),
118        }
119    }
120}
121
122#[derive(Debug)]
123struct CacheIoWorkerShared<Output> {
124    in_flight: Mutex<HashMap<CacheIoOperationKey, Arc<CacheIoCompletion<Output>>>>,
125    execution: Mutex<CacheIoExecutionState>,
126    space_available: Condvar,
127    stopping: AtomicBool,
128    shutdown_polling: AtomicBool,
129}
130
131impl<Output> CacheIoWorkerShared<Output> {
132    fn new(capacity: usize) -> Result<Self, CacheIoWorkerError> {
133        Ok(Self {
134            in_flight: Mutex::new(HashMap::new()),
135            execution: Mutex::new(CacheIoExecutionState::new(capacity)?),
136            space_available: Condvar::new(),
137            stopping: AtomicBool::new(false),
138            shutdown_polling: AtomicBool::new(false),
139        })
140    }
141}
142
143/// Exact completion ownership for one coalesced cache I/O operation.
144pub struct CacheIoTicket<Output> {
145    /// Exact logical operation identity.
146    pub key: CacheIoOperationKey,
147    completion: Arc<CacheIoCompletion<Output>>,
148    shared: Arc<CacheIoWorkerShared<Output>>,
149}
150
151impl<Output> Clone for CacheIoTicket<Output> {
152    fn clone(&self) -> Self {
153        Self {
154            key: self.key.clone(),
155            completion: Arc::clone(&self.completion),
156            shared: Arc::clone(&self.shared),
157        }
158    }
159}
160
161impl<Output> std::fmt::Debug for CacheIoTicket<Output> {
162    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
163        formatter
164            .debug_struct("CacheIoTicket")
165            .field("key", &self.key)
166            .finish_non_exhaustive()
167    }
168}
169
170impl<Output: Clone> CacheIoTicket<Output> {
171    /// Waits for the logical output or cancellation.
172    pub fn wait(&self) -> Result<Output, CacheIoWorkerError> {
173        self.completion.wait(self.key.generation)
174    }
175
176    /// Cancels prepared, queued, or in-flight work exactly once.
177    pub fn cancel(&self) -> bool {
178        let Ok(mut execution) = self.shared.execution.lock() else {
179            return false;
180        };
181        let cancelled = execution.cancel(&self.key) && self.completion.cancel();
182        self.shared.space_available.notify_all();
183        cancelled
184    }
185
186    /// Waits until all backend task inputs and retained resources are dropped.
187    pub fn wait_for_task_resources(&self) -> Result<(), CacheIoWorkerError> {
188        self.completion.wait_for_task_resources()
189    }
190
191    /// Returns whether two tickets join the same exact completion owner.
192    pub fn shares_completion_with(&self, other: &Self) -> bool {
193        Arc::ptr_eq(&self.completion, &other.completion)
194    }
195}
196
197/// Prepared cache I/O that admits physical work only when explicitly enqueued.
198pub struct CacheIoSubmission<Task, Output> {
199    /// Ticket shared by the operation owner and all exact-key joiners.
200    pub ticket: CacheIoTicket<Output>,
201    sender: mpsc::Sender<CacheIoWorkerRequest<Task, Output>>,
202    shared: Arc<CacheIoWorkerShared<Output>>,
203    unsent: Option<CacheIoWorkerRequest<Task, Output>>,
204    joined_task: Option<Task>,
205    /// Whether this submission joined an already prepared exact operation.
206    pub joined: bool,
207}
208
209/// Physical admission observations for one submission.
210#[derive(Debug, Clone, Copy, Eq, PartialEq)]
211pub struct CacheIoSubmissionOutcome {
212    /// Whether this submission joined an existing exact operation.
213    pub joined: bool,
214    /// Whether finite queue capacity delayed physical admission.
215    pub backpressure: bool,
216    /// Largest observed physical queue occupancy.
217    pub peak_occupancy: usize,
218}
219
220impl<Task, Output: Clone> CacheIoSubmission<Task, Output> {
221    /// Returns the unused backend task when this submission joined existing work.
222    ///
223    /// Backends may disarm task-local rollback guards before the unused task is
224    /// dropped; the task is never physically executed.
225    pub fn joined_task_mut(&mut self) -> Option<&mut Task> {
226        self.joined_task.as_mut()
227    }
228
229    /// Admits this prepared task, blocking only on finite queue capacity.
230    pub fn enqueue(mut self) -> Result<CacheIoSubmissionOutcome, CacheIoWorkerError> {
231        let mut backpressure = false;
232        if let Some(request) = self.unsent.take() {
233            let mut execution = match self.shared.execution.lock() {
234                Ok(execution) => execution,
235                Err(_) => {
236                    drop(request);
237                    self.ticket.completion.release_task_resources();
238                    return Err(CacheIoWorkerError::Poisoned);
239                }
240            };
241            loop {
242                if self.shared.stopping.load(Ordering::Acquire) {
243                    execution.cancel(&self.ticket.key);
244                    drop(execution);
245                    drop(request);
246                    self.ticket
247                        .completion
248                        .finish(Err("cache I/O physical worker stopped".into()));
249                    self.ticket.completion.release_task_resources();
250                    retire_completion(&self.shared, &self.ticket.key, &self.ticket.completion);
251                    return Err(CacheIoWorkerError::OperationFailed(
252                        "cache I/O physical worker stopped".into(),
253                    ));
254                }
255                match execution.admit(&self.ticket.key)? {
256                    CacheIoAdmission::Admitted => {
257                        if self.sender.send(request).is_err() {
258                            execution.rollback_admission(&self.ticket.key)?;
259                            self.ticket
260                                .completion
261                                .finish(Err("cache I/O physical worker stopped".into()));
262                            self.ticket.completion.release_task_resources();
263                        }
264                        break;
265                    }
266                    CacheIoAdmission::AtCapacity => {
267                        backpressure = true;
268                        // Opt-in shutdown cannot lock execution from Drop:
269                        // backend task cleanup may itself need a native lock.
270                        // A bounded check also closes a shutdown notification
271                        // racing between the predicate check and this wait.
272                        let waited = if self.shared.shutdown_polling.load(Ordering::Acquire) {
273                            self.shared
274                                .space_available
275                                .wait_timeout(execution, std::time::Duration::from_millis(25))
276                                .map(|(execution, _)| execution)
277                                .map_err(|_| ())
278                        } else {
279                            self.shared.space_available.wait(execution).map_err(|_| ())
280                        };
281                        execution = match waited {
282                            Ok(execution) => execution,
283                            Err(_) => {
284                                drop(request);
285                                self.ticket.completion.release_task_resources();
286                                return Err(CacheIoWorkerError::Poisoned);
287                            }
288                        };
289                    }
290                    CacheIoAdmission::Cancelled => {
291                        drop(request);
292                        self.ticket.completion.release_task_resources();
293                        break;
294                    }
295                }
296            }
297            drop(execution);
298        }
299        Ok(CacheIoSubmissionOutcome {
300            joined: self.joined,
301            backpressure,
302            peak_occupancy: self
303                .shared
304                .execution
305                .lock()
306                .map_err(|_| CacheIoWorkerError::Poisoned)?
307                .peak_queued(),
308        })
309    }
310}
311
312impl<Task, Output> Drop for CacheIoSubmission<Task, Output> {
313    fn drop(&mut self) {
314        let Some(request) = self.unsent.take() else {
315            return;
316        };
317        if let Ok(mut execution) = self.shared.execution.lock() {
318            execution.cancel(&self.ticket.key);
319        }
320        drop(request);
321        self.ticket.completion.release_task_resources();
322        retire_completion(&self.shared, &self.ticket.key, &self.ticket.completion);
323    }
324}
325
326/// Generic bounded background worker over opaque backend task and output types.
327pub struct CacheIoWorker<Task, Output> {
328    sender: mpsc::Sender<CacheIoWorkerRequest<Task, Output>>,
329    handle: Mutex<Option<JoinHandle<()>>>,
330    shared: Arc<CacheIoWorkerShared<Output>>,
331    nonblocking_drop: bool,
332}
333
334impl<Task, Output> std::fmt::Debug for CacheIoWorker<Task, Output> {
335    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
336        formatter
337            .debug_struct("CacheIoWorker")
338            .finish_non_exhaustive()
339    }
340}
341
342impl<Task, Output> CacheIoWorker<Task, Output>
343where
344    Task: Send + 'static,
345    Output: Clone + Send + 'static,
346{
347    /// Starts a bounded worker using statically dispatched task and cleanup functions.
348    pub fn new(
349        capacity: usize,
350        thread_name: impl Into<String>,
351        execute: fn(Task) -> Result<Output, String>,
352        discard: fn(Output),
353    ) -> Result<Self, CacheIoWorkerError> {
354        let thread_name = thread_name.into();
355        let (sender, receiver) = mpsc::channel::<CacheIoWorkerRequest<Task, Output>>();
356        let shared = Arc::new(CacheIoWorkerShared::new(capacity)?);
357        let worker_shared = Arc::clone(&shared);
358        let handle = thread::Builder::new()
359            .name(thread_name.clone())
360            .spawn(move || {
361                while let Ok(request) = receiver.recv() {
362                    match request {
363                        CacheIoWorkerRequest::Operation {
364                            key,
365                            task,
366                            completion,
367                        } => {
368                            if worker_shared.stopping.load(Ordering::Acquire) {
369                                if let Ok(mut execution) = worker_shared.execution.lock() {
370                                    execution.cancel(&key);
371                                    // Consume the physical queue slot and move
372                                    // cancellation to a retireable state.
373                                    let _ = execution.begin(&key);
374                                }
375                                drop(task);
376                                completion.finish(Err("cache I/O physical worker stopped".into()));
377                                retire_completion(&worker_shared, &key, &completion);
378                                completion.release_task_resources();
379                                continue;
380                            }
381                            let start = worker_shared
382                                .execution
383                                .lock()
384                                .map_err(|_| CacheIoWorkerError::Poisoned)
385                                .and_then(|mut execution| {
386                                    execution.begin(&key).map_err(Into::into)
387                                });
388                            worker_shared.space_available.notify_all();
389                            match start {
390                                Ok(CacheIoStartDisposition::Execute) => {}
391                                Ok(CacheIoStartDisposition::Discard) => {
392                                    drop(task);
393                                    completion.release_task_resources();
394                                    retire_completion(&worker_shared, &key, &completion);
395                                    continue;
396                                }
397                                Err(error) => {
398                                    drop(task);
399                                    completion.finish(Err(error.to_string()));
400                                    completion.release_task_resources();
401                                    retire_completion(&worker_shared, &key, &completion);
402                                    continue;
403                                }
404                            }
405                            let result = catch_unwind(AssertUnwindSafe(|| execute(*task)))
406                                .unwrap_or_else(|_| {
407                                    Err("cache I/O physical worker operation panicked".into())
408                                });
409                            let disposition = worker_shared
410                                .execution
411                                .lock()
412                                .map_err(|_| CacheIoWorkerError::Poisoned)
413                                .and_then(|mut execution| {
414                                    execution.complete(&key).map_err(Into::into)
415                                });
416                            if !matches!(disposition, Ok(CacheIoCompletionDisposition::Publish))
417                                || completion.is_ready()
418                            {
419                                if let Ok(output) = result {
420                                    discard(output);
421                                }
422                            } else {
423                                completion.finish(result);
424                            }
425                            completion.release_task_resources();
426                            retire_completion(&worker_shared, &key, &completion);
427                        }
428                        CacheIoWorkerRequest::Stop => break,
429                    }
430                }
431            })
432            .map_err(|source| CacheIoWorkerError::Spawn {
433                thread_name,
434                source,
435            })?;
436        Ok(Self {
437            sender,
438            handle: Mutex::new(Some(handle)),
439            shared,
440            nonblocking_drop: false,
441        })
442    }
443
444    /// Requests shutdown without joining from this handle's destructor.
445    ///
446    /// The worker owns in-flight tasks until they finish and discards queued
447    /// tasks itself. Prepared submissions reject admission after shutdown.
448    pub fn with_nonblocking_drop(mut self) -> Self {
449        self.nonblocking_drop = true;
450        self.shared.shutdown_polling.store(true, Ordering::Release);
451        self
452    }
453
454    /// Prepares new work or joins an exact operation already owned by the worker.
455    pub fn prepare(
456        &self,
457        key: CacheIoOperationKey,
458        task: Task,
459    ) -> Result<CacheIoSubmission<Task, Output>, CacheIoWorkerError> {
460        let mut execution = self
461            .shared
462            .execution
463            .lock()
464            .map_err(|_| CacheIoWorkerError::Poisoned)?;
465        let preparation = execution.prepare(key.clone());
466        let mut completions = self
467            .shared
468            .in_flight
469            .lock()
470            .map_err(|_| CacheIoWorkerError::Poisoned)?;
471        if preparation == CacheIoPreparation::Joined {
472            let completion = completions
473                .get(&key)
474                .expect("runtime joined key has an exact completion");
475            return Ok(CacheIoSubmission {
476                ticket: CacheIoTicket {
477                    key,
478                    completion: Arc::clone(completion),
479                    shared: Arc::clone(&self.shared),
480                },
481                sender: self.sender.clone(),
482                shared: Arc::clone(&self.shared),
483                unsent: None,
484                joined_task: Some(task),
485                joined: true,
486            });
487        }
488        let completion = Arc::new(CacheIoCompletion::default());
489        completions.insert(key.clone(), Arc::clone(&completion));
490        drop(completions);
491        drop(execution);
492        let request = CacheIoWorkerRequest::Operation {
493            key: key.clone(),
494            task: Box::new(task),
495            completion: Arc::clone(&completion),
496        };
497        Ok(CacheIoSubmission {
498            ticket: CacheIoTicket {
499                key,
500                completion,
501                shared: Arc::clone(&self.shared),
502            },
503            sender: self.sender.clone(),
504            shared: Arc::clone(&self.shared),
505            unsent: Some(request),
506            joined_task: None,
507            joined: false,
508        })
509    }
510
511    /// Releases exact-key ownership after task resources are safe to drop.
512    pub fn retire(&self, ticket: &CacheIoTicket<Output>) {
513        retire_completion(&self.shared, &ticket.key, &ticket.completion);
514    }
515}
516
517impl<Task, Output> Drop for CacheIoWorker<Task, Output> {
518    fn drop(&mut self) {
519        if self.nonblocking_drop {
520            self.shared.stopping.store(true, Ordering::Release);
521            self.shared.space_available.notify_all();
522            // Do not put Stop ahead of a racing prepared sender. Disconnect
523            // after every remaining prepared handle has rejected admission;
524            // the receiver owns and resolves any already-enqueued message.
525        } else {
526            let _ = self.sender.send(CacheIoWorkerRequest::Stop);
527        }
528        if let Ok(handle) = self.handle.get_mut() {
529            if let Some(handle) = handle.take() {
530                if !self.nonblocking_drop {
531                    let _ = handle.join();
532                }
533            }
534        }
535    }
536}
537
538fn retire_completion<Output>(
539    shared: &CacheIoWorkerShared<Output>,
540    key: &CacheIoOperationKey,
541    completion: &Arc<CacheIoCompletion<Output>>,
542) {
543    let retired = if let Ok(mut execution) = shared.execution.lock() {
544        execution.retire(key).unwrap_or(false)
545    } else {
546        false
547    };
548    if retired {
549        shared.space_available.notify_all();
550        if let Ok(mut in_flight) = shared.in_flight.lock() {
551            if in_flight
552                .get(key)
553                .is_some_and(|current| Arc::ptr_eq(current, completion))
554            {
555                in_flight.remove(key);
556            }
557        }
558    }
559}
560
561/// Failure in generic cache I/O worker coordination or task execution.
562#[derive(Debug, thiserror::Error)]
563pub enum CacheIoWorkerError {
564    /// The worker's synchronization state was poisoned.
565    #[error("cache I/O worker synchronization state is poisoned")]
566    Poisoned,
567    /// A task returned a backend-specific failure string.
568    #[error("cache I/O operation failed: {0}")]
569    OperationFailed(String),
570    /// Cancellation won for this generation.
571    #[error("cache I/O operation was cancelled for generation {generation}")]
572    Cancelled {
573        /// Cancelled model/cache generation.
574        generation: u64,
575    },
576    /// The physical worker thread could not be started.
577    #[error("failed to start cache I/O worker {thread_name}: {source}")]
578    Spawn {
579        /// Requested worker thread name.
580        thread_name: String,
581        /// Underlying thread creation failure.
582        #[source]
583        source: std::io::Error,
584    },
585    /// The exact admission/cancellation state transition was invalid.
586    #[error(transparent)]
587    Execution(#[from] CacheIoExecutionStateError),
588}
589
590#[cfg(test)]
591mod tests {
592    use super::*;
593    use crate::cache::CacheIoOperationKind;
594    use eredu_core::cache::{CacheBlockId, CacheRepresentation};
595    use std::{sync::mpsc, time::Duration};
596
597    enum Task {
598        Value(u64),
599        Pause(mpsc::Sender<()>, mpsc::Receiver<()>),
600        Panic,
601    }
602
603    fn execute(task: Task) -> Result<u64, String> {
604        match task {
605            Task::Value(value) => Ok(value),
606            Task::Pause(started, release) => {
607                let _ = started.send(());
608                let _ = release.recv();
609                Ok(0)
610            }
611            Task::Panic => panic!("injected worker panic"),
612        }
613    }
614
615    fn discard(_value: u64) {}
616
617    fn key(block: i64) -> CacheIoOperationKey {
618        CacheIoOperationKey {
619            generation: 7,
620            id: CacheBlockId {
621                session_id: 1,
622                global_layer: 0,
623                representation: CacheRepresentation::KeyValue,
624                start: block,
625                end: block + 1,
626                rank: None,
627            },
628            kind: CacheIoOperationKind::Read,
629        }
630    }
631
632    #[test]
633    fn worker_coalesces_and_contains_task_panics() {
634        let worker = CacheIoWorker::new(1, "cache-worker-test", execute, discard).unwrap();
635        let first = worker.prepare(key(0), Task::Value(9)).unwrap();
636        let first_ticket = first.ticket.clone();
637        let joined = worker.prepare(key(0), Task::Value(10)).unwrap();
638        let joined_ticket = joined.ticket.clone();
639        assert!(joined.joined);
640        first.enqueue().unwrap();
641        joined.enqueue().unwrap();
642        assert_eq!(first_ticket.wait().unwrap(), 9);
643        assert_eq!(joined_ticket.wait().unwrap(), 9);
644        assert!(first_ticket.shares_completion_with(&joined_ticket));
645        worker.retire(&first_ticket);
646
647        let panicking = worker.prepare(key(1), Task::Panic).unwrap();
648        let ticket = panicking.ticket.clone();
649        panicking.enqueue().unwrap();
650        assert!(matches!(
651            ticket.wait(),
652            Err(CacheIoWorkerError::OperationFailed(message))
653                if message.contains("operation panicked")
654        ));
655        worker.retire(&ticket);
656    }
657
658    #[test]
659    fn cancellation_wakes_a_backpressured_submission() {
660        let worker =
661            Arc::new(CacheIoWorker::new(1, "cache-worker-cancel-test", execute, discard).unwrap());
662        let (started_tx, started_rx) = mpsc::channel();
663        let (release_tx, release_rx) = mpsc::channel();
664        let blocker = worker
665            .prepare(key(0), Task::Pause(started_tx, release_rx))
666            .unwrap();
667        let blocker_ticket = blocker.ticket.clone();
668        blocker.enqueue().unwrap();
669        started_rx.recv().unwrap();
670
671        let queued = worker.prepare(key(1), Task::Value(1)).unwrap();
672        queued.enqueue().unwrap();
673        let blocked = worker.prepare(key(2), Task::Value(2)).unwrap();
674        let blocked_ticket = blocked.ticket.clone();
675        let (outcome_tx, outcome_rx) = mpsc::channel();
676        let enqueue = std::thread::spawn(move || outcome_tx.send(blocked.enqueue()).unwrap());
677        assert!(outcome_rx.recv_timeout(Duration::from_millis(20)).is_err());
678        assert!(blocked_ticket.cancel());
679        assert!(
680            outcome_rx
681                .recv_timeout(Duration::from_secs(1))
682                .unwrap()
683                .unwrap()
684                .backpressure
685        );
686        enqueue.join().unwrap();
687        assert!(matches!(
688            blocked_ticket.wait(),
689            Err(CacheIoWorkerError::Cancelled { generation: 7 })
690        ));
691        release_tx.send(()).unwrap();
692        assert_eq!(blocker_ticket.wait().unwrap(), 0);
693    }
694
695    #[test]
696    fn nonblocking_drop_retains_active_task_and_retires_queued_work() {
697        let worker = CacheIoWorker::new(1, "cache-worker-detach", execute, discard)
698            .unwrap()
699            .with_nonblocking_drop();
700        let (started_tx, started_rx) = mpsc::channel();
701        let (release_tx, release_rx) = mpsc::channel();
702        let active = worker
703            .prepare(key(0), Task::Pause(started_tx, release_rx))
704            .unwrap();
705        let active_ticket = active.ticket.clone();
706        active.enqueue().unwrap();
707        started_rx.recv_timeout(Duration::from_secs(1)).unwrap();
708        let queued = worker.prepare(key(1), Task::Value(1)).unwrap();
709        let queued_ticket = queued.ticket.clone();
710        queued.enqueue().unwrap();
711        let prepared = worker.prepare(key(2), Task::Value(2)).unwrap();
712        let prepared_ticket = prepared.ticket.clone();
713
714        let (dropped_tx, dropped_rx) = mpsc::channel();
715        thread::spawn(move || {
716            drop(worker);
717            let _ = dropped_tx.send(());
718        });
719        let dropped_before_release = dropped_rx.recv_timeout(Duration::from_secs(1));
720        let active_retained = !*active_ticket.completion.released.lock().unwrap();
721        let rejected_prepared = prepared.enqueue();
722        let prepared_released = *prepared_ticket.completion.released.lock().unwrap();
723        release_tx.send(()).unwrap();
724        dropped_before_release.unwrap();
725        assert!(active_retained);
726        assert!(matches!(
727            rejected_prepared,
728            Err(CacheIoWorkerError::OperationFailed(_))
729        ));
730        assert!(prepared_released);
731        assert!(matches!(
732            prepared_ticket.wait(),
733            Err(CacheIoWorkerError::OperationFailed(_))
734        ));
735        assert_eq!(active_ticket.wait().unwrap(), 0);
736        assert!(matches!(
737            queued_ticket.wait(),
738            Err(CacheIoWorkerError::OperationFailed(_))
739        ));
740        active_ticket.wait_for_task_resources().unwrap();
741        queued_ticket.wait_for_task_resources().unwrap();
742        assert!(queued_ticket.shared.in_flight.lock().unwrap().is_empty());
743    }
744
745    #[test]
746    fn nonblocking_shutdown_wakes_backpressured_submission_before_active_task_finishes() {
747        let worker = CacheIoWorker::new(1, "cache-worker-detach-backpressure", execute, discard)
748            .unwrap()
749            .with_nonblocking_drop();
750        let (started_tx, started_rx) = mpsc::channel();
751        let (release_tx, release_rx) = mpsc::channel();
752        let active = worker
753            .prepare(key(0), Task::Pause(started_tx, release_rx))
754            .unwrap();
755        active.enqueue().unwrap();
756        started_rx.recv_timeout(Duration::from_secs(1)).unwrap();
757        worker
758            .prepare(key(1), Task::Value(1))
759            .unwrap()
760            .enqueue()
761            .unwrap();
762        let blocked = worker.prepare(key(2), Task::Value(2)).unwrap();
763        let (outcome_tx, outcome_rx) = mpsc::channel();
764        thread::spawn(move || {
765            let _ = outcome_tx.send(blocked.enqueue());
766        });
767        assert!(outcome_rx.recv_timeout(Duration::from_millis(20)).is_err());
768        drop(worker);
769        let outcome_before_release = outcome_rx.recv_timeout(Duration::from_secs(1));
770        release_tx.send(()).unwrap();
771        assert!(matches!(
772            outcome_before_release.unwrap(),
773            Err(CacheIoWorkerError::OperationFailed(_))
774        ));
775    }
776}
eredu_runtime/cache/worker.rs

eredu_runtime/cache/
worker.rs
