Hæng i Python-scriptet ved hjælp af SQLAlchemy og multiprocessing

Jeg tror på TypeError kommer fra multiprocessing 's get .

Jeg har fjernet al DB-koden fra dit script. Tag et kig på dette:

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

Bruger r.wait returnerer det forventede resultat, men ved hjælp af r.get rejser TypeError . Som beskrevet i pythons dokumenter , brug r.wait efter en map_async .

Rediger :Jeg er nødt til at ændre mit tidligere svar. Jeg tror nu på TypeError kommer fra SQLAlchemy. Jeg har ændret mit script for at genskabe fejlen.

Rediger 2 :Det ser ud til, at problemet er, at multiprocessing.pool spiller ikke godt, hvis en arbejder rejser en undtagelse, hvis konstruktør kræver en parameter (se også her ).

Jeg har ændret mit manuskript for at fremhæve dette.

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

I dit tilfælde, da din kode rejser en SQLAlchemy-undtagelse, er den eneste løsning, jeg kan komme i tanke om, at fange alle undtagelserne i do funktion og re-hæve en normal Exception i stedet. Noget som dette:

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

Rediger 3 :så det ser ud til at være en fejl med Python , men korrekte undtagelser i SQLAlchemy ville løse det:derfor har jeg rejst problemet med SQLAlchemy også.

Som en løsning på problemet tror jeg, at løsningen er i slutningen af ​​Rediger 2 ville gøre (indpakning af tilbagekald i try-except og re-raise).