Небольшая справка

Начиная с версии 4.2.6 контроль за интенсивностью входящего трафика (как raw, так и sop) в SObjectizer выполняется с помощью т.н. порогов ввода-вывода. Порог -- это пара значений <количество пакетов, объем пакетов>. Транспортный агент подсчитывает, сколько пакетов он извлек из канала и их суммарный объем. Как только какое-то из этих значений превышает соответствующее значение в заданном для канала пороге объявляется, что порог входящих данных превышен и канал блокируется. Чтение из канала останавливается до разблокирования канала, которое кто-то должен выполнить посредством отсылки сообщения msg_unblock_channel транспортному агенту. После разблокирования канала транспортный агент обнуляет свои счетчики и возобновляет операции чтения.

Мотивация

В версиях до 4.4.0-beta3 (включительно) контроль за входящим трафиком осуществлялся следующим образом:

• для raw-каналов транспортный агент выставлял признак блокированности канала в сообщении msg_raw_package. Соответственно, прикладной агент, обрабатывающий входящий raw-трафик, при получении msg_raw_package обнаруживал этот признак и отсылал транспортному агенту сообщение msg_unblock_channel. Таким образом прикладной агент получал какие-то гарантии, что он не будет получать входящие пакеты быстрее, чем будет способен их обработать;
• для sop-каналов транспортный агент выставлял признак блокированности канала в сообщении msg_sop_package, которое отсылалось агенту-коммуникатору. Агент-коммуникатор разбирал полученные данные, разбирал их, отсылал все сообщения, обнаруженные во входящем пакете и только после этого разблокировал канал с помощью сообщения msg_unblock_channel.

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

В очередной бета-версии 4.4.0 транспортная подсистема SObjectizer очень серьезно перестраивается. Так, теперь транспортные агенты сами будут разбирать входящие пакеты и генерировать содержащиеся в них сообщения. Соответственно, агент-коммуникатор уже не сможет управлять интенсивностью входящего SOP-трафика, поскольку транспортный агент даже не сможет никому сказать, что канал оказался заблокированным. А это ведет к том, что при интенсивном входящем SOP-трафике транспортный агент, работающий на отдельной нити (активный или член активной группы) способен вычитывать и разбирать входящие пакеты быстрее, чем прикладные агенты будут обрабатывать содержащиеся в этих пакетах сообщения.

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

Предлагаемое решение

Предлагаемое решение заключается в следующем:

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

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

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

Технические детали

Реализовать данную схему можно добавив в SObjectizer интерфейс:

class msg_destroy_notificator_t
  {
  public :
    virtual void
    on_destroy() = 0;
  };

И позволив назначение реализующих данный интерфейс объектов экземплярам класса so_4::rt::msg_data_t (или so_4::rt::impl::msg_data_impl_t). Например, таким образом:

class msg_data_t
  {
  private :
    ...
    std::auto_ptr< msg_destroy_notificator_t > m_destroy_notificator;

  public :
    ...
    ~msg_data_t()
      {
        if( m_destroy_notificator.get() )
          m_destroy_notificator->on_destroy();
      }
    ...
    void
    set_destroy_notificator(
      std::auto_ptr< msg_destroy_notificator_t > destroy_notificator )
      {
        m_destroy_notificator = destroy_notificator;
      }
    ...
  };

Таким образом обеспечивается автоматическое задействование объекта-уведомителя при разрушении экземпляра msg_data_t (т.е. сразу после завершения обработки сообщения всеми его подписчиками).