Python enqueue tasks and get results in order

Question

I would like to have multiple threads performing tasks, but I would also like to get the results in order.

Take a simple sample code:

from threading import Thread
import queue
import time


class TaskQueue(queue.Queue):
    def __init__(self, num_workers=1):
        queue.Queue.__init__(self)
        self.num_workers = num_workers
        self.start_workers()

    def add_task(self, task, *args, **kwargs):
        args = args or ()
        kwargs = kwargs or {}
        self.put((task, args, kwargs))

    def start_workers(self):
        for i in range(self.num_workers):
            t = Thread(target=self.worker)
            t.daemon = True
            t.start()

    def worker(self):
        while True:
            ##tupl = self.get()  # REMOVED
            item, args, kwargs = self.get()
            item(*args, **kwargs)
            self.task_done()


def task(task_num, sleep_time):
    time.sleep(sleep_time)
    print("Task #{} sleeping {}".format(task_num, sleep_time))

q = TaskQueue(num_workers=2)

for t, s in zip([1,2,3,4,5,6,7,8,9], [9,8,7,6,5,4,3,2,1]):
    q.add_task(task, t, s)

q.join()  # block until all tasks are done
print("All Done!!")

Where I am adding tasks, with associated task number and each taking different execution time (sleeps).

I have three issues/questions.

~~1) I am not even getting all the outputs (without even considering the order). Currently I am just getting the output:~~

Task #4 sleeping 6 Task #2 sleeping 8 Task #6 sleeping 4 Task #8 sleeping 2

Seems I am not getting the odd tasks, maybe everything coming from the other worker. Why is that and how can I get them?

The program just hangs afterwards. I am assuming since the worker blocks until it gets something from the queue. And if the queue is empty, just waits forever. How can I update it so it will quit or hit "All Done!!" once there is no more tasks in the queue.

How can I have it print the tasks in order? Basically I want the results to be:

Task #1 sleeping 9
Task #2 sleeping 8
Task #3 sleeping 7
Task #4 sleeping 6
Task #5 sleeping 5
Task #6 sleeping 4
Task #7 sleeping 3
Task #8 sleeping 2
Task #9 sleeping 1

Also assume the task results are quite large and the number of tasks itself is a lot, thus I dont really want to have them all saved in memory then do some ordering. I should know the number of tasks added into the queue, and would just like to utilize those on what to print first. Saving some in memory temporarily is acceptable. I know in the current example you kind of have to save some first, since the first task takes the longest. You can assume that the execution time (or sleep in this case) will be random per tasks.

Currently using Python 3.7

---EDIT---

Removing tupl = self.get() from the above code solved question #1 and #2. Thus only question #3 is remaining. Any ideas/solution are welcome

user_1179317 · Answer

Here's the answer to my question. Using two queues (for the jobs and for the results). Answers are extracted from the results queue and saved in a dictionary. They are printed out in order and deleted accordingly.

Taking 23 seconds for 2 workers, where 1 worker or just synchronous execution takes 45 seconds:

from threading import Thread
import queue
import time
import datetime

class TaskQueue():
    def __init__(self, num_workers=1):
        self.num_workers = num_workers
        self.total_num_jobs = 0
        self.jobs_completed = 0
        self.answers_sent = 0
        self.jobs = queue.Queue()
        self.results = queue.Queue()
        self.start_workers()

    def add_task(self, task, *args, **kwargs):
        args = args or ()
        kwargs = kwargs or {}
        self.total_num_jobs += 1
        self.jobs.put((task, args, kwargs))

    def start_workers(self):
        for i in range(self.num_workers):
            t = Thread(target=self.worker)
            t.daemon = True
            t.start()

    def worker(self):
        while True:
            item, args, kwargs = self.jobs.get()
            item(*args, **kwargs)
            self.jobs_completed += 1
            self.jobs.task_done()

    def get_answers(self):
        while self.answers_sent < self.total_num_jobs or self.jobs_completed == 0:
            yield self.results.get()
            self.answers_sent += 1
            self.results.task_done()


def task(task_num, sleep_time, q):
    time.sleep(sleep_time)
    ans = "Task #{} sleeping {}".format(task_num, sleep_time)
    q.put((task_num, ans))


if __name__ == "__main__":
    start = datetime.datetime.now()
    h = TaskQueue(num_workers=2)
    q = h.results
    answers = {}
    curr_task = 1

    for t, s in zip([1,2,3,4,5,6,7,8,9], [9,8,7,6,5,4,3,2,1]):
        h.add_task(task, t, s, q)

    for task_num, ans in h.get_answers():
        answers[task_num] = ans
        if curr_task in answers:
            print(answers[curr_task])
            del answers[curr_task]
            curr_task += 1

    # Print remaining items (if any)
    for k, v in sorted(answers.items()):
        print(v)

    h.jobs.join()  # block until all tasks are done

    print("All done")
    print("Total Execution: {}".format(datetime.datetime.now() - start))

Python enqueue tasks and get results in order

1 Answers