How to run Python custom objects in separate processes, all working on a shared events queue?

Processes require a function to run, which defines what the process is actually doing. Once this function exits (and there are no non-daemon threads) the process is done. This is similar to how Python itself always executes a __main__ script.

If you do mp.Process(target=CustomObject, args=(events_queue,)) that just tells the process to call CustomObject - which instantiates it once and then is done. This is not what you want, unless the class actually performs work when instantiated - which is a bad idea for other reasons.

Instead, you must define a main function or method that handles what you need: "communicate with events shared through the events_queue". This function should listen to the queue and take action depending on the events received.

A simple implementation looks like this:

import os, time
from multiprocessing import Queue, Process


class Worker:
    # separate input and output for simplicity
    def __init__(self, commands: Queue, results: Queue):
        self.commands = commands
        self.results = results

    # our main function to be run by a process
    def main(self):
        # each process should handle more than one command
        while True:
            value = self.commands.get()
            # pick a well-defined signal to detect "no more work"
            if value is None:
                self.results.put(None)
                break
            # do whatever needs doing
            result = self.do_stuff(value)
            print(os.getpid(), ':', self, 'got', value, 'put', result)
            time.sleep(0.2)  # pretend we do something
            # pass on more work if required
            self.results.put(result)

    # placeholder for what needs doing
    def do_stuff(self, value):
        raise NotImplementedError

This is a template for a class that just keeps on processing events. The do_stuff method must be overloaded to define what actually happens.

class AddTwo(Worker):
    def do_stuff(self, value):
        return value + 2


class TimesThree(Worker):
    def do_stuff(self, value):
        return value * 3


class Printer(Worker):
    def do_stuff(self, value):
        print(value)

This already defines fully working process payloads: Process(target=TimesThree(in_queue, out_queue).main) schedules the main method in a process, listening for and responding to commands.

Running this mainly requires connecting the individual components:

if __name__ == '__main__':
    # bookkeeping of resources we create
    processes = []
    start_queue = Queue()
    # connect our workers via queues
    queue = start_queue
    for element in (AddTwo, TimesThree, Printer):
        instance = element(queue, Queue())
        # we run the main method in processes
        processes.append(Process(target=instance.main))
        queue = instance.results
    # start all processes
    for process in processes:
        process.start()
    # send input, but do not wait for output
    start_queue.put(1)
    start_queue.put(248124)
    start_queue.put(-256)
    # send shutdown signal
    start_queue.put(None)
    # wait for processes to shutdown
    for process in processes:
        process.join()

Note that you do not need classes for this. You can also compose functions for a similar effect, as long as everything is pickle-able:

import os, time
from multiprocessing import Queue, Process

def main(commands, results, do_stuff):
    while True:
        value = commands.get()
        if value is None:
            results.put(None)
            break
        result = do_stuff(value)
        print(os.getpid(), ':', do_stuff, 'got', value, 'put', result)
        time.sleep(0.2)
        results.put(result)

def times_two(value):
    return value * 2


if __name__ == '__main__':
    in_queue, out_queue = Queue(), Queue()
    worker = Process(target=main, args=(in_queue, out_queue, times_two))
    worker.start()
    for message in (1, 3, 5, None):
        in_queue.put(message)
    while True:
        reply = out_queue.get()
        if reply is None:
            break
        print('result:', reply)