Как в одном потоке получить данные из других потоков?

Question

Вот мой код, который иллюстрирует проблему:

from threading import Thread
import queue
import time

q = queue.Queue()


class Worker(Thread):
    def __init__(self, name):
        Thread.__init__(self)
        self.name = name

    def run(self):
        if self.name == '0':
            update_tasks()
        else:
            while True:
                task = q.get()
                result = do_work(task)
                if not result:
                    q.put(task)
                
            
def update_tasks():
    while True:
        tasks = set(get_new_tasks()) - set(get_completed_tasks) - set(list(q.queue))
        for task in tasks:
            q.put(task)
        time.sleep(10)


def run_threads():
    threads = []
    for i in range(15):
        my_thread = Worker(str(i))
        my_thread.start()
        threads.append(my_thread)
    for thread in threads:
        thread.join()

run_threads()

Я запускаю 15 потоков. Из них 1 поток я отправляю пополнять количество задач функцией update_tasks() А остальные 14 должны решать эти задачи функцией do_work() Новые задачи поступают каждые 10 секунд. При этом, в свежеполученных задачах постоянно приходят те, что уже выполнены. Поэтому функция update_tasks() сравнивает свежеполученные задачи с теми, что уже сделаны, и теми, что находятся сейчас в очереди.

А теперь сама проблема: do_work() решает задачу примерно за 1 минуту и 14 задач, которые сейчас в работе снова поступают в список новых задач. Потому что они еще не появились в списке решенных задач. Но и в q.queue их уже нет. Эта конструкция их пропускает:

asks = set(get_new_tasks()) - set(get_completed_tasks) - set(list(q.queue))

Подскажите, пожалуйста, как можно одним потоком получить задачи, которые в это время выполняют другие потоки, для того чтобы избежать повторного выполнения одних и тех же задач?

Answer 1

исходя из вашего кода, зачем вообще проверять выполнение задач в воркерах, не проще ли делать это в update_tasks, она ведь и так знает что отправляла воркерам, а безрезультатные task добавляет сам воркер

def update_tasks():
    _tasks = set()  # ожидающие выполнения или уже выполненные task

    while True:
        tasks = get_new_tasks() - _tasks  # только новые task

        if tasks:
            _tasks.update(tasks)
            for task in tasks:
                q.put(task)

БЛОГ НА HUSL

Как в одном потоке получить данные из других потоков?

Ответы (1 шт):