Czy mogę wywoływać wątek cyklicznie z poziomu wątku?

Question

Próbuję przenieść próbki z wątku A („Pozyskiwanie”) do wątku B („P300”) przy użyciu, queueale nie mogę odczytać żadnych danych w wątku B, chociaż próbki są przydzielane w wątku A. Sądząc po moich wynikach, Myślę, że mój wątek B pędzi i testuje rzeczy, zanim mój wątek A zacznie wprowadzać dane.

Zobacz przybliżenie struktury mojego kodu poniżej:

import threading
import queue
from queue import Empty
import numpy as np
import warnings
warnings.filterwarnings("error")

class AcqThread(threading.Thread):
    def __init__(self, dataOutQ1, dataOutQ2, stopQ1, stopQ2, saveQ):
        threading.Thread.__init__(self)
        self.stopQ2 = stopQ2
        self.stopQ1 = stopQ1
        self.dataOutQ2 = dataOutQ2
        self.dataOutQ1 = dataOutQ1
        self.saveQ = saveQ

    def run(self):
        Acquisition(inlet, self.dataOutQ1, self.dataOutQ2, self.stopQ1, self.stopQ2, self.saveQ)

class P300Thread(threading.Thread):
    def __init__(self, dataInQ, featureQ, stopQ):
        threading.Thread.__init__(self)
        self.dataInQ = dataInQ
        self.featureQ = featureQ
        self.stopQ = stopQ

    def run(self):
        P300fun(self.dataInQ, self.featureQ, self.stopQ)

threadLock = threading.Lock()
SaveQ = queue.Queue()
DataOutQ1 = queue.Queue()
DataOutQ2 = queue.Queue()
StopQ1 = queue.Queue()
StopQ2 = queue.Queue()
FeatQ1 = queue.Queue()
StopQ1.put(0)
StopQ2.put(0)
#
AcqTh = AcqThread(DataOutQ1, DataOutQ2, StopQ1, StopQ2, SaveQ)
P300Th = P300Thread(DataOutQ1, FeatQ1, StopQ1)

def Acquisition(inlet, dataOutQ1, dataOutQ2, stopQ1, stopQ2, saveQ):
    i = 0
    print('Starting...')
    while i<1250: #i is the number of samples
        sample, timestamp = inlet.pull_sample() #samples coming in @ 250Hz
        ##Normalization, filtering##
        threadLock.acquire()
        dataOutQ1.put([filtsamples[:,-250:], rawtimestamps[-250:]]) #I only need the last 250 samples
        threadLock.release()
        i += 1

def P300fun(dataInQ, featureQ, stopQ):
    p300sample = []
    p300timestamp = []
    print(f"Is DataInQ size true? {DataOutQ1.qsize()}")
    print("Is dataInQ emtpy?", DataOutQ1.empty())
    while dataInQ.qsize(): #or while not dataqueue.empty():
        try:
            print("DataInQ has data")
            ss, ts = dataInQ.get(0) 
            print('<>P300\n>>Samples [', ss, ']\nTimestamp [', ts, ']')
        except Empty:
            return
    print('Thread Finished')

if __name__ == '__main__':
    print('Looking for an EEG stream...')
    streams = resolve_stream('type', 'EEG')
    inlet = StreamInlet(streams[0])
    print('Connected!\n')

    AcqTh.start()
    P300Th.start()

    AcqTh.join()
    P300Th.join()

    print("\n\n>>>DONE<<<\n\n")

I wyjście:

Looking for an EEG stream...
Connected!

Is DataInQ size true? 0
Starting...
Is dataInQ emtpy? True
Thread Finished

>>>DONE<<<

W moich badaniach pytanie 1 wydaje się przedstawiać podobny problem, ale wydaje się, że problem dotyczył części przetwarzania obrazu (i używają multiprocessingpakietu). Pytanie 2 wydaje się mieć problem ze współbieżnością, co może być moim problemem, ale nie jestem pewien, jak przetłumaczyć to na mój problem, daj mi znać, jeśli się mylę, chociaż). Pytanie 3 miało tylko problem z kolejnością argumentów, więc myślę, że nie ma tutaj zastosowania.

Jak mam się do tego zabrać? Czy powinienem cyklicznie wywoływać wątek B z poziomu wątku A? Czy potrzebuję pętli lub opóźnienia w wątku B? Czy jest jakiś problem z .join()częścią? Będę musiał dodać więcej wątków w najbliższej przyszłości, więc dobrze byłoby najpierw wymyślić, jak pracować tylko z dwoma ...

