python поток не завершается с atexit

Question

Вот мой сценарий. Когда я запускаю его в оболочке, он просто зависает на неопределенное время, тогда как я ожидал, что он завершится чисто.

import logging
from logging import StreamHandler
import pymsteams
import queue
import threading
import atexit


class TeamsHandler(StreamHandler):
    def __init__(self, channel_url):
        super().__init__()
        self.channel_url = channel_url
        self.queue = queue.Queue()
        self.thread = threading.Thread(target=self._worker)
        self.thread.start()
        atexit.register(self.queue.put, None)

    def _worker(self):
        while True:
            record = self.queue.get()
            if record is None:
                break
            msg = self.format(record)
            print(msg)

    def emit(self, record):
        # enqueue the record to log and return control to the caller
        self.queue.put(record)


if __name__ == "__main__":
    my_logger = logging.getLogger('TestLogging')
    my_logger.setLevel(logging.DEBUG)
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    my_logger.addHandler(console_handler)

    CHANNEL_ID = "not_used_anyway"
    teamshandler = TeamsHandler(CHANNEL_ID)
    teamshandler.setFormatter(logging.Formatter('%(levelname)s %(message)s'))
    teamshandler.setLevel(logging.DEBUG)
    my_logger.addHandler(teamshandler)
    for i in range(1, 2):
        my_logger.error(f"this is an error [{i}]")
        my_logger.info(f"this is an info [{i}]")

Запись None, которая должна быть отправлена ​​atexit (строка 28), никогда не приходит, поэтому поток остается открытым навсегда.

Как убедиться, что программа завершается без ошибок, изменяя только TeamsHandler?

Answer 1

Как указывает avysk, проблема, вероятно, в том, что обработчики atexit срабатывают слишком поздно, после того, как ожидание не-демонических потоков уже (должно быть) выполнено, что приводит к взаимоблокировке.

На вашем месте я бы просто добавил вызов типа TeamsHandler.finish() в конце блока if __name__ == '__main__' и изменил бы TeamsHandler в этих строках (непроверено):

_queues = []

class TeamsHandler(StreamHandler):
    def __init__(self, channel_url):
        super().__init__()
        self.channel_url = channel_url
        self.queue = queue.Queue()
        self.thread = threading.Thread(target=self._worker)
        self.thread.start()
        _queues.append(self.queue)

    def _worker(self):
        while True:
            record = self.queue.get()
            if record is None:
                break
            msg = self.format(record)
            print(msg)

    def emit(self, record):
        # enqueue the record to log and return control to the caller
        self.queue.put(record)

    @staticmethod
    def finish(self):
        for q in _queues:
            q.put(None)
        del _queues[:]

Answer 2

У меня что-то работает, посмотрите:

import queue
import threading


class Worker:
    def __init__(self):
        self.queue = queue.Queue()
        threading.Thread(target=self._worker).start()

    def _worker(self):
        print("starting thread")
        while True:
            record = self.queue.get()
            if record is None:
                print("exiting")
                break
            print(f"Got message: {record}")

    def emit(self, record):
        self.queue.put(record)


class Wrapper:
    def __init__(self):
        self._worker = Worker()

    def __del__(self):
        print("Wrapper is being deleted")
        self._worker.emit(None)

    def emit(self, record):
        self._worker.emit(record)

def main():
    worker = Wrapper()
    worker.emit("foo")
    worker.emit("bar")
    print("main exits")


if __name__ == "__main__":
    main()

Дело в том, что когда main завершается, worker (который является экземпляром Wrapper) выходит за рамки, и вызывается его метод __del__, и он отправляет сообщение остановки реальному рабочему объекту.

Результаты выполнения этого кода (строки «Получено сообщение», конечно, могут быть в разных местах):

starting thread
main exits
Wrapper is being deleted
Got message: foo
Got message: bar
exiting