Event Model Based on async and await
Back in 2012, when the oil price was still three-digit, and the grass was greener, Microsoft released .NET 4.5, and with it the async / await design. Quite a few articles have already been written about it ( Async in C # ), and most C # developers have studied it well. But have all the use cases been considered, is it possible to squeeze a little more out of await?
The most obvious use of this construct is to wait for some asynchronous operation to complete. The first thing that comes to mind is waiting for I / O. For example, we sent a request to a client and expect a response, then using await we can continue to execute the code after receiving the response, and the code itself will look synchronous. But what if, while waiting, it becomes necessary to interrupt this operation? Then we will have to use the CancellationToken, and if there are several such operations, then the tokens will need to be linked or one common token will be used. In this case, the reason for the cancellation will be hidden from the code using this CancellationToken. In addition to cancellation, the code must support handling of connection loss, timeout, returned errors, etc.
In the classic version, this will result in the use of a CancellationToken to handle cancellation, a try catch to handle a disconnect, and an analysis code for the returned data to evaluate the result of the request. But is it possible to fit all this into a single paradigm? In this article, I propose to consider an alternative event-based approach using async / await syntactic sugar.
Everything necessary for the event model on async / await was designed as an Eventing library and posted on GitHub under the MIT license.
The library has been tested and successfully used on the combat system for more than two years.
The example using Eventing described at the beginning will look like this:
Here we use EventManager, an event manager that implements the IEventManager interface, to wait for MessageReceived and CancelRequested events with a timeout of 50 seconds. Using the WaitFor call, we subscribe to the specified events, and the await call blocks the further execution of the code (but not the stream). It will remain locked until one of the specified events occurs or the timeout expires, after which execution continues in the current synchronization context. But what if the connection with the client is lost during the formation of the subscription? In this case, the code will freeze for 50 seconds, as the client disconnect event will be missed. Let's fix this:
Here we added the ClientDisconnected event and separated the creation of the awaitable eventAwait variable and directly waiting for the event. If we did not separate them, then the client could disconnect after checking client.Connected and waiting for the event, which would lead to the loss of the event. An event filter has also been added that excludes ClientDisconnected events not related to the current client.
To do this, create a class implementing IEvent:
And then call IEventManager.RaiseEvent, for example:
Inheriting from IEvent separates events from other classes and prevents the use of inappropriate instances in the RaiseEvent method. Inheritance is also supported:
If you have a complex system in which there are many simultaneous expected events, using the CancelRequested event instead of the cancellation tokens will allow you to avoid scamming and linking the global and local CancellationToken. This is important, as complex linking increases the likelihood of skipping a memory leak due to token retention.
Some events are periodic, such events can be received using IEventManager.StartReceiving method:
The handler will be called in the context synchronization context for each T event that matches the filter filter, if one is specified. If no synchronization context is specified, then SynchronizationContext.Current will be used.
It uses the same task mechanism on which async / await is based. When WaitFor is called, the event manager creates a task using TaskCompletionSource and subscribes to the selected event types in the message bus.
When an event is generated, the RaiseEvent method is called, which passes the event to the bus, and it selects subscriptions in accordance with the type of event, in which eventTypes includes this type. Next, the subscription handler is called and if it satisfies the filter, the result of the task execution is set and unlocks the await call.
In MessageSubscription.ProccessMessage, the message is passed to the user-defined synchronization context, which avoids delays in sending the message.
Everyone who has worked with async / await knows that after await is completed, the code continues its execution not in the current thread, but in the current synchronization context. This can be a problem if you subscribe to an event using StartReceiving and then call WaitFor, which will cause the class code to be executed simultaneously in different threads (the event handler from StartReceiving and the code after await // how scary to live!). This can be easily fixed with a single-threaded synchronization context included in the library:
Thus, the client will always be executed in the “Client thread” thread, and the server in the “Server thread”. You can write multi-threaded code without thinking about race condition. As a bonus, this will maximize the utilization of a single stream.
The main advantage is the simplicity and testability of the code. If you can argue about the first, everyone understands simplicity in his own way, then with the second paragraph everything is obvious. A multi-threaded application can be tested in one thread, emulating any sequence of events, and for this it is not necessary to create mock objects, any interaction can be reduced to events, and their verification to a RaiseEvent call. NUnit example:
In order not to overwhelm the article with listings, I will provide only a short textual description of one of the systems where Eventing is used. This is a horizontally scalable distributed system consisting of four types of nodes, one of which is a master. The master continuously communicates with all nodes and controls the execution of various operations on them. Each operation can be represented in the form of a finite state machine, where the transition is the occurrence of an event (including a timeout or cancellation). Although for each operation it was possible to implement the automaton in its classical form (which we initially did), it turned out to be much simpler to represent it using Eventing, where the current state was determined by the code execution point, rather than a separate variable. Moreover, at each step all expected events were explicitly listed, which simplified the testing of the white box.
The article discusses the key features and options for using the Eventing library. The library does not pretend to be universal and support highly loaded systems, but calls for a slightly different look at familiar things, and allows you to write code that is safe and easily tested from the point of view of multithreading.
The most obvious use of this construct is to wait for some asynchronous operation to complete. The first thing that comes to mind is waiting for I / O. For example, we sent a request to a client and expect a response, then using await we can continue to execute the code after receiving the response, and the code itself will look synchronous. But what if, while waiting, it becomes necessary to interrupt this operation? Then we will have to use the CancellationToken, and if there are several such operations, then the tokens will need to be linked or one common token will be used. In this case, the reason for the cancellation will be hidden from the code using this CancellationToken. In addition to cancellation, the code must support handling of connection loss, timeout, returned errors, etc.
In the classic version, this will result in the use of a CancellationToken to handle cancellation, a try catch to handle a disconnect, and an analysis code for the returned data to evaluate the result of the request. But is it possible to fit all this into a single paradigm? In this article, I propose to consider an alternative event-based approach using async / await syntactic sugar.
Eventing Library.
Everything necessary for the event model on async / await was designed as an Eventing library and posted on GitHub under the MIT license.
The library has been tested and successfully used on the combat system for more than two years.
Using
The example using Eventing described at the beginning will look like this:
var @event = await this.EventManager.WaitFor(TimeSpan.FromSeconds(50));
if (@event == null)
Log.Info("timeout");
else if (@event is CancelRequested)
Log.Info("Cancelled, reason: {0}", ((CancelRequested) @event).Reason);
else
Log.Info("Message received");
Here we use EventManager, an event manager that implements the IEventManager interface, to wait for MessageReceived and CancelRequested events with a timeout of 50 seconds. Using the WaitFor call, we subscribe to the specified events, and the await call blocks the further execution of the code (but not the stream). It will remain locked until one of the specified events occurs or the timeout expires, after which execution continues in the current synchronization context. But what if the connection with the client is lost during the formation of the subscription? In this case, the code will freeze for 50 seconds, as the client disconnect event will be missed. Let's fix this:
// Создание подписки
var eventAwait = this.EventManager.WaitFor(TimeSpan.FromSeconds(50),
e => !(e is ClientDisconnected) || ((ClientDisconnected)e).id == client.Id); // Фильтр события
if (!client.Connected || cancelRequested) {
// Случай отключения клиента или запроса на отмену во время создания подписки
Log.Info("Client disconnected or cancel requested");
return;
}
// Прерывание кода до наступления события
var @event = await eventAwait;
...
Here we added the ClientDisconnected event and separated the creation of the awaitable eventAwait variable and directly waiting for the event. If we did not separate them, then the client could disconnect after checking client.Connected and waiting for the event, which would lead to the loss of the event. An event filter has also been added that excludes ClientDisconnected events not related to the current client.
How to create an event?
To do this, create a class implementing IEvent:
class CancelRequested : IEvent {
public string Reason { get; set; }
}
And then call IEventManager.RaiseEvent, for example:
this.EventManager.RaiseEvent(new CancelRequested()).
Inheriting from IEvent separates events from other classes and prevents the use of inappropriate instances in the RaiseEvent method. Inheritance is also supported:
class UserCancelRequested : CancelRequested {
}
class SystemCancelRequested : CancelRequested {
}
var @event = await this.EventManager.WaitFor();
if (@event is UserCancelRequested)
...
If you have a complex system in which there are many simultaneous expected events, using the CancelRequested event instead of the cancellation tokens will allow you to avoid scamming and linking the global and local CancellationToken. This is important, as complex linking increases the likelihood of skipping a memory leak due to token retention.
How to subscribe to an event?
Some events are periodic, such events can be received using IEventManager.StartReceiving method:
void StartReceiving(Action handler, object listener, Func filter = null, SynchronizationContext context = null)
where T : IEvent;
The handler will be called in the context synchronization context for each T event that matches the filter filter, if one is specified. If no synchronization context is specified, then SynchronizationContext.Current will be used.
How it works?
It uses the same task mechanism on which async / await is based. When WaitFor is called, the event manager creates a task using TaskCompletionSource and subscribes to the selected event types in the message bus.
// EventManager.cs, создание подписки
var taskCompletionSource = new TaskCompletionSource();
var subscription = new MessageSubscription(
subscriptionId,
message => {
var @event = message as IEvent;
if (filter != null && !filter(@event))
return;
// Устанавливаем результат исполнения задачи
if (taskCompletionSource.TrySetResult(@event))
this.trace.TraceEvent(TraceEventType.Information, 0, "Wait ended: '{0}' - '{1}'",
subscriptionId, message.GetType());
},
this, UnsubscribePolicy.Auto, this.defaultContext, eventTypes);
this.messageBus.Subscribe(subscription);
...
return taskCompletionSource.Task;
When an event is generated, the RaiseEvent method is called, which passes the event to the bus, and it selects subscriptions in accordance with the type of event, in which eventTypes includes this type. Next, the subscription handler is called and if it satisfies the filter, the result of the task execution is set and unlocks the await call.
// EventManager.cs, генерация события
public void RaiseEvent(IEvent @event) {
this.trace.TraceEvent(TraceEventType.Information, 0, "Event: {0}", @event);
this.messageBus.Send(@event);
}
// MessageBus.cs, отправка сообщения
public void Send(object message) {
var messageType = message.GetType();
IMessageSubscription[] subscriptionsForMessage;
lock (this.subscriptions) {
subscriptionsForMessage = this.subscriptions
.Where(s => s.MessagesTypes.Any(type => messageType == type || type.IsAssignableFrom(messageType)))
.ToArray();
}
...
foreach (var subscription in subscriptionsForMessage)
subscription.ProccessMessage(message);
this.UnsubscribeAutoSubscriptions(subscriptionsForMessage);
...
// MessageSubscription.cs
public void ProccessMessage(object message) {
var messageHandler = this.handler;
this.SynchronizationContext.Post(o => messageHandler(message), null);
}
In MessageSubscription.ProccessMessage, the message is passed to the user-defined synchronization context, which avoids delays in sending the message.
Spare my class from multithreading!
Everyone who has worked with async / await knows that after await is completed, the code continues its execution not in the current thread, but in the current synchronization context. This can be a problem if you subscribe to an event using StartReceiving and then call WaitFor, which will cause the class code to be executed simultaneously in different threads (the event handler from StartReceiving and the code after await // how scary to live!). This can be easily fixed with a single-threaded synchronization context included in the library:
this.serverSynchronizationContext = new SingleThreadSynchronizationContext("Server thread");
this.clientSynchronizationContext = new SingleThreadSynchronizationContext("Client thread");
this.serverSynchronizationContext.Post(async o => await this.RunServer(), null);
this.clientSynchronizationContext.Post(async o => await this.RunClient(), null);
Thus, the client will always be executed in the “Client thread” thread, and the server in the “Server thread”. You can write multi-threaded code without thinking about race condition. As a bonus, this will maximize the utilization of a single stream.
What is the advantage?
The main advantage is the simplicity and testability of the code. If you can argue about the first, everyone understands simplicity in his own way, then with the second paragraph everything is obvious. A multi-threaded application can be tested in one thread, emulating any sequence of events, and for this it is not necessary to create mock objects, any interaction can be reduced to events, and their verification to a RaiseEvent call. NUnit example:
///
/// This test demonstrates how to test application that uses Eventing
/// All code executes sequently in one thread
///
[TestFixture]
public class TestCase : TestBase {
[Test]
public async Task ClientDoesWork() {
var client = new Client(this.EventManager);
var doWorkAwaitable = client.DoWork();
this.EventManager.RaiseEvent(new Connected());
// We can debug and step into
this.React();
await doWorkAwaitable;
Assert.AreEqual(true, client.Connected);
}
}
How can this be used?
In order not to overwhelm the article with listings, I will provide only a short textual description of one of the systems where Eventing is used. This is a horizontally scalable distributed system consisting of four types of nodes, one of which is a master. The master continuously communicates with all nodes and controls the execution of various operations on them. Each operation can be represented in the form of a finite state machine, where the transition is the occurrence of an event (including a timeout or cancellation). Although for each operation it was possible to implement the automaton in its classical form (which we initially did), it turned out to be much simpler to represent it using Eventing, where the current state was determined by the code execution point, rather than a separate variable. Moreover, at each step all expected events were explicitly listed, which simplified the testing of the white box.
Conclusion
The article discusses the key features and options for using the Eventing library. The library does not pretend to be universal and support highly loaded systems, but calls for a slightly different look at familiar things, and allows you to write code that is safe and easily tested from the point of view of multithreading.
