Skip to content

Instantly share code, notes, and snippets.

@lucaspar

lucaspar/rlock-demo.py

Last active August 6, 2025 14:00
Show Gist options
  • Save lucaspar/9560f11cfefdb19375c22b57f2103c76 to your computer and use it in GitHub Desktop.
Save lucaspar/9560f11cfefdb19375c22b57f2103c76 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. lucaspar revised this gist Aug 6, 2025. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion rlock-demo.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,5 +1,5 @@
    # To run this demo:
    # uv run rlock-demo.py
    # uv run https://gist.github.com/lucaspar/9560f11cfefdb19375c22b57f2103c76

    # /// script
    # requires-python = ">=3.13"
  2. lucaspar revised this gist Jan 11, 2025. 1 changed file with 3 additions and 0 deletions.
    3 changes: 3 additions & 0 deletions rlock-demo.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,3 +1,6 @@
    # To run this demo:
    # uv run rlock-demo.py

    # /// script
    # requires-python = ">=3.13"
    # dependencies = ["rich"]
  3. lucaspar created this gist Jan 11, 2025.
    59 changes: 59 additions & 0 deletions rlock-demo.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,59 @@
    # /// 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()