threading.py

threading.py – Background task system with priority, frame‑based scheduling,
and resource monitoring.

Provides:
- TaskPriority: LOW, NORMAL, HIGH
- BackgroundTaskManager: schedule tasks with interval_frames, track CPU/memory,
 cancel, and retrieve status.
- Integration with PerformanceMonitor for per‑task timing.

class TaskPriority

Description

Task priority – lower integer = higher priority.

Attributes
LOW: Any = 0
NORMAL: Any = 1
HIGH: Any = 2
Methods

No methods defined.

class TaskInfo

Description

Metadata for a scheduled task.

Attributes
id: str = None
name: str = None
priority: TaskPriority = None
interval_frames: int = 0
last_run_frame: int = 0
total_runs: int = 0
cpu_time_ms: float = 0.0
memory_peak_bytes: int = 0
active: bool = True
result: Any = None
error: Optional[Exception] = None
_fn: Optional[Callable] = None
_args: tuple = ()
_kwargs: dict = field(default_factory=dict)
Methods

No methods defined.

class BackgroundTaskManager

Description

Manages background tasks with priority, frame‑based scheduling, and resource tracking.
Tasks are executed in a thread pool, but scheduling decisions (when to run) are made
on the main thread via update().

Methods
def __init__(self: Any, max_workers: int = 4, monitor: Optional['PerformanceMonitor'] = None) -> Any
Args:
   max_workers: Number of worker threads in the pool.
   monitor: Optional PerformanceMonitor instance for timing.
def schedule(self: Any, fn: Callable, name: str = '', priority: TaskPriority = TaskPriority.NORMAL, interval_frames: int = 0, args: tuple = (), kwargs: dict = None) -> str
Schedule a task. If interval_frames > 0, it will repeat every that many frames.
Returns a task ID that can be used to cancel or check status.
def cancel(self: Any, task_id: str) -> bool
Cancel a scheduled task (even if already running).
def update(self: Any, dt: float) -> Any
Call this every frame on the main thread.
- Increments frame counter.
- Enqueues recurring tasks that are due.
- Processes completed results (updates TaskInfo with result/error).
def get_status(self: Any) -> Dict[str, Dict[str, Any]]
Return a summary of all tasks for debugging.
def shutdown(self: Any, wait: bool = True) -> Any
Shut down the thread pool and consumer thread.
def _enqueue_task(self: Any, task_id: str, info: TaskInfo) -> Any
Put a task into the priority queue for the consumer.
def _consumer_loop(self: Any) -> Any
Runs in a background thread: fetches tasks from priority queue and submits to executor.
def _on_task_done(self: Any, task_id: str, future: concurrent.futures.Future) -> Any
Callback when a future completes.
def __del__(self: Any) -> Any
No documentation
Back to Utils Module