[prog.thougths] Попытка спроектировать механизм взаимодействия процессов на одной ноде через shared memory по принципу single-producer/multi-consumers

С конца прошлой недели занимаюсь проектированием механизма, посредством которого процесс-producer смог бы передавать через shared memory информацию одному или нескольким процессам-consumer-ам. Вроде бы придумалось какое-то решение, которое выглядит рабочим. Данный пост является попыткой еще одной проверки этого решения: если получится упомянутый механизм внятно описать словами, то есть хорошая вероятность, что он будет воплощен в коде.

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

В общем, всех заинтересовавшихся приглашаю под кат.

Типа постановка задачи

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

Есть процесс-producer, который производит некие объекты с информацией. Эта информация должна поступать процессам-consumer-ам. У одного producer-а может быть сразу несколько consumer-ов.

Темп генерации объектов -- не больше 50-60 в секунду на начальном этапе, возможно, затем будет 100-150 в секунду.

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

Собственно, shared memory рассматривается именно из-за размера передаваемых объектов. Как-то не хочется пропихивать сотни мегабайт данных между процессами, работающими на одной ноде, через IPC-механизмы типа сокетов или пайпов.

Обязательные граничные условия

Есть важные особенности, которые оказывают серьезное влияние на итоговое решение.

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

Во-вторых, нет требований надежной доставки объектов. Т.е. если объект опубликовали, но информация об этом не дошла до consumer-ов по каким-то причинам (например, consumer-ов еще нет), то это не страшно. Объект будет просто потерян. Такие потери не есть хорошо, но и страшного в этом нет, т.к. producer все равно постоянно генерирует новые объекты с более актуальной информацией.

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

Так что одно из главных требований к обсуждаемому механизму -- это способность пережить падение+рестарт или зависание любого из процессов (будь то producer или consumer).

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

Есть контроль за жизнеспособностью producer-а и consumer-ов

Процессы producer и consumer работают под присмотром супервизора, который их запускает и затем следит за их жизнеспособностью. Так что если какой-то из этих процессов аварийно завершит работу, то супервизор перезапустит упавший процесс.

Предполагается, что супервизор будет иметь какие-то средства для проверки того, что процесс не завис и работает нормально (посредством каких-то прикладных ping-ов). Так что есть надежда, что супервизор сможет перезапускать и зависшие процессы (например, вошедшие в бесконечный цикл или поймавшие дедлок).

Предполагаемый механизм обмена данными

Описание разбито на две части.

В первой части описывается принцип работы обсуждаемого механизма. Этот механизм нуждается в каком-то способе обмена некими управляющими сообщениями между процессами. И этот способ не shared memory, а что-то из категории сокетов, именованных каналов (named pipes) или POSIX message queues. Что именно из IPC будет использоваться не оказывает непосредственного влияния на принцип работы самого механизма, поэтому в первой части описания просто предполагается, что такой способ есть.

Во второй части обсуждается, как именно может быть реализован обмен сообщениями между процессами с использованием имеющихся в Linux/Windows механизмов IPC.

Принцип работы механизма обмена данными через shared memory

Предпосылки

У процесса-producer-а есть некая точка входа с фиксированным именем, через которую процессы-consumer-ы могут присылать producer-у свои сообщения. Для того, чтобы consumer мог получать данные от producer-а, consumer должен знать имя точки входа. Каким именно способом consumer получит это имя мы не рассматриваем (через параметры командной строки, из конфигурационного файла, из переменных окружения и т.д.).

У каждого consumer-а есть некий собственный идентификатор. Что-то вроде короткого имени. У всех consumer-ов, которые хотят получать информацию от общего producer-а, должны быть уникальные (не совпадающие) идентификаторы. В противном случае producer не сможет нормально взаимодействовать с consumer-ами. Каким именно способом обеспечивается уникальность идентификаторов consumer-ов мы не рассматриваем, считаем, что это свойство тем или иным способом обеспечивается.