Wszelka pomoc jest wdzięczna!

Answer 1

Bycie noobem może być trudne ... Więc odpowiem na własne pytanie, aby pomóc innym początkującym, którzy również mogą napotkać ten problem.

Cóż, na początek: nie, nie jest możliwe powtarzające się wywoływanie wątku z poziomu wątku, ponieważ każdy wątek można wywołać tylko raz.

Istnieje jednak sposób, aby zapobiec zakończeniu wątku, zmuszając ich do czekania na wyzwalacze, które pozwolą im kontynuować. Po dalszych badaniach natknąłem się na to pytanie, które pokazało mi, że istnieje sposób na tworzenie wydarzeń dla wątków. Dokumentację można znaleźć tutaj . I jest to całkiem proste: obiekty zdarzeń zachowują się jak flagi i mogą być set()(wskazując True) lub clear()(wskazując False, co jest wartością oryginalną). Aby przetestować zdarzenie, można użyć is_set()metody dla problemów logicznych lub użyć wait()metody zamiast licznika czasu. W moim przypadku uratowało mi to kilka kolejek, z których miałem zamiar skorzystać:

import threading
import queue
from queue import Empty
import numpy as np


class AcqThread(threading.Thread):
    def __init__(self, dataOutQ1, dataOutQ2, saveQ):
        threading.Thread.__init__(self)
        self.dataOutQ2 = dataOutQ2
        self.dataOutQ1 = dataOutQ1
        self.saveQ = saveQ

    def run(self):
        Acquisition(inlet, self.dataOutQ1, self.dataOutQ2, self.saveQ)

class P300Thread(threading.Thread):
    def __init__(self, dataInQ, featureQ):
        threading.Thread.__init__(self)
        self.dataInQ = dataInQ
        self.featureQ = featureQ

    def run(self):
        P300fun(self.dataInQ, self.featureQ)

threadLock = threading.Lock()
SaveQ = queue.Queue()
DataOutQ1 = queue.Queue()
DataOutQ2 = queue.Queue()
FeatQ1 = queue.Queue()
FeatQ2 = queue.Queue()

#NEW:: initializes Events
E = threading.Event()
EP300 = threading.Event()
#
AcqTh = AcqThread(DataOutQ1, DataOutQ2, SaveQ)
P300Th = P300Thread(DataOutQ1, FeatQ1)

I pozwala mi "wywoływać" wątek B "cyklicznie", ponieważ zachowuje mój pierwszy, gdy jest aktywny (z powodu zdarzenia E) i wchodzi do części przetwarzania tylko wtedy, gdy zdarzenie EP300 jest ustawione. Następnie EP300 jest czyszczony po zakończeniu procesu:

def Acquisition(inlet, dataOutQ1, dataOutQ2 saveQ):
    i = 0
    print('Starting...')
    while i<1250:
        sample, timestamp = inlet.pull_sample()
        ##Normalization, filtering##
        if _condition_:
            threadLock.acquire()
            dataOutQ1.put([filtsamples[:,-250:], rawtimestamps[-250:]])
            threadLock.release()
            EP300.set() #NEW:: allows the P300 function to collect data from queue
        i += 1
    E.set() #NEW:: flaggs end data collection

def P300fun(dataInQ, featureQ):
    p300sample = []
    p300timestamp = []
    while not E.is_set(): #NEW:: loop until collection is ended
        if EP300.is_set(): #NEW:: activated when Event is triggered
            while dataInQ.qsize():
                try:
                    print("DataInQ has data")
                    ss, ts = dataInQ.get(0) 
                    print('<>P300\n>>Samples [', ss, ']\nTimestamp [', ts, ']')
                except Empty:
                    return
        if not E.is_set(): #NEW:: Event is cleared in case data collection is not over, waiting for a new set()
            EP300.clear()
    print('Thread Finished')

if __name__ == '__main__':
    print('Looking for an EEG stream...')
    streams = resolve_stream('type', 'EEG')
    inlet = StreamInlet(streams[0])
    print('Connected!\n')

    AcqTh.start()
    P300Th.start()

    AcqTh.join()
    P300Th.join()

    print("\n\n>>>DONE<<<\n\n")