проблема в многопоточном скрипте

[robisho]

Добрый день. Многопоточный скрипт в цикле обрабатывает mail_lst. Если запихать список длиннее 670 строк - получаю ошибку RuntimeError: can't start new thread

semaphore = threading.BoundedSemaphore(S_THREADS)
thr_lst = []
for index, mail in enumerate(mail_lst, 1):
    try:
        thr = threading.Thread(
            target=main,
            args=(mail, index, semaphore),
            daemon=True
        )
        thr_lst.append(thr)
        thr.start()
    except Exception as ex:
        logger.exception(f"Index={index} >>> {ex}")
for i_thr in thr_lst:
    i_thr.join(10.0)

подскажите, пожалуйста, почему это происходит и как это пофиксить. спасибо.

 

[regnor]

что в S_THREADS ?
возможно, запускаете слишком много потоков, существует ограничение потоков для одного процесса

 

[robisho]

что в S_THREADS ?
возможно, запускаете слишком много потоков, существует ограничение потоков для одного процесса

это количество потоков, выбирается вручную. ввожу от 5 до 30. если список короткий - всё работает, а с длинным списком проблема.

 

[regnor]

это количество потоков, выбирается вручную. ввожу от 5 до 30. если список короткий - всё работает, а с длинным списком проблема.

там семафор - то есть это кол во работащих потоков, а активные потоки все, которые вы запускаете
то есть в списке 670 элементов, запускается 670 потоков, но работают только то количество, которое в S_THREADS, остальные ждут

ну очевидно же, что проблема в количестве потоков, маленький список - мало потоков, большой список - много потоков...
посмотрите активные потоки на каждой итерации

как это пофиксить. спасибо.

делить список и запускать по 100, 200 потоков
или использовать asyncio

 

[robisho]

разбил список на части

semaphore = threading.BoundedSemaphore(S_THREADS)
temp_mail_lst = []
for index, mail in enumerate(mail_lst, 1):
    temp_mail_lst.append((index, mail))
    if len(temp_mail_lst) == 100 or (index == len(mail_lst)):
        print('Length Temp List', len(temp_mail_lst))
        thr_lst = []
        for element in temp_mail_lst:
            index, mail = element[0], element[1]
            thr = threading.Thread(
                target=main,
                args=(mail, index, semaphore),
                daemon=True
            )
            thr_lst.append(thr)
            thr.start()
        for i_thr in thr_lst:
            i_thr.join(10.0)
        temp_mail_lst = []

первая часть в 100 элементов обрабатывается нормально, но потом начинаются чудеса. например, обрабатывается вторая часть, т.е. скрипт сообщает, что "Length Temp List 100" второй раз, но обработка подтормаживает, как будто работает в один поток, хотя количество потоков выбрано 10. в консоль выводится номер строки из общего списка mail_lst, ожидаем, что следующее сообщение "Length Temp List 100" будет выведено после 200го номера строки списка, но это сообщение вывелось после 176 номера строки. как то вышло, что скрипт набрал следующую партию строк не закончив предыдущую часть... но обрабатывается после 176 строки всё-же 177 и т.д. и очень медленно.

решил выбросить семафор нафиг.

temp_mail_lst = []
for index, mail in enumerate(mail_lst, 1):
    temp_mail_lst.append((index, mail))
    if len(temp_mail_lst) == S_THREADS or (index == len(mail_lst)):
        print('Length Temp List', len(temp_mail_lst))
        thr_lst = []
        for element in temp_mail_lst:
            index, mail = element[0], element[1]
            thr = threading.Thread(
                target=main,
                args=(mail, index),
                daemon=True
            )
            thr_lst.append(thr)
            thr.start()
        for i_thr in thr_lst:
            i_thr.join(10.0)
        temp_mail_lst = []

теперь никаких задержек, сразу набирается пулл, а потом выводится результат, если он есть. но я хотел бы понять, почему с семафором не получилось. может кто подскажет? спасибо.