Многопоточность в Photon

О чём статья

В этой статье мы поговорим о многопоточности в серверной части.

• как реализована

• как используется

• что можно сделать

• что мы сами изобрели

Все эти вопросы актуальны только если вы разрабатываете что-то непосредственно для серверной части - модифицируете код SDK, пишите свой плагин или вообще делаете что-то своё.

Как в Photon решается вопрос с многопоточностью?

Серверное приложение на фотоне принимает запросы от множества клиентских соединений. Буду называть такие соединения пирами. Эти запросы образуют очереди. По одной на каждый пир. Если пиры подключены к одной комнате, их очереди объединяются в одну - очередь комнаты. Таких комнат набирается до нескольких тысяч и их очереди запросов обрабатываются тоже параллельно.

В качестве основы для реализации очередей задачи в Photon была взята библиотека retlang, которая была разработана на базе библиотеки Jetlang.

Почему не используем Task и async/await

Поэтому поводу есть следующие соображения:

1. Photon начали разрабатывать до появления этих штук

2. Количество задач, которые выполняются файберами, огромно - десятки тысяч в секунду. Поэтому добавлять ещё одну абстракцию, которая, как мне кажется, ещё и GC нагружает, не было смысла. Абстракция файберов гораздо тоньше, если так можно выразится.

3. Наверняка есть TaskScheduler, который делает тоже самое что и файберы и я про него узнал бы в комментах, но в общем-то переизобретать велосипед не хотелось и не хочется.

Что такое Fiber?

Файбер это класс, который реализует очередь команд. Команды ставятся в очередь и исполняются одна за другой - FIFO. Можно сказать, что тут реализован шаблон multiple writers-single reader. Я ещё раз хочу обратить внимание на то, что команды исполняются в той последовательности, в которой поступили, т.е. одна за другой. На этом основывается безопасность доступа к данным в многопоточной среде.

Хотя в Photon мы используем только один файбер, а именно PoolFiber, библиотека предоставляет их пять. Все они реализуют интерфейс IFiber. Вот коротко о них.

• ThreadFiber - это IFiber, опирающийся на выделенный поток. Используется для частых и чувствительных к быстродействию операций.

• PoolFiber - это IFiber, опирающийся на пул потоков .NET. Выполнение всё равно происходит последовательно и только в одном потоке за раз. Используйте его для нечастых и менее чувствительных к производительности операций. Или когда желательно не увеличивать количество потоков (Наш случай).

• FormFiber/DispatchFiber - это IFiber, опирающийся на механизм сообщений WinForms/WPF. FormFiber/DispatchFiber полностью удаляют необходимость в вызове Invoke или BeginInvoke чтобы коммуницировать с окном из другого потока.

• StubFiber - очень полезен для детерминированного тестирования. Предоставляется точный контроль, чтобы сделать тестирование опережений (races) простым. Исполнение всех задач происходит в вызывающем потоке.

Про PoolFiber

Раскрою тему про выполнение задач вы PoolFiber. Хоть он и использует пул потоков, задачи в нём всё равно выполняются последовательно и используется только один поток за раз. Работает это так:

1. мы ставим в файбер задачу и она начинает исполнятся. Для этого вызывается ThreadPool.QueueUserWorkItem. И в какой-то момент выбирается один поток из пула и он выполняет эту задачу.

2. Если пока первая задача выполнялась мы поставили ещё несколько задач, то по окончании выполнения первой задачи, все новый забираются из очереди и снова вызывается ThreadPool.QueueUserWorkItem, чтобы все эти задачи отправились на исполнение. Для них будет выбран новый поток из пула. И когда он закончит, если в очереди есть задачи всё повторяется с начала.

Т.е. при том, что каждый раз новый пакет задач выполняет новый поток из пула, в каждый момент времени он один. Поэтому, если все задачи по работе с игровой комнатой ставятся в её файбер, из них(задач) можно безопасно обращаться к данным комнаты. Если к какому-то объекту обращаются из задач, выполняющихся в разных файберах, то тогда обязательно нужна синхронизация.

Почему PoolFiber

