One programming model to rule them all
\Rightarrow it is an abstraction over concurrency, it does not necessarily mean that the program is executed in parallel.
import requests
1response = requests.get('https://www.example.com')
items = response.headers.items()
2headers = [f'{key}: {header}' for key, header in items]
3formatted_headers = '\n'.join(headers)
with open('headers.txt', 'w') as file:
4 file.write(formatted_headers)We will use extensively the bakery metaphor.
One baker and two cakes to prepare.
\Rightarrow Switching between tasks is concurrency (or concurrent behavior).
Two bakers and two cakes to prepare.
\Rightarrow Doing multiple tasks in parallel is parallelism (or parallel behavior).
From [1]
From [1]
While parallelism implies concurrency, concurrency does not always imply parallelism.
\Rightarrow Concurrency is a broader concept than parallelism.
Main Process
βββββββββββββββ
β β
β CPU β
βββββββββββββββ€
β β
β Memory β
βββ¬βββββ¬βββββ¬ββ
β β β
β β β
βββββββββββββ β βββββββββββββ
β β β
ββββββββΌβββββββ ββββββββΌβββββββ ββββββββΌβββββββ
β β β β β β
β CPU β β CPU β β CPU β
βββββββββββββββ€ βββββββββββββββ€ βββββββββββββββ€
β β β β β β
β Memory β β Memory β β Memory β
βββββββββββββββ βββββββββββββββ βββββββββββββββ
Process 1 Process 1 Process 1ββββββββββββββββββββββββββββββββββββββββ
β MAIN PROCESS β
β β
β β
β ββββββββββββ β
β β β β
β β CPU β βββββββββββββ β
β β β β β β
β ββββββββββββ β Memory β β
β β β β
β βββββββββββββ β
β β
β β
β β
β βββ βββ βββ β
β ββΌβ ββΌβ ββΌβ β
β ββ΄β ββ΄β ββ΄β β
β βΌβΌβΌ βΌβΌβΌ βΌβΌβΌ β
β Thread 1 Thread 2 Thread 3 β
β βββ βββ βββ β
β ββΌβ ββΌβ ββΌβ β
β ββ΄β ββ΄β ββ΄β β
β βΌβΌβΌ βΌβΌβΌ βΌβΌβΌ β
β β
ββββββββββββββββββββββββββββββββββββββββimport os
import threading
print(f'Python process running with process id: {os.getpid()}')
total_threads = threading.active_count()
thread_name = threading.current_thread().name
print(f'Python is currently running {total_threads} thread(s)')
print(f'The current thread is {thread_name}')Python process running with process id: 132369
Python is currently running 8 thread(s)
The current thread is MainThreadimport multiprocessing
import os
def hello_from_process():
print(f'Hello from child process {os.getpid()}!')
if __name__ == '__main__':
hello_process = multiprocessing.Process(target=hello_from_process)
hello_process.start()
print(f'Hello from parent process {os.getpid()}')
hello_process.join()Hello from child process 132419!
Hello from parent process 132369import threading
def hello_from_thread():
print(f'Hello from thread {threading.current_thread()}!')
hello_thread = threading.Thread(target=hello_from_thread)
hello_thread.start()
total_threads = threading.active_count()
thread_name = threading.current_thread().name
print(f'Python is currently running {total_threads} thread(s)')
print(f'The current thread is {thread_name}')
hello_thread.join()Hello from thread <Thread(Thread-6 (hello_from_thread), started 140617705440832)>!
Python is currently running 8 thread(s)
The current thread is MainThreadAka Global Interpreter Lock
. . .
. . .
. . .
Multi-threading doesnβt guarantee parallel executienβ¦
\Longrightarrow Python seems to have started off with the wrong foot by a long wayβ¦
But wait!
Actually we can run (real) parallel programs with the multiprocessing package.
\Rightarrow But this is an βOS levelβ multiprocessing, with associated huge overhead (relatively)
Python actually releases the GIL when executing everything that is not Python code (e.g. C/C++ extensions and libraries)
\Rightarrow It means we can parallelize our code by using I/O bound and CPU bound libraries that release the GIL (this is the case for most of them)
asyncio
From [1]
From [1]
from collections import deque
messages = deque()
while True:
if messages:
message = messages.pop()
process_message(message)\Rightarrow In asyncio, the event loop is queue of tasks instead of messages, Tasks are wrapped coroutines.
def make_request():
cpu_bound_setup()
io_bound_web_request()
cpu_bound_postprocess()
task_one = make_request()
task_two = make_request()
task_three = make_request()
asyncio CoroutinesTo define a coroutine, we use the async def syntax.
async def my_coroutine() -> None
print('Hello world!')async def coroutine_add_one(number: int) -> int:
return number + 1
def add_one(number: int) -> int:
return number + 1
1function_result = add_one(1)
2coroutine_result = coroutine_add_one(1)
print(f'Function result is {function_result}\n\
and the type is {type(function_result)}')
print(f'Coroutine result is {coroutine_result}\n\
and the type is {type(coroutine_result)}')Function result is 2
and the type is <class 'int'>
Coroutine result is <coroutine object coroutine_add_one at 0x7fe3fdddb640>
and the type is <class 'coroutine'>From [1]
You need an event loop.
import asyncio
async def coroutine_add_one(number: int) -> int:
return number + 1
1result = asyncio.run(coroutine_add_one(1))
print(result)This code will not work in a Jupyter notebook, because the event loop is already running (by Jupyter itself). So you just have to replace the line 4 by:
result = await coroutine_add_one(1)await keywordimport asyncio
async def add_one(number: int) -> int:
return number + 1
async def main() -> None:
1 one_plus_one = await add_one(1)
2 two_plus_one = await add_one(2)
print(one_plus_one)
print(two_plus_one)
3await main()add_one(1).
add_one(2).
main(). (outside of a Jupyter notebook, you have to launch the event loop somewhere, like asyncio.run(main()) instead of await main())
2
3
From [1]
asyncio.sleepimport asyncio
async def hello_world_message() -> str:
1 await asyncio.sleep(1)
return 'Hello World!'
async def main() -> None:
2 message = await hello_world_message()
print(message)
await main()hello_world_message for 1 second.
main until hello_world_message is finished.
Hello World!delay(seconds)import asyncio
1async def delay(delay_seconds: int) -> int:
2 print(f'sleeping for {delay_seconds} second(s)')
await asyncio.sleep(delay_seconds)
print(f'finished sleeping for {delay_seconds} second(s)')
3 return delay_secondsimport asyncio
async def add_one(number: int) -> int:
return number + 1
async def hello_world_message() -> str:
await delay(1)
return 'Hello World!'
async def main() -> None:
1 message = await hello_world_message()
2 one_plus_one = await add_one(1)
print(one_plus_one)
print(message)
await main()main until hello_world_message is finished.
main until add_one is finished.
sleeping for 1 second(s)
finished sleeping for 1 second(s)
2
Hello World!
From [1]
Moving away from sequential execution and run add_one and hello_world_message concurrently.
For that we needβ¦
So far we just learned how to create coroutines and put then in the event loop.
Tasks are a way to schedule coroutines concurrently.
\Rightarrow Tasks are wrapped coroutines which are scheduled to run in the event loop as soon as possible.
import asyncio
async def main():
sleep_for_three = asyncio.create_task(delay(3))
print(type(sleep_for_three))
result = await sleep_for_three
print(result)
await main()<class '_asyncio.Task'>
sleeping for 3 second(s)
finished sleeping for 3 second(s)
3import asyncio
async def main():
sleep_for_three = \
asyncio.create_task(delay(3))
sleep_again = \
asyncio.create_task(delay(3))
sleep_once_more = \
asyncio.create_task(delay(3))
await sleep_for_three
await sleep_again
await sleep_once_more
await main()sleeping for 3 second(s)
sleeping for 3 second(s)
sleeping for 3 second(s)
finished sleeping for 3 second(s)
finished sleeping for 3 second(s)
finished sleeping for 3 second(s)
From [1]
import asyncio
async def hello_every_second():
for i in range(2):
await asyncio.sleep(1)
print("I'm running other code while I'm waiting!")
async def main():
first_delay = asyncio.create_task(delay(3))
second_delay = asyncio.create_task(delay(3))
await hello_every_second()
await first_delay
await second_delay
await main()sleeping for 3 second(s)
sleeping for 3 second(s)
I'm running other code while I'm waiting!
I'm running other code while I'm waiting!
finished sleeping for 3 second(s)
finished sleeping for 3 second(s)
From [1]
import asyncio
from asyncio import CancelledError
async def main():
long_task = asyncio.create_task(delay(10))
seconds_elapsed = 0
while not long_task.done():
print('Task not finished, checking again in a second.')
await asyncio.sleep(1)
seconds_elapsed = seconds_elapsed + 1
if seconds_elapsed == 5:
long_task.cancel()
try:
await long_task
except CancelledError:
print('Our task was cancelled')
await main()Task not finished, checking again in a second.
sleeping for 10 second(s)
Task not finished, checking again in a second.
Task not finished, checking again in a second.
Task not finished, checking again in a second.
Task not finished, checking again in a second.
Task not finished, checking again in a second.
Our task was cancelledimport asyncio
async def main():
delay_task = asyncio.create_task(delay(2))
try:
result = await asyncio.wait_for(delay_task, timeout=1)
print(result)
except asyncio.exceptions.TimeoutError:
print('Got a timeout!')
print(f'Was the task cancelled? {delay_task.cancelled()}')
await main()sleeping for 2 second(s)
Got a timeout!
Was the task cancelled? Truefrom asyncio import Future
my_future = Future()
print(f'Is my_future done? {my_future.done()}')
my_future.set_result(42)
print(f'Is my_future done? {my_future.done()}')
print(f'What is the result of my_future? {my_future.result()}')Is my_future done? False
Is my_future done? True
What is the result of my_future? 42from asyncio import Future
import asyncio
def make_request() -> Future:
future = Future()
1 asyncio.create_task(set_future_value(future))
return future
async def set_future_value(future) -> None:
2 await asyncio.sleep(1)
future.set_result(42)
async def main():
future = make_request()
print(f'Is the future done? {future.done()}')
3 value = await future
print(f'Is the future done? {future.done()}')
print(value)
await main()Is the future done? False
Is the future done? True
42
async def, and can be awaited to allow other coroutines to run.
import functools
import time
from typing import Callable, Any
def async_timed():
def wrapper(func: Callable) -> Callable:
@functools.wraps(func)
async def wrapped(*args, **kwargs) -> Any:
print(f'starting {func} with args {args} {kwargs}')
start = time.time()
try:
return await func(*args, **kwargs)
finally:
end = time.time()
total = end - start
print(f'finished {func} in {total:.4f} second(s)')
return wrapped
return wrapperOfficial Python documentation for decorators
import asyncio
@async_timed()
async def delay(delay_seconds: int) -> int:
print(f'sleeping for {delay_seconds} second(s)')
await asyncio.sleep(delay_seconds)
print(f'finished sleeping for {delay_seconds} second(s)')
return delay_seconds
@async_timed()
async def main():
task_one = asyncio.create_task(delay(2))
task_two = asyncio.create_task(delay(3))
await task_one
await task_two
await main()starting <function main at 0x7fe3fdc84d60> with args () {}
starting <function delay at 0x7fe3fdc86c00> with args (2,) {}
sleeping for 2 second(s)
starting <function delay at 0x7fe3fdc86c00> with args (3,) {}
sleeping for 3 second(s)
finished sleeping for 2 second(s)
finished <function delay at 0x7fe3fdc86c00> in 2.0032 second(s)
finished sleeping for 3 second(s)
finished <function delay at 0x7fe3fdc86c00> in 3.0017 second(s)
finished <function main at 0x7fe3fdc84d60> in 3.0018 second(s)asyncio.gatherasyncio.gather() runs multiple asynchronous operations, wraps a coroutine as a task, and returns a list of results in the same order of awaitables.
import asyncio
async def call_api(message, result, delay=3):
print(message)
await asyncio.sleep(delay)
return result
async def main():
return await asyncio.gather(
call_api('Calling API 1 ...', 1),
call_api('Calling API 2 ...', 2)
)
await main()Calling API 1 ...
Calling API 2 ...[1, 2]asyncio.gather takes a tuple of awaitables, not a list of awaitables, but returns a list of results in the same order of awaitables.
If you want to pass a list, use the * operator to unpack it as a tuple.
import asyncio
@async_timed()
async def cpu_bound_work() -> int:
counter = 0
for i in range(100000000):
counter = counter + 1
return counter
@async_timed()
async def main():
task_one = asyncio.create_task(cpu_bound_work())
task_two = asyncio.create_task(cpu_bound_work())
await task_one
await task_two
await main()starting <function main at 0x7fe3fdc87c40> with args () {}
starting <function cpu_bound_work at 0x7fe3fdc87a60> with args () {}
finished <function cpu_bound_work at 0x7fe3fdc87a60> in 3.7527 second(s)
starting <function cpu_bound_work at 0x7fe3fdc87a60> with args () {}
finished <function cpu_bound_work at 0x7fe3fdc87a60> in 4.1736 second(s)
finished <function main at 0x7fe3fdc87c40> in 7.9269 second(s)import asyncio
import requests
@async_timed()
async def get_example_status() -> int:
return requests.get('http://www.example.com').status_code
@async_timed()
async def main():
task_1 = asyncio.create_task(get_example_status())
task_2 = asyncio.create_task(get_example_status())
task_3 = asyncio.create_task(get_example_status())
await task_1
await task_2
await task_3
await main()starting <function main at 0x7fe4559e0ae0> with args () {}
starting <function get_example_status at 0x7fe3fdc86e80> with args () {}
finished <function get_example_status at 0x7fe3fdc86e80> in 0.3771 second(s)
starting <function get_example_status at 0x7fe3fdc86e80> with args () {}
finished <function get_example_status at 0x7fe3fdc86e80> in 0.4965 second(s)
starting <function get_example_status at 0x7fe3fdc86e80> with args () {}
finished <function get_example_status at 0x7fe3fdc86e80> in 0.4975 second(s)
finished <function main at 0x7fe4559e0ae0> in 1.3716 second(s)import requests
def get_status_code(url: str) -> int:
response = requests.get(url)
return response.status_code
url = 'https://www.example.com'
print(get_status_code(url))
print(get_status_code(url))200
200import time
import requests
from concurrent.futures import ThreadPoolExecutor
def get_status_code(url: str) -> int:
response = requests.get(url)
return response.status_code
start = time.time()
with ThreadPoolExecutor() as pool:
urls = ['https://www.example.com' for _ in range(10)]
results = pool.map(get_status_code, urls)
for result in results:
# print(result)
pass
end = time.time()
print(f'finished requests in {end - start:.4f} second(s)')finished requests in 0.7330 second(s)start = time.time()
urls = ['https://www.example.com' for _ in range(10)]
for url in urls:
result = get_status_code(url)
# print(result)
end = time.time()
print(f'finished requests in {end - start:.4f} second(s)')finished requests in 7.8103 second(s)asyncioimport functools
import requests
import asyncio
from concurrent.futures import ThreadPoolExecutor
def get_status_code(url: str) -> int:
response = requests.get(url)
return response.status_code
@async_timed()
async def main():
loop = asyncio.get_running_loop()
with ThreadPoolExecutor() as pool:
urls = ['https://www.example.com' for _ in range(10)]
tasks = [loop.run_in_executor(pool, functools.partial(get_status_code, url)) for url in urls]
results = await asyncio.gather(*tasks)
print(results)
await main()starting <function main at 0x7fe455a29580> with args () {}
[200, 200, 200, 200, 200, 200, 200, 200, 200, 200]
finished <function main at 0x7fe455a29580> in 0.8459 second(s)Letβs define a big matrix on which we will compute the mean of each row.
import numpy as np
data_points = 400000000
rows = 50
columns = int(data_points / rows)
matrix = np.arange(data_points).reshape(rows, columns)Now process the matrix sequentially.
s = time.time()
res_seq = np.mean(matrix, axis=1)
e = time.time()
print(e - s)1.312664270401001And then the same with multithreading (we check that the results are exactly the same).
import functools
from concurrent.futures.thread import ThreadPoolExecutor
import asyncio
def mean_for_row(arr, row):
return np.mean(arr[row])
@async_timed()
async def main():
loop = asyncio.get_running_loop()
with ThreadPoolExecutor() as pool:
tasks = []
for i in range(rows):
mean = functools.partial(mean_for_row, matrix, i)
tasks.append(loop.run_in_executor(pool, mean))
return await asyncio.gather(*tasks)
res_threads = np.array(await main())
np.testing.assert_array_equal(res_seq, res_threads)starting <function main at 0x7fe455a2b060> with args () {}
finished <function main at 0x7fe455a2b060> in 0.2386 second(s)Everything is awaitable (coroutines, futures, tasks), i.e. can be simply run with await.
a task is a coroutine wrapped in a future, and scheduled to run in the event loop.
asyncio is a single-threaded asynchronous programming library, providing a simple way to write concurrent code for I/O bound tasks.
\Rightarrow Weβll see later that this programming model can be used for parallelism as well, and very easily.