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

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

В разделе 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;←┘

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

sender

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

sender

namespace messaging {

 class sender {│

 queue* q;  ←┘

public:   │

 sender() :←┘

 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; ←

public:        │

 operator 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;             │

                    │

 dispatcher(dispatcher const&)=delete;←┘

 dispatcher& operator=(dispatcher const&)=delete;

 template<

  typename Dispatcher,│

  typename Msg,    │

  typename Func>   ←┘

 friend class TemplateDispatcher;

 void wait_and_dispatch()

 {     

  for (;;) {←┘

  auto msg = q->wait_and_pop();

  dispatch(msg);

  }

 }       

         │

 bool dispatch (←┘

  std::shared_ptr const& msg) {

 if (dynamic_cast*>(msg.get())) {

  throw close_queue();

 }

 return false;

 }

public:              │

 dispatcher(dispatcher&& other):←┘

 q(other.q), chained(other.chained) {│

  other.chained = true;        ←┘

 }

 explicit dispatcher(queue* q_): q(q_), chained(false) {}

 template

 TemplateDispatcher

 handle(Func&& f)←┐

 {        

  return TemplateDispatcher(

  q, this, std::forward(f));

 }

 ~dispatcher() noexcept(false)←┐

 {              

  if (!chained) {

  wait_and_dispatch();

  }

 }

};

}

Экземпляр

dispatcher

wait()

wait_and_dispatch()

dispatch()

dispatch()

close_queue

false

close_queue

noexcept(false)

noexcept(true)

close_queue

Но просто вызывать функцию

wait()

handle()

handle()

dispatcher

TemplateDispatcher

chained

TemplateDispatcher

Листинг С.5. Шаблон класса

TemplateDispatcher

namespace messaging {

template<

 typename PreviousDispatcher, typename Msg, typename Func>

class TemplateDispatcher {

 queue* q;

 PreviousDispatcher* prev;

 Func f;

 bool chained;

 TemplateDispatcher(TemplateDispatcher const&) = delete;

 TemplateDispatcher& operator=(

  TemplateDispatcher const&) = delete;

 template<

  typename Dispatcher, typename OtherMsg, typename OtherFunc>

 friend class TemplateDispatcher;←┐

 void wait_and_dispatch()     │

 {                 │

  for (;;) {

  auto msg = q->wait_and_pop();

  if (dispatch(msg))←┐

  break;       │

  }         

 }

 bool dispatch(std::shared_ptr const& msg) {

  if (wrapped_message* wrapper =

  dynamic_cast*>(

  msg.get())) {    ←┐

  f(wrapper->contents);│

  return true;     │

  }          

 else {

  return prev->dispatch(msg);←┐

 }              

 }

public:

 TemplateDispatcher(TemplateDispatcher&& other):

  q(other.q), prev(other.prev), f(std::move(other.f)),

  chained(other.chained) {

  other.chained = true;

 }

 TemplateDispatcher(

  queue* q_, PreviousDispatcher* prev_, Func&& f_):

  q(q_), prev(prev_), f(std::forward(f_)), chained(false)

 {

  prev_->chained = true;

 }

 template

 TemplateDispatcher

 handle(OtherFunc&& of)←┐

 {           

  return TemplateDispatcher<

  TemplateDispatcher, OtherMsg, OtherFunc>(

   q, this, std::forward(of));

 }

 ~TemplateDispatcher() noexcept(false)←┐

 {                   │

  if (!chained) {           

  wait_and_dispatch();

  }

 }

};

}

Шаблон класса

TemplateDispatcher<>

dispatcher

wait_and_dispatch()

Поскольку мы не возбуждаем исключения, если сообщение обработало, то теперь в цикле (1) нужно проверять, обработали мы сообщение или нет. Обработка прекращается, как только сообщение успешно обработало, чтобы в следующий раз можно было ждать очередного набора сообщений. Если найдено соответствие указанному типу сообщения, то вызывается предоставленная функция (2), а не возбуждается исключение (хотя функция-обработчик может и сама возбудить исключение). Если же соответствие не найдено, то мы передаем сообщение предыдущему диспетчеру в цепочке (3). В самом первом экземпляре это будет объект

dispatcher

handle()

TemplateDispatcher<>

dispatcher

close_queue

noexcept(false)

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

И чтобы закончить пример из главы 4, в листинге С.6 приведён код сообщений, в листингах С.7, С.8 и С.9 — различные конечные автоматы, а в листинге С.10 — управляющая программа.

Листинг С.6. Сообщения банкомата

struct withdraw {

 std::string account;

 unsigned amount;

 mutable messaging::sender atm_queue;

 withdraw(std::string const& account_,

  unsigned amount_, messaging::sender atm_queue_):

  account(account_), amount(amount_), atm_queue(atm_queue_) {}

};

struct withdraw_ok {};

struct withdraw_denied {};

struct cancel_withdrawal {

 std::string account;