Начальные действия producer-а при старте (после рестарта)

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

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

Подключение producer-а к consumer-у

Чтобы получать информацию от producer-а consumer должен подключиться к producer-у. Для этого consumer выполняет следующие действия:

1. Создает для себя новую точку входа с уникальным именем;
2. Отсылает на точку входа producer-а сообщение attach_request, в котором передает свой идентификатор и имя только что созданной точки входа;
3. Producer отсылает на точку входа consumer-а ответное сообщение attach_ack, в котором передает имя файла, который consumer-у нужно отобразить в собственное адресное пространство. Так же в attach_ack сообщается session_id, который используется в последующем обмене сообщениями между producer-ом и consumer-ом;
4. Если consumer не получает ответа от producer-а за положенное время, то consumer возвращается к шагу №1;
5. Если consumer вовремя получил от producer-а attach_ack, то consumer открывает файл с именем, полученным от producer-а, и осуществляет отображение содержимого этого файла в собственное адресное пространство. Если при этом возникает какая-то ошибка, то consumer возвращается к шагу №1.

Процедура публикации новых объектов producer-ом

Публикация объекта выполняется в две стадии.

На первой стадии внутри producer-а кто-то запрашивает блок памяти некоторого размера. Producer пытается найти область подходящего размера внутри арены. Если такая область есть, то она оформляется в виде нового блока, который находится в статусе "выделен-но-не-опубликован".

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

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

На второй стадии producer отсылает всем подключенным consumer-ам сообщение publish_announce. Сообщение publish_announce отсылается в ту точку входа, которую consumer сообщил при подключении. Если запись publish_announce для какого-то consumer-а завершается неудачно, то producer считает этого consumer-а неработоспособным и предпринимает соответствующие действия (описываются ниже).

Каждый consumer должен в течении заданного времени ответить сообщением announce_ack. Всем consumer-ам, которые успели это сделать, отсылается сообщение publish, в котором содержится информация об опубликованном блоке.

Если consumer не успел прислать announce_ack, то сообщение publish такому consumer-у не отсылается.

По результатам отсылки publish-ей для опубликованного объекта выставляется значение счетчика ссылок: счетчик ссылок для опубликованного объекта равен количеству отосланных publish-е (отосланных без ошибок).

Так, если есть пять consumer-ов, на publish_announce было вовремя получено три announce_ack и все три publish-а были отосланы успешно, то счетчик ссылок для опубликованного объекта будет равен трем.

После публикации статус блока памяти в арене, в котором находится опубликованный объект, меняется на "опубликован".

Процесс-producer для каждого consumer-а ведет список объектов, которые были опубликованы для этого consumer-а (т.е. списки объектов, для которых этому consumer-у были отосланы сообщения publish). Благодаря этому списку producer точно знает, какие объекты были опубликованы для конкретного consumer-а.

Подтверждать получение сообщения publish consumer-у не нужно.

Когда consumer заканчивает использовать опубликованный объект, он отсылает producer-у сообщение release, в котором указывается, какой именно объект освобождается.

Получив release producer:

• вычеркивает объект из списка опубликованных для данного consumer-а объектов;
• уменьшает счетчик ссылок для опубликованного объекта. Если счетчик ссылок обнуляется, то блок памяти, принадлежавший объекту, помечается как свободный и возвращается в арену для последующего переиспользования.

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

Поведение producer-а при отсутствии свободного места в текущей арене

Рано или поздно может возникнуть ситуация, когда consumer-ы будут освобождать опубликованные объекты медленнее, чем producer-у нужно создавать и публиковать новые объекты. Например, если какой-то consumer притормозит, подвиснет или аварийно прекратит свою работу.

В этом случае свободная память в арене закончится и producer не сможет выделять новые блоки памяти под публикуемые данные.

В такой ситуации producer начинает отсчет времени, в течении которого запросы на выделение блока памяти завершаются неудачно. Когда это время превысит некоторый заданный предел, producer выполняет следующее:

