Fixing Celery Tasks with Asyncio Event Loop Issues

The Problem

When using asyncio with Celery tasks, you might encounter this error:

RuntimeError: Event loop is closed

This typically occurs when:

1. Running multiple async operations in a Celery task
2. Using run_async helper functions
3. Creating/closing event loops multiple times within the same task

The error happens because each async operation tries to create and close its own event loop, but subsequent operations find the loop closed.

Quick Solution

Copy this pattern for your Celery tasks:

@contextlib.contextmanager
def task_event_loop():
    """Context manager for task-level event loop"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        yield loop
    finally:
        try:
            # Cancel pending tasks
            pending = asyncio.all_tasks(loop)
            if pending:
                loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
            if not loop.is_closed():
                loop.close()
        except Exception as e:
            logger.warning(f"Error during event loop cleanup: {str(e)}")

@celery_app.task
def my_celery_task():
    with task_event_loop() as loop:
        # Run all async operations with the same loop
        result1 = loop.run_until_complete(async_operation1())
        result2 = loop.run_until_complete(async_operation2())
        result3 = loop.run_until_complete(async_operation3())

Understanding the Issue

What Changed?

• Celery tasks are synchronous by nature
• Asyncio operations require an event loop
• Each async operation needs a running event loop
• Closing loops prematurely breaks subsequent operations

Why It Happens

• Default approach creates/closes loops per operation
• Subsequent operations find closed loops
• Resource cleanup conflicts with operation needs
• No task-level loop management

The Solution

1. Create a Task-Level Event Loop Manager

@contextlib.contextmanager
def task_event_loop():
    """Context manager for task-level event loop"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        yield loop
    finally:
        try:
            # Clean up pending tasks
            pending = asyncio.all_tasks(loop)
            if pending:
                loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
            if not loop.is_closed():
                loop.close()
        except Exception as e:
            logger.warning(f"Error during event loop cleanup: {str(e)}")

2. Update Your Celery Task

@celery_app.task
def process_async_task(task_id: str, data: Dict[str, Any]):
    try:
        with task_event_loop() as loop:
            # Update status
            loop.run_until_complete(
                update_status_async(task_id, "PROCESSING")
            )
            
            # Process data
            result = loop.run_until_complete(
                process_data_async(data)
            )
            
            # Update results
            loop.run_until_complete(
                update_results_async(task_id, result)
            )
            
        return {"status": "success", "result": result}
        
    except Exception as e:
        # Handle errors
        with task_event_loop() as loop:
            loop.run_until_complete(
                update_error_async(task_id, str(e))
            )
        raise

3. Handle Synchronous Operations

@celery_app.task
def mixed_operations_task(task_id: str, data: Dict[str, Any]):
    # Async operations within event loop
    with task_event_loop() as loop:
        result = loop.run_until_complete(async_operations())
    
    # Synchronous operations outside loop
    process_sync_data(result)
    call_webhook(task_id)

Common Mistakes to Avoid

1. Creating Multiple Loops

# ❌ Wrong
result1 = run_async(operation1())  # Creates/closes loop
result2 = run_async(operation2())  # Creates/closes loop

# ✅ Correct
with task_event_loop() as loop:
    result1 = loop.run_until_complete(operation1())
    result2 = loop.run_until_complete(operation2())

2. Mixing Loop Management

# ❌ Wrong
loop = asyncio.new_event_loop()
result = run_async(operation())  # Creates another loop

# ✅ Correct
with task_event_loop() as loop:
    result = loop.run_until_complete(operation())

3. Not Handling Cleanup

# ❌ Wrong
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(operation())
# Loop never closed!

# ✅ Correct
with task_event_loop() as loop:
    result = loop.run_until_complete(operation())
# Loop automatically cleaned up

When to Use This Pattern

Apply this solution when:

• Your Celery tasks contain multiple async operations
• You're getting "Event loop is closed" errors
• You need to manage async resources properly
• You're mixing sync and async operations

Verifying the Fix

After applying the fix:

1. Event loop errors should be gone
2. Tasks should complete successfully
3. Resources should be properly cleaned up
4. Error handling should work as expected

Best Practices

1. Use Context Managers

   • Always use task_event_loop() context manager
   • Let Python handle loop cleanup
   • Ensure proper resource management

2. Keep Loops Task-Scoped

   • One loop per task execution
   • Don't share loops between tasks
   • Don't store loops in global variables

3. Handle Errors Properly

   • Use try/except blocks
   • Clean up resources in error cases
   • Log errors with proper context

4. Separate Concerns

   • Keep async operations together
   • Handle sync operations outside the loop
   • Use proper error boundaries

🔑 Key Takeaway: This pattern is specific to Celery tasks using asyncio. For other async frameworks or pure asyncio applications, different patterns might be more appropriate.