Threaded stoppable function

Zamiast podklasowania threading.Thread, można zmodyfikować funkcję, aby umożliwić zatrzymuję się przy fladze.

Potrzebujemy obiektu, dostępnego dla uruchomionej funkcji, dla którego ustawiamy flagę, aby przestała działać.

Możemy użyć threading.currentThread() obiektu.

import threading
import time


def doit(arg):
    t = threading.currentThread()
    while getattr(t, "do_run", True):
        print ("working on %s" % arg)
        time.sleep(1)
    print("Stopping as you wish.")


def main():
    t = threading.Thread(target=doit, args=("task",))
    t.start()
    time.sleep(5)
    t.do_run = False
    t.join()

if __name__ == "__main__":
    main()

Sztuczka polega na tym, że uruchomiony wątek może mieć dołączone dodatkowe właściwości. Rozwiązanie buduje na podstawie założeń:

• wątek ma właściwość "do_run" z wartością domyślną True
• proces macierzysty może przypisać do rozpoczętego wątku właściwość "do_run" do False.

Uruchamiając kod, otrzymujemy następujące wyjście:

$ python stopthread.py                                                        
working on task
working on task
working on task
working on task
working on task
Stopping as you wish.

Pill to kill-using Event

Inną alternatywą jest użycie threading.Event jako argumentu funkcji. Jest przez domyślna False, ale zewnętrzny proces może ją "ustawić" (na True) i funkcja może dowiedz się o tym za pomocą funkcji wait(timeout).

Możemy wait z zerowym timeoutem, ale możemy również użyć go jako timera uśpienia (użyte poniżej).

def doit(stop_event, arg):
    while not stop_event.wait(1):
        print ("working on %s" % arg)
    print("Stopping as you wish.")


def main():
    pill2kill = threading.Event()
    t = threading.Thread(target=doit, args=(pill2kill, "task"))
    t.start()
    time.sleep(5)
    pill2kill.set()
    t.join()

Edit: próbowałem tego w Pythonie 3.6. stop_event.wait() blokuje Zdarzenie (a więc pętlę while) do momentu zwolnienia. Nie zwraca wartości logicznej. Zamiast tego działa stop_event.is_set().

Zatrzymanie wielu wątków za pomocą jednej pigułki

Przewaga pigułki do zabijania jest lepiej widoczna, jeśli musimy zatrzymać wiele wątków na raz, jak jedna tabletka będzie działać dla wszystkich.

doit w ogóle się nie zmieni, tylko main obsługuje wątki nieco inaczej.

def main():
    pill2kill = threading.Event()
    tasks = ["task ONE", "task TWO", "task THREE"]

    def thread_gen(pill2kill, tasks):
        for task in tasks:
            t = threading.Thread(target=doit, args=(pill2kill, task))
            yield t

    threads = list(thread_gen(pill2kill, tasks))
    for thread in threads:
        thread.start()
    time.sleep(5)
    pill2kill.set()
    for thread in threads:
        thread.join()

Czytałem inne pytania na Stack, ale nadal byłem trochę zdezorientowany w komunikowaniu się między klasami. Oto jak podszedłem:

Używam listy do przechowywania wszystkich wątków w metodzie __init__ mojej klasy wxFrame: self.threads = []

Zgodnie z zaleceniami w Jak zatrzymać pętlę wątku w Pythonie? używam sygnału w mojej klasie wątku, który jest ustawiony na True podczas inicjalizacji klasy wątku.

class PingAssets(threading.Thread):
    def __init__(self, threadNum, asset, window):
        threading.Thread.__init__(self)
        self.threadNum = threadNum
        self.window = window
        self.asset = asset
        self.signal = True

    def run(self):
        while self.signal:
             do_stuff()
             sleep()

I mogę zatrzymać te wątki przez iterację nad moimi wątkami:

def OnStop(self, e):
        for t in self.threads:
            t.signal = False

Miałem inne podejście. Podklasowałem klasę wątku, a w konstruktorze utworzyłem obiekt Event. Następnie napisałem własną metodę join (), która najpierw ustawia to zdarzenie, a następnie wywołuje samą wersję rodzica.

Oto moja klasa, której używam do komunikacji z portami szeregowymi w aplikacji wxPython:

import wx, threading, serial, Events, Queue

class PumpThread(threading.Thread):

    def __init__ (self, port, queue, parent):
        super(PumpThread, self).__init__()
        self.port = port
        self.queue = queue
        self.parent = parent

        self.serial = serial.Serial()
        self.serial.port = self.port
        self.serial.timeout = 0.5
        self.serial.baudrate = 9600
        self.serial.parity = 'N'

        self.stopRequest = threading.Event()

    def run (self):
        try:
            self.serial.open()
        except Exception, ex:
            print ("[ERROR]\tUnable to open port {}".format(self.port))
            print ("[ERROR]\t{}\n\n{}".format(ex.message, ex.traceback))
            self.stopRequest.set()
        else:
            print ("[INFO]\tListening port {}".format(self.port))
            self.serial.write("FLOW?\r")

        while not self.stopRequest.isSet():
            msg = ''
            if not self.queue.empty():
                try:
                    command = self.queue.get()
                    self.serial.write(command)
                except Queue.Empty:
                    continue

            while self.serial.inWaiting():
                char = self.serial.read(1)
                if '\r' in char and len(msg) > 1:
                    char = ''
                    #~ print('[DATA]\t{}'.format(msg))
                    event = Events.PumpDataEvent(Events.SERIALRX, wx.ID_ANY, msg)
                    wx.PostEvent(self.parent, event)
                    msg = ''
                    break
                msg += char
        self.serial.close()

    def join (self, timeout=None):
        self.stopRequest.set()
        super(PumpThread, self).join(timeout)

    def SetPort (self, serial):
        self.serial = serial

    def Write (self, msg):
        if self.serial.is_open:
            self.queue.put(msg)
        else:
            print("[ERROR]\tPort {} is not open!".format(self.port))

    def Stop(self):
        if self.isAlive():
            self.join()

Kolejka służy do wysyłania wiadomości do portu, a główna pętla odbiera odpowiedzi z powrotem. Nie użyłem żadnego seryjnego.metoda readline (), ze względu na inny znak końca linii, i uznałem, że używanie klas io jest zbyt kłopotliwe.