Многопроцессорность и соединения с базой данных Django

Задний план:

Я работаю над проектом, который использует Django с базой данных Postgres. Мы также используем mod_wsgi в случае, если это имеет значение, так как некоторые из моих поисковых запросов упоминали об этом. При отправке веб-формы представление Django запускает работу, которая займет значительное количество времени (больше, чем хотелось бы ждать пользователю), поэтому мы запускаем работу с помощью системного вызова в фоновом режиме. Выполняемое задание должно иметь возможность чтения и записи в базу данных. Поскольку это задание занимает много времени, мы используем многопроцессорность для параллельного выполнения его частей.

Проблема:

У сценария верхнего уровня есть соединение с базой данных, и когда он порождает дочерние процессы, кажется, что соединение родителя доступно для детей. Затем есть исключение о том, как SET TRANSACTION ISOLATION LEVEL должен вызываться перед запросом. Исследования показали, что это происходит из-за попытки использовать одно и то же соединение с базой данных в нескольких процессах. Один поток, который я обнаружил, предлагал вызвать connection.close () в начале дочерних процессов, чтобы Django автоматически создавал новое соединение, когда оно ему нужно, и, следовательно, каждый дочерний процесс будет иметь уникальное соединение, то есть не разделяемое. У меня это не сработало, так как вызов connection.close () в дочернем процессе заставил родительский процесс пожаловаться на потерю соединения.

Другие результаты:

Некоторые вещи, которые я прочитал, похоже, указывают на то, что вы действительно не можете этого сделать, и что multiprocessing, mod_wsgi и Django плохо работают вместе. Мне кажется, в это трудно поверить.

Некоторые предлагали использовать сельдерей, что могло бы быть долгосрочным решением, но я не могу установить сельдерей в настоящее время, ожидая некоторых процессов утверждения, поэтому сейчас это не вариант.

Нашел несколько ссылок на SO и в других местах о постоянных соединениях с базой данных, что, по моему мнению, является другой проблемой.

Также нашел ссылки на psycopg2.pool и pgpool и кое-что о вышибале. По общему признанию, я не понимал большей части того, что читал по ним, но это определенно не бросалось мне в глаза как то, что я искал.

Текущий «Рабочий процесс»:

На данный момент я вернулся к последовательному запуску вещей, и это работает, но медленнее, чем хотелось бы.

Любые предложения относительно того, как я могу использовать многопроцессорную обработку для параллельной работы? Похоже, что если бы у меня был родитель и двое детей, которые имели бы независимые подключения к базе данных, все было бы хорошо, но я не могу получить такое поведение.

Спасибо и извините за длину!

Многопроцессорность копирует объекты соединения между процессами, потому что она разделяет процессы и, следовательно, копирует все файловые дескрипторы родительского процесса. При этом соединение с SQL-сервером - это просто файл, вы можете увидеть его в linux в / proc // fd / .... любой открытый файл будет совместно использоваться разветвленными процессами. Вы можете узнать больше о разветвлении здесь .

Мое решение заключалось в том, чтобы просто закрыть соединение с БД непосредственно перед запуском процессов, каждый процесс воссоздает само соединение, когда оно понадобится (проверено в django 1.4):

from django import db
db.connections.close_all()
def db_worker():      
    some_paralell_code()
Process(target = db_worker,args = ())

Pgbouncer / pgpool не связан с потоками в смысле многопроцессорности. Это скорее решение, чтобы не закрывать соединение при каждом запросе = ускорение подключения к postgres при высокой нагрузке.

Обновить:

Чтобы полностью устранить проблемы с подключением к базе данных, просто переместите всю логику, связанную с базой данных, в db_worker - я хотел передать QueryDict в качестве аргумента ... Лучшая идея - просто передать список идентификаторов ... См. QueryDict и values_list ('id', flat = Верно), и не забудьте превратить его в список! list (QueryDict) перед переходом к db_worker. Благодаря этому мы не копируем подключение к базе данных моделей.

def db_worker(models_ids):        
    obj = PartModelWorkerClass(model_ids) # here You do Model.objects.filter(id__in = model_ids)
    obj.run()


model_ids = Model.objects.all().values_list('id', flat=True)
model_ids = list(model_ids) # cast to list
process_count = 5
delta = (len(model_ids) / process_count) + 1

# do all the db stuff here ...

# here you can close db connection
from django import db
db.connections.close_all()

for it in range(0:process_count):
    Process(target = db_worker,args = (model_ids[it*delta:(it+1)*delta]))   

При использовании нескольких баз данных следует закрыть все соединения.

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

РЕДАКТИРОВАТЬ

Используйте то же, что и @lechup, чтобы закрыть все соединения (не уверен, с какой версии django был добавлен этот метод):

from django import db
db.connections.close_all()

Для Python 3 и Django 1.9 это сработало для меня:

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)

Обратите внимание, что без django.setup () я не смог бы заставить это работать. Я предполагаю, что что-то нужно снова инициализировать для многопроцессорной обработки.

У меня были проблемы с "закрытым соединением" при последовательном запуске тестовых примеров Django . Помимо тестов, существует еще один процесс, намеренно изменяющий базу данных во время выполнения теста. Этот процесс запускается в каждом тестовом примере setUp ().

Простым решением было унаследовать мои тестовые классы от TransactionTestCaseвместо TestCase. Это гарантирует, что база данных действительно была записана, а другой процесс имеет актуальное представление о данных.

(не лучшее решение, но возможное решение)

Если вы не можете использовать сельдерей, возможно, вы могли бы реализовать свою собственную систему очередей, в основном добавляя задачи в какую-то таблицу задач и имея обычный cron, который их выбирает и обрабатывает? (через команду управления)

Привет, я столкнулся с этой проблемой и смог решить ее, выполнив следующие действия (мы реализуем ограниченную систему задач)

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

Честно говоря, чтобы предотвратить состояние гонки (с несколькими одновременными пользователями), было бы лучше вызвать database.close () как можно быстрее после того, как вы разветвите процесс. Тем не менее, все еще может быть вероятность, что другой пользователь где-то в конце линии полностью сделает запрос к db, прежде чем у вас будет возможность очистить базу данных.

Честно говоря, было бы безопаснее и разумнее, если бы вилка не вызывала команду напрямую, а вместо этого вызывала сценарий в операционной системе, чтобы порожденная задача выполнялась в собственной оболочке django!

Вы можете предоставить Postgre больше ресурсов, в Debian / Ubuntu вы можете редактировать:

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

заменив 9.4 вашей версией postgre.

Вот несколько полезных строк, которые для этого следует обновить примерами значений, названия говорят сами за себя:

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

Будьте осторожны, не увеличивайте эти параметры слишком сильно, так как это может привести к ошибкам, когда Postgre попытается использовать больше ресурсов, чем доступно. Приведенные выше примеры отлично работают на машине Debian с 8 ГБ оперативной памяти, оснащенной 4 ядрами.

Если вам нужен только параллелизм ввода-вывода, а не параллелизм обработки, вы можете избежать этой проблемы, переключив свои процессы на потоки. Заменить

from multiprocessing import Process

с участием

from threading import Thread

ThreadОбъект имеет тот же интерфейс,Procsess

Если вы также используете пул соединений, у нас сработало следующее: принудительное закрытие соединений после разветвления. Раньше вроде не помогало.

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

connections[DEFAULT_DB_ALIAS].dispose()