Идея добавить в EventManager поддержку конкурентности/асинхронности не давала мне покоя еще со времен DGL, и вот, наконец, мне это удалось. Точнее, асинхронный слой удобно лег поверх стандартной событийной шины, и для этого не пришлось менять в ней практически ничего.
Перед тем, как перейти к деталям, объясню в двух словах, как работает менеджер событий, и чем проблематична его синхронная природа. Менеджер событий служит, главным образом, прослойкой между SDL и объектной структурой приложения. SDL предполагает единственного слушателя, который опрашивает события через SDL_PollEvent — менеджер событий Dagon агрегирует события SDL в собственную очередь, позволяя неограниченному множеству других объектов-слушателей (наследующих от EventListener) опрашивать события и реагировать на них путем динамической диспетчеризации.
Особенность в том, что все это делается синхронно — то есть, очередь событий EventManager живет в течение одной итерации игрового цикла, и все слушатели должны синхронно обрабатывать события в течение той же итерации. Это на самом деле неплохо, если методы-обработчики событий имеют реалтаймовую производительность — в основной игровой логике так оно и бывает, время отклика состояния на события ввода должно быть пренебрежимо мало. Но что, если вам нужно выполнить что-то относительно тяжелое? Например, сделать сетевой запрос или сохранить данные в файл? В этом случае поток может быть заблокирован, рендеринг застынет, и приложение на какое-то время перестанет реагировать на ввод, что не очень приятно для пользователя — особенно если это происходит посреди игрового процесса.
На помощь приходит многопоточность (которую еще называют конкурентностью, если речь идет о разделении времени, а не о строго параллельном выполнении потоков, как на GPU). Если выполнять блокирующие задачи в отдельном потоке, то игра не будет лагать, пользователь ничего не заметит.
Потоки создавать легко, и они полностью решают проблему синхронности, но они приносят с собой другую — resource contention (конкуренцию за ресурс). Это явление, когда два и более потоков могут одновременно читать и записывать одни и те же данные в памяти или аналогично использовать тот или иной общий ресурс (файл, какой-либо интерфейс и т.д.). Это приводит к состоянию гонки, когда результат программы зависит от непредсказуемого порядка выполнения операций различными потоками. А без этого потоки почти бесполезны, если только вы не решаете идеально распараллеливаемую задачу, когда каждый поток читает неизменяемые данные и пишет в свой приватный участок памяти. Игровую логику распараллелить непросто — состояние в играх, как правило, общее, его не поделишь. Для решения этой проблемы есть системные блокировки, мьютексы, которые позволяют «приватизировать» чтение и запись в определенном участке кода для одного потока, заблокировав остальные, но это может приводить к другим трудно решаемым проблемам вроде взаимных блокировок. К тому же мьютексы не бесплатны, и оверхед на блокировки, если ими злоупотреблять, может запросто перекрыть выгоду от использования потоков. Многопоточность — это настоящее минное поле!
К счастью, производители процессоров и тут предусмотрели решение — атомарные операции. Это такие операции, которые гарантированно выполняются целиком, не прерываясь другими потоками. Атомарное чтение гарантирует, что переменная не будет перезаписываться, пока ее читают. Атомарная запись — что другие потоки не будут читать переменную, пока она записывается. На x86 многие операции атомарны, такие, как 32-битное присваивание, но в общем случае на это полагаться нельзя, так как атомарность почти никогда не гарантируется на уровне ЯП — нужно использовать специальные интринсики, которые транслируются в атомарные инструкции. В D они предоставляются модулем core.atomic.
При помощи атомарных операций можно реализовать потоково-безопасные структуры данных, которые позволяют читать и писать из разных потоков, не боясь состояний гонки. Эти структуры, в свою очередь, лежат в основе конкурентного программирования.
В Dagon я реализовал межпоточную систему обмена сообщениями на основе SPSC-очередей. Фишка в том, что если очередь на кольцевом буфере всегда читается одним потоком и записывается другим (не больше!), она реализуется на атомиках как lock-free и wait-free алгоритм — то есть, операции над ней не блокируют поток и завершаются за детерминированное время, что идеально для реалтайма. Процесс аналогичен работе почтовой службы. Поток — вернее, объект Service, работающий в своем потоке — имеет две очереди, похожие на два почтовых ящика: для входящих сообщений (inbox) и исходящих (outbox). Когда сервис хочет отправить сообщение, он добавляет новый элемент в свою outbox-очередь. Специальный брокер в главном потоке, как почтальон, забирает сообщение и переправляет его в inbox-очередь сервиса-адресата. Тот, в свою очередь, поллит (то есть, опрашивает) свой inbox и реагирует на входящие сообщения.
Особенность моей реализации — она «дружит» с EventManager, в том смысле, что события из стандартной шины уходят в брокер, а сообщения от сервисов, если они адресованы синхронным компонентам приложения, приходят из бокера в стандартную шину. Помимо сообщений, поддерживаются задачи — можно асинхронно делегировать сервисам любой метод путем создания события типа EventType.Task. Брокер в этом случае выступает планировщиком, выбирая один из сервисов-воркеров для выполнения задачи по принципу round-robin (то есть, всем по кругу). Сервисы же могут выполнить свой метод синхронно, выставив задачу стандартной шине.
Адресация осуществляется при помощи строковых имен — любой синхронный EventListener и асинхронный сервис может иметь уникальный (или неуникальный) адрес для получения сообщений. Маршрутизация (от шины к брокеру и обратно) осуществляется при помощи целочисленной метки, которую я назвал доменом (domain). Если событие имеет отрицательный домен, то оно предназначено для сервиса, если положительный — для EventListener. Благодаря такому разделению системе легко разобраться, что куда отправлять, брокер при отправке может отсеивать определенные события, чтобы не создавать feedback loop. Использование знака домена лежит в основе математической логики маршрутизации — если произведение доменов положительно, то события идут в одном направлении, и наоборот. Нулевой домен трактуется как циркуляр, то есть, событие, предназначенное для всех систем. События ввода всегда являются циркулярами, сообщения могут ими быть, задачи — не могут (то есть, они идут либо в шину, либо в брокер).
При этом числа доменов могут быть любыми, их смысл остается на усмотрение разработчика приложения. Можно, например, делить сообщения на классы или придумать какие-то другие применения.
Примеры
Простой сервис, который логирует все входящие сообщения и отвечает одним и тем же текстом:
class Responder: Service
{
this(string address, MessageBroker broker, Owner owner)
{
super(address, broker, owner);
}
override void onMessage(
uint domain, string sender, string message, void* payload)
{
queueLog(LogLevel.Info, message);
send(sender, "Hi!", null, MessageDomain.MainThread);
}
}
class TestScene: Scene
{
Responder responder;
override void afterLoad()
{
// важно для получения сообщений!
address = "Scene";
responder =
New!Responder("Responder", eventManager.messageBroker, this);
responder.run();
// брокер по умолчанию выключен
eventManager.messageBroker.enabled = true;
}
override void onKeyDown(int key)
{
send("Responder", "Hello, World!");
}
override void onMessage(
int domain, string sender, string message, void* payload)
{
logInfo("Scene received from ", sender, ": ", message);
}
}
Пример из реальной жизни — простой сетевой клиент, который умеет только отправлять запросы через блокирующий сокет. Сообщения одному и тому же сервису обрабатываются по очереди, так что конфликт за порт исключен.
import std.socket: TcpSocket, InternetAddress;
class NetworkClient: Service
{
string host;
ushort port;
this(string address, MessageBroker broker, Owner owner,
string host, ushort port)
{
super(address, broker, owner);
this.host = host;
this.port = port;
}
override void onMessage(
int domain, string sender, string message, void* payload)
{
auto soc = new TcpSocket();
try
{
soc.connect(new InternetAddress(host, port));
soc.send(message);
soc.close();
}
catch(Exception e)
{
queueLog(LogLevel.Warning, e.msg);
}
}
}
Воркеры — сервисы, специально предназначенные для выполнения задач:
class TestScene: Scene
{
Worker worker1, worker2;
override void afterLoad()
{
address = "Scene";
worker1 = New!Worker("Worker1", eventManager.messageBroker, this);
worker2 = New!Worker("Worker2", eventManager.messageBroker, this);
worker1.run();
worker2.run();
eventManager.messageBroker.enabled = true;
}
void doSomethingThatBlocks(Object obj, void* payload)
{
queueLog(LogLevel.Info, "doSomethingThatBlocks");
}
override void onKeyDown(int key)
{
queueTask(&doSomethingThatBlocks);
}
}
В этих примерах используется queueLog вместо обычных функций log*, потому что логгер потоко-небезопасен — для асинхронного логирования я еще несколько релизов назад предусмотрел альтернативный механизм. Вместо прямого логирования следует добавлять событие лога в очередь, и EventManager затем его обработает, когда сможет, безопасно перенаправив сообщение логгеру. Недостатком этого является невозможность вариативного вызова с несколькими параметрами — можно передать только одну строку, но это лучше, чем ничего.
