Celery Task Coordination: Patterns for Multi-Worker Task Synchronization

In this tutorial, you will learn about Celery Task Coordination: Patterns for Multi. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery task coordination enables multiple workers to synchronize on shared state using distributed primitives like barriers, semaphores, counters, and locks, ensuring correct ordering and resource access in parallel task execution.

flowchart LR
    T1[Task A] -->|wait| B[Barrier N=3]
    T2[Task B] -->|wait| B
    T3[Task C] -->|wait| B
    B -->|all ready| Continue[Continue]
    T4[Task D] -->|acquire| L[Distributed Lock]
    T5[Task E] -->|acquire| L
    L -->|one at a time| CS[Critical Section]

What You'll Learn

• Distributed barriers with Redis
• Semaphores for resource limits
• Atomic counters for progress tracking
• Distributed locks with Redis and database
• Coordination failure handling

Why It Matters

Without coordination, parallel tasks race on shared resources, corrupt data, duplicate work, and exceed rate limits. Distributed coordination primitives prevent these issues while keeping tasks decoupled and scalable.

Real-World Use

DodaTech's batch file scanner uses a Redis barrier to coordinate 10 parallel scan tasks. When all 10 complete, the coordinator task generates a combined report. A Semaphore limits concurrent database writes to 5, preventing Connection Pool exhaustion.

Distributed Barrier

import redis
from celery import Celery

app = Celery('coordination', broker='redis://localhost:6379/0')
cache = redis.Redis(host='localhost', port=6379, db=1)

BARRIER_KEY = 'barrier:scan_complete'
BARRIER_COUNT = 3

@app.task
def scan_chunk(chunk_id, total_chunks):
    import time
    time.sleep(0.1)
    print(f"Scanned chunk {chunk_id}/{total_chunks}")

    remaining = cache.decr(f'{BARRIER_KEY}:remaining')
    if remaining <= 0:
        print(f"All {total_chunks} chunks scanned, proceeding")
        generate_report.delay()
        return True
    print(f"Waiting for {remaining} more chunks")
    return False

@app.task
def generate_report():
    print("Generating combined scan report")
    return "Report generated"

cache.set(f'{BARRIER_KEY}:remaining', BARRIER_COUNT)
for i in range(BARRIER_COUNT):
    scan_chunk.delay(i + 1, BARRIER_COUNT)
print(f"Submitted {BARRIER_COUNT} scan tasks")

Expected output:

Submitted 3 scan tasks
Scanned chunk 1/3
Waiting for 2 more chunks
Scanned chunk 2/3
Waiting for 1 more chunks
Scanned chunk 3/3
All 3 chunks scanned, proceeding
Generating combined scan report

Distributed Semaphore

import redis
from celery import Celery
import time

app = Celery('coordination', broker='redis://localhost:6379/0')
cache = redis.Redis(host='localhost', port=6379, db=1)

SEMAPHORE_KEY = 'semaphore:db_writes'
MAX_CONCURRENT = 3

@app.task
def write_to_database(data_id):
    acquired = cache.lpush(SEMAPHORE_KEY, data_id) <= MAX_CONCURRENT
    if not acquired:
        cache.lrem(SEMAPHORE_KEY, 1, data_id)
        print(f"Data {data_id}: waiting (too many concurrent writes)")
        write_to_database.retry(countdown=2, max_retries=5)
        return

    try:
        time.sleep(0.3)
        print(f"Data {data_id}: written to database")
    finally:
        cache.lrem(SEMAPHORE_KEY, 1, data_id)

cache.delete(SEMAPHORE_KEY)
for i in range(10):
    write_to_database.delay(i)
print(f"Submitted 10 write tasks (max {MAX_CONCURRENT} concurrent)")

Expected output:

Submitted 10 write tasks (max 3 concurrent)
Data 0: written to database
Data 1: written to database
Data 2: written to database
Data 3: waiting (too many concurrent writes)
Data 0: written to database
Data 3: written to database
...

Distributed Counter

import redis
from celery import Celery

app = Celery('coordination', broker='redis://localhost:6379/0')
cache = redis.Redis(host='localhost', port=6379, db=1)

COUNTER_KEY = 'counter:processed_items'
TOTAL_EXPECTED = 100

@app.task
def process_item(item_id):
    import time
    time.sleep(0.05)
    count = cache.incr(COUNTER_KEY)
    print(f"Processed item {item_id} ({count}/{TOTAL_EXPECTED})")

    if count >= TOTAL_EXPECTED:
        print(f"All {TOTAL_EXPECTED} items processed!")
        cache.delete(COUNTER_KEY)

cache.set(COUNTER_KEY, 0)
for i in range(TOTAL_EXPECTED):
    process_item.delay(i)
print(f"Submitted {TOTAL_EXPECTED} items for processing")

Expected output:

Submitted 100 items for processing
Processed item 5 (1/100)
Processed item 12 (2/100)
...
Processed item 42 (100/100)
All 100 items processed!

Common Mistakes

• Race conditions in barrier checks -- checking remaining == 0 and then acting can cause double execution if two tasks reach zero simultaneously. Use a Redis SETNX-based flag to ensure only one task acts.
• Semaphore without release -- if a task crashes after acquiring the semaphore, it leaks. Use a TTL on the semaphore key or a try/finally block to ensure release.
• Counter overflow -- atomic counters work but don't expire. Set a TTL or reset counter at the start of each batch to prevent stale counts from previous runs.
• Using local locks for distributed coordination -- threading.Lock only works within one Process. Use Redis locks or database advisory locks for multi-worker coordination.
• Forgetting Redis connection error handling -- network partitions cause coordination failures. Implement retry logic and fallback behaviors for all coordination primitives.

Practice Questions

1. How does a distributed barrier prevent race conditions in multi-worker synchronization?
2. What happens to a semaphore if a task crashes without releasing it?
3. Why is a Redis counter safer than a database counter for Celery coordination?
4. How do you ensure only one task acts when a barrier reaches zero?
5. What coordination primitive would you use to limit database connection usage?

Challenge

Build a multi-phase ETL pipeline with coordination: (1) Phase 1: 10 parallel extract tasks, (2) Barrier: wait for all extracts, (3) Phase 2: 5 parallel transform tasks, each processing a chunk, (4) Semaphore: limit concurrent database writes to 3, (5) Phase 3: single load task after all transforms complete. Use Redis for all coordination primitives.

FAQ

Mini Project

Build a distributed task coordinator library that wraps Redis primitives: (1) Barrier class with configurable count and automatic cleanup, (2) Semaphore class with acquire/release and TTL-based leak detection, (3) Counter class with batch tracking and completion callbacks, (4) Lock class with blocking and non-blocking acquire. Include comprehensive error handling for Redis failures.

What's Next

Continue with Late Acknowledgement to learn about reliable task execution. Then explore Task Expiration for time-bound task handling.