GCC Code Coverage Report

Directory: ./
File: src/msrproto/PeriodicSubscriptions.cpp
Date: 2026-11-04 12:22:15
Exec Total Coverage
Lines: 199 244 81.6%
Branches: 184 384 47.9%
Line Branch Exec Source
1 /*****************************************************************************
2 *
3 * Copyright (C) 2021 Bjarne von Horn (vh at igh dot de)
4 *
5 * This file is part of the PdCom library.
6 *
7 * The PdCom library is free software: you can redistribute it and/or modify
8 * it under the terms of the GNU Lesser General Public License as published by
9 * the Free Software Foundation, either version 3 of the License, or (at your
10 * option) any later version.
11 *
12 * The PdCom library is distributed in the hope that it will be useful, but
13 * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
14 * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
15 * License for more details.
16 *
17 * You should have received a copy of the GNU Lesser General Public License
18 * along with the PdCom library. If not, see <http://www.gnu.org/licenses/>.
19 *
20 *****************************************************************************/
21
22 #include "PeriodicSubscriptions.h"
23
24 #include "Channel.h"
25 #include "Subscription.h"
26
27 #include <cassert>
28 #include <cstring>
29 #include <pdcom5/Exception.h>
30 #include <sstream>
31
32 using PdCom::impl::MsrProto::DataDecoder;
33 using PdCom::impl::MsrProto::PeriodicSubscription;
34 using PdCom::impl::MsrProto::PeriodicSubscriptionsBase;
35 using PdCom::impl::MsrProto::PeriodicSubscriptionsWithGroup;
36 using PdCom::impl::MsrProto::PeriodicSubscriptionWithoutGroup;
37
38 using PdCom::impl::MsrProto::Variable;
39
40
41 52 PeriodicSubscriptionsWithGroup::transaction PeriodicSubscriptionsWithGroup::add(
42 PdCom::impl::Process::PendingCallbackQueue &queue,
43 bool notify_pending,
44 const std::shared_ptr<PeriodicSubscription> &s)
45 {
46 104 const auto var_ptr = s->variable_.lock();
47
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
 52 if (!var_ptr)
48 ✗ throw EmptyVariable();
49 52 const auto &var = static_cast<Channel const &>(*var_ptr);
50
1/2
✓ Branch 11 taken 52 times.
✗ Branch 12 not taken.
 52 const auto group_id = map_.find(s->subscriber_.getTransmission());
51
2/4
✓ Branch 3 taken 52 times.
✗ Branch 4 not taken.
✓ Branch 7 taken 52 times.
✗ Branch 8 not taken.
 104 auto map_it = map_.at(group_id).find(var.index_);
52
3/4
✓ Branch 3 taken 52 times.
✗ Branch 4 not taken.
✓ Branch 10 taken 41 times.
✓ Branch 11 taken 11 times.
 52 if (map_it == map_.at(group_id).end()) {
53 // not yet subscribed
54 41 unsigned decimation = s->subscriber_.getTransmission().getInterval()
55 41 / var.sample_time_.count()
56 41 + 0.5;
57 82 decimation += s->subscriber_.getTransmission().getInterval() > 0.0
58
3/4
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 39 times.
 41 and !decimation;
59 41 unsigned blocksize = 0;
60
61
1/2
✓ Branch 0 taken 41 times.
✗ Branch 1 not taken.
 41 if (decimation > 0) {
62 41 blocksize =
63 41 1.0 / var.sample_time_.count() / decimation / 25.0 + 0.5;
64
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 41 times.
 41 blocksize = std::min(var.bufsize / decimation, blocksize);
65 41 blocksize = std::min(blocksize, 1000U);
66 }
67
68
2/2
✓ Branch 0 taken 38 times.
✓ Branch 1 taken 3 times.
 41 if (!blocksize)
69 38 blocksize = 1;
70
1/2
✓ Branch 3 taken 41 times.
✗ Branch 4 not taken.
 82 time_vector_.resize(
71 82 std::max<size_t>(time_vector_.size(), blocksize + 2));
72
73
6/12
✓ Branch 3 taken 41 times.
✗ Branch 4 not taken.
✓ Branch 14 taken 41 times.
✗ Branch 15 not taken.
✓ Branch 18 taken 41 times.
✗ Branch 19 not taken.
✓ Branch 22 taken 41 times.
✗ Branch 23 not taken.
✓ Branch 33 taken 41 times.
✓ Branch 34 taken 41 times.
✗ Branch 45 not taken.
✗ Branch 46 not taken.
 41 map_.at(group_id).insert({var.index_, {blocksize, decimation, {{s}}}});
74
75
2/2
✓ Branch 0 taken 25 times.
✓ Branch 1 taken 16 times.
 41 if (notify_pending)
76
1/2
✓ Branch 8 taken 25 times.
✗ Branch 9 not taken.
 25 queue.push_back({s, PdCom::Subscription::State::Pending});
77 41 return {true, group_id, decimation, blocksize};
78 }
79 else {
80
2/2
✓ Branch 0 taken 10 times.
✓ Branch 1 taken 1 times.
 11 if (notify_pending
81
4/4
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 4 times.
✓ Branch 4 taken 6 times.
✓ Branch 5 taken 1 times.
 11 or map_it->second.subscriptions_.currentState
82 != PdCom::Subscription::State::Pending)
83
1/2
✓ Branch 10 taken 10 times.
✗ Branch 11 not taken.
 10 queue.push_back({s, map_it->second.subscriptions_.currentState});
84
1/2
✓ Branch 6 taken 11 times.
✗ Branch 7 not taken.
 11 map_it->second.subscriptions_.insert(s);
85 11 return {false, group_id, map_it->second.decimation_,
86 22 map_it->second.blocksize_};
87 }
88 }
89
90 PeriodicSubscriptionsWithGroup::transaction
91 27 PeriodicSubscriptionsWithGroup::remove(
92 const Variable &var,
93 PdCom::Transmission tm)
94 {
95 // remove'ing is okay with locked ps,
96 // because the subscription map is _not_ deleted
97 // and the forward map _not_ altered. So pointers stay valid.
98 27 const auto group_id = map_.erase(tm, var.index_);
99 27 return {group_id != 0, group_id, 0, 0};
100 }
101
102 53 PdCom::impl::MsrProto::PeriodicSubscriptionsBase::DataReceiveHandle::
103 DataReceiveHandle(
104 PeriodicSubscriptionsBase &ps,
105 ChannelSubscriptionMap &subscription_map,
106 VariableCache::Lock variable_cache_lock,
107 std::chrono::nanoseconds ts,
108 53 bool consistent_blocksize) :
109 ps_(&ps),
110 53 variable_cache_lock_(std::move(variable_cache_lock)),
111 subscription_map_(&subscription_map),
112 ts_(ts),
113 106 consistent_blocksize_(consistent_blocksize)
114 {
115
2/4
✗ Branch 1 not taken.
✓ Branch 2 taken 53 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 53 times.
 53 if (ps.locked_)
116 throw PdCom::InternalError(
117 ✗ "Receiving periodic data already in progress");
118 53 ps.locked_ = true;
119 53 }
120
121 55 void PeriodicSubscriptionsBase::DataReceiveHandle::readData(
122 const PdCom::impl::MsrProto::Channel &c,
123 const char *data)
124 {
125
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 55 times.
 55 if (!ps_) {
126 ✗ throw PdCom::InternalError("Operating on invalid DataReceiveHandle");
127 }
128
1/2
✓ Branch 2 taken 55 times.
✗ Branch 3 not taken.
 55 const Base64Info bi(data);
129
130
3/4
✓ Branch 4 taken 55 times.
✗ Branch 5 not taken.
✓ Branch 8 taken 47 times.
✓ Branch 9 taken 8 times.
 102 const auto sub_info_it = subscription_map_->find(c.index_);
131
2/2
✓ Branch 7 taken 8 times.
✓ Branch 8 taken 47 times.
 55 if (sub_info_it == subscription_map_->end())
132 8 return;
133
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 47 times.
 94 if (sub_info_it->second.subscriptions_.currentState
134 47 != PdCom::Subscription::State::Active)
135 ✗ return;
136
3/4
✗ Branch 1 not taken.
✓ Branch 2 taken 47 times.
✓ Branch 3 taken 7 times.
✓ Branch 4 taken 40 times.
 47 if (!consistent_blocksize_) {
137 7 current_blocksize_ = std::min<int>(
138 14 sub_info_it->second.blocksize_,
139
1/2
✗ Branch 5 not taken.
✓ Branch 6 taken 7 times.
 21 bi.encodedDataLength_ / c.type_info->element_size);
140 7 max_read_blocks_ = std::max(max_read_blocks_, current_blocksize_);
141 }
142
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 40 times.
 40 else if (current_blocksize_ == -1)
143 ✗ current_blocksize_ = sub_info_it->second.blocksize_;
144
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 40 times.
 40 else if (
145 40 current_blocksize_
146 40 != static_cast<int>(sub_info_it->second.blocksize_))
147 ✗ throw ProtocolError("Blocksize mismatch");
148
4/6
✓ Branch 5 taken 47 times.
✗ Branch 6 not taken.
✓ Branch 10 taken 47 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 47 times.
✓ Branch 14 taken 8 times.
 141 variable_cache_lock_[c.index_].readFromBase64(
149 94 data, bi.base64Length_, c, current_blocksize_);
150 }
151
152
153 45 void PeriodicSubscriptionsBase::DataReceiveHandle::readTimestamps(
154 const char *data)
155 {
156 static_assert(
157 sizeof(decltype(time_vector_)::value_type) == sizeof(int64_t),
158 "time_vector_ is not size compatible to time tag");
159
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 45 times.
 45 if (!ps_) {
160 ✗ throw PdCom::InternalError("Operating on invalid DataReceiveHandle");
161 }
162
2/4
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 45 times.
 45 if (!data || !*data)
163 ✗ throw PdCom::ProtocolError("Tag does not contain data");
164
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 45 times.
 45 const auto b64len = ::strlen(data);
165
1/2
✓ Branch 1 taken 45 times.
✗ Branch 2 not taken.
 45 if (current_blocksize_ == -1) {
166 45 current_blocksize_ = b64len * 3 / 4 / sizeof(int64_t);
167 45 max_read_blocks_ = current_blocksize_;
168 }
169 45 consistent_blocksize_ = true;
170
171
1/2
✓ Branch 5 taken 45 times.
✗ Branch 6 not taken.
 90 ps_->time_vector_.resize(
172 90 std::max<size_t>(current_blocksize_ + 1, ps_->time_vector_.size()));
173
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 45 times.
 90 if (!readFromBase64(
174 45 reinterpret_cast<char *>(ps_->time_vector_.data()), data,
175 45 b64len, sizeof(int64_t) * current_blocksize_)) {
176 ✗ throw ProtocolError("Invalid <time> child of <data>");
177 }
178 45 timestamps_read_ = true;
179 45 }
180
181 1 void PeriodicSubscriptionsWithGroup::dump(
182 std::vector<PdCom::Process::SubscriptionInfo> &ans,
183 const std::unordered_map<unsigned, Channel *> &channels) const
184 {
185 1 map_.dump(ans, channels);
186 1 }
187
188 51 void PeriodicSubscriptionsBase::DataReceiveHandle::endNewDataRecieving()
189 {
190
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 51 times.
 51 if (!ps_) {
191 ✗ throw PdCom::InternalError("Operating on invalid DataReceiveHandle");
192 }
193 // make sure to release the variable cache lock
194
1/2
✓ Branch 7 taken 51 times.
✗ Branch 8 not taken.
 102 auto cache = std::move(variable_cache_lock_);
195 // as this is the final call on receiving a new dataset,
196 // make sure to erase the ps pointer to mark us invalid afterwards
197 const struct PsDeleter
198 {
199 std::unique_ptr<PeriodicSubscriptionsBase, Unlocker> &ps_;
200 51 ~PsDeleter() { ps_.reset(); }
201 51 PsDeleter(std::unique_ptr<PeriodicSubscriptionsBase, Unlocker> &ps) :
202 51 ps_(ps)
203 51 {}
204
1/2
✓ Branch 4 taken 51 times.
✗ Branch 5 not taken.
 102 } psdeleter_ {ps_};
205 // subscriptions can dissappear during <data> tag parsing, so don't
206 // assume that subscriptions_ is not empty
207
3/6
✓ Branch 1 taken 51 times.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 51 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 51 times.
 51 if (!cache or current_blocksize_ == -1)
208 ✗ return;
209
210
2/2
✓ Branch 1 taken 64 times.
✓ Branch 2 taken 51 times.
 115 for (int block = 0; block < max_read_blocks_; ++block) {
211
2/2
✓ Branch 7 taken 67 times.
✓ Branch 8 taken 64 times.
 131 for (const auto &sub_map : *subscription_map_) {
212
1/2
✓ Branch 2 taken 67 times.
✗ Branch 3 not taken.
 67 const DataDecoder *var = cache.filled_cache(sub_map.first);
213 // no data recieved for given variable
214
2/2
✓ Branch 0 taken 66 times.
✓ Branch 1 taken 1 times.
 68 if (!var
215
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 66 times.
 66 || sub_map.second.subscriptions_.currentState
216 != PdCom::Subscription::State::Active)
217 1 continue;
218
1/2
✓ Branch 4 taken 66 times.
✗ Branch 5 not taken.
 66 sendDataToSubscribers(*var, sub_map.second, block);
219 }
220
1/2
✓ Branch 4 taken 64 times.
✗ Branch 5 not taken.
 64 ps_->subscriber_notifier_.notify();
221 64 ps_->subscriber_notifier_.clear();
222 }
223
1/2
✓ Branch 2 taken 51 times.
✗ Branch 3 not taken.
 51 current_blocksize_ = -1;
224 }
225
226 66 void PeriodicSubscriptionsBase::DataReceiveHandle::sendDataToSubscribers(
227 const DataDecoder &var,
228 const ChannelSubscripion &sub_map,
229 const int block)
230 {
231 66 bool size_check_done = false;
232
8/8
✓ Branch 6 taken 74 times.
✓ Branch 7 taken 64 times.
✓ Branch 12 taken 64 times.
✓ Branch 13 taken 2 times.
✓ Branch 15 taken 64 times.
✓ Branch 16 taken 2 times.
✓ Branch 19 taken 64 times.
✓ Branch 20 taken 2 times.
 138 for (const auto &weak_sub : sub_map.subscriptions_) {
233
3/4
✓ Branch 3 taken 74 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 72 times.
✓ Branch 7 taken 2 times.
 146 if (const auto sub = weak_sub.lock()) {
234
4/4
✓ Branch 5 taken 72 times.
✓ Branch 6 taken 2 times.
✓ Branch 8 taken 72 times.
✓ Branch 9 taken 2 times.
 146 const auto var_ptr = sub->variable_.lock();
235
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 74 times.
 74 if (!var_ptr)
236 ✗ throw EmptyVariable();
237
2/2
✓ Branch 0 taken 66 times.
✓ Branch 1 taken 8 times.
 74 if (!size_check_done) {
238
2/4
✗ Branch 1 not taken.
✓ Branch 2 taken 66 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 66 times.
 132 if (consistent_blocksize_
239
3/4
✓ Branch 0 taken 49 times.
✓ Branch 1 taken 17 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 49 times.
 115 && var.size() < sub_map.blocksize_
240
1/2
✓ Branch 5 taken 49 times.
✗ Branch 6 not taken.
 49 * var_ptr->totalSizeInBytes()) {
241 ✗ std::ostringstream os;
242 os << "size mismatch in variable cache, "
243 ✗ "expected "
244 ✗ << sub_map.blocksize_ * var_ptr->totalSizeInBytes()
245 ✗ << ", got " << var.size();
246 ✗ throw ProtocolError(os.str());
247 }
248
3/4
✓ Branch 7 taken 66 times.
✗ Branch 8 not taken.
✓ Branch 9 taken 2 times.
✓ Branch 10 taken 64 times.
 66 else if (var.size() < (block + 1) * var_ptr->totalSizeInBytes())
249 2 return;
250 64 size_check_done = true;
251 }
252
2/4
✓ Branch 11 taken 72 times.
✗ Branch 12 not taken.
✓ Branch 14 taken 72 times.
✗ Branch 15 not taken.
 72 sub->readData(var.data() + block * var_ptr->totalSizeInBytes());
253 72 std::chrono::nanoseconds ts;
254
3/4
✗ Branch 1 not taken.
✓ Branch 2 taken 72 times.
✓ Branch 3 taken 56 times.
✓ Branch 4 taken 16 times.
 72 if (timestamps_read_)
255
1/2
✓ Branch 4 taken 56 times.
✗ Branch 5 not taken.
 56 ts = ps_->time_vector_.at(block);
256 else
257 32 ts = ts_
258 64 + (block - max_read_blocks_ + 1)
259
1/2
✓ Branch 1 taken 16 times.
✗ Branch 2 not taken.
 32 * std::chrono::duration_cast<
260
1/2
✓ Branch 1 taken 16 times.
✗ Branch 2 not taken.
 32 std::chrono::nanoseconds>(
261
1/2
✓ Branch 9 taken 16 times.
✗ Branch 10 not taken.
 64 var_ptr->getSampleTime())
262 48 * sub_map.decimation_;
263
3/4
✓ Branch 8 taken 72 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 72 times.
✓ Branch 12 taken 2 times.
 72 ps_->subscriber_notifier_.insert(sub->subscriber_, ts);
264 }
265 }
266 }
267
268 PeriodicSubscriptionsBase::GroupId
269 52 PeriodicSubscriptionsWithGroup::GroupMap::find(PdCom::Transmission t)
270 {
271 // does a group already exists for given transmission?
272
1/2
✓ Branch 2 taken 52 times.
✗ Branch 3 not taken.
 104 const auto it = backward_.find(t);
273
2/2
✓ Branch 6 taken 19 times.
✓ Branch 7 taken 33 times.
 52 if (it != backward_.end()) {
274
3/6
✓ Branch 4 taken 19 times.
✗ Branch 5 not taken.
✗ Branch 7 not taken.
✓ Branch 8 taken 19 times.
✓ Branch 9 taken 19 times.
✗ Branch 10 not taken.
 38 if (forward_.at(it->second - 1).state
275 19 == ChannelSubscriptionMapImpl::Taken)
276 19 return it->second;
277 }
278 // find a free one;
279
2/2
✓ Branch 2 taken 8 times.
✓ Branch 3 taken 32 times.
 40 for (unsigned int i = 0; i < forward_.size(); ++i) {
280
3/4
✗ Branch 3 not taken.
✓ Branch 4 taken 8 times.
✓ Branch 5 taken 1 times.
✓ Branch 6 taken 7 times.
 8 if (forward_[i].state == ChannelSubscriptionMapImpl::Empty) {
281 1 const auto group_id = i + 1;
282
1/2
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
 1 backward_[t] = group_id;
283 1 forward_[i].state = ChannelSubscriptionMapImpl::Taken;
284 1 return group_id;
285 }
286 }
287 // allocate one at the back;
288
1/2
✓ Branch 2 taken 32 times.
✗ Branch 3 not taken.
 32 forward_.emplace_back();
289 32 forward_.back().state = ChannelSubscriptionMapImpl::Taken;
290
1/2
✓ Branch 4 taken 32 times.
✗ Branch 5 not taken.
 32 const auto group_id = backward_[t] = forward_.size();
291 32 return group_id;
292 }
293
294 PeriodicSubscriptionsBase::GroupId
295 27 PeriodicSubscriptionsWithGroup::GroupMap::erase(
296 PdCom::Transmission t,
297 PeriodicSubscriptionsBase::ChannelId channel_idx)
298 {
299
1/2
✓ Branch 2 taken 27 times.
✗ Branch 3 not taken.
 54 const auto backward_it = backward_.find(t);
300
1/2
✗ Branch 6 not taken.
✓ Branch 7 taken 27 times.
 27 if (backward_it == backward_.end()) {
301 ✗ assert(backward_it != backward_.end()
302 and "Inconsistent state, backward_ and forward_ out of sync");
303 ✗ return 0;
304 }
305
1/2
✓ Branch 4 taken 27 times.
✗ Branch 5 not taken.
 27 auto &channel_map = at(backward_it->second);
306
1/2
✓ Branch 2 taken 27 times.
✗ Branch 3 not taken.
 54 const auto map_it = channel_map.find(channel_idx);
307 // already cleaned? if so, nothing to do.
308
1/2
✗ Branch 6 not taken.
✓ Branch 7 taken 27 times.
 27 if (map_it == channel_map.end())
309 ✗ return 0;
310 // clean subscription map
311 const bool no_channel_subs_left =
312 27 map_it->second.subscriptions_.remove_expired();
313 // if channels left, also nothing to do.
314
2/2
✓ Branch 0 taken 5 times.
✓ Branch 1 taken 22 times.
 27 if (!no_channel_subs_left)
315 5 return 0;
316 // no subscriptions for this channel is left, we have to remember the
317 // group id
318 22 const auto ans = backward_it->second;
319 // now check whether any subscriptions are left for this transmission
320
2/2
✓ Branch 2 taken 14 times.
✓ Branch 3 taken 8 times.
 22 if (channel_map.guaranteed_empty()) {
321 // group id no longer in use, free up channel map
322 14 channel_map.state = ChannelSubscriptionMapImpl::AboutToBeDeleted;
323
1/2
✓ Branch 4 taken 14 times.
✗ Branch 5 not taken.
 14 backward_.erase(backward_it);
324 }
325 22 return ans;
326 }
327
328 41 void PeriodicSubscriptionsWithGroup::subscribeWasConfirmed(
329 PdCom::impl::Process::PendingCallbackQueue &q,
330 PeriodicSubscriptionsWithGroup::GroupId group_id,
331 PeriodicSubscriptionsWithGroup::ChannelId channel_id)
332 {
333
1/2
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
 41 auto &sub_map = map_.at(group_id);
334
2/4
✓ Branch 2 taken 41 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 41 times.
✗ Branch 6 not taken.
 82 const auto it = sub_map.find(channel_id);
335
1/2
✗ Branch 6 not taken.
✓ Branch 7 taken 41 times.
 41 if (it == sub_map.end())
336 ✗ return;
337
338
2/4
✓ Branch 3 taken 41 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 41 times.
✗ Branch 7 not taken.
 41 it->second.subscriptions_.broadcastState(
339 PdCom::Subscription::State::Active, q);
340 }
341
342 1 void PeriodicSubscriptionsWithGroup::GroupMap::dump(
343 std::vector<PdCom::Process::SubscriptionInfo> &ans,
344 const std::unordered_map<unsigned, Channel *> &channels) const
345 {
346
2/2
✓ Branch 6 taken 1 times.
✓ Branch 7 taken 1 times.
 2 for (const auto &all_groups : forward_) {
347
2/2
✓ Branch 5 taken 1 times.
✓ Branch 6 taken 1 times.
 2 for (const auto &sub_map : all_groups) {
348 1 const auto var = impl::Variable::toUApi(
349 3 std::static_pointer_cast<const MsrProto::Variable>(
350
2/4
✓ Branch 5 taken 1 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 1 times.
✗ Branch 10 not taken.
 5 channels.at(sub_map.first)->shared_from_this()));
351
2/2
✓ Branch 6 taken 1 times.
✓ Branch 7 taken 1 times.
 2 for (const auto &weak_sub : sub_map.second.subscriptions_) {
352
1/2
✓ Branch 3 taken 1 times.
✗ Branch 4 not taken.
 2 if (const auto sub = weak_sub.lock()) {
353
1/2
✓ Branch 10 taken 1 times.
✗ Branch 11 not taken.
 1 ans.emplace_back(sub->This_, &sub->subscriber_, var);
354 }
355 }
356 }
357 }
358 1 }
359
360 PeriodicSubscriptionWithoutGroup::transaction
361 7 PeriodicSubscriptionWithoutGroup::add(
362 Process::PendingCallbackQueue &q,
363 bool notify_pending,
364 const std::shared_ptr<PeriodicSubscription> &s)
365 {
366 14 const auto var_ptr = s->variable_.lock();
367
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 7 times.
 7 if (!var_ptr)
368 ✗ throw EmptyVariable();
369 7 const auto &var = static_cast<Channel const &>(*var_ptr);
370
1/2
✓ Branch 3 taken 7 times.
✗ Branch 4 not taken.
 14 auto map_it = map_.find(var.index_);
371
372
373 7 unsigned decimation = s->subscriber_.getTransmission().getInterval()
374 7 / var.sample_time_.count()
375 7 + 0.5;
376 14 decimation += s->subscriber_.getTransmission().getInterval() > 0.0
377
2/4
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 7 times.
 7 and !decimation;
378 7 unsigned blocksize = 0;
379
380
1/2
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
 7 if (decimation > 0) {
381 7 blocksize = 1.0 / var.sample_time_.count() / decimation / 25.0 + 0.5;
382
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 7 times.
 7 blocksize = std::min(var.bufsize / decimation, blocksize);
383 }
384
385
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 3 times.
 7 if (!blocksize)
386 4 blocksize = 1;
387
1/2
✓ Branch 10 taken 7 times.
✗ Branch 11 not taken.
 7 time_vector_.resize(std::max<size_t>(time_vector_.size(), blocksize + 2));
388
389
2/2
✓ Branch 6 taken 5 times.
✓ Branch 7 taken 2 times.
 7 if (map_it == map_.end()) {
390 // not yet subscribed
391
5/10
✓ Branch 10 taken 5 times.
✗ Branch 11 not taken.
✓ Branch 14 taken 5 times.
✗ Branch 15 not taken.
✓ Branch 18 taken 5 times.
✗ Branch 19 not taken.
✓ Branch 29 taken 5 times.
✓ Branch 30 taken 5 times.
✗ Branch 41 not taken.
✗ Branch 42 not taken.
 5 map_.insert({var.index_, {blocksize, decimation, {{s}}}});
392
393
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 3 times.
 5 if (notify_pending)
394
1/2
✓ Branch 8 taken 2 times.
✗ Branch 9 not taken.
 2 q.push_back({s, PdCom::Subscription::State::Pending});
395 5 return {true, decimation, blocksize};
396 }
397
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 2 times.
 2 else if (decimation < map_it->second.decimation_) {
398 // we have to resubscribe
399 ✗ map_it->second.subscriptions_.currentState =
400 PdCom::Subscription::State::Pending;
401
402 ✗ if (notify_pending)
403 ✗ q.push_back({s, PdCom::Subscription::State::Pending});
404 ✗ auto subs = std::move(map_it->second.subscriptions_);
405 ✗ subs.insert(s);
406 ✗ map_.erase(map_it);
407 ✗ map_.insert({var.index_, {blocksize, decimation, std::move(subs)}});
408 ✗ return {true, decimation, blocksize};
409 }
410 else {
411
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 if (notify_pending
412
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 2 or map_it->second.subscriptions_.currentState
413 != PdCom::Subscription::State::Pending)
414
1/2
✓ Branch 10 taken 2 times.
✗ Branch 11 not taken.
 2 q.push_back({s, map_it->second.subscriptions_.currentState});
415
1/2
✓ Branch 6 taken 2 times.
✗ Branch 7 not taken.
 2 map_it->second.subscriptions_.insert(s);
416 2 return {false, map_it->second.decimation_, map_it->second.blocksize_};
417 }
418 }
419
420 PeriodicSubscriptionWithoutGroup::transaction
421 ✗ PeriodicSubscriptionWithoutGroup::remove(const Variable &_var)
422 {
423 ✗ const auto it = map_.find(_var.index_);
424 ✗ if (it == map_.end() || !it->second.subscriptions_.remove_expired())
425 // not found or still subscribed by others
426 ✗ return {false, 0, 0};
427 else {
428 // now empty
429 ✗ map_.erase(it);
430 ✗ return {true, 0, 0};
431 }
432 }
433
434 5 void PeriodicSubscriptionWithoutGroup::subscribeWasConfirmed(
435 Process::PendingCallbackQueue &q,
436 ChannelId channel_id)
437 {
438
2/4
✓ Branch 2 taken 5 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 5 times.
✗ Branch 6 not taken.
 10 const auto it = map_.find(channel_id);
439
1/2
✗ Branch 6 not taken.
✓ Branch 7 taken 5 times.
 5 if (it == map_.end())
440 ✗ return;
441
442
2/4
✓ Branch 3 taken 5 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 5 times.
✗ Branch 7 not taken.
 5 it->second.subscriptions_.broadcastState(
443 PdCom::Subscription::State::Active, q);
444 }
445
446 ✗ void PeriodicSubscriptionWithoutGroup::dump(
447 std::vector<PdCom::Process::SubscriptionInfo> &ans,
448 const std::unordered_map<unsigned, Channel *> &channels) const
449 {
450 ✗ for (const auto &sub_map : map_) {
451 ✗ const auto var = impl::Variable::toUApi(
452 ✗ std::static_pointer_cast<const MsrProto::Variable>(
453 ✗ channels.at(sub_map.first)->shared_from_this()));
454 ✗ for (const auto &weak_sub : sub_map.second.subscriptions_) {
455 ✗ if (const auto sub = weak_sub.lock()) {
456 ✗ ans.emplace_back(sub->This_, &sub->subscriber_, var);
457 }
458 }
459 }
460 }
461