[Django]-Django multiprocessing and database connections

When using multiple databases, you should close all connections.

from django import db
for connection_name in db.connections.databases:
    db.connections[connection_name].close()

EDIT

Please use the same as @lechup mentionned to close all connections(not sure since which django version this method was added):

from django import db
db.connections.close_all()

For Python 3 and Django 1.9 this is what worked for me:

import multiprocessing
import django
django.setup() # Must call setup

def db_worker():
    for name, info in django.db.connections.databases.items(): # Close the DB connections
        django.db.connection.close()
    # Execute parallel code here

if __name__ == '__main__':
    multiprocessing.Process(target=db_worker)

Note that without the django.setup() I could not get this to work. I am guessing something needs to be initialized again for multiprocessing.

I had “closed connection” issues when running Django test cases sequentially. In addition to the tests, there is also another process intentionally modifying the database during test execution. This process is started in each test case setUp().

A simple fix was to inherit my test classes from TransactionTestCase instead of TestCase. This makes sure that the database was actually written, and the other process has an up-to-date view on the data.

Another way around your issue is to initialise a new connection to the database inside the forked process using:

from django.db import connection    
connection.connect()

(not a great solution, but a possible workaround)

if you can’t use celery, maybe you could implement your own queueing system, basically adding tasks to some task table and having a regular cron that picks them off and processes? (via a management command)

Hey I ran into this issue and was able to resolve it by performing the following (we are implementing a limited task system)

task.py

from django.db import connection

def as_task(fn):
    """  this is a decorator that handles task duties, like setting up loggers, reporting on status...etc """ 
    connection.close()  #  this is where i kill the database connection VERY IMPORTANT
    # This will force django to open a new unique connection, since on linux at least
    # Connections do not fare well when forked 
    #...etc

ScheduledJob.py

from django.db import connection

def run_task(request, job_id):
    """ Just a simple view that when hit with a specific job id kicks of said job """ 
    # your logic goes here
    # ...
    processor = multiprocessing.Queue()
    multiprocessing.Process(
        target=call_command,  # all of our tasks are setup as management commands in django
        args=[
            job_info.management_command,
        ],
        kwargs= {
            'web_processor': processor,
        }.items() + vars(options).items()).start()

result = processor.get(timeout=10)  # wait to get a response on a successful init
# Result is a tuple of [TRUE|FALSE,<ErrorMessage>]
if not result[0]:
    raise Exception(result[1])
else:
   # THE VERY VERY IMPORTANT PART HERE, notice that up to this point we haven't touched the db again, but now we absolutely have to call connection.close()
   connection.close()
   # we do some database accessing here to get the most recently updated job id in the database

Honestly, to prevent race conditions (with multiple simultaneous users) it would be best to call database.close() as quickly as possible after you fork the process. There may still be a chance that another user somewhere down the line totally makes a request to the db before you have a chance to flush the database though.

In all honesty it would likely be safer and smarter to have your fork not call the command directly, but instead call a script on the operating system so that the spawned task runs in its own django shell!

If all you need is I/O parallelism and not processing parallelism, you can avoid this problem by switch your processes to threads. Replace

from multiprocessing import Process

with

from threading import Thread

The Thread object has the same interface as Procsess

If you’re also using connection pooling, the following worked for us, forcibly closing the connections after being forked. Before did not seem to help.

from django.db import connections
from django.db.utils import DEFAULT_DB_ALIAS

connections[DEFAULT_DB_ALIAS].dispose()

One possibility is to use multiprocessing spawn child process creation method, which will not copy django’s DB connection details to the child processes. The child processes need to bootstrap from scratch, but are free to create/close their own django DB connections.

In calling code:

import multiprocessing
from myworker import work_one_item # <-- Your worker method

...

# Uses connection A
list_of_items = djago_db_call_one()

# 'spawn' starts new python processes
with multiprocessing.get_context('spawn').Pool() as pool:
    # work_one_item will create own DB connection
    parallel_results = pool.map(work_one_item, list_of_items)

# Continues to use connection A
another_db_call(parallel_results) 

In myworker.py:

import django. # <-\
django.setup() # <-- needed if you'll make DB calls in worker

def work_one_item(item):
   try:
      # This will create a new DB connection
      return len(MyDjangoModel.objects.all())

   except Exception as ex:
      return ex

Note that if you’re running the calling code inside a TestCase, mocks will not be propagated to the child processes (will need to re-apply them).

You could give more resources to Postgre, in Debian/Ubuntu you can edit :

nano /etc/postgresql/9.4/main/postgresql.conf

by replacing 9.4 by your postgre version .

Here are some useful lines that should be updated with example values to do so, names speak for themselves :

max_connections=100
shared_buffers = 3000MB
temp_buffers = 800MB
effective_io_concurrency = 300
max_worker_processes = 80

Be careful not to boost too much these parameters as it might lead to errors with Postgre trying to take more ressources than available. Examples above are running fine on a Debian 8GB Ram machine equiped with 4 cores.

Overwrite the thread class and close all DB connections at the end of the thread. Bellow code works for me:

class MyThread(Thread):
    def run(self):
        super().run()

        connections.close_all()

def myasync(function):
    def decorator(*args, **kwargs):
        t = MyThread(target=function, args=args, kwargs=kwargs)
        t.daemon = True
        t.start()

    return decorator

When you need to call a function asynchronized:

@myasync
def async_function():
    ...