concurrency

Идея добавить в 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 затем его обработает, когда сможет, безопасно перенаправив сообщение логгеру. Недостатком этого является невозможность вариативного вызова с несколькими параметрами — можно передать только одну строку, но это лучше, чем ничего.