Shared Memory

class SharedMemory:
    def __init__(self, name, flags=0, mode=0o600, size=0, read_only=False):
        """
        Create or open a named shared memory segment.
        
        Parameters:
        - name: str or None. If None, a random name is chosen. If str, should start with '/' (e.g., '/my_shared_memory')
        - flags: int, creation flags (O_CREAT, O_EXCL, O_CREX, O_TRUNC)
        - mode: int, permissions (octal, default 0o600)
        - size: int, size in bytes (0 to use existing size)
        - read_only: bool, whether to open in read-only mode
        
        Notes:
        - If size > 0, segment will be resized using ftruncate() regardless of new/existing status
        - O_TRUNC can be used to truncate existing segment to zero bytes (not supported on macOS)
        - On macOS, ftruncate() can only be called once during segment lifetime
        """

def close_fd(self):
    """
    Close the file descriptor associated with this SharedMemory object.
    
    Equivalent to calling os.close() on the fd attribute. Must be called explicitly -
    file descriptor is not closed automatically on garbage collection.
    
    Note: Closing the file descriptor has no effect on any mmap objects created from it.
    """

def fileno(self):
    """
    Returns the file descriptor for the shared memory segment.
    
    Returns:
    int: File descriptor (same as the fd property)
    
    This method allows SharedMemory objects to be used with functions
    that expect file-like objects with a fileno() method.
    """

def unlink(self):
    """
    Mark the shared memory segment for destruction.
    
    The segment is destroyed once all processes have unmapped it. Per POSIX spec,
    after unlinking, subsequent shm_open() calls with the same name will either
    fail (if O_CREAT not set) or create a new segment (if O_CREAT set).
    
    Note: Implementation behavior may vary across operating systems.
    """

def __str__(self):
    """
    String representation of the shared memory segment.
    
    Returns:
    str: Human-readable representation including name and size
    """

def __repr__(self):
    """
    Detailed string representation for debugging.
    
    Returns:
    str: Technical representation with class name and key attributes
    """

@property
def name(self):
    """
    The name provided in the constructor.
    
    Returns:
    str: Shared memory segment name
    """

@property
def mode(self):
    """
    The mode (permissions) provided in the constructor.
    
    Returns:
    int: File mode/permissions (e.g., 0o600)
    """

@property
def fd(self):
    """
    File descriptor representing the shared memory segment.
    
    Returns:
    int: File descriptor that can be used with os.read(), os.write(), mmap.mmap(), etc.
    """

@property
def size(self):
    """
    Current size of the shared memory segment in bytes.
    
    Returns:
    int: Size in bytes
    """

def unlink_shared_memory(name):
    """
    Convenience function to unlink a shared memory segment by name.
    
    Parameters:
    - name: str, shared memory segment name (e.g., '/my_shared_memory')
    
    Equivalent to opening the segment and calling unlink(), but more convenient
    when you only need to remove an existing segment.
    """

import posix_ipc
import os

# Create a 1KB shared memory segment
shm = posix_ipc.SharedMemory('/my_shm', posix_ipc.O_CREAT, size=1024)

# Write data using file descriptor
data = b'Hello, shared memory!'
os.write(shm.fd, data)

# Read data back
os.lseek(shm.fd, 0, os.SEEK_SET)  # Reset to beginning
read_data = os.read(shm.fd, len(data))
print(f"Read: {read_data.decode()}")

# Clean up
shm.close_fd()
shm.unlink()

import posix_ipc
import mmap
import struct

# Create shared memory segment
shm = posix_ipc.SharedMemory('/mapped_shm', posix_ipc.O_CREAT, size=4096)

# Memory map the segment
mm = mmap.mmap(shm.fd, shm.size)

# Write structured data directly to memory
mm[0:4] = struct.pack('i', 42)        # Write integer at offset 0
mm[4:12] = b'Hello!!!!'               # Write string at offset 4

# Read structured data
value = struct.unpack('i', mm[0:4])[0]  # Read integer
text = mm[4:12].rstrip(b'!')           # Read string

print(f"Value: {value}, Text: {text.decode()}")

# Clean up
mm.close()
shm.close_fd()
shm.unlink()

import posix_ipc
import mmap
import os
import time

# Producer process
def producer():
    # Create shared memory
    shm = posix_ipc.SharedMemory('/data_shm', posix_ipc.O_CREAT, size=1024)
    mm = mmap.mmap(shm.fd, shm.size)
    
    # Write data periodically
    for i in range(10):
        data = f"Message {i}".encode()
        mm[0:len(data)] = data
        mm[len(data)] = 0  # Null terminator
        time.sleep(1)
    
    mm.close()
    shm.close_fd()
    # Don't unlink - consumer will do it

# Consumer process
def consumer():
    # Open existing shared memory
    try:
        shm = posix_ipc.SharedMemory('/data_shm')
        mm = mmap.mmap(shm.fd, shm.size)
        
        # Read data
        for i in range(10):
            # Find null terminator
            null_pos = mm.find(b'\x00')
            if null_pos > 0:
                data = mm[0:null_pos].decode()
                print(f"Consumer read: {data}")
            time.sleep(1)
        
        mm.close()
        shm.close_fd()
        shm.unlink()  # Consumer cleans up
        
    except posix_ipc.ExistentialError:
        print("Shared memory not found")

import posix_ipc
import os

# Create segment with initial size
shm = posix_ipc.SharedMemory('/resizable_shm', posix_ipc.O_CREAT, size=512)
print(f"Initial size: {shm.size}")

# Resize using os.ftruncate() (alternative to size parameter)
os.ftruncate(shm.fd, 2048)
print(f"Resized to: {shm.size}")

# Note: On macOS, you can only call ftruncate() once per segment lifetime

shm.close_fd()
shm.unlink()

import posix_ipc
import mmap

# First process creates and writes data
shm_writer = posix_ipc.SharedMemory('/readonly_shm', posix_ipc.O_CREAT, size=1024)
mm_writer = mmap.mmap(shm_writer.fd, shm_writer.size)
mm_writer[0:12] = b'Hello World!'
mm_writer.close()
shm_writer.close_fd()

# Second process opens read-only
shm_reader = posix_ipc.SharedMemory('/readonly_shm', read_only=True)
mm_reader = mmap.mmap(shm_reader.fd, shm_reader.size, access=mmap.ACCESS_READ)

# Read data
data = mm_reader[0:12]
print(f"Read: {data.decode()}")

# Clean up
mm_reader.close()
shm_reader.close_fd()
posix_ipc.unlink_shared_memory('/readonly_shm')

import posix_ipc

try:
    # Try to open non-existent shared memory
    shm = posix_ipc.SharedMemory('/nonexistent')
except posix_ipc.ExistentialError:
    print("Shared memory does not exist")

try:
    # Try to create exclusive segment that already exists
    shm1 = posix_ipc.SharedMemory('/existing', posix_ipc.O_CREAT)
    shm2 = posix_ipc.SharedMemory('/existing', posix_ipc.O_CREX)
except posix_ipc.ExistentialError:
    print("Shared memory already exists")
    shm1.close_fd()
    shm1.unlink()

try:
    # Try to access segment without proper permissions
    shm = posix_ipc.SharedMemory('/restricted', posix_ipc.O_CREAT, mode=0o000)
    # Another process without permissions would get PermissionsError
except posix_ipc.PermissionsError:
    print("Permission denied")