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

Question

[HexUserHex]

Приветствую,
пробую написать многопоточный парсер на python, на текущий момент если честно даже не уверен что выбранный мною модуль threading является именно тем что мне нужно на самом деле, буду благодарен за совет или помощь более опытных разработчиков.

Цель:
Запустить парсинг с заранее установленным количеством потоков, и поддерживать заданное количество потоков на всем протяжении выполнения скрипта. В качестве примера я планирую парсить данные с ресурса где id инкрементируется.

Сильно упрощенный пример:

import threading
import time

def hard_func(i):
        print("very big code")
        return 1

if __name__ == "__main__":

    max_thread = 4
    threads = list()
    start_time = time.perf_counter()


    for i in range(0, 100000):
        print('Threads: ', str(threading.active_count()), ' i: ', str(i))
    
        if threading.active_count() < max_thread:
            print("Сreate and start thread %d.", i)
            x = threading.Thread(target=hard_func, args=(i,))
            threads.append(x)
            x.start()

    for index, thread in enumerate(threads):
        thread.join()

    end_time = time.perf_counter()
    elapsed_time = end_time - start_time
    print("Elapsed time: ", elapsed_time)

На данный момент скрипт выполняется быстрее с одним потоком чем с 2я и более.

Comments

[Ичи Ня]

Как вариант можно запустить несколько экземпляров кода. Да, это больше ресурсов больше "сожрёт", процентов на 50. Но зато реально запускать столько потоков, сколько надо.

Другой вариант через callback и async всё делать. Там простая проверка завершен код или нет.

Кидаю код одного из проектов:

import threading
import time


class ThreadingController:
    _threads = {}
    __thread_max = 1
    __delay_works = []
    __start_time = 0
    __run_delay_works = False

    def __init__(self, thread_max: int = 1):
        self.set_max_thread(thread_max=thread_max)
        self.__start_time = time.monotonic()

    def add_work(self, coro):
        self.clean_threads()
        thread_id = self.find_free_thread()

        if thread_id == 0 and self.__run_delay_works:
            count = len([x for x in self._threads if self._threads[x] is not None])
            if count == 0:
                self.run_delay_works()
            else:
                return False

        if len(self._threads) >= self.__thread_max and thread_id == 0:
            return False
        if thread_id == 0:
            thread_id = len(self._threads) + 1
        print(self.__thread_max, len(self._threads), thread_id, self._threads)
        threading.Event()
        t1 = threading.Thread(target=coro, args=[thread_id], daemon=True, name=str(thread_id))
        t1.start()
        threading.Event()
        self._threads[thread_id] = t1
        # t1.join()

    def add_delay_work(self, coro, *args, **kwargs):
        for work in self.__delay_works:
            if work['coro'] == coro and work['args'] == args and work['kwargs'] == kwargs:
                return
        self.__delay_works.append({'coro': coro, 'args': args, 'kwargs': kwargs})

    def run_delay_works(self):
        from ldplayer import LDPlayer
        from utils import close_process
        from gui import GUI
        GUI.log('')
        if self.__thread_max > 1:
            GUI.log('Запуск отложенных задач')
        LDPlayer().close_all()
        close_process(['adb.exe'])
        close_process()
        for work in self.__delay_works:
            if work['kwargs'].get('title'):
                GUI.log(work['kwargs'].get('title'))
            try:
                work['coro'](*work['args'], **work['kwargs'])
            except:
                pass
        self.__delay_works = []
        self.__start_time = time.monotonic()
        self.__run_delay_works = False

    def clean_threads(self):
        for idx in self._threads:
            if isinstance(self._threads[idx], threading.Thread) and not self._threads[idx].is_alive():
                self._threads[idx].join()
                self._threads[idx] = None

    def find_free_thread(self) -> int:
        delay_run_time = self.__thread_max * 600
        if delay_run_time > 3600:
            delay_run_time = 3600
        if time.monotonic() - self.__start_time > delay_run_time:
            self.__run_delay_works = True
            return 0
        alive_num = 1
        for idx in self._threads:
            if alive_num == idx and self._threads[idx] is not None:
                alive_num = 0
            if self._threads[idx] is None:
                alive_num = idx
        print(alive_num, self._threads)
        return alive_num


__all__ = ['ThreadingController']

Суть довольно простая: создаем контроллер (указывая максимум). А потом просто цикле запускаем th.add_work(work, ). Если есть свободный поток, он добавится, если нет, то ничего не делается. Раз в час прекращаются запускаться потоки, чтобы запустить эксклюзивные задачи в один поток (у меня банально - закрытие всех приложений и удаление временных файлов).

clean_threads - если задача выполнилась, то она будет "освобождена" и удалена из списка

Answer 1

[Artemii]

Можно попробовать multiprocessing.
Из-за ограничений GIL, threading, по факту, на процессоре выполняется всё равно по очереди. Лучшее его применение - это использование для запросов к внешнему источнику, и не дожидаясь ответа он сразу начинает новый процесс.

Comments

[Александр]

Используйте multiprocessing pool например

[HexUserHex]

Justa Gain, там работа с dlib и БД