• создает новый файл с уникальным именем, выполняет его отображение в память и инициализирует эту область как новую арену;
• старую арену producer перестает обслуживать (т.е. отменяет отображение в память для старого файла, закрывает этот файл и делает unlink для него);
• всем подключившимся consumer-ам отсылается команда reattach;
• producer удаляет всю информацию, относящуюся к подключившимся consumer-ам (т.е. producer отныне считает, что подключенных consumer-ов нет).

Живые consumer-ы получив команду reattach должны:

• освободить все опубликованные для них объекты;
• отменить отображение в память старого файла и закрыть этот файл;
• выполнить заново процедуру подключения к producer-у (при этом будет получено новое имя файла для отображения в память и новый идентификатор сессии).

Обязательства consumer-а по освобождению опубликованных объектов

Каждый consumer должен контролировать время, в течении которого используются опубликованные объекты. Когда это время превышает некий установленный порог, consumer должен аварийно завершить свою работу, т.к. consumer не может работать нормально (либо у него где-то что-то зависло и он не может обрабатывать публикуемые объекты с должной скоростью, либо же у него произошла утечка ресурсов, среди которых были и опубликованные объекты).

Действия producer-а при получении повторного сообщения attach_request от consumer-а

В сообщении attach_request содержится уникальный идентификатор consumer-а. Это позволяет producer-у определить, считается ли данный consumer уже подключенным или нет.

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

• проходит по списку опубликованных для этого consumer-а объектов и уменьшает счетчик ссылок для объектов на единицу. Если для какого-то объекта счетчик ссылок обнуляется, то блок памяти под этот объект объявляется свободным и возвращается в арену для последующего переиспользования;
• очищает список опубликованных для consumer-а объектов (считаем, что для этой новой инкарнации consumer-а еще ничего не было опубликовано).

Действия producer-а при невозможности отправить исходящее сообщение consumer-у

Если producer предпринимает попытку отослать consumer-у исходящее сообщение (например, publish_announce, publish, ping_ack, reattach) и получает ошибку отсылки (записи) сообщения, то producer специальным образом помечает этого consumer-а и перестает отсылать ему какие-либо сообщения вообще.

Если ошибка записи произошла из-за аварийного завершения работы consumer-а, то consumer рестартует и пришлет новый attach_request. Это даст понять producer-у, что старая информация о consumer-е перестала быть актуальной.

Если же ошибка записи вызвана какими-то другими причинами, а consumer на самом деле все еще работает, то отсутствие активности со стороны producer-а будет диагностирована consumer-ом (см. ниже).

Действия consumer-а при отсутствии активности со стороны producer-а

Каждый подключившийся к producer-у consumer должен контролировать время отсутствия входящих сообщений от producer-а. Если это время превышает некий установленный порог, то consumer должен отослать producer-у сообщение ping.

Если затем в течении длительного времени от producer-а не будет никаких входящих сообщений (будь то ping_ack, publish_announce или reattach), то consumer должен посчитать, что producer недееспособен (завершил работу или завис и вскоре будет перезапущен). Поэтому consumer должен начать процедуру нового подключения к consumer-у.

Может случиться так, что с producer-ом все нормально, просто по каким-то причинам нарушилось взаимодействие между producer-ом и consumer-ом (см. выше). В этом случае переподключение consumer-а к producer-у создаст новый канал связи с producer-ом, а producer актуализирует свою информацию о consumer-е.

При этом суммарное время до отправки ping-а и время ожидания ответа на ping должно превышать время, в течении которого consumer должен освободить опубликованные для него объекты.

Такое требование гарантирует то, что в случае переподключения consumer-а к продолжающему свою работу producer-у, у consumer-а нет ссылок на неосвобожденные опубликованные объекты. Все опубликованные объекты к этому времени уже должны быть освобождены или же consumer должен прекратить свою работу из-за невозможности освободить какой-то из объектов вовремя.

