multiprocessing + psycopg2 でハマったところ

やりたかったこと

複数回pd.read_sql実行する処理をmultiprocessingで並列処理したかった。

Versions

Name	Version
python	3.7.3
pandas	0.24.2
numpy	1.16.2
psycopg2-binary	2.8.4
PostgreSQL	11.5

成功例（非並列ver.）

コード

import multiprocessing
import pandas as pd
import numpy as np
import psycopg2

def get_connection():
	connection = psycopg2.connect(
						host='hostname',
						user='username',
						database='databasename',
						password='password')
	return connection

def function():
	for fuga in hoge:

		# fuga, pd.read_sql(sql='SELECT yyy...', con=conn)を含むなんらかの処理

	return

if __name__ == '__main__':
	with get_connection() as conn:
		hoge = np.ravel(pd.read_sql(sql='SELECT xxx...', con=conn).to_numpy())

		function

失敗例

上記成功例のコード最下部をmultiprocessingで書き直し、それに応じて変数を食えるようにfunction()を修正したところ、エラーが出力された。

コード

def function(i):
	fuga = hoge[i]

	# fuga, pd.read_sql(sql='SELECT yyy...', con=conn)を含むなんらかの処理

	return

if __name__ == '__main__':
	with get_connection() as conn:
		hoge = np.ravel(pd.read_sql(sql='SELECT xxx...', con=conn).to_numpy())

		with multiprocessing.Pool(processes=64) as pool:
			for _ in pool.imap_unordered(function, range(len(hoge))):
				pass

エラー抜粋

multiprocessing.pool.RemoteTraceback:

# 中略

psycopg2.OperationalError: lost synchronization with server: got message type

# 中略

psycopg2.InterfaceError: connection already closed

# 中略

pandas.io.sql.DatabaseError: Execution failed on sql: SELECT yyy...
lost synchronization with server: got message type
unable to rollback

# 中略

成功例

上記失敗例のコードにおいて、function()に接続情報を書くことで回避できた。

コード

def function(i):
	with get_connection() as conn:
		fuga = hoge[i]

		# fuga, pd.read_sql(sql='SELECT yyy...', con=conn)を含むなんらかの処理

	return

注意

並列数やfunction()の処理内容に依っては下記エラーが出るが、export OMP_NUM_THREADS=1とすることで回避できた。

OMP: Error #34: System unable to allocate necessary resources for OMP thread:
OMP: System error #11: Resource temporarily unavailable
OMP: Hint Try decreasing the value of OMP_NUM_THREADS.