Line data Source code
1 0 : // Distributed under the MIT License.
2 : // See LICENSE.txt for details.
3 :
4 : #pragma once
5 :
6 : #include <optional>
7 : #include <ostream>
8 : #include <pup.h>
9 : #include <type_traits>
10 : #include <unordered_map>
11 : #include <utility>
12 :
13 : #include "DataStructures/LinkedMessageId.hpp"
14 : #include "DataStructures/TaggedTuple.hpp"
15 : #include "Utilities/ErrorHandling/Assert.hpp"
16 : #include "Utilities/Serialization/PupStlCpp17.hpp"
17 : #include "Utilities/TMPL.hpp"
18 :
19 : /// \cond
20 : template <typename Id, typename QueueTags>
21 : class LinkedMessageQueue;
22 : /// \endcond
23 :
24 : /// \ingroup DataStructuresGroup
25 : /// A collection of queues of asynchronously received data
26 : ///
27 : /// This class is designed to collect asynchronously received
28 : /// messages, possibly from multiple sources, until sufficient data
29 : /// has been received for processing. Each tag in \p QueueTags
30 : /// defines a queue of messages of type `Tag::type`.
31 : template <typename Id, typename... QueueTags>
32 1 : class LinkedMessageQueue<Id, tmpl::list<QueueTags...>> {
33 : static_assert(sizeof...(QueueTags) > 0,
34 : "Must have at least one message source.");
35 :
36 : public:
37 0 : using IdType = Id;
38 0 : using queue_tags_list = tmpl::list<QueueTags...>;
39 :
40 : /// Insert data into a given queue at a given ID. All queues must
41 : /// receive data with the same collection of \p id_and_previous, but
42 : /// are not required to receive them in the same order.
43 : template <typename Tag>
44 1 : void insert(const LinkedMessageId<Id>& id_and_previous,
45 : typename Tag::type message);
46 :
47 : /// Insert multiple data at once into a given queue at a given ID. All queues
48 : /// must receive data with the same collection of \p id_and_previous, but are
49 : /// not required to receive them in the same order.
50 : ///
51 : /// \details Tags are inserted in the order they are passed in. Duplicate tags
52 : /// are not allowed.
53 : template <typename Tag1, typename Tag2, typename... Tags>
54 1 : void insert(const LinkedMessageId<Id>& id_and_previous,
55 : typename Tag1::type message_1, typename Tag2::type message_2,
56 : typename Tags::type... messages);
57 :
58 : /// The next ID in the received sequence, if all queues have
59 : /// received messages at that ID.
60 1 : std::optional<Id> next_ready_id() const;
61 :
62 : /// All the messages received at the time indicated by
63 : /// `next_ready_id()`. It is an error to call this function if
64 : /// `next_ready_id()` would return an empty optional. This function
65 : /// removes the stored messages from the queues, advancing the
66 : /// internal sequence ID.
67 1 : tuples::TaggedTuple<QueueTags...> extract();
68 :
69 : // NOLINTNEXTLINE(google-runtime-references)
70 0 : void pup(PUP::er& p) {
71 : p | previous_id_;
72 : p | messages_;
73 : }
74 :
75 : private:
76 : template <typename Tag>
77 0 : struct Optional {
78 0 : using type = std::optional<typename Tag::type>;
79 : };
80 :
81 0 : using OptionalTuple = tuples::TaggedTuple<Optional<QueueTags>...>;
82 :
83 0 : std::optional<Id> previous_id_{};
84 : std::unordered_map<std::optional<Id>, std::pair<Id, OptionalTuple>>
85 0 : messages_{};
86 : };
87 :
88 : template <typename Id, typename... QueueTags>
89 : template <typename Tag>
90 : void LinkedMessageQueue<Id, tmpl::list<QueueTags...>>::insert(
91 : const LinkedMessageId<Id>& id_and_previous, typename Tag::type message) {
92 : static_assert((... or std::is_same_v<Tag, QueueTags>), "Unrecognized tag.");
93 : const auto entry =
94 : messages_
95 : .insert({id_and_previous.previous,
96 : std::pair{id_and_previous.id, OptionalTuple{}}})
97 : .first;
98 : auto& [id, tuple] = entry->second;
99 : ASSERT(id_and_previous.id == id, "Received messages with different ids ("
100 : << id << " and " << id_and_previous.id
101 : << ") but the same previous id ("
102 : << id_and_previous.previous << ").");
103 : ASSERT(not tuples::get<Optional<Tag>>(tuple).has_value(),
104 : "Received duplicate messages at id "
105 : << id << " and previous id " << id_and_previous.previous << ".");
106 : tuples::get<Optional<Tag>>(tuple) = std::move(message);
107 : }
108 :
109 : template <typename Id, typename... QueueTags>
110 : template <typename Tag1, typename Tag2, typename... Tags>
111 : void LinkedMessageQueue<Id, tmpl::list<QueueTags...>>::insert(
112 : const LinkedMessageId<Id>& id_and_previous, typename Tag1::type message_1,
113 : typename Tag2::type message_2, typename Tags::type... messages) {
114 : static_assert(
115 : tmpl::size<
116 : tmpl::remove_duplicates<tmpl::list<Tag1, Tag2, Tags...>>>::value ==
117 : sizeof...(Tags) + 2,
118 : "Must have unique tags in LinkedMessageQueue insert.");
119 : insert<Tag1>(id_and_previous, std::move(message_1));
120 : insert<Tag2>(id_and_previous, std::move(message_2));
121 :
122 : [[maybe_unused]] const auto insert_pack =
123 : [this, &id_and_previous](const auto& tag_v, auto message) {
124 : (void)this;
125 : using tag = std::decay_t<decltype(tag_v)>;
126 : insert<tag>(id_and_previous, std::move(message));
127 : };
128 :
129 : EXPAND_PACK_LEFT_TO_RIGHT(insert_pack(Tags{}, std::move(messages)));
130 : }
131 :
132 : template <typename Id, typename... QueueTags>
133 : std::optional<Id>
134 : LinkedMessageQueue<Id, tmpl::list<QueueTags...>>::next_ready_id() const {
135 : const auto current_entry = messages_.find(previous_id_);
136 : if (current_entry == messages_.end()) {
137 : return {};
138 : }
139 : const auto& current_value = current_entry->second;
140 : const auto& [current_id, current_messages] = current_value;
141 : if ((... and
142 : tuples::get<Optional<QueueTags>>(current_messages).has_value())) {
143 : return current_id;
144 : } else {
145 : return {};
146 : }
147 : }
148 :
149 : template <typename Id, typename... QueueTags>
150 : tuples::TaggedTuple<QueueTags...>
151 : LinkedMessageQueue<Id, tmpl::list<QueueTags...>>::extract() {
152 : ASSERT(next_ready_id().has_value(),
153 : "Cannot extract before all messages have been received.");
154 : const auto current_entry = messages_.find(previous_id_);
155 : auto& current_value = current_entry->second;
156 : auto& [current_id, current_messages] = current_value;
157 : tuples::TaggedTuple<QueueTags...> result{};
158 : (void)(..., (tuples::get<QueueTags>(result) = std::move(
159 : *tuples::get<Optional<QueueTags>>(current_messages))));
160 : previous_id_ = current_id;
161 : messages_.erase(current_entry);
162 : return result;
163 : }
|