 unsigned amount;

 cancel_withdrawal(std::string const& account_,

  unsigned amount_):

  account(account_), amount(amount_) {}

};

struct withdrawal_processed {

 std::string account;

 unsigned amount;

 withdrawal_processed(std::string const& account_,

  unsigned amount_):

  account(account_), amount(amount_) {}

};

struct card_inserted {

 std::string account;

 explicit card_inserted(std::string const& account_):

  account(account_) {}

};

struct digit_pressed {

 char digit;

 explicit digit_pressed(char digit_):

  digit(digit_) {}

};

struct clear_last_pressed {};

struct eject_card {};

struct withdraw_pressed {

 unsigned amount;

 explicit withdraw_pressed(unsigned amount_):

  amount(amount_) {}

};

struct cancel_pressed {};

struct issue_money {

 unsigned amount;

 issue_money(unsigned amount_):

  amount(amount_) {}

};

struct verify_pin {

 std::string account;

 std::string pin;

 mutable messaging::sender atm_queue;

 verify_pin(std::string const& account_, std::string const& pin_,

  messaging::sender atm_queue_):

  account(account_), pin(pin_), atm_queue(atm_queue_) {}

};

struct pin_verified {};

struct pin_incorrect {};

struct display_enter_pin {};

struct display_enter_card {};

struct display_insufficient_funds {};

struct display_withdrawal_cancelled {};

struct display_pin_incorrect_message {};

struct display_withdrawal_options (};

struct get_balance {

 std::string account;

 mutable messaging::sender atm_queue;

 get_balance(

  std::string const& account_, messaging::sender atm_queue_):

  account(account_), atm_queue(atm_queue_) {}

};

struct balance {

 unsigned amount;

 explicit balance(unsigned amount_):

  amount(amount_) {}

};

struct display_balance {

 unsigned amount;

 explicit display_balance(unsigned amount_):

  amount(amount_) {}

};

struct balance_pressed {};

Листинг С.7. Конечный автомат банкомата

class atm {

 messaging::receiver incoming;

 messaging::sender bank;

 messaging::sender interface_hardware;

 void (atm::*state)();

 std::string account;

 unsigned withdrawal_amount;

 std::string pin;

 void process_withdrawal() {

  incoming.wait().handle(

  [&](withdraw_ok const& msg) {

   interface_hardware.send(

   issue_money(withdrawal_amount));

   bank.send(

   withdrawal_processed(account, withdrawal_amount));

   state = &atm::done_processing;

  }

  ).handle(

  [&](withdraw_denied const& msg) {

   interface_hardware.send(display_insufficient_funds());

   state = &atm::done_processing;

  }

  ).handle(

  [&](cancel_pressed const& msg) {

   bank.send(

   cancel_withdrawal(account, withdrawal_amount));

  interface_hardware.send(

   display_withdrawal_cancelled());

  state = &atm::done_processing;

  }

  );

 }

 void process_balance() {

  incoming.wait().handle(

  [&](balance const& msg) {

   interface_hardware.send(display_balance(msg.amount));

   state = &atm::wait_for_action;

  }

  ).handle(

  [&](cancel_pressed const& msg) {

   state = &atm::done_processing;

  }

  );

 }

 void wait_for_action() {

  interface_hardware.send(display_withdrawal_options());

  incoming.wait().handle(

  [&](withdraw_pressed const& msg) {

   withdrawal_amount = msg.amount;

   bank.send(withdraw(account, msg.amount, incoming));

   state = &atm::process_withdrawal;

  }

  ).handle(

  [&](balance_pressed const& msg) {

   bank.send(get_balance(account, incoming));

   state = &atm::process_balance;

  }

  ).handle(

  [&](cancel_pressed const& msg) {

  state = &atm::done_processing;

  }

  );

 }

 void verifying_pin() {

  incoming.wait().handle(

  [&](pin_verified const& msg) {

   state = &atm::wait_for_action;

  }

  ).handle(

  [&](pin_incorrect const& msg) {

   interface_hardware.send(

   display_pin_incorrect_message());

   state = &atm::done_processing;

  }

  ).handle(

  [&](cancel_pressed const& msg) {

   state = &atm::done_processing;

  }

  );

 }

 void getting_pin() {

  incoming.wait().handle(

  [&](digit_pressed const& msg) {

   unsigned const pin_length = 4;

   pin += msg.digit;

   if (pin.length() == pin_length) {

   bank.send(verify_pin(account, pin, incoming));

   state = &atm::verifying_pin;

   }

  }

  ).handle(

  [&](clear_last_pressed const& msg) {

   if (!pin.empty()) {

   pin.pop_back();

   }

  }

  ).handle(

  [&](cancel_pressed const& msg) {

   state = &atm::done_processing;

  }

  );

 }

 void waiting_for_card() {

  interface_hardware.send(display_enter_card());

  incoming.wait().handle(

  [&](card_inserted const& msg) {

   account = msg.account;

   pin = "";

   interface_hardware.send(display_enter_pin());

   state = &atm::getting_pin;

  }

  );

 }

