Пошаговое руководство. Создание пользовательского блока сообщений
В этом документе описывается, как создать пользовательский тип блока сообщений, который упорядочивает входящие сообщения по приоритету.
Хотя встроенные типы блоков сообщений предоставляют широкий спектр функциональных возможностей, вы можете создать собственный тип блока сообщений и настроить его в соответствии с требованиями приложения. Описание встроенных типов блоков сообщений, предоставляемых библиотекой асинхронных агентов, см. в разделе "Асинхронные блоки сообщений".
Необходимые компоненты
Ознакомьтесь со следующими документами перед началом работы с этим пошаговом руководстве.
Разделы
Это пошаговое руководство содержит следующие разделы:
Проектирование настраиваемого блока сообщений
Блоки сообщений участвуют в процессе отправки и получения сообщений. Блок сообщений, который отправляет сообщения, называется исходным блоком. Блок сообщений, который получает сообщения, называется целевым блоком. Блок сообщений, который отправляет и получает сообщения, называется блоком распространения. Библиотека агентов использует параллелизм абстрактного класса ::ISource для представления исходных блоков и параллелизма абстрактного класса ::ITarget для представления целевых блоков. Типы блоков сообщений, которые действуют как источники, производные от ISource
; типы блоков сообщений, которые действуют в качестве целевых объектов, производных от ITarget
.
Хотя тип блока сообщений можно наследить непосредственно от ISource
и ITarget
из библиотеки агентов определяет три базовых класса, которые выполняют большую часть функциональных возможностей, которые являются общими для всех типов блоков сообщений, например обработка ошибок и соединение блоков сообщений вместе в безопасном режиме параллелизма. Класс параллелизма::source_block является производным от ISource
и отправляет сообщения в другие блоки. Класс параллелизма::target_block является производным от ITarget
и получает сообщения из других блоков. Класс concurrency::p ropagator_block является производным от ISource
и ITarget
отправляет сообщения в другие блоки и получает сообщения из других блоков. Мы рекомендуем использовать эти три базовых класса для обработки сведений о инфраструктуре, чтобы сосредоточиться на поведении блока сообщений.
Классы source_block
target_block
и propagator_block
классы — это шаблоны, параметризованные в типе, который управляет подключениями или ссылками между исходными и целевыми блоками и типом, который управляет обработкой сообщений. Библиотека агентов определяет два типа, которые выполняют управление ссылками, параллелизм::single_link_registry и параллелизм::multi_link_registry. Класс single_link_registry
позволяет связать блок сообщений с одним источником или одним целевым объектом. Класс multi_link_registry
позволяет связать блок сообщений с несколькими источниками или несколькими целевыми объектами. Библиотека агентов определяет один класс, который выполняет управление сообщениями, параллелизм::ordered_message_processor. Класс ordered_message_processor
позволяет блокам сообщений обрабатывать сообщения в порядке их получения.
Чтобы лучше понять, как блоки сообщений связаны с их источниками и целевыми объектами, рассмотрим следующий пример. В этом примере показано объявление класса параллелизма::преобразователя .
template<
class _Input,
class _Output
>
class transformer : public propagator_block<
single_link_registry<ITarget<_Output>>,
multi_link_registry<ISource<_Input>>
>;
Класс transformer
является производным от propagator_block
и, следовательно, выступает как в качестве исходного блока, так и в качестве целевого блока. Он принимает сообщения типа _Input
и отправляет сообщения типа _Output
. Класс transformer
указывает single_link_registry
в качестве диспетчера ссылок для всех целевых блоков и multi_link_registry
в качестве диспетчера ссылок для любых исходных блоков. transformer
Поэтому объект может иметь до одного целевого объекта и неограниченного количества источников.
Класс, производный от source_block
шести методов: propagate_to_any_targets, accept_message, reserve_message, consume_message, release_message и resume_propagation. Класс, производный от target_block
этого, должен реализовать метод propagate_message и при необходимости реализовать метод send_message . Производный от propagator_block
функционально эквивалентен производным от обоих source_block
и target_block
.
Метод propagate_to_any_targets
вызывается средой выполнения для асинхронной или синхронной обработки всех входящих сообщений и распространения всех исходящих сообщений. Метод accept_message
вызывается целевыми блоками для приема сообщений. Многие типы блоков сообщений, например unbounded_buffer
, отправляют сообщения только в первый целевой объект, который получит его. Таким образом, он передает владение сообщением целевому объекту. Другие типы блоков сообщений, такие как параллелизм::overwrite_buffer, предлагают сообщения каждому из его целевых блоков. overwrite_buffer
Поэтому создает копию сообщения для каждого из его целевых объектов.
consume_message
Методы reserve_message
, release_message
и resume_propagation
методы позволяют блокам сообщений участвовать в резервировании сообщений. Целевые блоки вызывают reserve_message
метод, когда они предлагают сообщение и должны зарезервировать сообщение для последующего использования. После того как целевой блок резервирует сообщение, он может вызвать consume_message
метод для использования этого сообщения или release_message
метода для отмены резервирования. Как и в accept_message
случае с методом, реализация consume_message
может передавать владение сообщением или возвращать копию сообщения. После того как целевой блок использует или освобождает зарезервированное сообщение, среда выполнения вызывает resume_propagation
метод. Как правило, этот метод продолжает распространение сообщений, начиная с следующего сообщения в очереди.
Среда выполнения вызывает propagate_message
метод для асинхронной передачи сообщения из другого блока в текущий. Метод send_message
похож propagate_message
, за исключением того, что он синхронно, а не асинхронно отправляет сообщение в целевые блоки. Реализация по умолчанию send_message
отклоняет все входящие сообщения. Среда выполнения не вызывает ни один из этих методов, если сообщение не передает необязательную функцию фильтра, связанную с целевым блоком. Дополнительные сведения о фильтрах сообщений см. в разделе "Асинхронные блоки сообщений".
[В начало]
Определение класса priority_buffer
Класс priority_buffer
— это пользовательский тип блока сообщений, который сначала упорядочивает входящие сообщения по приоритету, а затем по порядку получения сообщений. Класс priority_buffer
напоминает класс параллелизма::unbounded_buffer , так как он содержит очередь сообщений, а также потому, что он действует как в качестве источника, так и целевого блока сообщений, и может иметь как несколько источников, так и несколько целевых объектов. Однако распространение unbounded_buffer
сообщений базируется только в том порядке, в котором он получает сообщения из своих источников.
Класс priority_buffer
получает сообщения типа std::кортежа , содержащие PriorityType
и Type
элементы. PriorityType
относится к типу, который имеет приоритет каждого сообщения; Type
ссылается на часть данных сообщения. Класс priority_buffer
отправляет сообщения типа Type
. Класс priority_buffer
также управляет двумя очередями сообщений: объект std::p riority_queue для входящих сообщений и объект std::queue для исходящих сообщений. Упорядочивание сообщений по приоритету полезно, если priority_buffer
объект получает несколько сообщений одновременно или при получении нескольких сообщений перед чтением сообщений потребителями.
Помимо семи методов, производных от propagator_block
класса, который должен реализовываться, priority_buffer
класс также переопределяет link_target_notification
методы и send_message
методы. Класс priority_buffer
также определяет два открытых вспомогательных метода, enqueue
а dequeue
также частный вспомогательный метод propagate_priority_order
.
В следующей процедуре описывается реализация priority_buffer
класса.
Определение класса priority_buffer
Создайте файл заголовка C++ и назовите его
priority_buffer.h
. Кроме того, можно использовать существующий файл заголовка, который является частью проекта.Добавьте
priority_buffer.h
следующий код.#pragma once #include <agents.h> #include <queue>
std
В пространстве имен определите специализации std::less и std::great, которые действуют на объектах concurrency::message.namespace std { // A specialization of less that tests whether the priority element of a // message is less than the priority element of another message. template<class Type, class PriorityType> struct less<concurrency::message<tuple<PriorityType,Type>>*> { typedef concurrency::message<tuple<PriorityType, Type>> MessageType; bool operator()(const MessageType* left, const MessageType* right) const { // apply operator< to the first element (the priority) // of the tuple payload. return (get<0>(left->payload) < get<0>(right->payload)); } }; // A specialization of less that tests whether the priority element of a // message is greater than the priority element of another message. template<class Type, class PriorityType> struct greater<concurrency::message<tuple<PriorityType, Type>>*> { typedef concurrency::message<std::tuple<PriorityType,Type>> MessageType; bool operator()(const MessageType* left, const MessageType* right) const { // apply operator> to the first element (the priority) // of the tuple payload. return (get<0>(left->payload) > get<0>(right->payload)); } }; }
Класс
priority_buffer
хранитmessage
объекты в объектеpriority_queue
. Эти специализации типов позволяют очереди приоритетов сортировать сообщения в соответствии с их приоритетом. Приоритетом является первый элементtuple
объекта.В пространстве имен объявите
concurrencyex
priority_buffer
класс.namespace concurrencyex { template<class Type, typename PriorityType = int, typename Pr = std::less<message<std::tuple<PriorityType, Type>>*>> class priority_buffer : public concurrency::propagator_block<concurrency::multi_link_registry<concurrency::ITarget<Type>>, concurrency::multi_link_registry<concurrency::ISource<std::tuple<PriorityType, Type>>>> { public: protected: private: }; }
Класс
priority_buffer
является производным отpropagator_block
. Поэтому он может отправлять и получать сообщения. Классpriority_buffer
может иметь несколько целевых объектов, получающих сообщения типаType
. Он также может содержать несколько источников, отправляющих сообщения типаtuple<PriorityType, Type>
.private
В разделеpriority_buffer
класса добавьте следующие переменные-члены.// Stores incoming messages. // The type parameter Pr specifies how to order messages by priority. std::priority_queue< concurrency::message<_Source_type>*, std::vector<concurrency::message<_Source_type>*>, Pr > _input_messages; // Synchronizes access to the input message queue. concurrency::critical_section _input_lock; // Stores outgoing messages. std::queue<concurrency::message<_Target_type>*> _output_messages;
Объект
priority_queue
содержит входящие сообщения;queue
объект содержит исходящие сообщения.priority_buffer
Объект может одновременно получать несколько сообщений;critical_section
объект синхронизирует доступ к очереди входных сообщений.В разделе определите
private
конструктор копирования и оператор назначения. Это предотвращаетpriority_queue
назначение объектов.// Hide assignment operator and copy constructor. priority_buffer const &operator =(priority_buffer const&); priority_buffer(priority_buffer const &);
public
В разделе определите конструкторы, которые являются общими для многих типов блоков сообщений. Также определите деструктор.// Constructs a priority_buffer message block. priority_buffer() { initialize_source_and_target(); } // Constructs a priority_buffer message block with the given filter function. priority_buffer(filter_method const& filter) { initialize_source_and_target(); register_filter(filter); } // Constructs a priority_buffer message block that uses the provided // Scheduler object to propagate messages. priority_buffer(concurrency::Scheduler& scheduler) { initialize_source_and_target(&scheduler); } // Constructs a priority_buffer message block with the given filter function // and uses the provided Scheduler object to propagate messages. priority_buffer(concurrency::Scheduler& scheduler, filter_method const& filter) { initialize_source_and_target(&scheduler); register_filter(filter); } // Constructs a priority_buffer message block that uses the provided // SchedulerGroup object to propagate messages. priority_buffer(concurrency::ScheduleGroup& schedule_group) { initialize_source_and_target(NULL, &schedule_group); } // Constructs a priority_buffer message block with the given filter function // and uses the provided SchedulerGroup object to propagate messages. priority_buffer(concurrency::ScheduleGroup& schedule_group, filter_method const& filter) { initialize_source_and_target(NULL, &schedule_group); register_filter(filter); } // Destroys the message block. ~priority_buffer() { // Remove all links. remove_network_links(); }
В разделе определите
public
методыenqueue
иdequeue
. Эти вспомогательные методы предоставляют альтернативный способ отправки сообщений и получения сообщений отpriority_buffer
объекта.// Sends an item to the message block. bool enqueue(Type const& item) { return concurrency::asend<Type>(this, item); } // Receives an item from the message block. Type dequeue() { return receive<Type>(this); }
В разделе определите
protected
propagate_to_any_targets
метод.// Transfers the message at the front of the input queue to the output queue // and propagates out all messages in the output queue. virtual void propagate_to_any_targets(concurrency::message<_Target_type>*) { // Retrieve the message from the front of the input queue. concurrency::message<_Source_type>* input_message = NULL; { concurrency::critical_section::scoped_lock lock(_input_lock); if (_input_messages.size() > 0) { input_message = _input_messages.top(); _input_messages.pop(); } } // Move the message to the output queue. if (input_message != NULL) { // The payload of the output message does not contain the // priority of the message. concurrency::message<_Target_type>* output_message = new concurrency::message<_Target_type>(get<1>(input_message->payload)); _output_messages.push(output_message); // Free the memory for the input message. delete input_message; // Do not propagate messages if the new message is not the head message. // In this case, the head message is reserved by another message block. if (_output_messages.front()->msg_id() != output_message->msg_id()) { return; } } // Propagate out the output messages. propagate_priority_order(); }
Метод
propagate_to_any_targets
передает сообщение, которое находится в передней части входной очереди в выходную очередь и распространяет все сообщения в выходной очереди.В разделе определите
protected
accept_message
метод.// Accepts a message that was offered by this block by transferring ownership // to the caller. virtual concurrency::message<_Target_type>* accept_message(concurrency::runtime_object_identity msg_id) { concurrency::message<_Target_type>* message = NULL; // Transfer ownership if the provided message identifier matches // the identifier of the front of the output message queue. if (!_output_messages.empty() && _output_messages.front()->msg_id() == msg_id) { message = _output_messages.front(); _output_messages.pop(); } return message; }
Когда целевой блок вызывает
accept_message
метод,priority_buffer
класс передает владение сообщением первому целевому блоку, который принимает его. (Это напоминает поведениеunbounded_buffer
.)В разделе определите
protected
reserve_message
метод.// Reserves a message that was previously offered by this block. virtual bool reserve_message(concurrency::runtime_object_identity msg_id) { // Allow the message to be reserved if the provided message identifier // is the message identifier of the front of the message queue. return (!_output_messages.empty() && _output_messages.front()->msg_id() == msg_id); }
Класс
priority_buffer
позволяет целевому блоку зарезервировать сообщение, если предоставленный идентификатор сообщения соответствует идентификатору сообщения, расположенного перед очередью. Другими словами, целевой объект может зарезервировать сообщение, еслиpriority_buffer
объект еще не получил дополнительное сообщение и еще не распространил текущий.В разделе определите
protected
consume_message
метод.// Transfers the message that was previously offered by this block // to the caller. The caller of this method is the target block that // reserved the message. virtual concurrency::message<Type>* consume_message(concurrency::runtime_object_identity msg_id) { // Transfer ownership of the message to the caller. return accept_message(msg_id); }
Целевой блок вызывает
consume_message
передачу собственности на сообщение, зарезервированное им.В разделе определите
protected
release_message
метод.// Releases a previous message reservation. virtual void release_message(concurrency::runtime_object_identity msg_id) { // The head message must be the one that is reserved. if (_output_messages.empty() || _output_messages.front()->msg_id() != msg_id) { throw message_not_found(); } }
Целевой блок вызывает
release_message
отмену резервирования сообщения.В разделе определите
protected
resume_propagation
метод.// Resumes propagation after a reservation has been released. virtual void resume_propagation() { // Propagate out any messages in the output queue. if (_output_messages.size() > 0) { async_send(NULL); } }
Среда выполнения вызывается
resume_propagation
после использования целевого блока или выпуска зарезервированного сообщения. Этот метод распространяет все сообщения, находящиеся в выходной очереди.В разделе определите
protected
link_target_notification
метод.// Notifies this block that a new target has been linked to it. virtual void link_target_notification(concurrency::ITarget<_Target_type>*) { // Do not propagate messages if a target block reserves // the message at the front of the queue. if (_M_pReservedFor != NULL) { return; } // Propagate out any messages that are in the output queue. propagate_priority_order(); }
Переменная-член
_M_pReservedFor
определяется базовым классомsource_block
. Эта переменная-член указывает на целевой блок, который содержит резервирование к сообщению, которое находится в передней части очереди выходных данных. Среда выполнения вызываетlink_target_notification
, когда новый целевой объект связан сpriority_buffer
объектом. Этот метод распространяет все сообщения, находящиеся в выходной очереди, если целевой объект не содержит резервирования.В разделе определите
private
propagate_priority_order
метод.// Propagates messages in priority order. void propagate_priority_order() { // Cancel propagation if another block reserves the head message. if (_M_pReservedFor != NULL) { return; } // Propagate out all output messages. // Because this block preserves message ordering, stop propagation // if any of the messages are not accepted by a target block. while (!_output_messages.empty()) { // Get the next message. concurrency::message<_Target_type> * message = _output_messages.front(); concurrency::message_status status = declined; // Traverse each target in the order in which they are connected. for (target_iterator iter = _M_connectedTargets.begin(); *iter != NULL; ++iter) { // Propagate the message to the target. concurrency::ITarget<_Target_type>* target = *iter; status = target->propagate(message, this); // If the target accepts the message then ownership of message has // changed. Do not propagate this message to any other target. if (status == accepted) { break; } // If the target only reserved this message, we must wait until the // target accepts the message. if (_M_pReservedFor != NULL) { break; } } // If status is anything other than accepted, then the head message // was not propagated out. To preserve the order in which output // messages are propagated, we must stop propagation until the head // message is accepted. if (status != accepted) { break; } } }
Этот метод распространяет все сообщения из выходной очереди. Каждое сообщение в очереди предлагается каждому целевому блоку, пока один из целевых блоков не принимает сообщение. Класс
priority_buffer
сохраняет порядок исходящих сообщений. Поэтому первое сообщение в выходной очереди должно приниматься целевым блоком, прежде чем этот метод предлагает любое другое сообщение целевым блокам.В разделе определите
protected
propagate_message
метод.// Asynchronously passes a message from an ISource block to this block. // This method is typically called by propagator_block::propagate. virtual concurrency::message_status propagate_message(concurrency::message<_Source_type>* message, concurrency::ISource<_Source_type>* source) { // Accept the message from the source block. message = source->accept(message->msg_id(), this); if (message != NULL) { // Insert the message into the input queue. The type parameter Pr // defines how to order messages by priority. { concurrency::critical_section::scoped_lock lock(_input_lock); _input_messages.push(message); } // Asynchronously send the message to the target blocks. async_send(NULL); return accepted; } else { return missed; } }
Метод
propagate_message
позволяетpriority_buffer
классу выступать в качестве приемника сообщений или целевого объекта. Этот метод получает сообщение, предлагаемое предоставленным блоком источника, и вставляет это сообщение в очередь приоритета. Затемpropagate_message
метод асинхронно отправляет все выходные сообщения в целевые блоки.Среда выполнения вызывает этот метод при вызове функции параллелизма::asend или при подключении блока сообщений к другим блокам сообщений.
В разделе определите
protected
send_message
метод.// Synchronously passes a message from an ISource block to this block. // This method is typically called by propagator_block::send. virtual concurrency::message_status send_message(concurrency::message<_Source_type>* message, concurrency::ISource<_Source_type>* source) { // Accept the message from the source block. message = source->accept(message->msg_id(), this); if (message != NULL) { // Insert the message into the input queue. The type parameter Pr // defines how to order messages by priority. { concurrency::critical_section::scoped_lock lock(_input_lock); _input_messages.push(message); } // Synchronously send the message to the target blocks. sync_send(NULL); return accepted; } else { return missed; } }
Метод
send_message
похожpropagate_message
на . Однако он отправляет выходные сообщения синхронно, а не асинхронно.Среда выполнения вызывает этот метод во время синхронной операции отправки, например при вызове функции параллелизма::send .
Класс priority_buffer
содержит перегрузки конструктора, типичные во многих типах блоков сообщений. Некоторые перегрузки конструктора принимают параллелизм::Scheduler или параллелизм::ScheduleGroup, которые позволяют блоку сообщений управлять определенным планировщиком задач. Другие перегрузки конструктора принимают функцию фильтра. Функции фильтрации позволяют блокам сообщений принимать или отклонять сообщение на основе полезных данных. Дополнительные сведения о фильтрах сообщений см. в разделе "Асинхронные блоки сообщений". Дополнительные сведения о планировщиках задач см. в разделе Планировщик задач.
priority_buffer
Так как класс упорядочивает сообщения по приоритету, а затем по порядку получения сообщений, этот класс наиболее полезен при асинхронном получении сообщений, например при вызове функции параллелизма::asend или при подключении блока сообщений к другим блокам сообщений.
[В начало]
Полный пример
В следующем примере показано полное определение priority_buffer
класса.
// priority_buffer.h
#pragma once
#include <agents.h>
#include <queue>
namespace std
{
// A specialization of less that tests whether the priority element of a
// message is less than the priority element of another message.
template<class Type, class PriorityType>
struct less<concurrency::message<tuple<PriorityType,Type>>*>
{
typedef concurrency::message<tuple<PriorityType, Type>> MessageType;
bool operator()(const MessageType* left, const MessageType* right) const
{
// apply operator< to the first element (the priority)
// of the tuple payload.
return (get<0>(left->payload) < get<0>(right->payload));
}
};
// A specialization of less that tests whether the priority element of a
// message is greater than the priority element of another message.
template<class Type, class PriorityType>
struct greater<concurrency::message<tuple<PriorityType, Type>>*>
{
typedef concurrency::message<std::tuple<PriorityType,Type>> MessageType;
bool operator()(const MessageType* left, const MessageType* right) const
{
// apply operator> to the first element (the priority)
// of the tuple payload.
return (get<0>(left->payload) > get<0>(right->payload));
}
};
}
namespace concurrencyex
{
// A message block type that orders incoming messages first by priority,
// and then by the order in which messages are received.
template<class Type,
typename PriorityType = int,
typename Pr = std::less<message<std::tuple<PriorityType, Type>>*>>
class priority_buffer : public concurrency::propagator_block<concurrency::multi_link_registry<concurrency::ITarget<Type>>,
concurrency::multi_link_registry<concurrency::ISource<std::tuple<PriorityType, Type>>>>
{
public:
// Constructs a priority_buffer message block.
priority_buffer()
{
initialize_source_and_target();
}
// Constructs a priority_buffer message block with the given filter function.
priority_buffer(filter_method const& filter)
{
initialize_source_and_target();
register_filter(filter);
}
// Constructs a priority_buffer message block that uses the provided
// Scheduler object to propagate messages.
priority_buffer(concurrency::Scheduler& scheduler)
{
initialize_source_and_target(&scheduler);
}
// Constructs a priority_buffer message block with the given filter function
// and uses the provided Scheduler object to propagate messages.
priority_buffer(concurrency::Scheduler& scheduler, filter_method const& filter)
{
initialize_source_and_target(&scheduler);
register_filter(filter);
}
// Constructs a priority_buffer message block that uses the provided
// SchedulerGroup object to propagate messages.
priority_buffer(concurrency::ScheduleGroup& schedule_group)
{
initialize_source_and_target(NULL, &schedule_group);
}
// Constructs a priority_buffer message block with the given filter function
// and uses the provided SchedulerGroup object to propagate messages.
priority_buffer(concurrency::ScheduleGroup& schedule_group, filter_method const& filter)
{
initialize_source_and_target(NULL, &schedule_group);
register_filter(filter);
}
// Destroys the message block.
~priority_buffer()
{
// Remove all links.
remove_network_links();
}
// Sends an item to the message block.
bool enqueue(Type const& item)
{
return concurrency::asend<Type>(this, item);
}
// Receives an item from the message block.
Type dequeue()
{
return receive<Type>(this);
}
protected:
// Asynchronously passes a message from an ISource block to this block.
// This method is typically called by propagator_block::propagate.
virtual concurrency::message_status propagate_message(concurrency::message<_Source_type>* message,
concurrency::ISource<_Source_type>* source)
{
// Accept the message from the source block.
message = source->accept(message->msg_id(), this);
if (message != NULL)
{
// Insert the message into the input queue. The type parameter Pr
// defines how to order messages by priority.
{
concurrency::critical_section::scoped_lock lock(_input_lock);
_input_messages.push(message);
}
// Asynchronously send the message to the target blocks.
async_send(NULL);
return accepted;
}
else
{
return missed;
}
}
// Synchronously passes a message from an ISource block to this block.
// This method is typically called by propagator_block::send.
virtual concurrency::message_status send_message(concurrency::message<_Source_type>* message,
concurrency::ISource<_Source_type>* source)
{
// Accept the message from the source block.
message = source->accept(message->msg_id(), this);
if (message != NULL)
{
// Insert the message into the input queue. The type parameter Pr
// defines how to order messages by priority.
{
concurrency::critical_section::scoped_lock lock(_input_lock);
_input_messages.push(message);
}
// Synchronously send the message to the target blocks.
sync_send(NULL);
return accepted;
}
else
{
return missed;
}
}
// Accepts a message that was offered by this block by transferring ownership
// to the caller.
virtual concurrency::message<_Target_type>* accept_message(concurrency::runtime_object_identity msg_id)
{
concurrency::message<_Target_type>* message = NULL;
// Transfer ownership if the provided message identifier matches
// the identifier of the front of the output message queue.
if (!_output_messages.empty() &&
_output_messages.front()->msg_id() == msg_id)
{
message = _output_messages.front();
_output_messages.pop();
}
return message;
}
// Reserves a message that was previously offered by this block.
virtual bool reserve_message(concurrency::runtime_object_identity msg_id)
{
// Allow the message to be reserved if the provided message identifier
// is the message identifier of the front of the message queue.
return (!_output_messages.empty() &&
_output_messages.front()->msg_id() == msg_id);
}
// Transfers the message that was previously offered by this block
// to the caller. The caller of this method is the target block that
// reserved the message.
virtual concurrency::message<Type>* consume_message(concurrency::runtime_object_identity msg_id)
{
// Transfer ownership of the message to the caller.
return accept_message(msg_id);
}
// Releases a previous message reservation.
virtual void release_message(concurrency::runtime_object_identity msg_id)
{
// The head message must be the one that is reserved.
if (_output_messages.empty() ||
_output_messages.front()->msg_id() != msg_id)
{
throw message_not_found();
}
}
// Resumes propagation after a reservation has been released.
virtual void resume_propagation()
{
// Propagate out any messages in the output queue.
if (_output_messages.size() > 0)
{
async_send(NULL);
}
}
// Notifies this block that a new target has been linked to it.
virtual void link_target_notification(concurrency::ITarget<_Target_type>*)
{
// Do not propagate messages if a target block reserves
// the message at the front of the queue.
if (_M_pReservedFor != NULL)
{
return;
}
// Propagate out any messages that are in the output queue.
propagate_priority_order();
}
// Transfers the message at the front of the input queue to the output queue
// and propagates out all messages in the output queue.
virtual void propagate_to_any_targets(concurrency::message<_Target_type>*)
{
// Retrieve the message from the front of the input queue.
concurrency::message<_Source_type>* input_message = NULL;
{
concurrency::critical_section::scoped_lock lock(_input_lock);
if (_input_messages.size() > 0)
{
input_message = _input_messages.top();
_input_messages.pop();
}
}
// Move the message to the output queue.
if (input_message != NULL)
{
// The payload of the output message does not contain the
// priority of the message.
concurrency::message<_Target_type>* output_message =
new concurrency::message<_Target_type>(get<1>(input_message->payload));
_output_messages.push(output_message);
// Free the memory for the input message.
delete input_message;
// Do not propagate messages if the new message is not the head message.
// In this case, the head message is reserved by another message block.
if (_output_messages.front()->msg_id() != output_message->msg_id())
{
return;
}
}
// Propagate out the output messages.
propagate_priority_order();
}
private:
// Propagates messages in priority order.
void propagate_priority_order()
{
// Cancel propagation if another block reserves the head message.
if (_M_pReservedFor != NULL)
{
return;
}
// Propagate out all output messages.
// Because this block preserves message ordering, stop propagation
// if any of the messages are not accepted by a target block.
while (!_output_messages.empty())
{
// Get the next message.
concurrency::message<_Target_type> * message = _output_messages.front();
concurrency::message_status status = declined;
// Traverse each target in the order in which they are connected.
for (target_iterator iter = _M_connectedTargets.begin();
*iter != NULL;
++iter)
{
// Propagate the message to the target.
concurrency::ITarget<_Target_type>* target = *iter;
status = target->propagate(message, this);
// If the target accepts the message then ownership of message has
// changed. Do not propagate this message to any other target.
if (status == accepted)
{
break;
}
// If the target only reserved this message, we must wait until the
// target accepts the message.
if (_M_pReservedFor != NULL)
{
break;
}
}
// If status is anything other than accepted, then the head message
// was not propagated out. To preserve the order in which output
// messages are propagated, we must stop propagation until the head
// message is accepted.
if (status != accepted)
{
break;
}
}
}
private:
// Stores incoming messages.
// The type parameter Pr specifies how to order messages by priority.
std::priority_queue<
concurrency::message<_Source_type>*,
std::vector<concurrency::message<_Source_type>*>,
Pr
> _input_messages;
// Synchronizes access to the input message queue.
concurrency::critical_section _input_lock;
// Stores outgoing messages.
std::queue<concurrency::message<_Target_type>*> _output_messages;
private:
// Hide assignment operator and copy constructor.
priority_buffer const &operator =(priority_buffer const&);
priority_buffer(priority_buffer const &);
};
}
В следующем примере параллельно выполняется несколько операций asend
параллелизма ::receive в объекте priority_buffer
.
// priority_buffer.cpp
// compile with: /EHsc
#include <ppl.h>
#include <iostream>
#include "priority_buffer.h"
using namespace concurrency;
using namespace concurrencyex;
using namespace std;
int wmain()
{
// Concurrently perform a number of asend and receive operations
// on a priority_buffer object.
priority_buffer<int> pb;
parallel_invoke(
[&pb] { for (int i = 0; i < 25; ++i) asend(pb, make_tuple(2, 36)); },
[&pb] { for (int i = 0; i < 25; ++i) asend(pb, make_tuple(0, 12)); },
[&pb] { for (int i = 0; i < 25; ++i) asend(pb, make_tuple(1, 24)); },
[&pb] {
for (int i = 0; i < 75; ++i) {
wcout << receive(pb) << L' ';
if ((i+1) % 25 == 0)
wcout << endl;
}
}
);
}
В этом примере создается следующий пример выходных данных.
36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36 36
24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24 24
12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12 12
priority_buffer
Класс упорядочивает сообщения сначала по приоритету, а затем по порядку, в котором он получает сообщения. В этом примере сообщения с более высоким числовым приоритетом вставляются в передней части очереди.
[В начало]
Компиляция кода
Скопируйте пример кода и вставьте его в проект Visual Studio или вставьте определение priority_buffer
класса в файл с именем priority_buffer.h
и тестовой программой в файле, который называется priority_buffer.cpp
, а затем выполните следующую команду в окне командной строки Visual Studio.
cl.exe /EHsc priority_buffer.cpp
См. также
Пошаговые руководства по среде выполнения с параллелизмом
Асинхронные блоки сообщений
Функции передачи сообщений