How Pixelle-Video Handles Task Cleanup: A Deep Dive into the Async TaskManager

Pixelle-Video automatically cleans up finished tasks through a periodic _cleanup_loop coroutine that removes completed, failed, or cancelled tasks after a configurable retention window, ensuring no resource leaks when the server shuts down.

The open-source Pixelle-Video project (AIDC-AI/Pixelle-Video) provides a robust background task system for AI video generation. Understanding how it handles task cleanup is critical for production deployments where memory management and resource constraints matter. This article examines the complete cleanup architecture from configuration to shutdown.

Core Cleanup Architecture

Pixelle-Video's cleanup system centers on three interlocking components defined in api/tasks/manager.py:

Component	Purpose	Key Method
Task lifecycle management	Create, track, and execute async tasks	create_task(), execute_task()
Periodic cleanup loop	Continuously purge stale tasks	_cleanup_loop()
Old task removal	Apply retention policy to finished tasks	_cleanup_old_tasks()

The TaskManager class orchestrates these operations through a singleton pattern exposed as task_manager throughout the application.

Starting and Stopping the Cleanup Loop

The cleanup lifecycle begins when the FastAPI server starts. In api/app.py, the lifespan context manager initializes the system:

# api/app.py (lines 68-77)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize the task manager

    await task_manager.start()
    yield
    # Shutdown: cancel all pending tasks and cleanup

    await task_manager.stop()

The start() method in api/tasks/manager.py (lines 46-71) creates the persistent cleanup task:

# api/tasks/manager.py – start() implementation

def start(self):
    """Start the background cleanup loop."""
    self._cleanup_task = asyncio.create_task(
        self._cleanup_loop(),
        name="task_cleanup_loop"
    )

When the server shuts down, stop() (lines 66-71) ensures graceful termination:

• Cancels the _cleanup_loop task
• Cancels all pending Future objects via future.cancel()
• Clears the internal _tasks and _task_futures dictionaries

This guarantees no dangling references or leaked asyncio tasks remain.

The Cleanup Loop: _cleanup_loop

The heart of the system is _cleanup_loop (lines 232-242), an infinite coroutine that drives periodic cleanup:

# api/tasks/manager.py – _cleanup_loop (lines 232-242)

async def _cleanup_loop(self):
    """Background task that periodically cleans up old tasks."""
    while True:
        try:
            await asyncio.sleep(self._cleanup_interval)
            await self._cleanup_old_tasks()
        except asyncio.CancelledError:
            # Expected on shutdown

            break
        except Exception as e:
            logger.error(f"Error in cleanup loop: {e}")

Key characteristics:

• Configurable interval: controlled by task_cleanup_interval (default 3600 seconds / 1 hour)
• Exception resilience: logs errors but continues running; only CancelledError breaks the loop
• Non-blocking: uses asyncio.sleep to yield control between runs

Applying Retention Policy: _cleanup_old_tasks

The actual removal logic resides in _cleanup_old_tasks (lines 445-461):

# api/tasks/manager.py – _cleanup_old_tasks (lines 445-461)

async def _cleanup_old_tasks(self):
    """Remove tasks that have exceeded the retention time."""
    cutoff = datetime.utcnow() - timedelta(seconds=self._retention_time)
    
    expired_tasks = [
        task_id for task_id, task in self._tasks.items()
        if task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED)
        and task.completed_at and task.completed_at < cutoff
    ]
    
    for task_id in expired_tasks:
        del self._tasks[task_id]
        if task_id in self._task_futures:
            self._task_futures[task_id].cancel()
            del self._task_futures[task_id]
    
    if expired_tasks:
        logger.info(f"Cleaned up {len(expired_tasks)} expired tasks")

This method implements a three-state cleanup eligibility:

Status	Eligible for Cleanup?
COMPLETED	✅ Yes
FAILED	✅ Yes
CANCELLED	✅ Yes
PENDING or RUNNING	❌ No

The retention calculation uses completed_at timestamp compared against task_retention_time (default 86400 seconds / 24 hours).

Configuration Settings

Cleanup behavior is controlled in api/config.py (lines 33-37):

# api/config.py – cleanup configuration (lines 33-37)

class APIConfig(BaseSettings):
    # How often to run the cleanup loop (seconds)

    task_cleanup_interval: int = 3600  # 1 hour

    
    # How long to keep finished task results (seconds)

    task_retention_time: int = 86400  # 24 hours

These environment-configurable values allow deployment-specific tuning:

• High-volume services: Reduce task_cleanup_interval to minutes
• Compliance requirements: Adjust task_retention_time for audit trails
• Memory-constrained environments: Lower retention to free RAM faster

Complete Cleanup Flow Example

Here's how task cleanup operates from creation to removal:

from api.tasks import task_manager, TaskType
import asyncio

async def demonstrate_cleanup():
    # 1. Create a video generation task

    task = task_manager.create_task(
        task_type=TaskType.VIDEO_GENERATION,
        request_params={"prompt": "A robot dancing in the rain"}
    )
    print(f"Created task: {task.task_id}")
    
    # 2. Execute the work (simulated here)

    await task_manager.execute_task(
        task_id=task.task_id,
        coro_func=asyncio.sleep,  # stand-in for actual video generation

        delay=0.1
    )
    
    # 3. Task is now COMPLETED with completed_at timestamp

    
    # 4. Cleanup loop (running every hour) will eventually call:

    #    _cleanup_old_tasks() → checks if completed_at < now - 24h

    #    → deletes task if expired

    
    # 5. On server shutdown, lifespan calls task_manager.stop():

    #    → cancels cleanup loop

    #    → cancels all futures

    #    → clears all dictionaries

# Run with: asyncio.run(demonstrate_cleanup())

Summary

• Start/stop lifecycle: task_manager.start() launches _cleanup_loop; task_manager.stop() cancels it and clears all state
• Periodic execution: _cleanup_loop sleeps for task_cleanup_interval (default 1 hour) between cleanup runs
• Retention policy: _cleanup_old_tasks removes tasks with COMPLETED, FAILED, or CANCELLED status when completed_at exceeds task_retention_time (default 24 hours)
• Graceful shutdown: All pending futures are cancelled and dictionaries cleared to prevent memory leaks
• Configuration: Cleanup timing is controlled via APIConfig in api/config.py

Frequently Asked Questions

How often does Pixelle-Video clean up finished tasks?

By default, the cleanup loop runs every 3600 seconds (1 hour). This is controlled by the task_cleanup_interval setting in api/config.py. You can reduce this to minutes for high-traffic deployments or increase it for lower resource usage.

What determines when a task becomes eligible for cleanup?

A task must meet two conditions: its status must be COMPLETED, FAILED, or CANCELLED (not PENDING or RUNNING), and its completed_at timestamp must be older than the retention window. The default task_retention_time is 86400 seconds (24 hours).

What happens to running tasks when the server shuts down?

When the FastAPI server shuts down, the lifespan context manager in api/app.py calls task_manager.stop(). This cancels the cleanup loop, cancels all pending Future objects via future.cancel(), and clears the internal _tasks and _task_futures dictionaries to ensure no resource leaks.

Can I customize the retention policy for different task types?

The current implementation in api/tasks/manager.py uses global task_cleanup_interval and task_retention_time values from APIConfig. There is no per-task-type retention in the base implementation. To implement custom retention, you would need to modify _cleanup_old_tasks to check task.task_type against a type-specific retention mapping.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}