How to Use The Deadlock Simulation Issue in Django Using Multiprocessing

As a Django developer, I recently tried to simulate a database deadlock for testing purposes. My goal was simple: create two processes that lock database rows in conflicting orders, then watch them grind to a halt. But when I ran the code. nothing happened. No deadlock, no errors just two processes finishing like old friends sharing a cup of coffee.

Here’s what went wrong, how I fixed it, and what you can learn from my struggle.

Django’s Database Connection Isn’t Fork-Safe

When using Python’s multiprocessing module, child processes inherit the parent’s database connection. This means both processes end up sharing the same connection, leading to sequential execution rather than true concurrency. Without real concurrent access, there’s no chance for a deadlock.

The Flawed Code

Here’s my initial (broken) implementation:

import time
from multiprocessing import Process
from django.db import connection, transaction
from myapp.models import Change

@transaction.atomic()
def process_1():
    change = Change.objects.select_for_update().get(id=1)
    time.sleep(5)  # Simulate work
    change = Change.objects.select_for_update().get(id=2)  # Oops, no conflict!

@transaction.atomic()
def process_2():
    change = Change.objects.select_for_update().get(id=2)
    time.sleep(5)
    change = Change.objects.select_for_update().get(id=1)

class DeadlockTestCase(TestCase):
    def test_deadlock(self):
        p1 = Process(target=process_1)
        p2 = Process(target=process_2)
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        # Tests passed, but no deadlock occurred!

Why This Fails

1. Shared Connections: Both processes inherited the same database connection, so their queries ran sequentially.
2. Decorator Misuse: Using @transaction.atomic() as a decorator outside Django’s TestCase context doesn’t properly isolate transactions across processes.
3. No Connection Reset: Django doesn’t automatically close/reopen connections in child processes.

Close Connections & Use Context Managers

To force true concurrency, each process must close the inherited connection and start fresh. Here’s the corrected code:

def process_1():
    connections.close_all()  # Critical: Reset connection
    with transaction.atomic():  # Use context manager, not decorator
        print("Process 1: Locking change 1")
        change = Change.objects.select_for_update().get(id=1)
        time.sleep(5)
        print("Process 1: Locking change 2")
        change = Change.objects.select_for_update().get(id=2)

def process_2():
    connections.close_all()
    with transaction.atomic():
        print("Process 2: Locking change 2")
        change = Change.objects.select_for_update().get(id=2)
        time.sleep(5)
        print("Process 2: Locking change 1")
        change = Change.objects.select_for_update().get(id=1)

Key Fixes

1. connections.close_all(): Resets the connection in each process, forcing Django to create a new one.
2. Context Managers Over Decorators: Using with transaction.atomic() ensures transactions are scoped to the process.
3. Logging: Added print() statements to visualize the locking sequence.

Practice Enhancements

To deepen your understanding, try these tweaks:

Add a Timeout

p1.join(timeout=3)
if p1.is_alive():
    print("Deadlock detected! Process 1 timed out.")

Deadlocks often cause timeouts—assert that one process fails.

2. Test Isolation Levels
   Modify your database settings to use 'isolation_level': 'REPEATABLE READ' (PostgreSQL) and see how locking behavior changes.
3. Wrap in a Management Command
   Move the test logic to a custom command:

# myapp/management/commands/deadlock_test.py
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    def handle(self, *args, **options):
        # Paste test logic here
        self.stdout.write("Deadlock simulation complete!")

Run it with python manage.py deadlock_test.

Final Thoughts

Simulating deadlocks in Django taught me two critical lessons:

• Database connections are not thread/process-safe by default. Always close them in child processes.
• Concurrency is tricky. Even small oversights (like a missing close_all()) can completely invalidate your tests.

If you’re working with multiprocessing in Django, treat database connections with suspicion. Test rigorously, log everything, and remember: what looks concurrent on the surface might just be sequential under the hood.