Terminating Subprocesses and Threads while they're calculating

[lvlanson]

I am working on a larger project, where I have 2 Threads (same process) and one separate process. One of the threads is the gui, the other thread is a sentinel thread, observing the subprocess, and the subprocess is doing some heavy lifting with neural networks. The architecture looks somewhat like this:



I need to be able to cancel the process of the neural network and respectively end the sentinel thread. I have created a small example which shows the architecture generally and what I approach to do.

from multiprocessing import Process, Queue
from threading import Thread
from time import sleep


class Worker(Process):
    # The worker resembles the neural network. It does some calculations and shares
    # the information via the queue.
    def __init__(self, queue: Queue):
        Process.__init__(self)
        self.queue = queue

    def run(self):
        i = 0
        while True:
            self.queue.put(i)
            i += 1

    def stop(self):
        # I used the stop function for trying out some things, like using a joinable 
        # queue and block execution as long as the queue is not empty, which is not 
        # working
        self.queue.put(None)
        self.terminate()


class Listener(Thread):
    # This class resembles the sentinel thread. It checks in an infinite loop for
    # messages. In the real application I send signals via the signals and slots
    # design pattern to the gui and display the sent information.

    def __init__(self):
        Thread.__init__(self)
        self.queue = Queue()
        self.worker = Worker(self.queue)

    def run(self):
        self.worker.start()
        while True:
            data = self.queue.get()
            if data is not None:
                print(data)
            else:
                break
        print("broken")

    def stop(self):
        self.worker.stop()


class System:
    # This class resembles the gui

    def __init__(self):
        self.listener = Listener()

    def start(self):
        self.listener.start()

    def stop(self):
        self.listener.stop()


if __name__ == "__main__":
    system = System()
    system.start()
    sleep(0.1)
    system.stop()

What is the problem
As long as a process reads or writes to the queue, and/or the queue is not emptied properly, one or both of the processes become zombie processes, which basically is a deadlock in some sense. Therefore I need to find a way to properly handle the queue when terminating the process, thus the processes terminate without errors.

What I have tried so far:

1. Using a Joinable Queue and join() for each task_done()
   
2. Rewriting the SIGTERM signalhandler to wait the queue to be emptied
   
3. Using a Joinable Queue and only join() within the SIGTERM signalhandler
   

The results:

1. The speed of the processing collapsed greatly, but termination worked properly
   
2. termination does not work the way I implemented it
   
3. Sometimes it worked, sometimes it did not. So no reliable output and knowledge from this method
   

An attempt for (3) is the following:

class Worker(Process):

    def __init__(self, queue: Queue):
        Process.__init__(self)
        self.queue = queue
        self.abort = False
        self.lock = Lock()
        signal(SIGTERM, self.stop)

    def run(self):
        i = 0
        while True:
            self.lock.acquire()
            if self.abort:
                break
            else:
                self.queue.put(i)
                i += 1
            self.lock.release()
        exit(0)

    def stop(self, sig, frame):
        self.abort = True
        self.queue.put(None)
        self.queue.join()
        exit(0)

[Gribouillis]

As a workaround, I suggest to use Queue.get() with a timeout, say half a second in the sentinel thread instead of a blocking Queue.get() and check if the worker process is still running when the timeout occurs. This gives the sentinel thread an option to exit instead of waiting for queue events when the worker thread is already dead.

[lvlanson]

As a workaround, I suggest to use Queue.get() with a timeout, say half a second in the sentinel thread instead of a blocking Queue.get() and check if the worker process is still running when the timeout occurs. This gives the sentinel thread an option to exit instead of waiting for queue events when the worker thread is already dead.

I guess this is a last resort option. There still would data remain in the multiprocessing queue. My instinct tells me that somehow interrupt handlers/ asynchronous event handlers should do the trick somehow.

[Gribouillis]

There are still things to try. You could call queue.close() and queue.join_thread() in the worker process when it receives the instruction to stop. This would wait until every data previously sent to the queue has been written in the underlying pipe. Also it would raise an exception if the run() function still tries to put() anything else in the queue.

[lvlanson]

There are still things to try. You could call queue.close() and queue.join_thread() in the worker process when it receives the instruction to stop. This would wait until every data previously sent to the queue has been written in the underlying pipe. Also it would raise an exception if the run() function still tries to put() anything else in the queue.

This solution works

from multiprocessing import Process, Lock
from threading import Thread
from time import sleep
from sys import exit
from multiprocessing import Queue


class Worker(Process):

    def __init__(self, queue: Queue):
        Process.__init__(self)
        self.queue = queue

    def run(self):
        i = 0
        try:
            while True:
                self.queue.put(i)
                i += 1
        except ValueError:
            pass

    def stop(self):
        self.queue.put("END")
        self.queue.close()
        self.queue.join_thread()
        self.terminate()


class Listener(Thread):

    def __init__(self):
        Thread.__init__(self)
        self.queue = Queue()
        self.worker = Worker(self.queue)
        self.is_running = True

    def run(self):
        self.worker.start()
        while self.is_running:
            try:
                data = self.queue.get()
                if data != "END":
                    pass
                else:
                    self.is_running = False
            except TypeError:
                self.is_running = False

    def stop(self):
        self.worker.stop()
        self.is_running = False


class System:

    def __init__(self):
        self.listener = Listener()

    def start(self):
        self.listener.start()

    def stop(self):
        self.listener.stop()


if __name__ == "__main__":
    system = System()
    system.start()
    sleep(0.1)
    system.stop()
    sleep(1)
    print(f"Process Alive: {system.listener.worker.is_alive()}")
    print(f"Thread Alive: {system.listener.is_alive()}")