multiprocessing-logging

import logging import multiprocessing import time from multiprocessing_logging import install_mp_handler import os def worker_function(name): logger = logging.getLogger() logger.info(f"Worker {name} starting (PID: {os.getpid()})") time.sleep(0.5) # Simulate work logger.info(f"Worker {name} finishing (PID: {os.getpid()})") if __name__ == "__main__": # Set up basic logging for the main process first logging.basicConfig( level=os.environ.get('LOG_LEVEL', 'INFO'), format='%(asctime)s - %(processName)s - %(levelname)s - %(message)s' ) logger = logging.getLogger() logger.info("Main process starting.") # IMPORTANT: install_mp_handler BEFORE creating any worker processes or Pool # This library only works with the 'fork' start method. # On some systems (e.g., macOS Python 3.8+, Python 3.14+ on POSIX), # 'fork' is not the default or is considered unsafe. # You might need to explicitly set 'fork' if your system supports it: # multiprocessing.set_start_method('fork', force=True) try: install_mp_handler(logger=logger) logger.info("Multiprocessing logging handler installed.") processes = [] for i in range(3): p = multiprocessing.Process(target=worker_function, args=(f"Task-{i}",)) processes.append(p) p.start() for p in processes: p.join() logger.info("All worker processes completed.") except AssertionError as e: logger.error(f"Failed to install multiprocessing handler: {e}. Check multiprocessing start method.") except Exception as e: logger.error(f"An unexpected error occurred: {e}") logger.info("Main process finishing.")