Singleton has same address in multiple processes, but behaves like separate objects

Somebody might be able to explain this with greater detail but I believe the problem -- and there is a problem which I will explain in a moment -- is that you have to first ask how are the arguments being passed from your main process to your new subprocesses. Well, it would seem that would depend on what platform you are running under, which you neglected to specify but I can infer is one that uses fork to create new processes. I believe that because your platform supports forking and therefore inherits the address space (read/only, copy on write) from the main process, no special mechanism such as pickle is required to serialize/deserialize the arguments and that is why the addresses are the same.

But here's the rub: Under Windows, which does not support fork and will rely on pickle for propagating Process arguments, the Message.__new__ method will be re-executed in each of the newly created processes. Not only will the singletons have their own unique addresses (unless a miracle occurs), they will be allocating their own multiprocessing.Pipe instances and then, of course, the application will not work at all as demonstrated below. You would be better off just passing multiprocessing.Connection objects to your subprocesses.

from multiprocessing import Process, Pipe
import logging
import time

class MessagePipe:
    """This implements a multiprocessing message pipe as a singleton and is intended for passing event information
    such as log messaged and exceptions between processes.
    """

    _instance = None
    _parent_connection = None
    _child_connection = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            print('__new__ is executing')
            cls._instance = super(MessagePipe, cls).__new__(cls, *args, **kwargs)
            cls._parent_connection, cls._child_connection = Pipe()
        return cls._instance

    @property
    def child_connection(self):
        return self._child_connection

    @property
    def parent_connection(self):
        return self._parent_connection

    def send(self, text="", exception=None, level=logging.INFO):
        message = {
            "text": text,
            "exception": exception,
            "level": level
        }
        self.child_connection.send(message)

    def recv(self):
        return self.parent_connection.recv()

def keep_receiving(pipe):
    for i in range(3):
        print("from rec_proc: ", pipe)
        print(pipe.recv())

def keep_sending(pipe):
    for i in range(3):
        print("from send_proc:", pipe)
        pipe.send(text=f"test {i + 1}")
        time.sleep(1)

def v2():
    send_proc = Process(target=keep_sending, args=(MessagePipe(), ))
    send_proc.start()

    rec_proc = Process(target=keep_receiving, args=(MessagePipe(), ))
    rec_proc.start()

    send_proc.join()
    rec_proc.join()


if __name__ == '__main__':
    v2()

Prints:

__new__ is executing
__new__ is executing
from send_proc: <__mp_main__.MessagePipe object at 0x00000283480705E0>
from rec_proc:  <__mp_main__.MessagePipe object at 0x00000249497A05E0>
from send_proc: <__mp_main__.MessagePipe object at 0x00000283480705E0>
from send_proc: <__mp_main__.MessagePipe object at 0x00000283480705E0>

rec_proc hangs because it is reading from a connection that nobody is writing to.

The problem can be somewhat mitigated by specifying pickle __setstate__ and __getstate__ methods for your MessagePipe class. This will not stop your __new__ method from being called and creating a new Pipe instances in the subprocesses but it will replace them with the serialized arguments that were passed.

from multiprocessing import Process, Pipe
import logging
import time

class MessagePipe:
    """This implements a multiprocessing message pipe as a singleton and is intended for passing event information
    such as log messaged and exceptions between processes.
    """

    _instance = None
    _parent_connection = None
    _child_connection = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            print('__new__ is executing')
            cls._instance = super(MessagePipe, cls).__new__(cls, *args, **kwargs)
            cls._parent_connection, cls._child_connection = Pipe()
        return cls._instance

    @property
    def child_connection(self):
        return self._child_connection

    @property
    def parent_connection(self):
        return self._parent_connection

    def send(self, text="", exception=None, level=logging.INFO):
        message = {
            "text": text,
            "exception": exception,
            "level": level
        }
        self.child_connection.send(message)

    def recv(self):
        return self.parent_connection.recv()

    def __getstate__(self):
        return (self._parent_connection, self._child_connection)

    def __setstate__(self, state):
        self._parent_connection, self._child_connection = state

def keep_receiving(pipe):
    for i in range(3):
        print("from rec_proc: ", pipe)
        print(pipe.recv())

def keep_sending(pipe):
    for i in range(3):
        print("from send_proc:", pipe)
        pipe.send(text=f"test {i + 1}")
        time.sleep(1)

