Celery Asynchronous Task Processing

This page provides an overview of how Celery is used in The Wall API for asynchronous task processing. It details Celery configuration, Redis integration as a message broker, worker setup, task orchestration, and queue management.

Table of Contents

1. Overview
2. Celery Configuration
   2.1 Core Celery Configuration
   2.2 Celery Application Setup
3. Celery System Architecture
   3.1 Celery Workers and Beat Scheduler
     3.1.1 Celery Worker 1
     3.1.2 Celery Worker 2
     3.1.3 Celery Beat - Periodic Task Scheduler
     3.1.4 Embedded Celery Test Worker
   3.2 Celery Queues
4. Celery Tasks
   4.1 Concurrent Tasks
     4.1.1 Task Priority Configuration
     4.1.2 List of Concurrent Tasks
     4.1.3 Abort Signals
   4.2 Sequential Tasks
   4.3 Test Task Wrappers

1. Overview

In The Wall API, Celery is configured to work with Redis as a message broker. It runs multiple task workers, each optimized for handling different types of workloads:

• Celery Worker 1 manages concurrent tasks, enabling parallel execution for high-throughput workloads.
• Celery Worker 2 handles sequential tasks, ensuring ordered execution.
• Celery Beat schedules periodic tasks, such as cleanup and archival jobs.

2. Celery Configuration

2.1 Core Celery Configuration

• Celery is configured in:

  settings.py

  
  

  CELERY_WORKER_PREFETCH_MULTIPLIER = 1
  CELERY_TASK_PRIORITY = {
      'HIGH': 0,     # Highest
      'MEDIUM': 5,
      'LOW': 9
  }

  env_utils.py

  
  

  def configure_connections_settings(DATABASES=None) -> dict:
    STARTED_FROM_CELERY_SERVICE = os.getenv('STARTED_FROM_CELERY_SERVICE', 'False') == 'True'
  
    if PROJECT_MODE == 'dev' and STARTED_FROM_CELERY_SERVICE and DATABASES:
        DATABASES['default']['HOST'] = 'postgres'
    
    CELERY_BROKER_URL = REDIS_URL.replace('REDIS_DB_NUMBER', REDIS_DB_NUMBER_CELERY)
  
    if PROJECT_MODE == 'dev' and STARTED_FROM_CELERY_SERVICE:
  
        CELERY_BROKER_URL = CELERY_BROKER_URL.replace('localhost', 'redis')
    CELERY_RESULT_BACKEND = CELERY_BROKER_URL
  
    result['CELERY_BROKER_URL'] = CELERY_BROKER_URL
    result['CELERY_RESULT_BACKEND'] = CELERY_RESULT_BACKEND
    result['CELERY_RESULT_EXPIRES'] = '600'

  • ✔ configure_connections_settings() isolates connection setup and supports the lightweight Celery Beat mode, allowing Beat to start without full Django initialization. It is reused in the postgres and redis services' connection setup.
  
  

• The values of these settings are loaded from .env files stored in the config/envs/ folder.

  Variable	Description
  CELERY_BROKER_URL	Message broker main URL
  CELERY_RESULT_BACKEND	Message broker URL for Celery task results storage
  CELERY_RESULT_EXPIRES	Duration (in seconds) before task results expire
  CELERY_TASK_PRIORITY	Task priority level
  CELERY_WORKER_PREFETCH_MULTIPLIER	Number of prefetched tasks before workers processing starts
  LIGHT_CELERY_CONFIG	Controls the lightweight Celery Beat setup
  STARTED_FROM_CELERY_SERVICE	Indicates whether the process was launched via Celery

  💡 config/envs/ folder breakdown → Environment Configuration Files Folder Structure.

  💡 For a detailed description of the Celery environment variables, refer to the Celery Task Processing Configuration section of the Environment Variables Index page.

Return to Top

2.2 Celery Application Setup

• celery.py - Celery Initialization

  • Initializes Celery in:

    • Full mode (Workers)

      • Loads Django settings.
      • Auto-discovers tasks across registered apps.

    • Lightweight mode (Beat)

      • Skips the full load of the Django settings, using LIGHT_CELERY_CONFIG and env_utils.configure_connections_settings() to reduce resource usage.
      • Imports the scheduled tasks manually.

      ✔ The lightweight mode is specifically designed to optimize Celery Beat, reducing memory usage and startup time by avoiding full Django initialization.
      
      

  • Configures the periodic task scheduler (Celery Beat), enabling scheduled executions.

  
  

