threading.py

Threading System - Concurrent Task Execution and Management

LOCATION: lunaengine/utils/threading.py

DESCRIPTION:
Provides a thread pool system for concurrent task execution, allowing
background processing without blocking the main game loop. Supports
task submission, execution, and result management.

KEY COMPONENTS:
- Task: Individual executable task with parameters and results
- ThreadPool: Managed pool of worker threads for concurrent execution
- Queue-based task distribution for efficient load balancing
- Exception handling and task result management

LIBRARIES USED:
- threading: Core threading functionality and thread management
- queue: Thread-safe task queue implementation
- typing: Type hints for callable functions and task parameters

USAGE:
>>> pool = ThreadPool(4)
>>> pool.start()
>>> task = Task(expensive_calculation, arg1, arg2)
>>> pool.submit(task)
>>> # Later: check task.result or task.exception
>>> pool.stop()

class Task

Description

No documentation

Methods
def __init__(self: Any, func: Callable, *args: tuple, **kwargs: dict) -> Any
No documentation
def execute(self: Any) -> Any
Execute the task

class ThreadPool

Description

No documentation

Methods
def __init__(self: Any, num_threads: int = 4) -> Any
No documentation
def start(self: Any) -> Any
Start the thread pool
def stop(self: Any) -> Any
Stop the thread pool
def submit(self: Any, task: Task) -> Any
Submit a task to the thread pool
def _worker(self: Any) -> Any
Worker thread function
Back to Utils Module