Moved SharedMemoryManager to managers module, tweak to fragile test.

Author: applio

Doc/library/multiprocessing.shared_memory.rst

@@ -20,7 +20,8 @@ and management of shared memory to be accessed by one or more processes
 on a multicore or symmetric multiprocessor (SMP) machine.  To assist with
 the life-cycle management of shared memory especially across distinct
 processes, a :class:`~multiprocessing.managers.BaseManager` subclass,
-:class:`SharedMemoryManager`, is also provided.
+:class:`SharedMemoryManager`, is also provided in the
+``multiprocessing.managers`` module.
 
 In this module, shared memory refers to "System V style" shared memory blocks
 (though is not necessarily implemented explicitly as such) and does not refer

Lib/multiprocessing/managers.py

@@ -1,5 +1,5 @@
 #
-# Module providing the `SyncManager` class for dealing
+# Module providing manager classes for dealing
 # with shared objects
 #
 # multiprocessing/managers.py
@@ -8,7 +8,8 @@
 # Licensed to PSF under a Contributor Agreement.
 #
 
-__all__ = [ 'BaseManager', 'SyncManager', 'BaseProxy', 'Token' ]
+__all__ = [ 'BaseManager', 'SyncManager', 'BaseProxy', 'Token',
+            'SharedMemoryManager' ]
 
 #
 # Imports
@@ -19,6 +20,7 @@
 import array
 import queue
 import time
+from os import getpid
 
 from traceback import format_exc
 
@@ -28,6 +30,11 @@
 from . import process
 from . import util
 from . import get_context
+try:
+    from . import shared_memory
+    HAS_SHMEM = True
+except ImportError:
+    HAS_SHMEM = False
 
 #
 # Register some things for pickling
@@ -1200,3 +1207,143 @@ class SyncManager(BaseManager):
 # types returned by methods of PoolProxy
 SyncManager.register('Iterator', proxytype=IteratorProxy, create_method=False)
 SyncManager.register('AsyncResult', create_method=False)
+
+#
+# Definition of SharedMemoryManager and SharedMemoryServer
+#
+
+if HAS_SHMEM:
+    class _SharedMemoryTracker:
+        "Manages one or more shared memory segments."
+
+        def __init__(self, name, segment_names=[]):
+            self.shared_memory_context_name = name
+            self.segment_names = segment_names
+
+        def register_segment(self, segment_name):
+            "Adds the supplied shared memory block name to tracker."
+            util.debug(f"Register segment {segment_name!r} in pid {getpid()}")
+            self.segment_names.append(segment_name)
+
+        def destroy_segment(self, segment_name):
+            """Calls unlink() on the shared memory block with the supplied name
+            and removes it from the list of blocks being tracked."""
+            util.debug(f"Destroy segment {segment_name!r} in pid {getpid()}")
+            self.segment_names.remove(segment_name)
+            segment = shared_memory.SharedMemory(segment_name)
+            segment.close()
+            segment.unlink()
+
+        def unlink(self):
+            "Calls destroy_segment() on all tracked shared memory blocks."
+            for segment_name in self.segment_names[:]:
+                self.destroy_segment(segment_name)
+
+        def __del__(self):
+            util.debug(f"Call {self.__class__.__name__}.__del__ in {getpid()}")
+            self.unlink()
+
+        def __getstate__(self):
+            return (self.shared_memory_context_name, self.segment_names)
+
+        def __setstate__(self, state):
+            self.__init__(*state)
+
+
+    class SharedMemoryServer(Server):
+
+        public = Server.public + \
+                 ['track_segment', 'release_segment', 'list_segments']
+
+        def __init__(self, *args, **kwargs):
+            Server.__init__(self, *args, **kwargs)
+            self.shared_memory_context = \
+                _SharedMemoryTracker(f"shmm_{self.address}_{getpid()}")
+            util.debug(f"SharedMemoryServer started by pid {getpid()}")
+
+        def create(self, c, typeid, *args, **kwargs):
+            """Create a new distributed-shared object (not backed by a shared
+            memory block) and return its id to be used in a Proxy Object."""
+            # Unless set up as a shared proxy, don't make shared_memory_context
+            # a standard part of kwargs.  This makes things easier for supplying
+            # simple functions.
+            if hasattr(self.registry[typeid][-1], "_shared_memory_proxy"):
+                kwargs['shared_memory_context'] = self.shared_memory_context
+            return Server.create(self, c, typeid, *args, **kwargs)
+
+        def shutdown(self, c):
+            "Call unlink() on all tracked shared memory, terminate the Server."
+            self.shared_memory_context.unlink()
+            return Server.shutdown(self, c)
+
+        def track_segment(self, c, segment_name):
+            "Adds the supplied shared memory block name to Server's tracker."
+            self.shared_memory_context.register_segment(segment_name)
+
+        def release_segment(self, c, segment_name):
+            """Calls unlink() on the shared memory block with the supplied name
+            and removes it from the tracker instance inside the Server."""
+            self.shared_memory_context.destroy_segment(segment_name)
+
+        def list_segments(self, c):
+            """Returns a list of names of shared memory blocks that the Server
+            is currently tracking."""
+            return self.shared_memory_context.segment_names
+
+
+    class SharedMemoryManager(BaseManager):
+        """Like SyncManager but uses SharedMemoryServer instead of Server.
+
+        It provides methods for creating and returning SharedMemory instances
+        and for creating a list-like object (ShareableList) backed by shared
+        memory.  It also provides methods that create and return Proxy Objects
+        that support synchronization across processes (i.e. multi-process-safe
+        locks and semaphores).
+        """
+
+        _Server = SharedMemoryServer
+
+        def __init__(self, *args, **kwargs):
+            BaseManager.__init__(self, *args, **kwargs)
+            util.debug(f"{self.__class__.__name__} created by pid {getpid()}")
+
+        def __del__(self):
+            util.debug(f"{self.__class__.__name__}.__del__ by pid {getpid()}")
+            pass
+
+        def get_server(self):
+            'Better than monkeypatching for now; merge into Server ultimately'
+            if self._state.value != State.INITIAL:
+                if self._state.value == State.STARTED:
+                    raise ProcessError("Already started SharedMemoryServer")
+                elif self._state.value == State.SHUTDOWN:
+                    raise ProcessError("SharedMemoryManager has shut down")
+                else:
+                    raise ProcessError(
+                        "Unknown state {!r}".format(self._state.value))
+            return self._Server(self._registry, self._address,
+                                self._authkey, self._serializer)
+
+        def SharedMemory(self, size):
+            """Returns a new SharedMemory instance with the specified size in
+            bytes, to be tracked by the manager."""
+            with self._Client(self._address, authkey=self._authkey) as conn:
+                sms = shared_memory.SharedMemory(None, create=True, size=size)
+                try:
+                    dispatch(conn, None, 'track_segment', (sms.name,))
+                except BaseException as e:
+                    sms.unlink()
+                    raise e
+            return sms
+
+        def ShareableList(self, sequence):
+            """Returns a new ShareableList instance populated with the values
+            from the input sequence, to be tracked by the manager."""
+            with self._Client(self._address, authkey=self._authkey) as conn:
+                sl = shared_memory.ShareableList(sequence)
+                try:
+                    dispatch(conn, None, 'track_segment', (sl.shm.name,))
+                except BaseException as e:
+                    sl.shm.unlink()
+                    raise e
+            return sl

