使用 SQLAlchemy 和多处理挂入 Python 脚本

考虑以下 Python 脚本，它使用 SQLAlchemy 和 Python 多处理模块。 这是 Debian squeeze 上的 Python 2.6.6-8+b1（默认）和 SQLAlchemy 0.6.3-3（默认）。 这是一些实际代码的简化版本。

import multiprocessing
from sqlalchemy import *
from sqlalchemy.orm import *
dbuser = ...
password = ...
dbname = ...
dbstring = "postgresql://%s:%s@localhost:5432/%s"%(dbuser, password, dbname)
db = create_engine(dbstring)
m = MetaData(db)

def make_foo(i):
    t1 = Table('foo%s'%i, m, Column('a', Integer, primary_key=True))

conn = db.connect()
for i in range(10):
    conn.execute("DROP TABLE IF EXISTS foo%s"%i)
conn.close()
db.dispose()

for i in range(10):
    make_foo(i)

m.create_all()

def do(kwargs):
    i, dbstring = kwargs['i'], kwargs['dbstring']

    db = create_engine(dbstring)
    Session = scoped_session(sessionmaker())
    Session.configure(bind=db)
    Session.execute("COMMIT; BEGIN; TRUNCATE foo%s; COMMIT;")
    Session.commit()
    db.dispose()

pool = multiprocessing.Pool(processes=5)               # start 4 worker processes
results = []
arglist = []
for i in range(10):
    arglist.append({'i':i, 'dbstring':dbstring})
r = pool.map_async(do, arglist, callback=results.append) # evaluate "f(10)" asynchronously
r.get()
r.wait()
pool.close()
pool.join()

该脚本挂起并显示以下错误消息。

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib/python2.6/multiprocessing/pool.py", line 259, in _handle_results
    task = get()
TypeError: ('__init__() takes at least 4 arguments (2 given)', <class 'sqlalchemy.exc.ProgrammingError'>, ('(ProgrammingError) syntax error at or near "%"\nLINE 1: COMMIT; BEGIN; TRUNCATE foo%s; COMMIT;\n        ^\n',))

当然，这里的语法错误是TRUNCATE foo%s;。我的问题是，为什么进程挂起，我可以说服它以错误退出，而不对我的代码进行重大修改吗？此行为与我的实际代码非常相似。

请注意，如果该语句被替换为类似的内容，则不会发生挂起print foobarbaz。另外，如果我们更换，挂起仍然会发生

Session.execute("COMMIT; BEGIN; TRUNCATE foo%s; COMMIT;")
Session.commit()
db.dispose()

by just Session.execute("TRUNCATE foo%s;")

我使用前一个版本，因为它更接近我的实际代码正在执行的操作。

另外，删除multiprocessing从图片中并连续循环遍历表格可以使挂起消失，并且它只是以错误退出。

