Как остановить threading.Thread (), заблокированный socket.socket.recv () / queue.Queue (). Get ()? - PullRequest
Новая
       8

Как остановить threading.Thread (), заблокированный socket.socket.recv () / queue.Queue (). Get ()?

0 голосов
/ 08 мая 2020

У меня есть программа, которая прослушивает пакеты, при получении интерпретирует их, а затем отправляет другой пакет. Sth вроде этого: 

  main
    |--- listener: threading.Thread(target=start_listening, args=(received), daemon=True)
    |--- sender: threading.Thread(target=start_sending, args=(sent), daemon=True)
    |
interpret_data()

received = sent = queue.Queue()
some_socket = socket.socket

def start_listening(where_to_add: queue.Queue):
    while True:
        data = some_socket.recv(X)
        if len(data) == 0: 
            received.put(0)
            break         

        received.put(data)

def start_sending(from_where_send: queue.Queue):
    while True:
        packet = from_where_send.get()
        some_socket.send(packet)

def interpret_data():
    """ Returns 1 when received no data for 10 seconds"""
    while True:
        try:
            data = received.get(timeout=10)
            # Interpret data

        except queue.Empty:
            stop_threads()
            return 1

def stop_threads():
    pass

Как лучше всего закрыть эти потоки?

Сначала я использовал queue.Queue (). Get () с тайм-аутом и проверкой __stop переменная, но до закрытия потока должен пройти тайм-аут. Малый тайм-аут - время процессора потрачено впустую, большой тайм-аут - поток не очень реагирует на остановку.

Я пробовал select.select ((some_socket,), timeout) с setblocking (False), но программа анализировала данные быстрее, чем принимала , поэтому возникало sth вроде «Невозможно удовлетворить получение данных с помощью неблокирующей функции (some_socket.recv ())».

Имеет ли смысл игнорировать их остановку и делать что-то вроде этого: 

  main 
    |---worker: threading.Thread()
    |       |--- listener: threading.Thread(target=start_listening, args=(received), daemon=True)
    |       |--- sender: threading.Thread(target=, args=(sent), daemon=True)
    |    interpret_data()
    |    return
    |
worker.wait()
...