A look at Tokio: how this asynchronous event handler works
And what is it used in the framework for private Exchum blockchains?
Tokio is a framework for developing scalable network applications on Rust using components for working with asynchronous I / O. Tokio often serves as the basis for other libraries and implementations of high-performance protocols. Despite the fact that it is a fairly young framework, it has already managed to become part of the middleware ecosystem.
Although Tokio is criticized for being too complex to master, it is already used in production environments, since the code written in Tokio is easier to maintain. For example, it has already been integrated into hyper , tower-grpc and Sonduit . We also turned to this solution when developing our Exonum platform.
Work on Exonum began in 2016, when Tokio did not exist yet, so we first used the Mio v0.5 library. With the advent of Tokio, it became clear that the Mio library used was outdated, moreover, with its help it was difficult to organize the event model Exonum. The model included several types of events (network messages, timeouts, messages from the REST API, etc.), as well as their sorting by priority.
')
Each event entails a change in the state of the node, which means they must be processed in one stream, in a certain order and according to one principle. On Mio, the processing scheme for each event had to be described manually, which, while maintaining the code (adding / changing parameters), could result in a large number of errors. Tokio made it possible to simplify this process with built-in functions.
Below we describe the components of the Tokio stack, which allow you to efficiently organize the processing of asynchronous tasks.
/ image by Kevin Dooley CC
Tokio architecture
At its core, Tokio is a “wrapper” over Mio. Mio is a Rust crack that provides an API for low-level I / O and is platform independent — it works with several tools: epoll on Linux, kqueue on Mac OS, or IOCP on Windows. Thus, the Tokio architecture can be represented as follows:
Futures
As can be seen from the diagram above, the main functional component of Tokio, is futures - this is crate Rust, which allows you to work with asynchronous code in a synchronous manner. In other words, the library makes it possible to operate with code that implements tasks that have not yet been performed, as if they were already completed.
Essentially, futures are values ​​that will be calculated in the future, but not yet known. In the futures format, you can present all sorts of events: database requests, timeouts, lengthy tasks for the CPU, reading information from a socket, etc., and synchronize them.
An example of a future in real life is the notification of delivery of a registered letter by mail: upon completion of delivery, the sender is sent a notification of the successful receipt by the addressee of the letter. After receiving the notification, the sender determines what action to take next.
Developer David Simmons ( David Simmons ), who collaborated with Intel, Genuity and Sparco Media, as an example of the organization of asynchronous I / O using futures leads the exchange of messages with an HTTP server.
Imagine that a server each time generates a new thread (thread) for an established connection. With synchronous I / O, the system first reads the bytes in order, then processes the information and writes the result back. At the same time, at the moment of reading / writing, the thread will not be able to continue execution (it is blocked) until the operation is completed. This leads to the fact that with a large number of connections, difficulties arise in scaling (the so-called C10k problem).
In the case of asynchronous processing, the thread enqueues the I / O request and continues execution (that is, it does not block). The system reads / writes after a while, and the thread asks if the query was executed before using the results. Thus, futures are able to perform different tasks, for example, one can read a request, the second can process it, and the third can form a response.
In the futures kreite, the type of Future is defined, which is the core of the entire library. This type is defined for objects that are not executed immediately, but after some time. Its main part is expressed in the code as follows:
trait Future { type Item; type Error; fn poll(&mut self) -> Poll<Self::Item, Self::Error>; fn wait(self) -> Result<Self::Item, Self::Error> { ... } fn map<F, U>(self, f: F) -> Map<Self, F> where F: FnOnce(Self::Item) -> U { ... } } The “heart” of the Future type is the poll () method. It is responsible for sending the completion indicator, call waiting, or counted value. In this case, futures are launched in the context of the task . A task is associated with only one future, but the latter may be complex, that is, contain several other futures combined by the join_all () or and_then () commands . For example:
let client_to_server = copy(client_reader, server_writer) .and_then(|(n, _, server_writer)| { shutdown(server_writer).map(move |_| n) }); The executor is responsible for coordinating the task / future. If there are several tasks running at the same time, and some of them are waiting for the results of external asynchronous events (for example, reading data from the network / socket), the executor must efficiently allocate processor resources for optimal performance. In practice, this occurs due to the "transfer" of processor power to tasks that can be performed while other tasks are blocked due to the lack of external data.
In the case of a deferred task, the executor obtains information that it can be done using the notify () method. An example would be the futures cracker who “wakes up” when you call wait () - the source code of the example is presented in the official Rust repository on GitHub:
pub fn wait_future(&mut self) -> Result<F::Item, F::Error> { ThreadNotify::with_current(|notify| { loop { match self.poll_future_notify(notify, 0)? { Async::NotReady => notify.park(), Async::Ready(e) => return Ok(e), } } }) } Streams
In addition to futures, Tokio works with other components for asynchronous I / O streams. While the future returns only one final result, stream works with a series of events and is able to return several results.
Again, a real-life example: periodic alerts from a temperature sensor can be represented as stream. The sensor will regularly send the temperature measurement value to the user at certain intervals.
The stream may look like this:
trait Stream { type Item; type Error; fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error>; } The mechanics of working with stream are identical to those used for futures: similar combinators are used to transform and modify stream details. Moreover, stream can easily be transformed into the future using the adapter into_future.
Below we will examine in detail the use of futures and stream in our Exonum framework.
Tokio in Exonum
As already mentioned, the Exonum developers decided to use the Tokio library to implement the event loop in the framework.
A simplified diagram of the organization of the event model in Exonum can be represented as follows:
Each node of the network exchanges messages with other nodes. All incoming messages are in the network event queue, where besides them are also internal events (timeouts and internal API events). Each type of event forms a separate stream. But the processing of such events, as noted earlier, is a synchronous process, since it entails changes in the state of the node. An Event Agregator combines several chains of events into one and sends them via the channel to the event loop, where they are processed in order of priority.
When communicating between nodes, Exonum performs the following related operations on each of them:
Connecting to node N (opening a socket, setting up a socket) -> Receiving messages of a node N (receiving bytes from a socket, splitting bytes into messages) -> Forwarding messages to the channel of the current node
let connect_handle = Retry::spawn(handle.clone(), strategy, action) .map_err(into_other) // Configure socket .and_then(move |sock| { sock.set_nodelay(network_config.tcp_nodelay)?; let duration = network_config.tcp_keep_alive.map(Duration::from_millis); sock.set_keepalive(duration)?; Ok(sock) }) // Connect socket with the outgoing channel .and_then(move |sock| { trace!("Established connection with peer={}", peer); let stream = sock.framed(MessagesCodec::new(max_message_len)); let (sink, stream) = stream.split(); let writer = conn_rx .map_err(|_| other_error("Can't send data into socket")) .forward(sink); let reader = stream.for_each(result_ok); reader .select2(writer) .map_err(|_| other_error("Socket error")) .and_then(|res| match res { Either::A((_, _reader)) => Ok("by reader"), Either::B((_, _writer)) => Ok("by writer"), }) }) Due to the fact that futures makes it possible to combine various abstractions in a chain without losing system performance, the program code is broken up into small functional blocks, and, therefore, it becomes easier to maintain.
Using the network is a nontrivial task. To work with a node, it is necessary to connect to it, as well as to ensure the logic of reconnections in case of a break :
.map_err(into_other) In addition, it is necessary to configure the socket :
.and_then(move |sock| { sock.set_nodelay(network_config.tcp_nodelay)?; let duration = network_config.tcp_keep_alive.map(Duration::from_millis); sock.set_keepalive(duration)?; Ok(sock) }) And parse incoming bytes as messages:
let (sink, stream) = stream.split(); Each of the above code blocks, in turn, consists of smaller sections. However, due to the fact that futures allows you to freely combine these blocks, we do not need to think about the internal structure of each of them.
In conclusion, I would like to note that at the moment Exonum as an API uses a slightly outdated version of iron based on the hyper library. However, now we are considering the option of switching to the pure hyper, which Tokio uses.
We offer you some more materials on the topic from our blog on Habré:
Source: https://habr.com/ru/post/351824/
All Articles