مناقشة تسريع Shutil.copytree

اكتب هنا

هذه مناقشة حول، راجع: https://discuss.python.org/t/speed-up-shutil-copytree/62078. إذا كان لديك أي أفكار، أرسل لي من فضلك!

خلفية

shutil هو نموذج مفيد جدًا في لغة بايثون. يمكنك العثور عليه في جيثب: https://github.com/python/cpython/blob/master/Lib/shutil.py

shutil.copytree هي وظيفة تقوم بنسخ مجلد إلى مجلد آخر.

في هذه الوظيفة، يتم استدعاء وظيفة _copytree للنسخ.

ماذا تفعل _copytree؟

1. تجاهل الملفات/الأدلة المحددة.
2. إنشاء أدلة الوجهة.
3. نسخ الملفات أو الأدلة أثناء التعامل مع الروابط الرمزية.
4. جمع الأخطاء التي تمت مواجهتها وإثارتها في النهاية (على سبيل المثال، مشكلات الأذونات).
5. نسخ البيانات الوصفية للدليل المصدر إلى الدليل الوجهة.

مشاكل

_سرعة Copytree ليست سريعة جدًا عندما يكون عدد الملفات كبيرًا أو يكون حجم الملف كبيرًا.

الاختبار هنا:

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)

النتيجة هي:

تستغرق عملية الكتابة في_5000_files 4.084963083267212 ثانية
تستغرق النسخة 27.12768316268921 ثانية

ماذا فعلت

تعدد الخيوط

أستخدم مؤشرات ترابط متعددة لتسريع عملية النسخ. وأعدت تسمية الوظيفة _copytree_single_threaded وأضف وظيفة جديدة _copytree_multithreaded. هنا هو 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

أضيف حكمًا لاختيار استخدام مؤشرات ترابط متعددة أم لا.

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)

امتحان

أكتب 50000 ملف في المجلد المصدر. علامة المقعد:

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")

اكتب في:

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()

مقارنة بين:

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

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

• "my_shutil" هو نسختي المعدلة من Shutil.

تكاليف النسخ 1 173.04780609999943 ثانية
تكاليف النسخ 2 155.81321870000102 ثانية

copy2 أسرع من Copy1 كثيرًا. يمكنك تشغيل عدة مرات.

المزايا والعيوب

استخدام مؤشرات الترابط المتعددة يمكن أن يسرع عملية النسخ. لكنه سيزيد من استخدام الذاكرة. لكننا لسنا بحاجة إلى إعادة كتابة الخيوط المتعددة في الكود.

غير متزامن

شكرًا لـ "باري سكوت". سأتبع اقتراحه:

قد تحصل على نفس التحسين مع تقليل الحمل باستخدام الإدخال/الإخراج غير المتزامن.

أكتب هذا الكود:

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()

الإخراج:

تكلفة النسخ 1 187.21 ثانية
تكاليف النسخ 2 244.33 ثانية
تكاليف النسخ 3 111.27 ثانية

يمكنك أن ترى أن الإصدار غير المتزامن أسرع من إصدار مؤشر الترابط الفردي. لكن إصدار الخيط الفردي أسرع من الإصدار متعدد الخيوط. (ربما بيئة الاختبار الخاصة بي ليست جيدة جدًا، يمكنك محاولة إرسال النتيجة كرد لي)

شكرًا لك باري سكوت!

المزايا والعيوب

يعد المزامنة خيارًا جيدًا. لكن لا يوجد حل مثالي. إذا وجدت بعض المشاكل، يمكنك أن ترسل لي كرد.

نهاية

هذه هي المرة الأولى التي أكتب فيها مناقشة على python.org. إذا كان هناك أي مشكلة، واسمحوا لي أن أعرف. شكرًا لك.

موقعي على جيثب: https://github.com/mengqinyuan
موقع التطوير الخاص بي: https://dev.to/mengqinyuan