1

estoy intentando correr un codigo con hilos pero a la hora de correrlo como son muchos archivos me salta un error cannot create new thread por lo que e intenado limitar el numero de hilos a 100 o a 10. estube leyendo como limitarlos y encontre que con threading.BoundedSemaphore(10) podria hacerlo pero aun asi sigue sin funcionarme. alguien podria ayudarme o orientarme en alguna forma para poder limitar los hilos a un numero especifico

if __name__ == '__main__':

    threadLimiter = threading.BoundedSemaphore(100)
    threadLimiter.acquire()
    try:
        for dirName, subdirList, fileList in os.walk(location):
            for fname in fileList:
                file_name = os.path.join(dirName, fname)
                thread = threading.Thread(target=run, args=(file_name,))
                thread.start()
    finally:
        threadLimiter.release()

1 respuesta 1

0

Aparentemente, lo que buscas es crear un patrón productor/consumidor con 10 hilos consumidores. Podrías implementarlo usando colas por ejemplo, tienes un ejemplo simplificado en la propia documentación oficial de queue.Queue.

No obstante, creo que es mucho más simple no "reinventar la rueda" y usar concurrent.futures.ThreadPoolExecutor. Realmente se trata de una interfaz de alto nivel que se implementa mediante multiprocessing.dummy (el cual podríamos también usar directamente) que emula a multiprocessing.Pool pero usando hilos. Una de las ventajas es que prácticamente el mismo código podemos hacerlo funcionar con procesos usando concurrent.futures.ProcessPoolExecutor.

Un ejemplo basado en tu ejemplo y usando un máximo de diez hilos (workers):

import concurrent.futures
import os


LOCATION = "/home"

def run(file):
    return os.stat(file).st_size


if __name__ == '__main__':
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = {}
        for dir_name, _, file_list in os.walk(LOCATION):
            for file_name in file_list:
                file = os.path.join(dir_name, file_name)
                futures[executor.submit(run, file)] = file

        # Si run() retorna algo podemos obtener la salida tan pronto este procesada
        for future in concurrent.futures.as_completed(futures):
            file = futures[future]
            try:
                data = future.result()
            except Exception as exc:
                print(f'{file!r} generated an exception: {exc}')
            else:
                print(f'{file!r} ----> {data} bytes')

En este caso, a modo de ejemplo y sin sentido alguno, run se limita a obtener el tamaño de cada archivo.

Básicamente se crean un máximo de 10 hilos que van consumiendo cada ruta que se genera, cuando un hilo queda libre toma otra ruta si está disponible y ejecuta run con ella, así hasta consumirlas todas.

Si run no retorna nada o no te interesa, puedes prescindir del diccionario, simplemente haz:

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    for dir_name, _, file_list in os.walk(LOCATION):
        for file_name in file_list:
            executor.submit(run, os.path.join(dir_name, file_name))

o algo más críptico usando concurrent.futures.Executor.map:

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    file_gen = (os.path.join(dir_name, file_name)
                    for dir_name, _, file_list in os.walk(LOCATION)
                        for file_name in file_list
                )
    executor.map(run, file_gen)

concurrent.futures está disponible a partir de Python 3.2, no obstante fue portado a Python 2, de forma que podemos disponer del mismo en Python 2.7 si se diera el caso instalando el paquete futures desde PyPi, p.ej:

 $ python2 -m pip install futures

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge you have read our privacy policy.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.