posix-ipc

import posix_ipc import os # Semaphore Example SEM_NAME = '/my_semaphore' try: sem = posix_ipc.Semaphore(SEM_NAME, posix_ipc.O_CREAT, initial_value=1) print(f"Semaphore '{SEM_NAME}' created/opened. Value: {sem.value}") sem.acquire() print(f"Semaphore acquired. Value: {sem.value}") sem.release() print(f"Semaphore released. Value: {sem.value}") finally: if 'sem' in locals() and sem.name == SEM_NAME: # Ensure it's our semaphore and still valid sem.close() sem.unlink() print(f"Semaphore '{SEM_NAME}' closed and unlinked.") # Shared Memory Example SHM_NAME = '/my_shared_memory' SIZE = 128 try: shm = posix_ipc.SharedMemory(SHM_NAME, posix_ipc.O_CREAT, size=SIZE) print(f"Shared Memory '{SHM_NAME}' created/opened with size {SIZE}.") # In a real app, you'd mmap this and write/read bytes # e.g., memory_map = mmap.mmap(shm.fd, shm.size) # memory_map.write(b'Hello from shared memory!') finally: if 'shm' in locals() and shm.name == SHM_NAME: shm.close_fd() # Close the file descriptor shm.unlink() print(f"Shared Memory '{SHM_NAME}' unlinked.") # Message Queue Example (Note: Message queues are not supported on macOS by posix-ipc) # This example will likely fail on macOS due to OS limitations. # For a runnable example, execute on Linux/WSL. MQ_NAME = '/my_message_queue' if os.uname().sysname != 'Darwin': # Check if not macOS try: mq = posix_ipc.MessageQueue(MQ_NAME, posix_ipc.O_CREAT, max_messages=10, max_message_size=64) print(f"Message Queue '{MQ_NAME}' created/opened.") message = b"Hello from message queue!" mq.send(message) print(f"Sent: {message.decode()}") received_message, _ = mq.receive() print(f"Received: {received_message.decode()}") except posix_ipc.PermissionsError as e: print(f"Caught PermissionsError for MessageQueue: {e}. This might happen if queue exists with different permissions.") except posix_ipc.ExistentialError as e: print(f"Caught ExistentialError for MessageQueue: {e}. This might happen if queue exists with incompatible settings.") except Exception as e: print(f"An unexpected error occurred with MessageQueue: {e}") finally: if 'mq' in locals() and mq.name == MQ_NAME: try: mq.close() mq.unlink() print(f"Message Queue '{MQ_NAME}' closed and unlinked.") except Exception as e: print(f"Error during MessageQueue cleanup: {e}") else: print("MessageQueue example skipped: Not supported on macOS by posix-ipc.")