def v2():
    send_proc = Process(target=keep_sending, args=(MessagePipe(), ))
    send_proc.start()

    rec_proc = Process(target=keep_receiving, args=(MessagePipe(), ))
    rec_proc.start()

    send_proc.join()
    rec_proc.join()


if __name__ == '__main__':
    v2()

Prints:

__new__ is executing
__new__ is executing
__new__ is executing
from send_proc: <__mp_main__.MessagePipe object at 0x00000220220F15E0>
from rec_proc:  <__mp_main__.MessagePipe object at 0x000001BB2F5C15E0>
{'text': 'test 1', 'exception': None, 'level': 20}
from rec_proc:  <__mp_main__.MessagePipe object at 0x000001BB2F5C15E0>
from send_proc: <__mp_main__.MessagePipe object at 0x00000220220F15E0>
{'text': 'test 2', 'exception': None, 'level': 20}
from rec_proc:  <__mp_main__.MessagePipe object at 0x000001BB2F5C15E0>
from send_proc: <__mp_main__.MessagePipe object at 0x00000220220F15E0>
{'text': 'test 3', 'exception': None, 'level': 20}

Another, more classic implementation of the singleton pattern that does not use method __new__ avoids creating new Pipe instances and their corresponding connections only to have them discarded by the piclke __setstate__ method when it restores the original serialized connections:

from multiprocessing import Process, Pipe
import logging
import time

class MessagePipe:
    """This implements a multiprocessing message pipe as a singleton and is intended for passing event information
    such as log messaged and exceptions between processes.
    """

    _instance = None
    _parent_connection = None
    _child_connection = None

    def __init__(self):
        raise RuntimeError('Call getInstance() instead')

    @classmethod
    def getInstance(cls):
        if cls._instance is None:
            print('Creating new instance.')
            cls._instance = cls.__new__(cls)
            cls._parent_connection, cls._child_connection = Pipe()
        return cls._instance

    @property
    def child_connection(self):
        return self._child_connection

    @property
    def parent_connection(self):
        return self._parent_connection

    def send(self, text="", exception=None, level=logging.INFO):
        message = {
            "text": text,
            "exception": exception,
            "level": level
        }
        self.child_connection.send(message)

    def recv(self):
        return self.parent_connection.recv()

    def __getstate__(self):
        return (self._parent_connection, self._child_connection)

    def __setstate__(self, state):
        self._parent_connection, self._child_connection = state

def keep_receiving(pipe):
    for i in range(3):
        print("from rec_proc: ", pipe)
        print(pipe.recv())

def keep_sending(pipe):
    for i in range(3):
        print("from send_proc:", pipe)
        pipe.send(text=f"test {i + 1}")
        time.sleep(1)

def v2():
    send_proc = Process(target=keep_sending, args=(MessagePipe.getInstance(), ))
    send_proc.start()

    rec_proc = Process(target=keep_receiving, args=(MessagePipe.getInstance(), ))
    rec_proc.start()

    send_proc.join()
    rec_proc.join()


if __name__ == '__main__':
    v2()

Prints:

Creating new instance.
from send_proc: <__mp_main__.MessagePipe object at 0x0000023459610520>
from rec_proc:  <__mp_main__.MessagePipe object at 0x0000022F2D8A1520>
{'text': 'test 1', 'exception': None, 'level': 20}
from rec_proc:  <__mp_main__.MessagePipe object at 0x0000022F2D8A1520>
from send_proc: <__mp_main__.MessagePipe object at 0x0000023459610520>
{'text': 'test 2', 'exception': None, 'level': 20}
from rec_proc:  <__mp_main__.MessagePipe object at 0x0000022F2D8A1520>
from send_proc: <__mp_main__.MessagePipe object at 0x0000023459610520>
{'text': 'test 3', 'exception': None, 'level': 20}

On any modern OS, these are virtual memory adresses. These addresses are local to each process' memory space. They almost never correspond to the same physical memory cell 0x7f805e769610. You need to google "virtual memory"

Modern OS use virtual memory. That means that physical memory pages are mapped into the process address space. Because of that, the loader can give to multiple instances of the same program the same virtual addresses. So in every process the singleton will have the same virtual address, but as those virtual addresses are mapped to different physical pages, they will point to distinct objects.

Singleton has same address in multiple processes, but behaves like separate objects

Related

Recent Posts