How To Get All Pool.apply_async Processes To Stop Once Any One Process Has Found A Match In Python

I have the following code that is leveraging multiprocessing to iterate through a large list and find a match. How can I get all processes to stop once a match is found in any one

Solution 1:

You can check this question to see an implementation example solving your problem.

This works also with concurrent.futures pool.

Just replace the map method with apply_async and iterated over your list from the caller.

Something like this.

for part in grouper(alphabet, part_size):
    pool.apply_async(do_job, part, callback=self.callback)

grouper recipe

Solution 2:

multiprocessing isn't really designed to cancel tasks, but you can simulate it for your particular case by using pool.imap_unordered and terminating the pool when you get a hit:

defdo_job(first_bits):
    for x in itertools.product(first_bits, *itertools.repeat(alphabet, num_parts-1)):
        # CHECK FOR MATCH HEREprint(''.join(x))
        if match:
            returnTrue# If we exit loop without a match, function implicitly returns falsy None for us# Factor out part getting to simplify imap_unordered usedefget_part(i):
    if i == num_parts - 1:
        first_bit = alphabet[part_size * i :]
    else:
        first_bit = alphabet[part_size * i : part_size * (i+1)]

if __name__ == '__main__':
    # with statement with Py3 multiprocessing.Pool terminates when block exitswith multiprocessing.Pool(processes=4) as pool:

        # Don't need special case for final block; slices can for gotmatch in pool.imap_unordered(do_job, map(get_part, range(num_parts))):
             if gotmatch:
                 breakelse:
             print("No matches found")

This will run do_job for each part, returning results as fast as it can get them. When a worker returns True, the loop breaks, and the with statement for the Pool is exited, terminate-ing the Pool (dropping all work in progress).

Note that while this works, it's kind of abusing multiprocessing; it won't handle canceling individual tasks without terminating the whole Pool. If you need more fine grained task cancellation, you'll want to look at concurrent.futures, but even there, it can only cancel undispatched tasks; once they're running, they can't be cancelled without terminating the Executor or using a side-band means of termination (having the task poll some interprocess object intermittently to determine if it should continue running).