Demo of `multiprocessing.RLock` with context managers

rlock-demo.py

# To run this demo:
#   uv run rlock-demo.py

# /// script
# requires-python = ">=3.13"
# dependencies = ["rich"]
# ///

import multiprocessing
import time
from multiprocessing import RLock

from rich.console import Console

SLEEP_TIME: int = 3
global_lock = RLock()

console = Console()


def demo_rlock_behavior(process_id: int, color: str) -> None:
    """Demonstrates the usage of a recursive lock (RLock) in a single process."""

    def log(msg: str) -> None:
        timestamp = time.strftime("%M:%S")
        prefix = f"{timestamp} Process {process_id:02}: "
        console.print(prefix + msg, style=color)

    log("\tAcquiring the lock for the first time (blocking).")
    with global_lock:
        log("\t\tLock acquired for the first time.")
        time.sleep(SLEEP_TIME)
        log("\t\tAttempting to acquire the lock again (should work immediately).")

        # attempt to acquire it again
        # this works because it's a recursive lock.
        with global_lock:
            log("\t\t\tLock acquired for the second time.")
            time.sleep(SLEEP_TIME)
        log("\t\tSecond lock released.")

        time.sleep(SLEEP_TIME)

    log("\tFirst lock released.")


def main() -> None:
    """Main entry point for the script."""

    console.print("Running RLock demo...", style="bold")

    process_one = multiprocessing.Process(target=demo_rlock_behavior, args=(1, "red"))
    process_two = multiprocessing.Process(target=demo_rlock_behavior, args=(2, "blue"))
    all_processes = [process_one, process_two]
    _ = [process.start() for process in all_processes]
    _ = [process.join() for process in all_processes]

    console.print("RLock demo complete.", style="bold")


if __name__ == "__main__":
    main()