• apps.py - Celery Tasks Discovery

  • The ready() method in apps.py automatically imports Celery tasks on startup to ensure they are discoverable when Celery starts.

  ✔ Without this import, Celery workers may fail to recognize certain task functions if they are not explicitly imported during initialization.

Return to Top

---

3. Celery System Architecture

The Celery setup in The Wall API consists of two main workers, a periodic task scheduler (Celery Beat), and an embedded Celery test worker.

3.1 Celery Workers and Beat Scheduler

3.1.1 Celery Worker 1

• Optimized for performance, enabling simultaneous task execution while preserving order in designated task flows.
• Ensures priority-based execution, processing higher-priority tasks ahead of others.

✔ The worker runs with an automatically determined concurrency level, allowing Celery to prioritize and balance multiple tasks dynamically.

3.1.2 Celery Worker 2

• Configured with --concurrency=1, enforcing strict sequential task execution.
• Handles lower-priority logging and maintenance tasks.

✔ This worker is specifically designed to prevent parallel execution conflicts, ensuring that tasks do not overlap.

3.1.3 Celery Beat - Periodic Task Scheduler

• Schedules periodic system maintenance tasks, such as:
  • Archiving and cleaning up logs
  • Deleting unused wall configurations

✔ Celery Beat ensures that scheduled maintenance tasks run automatically, reducing manual intervention and keeping the system optimized with minimal overhead.

3.1.4 Embedded Celery Test Worker

• Runs an in-process Celery worker that executes tasks in real conditions.
• Subscribes only to the test_queue, ensuring test tasks remain isolated from production workers.
• Spins up dynamically during test execution and shuts down afterward.
• Used in test_celery_concurrent_tasks.ConcurrentCeleryTasksTestBase, where it is started via the start_worker() API from celery.contrib.testing.worker.

✔ The Embedded Celery Test Worker allows test cases to execute real Celery tasks without requiring a full Celery worker instance, minimizing resource usage while maintaining execution fidelity.
✔ By isolating test-related tasks, it guarantees a reliable testing environment without affecting production operations.

Return to Top

3.2 Celery Queues

The Celery queues enable efficient task distribution across multiple workers. Each queue is designed to handle specific types of tasks, optimizing execution flow and resource allocation.

Queue Name	Purpose	Worker
concurrent_tasks	Manages parallel execution of prioritized tasks.	Worker 1
sequential_tasks	Handles strictly ordered tasks (processed sequentially).	Worker 2 (--concurrency=1)
test_queue	Isolates test-related tasks from normal usage.	Embedded Test Worker

✔ The queue structure ensures efficient execution by separating concurrent, sequential, and test tasks.
This prevents resource contention, maintains execution order where needed, and isolates testing from production workflows.

Return to Top

---

4. Celery Tasks

4.1 Concurrent Tasks

Sent to the concurrent_tasks queue to be processed by worker 1

4.1.1 Task Priority Configuration

• Priority levels are defined by CELERY_TASK_PRIORITY
• Task prefetching is limited (CELERY_WORKER_PREFETCH_MULTIPLIER=1), ensuring that workers fetch only one task at a time, preventing low-priority tasks from delaying high-priority execution.

✔ By using priority levels, the system ensures that critical tasks are executed as soon as possible without delaying lower-priority operations.

Return to Top

4.1.2 List of Concurrent Tasks

Concurrent tasks allow parallel execution, improving performance for compute-heavy operations.

orchestrate_wall_config_processing_task

• Trigger type:
  • Normal App Use.
• Sent from:
  • Directly sent from the application on initial upload involving full-range simulations, or during subsequent on-demand requests.
• Description:
  • Orchestrates the creation of multiple Walls, by sending a task group of multiple create_wall_task.
  • Listens for deletion signals and aborts execution if needed.

✔ A centralized orchestration task ensures proper tracking of task execution, handling failures, and managing retries in a structured way.
✔ State management prevents duplicate processing by locking WallConfig.status = CELERY_CALCULATION before task execution begins.
✔ Decoupling task scheduling from business logic keeps the system maintainable and allows for better scalability.

