2
class SThread(threading.Thread):
    '''worker поток'''
    def __init__(self, queue_in):
        threading.Thread.__init__(self)
        self.queue_in = queue_in

        self.setDaemon(True)
        self.start()

    def run(self):
        get_cmd, task_done = self.queue_in.get, self.queue_in.task_done

        while True:
            cmd = get_cmd()  # ожидать/получить команду

            try:  # выполнить и сохранить результат в cmd
                cmd['result'] = cmd['target'](*cmd.get('args', ()), **cmd.get('kwargs', {}))

            except Exception as ex:
                if cmd is None:  # выход
                    return
                else:
                    traceback.print_exc()
                    traceback.print_tb(ex.__traceback__, file=open('__err.log', 'a'))
                    threading.Thread(target=self.run).start()
            finally:
                task_done()

3 ответа 3

3

Чтобы сохранить информацию о текущем исключении в текущем потоке, можно вызывать exc_info = sys.exc_info(). Позже можно напечатать в привычном формате используя: traceback.print_exception(*exc_info). See How to print the full traceback without halting the program?


Оставляя за кадром разумность наследования от (list, threading.Thread, Attributes), обычно метод Thread.run() не перезапускается в случае исключения (поток просто умирает). Запустить ли новый поток c нуля в случае исключения, или же сбросить ли значения текущих переменных в self.run() в ручную и запустить код метода повторно (после поимки исключения) в простом цикле—зависит от конкретной задачи (как часто вы ожидаете исключения, насколько сложно состояние, сохраняемое в потоке).

Я вижу большое число переменных, которые ничего полезное не делают или пытаются реализовать уже существующую функциональность. Вместо наследования, лучше использовать делегирование к существующим объектам в таких случаях. Попробуйте сделать шаг назад и подумать что вы пытаетесь сделать в терминах конечного результата и какие уже существующие примитивы (из threading, concurrent.futures, других модулей) могут в этом помочь. Позже если вы много похожих задач решаете, можно общую функциональность в отдельные классы упаковать, но это преждевременное усложение создавать классы типа SThread, до того как реальная потребность продемонстрирована (повторением кода). И даже в этом случае, возможно лучшее разбиение на объекты, чем всё в один SThread класс побросать.

2

Насколько я понимаю тут есть 2 проблемы:

  1. Увидеть стектрейс, если что-то пошло не так.
  2. Как организовать перезапуск.

Увидеть стектрейс

Вы всегда можете его вывести самостоятельно.

import traceback
import logging
import sys

stack_tuple = traceback.extract_stack()
stack = stack_tuple[:-2]
tb = traceback.extract_tb(sys.exc_info()[2])
detail = traceback.format_exception_only(*sys.exc_info()[0:1])
logging.error('stack %s, tb %s, detail %s', stack, tb, detail)

Перезапуск

Если Вас устраивает ваша текущая схема, то вроде как она не плоха. Но, мне кажется, что при работе с несколькими потоками, удобнее схема, когда один из потоков является наблюдателем (supervisor). Он запускает остальные потоки (workers) и следит за их здоровьем и перезапускает при необходимости.

Кстати, говоря схемы презапусков, тоже могут быть разными, и ими проще управлять, когда есть supervisor.

0
1

Имеется такой пул. Вобщем то это работает, но изредка зависает. Вероятно здесь нужны какието блокировки, вопрос где и почему

import threading, queue, contextlib, tkinter
import time, random

class SThreadIOQueue:
    '''priority_DeQueue_in -> queue_out'''
    count = 0  # count -= 1 для de-queue для sort в PriorityQueue

    def __init__(self, queue_in: queue.PriorityQueue, queue_out: queue.Queue):
        self.queue_in = queue_in
        self.queue_out = queue_out

    def submit(self, target: callable, *args, **kwargs) -> None:
        '''выполнить target, последняя зашедшая, выполнится первой'''
        SThreadIOQueue.count -= 1  # отрицательные - как dequeue при sort PriorityQueue
        self.queue_in.put_nowait((SThreadIOQueue.count, dict(target=target, args=args, kwargs=kwargs)))

    def map(self, fn: callable, args: tuple) -> iter:
        '''выполнить target's и получить результаты'''
        e = 0
        for e, a in enumerate(args, start=1):
            self.submit(fn, a)

        for _ in range(e):
            yield self.queue_out.get()  # результат