В Photon повсеместно используются PoolFiber. В первую очередь как раз потому, что он не создаёт дополнительных потоков и своим файбером может обладать любой кому это нужно. Мы его, кстати, немного модифицировали и теперь его нельзя остановить. Т.е. PoolFiber.Stop не остановит исполнение текущих задач. Для нас это было важно.

Ставить задачи в файбер можно из какого угодно потока. Всё это потоко-безопасно. Задача, которая исполняется в текущий момент, тоже может ставить новые задачи в файбер, в котором она исполняется.

Поставить задачу в файбер можно тремя способами:

• поставить задачу в очередь

• поставить задачу в очередь, которая будет выполнена через некоторый интервал

• поставить задачу в очередь, которая будет выполняться регулярно.

На уровне кода это выглядит примерно так:

// поставили задачу в очередь
fiber.Enqueue(()=>{some action code;});

// поставили задачу в очередь, чтобы выполнилась через 10 секунд
var scheduledAction = fiber.Schedule(()=>{some action code;}, 10_000);
...
// останавливаем таймер
scheduledAction.Dispose()

// поставили задачу в очередь, чтобы выполнилась через 10 секунд и каждые 5
var scheduledAction = fiber.Schedule(()=>{some action code;}, 10_000, 5_000);
...
// останавливаем таймер
scheduledAction.Dispose()

Для задач, которые выполняются через какой-то интервал важно сохранить ссылку, которую вернул fiber.Schedule. Это единственный способ остановить выполнение такой задачи.

Executors

Теперь про экзекуторы. Это классы, которые собственно выполняют задачи. Они реализуют методы Execute(Action a) и Execute(List<Action> a). PoolFiber использует второй. Т.е. задачи пачкой попадают в экзекутор. Что с ними дальше происходит зависит от экзекутора. Поначалу мы использовали класс DefaultExecutor. Всё что он делает это:

        public void Execute(List<Action> toExecute)
        {
            foreach (var action in toExecute)
            {
                Execute(action);
            }
        }

        public void Execute(Action toExecute)
        {
            if (_running)
            {
                toExecute();
            }
        }

В реальной жизни этого оказалось недостаточно. Потому что в случае исключения в одном из 'action' все остальные из списка toExecute пропускались. Поэтому по умолчанию сейчас используется FailSafeBatchExecutor, который внутрь цикла добавляет ещё try/catch. Мы рекомендуем использовать именно этот экзекутор, если не нужно ничего особенного. Этот экзекутор мы добавили сами, поэтому его нет в тех версиях, которые можно найти например на github.

Что ещё мы сами изобрели

BeforeAfterExecutor

Позднее мы добавили ещё один экзекутор, чтобы решить наши задачи с логгированием. Называется он BeforeAfterExecutor. Он "обёртывает" переданный эму экзекутор. Если ничего не передали, то создаётся FailSafeBatchExecutor. Особенностью BeforeAfterExecutor является способность выполнять экшен перед выполнением списка задач и ещё один экшен после выполнения списка задач. Конструктор выглядит следующим образом:

public BeforeAfterExecutor(Action beforeExecute, Action afterExecute, IExecutor executor = null)

Для чего это используется. Файбер и экзекутор имеют одного владельца. При создании экзекутора ему передаётся два экшена. Первый добавляет пары ключ/значение в контекст потока, а второй удаляет их, тем самым выполняя функцию уборщика. Добавленные в контекст потока пары добавляются системой логирования к сообщениям и мы можем видеть некоторые мета данные того, кто сообщение оставил.

Пример:

var beforeAction = ()=>
{
  log4net.ThreadContext.Properties["Meta1"] = "value";
};

var afterAction = () => ThreadContext.Properties.Clear();

//создаём экзекутор
var e = new BeforeAfterExecutor(beforeAction, afterAction);

//создаём PoolFiber
var fiber = new PoolFiber(e);

Теперь если что-то логгируется из задачи, которая исполняется в fiber, log4net добавит тэг Meta1 со значением value.

ExtendedPoolFiber и ExtendedFailSafeExecutor

