General concepts
Concurrency vs Parallelism
Concurrency is when two or more tasks can start, run, and complete in overlapping time periods. It doesn’t necessarily mean they’ll ever both be running at the same instant. For example, multitasking on a single-core machine.
Parallelism is when tasks literally run at the same time, e.g., on a multicore processor.
Say you have a program that has two threads. The program can run in two ways:
1
2
3
4
5
6
7
8
9
10
11
|
Concurrency Concurrency + parallelism
(Single-Core CPU) (Multi-Core CPU)
___ ___ ___
|th1| |th1|th2|
| | | |___|
|___|___ | |___
|th2| |___|th2|
___|___| ___|___|
|th1| |th1|
|___|___ | |___
|th2| | |th2|
|
In both cases we have concurrency from the mere fact that we have more than one thread running.
If we ran this program on a computer with a single CPU core, the OS would be switching between the two threads, allowing one thread to run at a time.
If we ran this program on a computer with a multi-core CPU then we would be able to run the two threads in parallel - side by side at the exact same time.
Process vs Thread vs Coroutine
Process
A process “owns” all its assigned resources, like memory, file handles, sockets, device handles, etc., and these resources are all shared among its kernel threads.
Thread
Every process has at least one kernel thread. Usually, multiple threads are created within a process. Threads execute within a process and processes execute within the operating system kernel. With threads, the operating system switches running threads preemptively according to its scheduler, which is an algorithm in the operating system kernel.
Coroutine
With coroutines, the programmer and programming language determine when to switch coroutines; in other words, tasks are cooperatively multitasked by pausing and resuming functions at set points, typically (but not necessarily) within a single thread.
Coroutines can provide a very high level of concurrency with very little overhead. Remember a co-routine can still do concurrency without scheduler overhead - it simply manages the context-switching itself.
Another benefit is the much lower memory usage. With the threaded-model, each thread needs to allocate its own stack, and so your memory usage grows linearly with the number of threads you have. With co-routines, the number of routines you have doesn’t have a direct relationship with your memory usage.
Python Global Interpreter Lock
In CPython, the global interpreter lock, or GIL, is a mutex that protects access to Python objects, preventing multiple threads from executing Python bytecodes at once. The GIL prevents race conditions and ensures thread safety. In short, this mutex is necessary mainly because CPython’s memory management is not thread-safe.
When you look at a typical Python program—or any computer program for that matter—there’s a difference between those that are CPU-bound in their performance and those that are I/O-bound. For CPU-bound work you need to sidestep the GIL by launching multiple processes. If you want your application to make better use of the computational resources of multi-core machines, you are advised to use multiprocessing or concurrent.futures.ProcessPoolExecutor. However, threading is still an appropriate model if you want to run multiple I/O-bound tasks simultaneously.
Because of Python’s Global Interpreter Lock (GIL), the threads within each python process cannot truly run in parallel, it can run in concurrent.
The multiprocessing module allows you to create multiple processes, each of them with its own Python interpreter. For this reason, Python multiprocessing accomplishes process-based parallelism.
Threads Python cannot run in parallel anyway, so coroutine running in a single thread cannot run in parallel, they run concurrently. coroutines can provide a very high level of concurrency with very little overhead.
There are two common strategies for working around the limitations of the GIL. First, if you are working entirely in Python, you can use the multiprocessing module to create a process pool and use it like a co-processor. The second strategy for working around the GIL is to focus on C extension programming. The general idea is to move computationally intensive tasks to C, independent of Python, and have the C code release the GIL while it’s working. If you are using other tools to access C, such as the ctypes library or Cython, you may not need to do anything. For example, ctypes releases the GIL when calling into C by default.
1
2
3
4
5
6
7
8
9
10
|
#include "Python.h"
...
PyObject *pyfunc(PyObject *self, PyObject *args) {
...
Py_BEGIN_ALLOW_THREADS
// Threaded C code
...
Py_END_ALLOW_THREADS
...
}
|
Synchronous vs Asynchronous
Concurrency with threading
Starting and Stopping Threads
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
from threading import Thread
import time
class CountdownTask:
def __init__(self):
self._running = True
def terminate(self):
self._running = False
def run(self, n):
while self._running and n > 0:
print("T-minus", n)
n -= 1
time.sleep(5)
c = CountdownTask()
t = Thread(target=c.run, args=(10,))
t.start()
if t.is_alive():
print('Still running')
time.sleep(20)
# Signal termination
c.terminate()
# Wait for actual termination (if needed)
t.join()
if not t.is_alive():
print('Terminated')
|
Determining If a Thread Has Started
Using an Event to coordinate the startup of a thread:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
from threading import Thread, Event
import time
# Code to execute in an independent thread
def countdown(n, started_evt):
print("countdown starting")
started_evt.set()
while n > 0:
print("T-minus", n)
n -= 1
time.sleep(5)
# Create the event object that will be used to signal startup
started_evt = Event()
# Launch the thread and pass the startup event
print("Launching countdown")
t = Thread(target=countdown, args=(10,started_evt))
t.start()
# Wait for the thread to start
started_evt.wait()
print("countdown is running")
|
Using a Condition object implements a periodic timer that other threads can monitor to see whenever the timer expires:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
|
import threading
import time
class PeriodicTimer:
def __init__(self, interval):
self._interval = interval
self._flag = 0
self._cv = threading.Condition()
def start(self):
t = threading.Thread(target=self.run)
t.daemon = True
t.start()
def run(self):
'''
Run the timer and notify waiting threads after each interval
'''
while True:
time.sleep(self._interval)
with self._cv:
self._flag ^= 1
self._cv.notify_all()
def wait_for_tick(self):
'''
Wait for the next tick of the timer
'''
with self._cv:
last_flag = self._flag
while last_flag == self._flag:
self._cv.wait()
# Example use of the timer
ptimer = PeriodicTimer(5)
ptimer.start()
# Two threads that synchronize on the timer
def countdown(nticks):
while nticks > 0:
ptimer.wait_for_tick()
print("T-minus", nticks)
nticks -= 1
def countup(last):
n = 0
while n < last:
ptimer.wait_for_tick()
print("Counting", n)
n += 1
threading.Thread(target=countdown, args=(10,)).start()
threading.Thread(target=countup, args=(5,)).start()
|
A critical feature of Event objects is that they wake all waiting threads. If you are writing a program where you only want to wake up a single waiting thread, it is probably better to use a Semaphore or Condition object instead.
Each time the semaphore is released, only one worker will wake up and run.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
import threading
import time
# Worker thread
def worker(n, sema):
# Wait to be signalled
sema.acquire()
# Do some work
print("Working", n)
# Create some threads
sema = threading.Semaphore(0)
nworkers = 10
for n in range(nworkers):
t = threading.Thread(target=worker, args=(n, sema,))
t.daemon=True
t.start()
|
1
2
3
4
|
>>> sema.release()
Working 0
>>> sema.release()
Working 1
|
Communicating Between Threads
Perhaps the safest way to send data from one thread to another is to use a Queue from the queue library. To do this, you create a Queue instance that is shared by the threads. Threads then use put() or get() operations to add or remove items from the queue.
Queue instances already have all of the required locking, so they can be safely shared by as many threads as you wish.
A common solution to this problem is to rely on a special sentinel value, which when placed in the queue, causes consumers to terminate.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
from queue import Queue
from threading import Thread
import time
_sentinel = object()
# A thread that produces data
def producer(out_q):
n = 10
while n > 0:
# Produce some data
out_q.put(n)
time.sleep(2)
n -= 1
# Put the sentinel on the queue to indicate completion
out_q.put(_sentinel)
# A thread that consumes data
def consumer(in_q):
while True:
# Get some data
data = in_q.get()
# Check for termination
if data is _sentinel:
in_q.put(_sentinel)
break
# Process the data
print('Got:', data)
print('Consumer shutting down')
if __name__ == '__main__':
q = Queue()
t1 = Thread(target=consumer, args=(q,))
t2 = Thread(target=producer, args=(q,))
t1.start()
t2.start()
t1.join()
t2.join()
|
One caution with thread queues is that putting an item in a queue doesn’t make a copy
of the item. Thus, communication actually involves passing an object reference between threads. If you are concerned about shared state, it may make sense to only pass immutable data structures (e.g., integers, strings, or tuples) or to make deep copies of the queued items.
Thread communication with a queue is a one-way and nondeterministic process. Queue objects do provide some basic completion features, as illustrated by the task_done() and join() methods:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
from queue import Queue
from threading import Thread
# A thread that produces data
def producer(out_q):
while running:
# Produce some data
...
out_q.put(data)
# A thread that consumes data
def consumer(in_q):
while True:
# Get some data
data = in_q.get()
# Process the data
...
# Indicate completion
in_q.task_done()
# Create the shared queue and launch both threads
q = Queue()
t1 = Thread(target=consumer, args=(q,))
t2 = Thread(target=producer, args=(q,))
t1.start()
t2.start()
# Wait for all produced items to be consumed
q.join()
|
If a thread needs to know immediately when a consumer thread has processed a particular item of data, you should pair the sent data with an Event object that allows the producer to monitor its progress.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
from queue import Queue
from threading import Thread, Event
# A thread that produces data
def producer(out_q):
while running:
# Produce some data
...
# Make an (data, event) pair and hand it to the consumer
evt = Event()
out_q.put((data, evt))
...
# Wait for the consumer to process the item
evt.wait()
# A thread that consumes data
def consumer(in_q):
while True:
# Get some data
data, evt = in_q.get()
# Process the data
...
# Indicate completion
evt.set()
|
If you create a Queue with an optional size, such as Queue(N), it places a limit
on the number of items that can be enqueued before the put() blocks the producer.
Adding an upper bound to a queue might make sense if there is mismatch in speed
between a producer and consumer. For instance, if a producer is generating items at a
much faster rate than they can be consumed. On the other hand, making a queue block when it’s full can also have an unintended cascading effect throughout your program, possibly causing it to deadlock or run poorly. In general, the problem of “flow control” between communicating threads is a much harder problem than it seems. If you ever find yourself trying to fix a problem by fiddling with queue sizes, it could be an indicator of a fragile design or some other inherent scaling problem.
Both the get() and put() methods support nonblocking and timeouts.
Both of these options can be used to avoid the problem of just blocking indefinitely on a particular queuing operation.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
import queue
q = queue.Queue()
try:
data = q.get(block=False)
except queue.Empty:
...
try:
q.put(item, block=False)
except queue.Full:
...
try:
data = q.get(timeout=5.0)
except queue.Empty:
...
|
build a thread-safe priority queue:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
import heapq
import threading
import time
class PriorityQueue:
def __init__(self):
self._queue = []
self._count = 0
self._cv = threading.Condition()
def put(self, item, priority):
with self._cv:
heapq.heappush(self._queue, (-priority, self._count, item))
self._count += 1
self._cv.notify()
def get(self):
with self._cv:
while len(self._queue) == 0:
self._cv.wait()
return heapq.heappop(self._queue)[-1]
def producer(q):
print('Producing items')
q.put('C', 5)
q.put('A', 15)
q.put('B', 10)
q.put('D', 0)
q.put(None, -100)
def consumer(q):
time.sleep(5)
print('Getting items')
while True:
item = q.get()
if item is None:
break
print('Got:', item)
print('Consumer done')
if __name__ == '__main__':
q = PriorityQueue()
t1 = threading.Thread(target=producer, args=(q,))
t2 = threading.Thread(target=consumer, args=(q,))
t1.start()
t2.start()
t1.join()
t2.join()
|
Locking Critical Sections
lock critical sections of code to avoid race conditions.
A Lock guarantees mutual exclusion when used with the with statement—that is, only one thread is allowed to execute the block of statements under the with statement at a time. The with statement acquires the lock for the duration of the indented statements and releases the lock when control flow exits the indented block.
Thread scheduling is inherently nondeterministic. Locks should always be used whenever shared mutable state is accessed by multiple threads.
To avoid the potential for deadlock, programs that use locks should be written in a way
such that each thread is only allowed to acquire one lock at a time.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
import threading
class SharedCounter:
'''
A counter object that can be shared by multiple threads.
'''
def __init__(self, initial_value = 0):
self._value = initial_value
self._value_lock = threading.Lock()
def incr(self,delta=1):
'''
Increment the counter with locking
'''
with self._value_lock:
self._value += delta
def decr(self,delta=1):
'''
Decrement the counter with locking
'''
with self._value_lock:
self._value -= delta
def test(c):
for n in range(1000000):
c.incr()
for n in range(1000000):
c.decr()
if __name__ == '__main__':
c = SharedCounter()
t1 = threading.Thread(target=test, args=(c,))
t2 = threading.Thread(target=test, args=(c,))
t3 = threading.Thread(target=test, args=(c,))
t1.start()
t2.start()
t3.start()
print('Running test')
t1.join()
t2.join()
t3.join()
assert c._value == 0
print('Looks good!')
|
locks explicitly acquired and released:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
import threading
class SharedCounter:
'''
A counter object that can be shared by multiple threads.
'''
def __init__(self, initial_value = 0):
self._value = initial_value
self._value_lock = threading.Lock()
def incr(self,delta=1):
'''
Increment the counter with locking
'''
self._value_lock.acquire()
self._value += delta
self._value_lock.release()
def decr(self,delta=1):
'''
Decrement the counter with locking
'''
self._value_lock.acquire()
self._value -= delta
self._value_lock.release()
|
Locking with Deadlock Avoidance
As a rule of thumb, as long as you can ensure that threads can hold only one lock at a time, your program will be deadlock free.
In multithreaded programs, a common source of deadlock is due to threads that attempt to acquire multiple locks at once. For instance, if a thread acquires the first lock, but then blocks trying to acquire the second lock, that thread can potentially block the progress of other threads and make the program freeze.
One solution to deadlock avoidance is to assign each lock in the program a unique number, and to enforce an ordering rule that only allows multiple locks to be acquired in ascending order. This is surprisingly easy to implement using a context manager as follows:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
import threading
from contextlib import contextmanager
# Thread-local state to stored information on locks already acquired
_local = threading.local()
@contextmanager
def acquire(*locks):
# Sort locks by object identifier
locks = sorted(locks, key=lambda x: id(x))
# Make sure lock order of previously acquired locks is not violated
acquired = getattr(_local, 'acquired',[])
if acquired and max(id(lock) for lock in acquired) >= id(locks[0]):
raise RuntimeError('Lock Order Violation')
# Acquire all of the locks
acquired.extend(locks)
_local.acquired = acquired
try:
for lock in locks:
lock.acquire()
yield
finally:
# Release locks in reverse order of acquisition
for lock in reversed(locks):
lock.release()
del acquired[-len(locks):]
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
import threading
from deadlock import acquire
x_lock = threading.Lock()
y_lock = threading.Lock()
def thread_1():
while True:
with acquire(x_lock, y_lock):
print("Thread-1")
def thread_2():
while True:
with acquire(y_lock, x_lock):
print("Thread-2")
input('This program runs forever. Press [return] to start, Ctrl-C to exit')
t1 = threading.Thread(target=thread_1)
t1.daemon = True
t1.start()
t2 = threading.Thread(target=thread_2)
t2.daemon = True
t2.start()
import time
while True:
time.sleep(1)
|
A common deadlock detection and recovery scheme involves the use of a watchdog timer. Deadlock avoidance is a different strategy where locking operations are carried out in a manner that simply does not allow the program to enter a deadlocked state. Last, but not least, it should be noted that in order to avoid deadlock, all locking operations must be carried out using our acquire() function. If some fragment of code decided to acquire a lock directly, then the deadlock avoidance algorithm wouldn’t work.
Storing Thread-Specific State
Create a thread-local storage object using threading.local(). Attributes stored and read on this object are only visible to the executing thread and no others.
Under the covers, an instance of threading.local() maintains a separate instance dictionary for each thread. All of the usual instance operations of getting, setting, and deleting values just manipulate the per-thread dictionary. The fact that each thread uses a separate dictionary is what provides the isolation of data.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
from socket import socket, AF_INET, SOCK_STREAM
import threading
class LazyConnection:
def __init__(self, address, family=AF_INET, type=SOCK_STREAM):
self.address = address
self.family = AF_INET
self.type = SOCK_STREAM
self.local = threading.local()
def __enter__(self):
if hasattr(self.local, 'sock'):
raise RuntimeError('Already connected')
# each thread actually creates its own dedicated socket connection
self.local.sock = socket(self.family, self.type)
self.local.sock.connect(self.address)
return self.local.sock
def __exit__(self, exc_ty, exc_val, tb):
self.local.sock.close()
del self.local.sock
def test(conn):
from functools import partial
# Connection closed
with conn as s:
# conn.__enter__() executes: connection open
s.send(b'GET /index.html HTTP/1.0\r\n')
s.send(b'Host: www.python.org\r\n')
s.send(b'\r\n')
resp = b''.join(iter(partial(s.recv, 8192), b''))
# conn.__exit__() executes: connection closed
print('Got {} bytes'.format(len(resp)))
if __name__ == '__main__':
conn = LazyConnection(('www.python.org', 80))
t1 = threading.Thread(target=test, args=(conn,))
t2 = threading.Thread(target=test, args=(conn,))
t1.start()
t2.start()
t1.join()
t2.join()
|
Creating a Thread Pool
Generally, you only want to use thread pools for I/O-bound processing.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
from queue import Queue
def echo_client(q):
'''
Handle a client connection
'''
sock, client_addr = q.get()
print('Got connection from', client_addr)
while True:
msg = sock.recv(65536)
if not msg:
break
sock.sendall(msg)
print('Client closed connection')
sock.close()
def echo_server(addr, nworkers):
print('Echo server running at', addr)
# Launch the client workers
q = Queue()
for n in range(nworkers):
t = Thread(target=echo_client, args=(q,))
t.daemon = True
t.start()
# Run the server
sock = socket(AF_INET, SOCK_STREAM)
sock.bind(addr)
sock.listen(5)
while True:
client_sock, client_addr = sock.accept()
q.put((client_sock, client_addr))
echo_server(('',15000), 128)
|
Generally, you should avoid writing programs that allow unlimited growth in the number of threads.
If the size of the virtual memory is a concern, you can dial it down using the threading.stack_size() function.
1
2
|
import threading
threading.stack_size(65536)
|
Actor model
The actor model is a mathematical model of concurrent computation that treats an actor as the basic building block of concurrent computation.
An actor is a computational entity that, in response to a message it receives, can concurrently:
- send a finite number of messages to other actors;
- create a finite number of new actors;
- designate the behavior to be used for the next message it receives.
The actor model is characterized by inherent concurrency of computation within and among actors, dynamic creation of actors, inclusion of actor addresses in messages, and interaction only through direct asynchronous message passing with no restriction on message arrival order.
Thus an actor can only communicate with actors whose addresses it has. It can obtain those from a message it receives, or if the address is for an actor it has itself created.
In a nutshell, an actor is a concurrently executing task that simply acts upon messages sent to it. Communication with actors is one way and asynchronous.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
from queue import Queue
from threading import Thread, Event
# Sentinel used for shutdown
class ActorExit(Exception):
pass
class Actor:
def __init__(self):
self._mailbox = Queue()
def send(self, msg):
'''
Send a message to the actor
'''
self._mailbox.put(msg)
def recv(self):
'''
Receive an incoming message
'''
msg = self._mailbox.get()
if msg is ActorExit:
raise ActorExit()
return msg
def close(self):
'''
Close the actor, thus shutting it down
'''
self.send(ActorExit)
def start(self):
'''
Start concurrent execution
'''
self._terminated = Event()
t = Thread(target=self._bootstrap)
t.daemon = True
t.start()
def _bootstrap(self):
try:
self.run()
except ActorExit:
pass
finally:
self._terminated.set()
def join(self):
self._terminated.wait()
def run(self):
'''
Run method to be implemented by the user
'''
while True:
msg = self.recv()
# Sample ActorTask
class PrintActor(Actor):
def run(self):
while True:
msg = self.recv()
print("Got:", msg)
if __name__ == '__main__':
# Sample use
p = PrintActor()
p.start()
p.send("Hello")
p.send("World")
p.close()
p.join()
|
In this example, Actor instances are things that you simply send a message to using their send() method. The close() method is programmed to shut down the actor by placing a special sentinel value (ActorExit) on the queue. Users define new actors by inheriting from Actor and redefining the run() method to implement their custom processing. The usage of the ActorExit exception is such that user-defined code can be programmed to catch the termination request and handle it if appropriate (the exception is raised by the get() method and propagated).
For example, you could pass tagged messages in the form of tuples and have actors take different courses of action like this:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
from actor import Actor
class TaggedActor(Actor):
def run(self):
while True:
tag, *payload = self.recv()
getattr(self,"do_"+tag)(*payload)
# Methods correponding to different message tags
def do_A(self, x):
print("Running A", x)
def do_B(self, x, y):
print("Running B", x, y)
# Example
if __name__ == '__main__':
a = TaggedActor()
a.start()
a.send(('A', 1)) # Invokes do_A(1)
a.send(('B', 2, 3)) # Invokes do_B(2,3)
a.close()
a.join()
|
Here is a variation of an actor that allows arbitrary functions to be executed in a worker and results to be communicated back using a special Result object:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
|
from actor import Actor
from threading import Event
class Result:
def __init__(self):
self._evt = Event()
self._result = None
def set_result(self, value):
self._result = value
self._evt.set()
def result(self):
self._evt.wait()
return self._result
class Worker(Actor):
def submit(self, func, *args, **kwargs):
r = Result()
self.send((func, args, kwargs, r))
return r
def run(self):
while True:
func, args, kwargs, r = self.recv()
r.set_result(func(*args, **kwargs))
# Example use
if __name__ == '__main__':
worker = Worker()
worker.start()
r = worker.submit(pow, 2, 3)
print(r.result())
worker.close()
worker.join()
|
Last, but not least, the concept of “sending” a task a message is something that can be scaled up into systems involving multiple processes or even large distributed systems. For example, the send() method of an actor-like object could be programmed to transmit data on a socket connection or deliver it via some kind of messaging infrastructure (e.g., AMQP, ZMQ, etc.).
Publish/Subscribe model
To implement publish/subscribe messaging, you typically introduce a separate “exchange” or “gateway” object that acts as an intermediary for all messages. That is, instead of directly sending a message from one task to another, a message is sent to the exchange and it delivers it to one or more attached tasks.
Messages will be delivered to an exchange and the exchange will deliver them to attached subscribers.
The concept of tasks or threads sending messages to one another (often via queues) is easy to implement and quite popular. However, the benefits of using a public/subscribe (pub/sub) model instead are often overlooked.
- First, the use of an exchange can simplify much of the plumbing involved in setting up communicating threads. In practice, it can make it easier to decouple various tasks in the program.
- Second, the ability of the exchange to broadcast messages to multiple subscribers opens up new communication patterns.
- Last, but not least, a notable aspect of the implementation is that it works with a variety of task-like objects.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
from collections import defaultdict
class Exchange:
def __init__(self):
self._subscribers = set()
def attach(self, task):
self._subscribers.add(task)
def detach(self, task):
self._subscribers.remove(task)
def send(self, msg):
for subscriber in self._subscribers:
subscriber.send(msg)
# Dictionary of all created exchanges
_exchanges = defaultdict(Exchange)
# Return the Exchange instance associated with a given name
def get_exchange(name):
return _exchanges[name]
if __name__ == '__main__':
# Example task (just for testing)
class Task:
def __init__(self, name):
self.name = name
def send(self, msg):
print('{} got: {!r}'.format(self.name, msg))
task_a = Task('A')
task_b = Task('B')
exc = get_exchange('spam')
exc.attach(task_a)
exc.attach(task_b)
exc.send('msg1')
exc.send('msg2')
exc.detach(task_a)
exc.detach(task_b)
exc.send('msg3')
|
Polling Multiple Thread Queues
if you don’t use the socket technique, your only option is to write code that cycles through the queues and uses a timer:
1
2
3
4
5
6
7
8
9
|
import time
def consumer(queues):
while True:
for q in queues:
if not q.empty():
item = q.get()
print('Got:', item)
# Sleep briefly to avoid 100% CPU
time.sleep(0.01)
|
This might work for certain kinds of problems, but it’s clumsy and introduces other
weird performance problems.
A common solution to polling problems involves a little-known trick involving a hidden loopback network connection.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
import queue
import socket
import os
class PollableQueue(queue.Queue):
def __init__(self):
super().__init__()
# Create a pair of connected sockets
if os.name == 'posix':
self._putsocket, self._getsocket = socket.socketpair()
else:
# Compatibility on non-POSIX systems
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('127.0.0.1', 0))
server.listen(1)
self._putsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._putsocket.connect(server.getsockname())
self._getsocket, _ = server.accept()
server.close()
def fileno(self):
return self._getsocket.fileno()
def put(self, item):
super().put(item)
self._putsocket.send(b'x')
def get(self):
self._getsocket.recv(1)
return super().get()
# Example code that performs polling:
if __name__ == '__main__':
import select
import threading
import time
def consumer(queues):
'''
Consumer that reads data on multiple queues simultaneously
'''
while True:
can_read, _, _ = select.select(queues,[],[])
for r in can_read:
item = r.get()
print('Got:', item)
q1 = PollableQueue()
q2 = PollableQueue()
q3 = PollableQueue()
t = threading.Thread(target=consumer, args=([q1,q2,q3],))
t.daemon = True
t.start()
# Feed data to the queues
q1.put(1)
q2.put(10)
q3.put('hello')
q2.put(15)
# Give thread time to run
time.sleep(1)
|
Concurrency with multiprocessing
Launching a Daemon Process on Unix
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
|
#!/usr/bin/env python3
# daemon.py
import os
import sys
import atexit
import signal
def daemonize(pidfile, *, stdin='/dev/null',
stdout='/dev/null',
stderr='/dev/null'):
if os.path.exists(pidfile):
raise RuntimeError('Already running')
# First fork (detaches from parent)
try:
if os.fork() > 0:
raise SystemExit(0) # Parent exit
except OSError as e:
raise RuntimeError('fork #1 failed.')
os.chdir('/')
os.umask(0)
os.setsid()
# Second fork (relinquish session leadership)
try:
if os.fork() > 0:
raise SystemExit(0)
except OSError as e:
raise RuntimeError('fork #2 failed.')
# Flush I/O buffers
sys.stdout.flush()
sys.stderr.flush()
# Replace file descriptors for stdin, stdout, and stderr
with open(stdin, 'rb', 0) as f:
os.dup2(f.fileno(), sys.stdin.fileno())
with open(stdout, 'ab', 0) as f:
os.dup2(f.fileno(), sys.stdout.fileno())
with open(stderr, 'ab', 0) as f:
os.dup2(f.fileno(), sys.stderr.fileno())
# Write the PID file
with open(pidfile,'w') as f:
print(os.getpid(),file=f)
# Arrange to have the PID file removed on exit/signal
atexit.register(lambda: os.remove(pidfile))
# Signal handler for termination (required)
def sigterm_handler(signo, frame):
raise SystemExit(1)
signal.signal(signal.SIGTERM, sigterm_handler)
def main():
import time
sys.stdout.write('Daemon started with pid {}\n'.format(os.getpid()))
while True:
sys.stdout.write('Daemon Alive! {}\n'.format(time.ctime()))
time.sleep(10)
if __name__ == '__main__':
PIDFILE = '/tmp/daemon.pid'
if len(sys.argv) != 2:
print('Usage: {} [start|stop]'.format(sys.argv[0]), file=sys.stderr)
raise SystemExit(1)
if sys.argv[1] == 'start':
try:
daemonize(PIDFILE,
stdout='/tmp/daemon.log',
stderr='/tmp/dameon.log')
except RuntimeError as e:
print(e, file=sys.stderr)
raise SystemExit(1)
main()
elif sys.argv[1] == 'stop':
if os.path.exists(PIDFILE):
with open(PIDFILE) as f:
os.kill(int(f.read()), signal.SIGTERM)
else:
print('Not running', file=sys.stderr)
raise SystemExit(1)
else:
print('Unknown command {!r}'.format(sys.argv[1]), file=sys.stderr)
raise SystemExit(1)
|
Concurrency with coroutines
Simple Generators
Using a yield expression in a function’s body causes that function to be a generator function, and using it in an async def function’s body causes that coroutine function to be an asynchronous generator function.
1
2
3
4
5
|
def gen(): # defines a generator function
yield 123
async def agen(): # defines an asynchronous generator function
yield 123
|
1
2
3
4
5
|
def fib():
a, b = 0, 1
while True:
yield b
a, b = b, a+b
|
Coroutines via Enhanced Generators
How coroutines evolved from generators
-
Redefine yield to be an expression, rather than a statement.
A coroutine is syntactically like a generator: just a function with the yield keyword in its body. However, in a coroutine, yield usually appears on the right side of an expression, e.g. datum = yield, and it may or may not produce a value — if there is no expression after the yield keyword, the generator yields None.
-
The value passed to send() method on a generator defines what gets returned when the yield statement wakes back up. You can only call send() method if the coroutine is currently suspended.
1
2
3
4
5
6
7
8
9
10
|
inspect.getgeneratorstate(…)
'GEN_CREATED'
Waiting to start execution.
'GEN_RUNNING'
Currently being executed by the interpreter, only in a multi-threaded application
'GEN_SUSPENDED'
Currently suspended at a yield expression.
'GEN_CLOSED'
Execution has completed.
|
-
It’s also possible to use the throw() method of a generator to raise an arbitrary execution at the yield statement.
-
It’s also possible to execute a close() method on a generator to raise GeneratorExit exception at the yield statement, which stops execution. If desired, a generator can catch this exception and perform cleanup actions.
-
A generator can now return a value; previously, providing a value to the return
statement inside a generator raised a SyntaxError.
-
The yield from syntax enables complex generators to be refactored into smaller,
nested generators while avoiding a lot of boilerplate code previously required for a
generator to delegate to subgenerators.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
>>> def simple_coroutine():
... print('-> coroutine started')
... x = yield
... print('-> coroutine received:', x)
...
>>> my_coro = simple_coroutine()
>>> my_coro
<generator object simple_coroutine at 0x100c2be10>
# The first call is next(…) because the generator hasn’t started so it’s not waiting in a yield and we can’t send it any data initially.
>>> next(my_coro)
-> coroutine started
>>> my_coro.send(42)
-> coroutine received: 42
# now the coroutine resumes and runs until the next yield or termination.
Traceback (most recent call last):
...
StopIteration
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
>>> def simple_coro2(a):
... print('-> Started: a =', a)
... b = yield a
... print('-> Received: b =', b)
... c = yield a + b
... print('-> Received: c =', c)
...
>>> my_coro2 = simple_coro2(14)
>>> from inspect import getgeneratorstate
>>> getgeneratorstate(my_coro2)
'GEN_CREATED'
# Advance coroutine to first yield, yielding value of a and suspending to wait for value to be assigned to b.
>>> next(my_coro2)
-> Started: a = 14
14
>>> getgeneratorstate(my_coro2)
'GEN_SUSPENDED'
# The yield expression evaluates to 28 and that number is bound to b.
# the value of a + b is yielded (42) and the coroutine is suspended waiting for the value to be assigned to c.
>>> my_coro2.send(28)
-> Received: b = 28
42
>>> my_coro2.send(99)
-> Received: c = 99
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
StopIteration
>>> getgeneratorstate(my_coro2)
'GEN_CLOSED'
|
It’s crucial to understand that the execution of the coroutine is suspended exactly at the yield keyword. As mentioned before, in an assignment statement the code to the right of the = is evaluated before the actual assignment happens.
Decorators for coroutine priming
1
2
3
4
5
6
7
8
9
10
|
from functools import wraps
def coroutine(func):
"""Decorator: primes `func` by advancing to first `yield`"""
@wraps(func)
def primer(*args,**kwargs):
gen = func(*args,**kwargs)
next(gen)
return gen
return primer
|
1
2
3
4
5
6
7
8
9
10
|
# A simple consumer decorator that makes a generator function automatically advance to its first yield point when initially called
def consumer(func):
def wrapper(*args,**kw):
gen = func(*args, **kw)
gen.next()
return gen
wrapper.__name__ = func.__name__
wrapper.__dict__ = func.__dict__
wrapper.__doc__ = func.__doc__
return wrapper
|
The yield from syntax automatically primes the coroutine called by it, making it incompatible with decorators such as @coroutine. The asyncio.coroutine decorator is designed to work with yield form so it does not prime the coroutine.
Coroutine termination and exception handling
An unhandled exception within a coroutine propagates to the caller of the next or send which triggered it. An unhandled exception will kill a coroutine.
-
generator.throw(exc_type[, exc_value[, traceback]])
Causes the yield expression where the generator was paused to raise the exception
given. If the exception is handled by the generator, flow advances to the next
yield, and the value yielded becomes the value of the generator.throw call. If the exception is not handled by the generator, it propagates to the context of the caller.
-
generator.close()
Causes the yield expression where the generator was paused to raise a GeneratorExit exception. No error is reported to the caller if the generator does not handle that exception or raises StopIteration — usually by running to completion. When receiving a GeneratorExit the generator must not yield a value, otherwise a RuntimeError is raised. If any other exception is raised by the generator, it propagates to the caller.
1
2
3
4
5
6
7
8
9
10
11
12
13
|
class DemoException(Exception):
"""An exception type for the demonstration."""
def demo_exc_handling():
print('-> coroutine started')
while True:
try:
x = yield
except DemoException:
print('*** DemoException handled. Continuing...')
else:
print('-> coroutine received: {!r}'.format(x))
raise RuntimeError('This line should never run.')
|
If it’s necessary that some cleanup code is run no matter how the coroutine ends, you need to wrap the relevant part of the coroutine body in a try/finally block.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
class DemoException(Exception):
"""An exception type for the demonstration."""
def demo_finally():
print('-> coroutine started')
try:
while True:
try:
x = yield
except DemoException:
print('*** DemoException handled. Continuing...')
else:
print('-> coroutine received: {!r}'.format(x))
finally:
print('-> coroutine ending')
|
Returning a value from a coroutine
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from collections import namedtuple
Result = namedtuple('Result', 'count average')
def averager():
total = 0.0
count = 0
average = None
while True:
term = yield
if term is None:
break # <1>
total += term
count += 1
average = total/count
return Result(count, average)
|
Note that the value of the return expression is smuggled to the caller as an attribute of
the StopIteration exception.
1
2
3
4
5
6
7
8
9
10
11
12
|
>>> coro_avg = averager()
>>> next(coro_avg)
>>> coro_avg.send(10)
>>> coro_avg.send(30)
>>> coro_avg.send(6.5)
>>> try:
... coro_avg.send(None)
... except StopIteration as exc:
... result = exc.value
...
>>> result
Result(count=3, average=15.5)
|
Using yield from
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
from collections import Iterable
def flatten(items, ignore_types=(str, bytes)):
for x in items:
if isinstance(x, Iterable) and not isinstance(x, ignore_types):
yield from flatten(x)
else:
yield x
items = [1, 2, [3, 4, [5, 6], 7], 8]
# Produces 1 2 3 4 5 6 7 8
for x in flatten(items):
print(x)
items = ['Dave', 'Paula', ['Thomas', 'Lewis']]
for x in flatten(items):
print(x)
|
when a generator gen calls yield from subgen(), the subgen takes over and will yield values to the caller of gen; the caller will in effect drive subgen directly. Meanwhile gen will be blocked, waiting until subgen terminantes.
The delegating generator works as a pipe. One delegating generator uses yield from to call a subgenerator, which itself is a delegating generator calling another subgenerator with yield from and so on. Eventually this chain must end in a simple generator that uses just yield, but it may also end in any iterable object。
Every yield from chain must be driven by a client that calls next(…) or .send(…) on
the outermost delegating generator.
Examples
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
|
@consumer
def jpeg_writer(dirname):
fileno = 1
while True:
filename = os.path.join(dirname,"page%04d.jpg" % fileno)
write_jpeg((yield), filename)
fileno += 1
@consumer
def thumbnail_pager(pagesize, thumbsize, destination):
while True:
page = new_image(pagesize)
rows, columns = pagesize / thumbsize
pending = False
try:
for row in xrange(rows):
for column in xrange(columns):
thumb = create_thumbnail((yield), thumbsize)
page.write(
thumb, col*thumbsize.x, row*thumbsize.y )
pending = True
except GeneratorExit:
# close() was called, so flush any pending output
if pending:
destination.send(page)
# then close the downstream consumer, and exit
destination.close()
return
else:
# we finished a page full of thumbnails, so send it
# downstream and keep on looping
destination.send(page)
# Put them together to make a function that makes thumbnail
# pages from a list of images and other parameters.
#
def write_thumbnails(pagesize, thumbsize, images, output_dir):
pipeline = thumbnail_pager(
pagesize, thumbsize, jpeg_writer(output_dir)
)
for image in images:
pipeline.send(image)
pipeline.close()
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
# A simple co-routine scheduler or trampoline that lets coroutines call other coroutines by yielding the coroutine they wish to invoke.
import collections
class Trampoline:
"""Manage communications between coroutines"""
running = False
def __init__(self):
self.queue = collections.deque()
def add(self, coroutine):
"""Request that a coroutine be executed"""
self.schedule(coroutine)
def run(self):
result = None
self.running = True
try:
while self.running and self.queue:
func = self.queue.popleft()
result = func()
return result
finally:
self.running = False
def stop(self):
self.running = False
def schedule(self, coroutine, stack=(), val=None, *exc):
def resume():
value = val
try:
if exc:
value = coroutine.throw(value,*exc)
else:
value = coroutine.send(value)
except:
if stack:
# send the error back to the "caller"
self.schedule(
stack[0], stack[1], *sys.exc_info()
)
else:
# Nothing left in this pseudothread to
# handle it, let it propagate to the
# run loop
raise
if isinstance(value, types.GeneratorType):
# Yielded to a specific coroutine, push the
# current one on the stack, and call the new
# one with no args
self.schedule(value, (coroutine,stack))
elif stack:
# Yielded a result, pop the stack and send the
# value to the caller
self.schedule(stack[0], stack[1], value)
# else: this pseudothread has ended
self.queue.append(resume)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
# coroutine function that echos data back on a connected
# socket
#
def echo_handler(sock):
while True:
try:
data = yield nonblocking_read(sock)
yield nonblocking_write(sock, data)
except ConnectionLost:
pass # exit normally if connection lost
# coroutine function that listens for connections on a
# socket, and then launches a service "handler" coroutine
# to service the connection
#
def listen_on(trampoline, sock, handler):
while True:
# get the next incoming connection
connected_socket = yield nonblocking_accept(sock)
# start another coroutine to handle the connection
trampoline.add( handler(connected_socket) )
# Create a scheduler to manage all our coroutines
t = Trampoline()
# Create a coroutine instance to run the echo_handler on
# incoming connections
#
server = listen_on(
t, listening_socket("localhost","echo"), echo_handler
)
# Add the coroutine to the scheduler
t.add(server)
# loop forever, accepting connections and servicing them
# "in parallel"
#
t.run()
|
Coroutines with async and await syntax
The following new syntax is used to declare a native coroutine:
1
2
|
async def read_data(db):
pass
|
async def functions are always coroutines, even if they do not contain await expressions.CO_COROUTINE is used to mark native coroutines (defined with new syntax).StopIteration exceptions are not propagated out of coroutines, and are replaced with a RuntimeError.
The following new syntax allows interoperability between existing generator-based coroutines in asyncio and native coroutines:
1
2
3
4
|
@types.coroutine
def process_data(db):
data = yield from read_data(db)
...
|
- The function applies
CO_ITERABLE_COROUTINE flag to generator- function’s code object, making it return a coroutine object.
1
2
3
|
async def read_data(db):
data = await db.fetch('SELECT ...')
...
|
await, similarly to yield from, suspends execution of read_data coroutine until db.fetch awaitable completes and returns the result data.
asynchronous context managers:
1
2
|
async with EXPR as VAR:
BLOCK
|
1
2
3
4
5
6
7
|
async def commit(session, data):
...
async with session.transaction():
...
await session.update(data)
...
|
Code that needs locking also looks lighter:
instead of:
1
2
|
with (yield from lock):
...
|
An example of asynchronous iterable:
1
2
3
4
5
6
7
8
9
10
11
12
13
|
class AsyncIterable:
def __aiter__(self):
return self
async def __anext__(self):
data = await self.fetch_data()
if data:
return data
else:
raise StopAsyncIteration
async def fetch_data(self):
...
|
A new statement for iterating through asynchronous iterators:
1
2
3
4
|
async for TARGET in ITER:
BLOCK
else:
BLOCK2
|
Coroutines are based on generators internally, thus they share the implementation. Similarly to generator objects, coroutines have throw(), send() and close() methods.
Using Generators As an Alternative to Threads
Specifically, the fundamental behavior of yield is that it causes a generator to suspend its execution. By suspending execution, it is possible to write a scheduler that treats generators as a kind of “task” and alternates their execution using a kind of cooperative task switching.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
# A very simple example of a coroutine/generator scheduler
# Two simple generator functions
def countdown(n):
while n > 0:
print("T-minus", n)
yield
n -= 1
print("Blastoff!")
def countup(n):
x = 0
while x < n:
print("Counting up", x)
yield
x += 1
from collections import deque
class TaskScheduler:
def __init__(self):
self._task_queue = deque()
def new_task(self, task):
'''
Admit a newly started task to the scheduler
'''
self._task_queue.append(task)
def run(self):
'''
Run until there are no more tasks
'''
while self._task_queue:
task = self._task_queue.popleft()
try:
# Run until the next yield statement
next(task)
self._task_queue.append(task)
except StopIteration:
# Generator is no longer executing
pass
# Example use
sched = TaskScheduler()
sched.new_task(countdown(10))
sched.new_task(countdown(5))
sched.new_task(countup(15))
sched.run()
|
In practice, you probably wouldn’t use generators to implement concurrency for something as simple as shown. Instead, you might use generators to replace the use of threads when implementing actors or network servers.
Actor Model Using generators
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
from collections import deque
class ActorScheduler:
def __init__(self):
self._actors = { } # Mapping of names to actors
self._msg_queue = deque() # Message queue
def new_actor(self, name, actor):
'''
Admit a newly started actor to the scheduler and give it a name
'''
self._msg_queue.append((actor,None))
self._actors[name] = actor
def send(self, name, msg):
'''
Send a message to a named actor
'''
actor = self._actors.get(name)
if actor:
self._msg_queue.append((actor,msg))
def run(self):
'''
Run as long as there are pending messages.
'''
while self._msg_queue:
actor, msg = self._msg_queue.popleft()
try:
actor.send(msg)
except StopIteration:
pass
# Example use
if __name__ == '__main__':
def printer():
while True:
msg = yield
print('Got:', msg)
def counter(sched):
while True:
# Receive the current count
n = yield
if n == 0:
break
# Send to the printer task
sched.send('printer', n)
# Send the next count to the counter task (recursive)
sched.send('counter', n-1)
sched = ActorScheduler()
# Create the initial actors
sched.new_actor('printer', printer())
sched.new_actor('counter', counter(sched))
# Send an initial message to the counter to initiate
sched.send('counter', 10000)
sched.run()
|
Concurrent Network Application Using Generators
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
from collections import deque
from select import select
# This class represents a generic yield event in the scheduler
class YieldEvent:
def handle_yield(self, sched, task):
pass
def handle_resume(self, sched, task):
pass
# Task Scheduler
class Scheduler:
def __init__(self):
self._numtasks = 0 # Total num of tasks
self._ready = deque() # Tasks ready to run
self._read_waiting = {} # Tasks waiting to read
self._write_waiting = {} # Tasks waiting to write
# Poll for I/O events and restart waiting tasks
def _iopoll(self):
rset,wset,eset = select(self._read_waiting,
self._write_waiting,[])
for r in rset:
evt, task = self._read_waiting.pop(r)
evt.handle_resume(self, task)
for w in wset:
evt, task = self._write_waiting.pop(w)
evt.handle_resume(self, task)
def new(self,task):
'''
Add a newly started task to the scheduler
'''
self._ready.append((task, None))
self._numtasks += 1
def add_ready(self, task, msg=None):
'''
Append an already started task to the ready queue.
msg is what to send into the task when it resumes.
'''
self._ready.append((task, msg))
# Add a task to the reading set
def _read_wait(self, fileno, evt, task):
self._read_waiting[fileno] = (evt, task)
# Add a task to the write set
def _write_wait(self, fileno, evt, task):
self._write_waiting[fileno] = (evt, task)
def run(self):
'''
Run the task scheduler until there are no tasks
'''
while self._numtasks:
if not self._ready:
self._iopoll()
task, msg = self._ready.popleft()
try:
# Run the coroutine to the next yield
r = task.send(msg)
if isinstance(r, YieldEvent):
r.handle_yield(self, task)
else:
raise RuntimeError('unrecognized yield event')
except StopIteration:
self._numtasks -= 1
# Example implementation of coroutine based socket I/O
class ReadSocket(YieldEvent):
def __init__(self, sock, nbytes):
self.sock = sock
self.nbytes = nbytes
def handle_yield(self, sched, task):
sched._read_wait(self.sock.fileno(), self, task)
def handle_resume(self, sched, task):
data = self.sock.recv(self.nbytes)
sched.add_ready(task, data)
class WriteSocket(YieldEvent):
def __init__(self, sock, data):
self.sock = sock
self.data = data
def handle_yield(self, sched, task):
sched._write_wait(self.sock.fileno(), self, task)
def handle_resume(self, sched, task):
nsent = self.sock.send(self.data)
sched.add_ready(task, nsent)
class AcceptSocket(YieldEvent):
def __init__(self, sock):
self.sock = sock
def handle_yield(self, sched, task):
sched._read_wait(self.sock.fileno(), self, task)
def handle_resume(self, sched, task):
r = self.sock.accept()
sched.add_ready(task, r)
# Wrapper around a socket object for use with yield
class Socket(object):
def __init__(self, sock):
self._sock = sock
def recv(self, maxbytes):
return ReadSocket(self._sock, maxbytes)
def send(self, data):
return WriteSocket(self._sock, data)
def accept(self):
return AcceptSocket(self._sock)
def __getattr__(self, name):
return getattr(self._sock, name)
if __name__ == '__main__':
from socket import socket, AF_INET, SOCK_STREAM
import time
# Example of a function involving generators. This should
# be called using line = yield from readline(sock)
def readline(sock):
chars = []
while True:
c = yield sock.recv(1)
if not c:
break
chars.append(c)
if c == b'\n':
break
return b''.join(chars)
# Echo server using generators
class EchoServer:
def __init__(self,addr,sched):
self.sched = sched
sched.new(self.server_loop(addr))
def server_loop(self,addr):
s = Socket(socket(AF_INET,SOCK_STREAM))
s.bind(addr)
s.listen(5)
while True:
c,a = yield s.accept()
print('Got connection from ', a)
self.sched.new(self.client_handler(Socket(c)))
def client_handler(self,client):
while True:
line = yield from readline(client)
if not line:
break
line = b'GOT:' + line
while line:
nsent = yield client.send(line)
line = line[nsent:]
client.close()
print('Client closed')
sched = Scheduler()
EchoServer(('',16000),sched)
sched.run()
|
Finally, if programming with generators, it is important to stress that there are some major limitations.
Concurrency with futures
Creating a Thread Pool
Generally, you only want to use thread pools for I/O-bound processing.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
from concurrent import futures
from flags import save_flag, get_flag, show, main
MAX_WORKERS = 20
def download_one(cc):
image = get_flag(cc)
show(cc)
save_flag(image, cc.lower() + '.gif')
return cc
# BEGIN FLAGS_THREADPOOL_AS_COMPLETED
def download_many(cc_list):
cc_list = cc_list[:5] # <1>
with futures.ThreadPoolExecutor(max_workers=3) as executor: # <2>
to_do = []
for cc in sorted(cc_list): # <3>
future = executor.submit(download_one, cc) # <4>
to_do.append(future) # <5>
msg = 'Scheduled for {}: {}'
print(msg.format(cc, future)) # <6>
results = []
for future in futures.as_completed(to_do): # <7>
res = future.result() # <8>
msg = '{} result: {!r}'
print(msg.format(future, res)) # <9>
results.append(res)
return len(results)
# END FLAGS_THREADPOOL_AS_COMPLETED
if __name__ == '__main__':
main(download_many)
|
a simple TCP server that uses a thread-pool to serve clients:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
from socket import AF_INET, SOCK_STREAM, socket
from concurrent.futures import ThreadPoolExecutor
def echo_client(sock, client_addr):
'''
Handle a client connection
'''
print('Got connection from', client_addr)
while True:
msg = sock.recv(65536)
if not msg:
break
sock.sendall(msg)
print('Client closed connection')
sock.close()
def echo_server(addr):
print('Echo server running at', addr)
pool = ThreadPoolExecutor(128)
sock = socket(AF_INET, SOCK_STREAM)
sock.bind(addr)
sock.listen(5)
while True:
client_sock, client_addr = sock.accept()
pool.submit(echo_client, client_sock, client_addr)
echo_server(('',15000))
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
from concurrent.futures import ThreadPoolExecutor
import urllib.request
def fetch_url(url):
u = urllib.request.urlopen(url)
data = u.read()
return data
pool = ThreadPoolExecutor(10)
# Submit work to the pool
a = pool.submit(fetch_url, 'http://www.python.org')
b = pool.submit(fetch_url, 'http://www.pypy.org')
# Get the results back
x = a.result()
y = b.result()
|
Parallel Programming Using Process Pool
Use futures.ProcessPoolExecutor to run CPU-intensive tasks.
If you are are trying to parallelize a list comprehension or a map() operation, you use pool.map(). Alternatively, you can manually submit single tasks using the pool.submit() method.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
|
import sys
import time
from concurrent import futures
from random import randrange
from arcfour import arcfour
JOBS = 12
SIZE = 2**18
KEY = b"'Twas brillig, and the slithy toves\nDid gyre"
STATUS = '{} workers, elapsed time: {:.2f}s'
def arcfour_test(size, key):
in_text = bytearray(randrange(256) for i in range(size))
cypher_text = arcfour(key, in_text)
out_text = arcfour(key, cypher_text)
assert in_text == out_text, 'Failed arcfour_test'
return size
def main(workers=None):
if workers:
workers = int(workers)
t0 = time.time()
with futures.ProcessPoolExecutor(workers) as executor:
actual_workers = executor._max_workers
to_do = []
for i in range(JOBS, 0, -1):
size = SIZE + int(SIZE / JOBS * (i - JOBS/2))
job = executor.submit(arcfour_test, size, KEY)
to_do.append(job)
for future in futures.as_completed(to_do):
res = future.result()
print('{:.1f} KB'.format(res/2**10))
print(STATUS.format(actual_workers, time.time() - t0))
if __name__ == '__main__':
if len(sys.argv) == 2:
workers = int(sys.argv[1])
else:
workers = None
main(workers)
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
import gzip
import io
import glob
from concurrent import futures
def find_robots(filename):
'''
Find all of the hosts that access robots.txt in a single log file
'''
robots = set()
with gzip.open(filename) as f:
for line in io.TextIOWrapper(f,encoding='ascii'):
fields = line.split()
if fields[6] == '/robots.txt':
robots.add(fields[0])
return robots
def find_all_robots(logdir):
'''
Find all hosts across and entire sequence of files
'''
files = glob.glob(logdir+"/*.log.gz")
all_robots = set()
with futures.ProcessPoolExecutor() as pool:
for robots in pool.map(find_robots, files):
all_robots.update(robots)
return all_robots
if __name__ == '__main__':
import time
start = time.time()
robots = find_all_robots("logs")
end = time.time()
for ipaddr in robots:
print(ipaddr)
print('Took {:f} seconds'.format(end-start))
|
Instead of blocking, you can also arrange to have a callback function triggered upon
completion instead. The user-supplied callback function receives an instance of Future that must be used to obtain the actual result (i.e., by calling its result() method).
1
2
3
4
5
6
|
def when_done(r):
print('Got:', r.result())
with ProcessPoolExecutor() as pool:
future_result = pool.submit(work, arg)
future_result.add_done_callback(when_done)
|
Concurrency with asyncio