pythonwindowspyqt5multiprocessinginterrupt

Python + PyQt : Interrupting another Process


I have a PyQt5 based app which is running on Windows. I have a main GUI process from which I start another process which runs a worker function. I wish to raise a KeyboardInterrupt in the worker process from my main process. Is this possible? Please see the code snippet below to see what I am trying. This works when I replace the PyQt part with a simple console interface. I want to make it work with the GUI though. Ideally, every time I click the button, the worker thread will print the value of the counter.

Edit: It is important to note here that I must have a separate process and not a thread as the simple ctr in the try block will be replaced by a more complex bit of code which might block the Process for a longer period of time. Also, the try block is used as it can be interrupted by an exception. Hence the interest in rasing an exception in another process.

import multiprocessing as mp

from PyQt5.QtWidgets import *


def worker():
    ctr = 0
    for i in range(100):
        try:
            while True:
                ctr += 1
        except KeyboardInterrupt:
            print(ctr)

def on_button_clicked():
    #This raises a Keyorad Interrupt in the main process.
    #It should somehow raise it in the worker process
    raise KeyboardInterrupt

if __name__ == '__main__':
    mp.freeze_support()
    proc = mp.Process(target=worker)
    proc.start()

    app = QApplication([])
    button = QPushButton('Click')

    button.clicked.connect(on_button_clicked)
    button.show()
    app.exec()

Solution

  • As suggested by @RaJa,I have used Eventand the _thread module along with another thread to solve the problem. The below code should help anyone on the lookout for similar answers.

    import multiprocessing as mp
    import threading
    from PyQt5.QtWidgets import *
    import _thread
    
    def other_thread(e):    
        while True:
            e.wait()
            e.clear()
            _thread.interrupt_main()
        
    def worker(e):
        input_thread = threading.Thread(target=other_thread, args=(e,))
        input_thread.start()
        ctr = 0
        for i in range(100):
            try:
                while True:
                    ctr += 1
            except KeyboardInterrupt:
                print(ctr)            
    
    def on_button_clicked(e):
        e.set()
    
    if __name__ == '__main__':
        mp.freeze_support()
        e = mp.Event()
        proc = mp.Process(target=worker, args=(e,))
        proc.start()
    
        app = QApplication([])
        button = QPushButton('Click')
    
        button.clicked.connect(lambda: on_button_clicked(e))
        button.show()
        app.exec()