Skip to content

qass.tools.analyzer.mainthread_hook

Classes:

Name Description
inMainThread

__all__ module-attribute 

__all__ = ['inMainThread']

inMainThread

Bases: QObject

Methods:

Name Description
__call__

Invokes a method on the main thread. Taking care of garbage collection "bugs".
__init__
execute

Attributes:

Name Type Description
called
synchron
Source code in src/qass/tools/analyzer/mainthread_hook.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class inMainThread(QObject):
    def __init__(self, synchron=True):
        super().__init__()
        self.synchron = synchron
        # move self QObject to the main thread
        main_thread = QGuiApplication.instance().thread()
        self.moveToThread(main_thread)

    called = Signal()

    def __call__(self, method: Callable):
        """
        Invokes a method on the main thread. Taking care of garbage collection "bugs".
        """
        def inner_func(*args, **kwargs):
            if QThread.currentThread() == QGuiApplication.instance().thread():
                return method(*args, **kwargs)

            self._method = method

            # do not garbage collect this self object until the function has finished.
            self.setParent(QGuiApplication.instance())

            self.params = args
            self.kwargs = kwargs

            self.stdout_redirector = OutputRedirect_Manager.getInstance().getCurrentOutRedirect()
            self.stderr_redirector = OutputRedirect_Manager.getInstance().getCurrentErrRedirect()
            self.debuggers = Analyzer.Debugger.DebuggerManager.getInstance().getCurrDebuggers()

            if self.synchron:
                # This function must not be called from the main thread - this would result in a deadlock
                QObject.disconnect(self, SIGNAL('called()'), self.execute)
                self.called.connect(self.execute, Qt.BlockingQueuedConnection)
                self.called.emit()
                # the result will only be available in the synchronous execution mode

                if self._exc_info is not None:
                    type, exc, traceback = self._exc_info
                    raise exc
                return self._result
            else:
                QObject.disconnect(self, SIGNAL('called()'), self.execute)
                self.called.connect(self.execute)
                self.called.emit()

        return inner_func

    @Slot()
    def execute(self):
        OutputRedirect_Manager.getInstance().registerStdOutHandler(self.stdout_redirector)
        OutputRedirect_Manager.getInstance().registerStdErrHandler(self.stderr_redirector)

        for dbg in self.debuggers:
            Analyzer.Debugger.DebuggerManager.getInstance().registerDebugger(dbg)

        self._result = None
        self._exc_info = None

        try:
            # actually execute the wrapped method
            self._result = self._method(*self.params, **self.kwargs)
        except Exception as e:
            import sys
            self._exception = e
            self._exc_info = sys.exc_info()
        finally:
            for dbg in self.debuggers:
                Analyzer.Debugger.DebuggerManager.getInstance().unregisterDebugger(dbg)

            OutputRedirect_Manager.getInstance().unregisterStdOutHandler(self.stdout_redirector)
            OutputRedirect_Manager.getInstance().unregisterStdErrHandler(self.stderr_redirector)
            # decref self -> trigger garbage collector
            self.setParent(None)

called class-attribute instance-attribute 

called = Signal()

synchron instance-attribute 

synchron = synchron

__call__ 

__call__(method: Callable)

Invokes a method on the main thread. Taking care of garbage collection "bugs".

Source code in src/qass/tools/analyzer/mainthread_hook.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __call__(self, method: Callable):
    """
    Invokes a method on the main thread. Taking care of garbage collection "bugs".
    """
    def inner_func(*args, **kwargs):
        if QThread.currentThread() == QGuiApplication.instance().thread():
            return method(*args, **kwargs)

        self._method = method

        # do not garbage collect this self object until the function has finished.
        self.setParent(QGuiApplication.instance())

        self.params = args
        self.kwargs = kwargs

        self.stdout_redirector = OutputRedirect_Manager.getInstance().getCurrentOutRedirect()
        self.stderr_redirector = OutputRedirect_Manager.getInstance().getCurrentErrRedirect()
        self.debuggers = Analyzer.Debugger.DebuggerManager.getInstance().getCurrDebuggers()

        if self.synchron:
            # This function must not be called from the main thread - this would result in a deadlock
            QObject.disconnect(self, SIGNAL('called()'), self.execute)
            self.called.connect(self.execute, Qt.BlockingQueuedConnection)
            self.called.emit()
            # the result will only be available in the synchronous execution mode

            if self._exc_info is not None:
                type, exc, traceback = self._exc_info
                raise exc
            return self._result
        else:
            QObject.disconnect(self, SIGNAL('called()'), self.execute)
            self.called.connect(self.execute)
            self.called.emit()

    return inner_func

__init__ 

__init__(synchron=True)
Source code in src/qass/tools/analyzer/mainthread_hook.py 
35
36
37
38
39
40
def __init__(self, synchron=True):
    super().__init__()
    self.synchron = synchron
    # move self QObject to the main thread
    main_thread = QGuiApplication.instance().thread()
    self.moveToThread(main_thread)

execute 

execute()
Source code in src/qass/tools/analyzer/mainthread_hook.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@Slot()
def execute(self):
    OutputRedirect_Manager.getInstance().registerStdOutHandler(self.stdout_redirector)
    OutputRedirect_Manager.getInstance().registerStdErrHandler(self.stderr_redirector)

    for dbg in self.debuggers:
        Analyzer.Debugger.DebuggerManager.getInstance().registerDebugger(dbg)

    self._result = None
    self._exc_info = None

    try:
        # actually execute the wrapped method
        self._result = self._method(*self.params, **self.kwargs)
    except Exception as e:
        import sys
        self._exception = e
        self._exc_info = sys.exc_info()
    finally:
        for dbg in self.debuggers:
            Analyzer.Debugger.DebuggerManager.getInstance().unregisterDebugger(dbg)

        OutputRedirect_Manager.getInstance().unregisterStdOutHandler(self.stdout_redirector)
        OutputRedirect_Manager.getInstance().unregisterStdErrHandler(self.stderr_redirector)
        # decref self -> trigger garbage collector
        self.setParent(None)