// Oak Video Editor - Non-Linear Video Editor // Copyright (C) 2026 Oak Team // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU General Public License for more details. // // You should have received a copy of the GNU General Public License // along with this program. If not, see . //! The `TaskManager` singleton, mirroring `src/task/src/taskmanager.h`. //! //! Holds the set of live [`crate::task::Task`] instances; starting a task //! hands its ownership to the manager, which also exposes the codec task //! submitter registration (see [`crate::codecbridge`]). //! //! CPP-PARITY: src/task/src/taskmanager.h //! //! ## Threading model //! //! [`TaskManager::add_task`] spawns a worker thread that runs the task's //! [`Task::start`]. The worker thread only touches the task (through a raw //! pointer) and never the manager, so the manager's own mutex is never held //! across a join in the C ABI paths (joins happen lock-free via //! [`TaskManager::drain_finished`] / [`TaskManager::take_thread_by_ptr`]). use std::sync::Mutex; use crate::error::{Error, Result}; use crate::task::Task; /// Process-wide singleton manager. C++ is a lazy singleton; the Rust side /// keeps a `Mutex>` so init/shutdown is explicit and races are /// rejected with `OAKTASK_E_STATE` (matching `oaktask_manager_init`). pub struct TaskManager { /// Live tasks, most-recently-added last (mirrors the C++ `task_list_`). tasks: Vec>, /// Worker threads, parallel to `tasks` (moved out before joining). threads: Vec>>, /// Whether the codec task submitter is currently registered. codec_submitter_registered: bool, } /// Boxable shim so tasks of heterogeneous concrete types can be stored in /// the manager's list. pub trait TaskBox: Send { /// Downcast/erase operations the manager needs. fn task(&self) -> &Task; } impl TaskBox for Task { fn task(&self) -> &Task { self } } /// The process-wide instance slot. A raw pointer (never freed while the /// slot is set) so [`TaskManager::instance`] can hand out a `&'static`. static INSTANCE: Mutex> = Mutex::new(None); /// `Send`/`Sync` wrapper for the manager pointer (the manager's state is /// guarded by the `INSTANCE` mutex; the pointer itself is plain). struct ManagerPtr(*mut TaskManager); // Safety: all access to the pointee goes through the `INSTANCE` mutex. unsafe impl Send for ManagerPtr {} unsafe impl Sync for ManagerPtr {} impl TaskManager { /// The process-wide instance; `None` until [`TaskManager::init`]. /// /// CPP-PARITY: src/task/src/taskmanager.h (instance) pub fn instance() -> Option<&'static TaskManager> { let guard = INSTANCE.lock().unwrap(); guard.as_ref().map(|p| unsafe { &*p.0 }) } /// Create the singleton and register the codec task submitter. Returns /// `Err(Error::State)` if already initialized (mirrors /// `oaktask_manager_init`). pub fn init() -> Result<()> { let mut guard = INSTANCE.lock().unwrap(); if guard.is_some() { return Err(Error::State); } let mgr = Box::into_raw(Box::new(TaskManager { tasks: Vec::new(), threads: Vec::new(), codec_submitter_registered: false, })); *guard = Some(ManagerPtr(mgr)); Ok(()) } /// Destroy the singleton. Idempotent. pub fn shutdown() { let ptr = { let mut guard = INSTANCE.lock().unwrap(); guard.take() }; if let Some(p) = ptr { // The Drop impl cancels and joins every worker thread. unsafe { drop(Box::from_raw(p.0)); } } } /// Run `f` with the singleton (when initialized), without taking /// ownership. Used by the C ABI layer. pub fn with_manager(f: impl FnOnce(&TaskManager) -> R) -> Option { let guard = INSTANCE.lock().unwrap(); guard.as_ref().map(|p| f(unsafe { &*p.0 })) } /// Run `f` with the singleton mutably (when initialized). Used by the C /// ABI layer. The caller must not call [`TaskManager::with_manager`] /// from a worker callback while this is held (same rule as the C++ /// `mutex_`). pub fn with_manager_mut(f: impl FnOnce(&mut TaskManager) -> R) -> Option { let mut guard = INSTANCE.lock().unwrap(); guard.as_mut().map(|p| f(unsafe { &mut *p.0 })) } /// Add a task, taking ownership (the manager now owns it). A worker /// thread is spawned that runs [`Task::start`] on the task. /// /// The worker runs through a raw pointer while the box lives in /// `self.tasks`; the shared reference produced by `task()` ends before /// the pointer is used, so the cast is safe in this exclusively-owned /// setting. #[allow(invalid_reference_casting)] pub fn add_task(&mut self, task: Box) { let ptr = task.task() as *const Task as usize as *mut Task; self.tasks.push(task); // The worker only touches the task through the raw pointer; it never // reaches the manager, so the box stays alive in `self.tasks`. let ptr_usize = ptr as usize; let handle = std::thread::spawn(move || { unsafe { // The manager tracks success/failure through the task's own // finished state; the Result is intentionally ignored. let _ = (&mut *(ptr_usize as *mut Task)).start(); } }); self.threads.push(Some(handle)); } /// Cancel the task at `index`. #[allow(invalid_reference_casting)] pub fn cancel_task(&mut self, index: usize) -> Result<()> { let task = self.tasks.get(index).ok_or(Error::NotFound)?; let ptr = task.task() as *const Task as usize as *mut Task; unsafe { (&mut *ptr).cancel(); } Ok(()) } /// Cancel the task at `index` and block until it finishes. pub fn cancel_task_and_wait(&mut self, index: usize) -> Result<()> { self.cancel_task(index)?; let handle = self.threads.get_mut(index).and_then(|h| h.take()); if let Some(h) = handle { let _ = h.join(); } Ok(()) } /// Remove finished tasks from the list. pub fn delete_finished(&mut self) { for (_task, handle) in self.drain_finished() { let _ = handle.join(); } } /// Number of live tasks. pub fn get_task_count(&self) -> usize { self.tasks.len() } /// Raw pointer to the task at `index` (stable while the manager owns /// it). Used by `oaktask_manager_at` to build a borrowed task handle. pub fn task_ptr_at(&self, index: usize) -> Result<*mut Task> { let task = self.tasks.get(index).ok_or(Error::NotFound)?; Ok(task.task() as *const Task as *mut Task) } /// Index of the task with the given address. pub fn find_index(&self, ptr: *const Task) -> Option { self.tasks .iter() .position(|t| t.task() as *const Task == ptr) } /// Cancel the task with the given address (no-op when absent). pub fn cancel_task_by_ptr(&mut self, ptr: *const Task) { if let Some(index) = self.find_index(ptr) { let _ = self.cancel_task(index); } } /// Move the worker thread of the task with the given address out of the /// list so the caller can join it without holding the manager lock. pub fn take_thread_by_ptr(&mut self, ptr: *const Task) -> Option> { let index = self.find_index(ptr)?; self.threads.get_mut(index).and_then(|h| h.take()) } /// Remove every finished task (and its worker thread) from the list, /// returning the task boxes and thread handles so the caller can join /// them lock-free (the boxes must stay alive until the threads exit). pub fn drain_finished(&mut self) -> Vec<(Box, std::thread::JoinHandle<()>)> { let mut out = Vec::new(); let mut i = 0; while i < self.tasks.len() { if self.tasks[i].task().is_finished() { let task = self.tasks.remove(i); let handle = self.threads.remove(i); if let Some(handle) = handle { out.push((task, handle)); } } else { i += 1; } } out } /// Whether this manager registered the codec submitter itself. pub fn codec_submitter_registered(&self) -> bool { self.codec_submitter_registered } /// Mark whether this manager registered the codec submitter. pub fn set_codec_submitter_registered(&mut self, registered: bool) { self.codec_submitter_registered = registered; } } impl Drop for TaskManager { fn drop(&mut self) { // Cancel every task, then join every worker (mirrors the C++ // destructor: cancel all first, then join all). #[allow(invalid_reference_casting)] for task in &self.tasks { let ptr = task.task() as *const Task as usize as *mut Task; unsafe { (&mut *ptr).cancel(); } } for handle in self.threads.iter_mut() { if let Some(h) = handle.take() { let _ = h.join(); } } } }