examples for ch.20: Concurrency Models

20-concurrency/primes/primes.py

@@ -0,0 +1,51 @@
+import math
+import itertools
+
+
+PRIME_FIXTURE = [
+    (2, True),
+    (142702110479723, True),
+    (299593572317531, True),
+    (3333333333333301, True),
+    (3333333333333333, False),
+    (3333335652092209, False),
+    (4444444444444423, True),
+    (4444444444444444, False),
+    (4444444488888889, False),
+    (5555553133149889, False),
+    (5555555555555503, True),
+    (5555555555555555, False),
+    (6666666666666666, False),
+    (6666666666666719, True),
+    (6666667141414921, False),
+    (7777777536340681, False),
+    (7777777777777753, True),
+    (7777777777777777, False),
+    (9999999999999917, True),
+    (9999999999999999, False),
+]
+
+NUMBERS = [n for n, _ in PRIME_FIXTURE]
+
+# tag::IS_PRIME[]
+def is_prime(n) -> bool:
+    if n < 2:
+        return False
+    if n == 2:
+        return True
+    if n % 2 == 0:
+        return False
+
+    root = int(math.floor(math.sqrt(n)))
+    for i in range(3, root + 1, 2):
+        if n % i == 0:
+            return False
+    return True
+# end::IS_PRIME[]
+
+if __name__ == '__main__':
+
+    for n, prime in PRIME_FIXTURE:
+        prime_res = is_prime(n)
+        assert prime_res == prime
+        print(n, prime)

20-concurrency/primes/procs.py

@@ -0,0 +1,46 @@
+# tag::PRIMES_PROC_TOP[]
+from time import perf_counter
+from typing import Tuple, List, NamedTuple
+from multiprocessing import Process, SimpleQueue  # <1>
+from multiprocessing import queues  # <2>
+
+from primes import is_prime, NUMBERS
+
+class Result(NamedTuple):  # <3>
+    flag: bool
+    elapsed: float
+
+JobQueue = queues.SimpleQueue[Tuple[int, Result]]  # <4>
+
+def check(n: int) -> Result:  # <5>
+    t0 = perf_counter()
+    res = is_prime(n)
+    return Result(res, perf_counter() - t0)
+
+def job(n: int, results: JobQueue) -> None:  # <6>
+    results.put((n, check(n)))  # <7>
+# end::PRIMES_PROC_TOP[]
+
+# tag::PRIMES_PROC_MAIN[]
+def main() -> None:
+    t0 = perf_counter()
+    results: JobQueue = SimpleQueue()  # <1>
+    workers: List[Process] = []  # <2>
+
+    for n in NUMBERS:
+        worker = Process(target=job, args=(n, results))  # <3>
+        worker.start()  # <4>
+        workers.append(worker)  # <5>
+
+    for _ in workers:  # <6>
+        n, (prime, elapsed) = results.get()  # <7>
+        label = 'P' if prime else ' '
+        print(f'{n:16}  {label} {elapsed:9.6f}s')
+
+
+    time = perf_counter() - t0
+    print('Total time:', f'{time:0.2f}s')
+
+if __name__ == '__main__':
+    main()
+# end::PRIMES_PROC_MAIN[]

20-concurrency/primes/sequential.py

@@ -0,0 +1,26 @@
+from time import perf_counter
+from typing import NamedTuple
+
+from primes import is_prime, NUMBERS
+
+class Result(NamedTuple):  # <1>
+    flag: bool
+    elapsed: float
+
+def check(n: int) -> Result:  # <2>
+    t0 = perf_counter()
+    flag = is_prime(n)
+    return Result(flag, perf_counter() - t0)
+
+def main() -> None:
+    t0 = perf_counter()
+    for n in NUMBERS:  # <3>
+        prime, elapsed = check(n)
+        label = 'P' if prime else ' '
+        print(f'{n:16}  {label} {elapsed:9.6f}s')
+
+    elapsed = perf_counter() - t0  # <4>
+    print('Total time:', f'{elapsed:0.2f}s')
+
+if __name__ == '__main__':
+    main()

20-concurrency/primes/spinner_async_nap.py

@@ -0,0 +1,58 @@
+# spinner_async_experiment.py
+
+# credits: Example by Luciano Ramalho inspired by
+# Michele Simionato's multiprocessing example in the python-list:
+# https://mail.python.org/pipermail/python-list/2009-February/675659.html
+
+import asyncio
+import itertools
+import math
+
+# tag::SPINNER_ASYNC_NAP[]
+async def is_prime(n):
+    if n < 2:
+        return False
+    if n == 2:
+        return True
+    if n % 2 == 0:
+        return False
+
+    sleep = asyncio.sleep  # <1>
+    root = int(math.floor(math.sqrt(n)))
+    for i in range(3, root + 1, 2):
+        if n % i == 0:
+            return False
+        if i % 100_000 == 1:  # <2>
+            await sleep(0)
+    return True
+# end::SPINNER_ASYNC_NAP[]
+
+
+async def spin(msg: str) -> None:
+    for char in itertools.cycle(r'\|/-'):
+        status = f'\r{char} {msg}'
+        print(status, flush=True, end='')
+        try:
+            await asyncio.sleep(.1)
+        except asyncio.CancelledError:
+            break
+    blanks = ' ' * len(status)
+    print(f'\r{blanks}\r', end='')
+
+async def slow() -> int:
+    await is_prime(5_000_111_000_222_021)  # <4>
+    return 42
+
+async def supervisor() -> int:
+    spinner = asyncio.create_task(spin('thinking!'))  # <1>
+    print('spinner object:', spinner)  # <2>
+    result = await slow()  # <3>
+    spinner.cancel()  # <5>
+    return result
+
+def main() -> None:
+    result = asyncio.run(supervisor())
+    print('Answer:', result)
+
+if __name__ == '__main__':
+    main()

