[解決済み] Djangoのマルチプロセッシングとデータベース接続

質問

背景について。

私は、PostgresデータベースでDjangoを使用するプロジェクトに取り組んでいます。 また、いくつかのウェブ検索で言及されているので、一応 mod_wsgi も使用しています。 Web フォームの送信時に、Django のビューはかなりの時間 (ユーザが待ちたい時間以上) を要するジョブをキックするので、バックグラウンドでシステムコールを介してジョブをキックオフします。 現在実行されているジョブは、データベースへの読み書きができる必要があります。 このジョブには非常に長い時間がかかるため、マルチプロセシングを使用して一部を並行して実行します。

問題です。

トップレベルのスクリプトがデータベース接続を持ち、子プロセスを生成するとき、親の接続が子プロセスで利用可能であると思われます。 そして、SET TRANSACTION ISOLATION LEVEL がクエリの前に呼び出されなければならない方法についての例外が発生します。 調査によると、これは複数のプロセスで同じデータベース接続を使用しようとすることが原因であることがわかりました。 私が見つけたあるスレッドでは、子プロセスの開始時に connection.close() を呼び出すことで、Django が必要な時に自動的に新しい接続を作成し、各子プロセスが固有の接続を持つようにする、つまり、共有しないようにすることを提案していました。 子プロセスで connection.close() を呼び出すと、親プロセスが接続が切れたと文句を言うので、これはうまくいきませんでした。

その他の発見。

私が読んだいくつかのものは、これを実際に行うことはできないし、マルチプロセシング、mod_wsgi、および Django は一緒にうまく機能しないことを示すようでした。 これは信じがたいことだと思います。

いくつかの人は celery を使うことを提案し、それは長期的な解決策かもしれませんが、私はいくつかの承認プロセスを保留して、現時点では celery をインストールすることができないので、今はオプションではありません。

SO や他の場所で、永続的なデータベース接続に関するいくつかの参照を見つけましたが、これは別の問題であると思います。

また、psycopg2.pool と pgpool への参照と bouncer についての何かを見つけました。確かに、私はそれらについて読んでいたもののほとんどを理解していませんでしたが、私が探していたものであるとして、確かにそれは私の目に飛び込んできませんでした。

現在の "ワークアラウンド"。

今のところ、私は物事を直列に実行することに戻しており、それは動作しますが、私が望むよりも遅いです。

並行して実行するためにマルチプロセシングを使用する方法について、何か提案はありますか? 親と 2 つの子のすべてにデータベースへの独立した接続を持たせることができれば、物事はうまくいくように思えますが、そのような動作を得ることができないようです。

ありがとうございます、そして長くてすみません!

どのように解決するのですか？

マルチプロセシングはプロセスをフォークするため、プロセス間で接続オブジェクトをコピーし、したがって、親プロセスのすべてのファイル記述子をコピーします。とはいえ、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は、マルチプロセッシングの意味でスレッドと連携しているわけではありません。むしろ、高負荷時の接続を高速化するために、リクエスト毎に接続を閉じないようにするためのソリューションです。

更新しました。

データベース接続の問題を完全に取り除くには、データベースと接続するすべてのロジックをdb_workerに移動するだけです - 引数としてQueryDictを渡したかったのですが... もっと良いアイデアは、単にIDのリストを渡すことです... 参照 クエリディクト と values_list('id', flat=True) を参照してください。 db_worker に渡す前に list(QueryDict) に変換することを忘れないでください。そのおかげで、モデルのデータベース接続をコピーする必要がありません。

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]))