2

I have a python flask app. I would like to use concurrency when responding to a specific route without creating extra threads on every request.

There's a route defined as follows:

def sentence_numfound(path):
    nf = util.NumFound(path)
    return json.dumps(nf.results(path))

nf.results() needs to issue multiple http requests before returning, and I would like to do them in parallel. Currently I'm doing this:

class NumFound:
    def __init__(self, path):
        queries = get_queries(path) # A list
        self.__results = [{}] * len(queries)
        self.queue = Queue.Queue()
        for i, q in enumerate(queries):
            self.queue.put((i, q))

    def results(self):
        num_workers = 31
        for i in range(num_workers):
            t = threading.Thread(target=self.worker)
            t.daemon = True
            t.start()
        self.queue.join()
        return self.__results

    def worker(self):
        while True:
            i, q = self.queue.get()
            self.__results[i] = foo(q)
            self.queue.task_done()

The problem is that new threads are created on every request and there's no way to close them. Eventually the route fails with an error because python can't create any more threads.

Is there a simple way to reuse the threads? Or another way to achieve the concurrency?

3
  • I'm sorry it's not an answer. Why do you want to create thread? Werkzeug, Apahce, nginx ... have this feature already.
    – emesday
    Apr 18, 2014 at 16:37
  • Apache etc have the ability to create threads for incoming requests, but my code needs to make many outgoing requests for a single incoming request. Network latency is the bottleneck, and concurrency gives a huge speedup.
    – elplatt
    Apr 18, 2014 at 17:00
  • Also not an answer, but have you considered using gevent/green threads? For something heavily i/o bound, it might work nicely: sdiehl.github.io/gevent-tutorial You could also punt and use something like celery - the route returns a spinner and then redirects to the final page when everything returns. Apr 18, 2014 at 19:41

1 Answer 1

4

I think you will get close to your implementation using a multiprocessing.Pool.

You can set up a pool of workers as follows:

from multiprocessing import Pool
pool = Pool(processes=31)

Then all your route needs to do is submit the jobs to the pool and wait for all of them to be done. I can't test this because you haven't provided enough code, but it may look more or less like this:

def sentence_numfound(path):
    return jsonify(pool.map(foo, get_queries(path)))

This basically calls foo(query) for each query in the processes owned by the pool, all in parallel. The map() call will return when all the jobs are done. The return value is an array with the results, in the same order as the input array.

I hope this helps!

1
  • Works great! I just had to make sure to define foo() before creating the Pool().
    – elplatt
    Apr 21, 2014 at 20:20

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.