Блокировать основной поток, пока фоновый поток Python не завершит побочную задачу

Question

У меня есть потоковое приложение на Python с длительным циклом mainloop в фоновом потоке Этот фоновый mainloop на самом деле является вызовом pyglet.app.run () , который управляет окном GUI, а также может быть настроен на периодический вызов другого кода. Мне нужно, чтобы функция do_stuff(duration) вызывалась по желанию из основного потока для запуска анимации в графическом интерфейсе, ожидания ее остановки и возврата. Фактическая анимация должна выполняться в фоновом потоке, поскольку библиотека GUI не может обрабатывать управление отдельными потоками.

Я считаю, что мне нужно сделать что-то вроде этого:

import threading

class StuffDoer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.max_n_times = 0
        self.total_n_times = 0
        self.paused_ev = threading.Event()

    def run(self):
        # this part is outside of my control
        while True:
            self._do_stuff()
            # do other stuff

    def _do_stuff(self):
        # this part is under my control
        if self.paused_ev.is_set():
            if self.max_n_times > self.total_n_times:
                self.paused_ev.clear()
        else:
            if self.total_n_times >= self.max_n_times:
                self.paused_ev.set()
        if not self.paused_ev.is_set():
            # do stuff that must execute in the background thread
            self.total_n_times += 1

sd = StuffDoer()
sd.start()

def do_stuff(n_times):
    sd.max_n_times += n_times
    sd.paused_ev.wait_for_clear()   # wait_for_clear() does not exist
    sd.paused_ev.wait()
    assert (sd.total_n_times == sd.max_n_times)

РЕДАКТИРОВАТЬ: используйте max_n_times вместо stop_time, чтобы выяснить, почему Thread.join(duration) не сработает.

Из документации по threading.Event :

ожидание ([время])

Блокировать, пока внутренний флаг не станет истинным. Если внутренний флаг имеет значение true при входе, Вернись немедленно. В противном случае заблокировать пока другой поток не вызовет set () для установите флаг в true или пока необязательный тайм-аут.

Я обнаружил, что могу получить искомое поведение, если у меня есть пара событий paused_ev и not_paused_ev и использование not_paused_ev.wait(). Я мог бы почти просто использовать Thread.join(duration), за исключением того, что он должен возвращать точно только тогда, когда фоновый поток на самом деле регистрирует, что время истекло. Есть ли какой-то другой объект синхронизации или другая стратегия, которую я должен использовать вместо этого?

Я также был бы открыт для аргументов, что я все это неправильно рассматриваю, при условии, что они хорошие аргументы.

Answer 1

Надеюсь, что я получу некоторую ревизию или дополнительную информацию из моего комментария, но мне интересно, не переборщаете ли вы вещи, используя подкласс Thread. Вы можете делать такие вещи:

class MyWorker(object):
  def __init__(self):
    t = Thread(target = self._do_work, name "Worker Owned Thread")

    t.daemon = True

    t.start()

  def _do_work(self):
    While True:
      # Something going on here, forever if necessary.  This thread
      # will go away if the other non-daemon threads terminate, possibly
      # raising an exception depending this function's body.

Я считаю, что это имеет больше смысла, когда метод, который вы хотите запустить, является чем-то более подходящим для функции-члена некоторого другого класса, чем для метода run в потоке. Кроме того, это избавляет вас от необходимости инкапсулировать кучу бизнес-логики внутри потока. Все ИМО, конечно.

Answer 2

Я использовал очередь, аналогичную предложенной @wberry, и использовал Queue.task_done и Queue.wait:

import Queue
import threading

class StuffDoer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.setDaemon(True)
        self.max_n_times = 0
        self.total_n_times = 0
        self.do_queue = Queue.Queue()

    def run(self):
        # this part is outside of my control
        while True:
            self._do_stuff()
            # do other stuff

    def _do_stuff(self):
        # this part is under my control
        if self.total_n_times >= self.max_n_times:
            try:
                self.max_n_times += self.do_queue.get(block=False)
            except Queue.Empty, e:
                pass
        if self.max_n_times > self.total_n_times:
            # do stuff that must execute in the background thread
            self.total_n_times += 1
            if self.total_n_times >= self.max_n_times:
                self.do_queue.task_done()

sd = StuffDoer()
sd.start()

def do_stuff(n_times):
    sd.do_queue.put(n_times)
    sd.do_queue.join()
    assert (sd.total_n_times == sd.max_n_times)

Answer 3

Похоже, что ваш поток анимации GUI использует спин-блокировку в цикле while True. Это можно предотвратить, используя поточно-ориентированные очереди. Исходя из моего прочтения вашего вопроса, этот подход будет функционально эквивалентным и эффективным.

Я опускаю некоторые детали вашего кода выше, которые не изменились бы. Я также предполагаю, что метод run (), который вы не контролируете, использует значение self.stop_time для своей работы; в противном случае нет необходимости в поточной очереди.

from Queue import Queue
from threading import Event

class StuffDoer:
  def __init__(self, inq, ready):
    self.inq = inq
    self.ready = ready
  def _do_stuff(self):
    self.ready.set()
    self.stop_time = self.inq.get()

GUIqueue = Queue()
control = Event()

sd = StuffDoer(GUIqueue, control)

def do_stuff(duration):
  control.clear()
  GUIqueue.put(time.time() + duration)
  control.wait()

Answer 4

Я принял решение на основе @ gddc совета по этому вопросу.Вот мой код:

threads = []
# initializing aux thread(s) in the main thread ...
t = threading.Thread(target=ThreadF, args=(...))
#t.setDaemon(True) # I'm not sure does it really needed
t.start()
threads.append(t.ident)

# Block main thread
while filter(lambda thread: thread.ident in threads, threading.enumerate()):
    time.sleep(10)

Также вы можете использовать Thread.join для блокировки основного потока - это лучший способ.