асинхронная сериализация - C/C++

Допустим у нас есть stream socket (например boost::asio::ip::tcp::socket) с асинхронными операциями чтения/записи, а-ля async_read/async_write.
По этому потоку нужно гонять туда-сюда древовидные структуры, то есть обыкновенная сериализация.
Эти структуры не trivially copyable (как минимум потому, что содержат массивы переменной длины) — следовательно zero-copy никак не получится.

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

Эти структуры могут занимать от пары кибибайт до нескольких мебибайт, а соединений — десятки тысяч.
Возможно несколько уменьшить масштаб проблемы распилив крупные корневые структуры на несколько маленьких.

Более экономичным был бы вариант с использованием небольшого буфера фиксированного размера, например 32KiB — и соответственно поэтапная сериализация. То есть прочитали один chunk и сразу его десериализовали (и vice versa для сериализации).
Также это избавило бы от необходимости искать место для буфера переменного размера.

Варианты которые позволили бы использовать фиксированный буфер:

Отдельный поток для сериализации. Это, естественно, совсем не интересно.

Stackful Coroutines — на мой взгляд самый лучший вариант. Да и к тому же они есть не просто в Boost, а даже прям в Boost.Asio.

Описание схемы сериализации в специальном формате, например через expression templates (а-ля Boost.Proto), через гетерогенные последовательности (а-ля Boost.Fusion, тем более сейчас через него всё и сериализуется), либо через препроцессор (BOOST_PP_SEQ или X-Macro) — и последующий распил считывания кусочков данных на continuations.

Внешняя генерация необходимого кода из схемы — а-ля Apache Thrift или Protocol Buffers (кстати, а может там уже есть асинхронная сериализация?).

Собственно у меня два вопроса:
1. Какие есть ещё альтернативы? Возможно есть какие-то стандартные/распространённые решения?
2. Встречалась ли вам такая задача? В каком контексте? И как она была решена?

Спасибо.

Здравствуйте, Evgeny.Panasyuk, Вы писали:

EP>1. Какие есть ещё альтернативы?
да, после сегодняшнего коммита в YAS, у тебя есть возможность реализовать свои istream и ostream типы так, чтоб они напрямую читали/писали в сокет.

EP>2. Встречалась ли вам такая задача?
да, недавно только закончил с ее реализацией.
уже кратко упоминал о ней:

недели две назад закончил реализацию некоторого механизма зеркалирования и отката состояния группы игровых серверов.

>В каком контексте?
в контексте MMO игровых серверов =)

>И как она была решена?
так, как я описал выше.
(а в YAS я закоммитил это решение позже потому, что YAS все же является форком коммерческого проекта. по этому, этот функционал был сначала реализован в коммерческом прародителе.)

я, кажется понял, что ты имеешь ввиду упоминая корутины...
типа этого:

container type data;
oarchive oa;
for ( const auto &it: data ) {
   oa & it;
   yield();
}

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

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

Эти структуры могут занимать от пары кибибайт до нескольких мебибайт, а соединений — десятки тысяч.

Пока шёл на тренировку пришёл в голову ещё один вариант:

Использовать deque как буфер, а не vector. В этом случае chunk'и можно деаллоцировать по мере десериализации. Это немного уменьшает peak memory usage, но всё равно не супер айс.
Плюс, чанки от разных дэк можно хранить в одном freelist'е — аллокация/деаллокация будет копеечной.
Плюс, можно сделать оптимизацию в виде zero-copy в некоторых местах — если в какой-то момент у нас идут POD'ы, то чанки можно splice'ить прямо в target дэку.

Здравствуйте, niXman, Вы писали:

X>собственно говоря, корутины для этого не нужны. если ты предполагаешь отправлять в сокет каждый такой элемент, то следующий ты можешь брать в хендлере операции асинхронной записи.
fixed
X>но тут мне не понятно, как быть в случае, когда ты сериализуешь свое дерево, а во время охидяния хендлера происходит изменение этого дерева? или на момент когда ты начинаешь сериализовать это дерево, оно не будет изменяться до тех пор, пока сериализация не завершится? но даже в этом случае я не очень понимаю, для чего корутины

