[FIXED] Concurrent futures submit task to process pool non-blockingly

Issue

This question is Python 2-specific, using the community maintained backport of concurrent.futures.

I'm trying to use a ProcessPoolExecutor (with maxWorkers trivially set to 2) to run two tasks in parallel. Those tasks are both Python functions, and I'd like each of them to run in their own process. They don't need to coordinate with each other (I don't even need to know the exit status). I just want to be able to launch processes simultaneously and limit the number of processes that run in parallel at any given moment in time.

import concurrent.futures as futures
import time


def do_stuff(name):
    for x in range(10):
        print name, x
        time.sleep(1)


pool = futures.ProcessPoolExecutor(max_workers=2)
pool.submit(do_stuff("a"))
print "a submitted!"
pool.submit(do_stuff("b"))

This, however, prints

a 0 a 1 ... a 9 a submitted! b 0 b 1 ... b 9

Why is submit a blocking operation? Is there a nonblocking equivalent?

Here's an example using the multiprocessing library that has the behavior I want. It starts each process nonblockingly and then calls join (which is presumably just a thin wrapper around waitpid(2)). However, this technique does not give me the ability to limit the number of processes that run in parallel at any given moment.

import multiprocessing
import time


def do_stuff(name):
    for x in range(10):
        print name, x
        time.sleep(1)


proc_a = multiprocessing.Process(target=do_stuff, args="a")
proc_b = multiprocessing.Process(target=do_stuff, args="b")
proc_a.start()
proc_b.start()
proc_a.join()
proc_b.join()

Solution

The serial print out (instead of a concurrent one) from your code appears to be caused by the wrong syntax you had used to submit the argument of your function. Argument should be separated from function using a comma for Executor.submit(). Try this version instead.

import concurrent.futures as futures
import time

def do_stuff(name):
    for x in range(10):
        print name, x
        time.sleep(1)

pool = futures.ProcessPoolExecutor(max_workers=2)
pool.submit(do_stuff, "a")
print "a submitted!"
pool.submit(do_stuff, "b")
print "b submitted!"

Also, I recommend using the "with" statement to manage your submissions whenever possible as it will ensure proper closure/shutdown of the concurrent.futures.Executor. This was mentioned in the document.

with futures.ProcessPoolExecutor(max_workers=2) as executor:
    executor.submit(do_stuff, "a") 
    print "a submitted!"
    executor.submit(do_stuff, "b")
    print "b submitted!"

Answered By - Sun Bear