class SThread(threading.Thread, SThreadIOQueue):
    '''worker поток Thread пула'''
    def __init__(self, queue_in: queue.PriorityQueue, queue_out: queue.Queue, pool=None):
        threading.Thread.__init__(self)
        SThreadIOQueue.__init__(self, queue_in, queue_out)
        self.setDaemon(True)

        self.task = {}  # текущая задача
        self.pool = pool
        # тут, тк RuntimeError: main thread is not in main loop
        self._timeout = SThreadExitTimeout.get()
        self._size_min = SThreadPoolSizeMin.get()

        self.start()

    def run(self) -> None:
        '''поток'''
        try:
            while self.pool.working:
                out = None
                try:  # получить задачу
                    _num, self.task = self.queue_in.get(timeout=self._timeout)
                    try:  # выполнить задачу
                        out = self.task['target'](*self.task['args'], **self.task['kwargs'])
                    finally:  # вернуть результат
                        self.queue_out.put_nowait(out)
                        self.queue_in.task_done()
                        self.task = {}
                except queue.Empty:  # таймаут
                    if len(self.pool.threads) > self._size_min:
                        return
        finally:  # выход потока
            self.pool.remove_thread(self)
            self.task = {}


class SThreadPool(SThreadIOQueue):
    '''threading.Thread пул'''
    def __init__(self, size=0, parent=None):
        SThreadIOQueue.__init__(self, queue_in=queue.PriorityQueue(), queue_out=queue.Queue())
        self.parent = parent
        self.working = True

        self.threads = []
        self.size = size

        for _ in range(self.size):
            self.add_thread()

        auto_size_SThreadPool(self)

    def new_thread(self) -> SThread:
        '''создать новый поток'''
        th = SThread(self.queue_in, self.queue_out, pool=self)
        print('>', th.name)
        return th

    def add_thread(self) -> None:
        '''сохранить поток'''
        self.threads.append(self.new_thread())
        self.size = len(self.threads)

    def remove_thread(self, th: SThread) -> None:
        '''удалить поток'''
        with contextlib.suppress(ValueError):
            self.threads.remove(th)
            self.size = len(self.threads)
            print('<', th.name)

    def close(self) -> None:
        self.working = False
        for _ in range(SThreadPoolSizeMax.get()):
            self.queue_in.put_nowait((0, None))


def auto_size_SThreadPool(pool: SThreadPool, th_count=0) -> None:
    '''создать новый поток, если есть очередь, недостигнут maxsize, и все потоки заняты'''
    qsize = pool.queue_in.qsize()
    if qsize and (pool.size <= SThreadPoolSizeMax.get()) and all(th.task for th in pool.threads):
        th_count = divmod(qsize, SThreadPoolAddMinQSize.get())[0]  # 1 поток на каждый MinQSize
        if not th_count:
            th_count = 1
        else:
            max_th = SThreadPooMaxAddThread.get()
            if th_count > max_th:
                th_count = max_th

        for _ in range(th_count):
            if len(pool.threads) < SThreadPoolSizeMax.get():  # maxsize
                pool.add_thread()  # создать
            else:
                break

    if pool.working:  # перезапуск auto_size_SThreadPool
        Tk.after(SThreadAutoSizeTimeOut.get(), auto_size_SThreadPool, pool)


def func(t):
    time.sleep(t)
    return t, threading.current_thread()


if __name__ == '__main__':
    Tk = tkinter.Tk()
    SThreadPoolSizeMin = tkinter.IntVar(value=0)  # pool min size
    SThreadPoolSizeMax = tkinter.IntVar(value=5)  # pool max size
    SThreadExitTimeout = tkinter.IntVar(value=1)  # таймаут выхода, бездействующих потоков
    SThreadPoolAddMinQSize = tkinter.IntVar(value=5)  # мин длина очереди, для добавления, более одного потока
    SThreadPooMaxAddThread = tkinter.IntVar(value=2)  # не добавлять более потоков, за раз
    SThreadAutoSizeTimeOut = tkinter.IntVar(value=1000)  # период перезапуска auto_size_SThreadPool

    POOL = SThreadPool()
    auto_size_SThreadPool(POOL)

    def test(args=(random.randrange(5) for _ in range(50))):
        for t in POOL.map(func, args):
            print(t)
    POOL.submit(test)
    Tk.mainloop()
1
  • Если вы имеете в виду NamedThread — то кажется, что да, только порядок несколько иной. Было бы хорошо просто сделать еще один класс —отнаследоваться от ` threading.Thread` — в который инкапсулировать логику упарвления потоками. И в SThread — запускать только сам супервизор, который запустит все остальное. Т.е. можно было бы ваш исходный класс разбить на 2. В одном бизнес-логика в другом логика работы с потоками. Но это так — ради красоты мира. 3 авг 2016 в 22:55

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.