Параллельное программирование на С++ в действии

В листинге С.1 показан код очереди сообщений. Сообщения хранятся в списке и представлены указателями на базовый класс. Сообщения конкретного типа обрабатываются шаблонным классом, производным от этого базового класса. В момент помещения сообщения в очередь конструируется подходящий экземпляр обертывающего класса и сохраняется указатель на него; операция извлечения возвращает именно этот указатель. Поскольку в классе message_base нет функций-членов, извлекающий поток должен привести указатель к нужному типу wrapped_message, прежде чем сможет получить хранящееся сообщение.

Листинг С.1. Простая очередь сообщений

#include

#include

#include

#include

namespace messaging

{                     │Базовый класс

struct message_base {←┘элементов очереди

 virtual ~message_base() {}

};

template │Для каждого типа сообщений

struct wrapped_message:←┘имеется специализация

message_base {

 Msg contents;

 explicit wrapped_message(Msg const& contents_):

  contents(contents_) {}

};

            │Наша очередь

class queue←┘сообщений

{                                              │В настоящей

 std::mutex m;                                 │очереди хранят-

 std::condition_variable с;                    │ся указатели на

 std::queue > q;←┘message_base

public:

 template               │Обернуть добав-

 void push(T const& msg)            │ленное сообще-

 {                                  │ние и сохранить

  std::lock_guard lk(m);│указатель

  q.push( ←┘

   std::make_shared >(msg));

  с.notify_all();

 }

 std::shared_ptr wait_and_pop()│Блокирует до

 {                                           │появления в

  std::unique_lock lk(m);        │очереди хотя бы

  c.wait(lk, [&]{ return !q.empty(); });    ←┘одного элемента

  auto res = q.front();

  q.pop();

  return res;

 }

};

}

Отправкой сообщений занимается объект класса sender, показанного в листинге С.2. Это не более чем тонкая обертка вокруг очереди сообщений, которая позволяет только добавлять сообщения. При копировании экземпляров sender копируется только указатель на очередь, а не сама очередь.

Листинг С.2. Класс sender

namespace messaging {

 class sender {│

sender обертывает указатель

 queue* q;    ←┘на очередь

public:     │У сконструированного по умолчанию

 sender() :←┘sender'a нет очереди

 q(nullptr) {}

                             │Разрешаем конструирование

 explicit sender(queue* q_):←┘из указателя на очередь

  q(q_) {}

 template

 void send(Message const& msg) {

  if (q){        │Отправка сообщения сводится

   q->push(msg);←┘к помещению его в очередь

  }

 }

};

}

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

Листинг С.3. Класс receiver

namespace messaging {

class receiver {

 queue q; ←receiver владеет очередью

public:              │Разрешить неявное преобразование в объект

 operator sender() {←┘sender, ссылающийся на эту очередь

  return sender(&q);

 }

                     │При обращении к функции ожидания

 dispatcher wait() {←┘очереди создается диспетчер

  return dispatcher(&q);

 }

};

}