Python implementation of the event-driven paradigm using coroutines

An article on how to use Python extended generators to create your own coroutine implementation that switches on receiving events. The simplicity of the code of the resulting module will pleasantly surprise you and clarify the new and little-used features of the language that can be obtained using such generators. The article will help to understand how this works in serious implementations: asyncio , tornado , etc.

Theoretical moments and disclaimer

The concept of coroutine has a very broad interpretation, so you should decide what characteristics they will have in our implementation:

• Performed together in the same thread;
• Execution may be interrupted to wait for a specific event;
• Execution may resume after receiving the expected event;
• May return result on completion.

As a result, we get: event-oriented programming without callback functions and cooperative multitasking . The effect of using such a programming paradigm will be significant only for tasks that react to uneven events. First of all, these are I / O processing tasks: network servers, user interfaces, etc. Another possible use case is the task of calculating the state of characters in the game world. But it is categorically not suitable for tasks that produce long calculations.
It should be clearly understood that while the coroutine that is being executed is not interrupted while waiting for an event, all the others are in a stop state, even if the event they have expected has already occurred.

The basis of everything

In Python, generators are a good basis for all of this, if properly prepared in the literal and figurative sense. More precisely, the extended generators, the API of which was finally formed in the Python version 3.3. In previous versions, the return of the value (result) on the completion of the generator was not implemented and there was no convenient mechanism for calling one generator from another. However, coroutine implementations were before, but due to the limitations of conventional generators, they were not as “beautiful” as what we have. A very nice article on this topic, “A Curious Course on Coroutines and Concurrency”, is its only drawback, because there is no updated version. Such where the implementation of coroutine in python uses the latest innovations in the language, in particular in the API Enhanced Python Generators. The features of the extended generators that we need are discussed below.
The transfer of messages to the coroutine will be built on the possibility of setting the state of the generator. Copy the code below into the Python interpreter window running version 3.3 and above.

def gen_factory(): state = None while True: print("state:", state) state = yield state gen = gen_factory() 

The generator is created, it must be run.

 >>> next(gen) state: None 

Received the original state. Change the state:

 >>> gen.send("OK") state: OK 'OK' 

We see that the state has changed and returned as a result. The following send calls will return the status they have already passed.

Why do we need all this?

Imagine the task: to send greetings to Petrov every two seconds, to Ivanov every three seconds, and to the whole world every five seconds. In the form of Python code, you can think of something like this:

 def hello(name, timeout): while True: sleep(timeout) print(", {}!".format(name)) hello("", 2.0) hello("", 3.0) hello("", 5.0) 

It looks good, but only Petrov will receive greetings. But! A small modification that does not affect the clarity of the code, but on the contrary is a clarifying our thought, and this can already work as expected.

 @coroutine def hello(name, timeout): while True: yield from sleep(timeout) print(", {}!".format(name)) hello("", 2.0) hello("", 3.0) hello("", 5.0) run() 

The code turned out to be in the pythonic way style - it clearly illustrates the problem, linear without calbek, without unnecessary frills with objects, any comments in it are superfluous. It remains only to implement the coroutine decorator, its version of the sleep function and the run function. In the implementation, of course, no frills will not do. But this is also a pythonic way, hiding all the magic behind the facade of the library modules.
')

The most interesting

We call the module with the implementation of simple - concurrency, with the meaning and reflects the fact that it is in fact, will be the implementation of cooperative multitasking. It is clear that the decorator will have to make a generator from a normal function and start it (make the first call next). The construction of the language yield from forwarding the call to the next generator. That is, the sleep function should create a generator in which you can hide all the magic. Only the code of the received event will return to the generator that caused it. Here, the returned result is not processed, the code here can get essentially only one result, meaning that the timeout has expired. Waiting for the same I / O can return different types of events, for example (read / write / timeout). Moreover, generators generated by functions like sleep can return any type of data on yield from and, accordingly, their functionality can be not limited to waiting for events. The run function will start the event dispatcher, its task is to receive the event from the outside and / or generate it inside, determine its recipient and send it.
Let's start with the decorator:

 class coroutine(object): """       .""" _current = None def __init__(self, callable): self._callable = callable def __call__(self, *args, **kwargs): corogen = self._callable(*args, **kwargs) cls = self.__class__ if cls._current is None: try: cls._current = corogen next(corogen) finally: cls._current = None return corogen 

It is designed as a class, a typical technique, as promised, it creates and starts a generator. The construction with _current is added in order to avoid starting the generator, if the decorated function that creates it is called inside the body of another generator. In this case, the first call will be made. It will also help to figure out which generator the event should be transmitted to, so that it falls along the chain into the generator created by the sleep function.

 def sleep(timeout): """     " ".""" corogen = coroutine._current dispatcher.setup_timeout(corogen, timeout) revent = yield return revent 

Here we see the call dispatcher.setup_sleep, this informs the event dispatcher that the generator is waiting for a “time-out” event after the number of seconds specified by the timeout parameter.

 from collections import deque from time import time, sleep as sys_sleep class Dispatcher(object): """   .""" def __init__(self): self._pending = deque() self._deadline = time() + 3600.0 def setup_timeout(self, corogen, timeout): deadline = time() + timeout self._deadline = min([self._deadline, deadline]) self._pending.append([corogen, deadline]) self._pending = deque(sorted(self._pending, key=lambda a: a[1])) def run(self): """   .""" while len(self._pending) > 0: timeout = self._deadline - time() self._deadline = time() + 3600.0 if timeout > 0: sys_sleep(timeout) while len(self._pending) > 0: if self._pending[0][1] <= time(): corogen, _ = self._pending.popleft() try: coroutine._current = corogen corogen.send("timeout") except StopIteration: pass finally: coroutine._current = None else: break dispatcher = Dispatcher() run = lambda: dispatcher.run() 

There is nothing unusual in the event manager code either. Where to send events is determined using the class variable coroutine._current. When the module is loaded, an instance of the class is created, in the working implementation it should of course be a singleton. The collections.deque class is used instead of the list, as it is faster and more useful with its popleft method. Well, that's all, and there is no special magic. All of it is in fact hidden even deeper in the implementation of advanced Python generators. They can only be properly cooked.

Outro

If the topic is interesting, you can continue with the implementation of waiting for input / output events with an asynchronous TCP Echo server as an example. With a real event dispatcher, implemented as a dynamic library written in a different language faster than Python.