Элегантные способы уведомить потребителя, когда производитель сделан?

Question

Я реализую concurrent_blocking_queue с минимальными функциями:

//a thin wrapper over std::queue
template<typename T>
class concurrent_blocking_queue
{
    std::queue<T> m_internal_queue;
     //...
 public:
     void add(T const & item);
     T&   remove();
     bool empty();
};

Я намереваюсь использовать это для проблемы производитель-потребитель (я думаю, это где используются такие структуры данных?). Но я застрял на одной проблеме:

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

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

Answer 1

В прошлом я просто представил фиктивный "готовый" продукт. Поэтому, если производитель может создавать «продукты», скажем, типа A и типа B, я изобрел тип «done». Когда потребитель сталкивается с продуктом типа «готово», он знает, что дальнейшая обработка больше не требуется.

Answer 2

Это правда, что обычно ставят в очередь специальное сообщение "мы закончили";однако я думаю, что первоначальное желание OP создать внеполосный индикатор является разумным.Посмотрите на сложность, которую люди рассматривают, чтобы настроить внутриполосное сообщение о завершении!Типы прокси, шаблоны;печаль во благо.Я бы сказал, что метод done() проще и проще, и он делает общий случай (мы еще не закончили) быстрее и чище.

Я согласен с kids_fox, что try_remove возвращает ошибкукод, если очередь сделана, предпочтительнее, но это стилистически и YMMV.

Редактировать: Бонусные баллы за реализацию очереди, которая отслеживает, сколько производителей осталось в ситуации с несколькими производителями, и вызывает выполненное исключение, если все производители бросили полотенце ;-) Не собираюсь делать это с внутриполосными сообщениями!

Answer 3

В моих очередях обычно используются указатели (с std::auto_ptr в интерфейсе, чтобы четко указать, что отправитель может больше не обращаться к указателю);по большей части объекты в очереди были полиморфными, поэтому динамическое размещение и ссылочная семантика были необходимы в любом случае.В противном случае не должно быть слишком сложно добавить флаг «конец файла» в очередь.Вам потребуется специальная функция на стороне производителя (close?), Чтобы установить ее (используя точно такие же блокирующие примитивы, что и при записи в очередь), и цикл в функции удаления должен ждать, пока что-то будеттам, или очередь будет закрыта.Конечно, вам нужно будет вернуть значение Fallible, чтобы читатель мог знать, было ли чтение успешным или нет.Кроме того, не забывайте, что в этом случае вам нужно notify_all, чтобы гарантировать, что все процессы, ожидающие в состоянии, активизируются.

Кстати: я не совсем понимаю, как ваш интерфейс реализуем.Что означает T&, возвращаемое remove.По сути, remove должно быть примерно таким:

Fallible<T>
MessageQueue<T>::receive()
{
    ScopedLock l( myMutex );
    while ( myQueue.empty() && ! myIsDone )
        myCondition.wait( myMutex );
    Fallible<T> results;
    if ( !myQueue.empty() ) {
        results.validate( myQueue.top() );
        myQueue.pop();
    }
    return results;
}

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

Для остальных:

void
MessageQueue<T>::send( T const& newValue )
{
    ScopedLock l( myMutex );
    myQueue.push( newValue );
    myCondition.notify_all();
}

void
MessageQueue<T>::close()
{
    ScopedLock l( myMutex );
    myIsDone = true;
    myCondition.notify_all();
}

Answer 4

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

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

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

Как потребители собираются уведомить, что они обработали свое исключение и собираются умереть?Достаточно ли "умереть" или вам нужно подождать с дескриптором потока?Если вам нужно ждать на дескрипторе потока, что будет делать ожидание - поток, запрашивающий отключение очереди или последний потребительский поток, чтобы уведомить?

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

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

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