Según lo que he leído, por ejemplo aquí , entiendo que las operaciones de E/S liberan el GIL. Por lo tanto, si tengo que leer una gran cantidad de archivos en el sistema de archivos local, entiendo que una ejecución con subprocesos debería acelerar las cosas.
Para probar esto, tengo una carpeta ( input ) con aproximadamente ~ 100k archivos, cada archivo tiene solo una línea con un número entero aleatorio. Tengo dos funciones: una "secuencial" y otra "concurrente" que solo suman todos los números
import glob import concurrent.futures ALL_FILES = glob.glob('./input/*.txt') def extract_num_from_file(fname): #time.sleep(0.1) with open(fname, 'r') as f: file_contents = int(f.read().strip()) return file_contents def seq_sum_map_based(): return sum(map(extract_num_from_file, ALL_FILES)) def conc_sum_map_based(): with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: return sum(executor.map(extract_num_from_file, ALL_FILES))Si bien ambas funciones me dan el mismo resultado, la versión "concurrente" es aproximadamente 3-4 veces más lenta.
In [2]: %timeit ss.seq_sum_map_based() 3.77 s ± 50.2 ms per loop (mean ± std. dev. of 7 runs, 1 loop each) In [3]: %timeit ss.conc_sum_map_based() 12.8 s ± 240 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)¿Hay algo mal con mi código o en mi entendimiento?
Nota: Lo siguiente solo se aplica a los HDD, que tienen partes móviles que pueden afectar el rendimiento de lectura, no a los SDD. La naturaleza de la gran diferencia de rendimiento me deja claro que se trata de un problema relacionado con el HDD, por lo que esta información opera bajo esa suposición.
El problema es que, si bien los subprocesos pueden operar en paralelo, los datos deben leerse secuencialmente desde el disco duro, ya que solo hay un cabezal de lectura singular. Sin embargo, lo peor es que, dado que ha paralelizado las operaciones de E/S, el sistema operativo subyacente programará estas tareas de E/S de modo que estos archivos se procesen solo parcialmente antes de cambiar a otro subproceso, después de todo, incluso si solo tiene un solo entero, los encabezados de los archivos también deben procesarse, lo que hace que la cabeza de lectura salte mucho más salvajemente que en su código estrictamente secuencial. Todo esto da como resultado una sobrecarga masivamente mayor en comparación con simplemente leer la totalidad de cada archivo en secuencia, lo que no requiere tantos saltos.
Esto no sería un gran problema si, por ejemplo, tuviera un solo subproceso cargando grandes cantidades de datos del disco mientras que un segundo subproceso realiza un procesamiento intensivo del mismo, ya que eso permitiría el procesamiento intensivo del tiempo. para continuar desbloqueado por las operaciones de E/S. Su escenario particular es simplemente un caso muy, muy malo en el que ha renunciado a un cuello de botella GIL a cambio de un cuello de botella de E/S terriblemente lento.
En resumen, entendió correctamente que las operaciones de E/S liberan el GIL, simplemente llegó a una conclusión incorrecta sobre la paralelización de las lecturas de archivos.
La otra respuesta decía esto bastante bien:
ha renunciado a un cuello de botella GIL a cambio de un cuello de botella de E/S horriblemente lento. En resumen, entendió correctamente que las operaciones de E/S liberan el GIL, simplemente llegó a una conclusión incorrecta sobre la paralelización de las lecturas de archivos.
Agregaré que la lectura de archivos encadenados puede ser más eficaz si tiene E/S de sobra, como podría tener en un SSD muy rápido.
Probé esto en un SSD relativamente rápido (Samsung 970 EVO 1TB; disco secundario sin actividad) leyendo ~1000 archivos de varios tamaños, con un promedio de 8800 caracteres. En las pruebas, obtuve un mejor rendimiento con más subprocesos... pero los rendimientos decrecientes se activan rápidamente.
In [1]: len(ALL_FILES) 995 In [2]: %timeit ThreadedFileReader(ALL_FILES, n=1).join() # single threaded 61.8 ms ± 305 µs per loop (mean ± std. dev. of 7 runs, 10 loops each) In [3]: %timeit ThreadedFileReader(ALL_FILES, n=2).join() 54.7 ms ± 158 µs per loop (mean ± std. dev. of 7 runs, 10 loops each) In [4]: %timeit ThreadedFileReader(ALL_FILES, n=3).join() 56.1 ms ± 135 µs per loop (mean ± std. dev. of 7 runs, 10 loops each) In [5]: %timeit ThreadedFileReader(ALL_FILES, n=4).join() 57.8 ms ± 131 µs per loop (mean ± std. dev. of 7 runs, 10 loops each) In [6]: %timeit ThreadedFileReader(ALL_FILES, n=5).join() 58.9 ms ± 236 µs per loop (mean ± std. dev. of 7 runs, 10 loops each) In [7]: %timeit ThreadedFileReader(ALL_FILES, n=50).join() 68.6 ms ± 378 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)Entonces, su idea en principio es buena, pero solo si tiene suficiente E/S de sobra, en comparación con la lectura secuencial. A menos que tenga un almacenamiento muy rápido, es poco probable que necesite más de uno o dos subprocesos adicionales. Si su almacenamiento no es rápido en absoluto, el enfoque de subproceso único puede ser el camino a seguir.
Recuerde que si tiene muchos subprocesos que leen archivos al mismo tiempo, especialmente archivos pequeños, es probable que se vea obstaculizado por las capacidades de lectura aleatoria de su unidad. Comparativamente, es probable que un enfoque de subproceso único en archivos grandes se atasque más cerca de las capacidades de lectura secuencial de su unidad. Dependiendo del hardware, estas podrían ser clasificaciones de rendimiento sustancialmente diferentes.
Según el hardware y las características de los datos que esté leyendo, los beneficios del rendimiento de lectura secuencial pueden superar cualquier beneficio potencial de las lecturas en paralelo.
Para completar, el código que usé para probar esto está a continuación, aunque no tiene una relación particular con la respuesta.
class ThreadedFileReader: def __init__(self, files, n=5): self.files = deque(files) self.threads = [] self.results = queue.Queue() for _ in range(n): t = threading.Thread(target=self.worker) t.start() self.threads.append(t) def worker(self): while self.files: fname = self.files.pop() with open(fname, encoding='utf-8') as f: data = f.read() self.results.put(len(data)) return def join(self): for t in self.threads: t.join()