Реакция на перегрузку агента (круг второй)
Yauheni Akhotnikau
Появилось время изложить мысли по поводу реакции ядра SObjectizer на
перегрузку агента. Это далеко не полностью завершенная модель поведения
SObjectizer, а соображения по поводу возможности приостановки
агента-отправителя сообщения в случае перегрузки.
Итак, основной постулат состоит в том, что в приостановке отправителя есть
смысл только тогда, когда приостановка выполняется внутри метода send.
Т.е. вызывается send, внутри него выясняется, что это сообщение вызывает
перегрузку, и send "засыпает" до тех пор, пока условия перегрузки не
исчезнут (ну или не истечет тайм-аут ожидания).
Так вот, если отталкиваться от этого постулата, то оказывается, что есть
несколько контекстов, в которых приостановка отправителя недопустима.
1. Когда агент-отправитель и агент-получатель работают на одной нити
диспетчера. Например, они оба являются пассивными агентами или членами
одной активной группы.
2. Когда сообщение генерируется нитью таймера. Т.е. на очередном тике
таймера определяется, что должен быть отослан экземпляр
отложенного/периодического сообщения. Нить таймера не должна
приостанавливаться, даже это генерируемые им сообщения вызывают перегрузку
части агентов приложения.
3. Когда сообщение генерируется агентом, вроде нынешнего
агента-коммуникатора, обслуживающего SOP-каналы. Этому агенту поступают на
обработку сообщения глобальных агентов, он их сериализует и пересылают
нужным транспортным агентам. А уже транспортные агенты пишут
сериализованные пакеты в канал. Так вот такой агент-сериализатор не должен
приостанавливаться, даже если какой-то транспортный агент по каким-то
причинам не успевает разгребать свою очередь сообщений.
4. Когда происходит отсылка широковещательного сообщения. Например,
сообщение M отсылается агентом A и обрабатывается агентами B, C и D. У
агента C образовалась перегрузка. Из-за сообщений, которые ему раньше
наотсылал какой-то агент E. Я считаю, что не следует проистанавливать
работу агента A. Хотя это спорный момент.
Так вот, исходя из того, что контекстов, на которых возможность
приостановки отправителя слишком много, я предлагаю вообще исключить из
SObjectizer такую функциональность. Т.е. защита от перегрузки заключается
всего лишь в одном действии -- выбрасывании лишнего сообщения.
Несколько следствий из этого вывода.
Следствие 1. Потеря сообщения для какого-то агента может требовать
соответствующей реакции от этого агента. Представим себе транспортного
агента, который обслуживает raw-канал. Например, с его помощи реализуется
SOAP-транспорт. На агента может идти поток сообщений, каждое из которых
содержит часть SOAP-пакета. Если одно из этих сообщений будет определено
как "лишнее" и выброшено, то агенту нет смысла доставлять оставшиеся
сообщения. Агент должен понять, что часть исходящего трафика из-за
перегрузки была потеряна и ему остается только разорвать существующее
соединение.
Следствие 2. Не понятно, имеет ли смысл выбрасывать из send исключения,
которые скажут отправителю, что сообщение просто потеряно. В случае
широковещательной рассылки смысла в этом исключении не много (опять же,
должно это исключение прерывать широковещательную отсылку или же его нужно
инициировать только после обработки всех получателей). В случае
целенаправленной отсылки в этом может быть смысл, если только момент
выбрасывания сообщения определяется прямо внутри send. Если же это
происходит чуть позже (например, уже диспетчером при попытке вызова
обработчика сообщения), то исключение вообще не может быть передано
отправителю.
Следствие 3. Механизм приостановки отправителя все-таки нужен в ряде
задач. Например, при организации flow-based подходов. В таких случаях
приостановка может быть реализована путем предоставления специальных
объектов-каналов (channels, pipelines). Использование этих каналов будет
требовать от программиста заботы о том, чтобы читатель и писатель работали
на разных контекстах.
Вот как может работать подобный канал. Канал, это очередь сообщений. При
постановке в него сообщения объект отсылает уведомление агенту-получателю.
Когда уведомление доходит до получателя, получатель извлекает свое
сообщение из канала. Если выясняется, что канал переполнен, то писатель
приостанавливается на операции сохранения нового сообщения в канале.
В принципе, в коде эта штука может выглядеть так:
// Сообщение, которым будут обмениваться агент-парсер csv-файла
// и агент импорта записей из csv-файла в БД.
struct msg_next_csv_records { ... };
// Канал, который хранит сообщения с порциями csv-файла.
typedef buffered_channel_t< msg_next_csv_records > csv_records_channel_t;
// Действия агента-парсера.
class a_csv_parser_t : public so_5::agent_t {
...
void evt_parse_file(...) {
while( <не конец файла> ) {
<формирование очередной порции>
// Порция должна быть отправлена обработчику.
records_channel_.write( next_portion );
}
}
};
// Действия агента-импортера.
class a_db_importer_t : public so_5::agent_t {
...
void evt_next_records( const csv_records_channel_t::notify_t & notify ) {
// Извлекаем соответствующее сообщение из канала.
so_5::msg_handle_t< msg_next_csv_records > records =
notify.extract_object();
<обработка очередной порции данных>
}
};
Т.е. в очередь агенту ставится экземпляр сообщения
buffered_channel_t<T>::notify_t. Пока этот экземпляр идет получателю,
объект типа T остается в канале. Но, как только для notify_t будет вызван
extract_object, объект T будет изъят их канала. Соотвественно, именно в
этот момент, отправитель будет разбужен, если он был приостановлен из-за
переполнения канала.
И еще один момент. Ранее мы обсуждали возможность того, чтобы
агент-получатель мог почистить свою очередь сообщений при возникновении
перегрузки. Как раз то, что нить таймера должна отсылать сообщения
максимально быстро, похоже, ставит крест на этой возможности. Будет совсем
плохо, если нить таймера окажется приостановленной из-за того, что
какой-то перегруженный агент начнет перетряхивать несколько тысяч
сообщений в своей очереди.
--
Regards,
Yauheni Akhotnikau
Chief Developer
Intervale, http://www.intervale.ru
e-mail:eao...@intervale.ru <mailto:eao...@intervale.ru>