Multiprocessing¶

from pathos.multiprocessing import ProcessingPool as Pool
from pathos.multiprocessing import Process
import multiprocessing as mp

def start_arr_manager(arr_config, shutdown_event):
    """Entry point for Arr manager process."""
    manager = create_arr_manager(arr_config)
    manager.run_loop(shutdown_event)

def main():
    # Create shutdown event (shared between processes)
    manager = mp.Manager()
    shutdown_event = manager.Event()

    processes = []

    # Start WebUI process
    webui_process = Process(
        target=start_webui,
        args=(CONFIG, shutdown_event),
        name="WebUI"
    )
    webui_process.start()
    processes.append(webui_process)

    # Start Arr manager processes
    for arr_config in CONFIG.get_arr_instances():
        arr_process = Process(
            target=start_arr_manager,
            args=(arr_config, shutdown_event),
            name=f"ArrManager-{arr_config.Name}"
        )
        arr_process.start()
        processes.append(arr_process)

    # Monitor processes
    monitor_processes(processes, shutdown_event)

import multiprocessing as mp

# Create shared event for shutdown coordination
manager = mp.Manager()
shutdown_event = manager.Event()

# In main process
shutdown_event.set()  # Signal all processes to stop

# In child process
while not shutdown_event.is_set():
    # Keep running
    pass

import threading

# Global lock for database access
db_lock = threading.RLock()

@contextmanager
def locked_database():
    """Context manager for safe database access."""
    with db_lock:
        yield

from qBitrr.db_lock import locked_database

# Process A
with locked_database():
    DownloadsModel.create(hash="abc", name="Movie")

# Process B (waits for Process A)
with locked_database():
    download = DownloadsModel.get(hash="abc")

import signal

def signal_handler(signum, frame):
    """Handle shutdown signals."""
    logger.info(f"Received signal {signum}, shutting down...")

    # Signal all processes to stop
    shutdown_event.set()

    # Wait for processes to finish
    for process in processes:
        process.join(timeout=30)

        if process.is_alive():
            logger.warning(f"{process.name} did not stop, forcing...")
            process.terminate()
            process.join(timeout=5)

            if process.is_alive():
                logger.error(f"{process.name} won't die, killing...")
                process.kill()

    logger.info("All processes stopped")
    sys.exit(0)

# Register handlers
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)

def monitor_processes(processes, shutdown_event):
    """Monitor child processes and restart on failure."""
    while not shutdown_event.is_set():
        for i, process in enumerate(processes):
            if not process.is_alive() and not shutdown_event.is_set():
                logger.error(f"{process.name} died, restarting...")

                # Create new process with same config
                new_process = Process(
                    target=process._target,
                    args=process._args,
                    name=process.name
                )
                new_process.start()
                processes[i] = new_process

        time.sleep(30)  # Check every 30 seconds

import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    filename=log_path,
    maxBytes=10485760,  # 10 MB
    backupCount=5,
    mode='a'
)
handler.setLevel(logging.DEBUG)
handler.setFormatter(formatter)

# Thread-safe by default
logger.addHandler(handler)

def start_arr_manager(arr_config, shutdown_event):
    """Process entry point with error handling."""
    try:
        manager = create_arr_manager(arr_config)
        manager.run_loop(shutdown_event)
    except Exception as e:
        logger.exception(f"Fatal error in {arr_config.Name}: {e}")
        # Process exits, main process will restart it
        sys.exit(1)

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Per-process session
session = requests.Session()

# Connection pooling
adapter = HTTPAdapter(
    pool_connections=10,
    pool_maxsize=20,
    max_retries=Retry(total=3, backoff_factor=1)
)

session.mount('http://', adapter)
session.mount('https://', adapter)