#!/usr/bin/env python3
-"""
-The MIT License (MIT)
+"""The MIT License (MIT)
Copyright (c) 2020 LuizaLabs
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
-This class is based on https://github.com/luizalabs/shared-memory-dict.
-For details about what is preserved from the original and what was changed
-by Scott, see NOTICE at the root of this module.
+This class is based on
+https://github.com/luizalabs/shared-memory-dict. For details about
+what is preserved from the original and what was changed by Scott, see
+`NOTICE
+<https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=NOTICE;hb=HEAD>`_
+at the root of this module.
+
"""
import pickle
class SharedDict(object):
"""This class emulates the dict container but uses a
- Multiprocessing.SharedMemory region to back the dict such that it
+ `Multiprocessing.SharedMemory` region to back the dict such that it
can be read and written by multiple independent processes at the
same time. Because it constantly de/re-serializes the dict, it is
much slower than a normal dict.
+ Example usage... one process should set up the shared memory::
+
+ from pyutils.collectionz.shared_dict import SharedDict
+
+ shared_memory_id = 'SharedDictIdentifier'
+ shared_memory_size_bytes = 4096
+ shared_memory = SharedDict(shared_memory_id, shared_memory_size_bytes)
+
+ Other processes can then attach to the shared memory by
+ referencing its name. Don't try to pass the :class:`SharedDict` itself to
+ a child process. Rather, just pass its name string. You can create
+ child processes any way that Python supports. The
+ `wordle example <https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=examples/wordle/wordle.py;h=df9874ee0b309e7a70a5a7c8900629869def3928;hb=HEAD>`__ uses the
+ parallelize framework with `SharedDict` but a simple `subprocess.run`,
+ `exec_utils`, `ProcessExecutor`, whatever::
+
+ from pyutils import exec_utils
+
+ processes = []
+ for i in range(10):
+ processes.append(
+ exec_utils.cmd_in_background(
+ f'myhelper.py --number {i} --shared_memory={shared_memory_id}'
+ )
+ )
+
+ In the child process, attach the already created :class:`SharedDict`
+ using its name. A size is not necessary when attaching to an
+ already created shared memory region -- it cannot be resized after
+ creation. The name must be the same exact name that was used to
+ create it originally::
+
+ from pyutils.collectionz.shared_dict import SharedDict
+
+ shared_memory_id = config.config['shared_memory']
+ shared_memory = SharedDict(shared_memory_id)
+
+ The children processes (and parent process, also) can now just use
+ the shared memory like a normal `dict`::
+
+ if shared_memory[work_id] is None:
+ result = do_expensive_work(work_id)
+ shared_memory[work_id] = result
+
+ .. note::
+
+ It's pretty slow to mutate data in the shared memory. First,
+ it needs to acquire an exclusive lock. Second, it essentially
+ pickles an entire dict into the shared memory region. So this
+ is not a data structure that is going to win awards for speed.
+ But it is a very convenient way to have a shared cache, for
+ example. See the wordle example for a real life program using
+ `SharedDict` this way. It basically saves the result of large
+ computations in a `SharedDict` thereby allowing all threads to
+ avoid recomputing that same expensive computation. In this
+ scenario the slowness of the dict writes are more than paid
+ for by the avoidence of duplicated, expensive work.
+
+ Finally, someone (likely the main process) should call the :meth:`cleanup`
+ method when the shared memory region is no longer needed::
+
+ shared_memory.cleanup()
+
+ See also the `shared_dict_test.py <https://wannabe.guru.org/gitweb/?p=pyutils.git;a=blob_plain;f=tests/collectionz/shared_dict_test.py;h=0a684f4835554553018cefbc114034c2dc405794;hb=HEAD>`__ for an
+ example of using this class.
+
+ ---
"""
- NULL_BYTE = b'\x00'
+ NULL_BYTE = b"\x00"
LOCK = RLock()
def __init__(
name: Optional[str] = None,
size_bytes: Optional[int] = None,
) -> None:
- """
- Creates or attaches a shared dictionary back by a SharedMemory buffer.
- For create semantics, a unique name (string) and a max dictionary size
- (expressed in bytes) must be provided. For attach semantics, these are
- ignored.
+ """Creates or attaches a shared dictionary back by a
+ :class:`SharedMemory` buffer. For create semantics, a unique
+ name (string) and a max dictionary size (expressed in bytes)
+ must be provided. For attach semantics size is ignored.
- The first process that creates the SharedDict is responsible for
- (optionally) naming it and deciding the max size (in bytes) that
- it may be. It does this via args to the c'tor.
+ .. warning::
- Subsequent processes may safely omit name and size args.
+ Size is ignored on attach operations. The size of the
+ shared memory region cannot be changed once it has been
+ created.
+
+ The first process that creates the :class:`SharedDict` is
+ responsible for (optionally) naming it and deciding the max
+ size (in bytes) that it may be. It does this via args to the
+ c'tor.
+
+ Subsequent processes may safely the size arg.
+
+ Args:
+ name: the name of the shared dict, only required for initial caller
+ size_bytes: the maximum size of data storable in the shared dict,
+ only required for the first caller.
"""
assert size_bytes is None or size_bytes > 0
self.name = self.shared_memory.name
def get_name(self):
- """Returns the name of the shared memory buffer backing the dict."""
+ """
+ Returns:
+ The name of the shared memory buffer backing the dict.
+ """
return self.name
def _get_or_create_memory_block(
name: Optional[str] = None,
size_bytes: Optional[int] = None,
) -> shared_memory.SharedMemory:
+ """Internal helper."""
try:
return shared_memory.SharedMemory(name=name)
except FileNotFoundError:
return shared_memory.SharedMemory(name=name, create=True, size=size_bytes)
def _ensure_memory_initialization(self):
+ """Internal helper."""
with SharedDict.LOCK:
memory_is_empty = (
- bytes(self.shared_memory.buf).split(SharedDict.NULL_BYTE, 1)[0] == b''
+ bytes(self.shared_memory.buf).split(SharedDict.NULL_BYTE, 1)[0] == b""
)
if memory_is_empty:
self.clear()
def _write_memory(self, db: Dict[Hashable, Any]) -> None:
+ """Internal helper."""
data = self._serializer.dumps(db)
with SharedDict.LOCK:
try:
raise ValueError("exceeds available storage") from e
def _read_memory(self) -> Dict[Hashable, Any]:
+ """Internal helper."""
with SharedDict.LOCK:
return self._serializer.loads(self.shared_memory.buf.tobytes())
@contextmanager
def _modify_dict(self):
+ """Internal helper."""
with SharedDict.LOCK:
db = self._read_memory()
yield db
def close(self) -> None:
"""Unmap the shared dict and memory behind it from this
- process. Called by automatically __del__"""
- if not hasattr(self, 'shared_memory'):
+ process. Called by automatically :meth:`__del__`.
+ """
+ if not hasattr(self, "shared_memory"):
return
self.shared_memory.close()
def cleanup(self) -> None:
- """Unlink the shared dict and memory behind it. Only the last process should
- invoke this. Not called automatically."""
- if not hasattr(self, 'shared_memory'):
+ """Unlink the shared dict and memory behind it. Only the last process
+ should invoke this. Not called automatically."""
+ if not hasattr(self, "shared_memory"):
return
with SharedDict.LOCK:
self.shared_memory.unlink()
def clear(self) -> None:
- """Clear the dict."""
+ """Clears the shared dict."""
self._write_memory({})
def copy(self) -> Dict[Hashable, Any]:
- """Returns a shallow copy of the dict."""
+ """
+ Returns:
+ A shallow copy of the shared dict.
+ """
return self._read_memory()
def __getitem__(self, key: Hashable) -> Any:
return repr(self._read_memory())
def get(self, key: str, default: Optional[Any] = None) -> Any:
- """Gets the value associated with key or a default."""
+ """
+ Args:
+ key: the key to lookup
+ default: the value returned if key is not present
+
+ Returns:
+ The value associated with key or a default.
+ """
return self._read_memory().get(key, default)
def keys(self) -> KeysView[Hashable]: