All geek questions in one place

Python parallel execution - how to debug effectively?

The following is a Python issue that demonstrates how to execute a function funcin parallel using multiprocessing.Pool. The number of elements Npto iterate. The function funcsimply returns Npminus the iterability index. As you can see, I use the queue to return values ​​from the function when working in parallel mode.

If I install runParallel=False, the program can be executed in sequential mode.

The program works fine, for runParallel=Falseand runParallel=True, but now the main problem arises: As you can see below, if the parameter is problemIndexslightly lower Np(for example, problemIndex=7), then I am making a floating-point exception. I am divided by zero - stupid me :-)

On startup, runParallel=FalseI can see the line number of the source error code, and I will catch the error directly.

$ python map.py 
Traceback (most recent call last):
  File "map.py", line 63, in <module>
    a = func(argList[p])
  File "map.py", line 22, in func
    ret = 1/(args["index"]-args["problemIndex"])
ZeroDivisionError: integer division or modulo by zero

Nice!

However, for runParallel=TrueI just finished the printed “Bummer” section without indicating the source of the error. Annoyingly!

My question is: for runParallel=True, how can I effectively debug this and get the line number from the buggy line of code from Pool()?

#!/usr/bin/python
# map.py
import time
import multiprocessing
import sys
import random

# Toggle whether we run parallel or not
runParallel = True

# Problematic index - if less than Np we create an exception
problemIndex = 13

# Number of compute problems
Np = 10

def func(args):
    # Emulate that the function might be fast or slow
    time.sleep(random.randint(1,4))
    ret  = args["Np"] - args["index"]
    # Emulate a bug 
    if args["index"]==args["problemIndex"]:
        ret = 1/(args["index"]-args["problemIndex"])
    # Return data
    if args["runParallel"]:
        # We use a queue thus ordering may not be protected
        args["q"].put((args["index"],ret))
    else:
        return ret

# Return queue used when running parallel
manager = multiprocessing.Manager()
q = manager.Queue()

# Build argument lists
argList = []
for i in range(Np):
    args={}
    args["index"] = i # index
    args["Np"] = Np   # Number of problems
    args["q"] = q     # return queue for parallel execution mode
    args["problemIndex"] = problemIndex  # if index == problemIndex then func will malfunction
    args["runParallel"] = runParallel    # should we run parallel
    argList.append(args)

#should we run parallel
if runParallel:
    # Run 10 processes in parallel
    p = multiprocessing.Pool(processes=10)
    ret = p.map_async(func, argList)
    ret.wait()
    qLen = q.qsize()
    p.close()    
    if not qLen == Np:
        print "Bummer - one of more worker threads broke down",Np,qLen
        sys.exit(0)

resultVector = [None]*Np
for p in range(Np):
    if runParallel:
        (i,a) = q.get(timeout=0.1)
    else:
        i = p
        a = func(argList[p])
    resultVector[i] = a

for i in range(Np):
    print "Index", i, "gives",resultVector[i]
+4
source share
3 answers

, . /, , traceback.format_exc . , .

import traceback
import multiprocessing as mp
import time

def mpFunctionReportError(kwargs):
    '''
    wrap any function and catch any errors from f, 
    putting them in pipe instead of raising
    kwargs must contain 'queue' (multiprocessing queue) 
                    and 'f' function to be run
    '''
    queue = kwargs.pop('queue')
    f = kwargs.pop('f')
    rslt=None
    try:
        rslt = f(**kwargs)
        queue.put(rslt)
    except Exception, e:
        queue.put([e,traceback.format_exc(e)])
    return

def doNothing(a):
    return a

def raiseException(a):
    a='argh'
    raise ValueError('this is bad')


manager = mp.Manager()
outQ = manager.Queue()
p = mp.Pool(processes=4)

ret = p.map_async(mpFunctionReportError,iterable=[dict(f=doNothing,queue=outQ,a='pointless!') for i in xrange(4)])
ret.wait()
time.sleep(1)
for i in xrange(4):
    print(outQ.get_nowait())

ret = p.map_async(mpFunctionReportError,iterable=[dict(f=raiseException,queue=outQ,a='pointless!') for i in xrange(2)])
ret.wait()
time.sleep(1)
for i in xrange(2):
    e,trace = outQ.get_nowait()
    print(e)
    print(trace)

:

pointless!
pointless!
pointless!
pointless!
this is bad
Traceback (most recent call last):
  File "/home/john/projects/mpDemo.py", line 13, in mpFunctionReportError
    rslt = f(**kwargs)
  File "/home/john/projects/mpDemo.py", line 24, in raiseException
    raise ValueError('this is bad')
ValueError: this is bad

