Merge pull request #424 from realpython/python-parallel-processing

Materials for Bypassing the GIL

python-parallel-processing/01_java_vs_python/Fibonacci.java

@@ -0,0 +1,12 @@
+public class Fibonacci {
+    public static void main(String[] args) {
+        int cpus = Runtime.getRuntime().availableProcessors();
+        for (int i = 0; i < cpus; i++) {
+            new Thread(() -> fib(45)).start();
+        }
+    }
+    private static int fib(int n) {
+        return n < 2 ? n : fib(n - 2) + fib(n - 1);
+    }
+}
+

python-parallel-processing/01_java_vs_python/fibonacci.py

@@ -0,0 +1,10 @@
+import os
+import threading
+
+
+def fib(n):
+    return n if n < 2 else fib(n - 2) + fib(n - 1)
+
+
+for _ in range(os.cpu_count()):
+    threading.Thread(target=fib, args=(35,)).start()

python-parallel-processing/02_process-based_parallelism/01_fibonacci_multiprocessing.py

@@ -0,0 +1,10 @@
+import multiprocessing
+
+
+def fib(n):
+    return n if n < 2 else fib(n - 2) + fib(n - 1)
+
+
+if __name__ == "__main__":
+    for _ in range(multiprocessing.cpu_count()):
+        multiprocessing.Process(target=fib, args=(35,)).start()

python-parallel-processing/02_process-based_parallelism/02_fibonacci_multiprocessing_pool.py

@@ -0,0 +1,12 @@
+import multiprocessing
+
+
+def fib(n):
+    return n if n < 2 else fib(n - 2) + fib(n - 1)
+
+
+if __name__ == "__main__":
+    with multiprocessing.Pool(processes=4) as pool:
+        results = pool.map(fib, range(40))
+        for i, result in enumerate(results):
+            print(f"fib({i}) = {result}")

python-parallel-processing/02_process-based_parallelism/03_fibonacci_concurrent_futures_process_pool.py

@@ -0,0 +1,12 @@
+from concurrent.futures import ProcessPoolExecutor
+
+
+def fib(n):
+    return n if n < 2 else fib(n - 2) + fib(n - 1)
+
+
+if __name__ == "__main__":
+    with ProcessPoolExecutor(max_workers=4) as executor:
+        results = executor.map(fib, range(40))
+        for i, result in enumerate(results):
+            print(f"fib({i}) = {result}")

python-parallel-processing/02_process-based_parallelism/04_echo_benchmark.py

@@ -0,0 +1,17 @@
+import time
+from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
+
+
+def echo(data):
+    return data
+
+
+if __name__ == "__main__":
+    data = [complex(i, i) for i in range(15_000_000)]
+    for executor in ThreadPoolExecutor(), ProcessPoolExecutor():
+        t1 = time.perf_counter()
+        with executor:
+            future = executor.submit(echo, data)
+            future.result()
+        t2 = time.perf_counter()
+        print(f"{type(executor).__name__:>20s}: {t2 - t1:.2f}s")

python-parallel-processing/03_gil-immune_library/numpy_threads.py

@@ -0,0 +1,5 @@
+import numpy as np
+
+rng = np.random.default_rng()
+matrix = rng.random(size=(5000, 5000))
+matrix @ matrix

python-parallel-processing/03_gil-immune_library/requirements.txt

@@ -0,0 +1,2 @@
+numpy
+

python-parallel-processing/04_extension_module/compile.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+gcc $(python3-config --cflags) -shared -fPIC -O3 -o fibmodule.so fibmodule.c
+

python-parallel-processing/04_extension_module/fibmodule.c

@@ -0,0 +1,37 @@
+#include <Python.h>
+
+int fib(int n) {
+    return n < 2 ? n : fib(n - 2) + fib(n - 1);
+}
+
+static PyObject* fibmodule_fib(PyObject* self, PyObject* args) {
+    int n, result;
+
+    if (!PyArg_ParseTuple(args, "i", &n)) {
+        return NULL;
+    }
+
+    Py_BEGIN_ALLOW_THREADS
+    result = fib(n);
+    Py_END_ALLOW_THREADS
+
+    return Py_BuildValue("i", result);
+}
+
+static PyMethodDef fib_methods[] = {
+    {"fib", fibmodule_fib, METH_VARARGS, "Calculate the nth Fibonacci"},
+    {NULL, NULL, 0, NULL}
+};
+
+static struct PyModuleDef fibmodule = {
+    PyModuleDef_HEAD_INIT,
+    "fibmodule",
+    "Efficient Fibonacci number calculator",
+    -1,
+    fib_methods
+};
+
+PyMODINIT_FUNC PyInit_fibmodule(void) {
+    return PyModule_Create(&fibmodule);
+}
+

python-parallel-processing/04_extension_module/fibonacci_ext.py

@@ -0,0 +1,7 @@
+import os
+import threading
+
+import fibmodule
+
+for _ in range(os.cpu_count()):
+    threading.Thread(target=fibmodule.fib, args=(45,)).start()

python-parallel-processing/05_extension_module_cython/compile.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+cythonize --inplace --annotate -3 fibmodule.py
+

python-parallel-processing/05_extension_module_cython/fibmodule.py

@@ -0,0 +1,20 @@
+import cython
+
+
+@cython.ccall
+def fib(n: cython.int) -> cython.int:
+    with cython.nogil:
+        return _fib(n)
+
+
+@cython.cfunc
+@cython.nogil
+@cython.exceptval(check=False)
+def _fib(n: cython.int) -> cython.int:
+    return n if n < 2 else _fib(n - 2) + _fib(n - 1)
+
+
+if cython.compiled:
+    print("Cython compiled this module")
+else:
+    print("Cython didn't compile this module")