create_wall_task

• Trigger type:
  • Normal App Use.
• Sent from:
  • Task Group Execution from orchestrate_wall_config_processing_task.
• Description:
  • Processes and stores an individual Wall.

✔ Encapsulates the core logic for processing a wall.

delete_unused_wall_configs_task

• Trigger type:
  • Beat or Normal App Use.
• Sent from:
  • Directly sent from the application when a user deletes the last wall configuration file, related to a WallConfig.
  • Beat Scheduler - DB cleanup task.
• Description:
  • Initiates the deletion of WallConfig instances orphaned by:
    • On-demand deletion of the last wall configuration file associated with a WallConfig (triggered by the application).
    • Cascading deletion of uploaded files when a user deletes their account without explicitly removing their files (scheduled cleanup).
  • Sends a task chain of multiple wall_config_deletion_task.

✔ Triggered by both on-demand deletion requests and scheduled cleanup, ensuring timely removal of orphaned data.
✔ Uses task chaining to ensure controlled and sequential deletion of related objects.

wall_config_deletion_task

• Trigger type:
  • Beat or Normal App Use.
• Sent from:
  • Task Chain Execution from delete_unused_wall_configs_task.
• Description:
  • Deletes a single wall configuration instance.

✔ Encapsulates the core logic for deleting a wall configurations.

Return to Top

4.1.3 Abort Signals

An abort mechanism is implemented to prevent unnecessary computation and ensure clean termination when a deletion request is detected.
The orchestrate_wall_config_processing_task task:

• Monitors for deletion signals (WallConfig.deletion_initiated flag) to determine if an active processing task should be stopped.
• In case of a deletion signal during orchestration task execution, sends an abort signal (.abort()) to running create_wall_task tasks, ensuring that ongoing computations do not proceed unnecessarily.
• Ensures graceful termination, waiting for in-progress tasks to stop cleanly and log meaningful abort messages rather than forcefully interrupting execution.
• Returns structured abort states (OK_0, OK_1, OK_2), indicating the specific stage at which a task was interrupted.

✔ This mechanism prevents wasted computation and keeps the system responsive to real-time changes in processing needs.

✔ Instead of using Celery's basic .revoke() method, which abruptly stops execution, it ensures tasks terminate gracefully, allowing partial results to be handled correctly while maintaining system integrity.

✔ To facilitate the abort signal functionality, celery.contrib.abortable.AbortableTask is used in the task definitions within tasks.py.

💡 For more details on the abort signals implementation, refer to the Handling Abort Signals from Celery section of the Sequential-and-Concurrent-Simulation-Modes page.

Return to Top

4.2 Sequential Tasks

Sent to the sequential_tasks queue to be processed by worker 2

Task Name	Trigger Type	Description
archive_logs_task	Beat Schedule	Archives old log files.
clean_old_archives_task	Beat Schedule	Deletes outdated archived files.
log_error_task	Normal Application Use	Logs system errors for debugging.

✔ Sequential execution prevents race conditions when handling file operations and log management tasks.

💡 For a detailed breakdown of the error logging, refer to the Error Logging section of the Error Handling page.

💡 Envionment variables, for fine-grained control over the file maintenance and logging tasks → Environment Variables Index - Logging Configuration

Return to Top

4.3 Test Task Wrappers

Test task wrappers facilitate sending tasks to the test_queue, ensuring that real Celery execution is tested in an isolated environment without affecting production workflows.

• Wrap actual Celery tasks in tasks.py with @shared_task(queue='test_queue') to redirect execution to the test queue.
• Enable integration testing of Celery tasks, ensuring proper execution without interfering with the main workers.
• Executed within the Embedded Celery Test worker, which runs only during test execution and ensures that test-related Celery tasks remain fully isolated from production processing.
• Ensure behavior validation by mirroring actual task execution conditions while keeping test data separate.

✔ By using dedicated test wrappers, the system verifies Celery execution under real conditions while maintaining strict test isolation.

---

Next:   Sequential and Concurrent Simulation Modes
Back to: Redis - In-Memory Cache & Celery Message Broker

Return to Top
DTS Index
Home Page

---