this is bad
Traceback (most recent call last):
  File "/home/john/projects/mpDemo.py", line 13, in mpFunctionReportError
    rslt = f(**kwargs)
  File "/home/john/projects/mpDemo.py", line 24, in raiseException
    raise ValueError('this is bad')
ValueError: this is bad
+2

, :

def func(args):
  try:
    # Emulate that the function might be fast or slow
    time.sleep(random.randint(1,4))
    ret  = args["Np"] - args["index"]
    # Emulate a bug 
    if args["index"]==args["problemIndex"]:
      ret = 1/(args["index"]-args["problemIndex"])
    # Return data
    if args["runParallel"]:
      # We use a queue thus ordering may not be protected
      args["q"].put((args["index"],ret))
    else:
      return ret
  except Exception as e:
    logging.exception(e)
    raise

( problemIndex = 9):

ERROR:root:integer division or modulo by zero
Traceback (most recent call last):
  File "/home/rciorba/test.py", line 26, in func
    ret = 1/(args["index"]-args["problemIndex"])
ZeroDivisionError: integer division or modulo by zero
Bummer - one of more worker threads broke down 10 9
+1

, .

, , , .. "func", radu.ciorba. .

Johns 100% , , Johns. , !

#!/usr/bin/python
# map.py solution
import time
import multiprocessing
import sys
import random
import logging
import traceback

# Toggle whether we run parallel or not
runParallel = True

# Problematic index - if less than Np we create an exception
problemIndex = 14

# Number of compute problems
Np = 10

def func(args):
    # Emulate that the function might be fast or slow
    time.sleep(random.randint(1,4))
    ret  = args["Np"] - args["index"]

    # Emulate a bug 
    if args["index"]==args["problemIndex"]:
        ret = 1/(args["index"]-args["problemIndex"])

    # Return data
    return (args["index"],ret)

def mpFunctionReportError(args):
    rslt=None
    q = args["q"]
    rslt = {"index":args["index"],
            "args":None,
            "error":None, 
            "traceback":None}
    try:
        rslt["result"] = func(args)
        q.put(rslt)
    except Exception as e:
        rslt["result"] = None
        rslt["error"] = e
        rslt["args"] = str(args)
        rslt["traceback"] = traceback.format_exc(e)
        q.put(rslt)

# Return queue used when running parallel
manager = multiprocessing.Manager()
q = manager.Queue()

# Build argument lists
argList = []
for i in range(Np):
    args={}
    args["index"] = i # index
    args["Np"] = Np   # Number of problems
    args["q"] = q     # return queue for parallel execution mode
    args["problemIndex"] = problemIndex  # if index == problemIndex then func will malfunction
    args["runParallel"] = runParallel    # should we run parallel
    argList.append(args)


resultVector = [None]*Np

#should we run parallel
if runParallel:
    # Run 10 processes in parallel
    p = multiprocessing.Pool(processes=10)
    ret = p.map_async(mpFunctionReportError, argList)
    # Wait until error or done
    ret.wait()
    # Queue size
    qLen = q.qsize()
    p.close()    
    # List for the errors
    bugList = {}
    # Loop the queue
    for i in range(qLen):
        # Pop a value
        returnVal = q.get()
        # Check for the error code
        if returnVal["error"] is not None:
            bugList[returnVal["index"]] = returnVal
        else:
            resultVector[returnVal["index"]] = returnVal["result"]

    # Print the list of errors
    if bugList:        
        print "-"*70
        print "Some parts of the parallel execution broke down. Error list:"
        print "-"*70
        for i in bugList:
            print "Index :",bugList[i]["index"]
            print "Error code :",bugList[i]["error"]
            print "Traceback :",bugList[i]["traceback"]
            print "Args :",bugList[i]["args"]
            print "-"*70
        sys.exit(0)
else:
    for p in range(Np):
        resultVector[i] = func(argList[p])

for i in range(Np):
    print "Index", i, "gives",resultVector[i]

"runParallel = True" "problemIndex = 4",

----------------------------------------------------------------------
Some parts of the parallel execution broke down. Error list:
----------------------------------------------------------------------
Index : 4
Error code : integer division or modulo by zero
Traceback : Traceback (most recent call last):
  File "fix3.py", line 44, in mpFunctionReportError
    rslt["result"] = func(args)
  File "fix3.py", line 26, in func
    ret = 1/(args["index"]-args["problemIndex"])
ZeroDivisionError: integer division or modulo by zero

Args : {'Np': 10, 'index': 4, 'problemIndex': 4, 'q': <AutoProxy[Queue] object, typeid 'Queue' at 0xb708710c>, 'runParallel': True}
0
source

Source: https://habr.com/ru/post/1537886/

More articles:


All Articles