Многопоточный прогрессбар

Question

Пишу консольный многопоточный чекер для пркси, используя встроеный в python threading Раньше код выглядел так:

# Number of threads = Number of proxies
t_list = generate_threads(proxies, checker=checker, goods=goods)
for t in t_list:
    t.start()

for t in t_list:
    t.join()  # Waiting for process end checking
save_goods(goods, filename=goods_dest)

Но так как прокси много ~300 штук проверка идет порядка 30 - 40 секунд. И неплохо было бы запилить прогрессбар, который показывал бы состояние проверки.

Я решил использовать библиотеку tqdm. Переписал код так:

# Number of threads = Number of proxies
t_list = generate_threads(proxies, checker=checker, goods=goods)
for t in t_list:
    t.start()

for t in tqdm(t_list):
    t.join()  # Waiting for process end checking
save_goods(goods, filename=goods_dest)

Но у этого похода есть проблема. Т.к. потоки идут по очереди, и главный поток ждёт пока завершится следующий, то получается что если первый поток "долгий", то прогрессбар долго остаётся на 0, так как ждёт пока он выполнится. Так же я пытался вынести прогрессбар в отдельный процесс, но у меня ничего не получилось.

Answer 1

Тогда нужно проверять, какие потоки уже завершились с помощью метода is_alive(), прежде чем делать им join(). Мне кажется, можно сделать как-то так (код не проверял):

def get_finished_threads(threads):
    t_copy = threads.copy()
    while t_copy:
        finished = filter(lambda x: not x.is_alive(), t_copy)
        if not finished:
            time.sleep(0.1) # чтобы не съесть всё процессорное время на проверки
        for t in finished:
            t_copy.remove(t)
            yield t

# ...
for t in tqdm(get_finished_threads(t_list), total=len(t_list)):
    t.join()  # Waiting for process end checking

Answer 2

Можно воспользоваться явным изменением индикатора прогресса. Нужно завернуть функцию потока в обертку, которая по окончанию работы функции явно изменит индикатор:

from functools import wraps

lock = threading.Lock()

def execute_and_update_progress(f, progress):
    def wrapped(*args, **kwargs):
        try:
          return f(*args, **kwargs)
        finally: 
            with lock:  
                progress.update(1)
    return wrapped

# В предположении что checker это функция, которую выполняют потоки
with tqdm(total=len(proxies)) as progress:
    t_list = generate_threads(
        proxies, checker=execute_and_update_progress(checker, progress), goods=goods)
    for t in t_list:
        t.start()

    for t in t_list:
        t.join()

save_goods(goods, filename=goods_dest)

Второй вариант, это вместо явного создания потоков воспользоваться Executor-ами из concurrent.futures:

with ThreadPoolExecutor(max_workers=len(proxy)) as executor:
  lock = threading.Lock()
  futures = []
  with tqdm(total=len(proxies)) as progress:
    def update_progress():
       with lock:
         progress.update(1)
    for proxy in proxies: 
       # тут добавляем задачу на выполнение на proxy
       # нужно передать правильные параметры в функцию потока
       future = executor.submit(checker, param1, param2)
       future.add_done_callback(update_progress)
       futures.add(future)
    concurrent.futures.wait(futures)

БЛОГ НА HUSL

Многопоточный прогрессбар

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