我也对错误的形式感到困惑，特别是TypeError: ('__init__() takes at least 4 arguments (2 given)'少量。这个错误从何而来？看起来很可能是来自于某个地方multiprocessing code.

PostgreSQL 日志没有帮助。我看到很多像这样的行

2012-01-09 14:16:34.174 IST [7810] 4f0aa96a.1e82/1 12/583 0 ERROR:  syntax error at or near "%" at character 28
2012-01-09 14:16:34.175 IST [7810] 4f0aa96a.1e82/2 12/583 0 STATEMENT:  COMMIT; BEGIN; TRUNCATE foo%s; COMMIT;

但似乎没有其他相关的事情。

更新1：感谢lbolla和他的富有洞察力的分析，我能够提交Python 错误报告对这个。 看sbt该报告中的分析，以及here。另请参阅 Python 错误报告修复酸洗异常。因此，根据 sbt 的解释，我们可以重现原始错误

import sqlalchemy.exc
e = sqlalchemy.exc.ProgrammingError("", {}, None)
type(e)(*e.args)

这使

Traceback (most recent call last):
  File "<stdin>", line 9, in <module>
TypeError: __init__() takes at least 4 arguments (2 given)

更新 2：至少对于 SQLAlchemy，Mike Bayer 已修复此问题，请参阅错误报告StatementError 异常不可选取。。根据 Mike 的建议，我也向 psycopg2 报告了一个类似的错误，尽管我没有（也没有）有实际的损坏示例。无论如何，他们显然已经修复了这个问题，尽管他们没有提供修复的细节。看psycopg 异常无法被 pickle。为了更好地衡量，我还报告了一个 Python bugConfigParser 异常不可 pickle对应于lbolla提到的SO问题。看来他们想对此进行测试。

无论如何，这看起来在可预见的将来仍然会是一个问题，因为总的来说，Python 开发人员似乎没有意识到这个问题，因此也没有提防它。令人惊讶的是，似乎没有足够多的人使用多处理，因此这成为一个众所周知的问题，或者也许他们只是忍受它。我希望 Python 开发人员能够至少在 Python 3 中修复它，因为它很烦人。

我接受了 lbolla 的回答，因为如果没有他解释问题如何与异常处理相关，我可能无法理解这一点。我还要感谢 sbt，他解释说 Python 无法 pickle 异常是问题所在。我非常感谢他们俩，请投票支持他们的答案。谢谢。

更新 3：我发布了一个后续问题：捕获无法处理的异常并重新加注.

我相信TypeError来自multiprocessing's get.

我已经从您的脚本中删除了所有数据库代码。看看这个：

import multiprocessing
import sqlalchemy.exc

def do(kwargs):
    i = kwargs['i']
    print i
    raise sqlalchemy.exc.ProgrammingError("", {}, None)
    return i


pool = multiprocessing.Pool(processes=5)               # start 4 worker processes
results = []
arglist = []
for i in range(10):
    arglist.append({'i':i})
r = pool.map_async(do, arglist, callback=results.append) # evaluate "f(10)" asynchronously

# Use get or wait?
# r.get()
r.wait()

pool.close()
pool.join()
print results

Using r.wait返回预期的结果，但使用r.get raises TypeError。如中所述python 的文档, use r.wait之后map_async.

Edit: 我必须修改我之前的答案。我现在相信TypeError来自 SQLAlchemy。我修改了我的脚本以重现该错误。

Edit 2: 看来问题是这样的multiprocessing.pool如果任何工作人员引发其构造函数需要参数的异常，则效果不佳（另请参阅here).

我修改了我的脚本以突出这一点。

import multiprocessing

class BadExc(Exception):
    def __init__(self, a):
        '''Non-optional param in the constructor.'''
        self.a = a

class GoodExc(Exception):
    def __init__(self, a=None):
        '''Optional param in the constructor.'''
        self.a = a

def do(kwargs):
    i = kwargs['i']
    print i
    raise BadExc('a')
    # raise GoodExc('a')
    return i

pool = multiprocessing.Pool(processes=5)
results = []
arglist = []
for i in range(10):
    arglist.append({'i':i})
r = pool.map_async(do, arglist, callback=results.append)
try:
    # set a timeout in order to be able to catch C-c
    r.get(1e100)
except KeyboardInterrupt:
    pass
print results

在您的情况下，考虑到您的代码引发了 SQLAlchemy 异常，我能想到的唯一解决方案是捕获do功能并重新提高正常Exception反而。像这样的东西：

import multiprocessing

class BadExc(Exception):
    def __init__(self, a):
        '''Non-optional param in the constructor.'''
        self.a = a

def do(kwargs):
    try:
        i = kwargs['i']
        print i
        raise BadExc('a')
        return i
    except Exception as e:
        raise Exception(repr(e))

pool = multiprocessing.Pool(processes=5)
results = []
arglist = []
for i in range(10):
    arglist.append({'i':i})
r = pool.map_async(do, arglist, callback=results.append)
try:
    # set a timeout in order to be able to catch C-c
    r.get(1e100)
except KeyboardInterrupt:
    pass
print results

Edit 3: 所以，这似乎是一个Python 的错误，但是 SQLAlchemy 中的适当异常可以解决它：因此，我已经使用 SQLAlchemy 提出问题, too.

作为解决问题的方法，我认为最后的解决方案Edit 2会做（将回调包装在 try-except 中并重新引发）。