CODEWITHSUNDEEP
pythonpython-3.xconcurrencypython-multiprocessing
HighFlyingHawk
HighFlyingHawk

Reputation: 41

Catching exceptions in a multiprocessing loop

Is it possible to catch an exception raised from the loop in a callable from concurrent.futures.Executor such as this? --

with concurrent.futures.ThreadPoolExecutor(max_workers=num_cores) as executor:
    futures = {executor.submit(self._process_ticket, i) for i in items}
    concurrent.futures.wait(futures)

I am trying to loop through thousands of objects in a list to perform tasks on and just set up multiprocessing to process them quicker. This is working great but because of the objects I am working on there is a possibility an exception can be raised from the for loop during the process and because it is being called by concurrent.futures.Executor I am unable to catch it the way I was catching it previously.

Below is a simple example of how I was catching exceptions previously through a serialized process. I had to create a workaround to force the iteration to try and process the object again and continue from that step forward as the exception causes the loop to stop:

def task(self, items, step=None):
    # items = [list,of,many,objects]
    try:
        for i in range(0 if step is None else step, len(items)):
            with app.app_context():
                ## do things with items[i]..
    except Exception as e:
        self.task(items, i)

I'm basically wanting to force an object to be processed again in the event the loop is interrupted by an exception and continue the loop from that step.

Upvotes: 2

Views: 153

Answers (1)

HighFlyingHawk
HighFlyingHawk

Reputation: 41

Nevermind, figured this out a few days ago. Not sure if this is the best/recommended way to get this done but it seems to get the job done:

with concurrent.futures.ThreadPoolExecutor(max_workers=num_cores) as executor:
    try:
        for i in range(step, len(items)):
            futures = {executor.submit(self.task, items[i])}
    except:
        self.task(items, i)

    concurrent.futures.wait(futures)

Still open for suggestions/advice

Upvotes: 1

Related Questions