| Fast DDS Version 3.3.0 Fast DDS

Loading...
Searching...
No Matches
CacheChange.hpp
1// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
18
19#ifndef FASTDDS_RTPS_COMMON__CACHECHANGE_HPP
20#define FASTDDS_RTPS_COMMON__CACHECHANGE_HPP
21
22#include <atomic>
23#include <cassert>
24
25#include <fastdds/rtps/common/ChangeKind_t.hpp>
26#include <fastdds/rtps/common/FragmentNumber.hpp>
27#include <fastdds/rtps/common/InstanceHandle.hpp>
28#include <fastdds/rtps/common/SerializedPayload.hpp>
29#include <fastdds/rtps/common/Time_t.hpp>
30#include <fastdds/rtps/common/Types.hpp>
31#include <fastdds/rtps/common/VendorId_t.hpp>
32#include <fastdds/rtps/common/WriteParams.hpp>
33#include <fastdds/rtps/history/IPayloadPool.hpp>
34
35 namespace eprosima {
36 namespace fastdds {
37 namespace rtps {
38
39 struct CacheChange_t;
40
44 struct CacheChangeWriterInfo_t
45{
47 size_t num_sent_submessages = 0;
50 CacheChange_t* volatile previous = nullptr;
53 CacheChange_t* volatile next = nullptr;
55 std::atomic_bool is_linked {false};
56};
57
61 struct CacheChangeReaderInfo_t
62{
64 Time_t receptionTimestamp;
66 int32_t disposed_generation_count;
68 int32_t no_writers_generation_count;
70 uint32_t writer_ownership_strength;
71};
72
77 struct FASTDDS_EXPORTED_API CacheChange_t
78{
80 ChangeKind_t kind = ALIVE;
82 GUID_t writerGUID{};
84 InstanceHandle_t instanceHandle{};
86 SequenceNumber_t sequenceNumber{};
88 SerializedPayload_t serializedPayload{};
90 SerializedPayload_t inline_qos{};
92 bool isRead = false;
94 Time_t sourceTimestamp{};
96 fastdds::rtps::VendorId_t vendor_id = c_VendorId_Unknown;
97
98 union
99 {
100 CacheChangeReaderInfo_t reader_info;
101 CacheChangeWriterInfo_t writer_info;
102 };
103
104 WriteParams write_params{};
105 bool is_untyped_ = true;
106
111 CacheChange_t()
112 : writer_info()
113 {
114 inline_qos.encapsulation = DEFAULT_ENDIAN == LITTLEEND ? PL_CDR_LE : PL_CDR_BE;
115 }
116
117 CacheChange_t(
118 const CacheChange_t&) = delete;
119 const CacheChange_t& operator =(
120 const CacheChange_t&) = delete;
121
127 CacheChange_t(
128 uint32_t payload_size,
129 bool is_untyped = false)
130 : serializedPayload(payload_size)
131 , is_untyped_(is_untyped)
132 {
133 }
134
140 bool copy(
141 const CacheChange_t* ch_ptr)
142 {
143 kind = ch_ptr->kind;
144 writerGUID = ch_ptr->writerGUID;
145 instanceHandle = ch_ptr->instanceHandle;
146 sequenceNumber = ch_ptr->sequenceNumber;
147 sourceTimestamp = ch_ptr->sourceTimestamp;
148 reader_info.receptionTimestamp = ch_ptr->reader_info.receptionTimestamp;
149 write_params = ch_ptr->write_params;
150 isRead = ch_ptr->isRead;
151 vendor_id = ch_ptr->vendor_id;
152 fragment_size_ = ch_ptr->fragment_size_;
153 fragment_count_ = ch_ptr->fragment_count_;
154 first_missing_fragment_ = ch_ptr->first_missing_fragment_;
155
156 return serializedPayload.copy(&ch_ptr->serializedPayload, !ch_ptr->is_untyped_);
157 }
158
164 void copy_not_memcpy(
165 const CacheChange_t* ch_ptr)
166 {
167 kind = ch_ptr->kind;
168 writerGUID = ch_ptr->writerGUID;
169 instanceHandle = ch_ptr->instanceHandle;
170 sequenceNumber = ch_ptr->sequenceNumber;
171 sourceTimestamp = ch_ptr->sourceTimestamp;
172 reader_info.receptionTimestamp = ch_ptr->reader_info.receptionTimestamp;
173 write_params = ch_ptr->write_params;
174 isRead = ch_ptr->isRead;
175 vendor_id = ch_ptr->vendor_id;
176
177 // Copy certain values from serializedPayload
178 serializedPayload.encapsulation = ch_ptr->serializedPayload.encapsulation;
179 serializedPayload.is_serialized_key = ch_ptr->serializedPayload.is_serialized_key;
180
181 // Copy fragment size and calculate fragment count
182 setFragmentSize(ch_ptr->fragment_size_, false);
183 }
184
185 virtual ~CacheChange_t() = default;
186
191 uint32_t getFragmentCount() const
192 {
193 return fragment_count_;
194 }
195
200 uint16_t getFragmentSize() const
201 {
202 return fragment_size_;
203 }
204
209 bool is_fully_assembled()
210 {
211 return first_missing_fragment_ >= fragment_count_;
212 }
213
217 bool contains_first_fragment()
218 {
219 return 0 < first_missing_fragment_;
220 }
221
226 void get_missing_fragments(
227 FragmentNumberSet_t& frag_sns)
228 {
229 // Note: Fragment numbers are 1-based but we keep them 0 based.
230 frag_sns.base(first_missing_fragment_ + 1);
231
232 // Traverse list of missing fragments, adding them to frag_sns
233 uint32_t current_frag = first_missing_fragment_;
234 while (current_frag < fragment_count_)
235 {
236 frag_sns.add(current_frag + 1);
237 current_frag = get_next_missing_fragment(current_frag);
238 }
239 }
240
250 void setFragmentSize(
251 uint16_t fragment_size,
252 bool create_fragment_list = false)
253 {
254 fragment_size_ = fragment_size;
255 fragment_count_ = 0;
256 first_missing_fragment_ = 0;
257
258 if (fragment_size > 0)
259 {
260 // This follows RTPS 8.3.7.3.5
261 fragment_count_ = (serializedPayload.length + fragment_size - 1) / fragment_size;
262
263 if (create_fragment_list)
264 {
265 // Keep index of next fragment on the payload portion at the beginning of each fragment. Last
266 // fragment will have fragment_count_ as 'next fragment index'
267 size_t offset = 0;
268 for (uint32_t i = 1; i <= fragment_count_; i++, offset += fragment_size_)
269 {
270 set_next_missing_fragment(i - 1, i); // index to next fragment in missing list
271 }
272 }
273 else
274 {
275 // List not created. This means we are going to send this change fragmented, so it is already
276 // assembled, and the missing list is empty (i.e. first missing points to fragment count)
277 first_missing_fragment_ = fragment_count_;
278 }
279 }
280 }
281
282 bool add_fragments(
283 const SerializedPayload_t& incoming_data,
284 uint32_t fragment_starting_num,
285 uint32_t fragments_in_submessage)
286 {
287 uint32_t original_offset = (fragment_starting_num - 1) * fragment_size_;
288 uint32_t incoming_length = fragment_size_ * fragments_in_submessage;
289 uint32_t last_fragment_index = fragment_starting_num + fragments_in_submessage - 1;
290
291 // Validate payload types
292 if (serializedPayload.is_serialized_key != incoming_data.is_serialized_key)
293 {
294 return false;
295 }
296
297 // Validate fragment indexes
298 if (last_fragment_index > fragment_count_)
299 {
300 return false;
301 }
302
303 // validate lengths
304 if (last_fragment_index < fragment_count_)
305 {
306 if (incoming_data.length < incoming_length)
307 {
308 return false;
309 }
310 }
311 else
312 {
313 incoming_length = serializedPayload.length - original_offset;
314 }
315
316 if (original_offset + incoming_length > serializedPayload.length)
317 {
318 return false;
319 }
320
321 if (received_fragments(fragment_starting_num - 1, fragments_in_submessage))
322 {
323 memcpy(
324 &serializedPayload.data[original_offset],
325 incoming_data.data, incoming_length);
326 }
327
328 return is_fully_assembled();
329 }
330
331 private:
332
333 // Fragment size
334 uint16_t fragment_size_ = 0;
335
336 // Number of fragments
337 uint32_t fragment_count_ = 0;
338
339 // First fragment in missing list
340 uint32_t first_missing_fragment_ = 0;
341
342 uint32_t get_next_missing_fragment(
343 uint32_t fragment_index)
344 {
345 uint32_t* ptr = next_fragment_pointer(fragment_index);
346 return *ptr;
347 }
348
349 void set_next_missing_fragment(
350 uint32_t fragment_index,
351 uint32_t next_fragment_index)
352 {
353 uint32_t* ptr = next_fragment_pointer(fragment_index);
354 *ptr = next_fragment_index;
355 }
356
357 uint32_t* next_fragment_pointer(
358 uint32_t fragment_index)
359 {
360 size_t offset = fragment_size_;
361 offset *= fragment_index;
362 offset = (offset + 3u) & ~3u;
363 return reinterpret_cast<uint32_t*>(&serializedPayload.data[offset]);
364 }
365
375 bool received_fragments(
376 uint32_t initial_fragment,
377 uint32_t num_of_fragments)
378 {
379 bool at_least_one_changed = false;
380
381 if ((fragment_size_ > 0) && (initial_fragment < fragment_count_))
382 {
383 uint32_t last_fragment = initial_fragment + num_of_fragments;
384 if (last_fragment > fragment_count_)
385 {
386 last_fragment = fragment_count_;
387 }
388
389 if (initial_fragment <= first_missing_fragment_)
390 {
391 // Perform first = *first until first >= last_received
392 while (first_missing_fragment_ < last_fragment)
393 {
394 first_missing_fragment_ = get_next_missing_fragment(first_missing_fragment_);
395 at_least_one_changed = true;
396 }
397 }
398 else
399 {
400 // Find prev in missing list
401 uint32_t current_frag = first_missing_fragment_;
402 while (current_frag < initial_fragment)
403 {
404 uint32_t next_frag = get_next_missing_fragment(current_frag);
405 if (next_frag >= initial_fragment)
406 {
407 // This is the fragment previous to initial_fragment.
408 // Find future value for next by repeating next = *next until next >= last_fragment.
409 uint32_t next_missing_fragment = next_frag;
410 while (next_missing_fragment < last_fragment)
411 {
412 next_missing_fragment = get_next_missing_fragment(next_missing_fragment);
413 at_least_one_changed = true;
414 }
415
416 // Update next and finish loop
417 if (at_least_one_changed)
418 {
419 set_next_missing_fragment(current_frag, next_missing_fragment);
420 }
421 break;
422 }
423 current_frag = next_frag;
424 }
425 }
426 }
427
428 return at_least_one_changed;
429 }
430
431};
432
433} // namespace rtps
434} // namespace fastdds
435} // namespace eprosima
436
437#endif // FASTDDS_RTPS_COMMON__CACHECHANGE_HPP
eprosima::fastdds::BitmapRange::base
T base() const noexcept
Get base of the range.
Definition fixed_size_bitmap.hpp:131
eprosima::fastdds::BitmapRange::add
bool add(const T &item) noexcept
Adds an element to the range.
Definition fixed_size_bitmap.hpp:295
eprosima::fastdds::rtps::Time_t
Structure Time_t, used to describe times at RTPS protocol.
Definition Time_t.hpp:38
eprosima::fastdds::rtps::WriteParams
This class contains additional information of a CacheChange.
Definition WriteParams.hpp:36
eprosima::fastdds::rtps
Contains the RTPS protocol implementation.
eprosima::fastdds::rtps::LITTLEEND
@ LITTLEEND
Little endianness.
Definition Types.hpp:44
eprosima::fastdds::rtps::c_VendorId_Unknown
const VendorId_t c_VendorId_Unknown
Definition VendorId_t.hpp:34
eprosima::fastdds::rtps::VendorId_t
std::array< uint8_t, 2 > VendorId_t
Structure VendorId_t, specifying the vendor Id of the implementation.
Definition VendorId_t.hpp:32
eprosima::fastdds::rtps::DEFAULT_ENDIAN
constexpr Endianness_t DEFAULT_ENDIAN
Definition Types.hpp:80
eprosima::fastdds::rtps::ChangeKind_t
ChangeKind_t
Enumerates the different types of CacheChange_t.
Definition ChangeKind_t.hpp:38
eprosima::fastdds::rtps::ALIVE
@ ALIVE
ALIVE.
Definition ChangeKind_t.hpp:39
eprosima::fastdds::rtps::FragmentNumberSet_t
BitmapRange< FragmentNumber_t > FragmentNumberSet_t
Structure FragmentNumberSet_t, contains a group of fragmentnumbers.
Definition FragmentNumber.hpp:37
eprosima::fastdds
eprosima
eProsima namespace.
eprosima::fastdds::rtps::CacheChange_t
Structure CacheChange_t, contains information on a specific CacheChange.
Definition CacheChange.hpp:78
eprosima::fastdds::rtps::CacheChange_t::kind
ChangeKind_t kind
Kind of change, default value ALIVE.
Definition CacheChange.hpp:80
eprosima::fastdds::rtps::CacheChange_t::is_untyped_
bool is_untyped_
Definition CacheChange.hpp:105
eprosima::fastdds::rtps::CacheChange_t::contains_first_fragment
bool contains_first_fragment()
Checks if the first fragment is present.
Definition CacheChange.hpp:217
eprosima::fastdds::rtps::CacheChange_t::instanceHandle
InstanceHandle_t instanceHandle
Handle of the data associated with this change.
Definition CacheChange.hpp:84
eprosima::fastdds::rtps::CacheChange_t::isRead
bool isRead
Indicates if the cache has been read (only used in READERS).
Definition CacheChange.hpp:92
eprosima::fastdds::rtps::CacheChange_t::vendor_id
fastdds::rtps::VendorId_t vendor_id
Vendor Id of the writer that generated this change.
Definition CacheChange.hpp:96
eprosima::fastdds::rtps::CacheChange_t::sourceTimestamp
Time_t sourceTimestamp
Source TimeStamp.
Definition CacheChange.hpp:94
eprosima::fastdds::rtps::CacheChange_t::CacheChange_t
CacheChange_t(const CacheChange_t &)=delete
eprosima::fastdds::rtps::CacheChange_t::setFragmentSize
void setFragmentSize(uint16_t fragment_size, bool create_fragment_list=false)
Set fragment size for this change.
Definition CacheChange.hpp:250
eprosima::fastdds::rtps::CacheChange_t::reader_info
CacheChangeReaderInfo_t reader_info
Definition CacheChange.hpp:100
eprosima::fastdds::rtps::CacheChange_t::~CacheChange_t
virtual ~CacheChange_t()=default
eprosima::fastdds::rtps::CacheChange_t::inline_qos
SerializedPayload_t inline_qos
CDR serialization of inlined QoS for this change.
Definition CacheChange.hpp:90
eprosima::fastdds::rtps::CacheChange_t::copy_not_memcpy
void copy_not_memcpy(const CacheChange_t *ch_ptr)
Copy information form a different change into this one.
Definition CacheChange.hpp:164
eprosima::fastdds::rtps::CacheChange_t::getFragmentSize
uint16_t getFragmentSize() const
Get the size of each fragment this change is split into.
Definition CacheChange.hpp:200
eprosima::fastdds::rtps::CacheChange_t::getFragmentCount
uint32_t getFragmentCount() const
Get the number of fragments this change is split into.
Definition CacheChange.hpp:191
eprosima::fastdds::rtps::CacheChange_t::copy
bool copy(const CacheChange_t *ch_ptr)
Copy a different change into this one.
Definition CacheChange.hpp:140
eprosima::fastdds::rtps::CacheChange_t::add_fragments
bool add_fragments(const SerializedPayload_t &incoming_data, uint32_t fragment_starting_num, uint32_t fragments_in_submessage)
Definition CacheChange.hpp:282
eprosima::fastdds::rtps::CacheChange_t::writer_info
CacheChangeWriterInfo_t writer_info
Definition CacheChange.hpp:101
eprosima::fastdds::rtps::CacheChange_t::writerGUID
GUID_t writerGUID
GUID_t of the writer that generated this change.
Definition CacheChange.hpp:82
eprosima::fastdds::rtps::CacheChange_t::CacheChange_t
CacheChange_t(uint32_t payload_size, bool is_untyped=false)
Constructor with payload size.
Definition CacheChange.hpp:127
eprosima::fastdds::rtps::CacheChange_t::serializedPayload
SerializedPayload_t serializedPayload
Serialized Payload associated with the change.
Definition CacheChange.hpp:88
eprosima::fastdds::rtps::CacheChange_t::get_missing_fragments
void get_missing_fragments(FragmentNumberSet_t &frag_sns)
Fills a FragmentNumberSet_t with the list of missing fragments.
Definition CacheChange.hpp:226
eprosima::fastdds::rtps::CacheChange_t::sequenceNumber
SequenceNumber_t sequenceNumber
SequenceNumber of the change.
Definition CacheChange.hpp:86
eprosima::fastdds::rtps::CacheChange_t::CacheChange_t
CacheChange_t()
Default constructor.
Definition CacheChange.hpp:111
eprosima::fastdds::rtps::CacheChange_t::write_params
WriteParams write_params
Definition CacheChange.hpp:104
eprosima::fastdds::rtps::CacheChange_t::is_fully_assembled
bool is_fully_assembled()
Checks if all fragments have been received.
Definition CacheChange.hpp:209
eprosima::fastdds::rtps::CacheChangeReaderInfo_t
Specific information for a reader.
Definition CacheChange.hpp:62
eprosima::fastdds::rtps::CacheChangeReaderInfo_t::receptionTimestamp
Time_t receptionTimestamp
Reception TimeStamp (only used in Readers).
Definition CacheChange.hpp:64
eprosima::fastdds::rtps::CacheChangeReaderInfo_t::no_writers_generation_count
int32_t no_writers_generation_count
No-writers generation of the instance when this entry was added to it.
Definition CacheChange.hpp:68
eprosima::fastdds::rtps::CacheChangeReaderInfo_t::writer_ownership_strength
uint32_t writer_ownership_strength
Ownership stregth of its writer when the sample was received.
Definition CacheChange.hpp:70
eprosima::fastdds::rtps::CacheChangeReaderInfo_t::disposed_generation_count
int32_t disposed_generation_count
Disposed generation of the instance when this entry was added to it.
Definition CacheChange.hpp:66
eprosima::fastdds::rtps::CacheChangeWriterInfo_t
Specific information for a writer.
Definition CacheChange.hpp:45
eprosima::fastdds::rtps::CacheChangeWriterInfo_t::is_linked
std::atomic_bool is_linked
Used to know if the object is already in a list.
Definition CacheChange.hpp:55
eprosima::fastdds::rtps::CacheChangeWriterInfo_t::num_sent_submessages
size_t num_sent_submessages
Number of DATA / DATA_FRAG submessages sent to the transport (only used in Writers).
Definition CacheChange.hpp:47
eprosima::fastdds::rtps::CacheChangeWriterInfo_t::next
CacheChange_t *volatile next
Used to link with next node in a list.
Definition CacheChange.hpp:53
eprosima::fastdds::rtps::CacheChangeWriterInfo_t::previous
CacheChange_t *volatile previous
Used to link with previous node in a list.
Definition CacheChange.hpp:50
eprosima::fastdds::rtps::GUID_t
Structure GUID_t, entity identifier, unique in DDS-RTPS Domain.
Definition Guid.hpp:40
eprosima::fastdds::rtps::InstanceHandle_t
Struct InstanceHandle_t, used to contain the key for WITH_KEY topics.
Definition InstanceHandle.hpp:154
eprosima::fastdds::rtps::SequenceNumber_t
Structure SequenceNumber_t, different for each change in the same writer.
Definition SequenceNumber.hpp:38
eprosima::fastdds::rtps::SerializedPayload_t
Structure SerializedPayload_t.
Definition SerializedPayload.hpp:59
eprosima::fastdds::rtps::SerializedPayload_t::data
octet * data
Pointer to the data.
Definition SerializedPayload.hpp:68
eprosima::fastdds::rtps::SerializedPayload_t::encapsulation
uint16_t encapsulation
Encapsulation of the data as suggested in the RTPS 2.1 specification chapter 10.
Definition SerializedPayload.hpp:64
eprosima::fastdds::rtps::SerializedPayload_t::is_serialized_key
bool is_serialized_key
Whether the payload contains a serialized key, or the whole data.
Definition SerializedPayload.hpp:76
eprosima::fastdds::rtps::SerializedPayload_t::length
uint32_t length
Actual length of the data.
Definition SerializedPayload.hpp:66