Пояснения по принятым и описанным выше решениям

Взаимное обнаружение неработоспособности

Процесс producer контролирует работоспособность consumer-ов по наличию сообщений release. Если от consumer-а вовремя приходят release для опубликованных объектов, то значит с consumer-ом все нормально. Если ничего не публикуется, то смысла дополнительно следить за consumer-ами нет. Жив ли consumer или нет выяснится после попытки опубликовать новый объект.

Процесс consumer контролирует работоспособность producer-а за счет учета времени отсутствия сообщений от producer-а. Если producer перестал что-либо присылать и не отвечает на ping, значит старый producer прекратил свое существование и нужно начинать попытки подключения к новому producer-у (который либо уже запущен, либо будет запущен со временем).

Зачем нужен reattach и выделение новой арены?

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

Нас должны волновать две ситуации:

1. Какой-то consumer конкретно завис и его пока еще не рестартовали.
2. Какой-то consumer притормозил и не успевает освобождать опубликованные объекты. Однако, он все еще работает и опубликованные для него объекты этому consumer-у все еще нужны.

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

Мы могли бы подождать какой-то время (большее чем время, которое дается consumer-у на освобождение опубликованных объектов), после чего бы просто посчитали бы, что consumer уже мертв и все, что было опубликовано для consumer-а, должно быть освобождено. Это дало бы нам возможность использовать ту же арену не создавая новую.

Однако, такое ожидание не есть хорошо, т.к. в таком случае пострадают те consumer-ы, которые продолжают работать нормально.

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

Если те consumer-ы, которые не смогли сразу переподключиться просто временно подтормаживали, то они спокойно дообработают те объекты, которые для них были опубликованы в старой арене и уже затем переключатся на новую арену.

Те же consumer-ы, которые зависли намертво, будут просто рано или поздно убиты супервайзером. Но это уже никак не скажется на публикацию объектов через новую арену.

В приципе, у этого подхода есть опасность того, сперва один consumer начнет притормаживать и из-за него придется создать новую арену. Затем притормозит второй процесс и будет создана еще одна арена (при этом все еще будет существовать первая арена). Затем притормозит третий процесс и т.д. Арены будут плодиться и это будет вести к чрезмерному расходу памяти.

Но эта опасность в данном случае компенсируется тем, что consumer-ов будет мало (считанные единицы) и большому количеству арен просто неоткуда будет взяться.

Защита consumer-ов от перегрузки

В описанном механизме уже заложена одна степень защиты consumer-ов от слишком быстрого потока публикаций: сообщение publish отсылается только тем consumer-ам, которые успели вовремя ответить на publish_announce. Следовательно, если какой-то consumer начинает подтормаживать, то он будет присылать announce_ack-и с опозданием и новые publish-ы для него не будут отсылаться до тех пор, пока consumer не вернется к нормальному ритму и не сможет отсылать announce_ack вовремя.

Но эта защита может не сработать, если в consumer-е взаимодействие с producer-ом и обработка опубликованных объектов разнесена на разные рабочие нити. Так, одна нить общается с producer-ом и складывает указатели на опубликованные объекты в какую-то очередь. А вторая нить берет указатели из очереди и делает прикладную обработку опубликованных объектов.

В таком подходе скорость взаимодействия с producer-ом может не зависеть от скорости прикладной обработки.

Чтобы защититься от подобной реализации consumer-а можно сделать ограничение в producer-е: опубликовать для consumer-а можно не более N объектов. Если consumer "взял" N объектов, но еще не освободил ни один из них, то публикация N+1 объекта для этого consumer-а не выполняется.

Возможные способы реализации обмена сообщениями

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

Нужна поддержка Linux и Windows

Первая реализация описанного выше механизма должна сперва заработать под Linux-ом, а после того, как она докажет свою состоятельность на тестах, ее придется портировать еще и под Windows. Так что в основу обмена сообщениями должны быть положены средства, которые есть и в Linux, и в Windows, и которые работают более-менее схожим образом на обоих платформах.

