Simulación de uso de futures masivos

futur.py

# -*- coding: utf-8 -*-
"""
Created on Fri Jun 12 18:29:33 2015

@author: chema
"""

from concurrent.futures import (
    Future, ProcessPoolExecutor, ThreadPoolExecutor, TimeoutError, as_completed)
from itertools import islice
from os.path import getsize
from pathlib import Path
from random import random
import time
from typing import Generator, Iterable, List, TypeVar


def simula_timeout() -> int:
    # simulación de timeouts:
    # un 10% de las veces se produce una espera entre 1 y 9 segundos
    res = 0
    if random() > 0.9:
        espera = 10 * random()
        time.sleep(espera)
        res = int(espera)

    return res


def proc_file(root: Path) -> str:

    try:
        files = [f for f in root.glob("*") if f.is_file()]
        size = sum(getsize(f) for f in files)
        num = len(files)
        res = "*" * simula_timeout()
        res += f"{size:8} bytes en {num:3} ficheros de '{root}'"

    except:
        res = "ERROR"

    return res


T = TypeVar('T')

# Para evitar sobrecargar el proceso,
# el iterador se consume en chunks de `nitems` elementos
def chunks(it: Iterable[T], nitems=100) -> Iterable[List[T]]:
    while True:
        chunk = list(islice(it, nitems))
        if len(chunk) == 0:
            break
        yield chunk


def main():

    dir_input = "/home/chema/repos/SCM"

    # Elige uno de los dos tipos de ejecutores
    pool = ThreadPoolExecutor(40)
    #pool = ProcessPoolExecutor()

    with pool as ex:

        futures = (ex.submit(proc_file, root)
                   for root in Path(dir_input).glob("**/"))

        for (i, chunk) in enumerate(chunks(futures)):

            print(f"\n\n----------- CHUNK {i:2} -------------\n")

            try:
                for future in as_completed(chunk, timeout=8):
                    print(future.result())
            except TimeoutError:
                print("\n### ERROR: timeout exception\n")
                break


if __name__ == '__main__':
    main()