python-parallel-processing/05_extension_module_cython/fibmodule.pyx

@@ -0,0 +1,8 @@
+cpdef int fib(int n):
+    with nogil:
+        return _fib(n)
+
+
+cdef int _fib(int n) noexcept nogil:
+    return n if n < 2 else _fib(n - 2) + _fib(n - 1)
+

python-parallel-processing/05_extension_module_cython/fibonacci_ext.py

@@ -0,0 +1,7 @@
+import os
+import threading
+
+import fibmodule
+
+for _ in range(os.cpu_count()):
+    threading.Thread(target=fibmodule.fib, args=(45,)).start()

python-parallel-processing/05_extension_module_cython/requirements.txt

@@ -0,0 +1,2 @@
+cython
+

python-parallel-processing/06_ctypes/compile.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+gcc -shared -fPIC -O3 -o fibonacci.so fibonacci.c

python-parallel-processing/06_ctypes/fibonacci.c

@@ -0,0 +1,3 @@
+int fib(int n) {
+    return n < 2 ? n : fib(n - 2) + fib(n - 1);
+}

python-parallel-processing/06_ctypes/fibonacci_ctypes.py

@@ -0,0 +1,12 @@
+import ctypes
+import os
+import threading
+
+fibonacci = ctypes.CDLL("./fibonacci.so")
+
+fib = fibonacci.fib
+fib.argtypes = (ctypes.c_int,)
+fib.restype = ctypes.c_int
+
+for _ in range(os.cpu_count()):
+    threading.Thread(target=fib, args=(45,)).start()

python-parallel-processing/07_image_processing/image_processing.py

@@ -0,0 +1,126 @@
+# image_processing.py
+
+import argparse
+import pathlib
+import time
+import tkinter as tk
+import tkinter.ttk as ttk
+
+import numpy as np
+import PIL.Image
+import PIL.ImageTk
+
+import parallel
+
+
+class AppWindow(tk.Tk):
+    def __init__(self, image: PIL.Image.Image) -> None:
+        super().__init__()
+
+        # Main window
+        self.title("Exposure and Gamma Correction")
+        self.resizable(False, False)
+
+        # Parameters frame
+        self.frame = ttk.LabelFrame(self, text="Parameters")
+        self.frame.pack(fill=tk.X, padx=10, pady=10)
+        self.frame.columnconfigure(0, weight=0)
+        self.frame.columnconfigure(1, weight=1)
+
+        # EV slider
+        self.var_ev = tk.DoubleVar(value=0)
+        ev_label = ttk.Label(self.frame, text="Exposure:")
+        ev_label.grid(row=0, column=0, sticky=tk.W, padx=10, pady=10)
+        ev_slider = ttk.Scale(
+            self.frame,
+            from_=-1,
+            to=1,
+            orient=tk.HORIZONTAL,
+            variable=self.var_ev,
+        )
+        ev_slider.bind("<B1-Motion>", self.on_slide)
+        ev_slider.grid(row=0, column=1, sticky=tk.W + tk.E, padx=10, pady=10)
+
+        # Gamma slider
+        self.var_gamma = tk.DoubleVar(value=1)
+        gamma_label = ttk.Label(self.frame, text="Gamma:")
+        gamma_label.grid(row=1, column=0, sticky=tk.W, padx=10, pady=10)
+        gamma_slider = ttk.Scale(
+            self.frame,
+            from_=0.1,
+            to=2,
+            orient=tk.HORIZONTAL,
+            variable=self.var_gamma,
+        )
+        gamma_slider.bind("<B1-Motion>", self.on_slide)
+        gamma_slider.grid(
+            row=1, column=1, sticky=tk.W + tk.E, padx=10, pady=10
+        )
+
+        # Image preview
+        self.preview = ttk.Label(self, relief=tk.SUNKEN)
+        self.preview.pack(padx=10, pady=10)
+
+        # Status bar
+        self.var_status = tk.StringVar()
+        status_bar = ttk.Label(
+            self, anchor=tk.W, relief=tk.SUNKEN, textvariable=self.var_status
+        )
+        status_bar.pack(side=tk.BOTTOM, fill=tk.X)
+
+        # Image pixels
+        self.pixels = np.array(image)
+        self.update()
+        self.show_preview(image)
+
+        self.mainloop()
+
+    def on_slide(self, *args, **kwargs) -> None:
+        # Get parameters
+        ev = 2.0 ** self.var_ev.get()
+        gamma = 1.0 / self.var_gamma.get()
+
+        # Process pixels
+        t1 = time.perf_counter()
+        pixels = self.pixels.copy()
+        parallel.process(pixels, ev, gamma)
+        t2 = time.perf_counter()
+
+        # Render preview
+        image = PIL.Image.fromarray(pixels)
+        self.show_preview(image)
+        t3 = time.perf_counter()
+
+        # Update status
+        self.var_status.set(
+            f"Processed in {(t2 - t1) * 1000:.0f} ms "
+            f"(Rendered in {(t3 - t1) * 1000:.0f} ms)"
+        )
+
+    def show_preview(self, image: PIL.Image.Image) -> None:
+        scale = 0.75
+        offset = 2.0 * self.frame.winfo_height()
+        image.thumbnail(
+            (
+                int(self.winfo_screenwidth() * scale),
+                int(self.winfo_screenheight() * scale - offset),
+            )
+        )
+        image_tk = PIL.ImageTk.PhotoImage(image)
+        self.preview.configure(image=image_tk)
+        self.preview.image = image_tk
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("image_path", type=pathlib.Path)
+    return parser.parse_args()
+
+
+def main(args: argparse.Namespace) -> None:
+    with PIL.Image.open(args.image_path) as image:
+        AppWindow(image)
+
+
+if __name__ == "__main__":
+    main(parse_args())