Каждый consumer работает сразу с несколькими producer-ами

У одного producer-а может быть несколько consumer-ов. Об этом уже говорилось выше. Но сейчас нужно добавить, что каждый из предполагаемых consumer-ов должен будет получать информацию сразу от нескольких producer-ов. Т.е. будет какая-то такая схема:

Соответственно, есть вопрос о том, как контролировать IPC для общения сразу с несколькими consumer-ами и producer-ами.

Можно на каждое соединение между producer-ом и consumer-ом выделить отдельную рабочую нить, на которой будут выполняться блокирующие операции read/write (send/receive). Но более привлекательно пока что выглядит модель с использованием одной единственной нити, на которой операции read/write будут выполняться в неблокирующем режиме (посредством poll в Linux, к примеру).

Что задействовать для обмена сообщениями?

Описанный выше механизм предполагает обмен сообщениями attach_request, attach_ack, publish_announce, announce_ack, publish, release, ping, ping_ack.

Делать обмен этими сообщениями через shared memory не хочется, поскольку:

• я не припоминаю, чтобы в Linux-е был такой примитив межпроцессной синхронизации, как Event, доступный сразу для нескольких процессов (sem_open, sem_wait и пр. есть, а вот именно Event-а не припомню). Соответственно, мне не понятно, как можно сделать относительно дешевое ожидание поступления нового сообщения для процесса;
• мне непонятно как сделать так, чтобы один consumer на одной рабочей нити мог ждать уведомления сразу от нескольких producer-ов, если эти уведомления идут через shared memory. Могу ошибаться, но на Windows это можно сделать через именованные Event-ы и WaitForMultipleObject. Но вот под Linux-ом такой функциональности вроде как нет.

Поэтому на первый план выходят такие Linux-овые механизмы, как named pipes и POSIX message queue.

Пока я склоняюсь к тому, чтобы за основу взять named pipes. Из-за того, что контролировать доступность данных для чтения сразу для нескольких пайпов можно посредством poll. Что делает возможным вынесение работы со всеми producer-ами (или со всему consumer-ами) одну общую рабочую нить.

Использование named pipes для обмена сообщениями между producer-ом и consumer-ом может выглядеть следующим образом:

• процесс-producer создает named pipe с фиксированным для этого producer-а именем. Например, для producer-а с именем InStreamOne это может быть "./myprj-InStreamOne". Имя этого named pipe будет известно всем consumer-ам;
• процесс-producer открывает свой named pipe в режиме чтения. Процессы-consumer-ы открывают этот named pipe в режиме записи. Все сообщения, которые consumer-ы адресуют producer-у (attach_request, announce_ack, release, ping), записываются в этот named pipe;
• каждый consumer для взаимодействия с producer-ом создает свой собственный named pipe с уникальным именем. Например, если consumer с уникальным идентификатором ProcessorX хочет взаимодействовать с producer-ом с именем InStreamOne, то создается named pipe с именем "./myprj-ProcessorX-InStreamOne-PID-TIMESTAMP" (где вместо PID и TIMESTAMP подставляются соответствующие значения);
• consumer открывает свой named pipe в режиме чтения;
• consumer отсылает имя своего уникального named pipe в сообщении attach_request;
• producer открывает named pipe очередного consumer-а в режиме записи. Все сообщения, которые адресуются consumer-у (attach_ack, publish_announce, publish, ping_ack, reattach) отсылаются producer-ом в этот named pipe.

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

Если с ее реализацией возникнут какие-то проблемы, то придется смотреть в сторону POSIX message queue.

Как все это ляжет на имеющиеся в Windows средства я пока не прикидывал, так что даже описанный способ общения через named pipes может претерпеть существенные изменения.

Заключение

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

В любом случае спасибо, что дочитали.