python concurrent.futures пропускает процессы тайм-аута - PullRequest
Новая
       61

python concurrent.futures пропускает процессы тайм-аута

0 голосов
/ 02 февраля 2020

Я имею дело с тысячами URL-адресов изображений и хочу использовать concurrent.futures.ProcessPoolExecutor для ускорения.

Поскольку некоторые URL-адреса повреждены или изображения большие, функция процесса может зависнуть или неожиданно потреблять много времени на обработку. Я хочу добавить тайм-аут в функцию процесса, например, 10 секунд, чтобы избавиться от этих недопустимых изображений.

Я попытался установить параметр timeout в futures .as_completed, значение TimeoutException может быть успешно повышено. Тем не менее, кажется, что основной процесс все еще будет ожидать завершения дочернего процесса тайм-аута. Есть ли какой-либо подход к немедленному завершению дочернего процесса тайм-аута и добавлению следующего URL-адреса в пул? 

from concurrent import futures

def process(url):
    ### Some time consuming operation
    return result


def main():
    urls = ['url1','url2','url3',...,'url100']
    with futures.ProcessPoolExecutor(max_workers=10) as executor:
        future_list = {executor.submit(process, url):url for url in urls}
        results = []
        try:
            for future in futures.as_completed(future_list, timeout=10):
                results.append(future.result())
        except futures._base.TimeoutException:
            print("timeout")
    print(results)
if __name__ == '__main__':
    main()

В приведенном выше примере предположим, что у меня есть 100 URL-адресов, 10 из них недействительны и могут стоить много время, как получить список обработанных результатов остальных 90 URL?

1 Ответ

0 голосов
/ 03 февраля 2020

Не с библиотекой concurrent.futures.

Модуль pebble был разработан для преодоления такого ограничения. 

from pebble import ProcessPool
from concurrent.futures import TimeoutError

with process.ProcessPool() as pool:
    future = pool.schedule(function, args=(1,2), timeout=5)

    try:
        result = future.result()  # blocks until results are ready
    except TimeoutError as error:
        print("Function took longer than %d seconds" % error.args[1])
...