20-concurrency/primes/spinner_async_prime_no_spin.py

@@ -0,0 +1,40 @@
+# spinner_async_experiment.py
+
+# credits: Example by Luciano Ramalho inspired by
+# Michele Simionato's multiprocessing example in the python-list:
+# https://mail.python.org/pipermail/python-list/2009-February/675659.html
+
+import asyncio
+import itertools
+
+import primes
+
+async def spin(msg: str) -> None:
+    for char in itertools.cycle(r'\|/-'):
+        status = f'\r{char} {msg}'
+        print(status, flush=True, end='')
+        try:
+            await asyncio.sleep(.1)
+        except asyncio.CancelledError:
+            break
+    print('THIS WILL NEVER BE OUTPUT')
+
+# tag::SPINNER_ASYNC_EXPERIMENT[]
+async def slow() -> int:
+    primes.is_prime(5_000_111_000_222_021)  # <4>
+    return 42
+
+async def supervisor() -> int:
+    spinner = asyncio.create_task(spin('thinking!'))  # <1>
+    print('spinner object:', spinner)  # <2>
+    result = await slow()  # <3>
+    spinner.cancel()  # <5>
+    return result
+# end::SPINNER_ASYNC_EXPERIMENT[]
+
+def main() -> None:
+    result = asyncio.run(supervisor())
+    print('Answer:', result)
+
+if __name__ == '__main__':
+    main()

20-concurrency/primes/spinner_thread.py

@@ -0,0 +1,46 @@
+# spinner_thread.py
+
+# credits: Adapted from Michele Simionato's
+# multiprocessing example in the python-list:
+# https://mail.python.org/pipermail/python-list/2009-February/675659.html
+
+# tag::SPINNER_THREAD_TOP[]
+from threading import Thread, Event
+import itertools
+import time
+
+from primes import is_prime
+
+def spin(msg: str, done: Event) -> None:  # <1>
+    for char in itertools.cycle(r'\|/-'):  # <2>
+        status = f'\r{char} {msg}'  # <3>
+        print(status, end='', flush=True)
+        if done.wait(.1):  # <4>
+            break  # <5>
+    blanks = ' ' * len(status)
+    print(f'\r{blanks}\r', end='')  # <6>
+
+def slow() -> int:
+    is_prime(5_000_111_000_222_021)  # <7>
+    return 42
+# end::SPINNER_THREAD_TOP[]
+
+# tag::SPINNER_THREAD_REST[]
+def supervisor() -> int:  # <1>
+    done = Event()  # <2>
+    spinner = Thread(target=spin,
+                     args=('thinking!', done))  # <3>
+    print('spinner object:', spinner)  # <4>
+    spinner.start()  # <5>
+    result = slow()  # <6>
+    done.set()  # <7>
+    spinner.join()  # <8>
+    return result
+
+def main() -> None:
+    result = supervisor()  # <9>
+    print('Answer:', result)
+
+if __name__ == '__main__':
+    main()
+# end::SPINNER_THREAD_REST[]

20-concurrency/primes/threads.py

@@ -0,0 +1,41 @@
+from time import perf_counter
+from typing import Tuple, List, NamedTuple
+from threading import Thread
+from queue import SimpleQueue
+
+from primes import is_prime, NUMBERS
+
+class Result(NamedTuple):  # <3>
+    flag: bool
+    elapsed: float
+
+JobQueue = SimpleQueue[Tuple[int, Result]]  # <4>
+
+def check(n: int) -> Result:  # <5>
+    t0 = perf_counter()
+    res = is_prime(n)
+    return Result(res, perf_counter() - t0)
+
+def job(n: int, results: JobQueue) -> None:  # <6>
+    results.put((n, check(n)))  # <7>
+
+def main() -> None:
+    t0 = perf_counter()
+    results: JobQueue = SimpleQueue()  # <1>
+    workers: List[Thread] = []  # <2>
+
+    for n in NUMBERS:
+        worker = Thread(target=job, args=(n, results))  # <3>
+        worker.start()  # <4>
+        workers.append(worker)  # <5>
+
+    for _ in workers:  # <6>
+        n, (prime, elapsed) = results.get()  # <7>
+        label = 'P' if prime else ' '
+        print(f'{n:16}  {label} {elapsed:9.6f}s')
+
+    time = perf_counter() - t0
+    print('Total time:', f'{time:0.2f}s')
+
+if __name__ == '__main__':
+    main()