Есть ещё одна штука, которой не было в оригинальной версии retlang, и которую мы разработали позже. Этому предшествовала следующая история. Делюсь ей, чтобы и другим неповадно было. Была следующая задача. Есть PoolFiber (это тот, что работает поверх пула потоков .NET). В задаче, которая выполняется этим файбером, нам было необходимо синхронно выполнить HTTP запрос. Сделали просто:

1. перед выполнением запроса создаём event;

2. в другой файбер отправляется задача, выполняющая запрос, и, по завершению, ставящая event в сигнальное положение;

3. после этого встаём ожидать event.

Не лучшее с точки зрения масштабируемости решение начало давать неожиданный сбой. Оказалось, что задача, который мы ставим в другой файбер на шаге два, попадает в очередь того самого потока, который встал ждать event. Таким образом получили дедлок. Не всегда. Но достаточно часто, чтобы обеспокоиться этим.

Решение было реализовано в ExtendedPoolFiber и ExtendedFailSafeExecutor. Придумали ставить весь файбер на паузу. В этом состоянии он может накапливать новые задачи в очереди, но не исполняет их. Для того, чтобы поставить файбер на паузу вызывается метод Pause. Как только он вызван файбер (а именно экзекутор файбера) ждёт пока текущая задача выполнится и замирает. Все остальные задачи будут ждать первого из двух событий:

1. Вызов метода Resume

2. Таймаута (указывается при вызове метода Pause) В метод Resume можно поставить ещё и задачу, которая будет выполнена перед всеми, стоявшими в очереди задачами.

Мы используем этот трюк, когда плагину надо загрузить состояние комнаты, используя HTTP запрос. Чтобы игроки увидели обновлённое состояние комнаты сразу же, файбер комнаты ставится на паузу. При вызове метода Resume мы ставим ему задачу, который применяет загруженное состояние и все остальные задачи уже работают с обновлённым состоянием комнаты.

Кстати, необходимость ставить файбер на паузу окончательно убила возможность использовать ThreadFiber для очереди задач игровых комнат.

IFiberAction

IFiberAction - это эксперимент по сокращению нагрузки на GC. Мы не можем управлять процессом создания экшенов в .NET. Поэтому было решено заменить стандартные экшены на экземпляры класса, который реализует интерфейс IFiberAction. Предполагается, что экземпляры таких классов достаются из пула объектов и возвращаются туда сразу же после завершения. Этим и достигается снижение нагрузки на GC

Интерфейс IFiberAction выглядит следующим образом:

public interface IFiberAction
{
    void Execute()
    void Return()
}

Метод Execute содержит собственно, то что нужно исполнить. Метод Return вызывается после Execute, когда пришло время вернуть объект в пул.

Пример:

public class PeerHandleRequestAction : IFiberAction
{
    public static readonly ObjectPool<PeerHandleRequestAction> Pool = initialization;
    public OperationRequest Request {get; set;}
    public PhotonPeer Peer {get; set;}
    
    public void Execute()
    {
        this.Peer.HandleRequest(this.Request);
    }
    
    public void Return()
    {
        this.Peer = null;
        this.Request = null;
        
        Pool.Return(this);
    }
}

//теперь использование будет выглядит примерно так

var action = PeerHandleRequestAction.Pool.Get();
action.Peer = peer;
action.Request = request;

peer.Fiber.Enqueue(action);

Заключение

В качестве заключения коротко резюмирую то, о чём рассказал. Для обспечения потокобезопастности в фотон мы используем очереди задач, которые в нашем случае представлены файберами. Основной вид файбера, который мы используем это PoolFiber и его наследники. PoolFiber реализует очередь задач поверх стандартного пула потоков .NET. В силу дешевизны PoolFiber своим файбером могут обладать все, кому это необходимо. Если необходимо ставить очередь задач на паузу, используйте ExtendedPoolFiber.

Непосредственным выполнением задач в файберах занимаются экзекуторы, реализующие интефейс IExecutor. DefaultExecutor всем хорош, но в случае исключения теряет весь остаток задач, которые были переданы ему на исполнение. FailSafeExecutor видится в этом отношении разумным выбором. Если надо выполнить какое-то действие перед выполнением экзекутором пачки задач и после него, может пригодится BeforeAfterExecutor