pyqt-deref-from-thread.py

import time
from types import SimpleNamespace
from threading import Event
from concurrent.futures.thread import ThreadPoolExecutor

from PyQt5.QtCore import (
    Qt, QObject, QCoreApplication, QTimer, QEventLoop, pyqtSignal
)


class Emitter(QObject):
    signal = pyqtSignal()
    _p_signal = pyqtSignal()
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # queued signal -> signal connection
        self._p_signal.connect(self.signal, Qt.QueuedConnection)

    def schedule_emit(self):
        """Schedule `signal` emit"""
        self._p_signal.emit()

app = QCoreApplication([])
executor = ThreadPoolExecutor(max_workers=4)

def test():
    ref = SimpleNamespace()  # hold single reference to Emitter obj
    ref.obj = Emitter()
    # enqueue 200 meta call events to the obj
    for i in range(200):
        ref.obj.schedule_emit()

    # event signaling the event loop is about to be entered
    event = Event()

    def clear_obj(ref=ref):
        # wait for main thread to signal it is about to enter the event loop
        event.wait()
        # time.sleep(0) # yield thread
        # del ref.obj  # clear the last/single ref to obj
        del ref.obj

    executor.submit(clear_obj)

    loop = QEventLoop()
    QTimer.singleShot(0, loop.quit)
    # bytecode optimizations, reduce the time between event.set and exec to
    # minimum
    set = event.set
    exec = loop.exec

    set()    # signal/unblock the worker;
    exec()   # enter event loop to process the queued meta calls


iter_n = 0

while True:
    if iter_n % 10 == 0:
        print("iter:", iter_n, flush=True)
    test()
    iter_n += 1