Lib/multiprocessing/shared_memory.py

@@ -5,14 +5,11 @@
 """
 
 
-__all__ = [ 'SharedMemory', 'ShareableList',
-            'SharedMemoryServer', 'SharedMemoryManager' ]
+__all__ = [ 'SharedMemory', 'ShareableList' ]
 
 
 from functools import partial
 import mmap
-from .managers import dispatch, BaseManager, Server, State, ProcessError
-from . import util
 import os
 import errno
 import struct
@@ -53,7 +50,7 @@ class SharedMemory:
     shared memory block.
 
     Every shared memory block is assigned a unique name.  This enables
-    one process to create a shared memory block with a particular name 
+    one process to create a shared memory block with a particular name
     so that a different process can attach to that same shared memory
     block using that same name.
 
@@ -500,139 +497,3 @@ def index(self, value):
                 return position
         else:
             raise ValueError(f"{value!r} not in this container")
-
-
-class _SharedMemoryTracker:
-    "Manages one or more shared memory segments."
-
-    def __init__(self, name, segment_names=[]):
-        self.shared_memory_context_name = name
-        self.segment_names = segment_names
-
-    def register_segment(self, segment_name):
-        "Adds the supplied shared memory block name to tracker."
-        util.debug(f"Registering segment {segment_name!r} in pid {os.getpid()}")
-        self.segment_names.append(segment_name)
-
-    def destroy_segment(self, segment_name):
-        """Calls unlink() on the shared memory block with the supplied name
-        and removes it from the list of blocks being tracked."""
-        util.debug(f"Destroying segment {segment_name!r} in pid {os.getpid()}")
-        self.segment_names.remove(segment_name)
-        segment = SharedMemory(segment_name)
-        segment.close()
-        segment.unlink()
-
-    def unlink(self):
-        "Calls destroy_segment() on all currently tracked shared memory blocks."
-        for segment_name in self.segment_names[:]:
-            self.destroy_segment(segment_name)
-
-    def __del__(self):
-        util.debug(f"Called {self.__class__.__name__}.__del__ in {os.getpid()}")
-        self.unlink()
-
-    def __getstate__(self):
-        return (self.shared_memory_context_name, self.segment_names)
-
-    def __setstate__(self, state):
-        self.__init__(*state)
-
-
-class SharedMemoryServer(Server):
-
-    public = Server.public + \
-             ['track_segment', 'release_segment', 'list_segments']
-
-    def __init__(self, *args, **kwargs):
-        Server.__init__(self, *args, **kwargs)
-        self.shared_memory_context = \
-            _SharedMemoryTracker(f"shmm_{self.address}_{os.getpid()}")
-        util.debug(f"SharedMemoryServer started by pid {os.getpid()}")
-
-    def create(self, c, typeid, *args, **kwargs):
-        """Create a new distributed-shared object (not backed by a shared
-        memory block) and return its id to be used in a Proxy Object."""
-        # Unless set up as a shared proxy, don't make shared_memory_context
-        # a standard part of kwargs.  This makes things easier for supplying
-        # simple functions.
-        if hasattr(self.registry[typeid][-1], "_shared_memory_proxy"):
-            kwargs['shared_memory_context'] = self.shared_memory_context
-        return Server.create(self, c, typeid, *args, **kwargs)
-
-    def shutdown(self, c):
-        "Call unlink() on all tracked shared memory then terminate the Server."
-        self.shared_memory_context.unlink()
-        return Server.shutdown(self, c)
-
-    def track_segment(self, c, segment_name):
-        "Adds the supplied shared memory block name to Server's tracker."
-        self.shared_memory_context.register_segment(segment_name)
-
-    def release_segment(self, c, segment_name):
-        """Calls unlink() on the shared memory block with the supplied name
-        and removes it from the tracker instance inside the Server."""
-        self.shared_memory_context.destroy_segment(segment_name)
-
-    def list_segments(self, c):
-        """Returns a list of names of shared memory blocks that the Server
-        is currently tracking."""
-        return self.shared_memory_context.segment_names
-
-
-class SharedMemoryManager(BaseManager):
-    """Like SyncManager but uses SharedMemoryServer instead of Server.
-
-    It provides methods for creating and returning SharedMemory instances
-    and for creating a list-like object (ShareableList) backed by shared
-    memory.  It also provides methods that create and return Proxy Objects
-    that support synchronization across processes (i.e. multi-process-safe
-    locks and semaphores).
-    """
-
-    _Server = SharedMemoryServer
-
-    def __init__(self, *args, **kwargs):
-        BaseManager.__init__(self, *args, **kwargs)
-        util.debug(f"{self.__class__.__name__} created by pid {os.getpid()}")
-
-    def __del__(self):
-        util.debug(f"{self.__class__.__name__} told die by pid {os.getpid()}")
-        pass
-
-    def get_server(self):
-        'Better than monkeypatching for now; merge into Server ultimately'
-        if self._state.value != State.INITIAL:
-            if self._state.value == State.STARTED:
-                raise ProcessError("Already started SharedMemoryServer")
-            elif self._state.value == State.SHUTDOWN:
-                raise ProcessError("SharedMemoryManager has shut down")
-            else:
-                raise ProcessError(
-                    "Unknown state {!r}".format(self._state.value))
-        return self._Server(self._registry, self._address,
-                            self._authkey, self._serializer)
-
-    def SharedMemory(self, size):
-        """Returns a new SharedMemory instance with the specified size in
-        bytes, to be tracked by the manager."""
-        with self._Client(self._address, authkey=self._authkey) as conn:
-            sms = SharedMemory(None, create=True, size=size)
-            try:
-                dispatch(conn, None, 'track_segment', (sms.name,))
-            except BaseException as e:
-                sms.unlink()
-                raise e
-        return sms
-
-    def ShareableList(self, sequence):
-        """Returns a new ShareableList instance populated with the values
-        from the input sequence, to be tracked by the manager."""
-        with self._Client(self._address, authkey=self._authkey) as conn:
-            sl = ShareableList(sequence)
-            try:
-                dispatch(conn, None, 'track_segment', (sl.shm.name,))
-            except BaseException as e:
-                sl.shm.unlink()
-                raise e
-        return sl

Lib/test/_test_multiprocessing.py

@@ -3652,7 +3652,7 @@ def test_shared_memory_basics(self):
 
         # Attach to existing shared memory segment but specify a new size.
         same_sms = shared_memory.SharedMemory('test01_tsmb', size=20*sms.size)
-        self.assertEqual(same_sms.size, sms.size)  # Size was ignored.
+        self.assertLess(same_sms.size, 20*sms.size)  # Size was ignored.
         same_sms.close()
 
         if shared_memory._USE_POSIX:
@@ -3737,7 +3737,7 @@ def test_shared_memory_across_processes(self):
         sms.close()
 
     def test_shared_memory_SharedMemoryManager_basics(self):
-        smm1 = shared_memory.SharedMemoryManager()
+        smm1 = multiprocessing.managers.SharedMemoryManager()
         with self.assertRaises(ValueError):
             smm1.SharedMemory(size=9)  # Fails if SharedMemoryServer not started
         smm1.start()
@@ -3756,7 +3756,7 @@ def test_shared_memory_SharedMemoryManager_basics(self):
                 # No longer there to be attached to again.
                 absent_shm = shared_memory.SharedMemory(name=held_name)
 
-        with shared_memory.SharedMemoryManager() as smm2:
+        with multiprocessing.managers.SharedMemoryManager() as smm2:
             sl = smm2.ShareableList("howdy")
             shm = smm2.SharedMemory(size=128)
             held_name = sl.shm.name