Discutir sobre acelerar Shutil.copytree

Escribe aquí

Esta es una discusión sobre , consulte: https://discuss.python.org/t/speed-up-shutil-copytree/62078. Si tienes alguna idea, envíamela por favor.

Fondo

shutil es un módulo muy útil en Python. Puedes encontrarlo en github: https://github.com/python/cpython/blob/master/Lib/shutil.py

shutil.copytree es una función que copia una carpeta a otra carpeta.

En esta función, llama a la función _copytree para copiar.

¿Qué hace _copytree?

1. Ignorando archivos/directorios especificados.
2. Creando directorios de destino.
3. Copiar archivos o directorios mientras se manejan enlaces simbólicos.
4. Recopilar y generar errores encontrados (por ejemplo, problemas de permisos).
5. Replicando metadatos del directorio de origen al directorio de destino.

Problemas

La velocidad de _copytree no es muy rápida cuando la cantidad de archivos es grande o el tamaño del archivo es grande.

Prueba aquí:

import os
import shutil

os.mkdir('test')
os.mkdir('test/source')

def bench_mark(func, *args):
    import time
    start = time.time()
    func(*args)
    end = time.time()
    print(f'{func.__name__} takes {end - start} seconds')
    return end - start

# write in 3000 files
def write_in_5000_files():
    for i in range(5000):
        with open(f'test/source/{i}.txt', 'w') as f:
            f.write('Hello World'   os.urandom(24).hex())
            f.close()

bench_mark(write_in_5000_files)

def copy():
    shutil.copytree('test/source', 'test/destination')

bench_mark(copy)

El resultado es:

write_in_5000_files tarda 4,084963083267212 segundos
la copia tarda 27,12768316268921 segundos

lo que hice

subprocesos múltiples

Utilizo subprocesos múltiples para acelerar el proceso de copia. Y le cambio el nombre a la función _copytree_single_threaded y agrego una nueva función _copytree_multithreaded. Aquí está el copytree_multithreaded:

def _copytree_multithreaded(src, dst, symlinks=False, ignore=None, copy_function=shutil.copy2,
                            ignore_dangling_symlinks=False, dirs_exist_ok=False, max_workers=4):
    """Recursively copy a directory tree using multiple threads."""
    sys.audit("shutil.copytree", src, dst)

    # get the entries to copy
    entries = list(os.scandir(src))

    # make the pool
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # submit the tasks
        futures = [
            executor.submit(_copytree_single_threaded, entries=[entry], src=src, dst=dst,
                            symlinks=symlinks, ignore=ignore, copy_function=copy_function,
                            ignore_dangling_symlinks=ignore_dangling_symlinks,
                            dirs_exist_ok=dirs_exist_ok)
            for entry in entries
        ]

        # wait for the tasks
        for future in as_completed(futures):
            try:
                future.result()
            except Exception as e:
                print(f"Failed to copy: {e}")
                raise

Agrego un criterio para elegir usar multiproceso o no.

if len(entries) >= 100 or sum(os.path.getsize(entry.path) for entry in entries) >= 100*1024*1024:
        # multithreaded version
        return _copytree_multithreaded(src, dst, symlinks=symlinks, ignore=ignore,
                                        copy_function=copy_function,
                                        ignore_dangling_symlinks=ignore_dangling_symlinks,
                                        dirs_exist_ok=dirs_exist_ok)

else:
    # single threaded version
    return _copytree_single_threaded(entries=entries, src=src, dst=dst,
                                        symlinks=symlinks, ignore=ignore,
                                        copy_function=copy_function,
                                        ignore_dangling_symlinks=ignore_dangling_symlinks,
                                        dirs_exist_ok=dirs_exist_ok)

Prueba

Escribo 50000 archivos en la carpeta de origen. Punto de referencia:

def bench_mark(func, *args):
    import time
    start = time.perf_counter()
    func(*args)
    end = time.perf_counter()
    print(f"{func.__name__} costs {end - start}s")

Escribir:

import os
os.mkdir("Test")
os.mkdir("Test/source")

