Python ShareableList Example With JSON

client.py

"""Shared Memory Test Client
"""
import json
import sys
from multiprocessing.shared_memory import ShareableList


def client(shared_name):
    print(f"Client: connecting to list {shared_name}")
    shared_list = ShareableList(name=shared_name)

    json_str = shared_list[0]
    data = json.loads(json_str)

    print(f"Name: {data['name']}")
    print(f"Primes: {data['primes']}")


if __name__ == '__main__':
    if len(sys.argv) != 2:
        print("USAGE: client.py <shared_name>")
        sys.exit(1)
    client(sys.argv[1])

server.py

"""Shared Memory Test Server
"""
import json
from pathlib import Path
from multiprocessing.shared_memory import ShareableList
import subprocess
from threading import Thread, Event
import time


message = {
    "name": "This is a test message",
    "value": 42,
    "primes": [2, 3, 5, 7, 11, 13],
}


class SharedMemoryServer(Thread):
    """Share a list using shared memory."""

    def __init__(self):
        super().__init__()

        # And event so someone can call our self.get_shared_name() and be
        # sure everything is setup.
        self.ready = Event()

    def run(self):
        """Share out a list in a thread and wait."""
        json_str = json.dumps(message)

        shared_list = ShareableList([json_str])

        self.shared_name = shared_list.shm.name
        self.ready.set()

        time.sleep(10000000)

    def get_shared_name(self):
        """Wait and then return the shared name."""
        self.ready.wait()
        return self.shared_name


def start_client(shared_name):
    path = Path(__file__).resolve().parent / "client.py"
    args = ["python", str(path), shared_name]

    print(f"Server: running client {args}")
    subprocess.run(args)


def start_server():
    print(f"Server: starting")
    server = SharedMemoryServer()
    server.start()

    name = server.get_shared_name()