Python Shared Memory

from multiprocessing import shared_memory
import numpy as np

# Process A: Create shared memory block
shm = shared_memory.SharedMemory(create=True, size=1024, name="my_shared_block")
print(f"Created: {shm.name}, Size: {shm.size} bytes")

# Write data to shared memory
data = b"Hello from Process A!"
shm.buf[:len(data)] = data

# Process B: Attach to existing shared memory
shm_b = shared_memory.SharedMemory(name="my_shared_block")
message = bytes(shm_b.buf[:21]).decode('utf-8')
print(f"Read: {message}")  # Output: Hello from Process A!

# IMPORTANT: Cleanup (covered in detail later)
shm_b.close()
shm.close()
shm.unlink()  # Only creator should unlink!

import numpy as np
from multiprocessing import shared_memory, Process

def create_shared_array(shape, dtype=np.float64):
    """Create a NumPy array backed by shared memory."""
    size = np.prod(shape) * np.dtype(dtype).itemsize
    shm = shared_memory.SharedMemory(create=True, size=size)
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    return arr, shm

def attach_shared_array(name, shape, dtype=np.float64):
    """Attach to an existing shared NumPy array."""
    shm = shared_memory.SharedMemory(name=name)
    arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    return arr, shm

# Process A: Create shared array
arr, shm = create_shared_array((1000, 1000))
arr[:] = np.random.rand(1000, 1000)
print(f"Shared memory name: {shm.name}")
print(f"Array sum: {arr.sum():.2f}")

from contextlib import contextmanager
from multiprocessing import shared_memory

@contextmanager
def managed_shared_memory(name=None, create=False, size=0):
    """Context manager for automatic shared memory cleanup."""
    shm = None
    try:
        if create:
            shm = shared_memory.SharedMemory(create=True, size=size, name=name)
        else:
            shm = shared_memory.SharedMemory(name=name)
        yield shm
    finally:
        if shm is not None:
            shm.close()
            if create:
                try:
                    shm.unlink()
                except FileNotFoundError:
                    pass  # Already unlinked

# Usage - automatic cleanup guaranteed!
with managed_shared_memory(create=True, size=1024) as shm:
    shm.buf[0:5] = b"Hello"
    print(f"Name: {shm.name}")
# ✓ Automatically closed and unlinked here!

import numpy as np
from multiprocessing import shared_memory, Process, Lock
import time

def process_chunk(shm_name, shape, dtype, start_row, end_row, lock):
    """Process a chunk of the image (e.g., apply grayscale)."""
    # Attach to shared memory
    shm = shared_memory.SharedMemory(name=shm_name)
    img = np.ndarray(shape, dtype=dtype, buffer=shm.buf)

    # Process our assigned rows (no lock needed - non-overlapping regions!)
    for row in range(start_row, end_row):
        for col in range(shape[1]):
            # Convert RGB to grayscale
            r, g, b = img[row, col, 0], img[row, col, 1], img[row, col, 2]
            gray = int(0.299 * r + 0.587 * g + 0.114 * b)
            img[row, col] = [gray, gray, gray]

    # Report progress (needs lock for shared print)
    with lock:
        print(f"Processed rows {start_row}-{end_row}")

    shm.close()

def parallel_grayscale(image, num_workers=4):
    """Convert image to grayscale using multiple processes."""
    height, width, channels = image.shape

    # Create shared memory for the image
    shm = shared_memory.SharedMemory(create=True, size=image.nbytes)
    shared_img = np.ndarray(image.shape, dtype=image.dtype, buffer=shm.buf)
    shared_img[:] = image  # Copy data to shared memory

    lock = Lock()
    processes = []
    rows_per_worker = height // num_workers

    # Spawn worker processes
    for i in range(num_workers):
        start_row = i * rows_per_worker
        end_row = height if i == num_workers - 1 else (i + 1) * rows_per_worker

        p = Process(
            target=process_chunk,
            args=(shm.name, image.shape, image.dtype, start_row, end_row, lock)
        )
        processes.append(p)
        p.start()

    # Wait for all processes to complete
    for p in processes:
        p.join()

    # Copy result back
    result = shared_img.copy()

    # Cleanup
    shm.close()
    shm.unlink()

    return result

# Usage
if __name__ == "__main__":
    # Create a test image (1920x1080 RGB)
    test_image = np.random.randint(0, 256, (1080, 1920, 3), dtype=np.uint8)

    start = time.time()
    result = parallel_grayscale(test_image, num_workers=4)
    elapsed = time.time() - start

    print(f"Processed {test_image.size:,} pixels in {elapsed:.2f}s")