# Python Coroutine
[TOC]
> Coroutines are computer program components that allow execution to be suspended and resumed, generalizing subroutines for cooperative multitasking.
> Coroutines are well-suited for implementing familiar program components such as cooperative tasks, exceptions, event loops, iterators, infinite lists and pipes.
> - [wiki - Coroutine](https://en.wikipedia.org/wiki/Coroutine)
# Synchronous
一般不涉及 I/O 的 function, 如 `foo1()`, `foo2()` 會依序執行, 簡單
`sync.py`
```python=
#!/usr/bin/env python3
def foo1():
print('foo1')
def foo2():
print('foo2')
def main():
foo1()
foo2()
if __name__ == '__main__':
main()
```
```shell=
python sync.py
foo1
foo2
```
# Synchronous - I/O
當涉及 I/O, 若仍依序執行就會需要長時間等待, 下面用 `load_data()` 模擬長時間讀取檔案的情況
:::spoiler
這邊先用 synchronous 的方式實作 `sleep()`, 等後面介紹 asynchronous 再陸續改善
:::
`lib/sync.py`
```python=
import time
def timeit(func):
def wrapper(*args, **kwargs):
start_time = time.time()
print(f"Starting {time.ctime(start_time)}")
result = func(*args, **kwargs)
end_time = time.time()
print(f"Ending {time.ctime(end_time)}")
print(f"Time taken: {end_time - start_time:.2f} seconds")
return result
return wrapper
def sleep(secs):
now = time.time()
while time.time() < now + secs:
pass
def load_data(file, secs=5):
sleep(secs)
```
`sync.v1.py`
```python=
#!/usr/bin/env python3
from lib import sync
def main():
print('Loading data')
sync.load_data('a.txt')
print('Data loaded')
if __name__ == '__main__':
main()
```
執行後, 我們可以立刻看見 `Loading data`, 但需要等約 5 秒才能看見 `Data loaded`
```shell=
python sync.v1.py
Loading data 👈 print immediately
Data loaded 👈 print after ~5 seconds
```
# Synchronous - blocking
雖然上面 `load_data` 的作法 *執行上* 沒有問題: 我們確實把檔案讀進來了
但是如果我們有其他事情要處理呢?
假設我們要讀兩個檔案
`sync.v2.py`
```python=
#!/usr/bin/env python3
from lib import sync
def load_a():
sync.load_data('a.txt')
print('a.txt loaded')
def load_b():
sync.load_data('b.txt')
print('b.txt loaded')
def main():
print('Loading data')
load_a()
load_b()
print('Data loaded')
if __name__ == '__main__':
main()
```
這次我們依序讀了兩個檔案, 且兩次讀取都需要約 5 秒, 所以總共需要 10 秒
```shell=
Loading data 👈 print immediately
a.txt loaded 👈 print after ~5 seconds
b.txt loaded 👈 print after ~5 seconds
Data loaded 👈 print after ~10 seconds
```
TODO: 圖
可以看到, 由於我們是等 `a.txt` 讀取後才開始讀取 `b.txt`, 所以需要等 10 秒
若有更多的檔案, 或是更多的 I/O 要處理呢?
這樣處理的時間就會是所有 I/O 時間的總和
甚至在比較複雜的情境中, I/O 也會擋住其他任務的執行
那麼, 要怎麼改善呢?
# Parallelism
一種方式是透過平行化 (parallelism), 讓每個任務能夠同時執行
以下介紹兩種平行化的方式: multiprocessing, multithreading
## Multiprocessing
我們讓每個任務都起一個新的 process 處理
可以用 python 的 std [multiprocessing](https://docs.python.org/3/library/multiprocessing.html)
下面將執行所花費的時間印出方便觀察
`sync.v3.py`
```python=
#!/usr/bin/env python3
from multiprocessing import Pool
from lib import (sync, timeit)
def load_data(file):
sync.load_data(file)
print(f"{file} loaded")
@timeit
def main():
print('Loading data')
with Pool(2) as p: # 2 processes to reduce overhead
p.map(load_data, ['a.txt', 'b.txt'])
print('Data loaded')
if __name__ == '__main__':
main()
```
可以看到僅執行了約 5 秒
```shell=
Starting Mon Jul 22 14:03:11 2024
Loading data
a.txt loaded
b.txt loaded
Data loaded
Ending Mon Jul 22 14:03:16 2024
Time taken: 5.20 seconds
```
:::info
`multiprocessing` 的文件中有提到可以用 `concurrent.futures.ProcessPoolExecutor` 取代
後面講到 coroutine 時再介紹
> See also [concurrent.futures.ProcessPoolExecutor](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor) offers a higher level interface to push tasks to a background process without blocking execution of the calling process. Compared to using the Pool interface directly, the [concurrent.futures](https://docs.python.org/3/library/concurrent.futures.html#module-concurrent.futures) API more readily allows the submission of work to the underlying process pool to be separated from waiting for the results.
:::
## Multithreading
Multiprocessing 外, 另一種平行化的方式是 multithreading
由於 python (特指 CPython) 受限於 Global Interpreter Lock (GIL), 每次只會有一個 thread 拿到 lock 並執行
所以 multithreading 無法做到 *真正的* 平行化
其背後實際上是 Cooperative Multitasking 架構
因此本文不另外介紹, 有興趣請參考 [深入 GIL: 如何寫出快速且 thread-safe 的 Python – Grok the GIL: How to write fast and thread-safe Python](https://blog.louie.lu/2017/05/19/%E6%B7%B1%E5%85%A5-gil-%E5%A6%82%E4%BD%95%E5%AF%AB%E5%87%BA%E5%BF%AB%E9%80%9F%E4%B8%94-thread-safe-%E7%9A%84-python-grok-the-gil-how-to-write-fast-and-thread-safe-python/)
## Summary
雖然 multiprocessing 和 multithreading 都能將多個任務平行處理, 但也有各自的 overhead
Multiprocessing
- CPython 用 [fork](https://man7.org/linux/man-pages/man2/fork.2.html) 和 [exec](https://man7.org/linux/man-pages/man3/exec.3.html) 實做
- `fork` 會複製當前的 process 給 child process, 產生額外的 overhead
- Kernel-level context switch
Multithreading
- CPython 用 [pthread](https://man7.org/linux/man-pages/man7/pthreads.7.html) 實作
- 和當前 process 共用 memory space
- kernel-level context switch
- 理論上比 process context switch 快
雖然我們用平行化縮短了執行時間, 但是並未解決 I/O blocking 的問題
# Asynchronous
Process-based, thread-based, generator-based, coroutine
## Process-based
和 multiprocessing 類似, 我們可以 fork 出一個 process 執行 I/O 任務
等待結果時執行其他任務, I/O 結束後再繼續執行
`sync.v4.py`
```python=
#!/usr/bin/env python3
from multiprocessing import Process
from lib import (sync, timeit)
def load_data(file):
p = Process(target=sync.load_data, args=(file,))
p.start()
print(f"{file} loaded")
return p
@timeit
def main():
print('Loading data')
p1 = load_data('a.txt')
p2 = load_data('b.txt')
print('Do other tasks')
p1.join()
print('a.txt done')
p2.join()
print('b.txt done')
print('Data loaded')
if __name__ == '__main__':
main()
```
缺點是需要自己 join
```shell=
Starting Mon Jul 22 16:27:03 2024
Loading data
a.txt loaded
b.txt loaded
Do other tasks
a.txt done
b.txt done
Data loaded
Ending Mon Jul 22 16:27:08 2024
Time taken: 5.14 seconds
```
## Thread-based
和 process-based 類似, 但不需要複製當前的 process, 所以理論上會快一些
`sync.v5.py`
```python=
#!/usr/bin/env python3
from threading import Thread
from lib import (sync, timeit)
def load_data(file):
t = Thread(target=sync.load_data, args=(file,))
t.start()
print(f"{file} loaded")
return t
@timeit
def main():
print('Loading data')
t1 = load_data('a.txt')
t2 = load_data('b.txt')
print('Do other tasks')
t1.join()
print('a.txt done')
t2.join()
print('b.txt done')
print('Data loaded')
if __name__ == '__main__':
main()
```
結果如預期比 process-based 快一點點
```shell=
Starting Mon Jul 22 16:45:03 2024
Loading data
a.txt loaded
b.txt loaded
Do other tasks
a.txt done
b.txt done
Data loaded
Ending Mon Jul 22 16:45:08 2024
Time taken: 5.04 seconds
```
缺點是需要自己 join threads, 且即使有 GIL, 也可能寫出 thread-unsafe function
## Generator-based
### Generator
Python 提供 generator function 讓我們可以暫停 function 的執行並轉移控制權給 caller, 之後再從暫停的地方繼續執行
以 `fib()` generator function 為例, 我們印出每個 fibonacci 的數值
```python=
def fib(n):
a, b = 1, 0
for _ in range(n):
a, b = b, a + b
yield b
def main():
n = 5
gen = fib(n)
while True:
try:
print(next(gen))
except StopIteration:
break
```
```shell=
1
1
2
3
5
```
這個例子是從 `fib` generator function 將每次的結果回傳給 caller
Python 3.5 開始支援 generator [send(value)](https://peps.python.org/pep-0342/#new-generator-method-send-value) method
讓我們可以將數值從 caller 傳給 callee
來改寫一下前面的範例
```python=
def fib(n):
a = 0
b = 0
a = yield a
b = yield b
for _ in range(n):
yield b
a, b = b, a + b
yield b
def main():
n = 5
gen = fib(n)
gen.send(None)
gen.send(1)
gen.send(0)
while True:
try:
print(next(gen))
except StopIteration:
break
```
我們將初始值透過 generator 的 `send()` method 傳給 `fib`, 並印出前 n 個數值
這樣我們就達成了 Generator 間的溝通
### Refactor with generator functions
由於 process-based 和 thread-based 都有佔用資源和 kernel-level context switch overhead 的缺點
我們可以將 `sleep` 和 `load_data` 改寫成 generator function
`lib/async_impl.py`
```python=
def sleep(secs):
now = time.time()
while time.time() < now + secs:
yield
pass
def load_data(file, secs=5):
yield from sleep(secs)
```
`async_util.py`
```python=
from lib import (async_util, timeit)
def load_data(file):
yield from async_util.load_data(file)
print(f"{file} loaded")
@timeit
def main():
print('Loading data')
gen_a = load_data('a.txt')
gen_b = load_data('b.txt')
pending_tasks = [gen_a, gen_b]
tasks = {task: None for task in pending_tasks}
print('Do other tasks')
while pending_tasks:
for task in pending_tasks:
try:
tasks[task] = task.send(tasks[task])
except StopIteration as e:
pending_tasks.remove(task)
tasks[task] = e.args[0] if len(e.args) > 0 else None
print('Data loaded')
if __name__ == '__main__':
main()
```
```shell=
Starting Tue Jul 23 10:41:46 2024
Loading data
Do other tasks
b.txt loaded
a.txt loaded
Data loaded
Ending Tue Jul 23 10:41:51 2024
Time taken: 5.00 seconds
```
由於我們每次都會檢查 pending task 的狀態, 所以能夠充分利用 I/O 時閒置的時間去執行其他任務
可以看到執行時間又比 thread-based 快了一些
這就是 Coroutine!
## Coroutines with Event Loop
我們用 `event_loop` annotation 將 `main()` function 中的 `while` loop 封裝起來
並新增 `regular_task` 和 `async_task` annotations 將 wrapped function 加入 event loop 中
`event_loop` 會依序執行 regular 和 async functions
regular functions 執行一次即結束
async functions 則在每一次 loop 時都會從之前 pending (yield) 的地方繼續執行 (resume)
TODO: 圖
`lib/async_util.py`
```python=
import time
regular_tasks = []
async_tasks = []
def async_task(func):
global async_tasks
def wrapper(*args, **kwargs):
gen = func(*args, **kwargs)
async_tasks.append(gen)
return wrapper
def regular_task(func):
global regular_tasks
def wrapper(*args, **kwargs):
regular_tasks.append((func, args, kwargs))
return wrapper
def event_loop(target_func):
global regular_tasks, async_tasks
def wrapper(*args, **kwargs):
target_func(*args, **kwargs)
tasks = { task: None for task in async_tasks }
while async_tasks or regular_tasks:
for regular_task in regular_tasks:
func, args, kwargs = regular_task
try:
func(*args, **kwargs)
finally:
regular_tasks.remove(regular_task)
for async_task in async_tasks:
try:
tasks[async_task] = async_task.send(tasks[async_task])
except StopIteration as e:
tasks[async_task] = e.args[0] if len(e.args) > 0 else None
async_tasks.remove(async_task)
return wrapper
def sleep(secs):
now = time.time()
while time.time() < now + secs:
yield
pass
def load_data(file, secs=5):
yield from sleep(secs)
```
`async_util.py`
```python=
#!/usr/bin/env python3
from lib import (async_util, timeit)
@async_util.async_task
def load_data(file):
print(file)
yield from async.load_data(file)
print(f"{file} loaded")
@async_util.regular_task
def do_other_tasks():
print('Do other tasks')
@timeit
@async_util.event_loop
def main():
print('Loading data')
load_data('a.txt')
load_data('b.txt')
do_other_tasks()
print('Data loaded')
if __name__ == '__main__':
main()
```
我們將想要給 event loop 執行的 task 用 decorator 表示
最終會得到下面結果
```shell=
Starting Mon Jul 22 18:02:30 2024
Loading data
a.txt
b.txt
Do other tasks
a.txt loaded
b.txt loaded
Data loaded
Ending Mon Jul 22 18:02:35 2024
Time taken: 5.00 seconds
```
由於我們將 I/O 任務都交給 event loop 管理, 且 event loop 每次都會繼續 (resume) I/O 任務以確認其狀態
所以就不會再擋住其他任務的執行
這就是 asyncio 的雛型!
## Asyncio
先介紹一下 `asyncio` library
`asyncio` 提供了很多方便的 API 來達成 coroutines, 這邊解釋一些 `asyncio` 的物件
Awaitables
- 可以使用 `await` 表達式的物件
- 包含
- `Coroutines`
- `Tasks`
- `Futures`
Coroutines
- 用 `async/await` 表示, 取代前面的 generator functions
- 可以是 *coroutine function* 或是 *coroutine object*
- coroutine function
- 用 `async def` 宣告的 function
- coroutine object
- coroutine function 回傳的物件
Tasks
- 當 `coroutine` 物件被 `asyncio.create_task()` 加入 Task, 就代表已經加入排程
Futures
- low-level object
- 代表 awaitable object 最終的結果
- `asyncio` 用來支援 coroutine callback 的實作
- 少數 API 會回傳 Futures 物件, ex. `loop.run_in_executor()`
- 通常用在 multithreading 或 multiprocessing
- 詳見 [Executing code in thread or process pools](https://docs.python.org/3/library/asyncio-eventloop.html#executing-code-in-thread-or-process-pools)
:::spoiler
import asyncio
import concurrent.futures
def blocking_io():
# File operations (such as logging) can block the
# event loop: run them in a thread pool.
with open('/dev/urandom', 'rb') as f:
return f.read(100)
def cpu_bound():
# CPU-bound operations will block the event loop:
# in general it is preferable to run them in a
# process pool.
return sum(i * i for i in range(10 ** 7))
async def main():
loop = asyncio.get_running_loop()
## Options:
# 1. Run in the default loop's executor:
result = await loop.run_in_executor(
None, blocking_io)
print('default thread pool', result)
# 2. Run in a custom thread pool:
with concurrent.futures.ThreadPoolExecutor() as pool:
result = await loop.run_in_executor(
pool, blocking_io)
print('custom thread pool', result)
# 3. Run in a custom process pool:
with concurrent.futures.ProcessPoolExecutor() as pool:
result = await loop.run_in_executor(
pool, cpu_bound)
print('custom process pool', result)
if __name__ == '__main__':
asyncio.run(main())
:::
更多用法請見 [asyncio 官方文件](https://docs.python.org/3/library/asyncio-task.html)
前面我們用 generator 搭配 event loop 實做了簡單的 coroutine 架構
但是實作很複雜, 且不方便管理和維護
下面我們用 Python 提供的 `asyncio` library 和 `async def` keyword 來改寫前面的範例
`lib/async_util.py`
```python=
import time
def timeit(func):
async def wrapper(*args, **kwargs):
start_time = time.time()
print(f"Starting {time.ctime(start_time)}")
result = await func(*args, **kwargs)
end_time = time.time()
print(f"Ending {time.ctime(end_time)}")
print(f"Time taken: {end_time - start_time:.2f} seconds")
return result
return wrapper
async def sleep(secs):
await asyncio.sleep(secs)
async def load_data(file, secs=5):
await sleep(secs)
```
`async.py`
```python=
import asyncio
import time
async def do_other_tasks():
print('Do other tasks')
async def _load_data(file):
await load_data(file)
print(f"{file} loaded")
@timeit
async def main():
print('Loading data')
task1 = asyncio.create_task(_load_data('data.txt'))
task2 = asyncio.create_task(_load_data('data2.txt'))
task3 = asyncio.create_task(do_other_tasks())
await asyncio.gather(task1, task2, task3)
print('Data loaded')
if __name__ == '__main__':
asyncio.run(main())
```
`asyncio.run()` 會產生 event loop, 類似我們前面用的 `@event_loop` annotation
`asyncio.create_task()` 會將 `coroutine` 物件封裝成 `Task` 物件加入排程, 類似前面的 `@regular_task` 和 `@async_task` annotations
Refactor 完後看起來簡潔多了
```shell=
Starting Tue Jul 23 12:05:31 2024
Loading data
Do other tasks
data.txt loaded
data2.txt loaded
Data loaded
Ending Tue Jul 23 12:05:36 2024
Time taken: 5.02 seconds
```
執行時間也比 multiprocessing 和 multithreading 快
# Summary
我們介紹並比較了 Synchronous 和 Asynchronous 的差別
發現 coroutine 執行時間確實比 multiprocessing 和 threading 要快
在 Synchronous 的介紹中, 說明了如何透過平行化 (Parallelism) 改善 IO-bound functions 效能
並介紹兩種平行化的方式: multiprocessing, multithreading, 其中 multithreading 在 Python 因為 GIL (Global Interpreter Lock) 的關係, 其背後實作其實是 coroutine
在 Asynchronous 中, 我們分別用 processing, threading, generator 實作了 asynchronous function
並從 generator 開始, 實作出簡單版本的 coroutine 架構
最後介紹了 `asyncio` library 的 `Awaitables` 等物件, 並用其改寫前面自己刻的 coroutine 架構
和 JavaScript 原生就有 event loop 不同, Python 需要透過 `asyncio` 提供的 API 產生 event loop
但是搭配 `async def`, `async for`, `async with` 等 Coroutine syntax
Python 也能享受到 coroutine 帶來的好處
搭配 `multiprocessing` 和 `threading`, 又有更多的彈性
之後如果程式有 IO-bound / CPU-bound 的需求, 不妨試試 coroutine 吧!
# Reference
- [PEP 255 - Simple Generators](https://peps.python.org/pep-0255/)
- [PEP 342 - Coroutines via Enhanced Generators](https://peps.python.org/pep-0342/)
- [Coroutines and Tasks](https://docs.python.org/3/library/asyncio-task.html)
- [Compound statements](https://docs.python.org/3/reference/compound_stmts.html)
- [multiprocessing — Process-based parallelism](https://docs.python.org/3/library/multiprocessing.html)
- [深入 GIL: 如何寫出快速且 thread-safe 的 Python – Grok the GIL: How to write fast and thread-safe Python](https://blog.louie.lu/2017/05/19/%E6%B7%B1%E5%85%A5-gil-%E5%A6%82%E4%BD%95%E5%AF%AB%E5%87%BA%E5%BF%AB%E9%80%9F%E4%B8%94-thread-safe-%E7%9A%84-python-grok-the-gil-how-to-write-fast-and-thread-safe-python/)
- [[Python]Concurreny效能實測 — MultiThreading/MultiProcessing/Coroutine](https://wuyiru.medium.com/python-concurreny%E6%95%88%E8%83%BD%E5%AF%A6%E6%B8%AC-multithreading-multiprocessing-coroutine-6043793e0328)
Sign in with Wallet
Connect another wallet