Последовательная асинхронная сериализация, такты которой запускаются хендлерами из async_write(in_blum.. , с оглядкой на счётчики и флаги,
которые часть специализированного класса, и которые отрабатывают в отдельном потоке или даже процессе, запускаемом из хендлера async_accept?
Тут можно интересный конечный автомат наваять

Здравствуйте, niXman, Вы писали:

EP>>1. Какие есть ещё альтернативы?
X>да, после сегодняшнего коммита в YAS, у тебя есть возможность реализовать свои istream и ostream типы так, чтоб они напрямую читали/писали в сокет.

Асинхронно буферами фиксированного размера?

EP>>2. Встречалась ли вам такая задача?
X>да, недавно только закончил с ее реализацией.
X>уже кратко упоминал о ней:
X>

X>недели две назад закончил реализацию некоторого механизма зеркалирования и отката состояния группы игровых серверов.

Да, я видел — интересно. Это для fault tolerance? (никогда профессионально не занимался сетью)

Здравствуйте, niXman, Вы писали:

X>я, кажется понял, что ты имеешь ввиду упоминая корутины...
X>типа этого:
X>

X>container type data;
X>oarchive oa;
X>for ( const auto &it: data ) {
X>   oa & it;
X>   yield();
X>}
X>

X>?

Да, примерно. yield можно спрятать в архив — пользовательский код и не будет знать о том, что его прерывают.
То есть пользователь будет просто видеть/писать сериализацию в стиле Boost.Serialization-like.

Здравствуйте, Evgeny.Panasyuk, Вы писали:

EP>Асинхронно буферами фиксированного размера?
ну во-первых — это как тебе угодно. все что тебе потребуется, это реализовать свой ostream класс, с единственным необходимым YAS`у методом: 'std::size_t write(const void *ptr, const std::size_t size)'.
во-вторых — чтоб говорить о буферах/преаллокациях/пуле_буфферов — нужно знать скорость поступления данных, и скорость их выгрузки. это и определит стратегию.

EP>Это для fault tolerance?
да.

Здравствуйте, niXman, Вы писали:

X>собственно говоря, корутины для этого не нужны. если ты предполагаешь отправлять в сокет каждый такой элемент, то следующий ты можешь брать в хендлере операции асинхронной записи.

Собственно это то, как оно будет реализовано на нижнем уровне — скидываем один буфер, в хэндлере async_write — запись следующего и так далее.
Вопрос в том, как это сделать в удобном виде, а не в рукопашку раскладывать сериализацию каждой структуры на хэндлеры-continutaions.
Например, у нас есть:

template<typename Archive>
void serialize(Archive &ar, Widget &x)
{
    ar & x.a;
    ar & x.b;
    ar & x.c;
    ar & x.d;
}

плюс есть буфер фиксированного размера:

array<char, 1024> fixed_buffer;

этот буфер будет передаваться по сети через async_read/async_write.

Весь Widget больше чем буфер, и за один присест его не получится отправить/получить через fixed_buffer.
Задача сделать сериализацию по кускам, но не меняя код шаблона функции serialize. На stackful coroutines это делается элементарно.
Меня же интересует — существуют ли какие-нибудь другие удобные способы?

X>но тут мне не понятно, как быть в случае, когда ты сериализуешь свое дерево, а во время охидяния хендлера происходит изменение этого дерева? или на момент когда ты начинаешь сериализовать это дерево, оно не будет изменяться до тех пор, пока сериализация не завершится?

Нет, изменение дерева не происходит. Точнее если потребуется, то при записи можно освобождать уже записанные части (делать shrink_to_fit для уже записанных массивов и т.п.)

Здравствуйте, Evgeny.Panasyuk, Вы писали:

EP>Спасибо.

могу предложить использовать любую поточную сериализацию (YAS, protbuf), подставив декоратор, который нарезает на чанки и асинхронно пишет в сокет_стрим
тут только надо продумать два момента:
1) обработка ошибок при сериализации (часть данных отправили, остальное не отправим)
2) получателю придется все это склеивать и вряд ли получится поточно десериализовать

для чанков можно использовать спец аллокаторы (в студии >= 12 кстати классные аллокаторы поставляются: http://msdn.microsoft.com/en-us/library/ee292134.aspx )

EP>Внешняя генерация необходимого кода из схемы

считаю это наиболее удобным (по сравнению c препроцессором) способом создания RPC/serialization кода для C++
так я делал в коммерческом коде, и так делается в MS idl, google protobuf

Здравствуйте, niXman, Вы писали:

EP>>Асинхронно буферами фиксированного размера?
X>ну во-первых — это как тебе угодно. все что тебе потребуется, это реализовать свой ostream класс, с единственным необходимым YAS`у методом: 'std::size_t write(const void *ptr, const std::size_t size)'.

В этом конкретном случае — формат структур уже специфицирован, причём он довольно простой. Никаких особых фич как в Boost.Serialization — например отслеживание указателей, сериализация полиморфных типов — не требуется.
Сейчас всё ядро сериализации — это in/out archive (которые пришлось бы писать в любом случае) + рекурсивная сериализация Boost.Fusion Forward Sequence. На всё про всё — меньше ста строк.
Структуры определяются просто как BOOST_FUSION_DEFINE_STRUCT и уже готовы к сериализации.

Подменив архив на тот, который вызывает async_write/async_read, и передаёт yield как хэнделер — как раз и получится асинхронная сериализация. Вопрос же в том — есть ли другие способы.

Здравствуйте, Evgeny.Panasyuk, Вы писали:

EP>Собственно это то, как оно будет реализовано на нижнем уровне — скидываем один буфер, в хэндлере async_write — запись следующего и так далее.
EP>Вопрос в том, как это сделать в удобном виде, а не в рукопашку раскладывать сериализацию каждой структуры на хэндлеры-continutaions.
как вариант — написать обертку над yas::*_oarchive:

struct my_oarchive {
   template<typename T>
   my_oarchive& operator& (const T &o) {
      yas::*_oarchive::operator& (o);
      yield();
      return *this;
   }
};

EP>Меня же интересует — существуют ли какие-нибудь другие удобные способы?
мне такие не известны. если найдешь — расскажи плиз.

EP>Нет, изменение дерева не происходит.
так если у тебя неизменяемый объект — то что мешает таки реализовать свой ostream класс, и в нем, когда его внутренний буфер достигает предельного размера — отсылать собранные данные, и снова заполнять буфер следующими данными?
чтоб не было бауз в использовании сети — завести два буфера: первый отдаешь на отправку, второй — для сериализации в него. в хендлере отправки данных переключаешь буфера — второй на отправку, а первый для сериализации.

Здравствуйте, uzhas, Вы писали:

EP>>Спасибо.
U>могу предложить использовать любую поточную сериализацию (YAS, protbuf), подставив декоратор, который нарезает на чанки

Нарезать на чанки как раз не проблема. Как только заполнился буфер — сразу пишем.

U>и асинхронно пишет в сокет_стрим

Основной фокус как раз тут.
Вот вызвали async_write для первого чанка. Следующий async_write должен происходить из хэндлера записи.
А пока пишется — мы просто останавливаем сериализацию и отдаём управление io_service'у (thread-то всего один). Но — нам нужно не потерять callstack чтобы продолжить запись с правильной позиции в структуре.
Один из вариантов решения — это stackful coroutines, которые как раз сохраняют наш callstack до ~~лучших времён~~ вызова хэндлера.

Здравствуйте, niXman, Вы писали:

X>struct my_oarchive {
тут таки наследуем yas::*_oarchive

X>чтоб не было *бауз в использовании сети
пауз

Здравствуйте, Evgeny.Panasyuk, Вы

EP>Основной фокус как раз тут.
EP>Вот вызвали async_write для первого чанка. Следующий async_write должен происходить из хэндлера записи.

Почему должен? Я предлагаю делать полную сериализацию в одном потоке, а запись в сокет делать чанками асинхронно: async_write всегда вызывается из потока сериализации

Здравствуйте, niXman, Вы писали:

X>как вариант — написать обертку над yas::*_oarchive:
X>

X>struct my_oarchive {
X>   template<typename T>
X>   my_oarchive& operator& (const T &o) {
X>      yas::*_oarchive::operator& (o);
X>      yield();
X>      return *this;
X>   }
X>};
X>

Ну да — я это и имел в виду в первом сообщении. (ранее я показывал пример чего-то подобного, там даже ручной демультиплексор для наглядности).

EP>>Меня же интересует — существуют ли какие-нибудь другие удобные способы?
X>мне такие не известны. если найдешь — расскажи плиз.

Ок, хорошо.

EP>>Нет, изменение дерева не происходит.
X>так если у тебя неизменяемый объект — то что мешает таки реализовать свой ostream класс, и в нем, когда его внутренний буфер достигает предельного размера — отсылать собранные данные, и снова заполнять буфер следующими данными?
X>чтоб не было бауз в использовании сети — завести два буфера: первый отдаешь на отправку, второй — для сериализации в него. в хендлере отправки данных переключаешь буфера — второй на отправку, а первый для сериализации.

А как отдать управление io_service'у после первой отсылки? Ок, допустим мы как-то решили эту задачу. Но сериализация в большинстве случаев будет на порядки быстрее отправки/получения чанков.
Отправив первый чанк — мы заполним второй быстрее чем сообщение будет отправлено, и теперь у нас уже два неотправленных буфера. То есть мы вернулись к тому, с чего начали.

Здравствуйте, Evgeny.Panasyuk, Вы писали:

EP>Отправив первый чанк — мы заполним второй быстрее чем сообщение будет отправлено, и теперь у нас уже два неотправленных буфера. То есть мы вернулись к тому, с чего начали.

именно, поэтому может стоит рассмотреть как альтернативу модель реактора вместо проактора?
http://www.boost.org/doc/libs/1_55_0b1/doc/html/boost_asio/overview/core/reactor.html

Здравствуйте, uzhas, Вы писали:

EP>>Основной фокус как раз тут.
EP>>Вот вызвали async_write для первого чанка. Следующий async_write должен происходить из хэндлера записи.

U>Почему должен?

Во-первых, порядок двух асинхронных операций не определён. Во-вторых, async_write не атомарен, и сам состоит из нескольких операций типа async_write_some, которые могут перемешиваться с async_write_some от другого async_write.
Но даже если у нас есть строгий порядок асинхронных записей (допустим у нас есть очередь (переменного размера, что уже не хорошо)) — то всё равно не понятно как это будет работать:

async_write(buf1, ...);
async_write(buf2, ...);
async_write(buf3, ...);

В момент вызова каждого async_write у нас уже должен быть готов соответствующий буфер, чего мы изначально и пытались избежать.

Или для наглядности рассмотрим чтение — async_read. Сразу после запуска первого чтения мы не можем запустить второе, так как мы ещё первый кусок данных не распрасили.

U>Я предлагаю делать полную сериализацию в одном потоке, а запись в сокет делать чанками асинхронно: async_write всегда вызывается из потока сериализации

Поток в смысле отдельный thread для сериализации, который блокируется на записи/чтении? То есть вариант #1 из первого сообщения?
Это не подходит — у нас по условию задачи десятки тысяч соединений, плюс оверхед на межпоточную синхронизацию совсем не нужен.

Здравствуйте, antropolog, Вы писали:

EP>>Отправив первый чанк — мы заполним второй быстрее чем сообщение будет отправлено, и теперь у нас уже два неотправленных буфера. То есть мы вернулись к тому, с чего начали.
A>именно, поэтому может стоит рассмотреть как альтернативу модель реактора вместо проактора?
A>http://www.boost.org/doc/libs/1_55_0b1/doc/html/boost_asio/overview/core/reactor.html

Непонятно чем это лучше проактора для данной задачи.
По сути точно такая же неблокирующая операция, которая выполнится в будущем — нам точно также нужно сохранять текущий callstack и отдавать управление.

	От:	Evgeny.Panasyuk
	Дата:	27.12.13 16:35
	Оценка:

От:	niXman	https://github.com/niXman
Дата:	27.12.13 16:49
Оценка:	12 (1)

От:	niXman	https://github.com/niXman
Дата:	27.12.13 17:41
Оценка:

От:	niXman	https://github.com/niXman
Дата:	27.12.13 17:47
Оценка:

	От:	Evgeny.Panasyuk
	Дата:	27.12.13 19:23
	Оценка: