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

R>для 90х это было бы ОК, но на дворе-то 2010й
R>зачем было разводить столько графоманства, когда современными средствами задача решается тривиально?
R>пруф-код, слепленый за 5 ммин

R>

R>struct Subscriber {
R>    void OnSomething(int asContext) { printf("%i\n", asContext); }
R>};

R>int _tmain(int argc, _TCHAR* argv[])
R>{
R>    std::set<Subscriber*> subscribers;
R>    subscribers.insert(new Subscriber());

R>    int context = 7;
R>    task_group g;
R>    g.run( [subscribers,context] { for each(auto s in subscribers) { s->OnSomething(context); }} );
R>    g.wait();

R>    return 0;
R>}

Ваш пример не отображает проблеммную область.

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

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

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

gh2>2) избавитесь от связи (время жизни диспетчера) == (время, проведенное в критической секции)

Можно подробнее, я не понял в чем суть?

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

R>Вы досконально понимаете как устроена библиотека signal2 в boost?
R>лично меня пугает количество кода при вызове колбека, вот стек вызовов.

Это стек вызовов отладочной сборки, я так понимаю?

J>>с чего бы это буст был там недоступен? и что такое "нейтивные приложения" в контексте С++?
R>Ну я считаю что многопоточность это уже не совсем контекст c++, возможно я выбрал не совсем ту ветку форума, но более подходящей не нашел.
Если не интересует многопоточность, то есть Boost.Signals первой версии.
У нее проблемы с многопоточностью (как раз с отпиской клиентов), но проще реализация. Так что если не предполагается гонять все это в нескольких потоках (хотя у вас же там критические секции сплошь), то можно использовать ее.

R>Native приложение это приложение, которое может запускатья на ранней стадии загрузки виндовс (например консоль восстановления)
R>виндовый checkdisk так работает, он запускает c:\WINDOWS\system32\autochk.exe на ранней стадии загрузки. autochk.exe это и есть Native приложение.
R>Native приложение в основном пользуется лишь экспортом из ntdll.dll (недокументированными функциями), и ничего другого ему не доступно.
R>У нас по началу были большие трудности с CRT и STL, а вы про boost говорите...
Хм.. Какие именно трудности с STL? CRT — я еще могу понять, но STL — это же библиотека шаблонов Она сама по себе вообще ничего стороннего не зовет, кроме new/delete (и то, почти везде можно отдать свой аллокатор вместо стандартного, и тогда вообще прогулок в CRT не будет).

R>Интересно можно ли при написании модулей ядра в Linux пользоваться библиотеками STL и boost?
С этими вопросами вам лучше в "Низкоуровневое программирование".
Но я видел в англоязычных блогах сообщения, народ юзал буст для драйверов и был очень доволен.

You will always get what you always got
  If you always do  what you always did

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

R>Вы конечно-же со мной не согласитесь, но я считаю что c++ исключения это зло.
R>...

А откуда такая уверенность в том, что я с этим не соглашусь?

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

Правильно ли я понимаю, что в вашей модели, уже отписавшийся подписчик может получить уведомление?
В самом деле, в методе SendNotification вы делаете копию массива подписок под критической секцией.
После чего у вас некоторый слушатель может отписаться, но все равно получить уведомление.
Это не очень хорошо.

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

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

B>Правильно ли я понимаю, что в вашей модели, уже отписавшийся подписчик может получить уведомление?
B>В самом деле, в методе SendNotification вы делаете копию массива подписок под критической секцией.
B>После чего у вас некоторый слушатель может отписаться, но все равно получить уведомление.
B>Это не очень хорошо.

Еще мне не нравится, что метод Unsubscribe принимает на вход указатель-подписчика. Получается, что тот, кто отписывает, должен иметь этот указатель. Это не очень удобно. Обычно в таких случаях метод Subscribe возвращает некоторый дескриптор, который потом надо передавать в Unsubscribe. Наример, можно сделать так:

typedef struct HSUBSCRIBE__{int unused; }*HSUBSCRIBE;

class CNotificationDispatcher
{
...
private:
volatile long m_lSubsCounter;
...
};

CNotificationDispatcher::CNotificationDispatcher : m_lSubsCounter(0)
{
...
}

HSUBSCRIBE Subscribe(INotificationListener* pListener)
{
    if( S_OK )
    {
        return (HSUBSCRIBE)InterlockedIncrement( &m_lSubsCounter);
    }
    return (HSUBSCRIBE)0;
}

bool Unsubscribe(HSUBSCRIBE hSubs)
{
...
}

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

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

B>Правильно ли я понимаю, что в вашей модели, уже отписавшийся подписчик может получить уведомление?
B>В самом деле, в методе SendNotification вы делаете копию массива подписок под критической секцией.
B>После чего у вас некоторый слушатель может отписаться, но все равно получить уведомление.
B>Это не очень хорошо.

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

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

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

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

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

R>Это фундаметальная проблема такого подхода.
R>Решить её можно только перейдя к асинхронному обмену сообщениями. А именно с каждым объектом связывается понятие родительского потока, при нотификации родительским потокам рассылаются асинхронные сообщения. При их получении потоки проверяют, не отписан ли ещё каждый конкретный объект, если нет, то вызывается обработчик объекта. Соотв. отписывать можно только из родительского потока.

Это так же избавит от необходимости использования каких-либо мьютексов.

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

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

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

B>>Правильно ли я понимаю, что в вашей модели, уже отписавшийся подписчик может получить уведомление?
B>>В самом деле, в методе SendNotification вы делаете копию массива подписок под критической секцией.
B>>После чего у вас некоторый слушатель может отписаться, но все равно получить уведомление.
B>>Это не очень хорошо.

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

Есть много разных способов. Например, можно связать с каждым подписчиком класс-защелку. К ней должен иметь доступ как подписчик, так и уведомитель.
Защелка открывается в методе Subscribe после добавление подписчика. В методе Unsubscribe она закрывается уведомителем под критической секцией.
Класс подписчик в своем методе OnNotification должен вначале проверять, разрешен ли ему вызов. Если нет, то просто выходить.
Защелкой должен владеть подписчик. Можно просто создать обертку над подписчиком ( в принципе и умный указатель есть обертка ), который владеет и подписчиком
и защелкой. В более сложном варианте можно сделать защелку более сложной, поместив туда еще и событие. При вызове OnNotification, когда защелка открыта,
можно увеличивать счетчик использования защелки в начале и уменьшать в конце. А событие взводить, когда счетчик станет равным нулю. Тогда на событии можно ожидать,
когда подписчик закончит обрабатывать свой OnNotification.

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

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

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

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

R>Это фундаметальная проблема такого подхода.
R>Решить её можно только перейдя к асинхронному обмену сообщениями. А именно с каждым объектом связывается понятие родительского потока, при нотификации родительским потокам рассылаются асинхронные сообщения. При их получении потоки проверяют, не отписан ли ещё каждый конкретный объект, если нет, то вызывается обработчик объекта. Соотв. отписывать можно только из родительского потока.

R>

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

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

B>Есть много разных способов. Например, можно связать с каждым подписчиком класс-защелку. К ней должен иметь доступ как подписчик, так и уведомитель.
B>Защелка открывается в методе Subscribe после добавление подписчика. В методе Unsubscribe она закрывается уведомителем под критической секцией.
B>Класс подписчик в своем методе OnNotification должен вначале проверять, разрешен ли ему вызов. Если нет, то просто выходить.
B>Защелкой должен владеть подписчик. Можно просто создать обертку над подписчиком ( в принципе и умный указатель есть обертка ), который владеет и подписчиком
B>и защелкой. В более сложном варианте можно сделать защелку более сложной, поместив туда еще и событие. При вызове OnNotification, когда защелка открыта,
B>можно увеличивать счетчик использования защелки в начале и уменьшать в конце. А событие взводить, когда счетчик станет равным нулю. Тогда на событии можно ожидать,
B>когда подписчик закончит обрабатывать свой OnNotification.

Спасибо за идею, надо будет поразмышлять над ней...

B>Я в свое время много потрудился над данной задачей. Но у меня требования были пожестче. Там есть немало подводных камней, хотя на вид она кажется простой.

И это мне нравится больше всего

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

R>Это фундаметальная проблема такого подхода.
R>Решить её можно только перейдя к асинхронному обмену сообщениями. А именно с каждым объектом связывается понятие родительского потока, при нотификации родительским потокам рассылаются асинхронные сообщения. При их получении потоки проверяют, не отписан ли ещё каждый конкретный объект, если нет, то вызывается обработчик объекта. Соотв. отписывать можно только из родительского потока.

Не совсем понял идею родительских потоков (это как?)

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

Хочу спросить у форумчан: на сколько критично автоматическая отписка при уничтожении подписчиков или издателя?
Плюсы:
— программисту не надо постоянно следить за подписками, при уничтожении подписчика, объект сам отпишется от издателя (издатель отпишет/уведомит подписчиков соответственно);
— код бизнес логики не изобилует дополнительными проверками.

Минусы:
— дополнительный расход памяти на дополнительные списки (к каким издателям подписан подписчик), для встраиваемых систем может быть критично;
— расслабляет программиста, вместо продуманного алгоритма с четким временем жизни объектов, чаще пишется код по принципу "заработало и ладно".

Все выше сказанное — мое ИМХО, но жду ваших конструктивных ответов.

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

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

R>Мне кажется не всегда удобно пользоваться асинхронными сообщениями, например если надо у пользователя спросить что-то и в зависимости от ответа продолжить или остановится, с асинхронными сообщениями это будет сделать сложнее. Еще проще для понимания фиксированные состояния системы, если поток ожидает пока пройдет нотификация проще понимать что происходит в системе (проще фиксить баги в некоторых случаях к примеру)
R>Если выбирать между гарантией неполучения сообщения после вызова Unsubscribe в синхронных сообщениях и асинхронными сообщениями я скорее выбрал бы синхронные.

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

Хочу поблагодарить всех, кто обратил внимание на мой пост и написал свои замечания/советы.
также хочу представить на общий суд отредактированный код, который, надеюсь, учитывает большинство замечаний/предложений.
Отдельное спасибо хочу выразить пользователю remark за идею копирования списка при добавлении-удалении, правда мне предстоит непростая задача объяснить доходно об этой технологии, не уходя далеко от тематики статьи.
в общем вот, версия 1.0.0.2:

namespace Observer
{
    //////////////////////////
    // Subscriber
    //////////////////////////
    typedef unsigned __int64    SubscriberId;
    class CSubscriber
    {
    public:
        virtual ~CSubscriber(){}
        virtual void    OnNotification(void* pContext) = 0;
        SubscriberId    GetSubscriberId()   {return (SubscriberId)this;}
    };
    typedef boost::shared_ptr<CSubscriber> CSubscriberPtr;

    //////////////////////////////////////////////////////////////////////
    // Dispatcher
    ///////////////////////////////////

    class CDispatcher
    {
    private:
        typedef std::vector<CSubscriberPtr>         CSubscriberList;
        typedef boost::shared_ptr<CSubscriberList>  CSubscriberListPtr;
    public:
        SubscriberId Subscribe(CSubscriberPtr pNewSubscriber)
        {
            //Declaration of the next shared pointer before ScopeLocker
            //prevents release of subscribers from under lock
            CSubscriberListPtr pNewSubscriberList(new CSubscriberList());
            //Enter to locked section
            CScopeLocker ScopeLocker(m_Lock);
            if(m_pSubscriberList)
            {
                //Copy existing subscribers
                pNewSubscriberList->assign(m_pSubscriberList->begin(), m_pSubscriberList->end());
            }
            for(size_t i = 0; i < pNewSubscriberList->size(); ++i)
            {
                CSubscriberPtr pCurrSubscriber = (*pNewSubscriberList)[i];
                if(pCurrSubscriber->GetSubscriberId() == pNewSubscriber->GetSubscriberId())
                {
                    return 0;
                }
            }
            //Add new subscriber to new subscriber list
            pNewSubscriberList->push_back(pNewSubscriber);
            //Exchange subscriber lists
            m_pSubscriberList = pNewSubscriberList;
            return pNewSubscriber->GetSubscriberId();
        }
        bool Unsubscribe(SubscriberId id)
        {
            //Declaration of the next shared pointers before ScopeLocker
            //prevents release of subscribers from under lock
            CSubscriberPtr      pSubscriberToRelease;
            CSubscriberListPtr  pNewSubscriberList;
            //Enter to locked section
            CScopeLocker ScopeLocker(m_Lock);
            if(!m_pSubscriberList)
            {
                //No subscribers
                return false;
            }
            pNewSubscriberList = CSubscriberListPtr(new CSubscriberList());
            for(size_t i = 0; i < m_pSubscriberList->size(); ++i)
            {
                CSubscriberPtr pCurrSubscriber = (*m_pSubscriberList)[i];
                if(pCurrSubscriber->GetSubscriberId() == id)
                {
                    pSubscriberToRelease = pCurrSubscriber;
                }
                else
                {
                    pNewSubscriberList->push_back(pCurrSubscriber);
                }
            }
            //Exchange subscriber lists
            m_pSubscriberList = pNewSubscriberList;
            if(!pSubscriberToRelease.get())
            {
                return false;
            }
            return false;
        }
        void SendNotification(void* pContext)
        {
            CSubscriberListPtr pSubscriberList;
            {
                CScopeLocker ScopeLocker(m_Lock);
                if(!m_pSubscriberList)
                {
                    //No subscribers
                    return;
                }
                //Get shared pointer to an existing list of subscribers
                pSubscriberList = m_pSubscriberList;
            }
            //pSubscriberList pointer to copy of subscribers' list
            for(size_t i = 0; i < pSubscriberList->size(); ++i)
            {
                (*pSubscriberList)[i]->OnNotification(pContext);
            }
        }
    private:
        CSubscriberListPtr  m_pSubscriberList;
        CLock               m_Lock;
    };

}; //namespace Observer

в функции Unsubscribe я допустил ошибку:

R>        bool Unsubscribe(SubscriberId id)
              ...
R>            if(!pSubscriberToRelease.get())
R>            {
R>                return false;
R>            }
R>            return false; <- должно быть return true;
R>        }

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

R>Не совсем понял идею родительских потоков (это как?)

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

R>Мне кажется не всегда удобно пользоваться асинхронными сообщениями, например если надо у пользователя спросить что-то и в зависимости от ответа продолжить или остановится, с асинхронными сообщениями это будет сделать сложнее. Еще проще для понимания фиксированные состояния системы, если поток ожидает пока пройдет нотификация проще понимать что происходит в системе (проще фиксить баги в некоторых случаях к примеру)
R>Если выбирать между гарантией неполучения сообщения после вызова Unsubscribe в синхронных сообщениях и асинхронными сообщениями я скорее выбрал бы синхронные.

Синхронные сообщения безусловно удобнее, но дедлоки это их бич.

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

B>Есть много разных способов. Например, можно связать с каждым подписчиком класс-защелку. К ней должен иметь доступ как подписчик, так и уведомитель.
B>Защелка открывается в методе Subscribe после добавление подписчика. В методе Unsubscribe она закрывается уведомителем под критической секцией.
B>Класс подписчик в своем методе OnNotification должен вначале проверять, разрешен ли ему вызов. Если нет, то просто выходить.

Если он будет просто проверять в начале функции, то мы возвращаемся туда, откуда начали, — может работать нотификация уже отписанного объекта. Ровным счётом ничего не решается.
Какие ешё есть разные способы?


B>Защелкой должен владеть подписчик. Можно просто создать обертку над подписчиком ( в принципе и умный указатель есть обертка ), который владеет и подписчиком
B>и защелкой. В более сложном варианте можно сделать защелку более сложной, поместив туда еще и событие. При вызове OnNotification, когда защелка открыта,
B>можно увеличивать счетчик использования защелки в начале и уменьшать в конце. А событие взводить, когда счетчик станет равным нулю. Тогда на событии можно ожидать,
B>когда подписчик закончит обрабатывать свой OnNotification.

Это потенциальные дедлоки.

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

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

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