# write in 50000 files
def write_in_file():
    for i in range(50000):
         with open(f"Test/source/{i}.txt", 'w') as f:
             f.write(f"{i}")
             f.close()

Dos comparando:

def copy1():
    import shutil
    shutil.copytree('test/source', 'test/destination1')

def copy2():
    import my_shutil
    my_shutil.copytree('test/source', 'test/destination2')

• "my_shutil" es mi versión modificada de Shutil.

copia1 cuesta 173.04780609999943s
copiar2 cuesta 155.81321870000102s

copiar2 es mucho más rápido que copiar1. Puedes correr muchas veces.

Ventajas y desventajas

El uso de subprocesos múltiples puede acelerar el proceso de copia. Pero aumentará el uso de la memoria. Pero no necesitamos reescribir el subproceso múltiple en el código.

asíncrono

Gracias a "Barry Scott". Seguiré su sugerencia :

Es posible obtener la misma mejora con menos gastos generales utilizando E/S asíncrona.

Escribo este código:

import os
import shutil
import asyncio
from concurrent.futures import ThreadPoolExecutor
import time


# create directory
def create_target_directory(dst):
    os.makedirs(dst, exist_ok=True)

# copy 1 file
async def copy_file_async(src, dst):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, shutil.copy2, src, dst)

# copy directory
async def copy_directory_async(src, dst, symlinks=False, ignore=None, dirs_exist_ok=False):
    entries = os.scandir(src)
    create_target_directory(dst)

    tasks = []
    for entry in entries:
        src_path = entry.path
        dst_path = os.path.join(dst, entry.name)

        if entry.is_dir(follow_symlinks=not symlinks):
            tasks.append(copy_directory_async(src_path, dst_path, symlinks, ignore, dirs_exist_ok))
        else:
            tasks.append(copy_file_async(src_path, dst_path))

    await asyncio.gather(*tasks)
# choose copy method
def choose_copy_method(entries, src, dst, **kwargs):
    if len(entries) >= 100 or sum(os.path.getsize(entry.path) for entry in entries) >= 100 * 1024 * 1024:
        # async version
        asyncio.run(copy_directory_async(src, dst, **kwargs))
    else:
        # single thread version
        shutil.copytree(src, dst, **kwargs)
# test function
def bench_mark(func, *args):
    start = time.perf_counter()
    func(*args)
    end = time.perf_counter()
    print(f"{func.__name__} costs {end - start:.2f}s")

# write in 50000 files
def write_in_50000_files():
    for i in range(50000):
        with open(f"Test/source/{i}.txt", 'w') as f:
            f.write(f"{i}")

def main():
    os.makedirs('Test/source', exist_ok=True)
    write_in_50000_files()

    # 单线程复制
    def copy1():
        shutil.copytree('Test/source', 'Test/destination1')

    def copy2():
        shutil.copytree('Test/source', 'Test/destination2')

    # async
    def copy3():
        entries = list(os.scandir('Test/source'))
        choose_copy_method(entries, 'Test/source', 'Test/destination3')

    bench_mark(copy1)
    bench_mark(copy2)
    bench_mark(copy3)

    shutil.rmtree('Test')

if __name__ == "__main__":
    main()

Producción:

copia1 cuesta 187,21s
copiar2 cuesta 244,33s
copiar3 cuesta 111,27s

Puedes ver que la versión asíncrona es más rápida que la versión de un solo hilo. Pero la versión de un solo subproceso es más rápida que la versión de múltiples subprocesos. (Tal vez mi entorno de prueba no sea muy bueno, puedes intentar enviarme tu resultado como respuesta)

¡Gracias Barry Scott!

Ventajas y desventajas

Async es una buena opción. Pero ninguna solución es perfecta. Si encuentra algún problema, puede enviarme como respuesta.

Fin

Esta es la primera vez que escribo una discusión en python.org. Si hay algún problema, hágamelo saber. Gracias.

Mi Github: https://github.com/mengqinyuan
Mi desarrollador: https://dev.to/mengqinyuan