Параллельное программирование на С++ в действии. Практика разработки многопоточных программ - [163]

	Отсутствует	Отсутствует	Типы std::atomic_xxx, шаблон класса std::atomic<>, функция std::atomic_thread_fence()	Глава 5
Потокобезопасные контейнеры	Контейнеры в пакете java.util.concurrent.	Отсутствует	Отсутствует	Отсутствует	Главы 6 и 7
Будущие результаты	Интерфейс java.util.concurrent.future и ассоциированные с ним классы	Отсутствует	Шаблонные классы boost::unique_future<> и boost::shared_future<>	Шаблонные классы std::future<>, std::shared_future<>, и std::atomic_future<>	Глава 9
Пулы потоков	Класс java.util.concurrent.ThreadPoolExecutor	Отсутствует	Отсутствует	Отсутствует	Глава 9
Прерывание потока	Метод interrupt() класса java.lang.Thread	pthread_cancel()	Функция-член interrupt() класса boost::thread	Отсутствует

Приложение С. Каркас передачи сообщений и полный пример программы банкомата

В разделе 4.1 мы познакомились с каркасом передачи сообщений между потоками, продемонстрировав его на примере программы банкомата. В этом приложении приводится полный код примера, включая и код каркаса передачи сообщений.

В листинге С.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);
 }
};
}

Если sender только ссылается на очередь сообщений, то receiver ей владеет. Мы можем получить объект sender, ссылающийся на очередь, воспользовавшись неявным преобразованием. Процедура диспетчеризации сообщения начинается с обращения к функции wait(). При этом создается объект dispatcher, ссылающийся на очередь, которой владеет receiver. Класс dispatcher показан в следующем листинге; как видите, содержательная работа производится в его деструкторе. В данном случае работа состоит в ожидании сообщения и его диспетчеризации.

Листинг С.4. Класс dispatcher

namespace messaging {

class close_queue {}; ←Сообщение о закрытии очереди

class dispatcher {

 queue* q;                             │Экземпляры

 bool chained;                         │