Custom thread-pooling

Question

Any suggestions/improvements for the following custom thread-pool code?

import threading
from Queue import Queue

class Worker(threading.Thread):
    def __init__(self, function, in_queue, out_queue):
        self.function = function
        self.in_queue, self.out_queue = in_queue, out_queue
        super(Worker, self).__init__()

    def run(self):
        while True:
            if self.in_queue.empty(): break
            data = in_queue.get()
            result = self.function(*data)
            self.out_queue.put(result)
            self.in_queue.task_done()

def process(data, function, num_workers=1):
    in_queue = Queue()
    for item in data: 
        in_queue.put(item)

    out_queue = Queue(maxsize=in_queue.qsize())
    workers = [Worker(function, in_queue, out_queue) for i in xrange(num_workers)]

    for worker in workers: 
        worker.start()

    in_queue.join()

    while not out_queue.empty():
        yield out_queue.get()

I recommend looking into the multiprocessing.pool.ThreadPool object, also explained here. — Theuni
– Theuni, Commented Dec 7, 2012 at 21:03

Gareth Rees · Accepted Answer · 2016-09-26 14:07:52Z

The function concurrent.futures.ThreadPoolExecutor.map is built into Python 3 and does almost the same thing as the code in the post. If you're still using Python 2, then there's a backport of the concurrent.futures package on PyPI.
But considered as an exercise, the code here seems basically fine (apart from the lack of documentation), and the remaining comments are minor issues.
Worker.run could be simplified slightly by writing the loop condition like this:
```
while not self.in_queue.empty():
    # ...
```
There's no need to pass a maxsize argument to the Queue constructor—the default is that the queue is unbounded, which is fine.
The only use of the workers list is to start the workers. But it would be easier to start each one as you create it:
```
for _ in range(num_workers):
    Worker(function, in_queue, out_queue).start()
```
The results come out in random(-ish) order because threads are nondeterministic. But for many use cases you would like to know which input corresponds to which output, and so you'd like the outputs to be in the same order as the inputs (as they are in the case of concurrent.futures.ThreadPoolExecutor.map).

Stack Exchange Network

Custom thread-pooling

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Custom thread-pooling

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions