Merge pull request #605 from realpython/python-concurrency

Materials for Python Concurrency [UPDATE]

python-concurrency/README.md

@@ -0,0 +1,13 @@
+# Speed Up Your Python Program With Concurrency
+
+This folder contains the sample code for the [Speed Up Your Python Program With Concurrency](https://realpython.com/python-concurrency/) tutorial.
+
+To use this code, first create and activate a [virtual environment](https://realpython.com/python-virtual-environments-a-primer/). Then, install the required libraries into it:
+
+```sh
+$ python -m venv venv/
+$ source venv/bin/activate
+(venv) $ python -m pip install -r requirements.txt
+```
+
+This will ensure you have the required Python packages.

python-concurrency/cpu_asyncio.py

@@ -0,0 +1,18 @@
+import asyncio
+import time
+
+
+async def main():
+    start_time = time.perf_counter()
+    tasks = [fib(35) for _ in range(20)]
+    await asyncio.gather(*tasks, return_exceptions=True)
+    duration = time.perf_counter() - start_time
+    print(f"Computed in {duration} seconds")
+
+
+async def fib(n):
+    return n if n < 2 else await fib(n - 2) + await fib(n - 1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python-concurrency/cpu_non_concurrent.py

@@ -0,0 +1,17 @@
+import time
+
+
+def main():
+    start_time = time.perf_counter()
+    for _ in range(20):
+        fib(35)
+    duration = time.perf_counter() - start_time
+    print(f"Computed in {duration} seconds")
+
+
+def fib(n):
+    return n if n < 2 else fib(n - 2) + fib(n - 1)
+
+
+if __name__ == "__main__":
+    main()

python-concurrency/cpu_processes.py

@@ -0,0 +1,18 @@
+import concurrent.futures
+import time
+
+
+def main():
+    start_time = time.perf_counter()
+    with concurrent.futures.ProcessPoolExecutor() as executor:
+        executor.map(fib, [35] * 20)
+    duration = time.perf_counter() - start_time
+    print(f"Computed in {duration} seconds")
+
+
+def fib(n):
+    return n if n < 2 else fib(n - 2) + fib(n - 1)
+
+
+if __name__ == "__main__":
+    main()

python-concurrency/cpu_threads.py

@@ -0,0 +1,18 @@
+import concurrent.futures
+import time
+
+
+def main():
+    start_time = time.perf_counter()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
+        executor.map(fib, [35] * 20)
+    duration = time.perf_counter() - start_time
+    print(f"Computed in {duration} seconds")
+
+
+def fib(n):
+    return n if n < 2 else fib(n - 2) + fib(n - 1)
+
+
+if __name__ == "__main__":
+    main()

concurrency-overview/io_asyncio.py → python-concurrency/io_asyncio.py

@@ -1,30 +1,30 @@
-#!/usr/bin/env python3
 import asyncio
 import time
 
 import aiohttp
 
 
-async def download_site(session, url):
-    async with session.get(url) as response:
-        print("Read {0} from {1}".format(response.content_length, url))
+async def main():
+    sites = [
+        "https://www.jython.org",
+        "http://olympus.realpython.org/dice",
+    ] * 80
+    start_time = time.perf_counter()
+    await download_all_sites(sites)
+    duration = time.perf_counter() - start_time
+    print(f"Downloaded {len(sites)} sites in {duration} seconds")
 
 
 async def download_all_sites(sites):
     async with aiohttp.ClientSession() as session:
-        tasks = []
-        for url in sites:
-            task = asyncio.ensure_future(download_site(session, url))
-            tasks.append(task)
+        tasks = [download_site(url, session) for url in sites]
         await asyncio.gather(*tasks, return_exceptions=True)
 
 
+async def download_site(url, session):
+    async with session.get(url) as response:
+        print(f"Read {len(await response.read())} bytes from {url}")
+
+
 if __name__ == "__main__":
-    sites = [
-        "https://www.jython.org",
-        "http://olympus.realpython.org/dice",
-    ] * 80
-    start_time = time.time()
-    asyncio.get_event_loop().run_until_complete(download_all_sites(sites))
-    duration = time.time() - start_time
-    print(f"Downloaded {len(sites)} sites in {duration} seconds")
+    asyncio.run(main())

concurrency-overview/io_non_concurrent.py → python-concurrency/io_non_concurrent.py

@@ -1,12 +1,17 @@
-#!/usr/bin/env python3
 import time
 
 import requests
 
 
-def download_site(url, session):
-    with session.get(url) as response:
-        print(f"Read {len(response.content)} from {url}")
+def main():
+    sites = [
+        "https://www.jython.org",
+        "http://olympus.realpython.org/dice",
+    ] * 80
+    start_time = time.perf_counter()
+    download_all_sites(sites)
+    duration = time.perf_counter() - start_time
+    print(f"Downloaded {len(sites)} sites in {duration} seconds")
 
 
 def download_all_sites(sites):
@@ -15,12 +20,10 @@ def download_all_sites(sites):
             download_site(url, session)
 
 
+def download_site(url, session):
+    with session.get(url) as response:
+        print(f"Read {len(response.content)} bytes from {url}")
+
+
 if __name__ == "__main__":
-    sites = [
-        "https://www.jython.org",
-        "http://olympus.realpython.org/dice",
-    ] * 80
-    start_time = time.time()
-    download_all_sites(sites)
-    duration = time.time() - start_time
-    print(f"Downloaded {len(sites)} in {duration} seconds")
+    main()

python-concurrency/io_processes.py

@@ -0,0 +1,40 @@
+import atexit
+import multiprocessing
+import time
+from concurrent.futures import ProcessPoolExecutor
+
+import requests
+
+session: requests.Session
+
+
+def main():
+    sites = [
+        "https://www.jython.org",
+        "http://olympus.realpython.org/dice",
+    ] * 80
+    start_time = time.perf_counter()
+    download_all_sites(sites)
+    duration = time.perf_counter() - start_time
+    print(f"Downloaded {len(sites)} sites in {duration} seconds")
+
+
+def download_all_sites(sites):
+    with ProcessPoolExecutor(initializer=init_process) as executor:
+        executor.map(download_site, sites)
+
+
+def download_site(url):
+    with session.get(url) as response:
+        name = multiprocessing.current_process().name
+        print(f"{name}:Read {len(response.content)} bytes from {url}")
+
+
+def init_process():
+    global session
+    session = requests.Session()
+    atexit.register(session.close)
+
+
+if __name__ == "__main__":
+    main()