 void done_processing() {

  interface_hardware.send(eject_card());

  state = &atm::waiting_for_card;

 }

 atm(atm const&) = delete;

 atm& operator=(atm const&) = delete;

public:

 atm(messaging::sender bank_,

  messaging::sender interface_hardware_):

  bank(bank_), interface_hardware(interface_hardware_) {}

 void done() {

  get_sender().send(messaging::close_queue());

 }

 void run() {

  state = &atm::waiting_for_card;

  try {

  for (;;) {

   (this->*state)();

  }

  } catch(messaging::close_queue const&) {

  }

 }

 messaging::sender get_sender() {

  return incoming;

 }

};

Листинг С.8. Конечный автомат банка

class bank_machine {

 messaging::receiver incoming;

 unsigned balance;

public:

 bank_machine():

  balance(199) {}

 void done() {

  get_sender().send(messaging::close_queue());

 }

 void run() {

  try {

  for (;;) {

   incoming.wait().handle(

   [&](verify_pin const& msg) {

    if (msg.pin == "1937") {

    msg.atm_queue.send(pin_verified());

    } else {

    msg.atm_queue.send(pin_incorrect());

    }

   }

   ).handle(

   [&](withdraw const& msg) {

   if (balance >= msg.amount) {

    msg.atm_queue.send(withdraw_ok());

    balance -= msg.amount;

    } else {

    msg.atm_queue.send(withdraw_denied());

    }

   }

   ).handle(

   [&](get_balance const& msg) {

    msg.atm_queue.send(::balance(balance));

   }

   ).handle(

   [&](withdrawal_processed const& msg) {

   }

   ).handle(

   [&](cancel_withdrawal const& msg) {

   }

   );

  }

  } catch(messaging::close_queue const&) {

 }

 }

 messaging::sender get_sender() {

  return incoming;

 }

};

Листинг С.9. Конечный автомат пользовательского интерфейса

class interface_machine {

 messaging::receiver incoming;

public:

 void done() {

  get_sender().send(messaging::close_queue());

 }

 void run() {

  try {

  for (;;) {

   incoming.wait().handle (

   [&](issue_money const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout << "Issuing "

         << msg.amount << std::endl;

    }

   }

   ).handle(

   [&](display_insufficient_funds const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout << "Insufficient funds" << std::endl;

    }

   }

   ).handle(

   [&](display_enter_pin const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout

     << "Please enter your PIN (0-9)" << std::endl;

    }

   }

   ).handle(

   [&](display_enter_card const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout << "Please enter your card (I)"

         << std::endl;

    }

   }

   ).handle(

   [&](display_balance const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout

     << "The balance of your account is "

     << msg.amount << std::endl;

    }

   }

   ).handle(

   [&](display_withdrawal_options const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout << "Withdraw 50? (w)" << std::endl;

    std::cout << "Display Balance? (b)"

         << std::endl;

    std::cout << "Cancel? (c) " << std::endl;

    }

   }

   ).handle(

   [&](display_withdrawal_cancelled const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout << "Withdrawal cancelled"

         << std::endl;

    }

   }

   ).handle(

   [&](display_pin_incorrect_message const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout << "PIN incorrect" << std::endl;

    }

   }

   ).handle(

   [&](eject_card const& msg) {

    {

    std::lock_guard lk(iom);

    std::cout << "Ejecting card" << std::endl;

   }

   }

   );

  }

 } catch (messaging::close_queue&) {

  }

 }

 messaging::sender get_sender() {

  return incoming;

 }

};

Листинг С.10. Управляющая программа

int main() {

 bank_machine bank;

 interface_machine interface_hardware;

 atm machine(bank.get_sender(), interface_hardware.get_sender());

 std::thread bank_thread(&bank_machine::run, &bank);

 std::thread if_thread(&interface_machine::run,

  &interface_hardware);

 std::thread atm_thread(&atm::run, &machine);

 messaging::sender atmqueue(machine.get_sender());

 bool quit_pressed = false;

 while (!quit_pressed) {

  char c = getchar();

  switch(с) {

 case '0':

  case '1':

  case '2':

  case '3':

  case '4':

  case '5':

  case '6':

  case '7':

  case '8':

  case '9':

  atmqueue.send(digit_pressed(с));

  break;

  case 'b':

  atmqueue.send(balance_pressed());

  break;

  case 'w':

  atmqueue.send(withdraw_pressed(50));

  break;

  case 'с':

  atmqueue.send(cancel_pressed());

  break;

  case 'q':

  quit_pressed = true;

  break;

  case 'i':

  atmqueue.send(card_inserted("acc1234"));

  break;

  }

 }

 bank.done();

 machine.done();

 interface_hardware.done();

 atm_thread.join();

 bank_thread.join();

 if_thread.join();

}