Skip to content

Commit 4cfd8cc

Browse files
DanipizaDanipiza
authored
Fix DiscoveryDatabase endpoint cleanup for inactive updates and shutdown (#188)
* [#25493] Fix DiscoveryDatabase endpoint cleanup for inactive updates and shutdown Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#25493] Added tests Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#25493] Applied Uncrustify Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#25493] Review: Restored strict endpoint erase semantics for filtered GUID cleanup Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#25493] Review: Changed teardown removed endpoint not in DB from error to debug Signed-off-by: danipiza <dpizarrogallego@gmail.com> * [#25493] Applied review Signed-off-by: danipiza <dpizarrogallego@gmail.com> --------- Signed-off-by: danipiza <dpizarrogallego@gmail.com>
1 parent f74e24d commit 4cfd8cc

9 files changed

Lines changed: 386 additions & 34 deletions

File tree

‎ddspipe_core/include/ddspipe_core/dynamic/DiscoveryDatabase.hpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
#include <functional>
2020
#include <map>
2121
#include <mutex>
22+
#include <set>
2223
#include <shared_mutex>
2324
#include <string>
2425
#include <thread>
@@ -263,7 +264,7 @@ class DiscoveryDatabase
263264
//! Database of endpoints indexed by guid
264265
std::map<types::Guid, types::Endpoint> entities_;
265266

266-
//! Database of endpoints indexed by guid
267+
//! Database of filtered endpoint GUIDs
267268
std::set<types::Guid> entities_filter_;
268269

269270
//! Mutex to guard queries to the database

‎ddspipe_core/src/cpp/dynamic/DiscoveryDatabase.cpp‎

Lines changed: 57 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,12 @@ void DiscoveryDatabase::stop() noexcept
7676
{
7777
logDebug(DDSPIPE_DISCOVERY_DATABASE, "Processing thread routine already stopped.");
7878
}
79+
80+
{
81+
std::unique_lock<std::shared_timed_mutex> lock(mutex_);
82+
entities_.clear();
83+
entities_filter_.clear();
84+
}
7985
}
8086

8187
bool DiscoveryDatabase::topic_exists(
@@ -107,6 +113,7 @@ bool DiscoveryDatabase::add_endpoint_(
107113
{
108114
{
109115
std::unique_lock<std::shared_timed_mutex> lock(mutex_);
116+
entities_filter_.erase(new_endpoint.guid);
110117

111118
auto it = entities_.find(new_endpoint.guid);
112119
if (it != entities_.end())
@@ -115,8 +122,8 @@ bool DiscoveryDatabase::add_endpoint_(
115122
if (it->second.active)
116123
{
117124
throw utils::InconsistencyException(
118-
utils::Formatter() <<
119-
"Error adding Endpoint to database. Endpoint already exists." << new_endpoint);
125+
utils::Formatter()
126+
<< "Error adding Endpoint to database. Endpoint already exists." << new_endpoint);
120127
}
121128
else
122129
{
@@ -151,16 +158,43 @@ bool DiscoveryDatabase::add_endpoint_(
151158
bool DiscoveryDatabase::update_endpoint_(
152159
const Endpoint& endpoint_to_update)
153160
{
161+
if (!endpoint_to_update.active)
162+
{
163+
bool filtered_erased = false;
164+
bool endpoint_exists = false;
165+
166+
{
167+
std::unique_lock<std::shared_timed_mutex> lock(mutex_);
168+
filtered_erased = entities_filter_.erase(endpoint_to_update.guid) > 0;
169+
endpoint_exists = entities_.find(endpoint_to_update.guid) != entities_.end();
170+
}
171+
172+
if (!endpoint_exists)
173+
{
174+
if (filtered_erased)
175+
{
176+
return true;
177+
}
178+
179+
throw utils::InconsistencyException(
180+
utils::Formatter()
181+
<< "Error updating Endpoint in database. Endpoint entry not found." << endpoint_to_update);
182+
}
183+
184+
return erase_endpoint_(endpoint_to_update) == utils::ReturnCode::RETCODE_OK;
185+
}
186+
154187
{
155188
std::unique_lock<std::shared_timed_mutex> lock(mutex_);
189+
entities_filter_.erase(endpoint_to_update.guid);
156190

157191
auto it = entities_.find(endpoint_to_update.guid);
158192
if (it == entities_.end())
159193
{
160194
// Entry not found
161195
throw utils::InconsistencyException(
162-
utils::Formatter() <<
163-
"Error updating Endpoint in database. Endpoint entry not found." << endpoint_to_update);
196+
utils::Formatter()
197+
<< "Error updating Endpoint in database. Endpoint entry not found." << endpoint_to_update);
164198
}
165199
else
166200
{
@@ -185,26 +219,34 @@ bool DiscoveryDatabase::update_endpoint_(
185219
utils::ReturnCode DiscoveryDatabase::erase_endpoint_(
186220
const Endpoint& endpoint_to_erase)
187221
{
222+
bool endpoint_erased = false;
223+
188224
{
189225
std::unique_lock<std::shared_timed_mutex> lock(mutex_);
190226

191227
EPROSIMA_LOG_INFO(DDSPIPE_DISCOVERY_DATABASE, "Erasing Endpoint " << endpoint_to_erase << ".");
192228

193229
auto erased = entities_.erase(endpoint_to_erase.guid);
230+
endpoint_erased = erased > 0;
194231

195232
if (erased == 0)
196233
{
197234
throw utils::InconsistencyException(
198-
utils::Formatter() <<
199-
"Error erasing Endpoint " << endpoint_to_erase <<
200-
" from database. Endpoint entry not found.");
235+
utils::Formatter()
236+
<< "Error erasing Endpoint " << endpoint_to_erase
237+
<< " from database. Endpoint entry not found.");
201238
}
239+
240+
entities_filter_.erase(endpoint_to_erase.guid);
202241
}
203242

204-
std::lock_guard<std::mutex> lock(callbacks_mutex_);
205-
for (auto erased_endpoint_callback : erased_endpoint_callbacks_)
243+
if (endpoint_erased)
206244
{
207-
erased_endpoint_callback(endpoint_to_erase);
245+
std::lock_guard<std::mutex> lock(callbacks_mutex_);
246+
for (auto erased_endpoint_callback : erased_endpoint_callbacks_)
247+
{
248+
erased_endpoint_callback(endpoint_to_erase);
249+
}
208250
}
209251

210252
return utils::ReturnCode::RETCODE_OK;
@@ -213,6 +255,7 @@ utils::ReturnCode DiscoveryDatabase::erase_endpoint_(
213255
void DiscoveryDatabase::add_filtered_endpoint(
214256
const types::Guid guid)
215257
{
258+
std::unique_lock<std::shared_timed_mutex> lock(mutex_);
216259
entities_filter_.insert(guid);
217260
}
218261

@@ -237,6 +280,7 @@ void DiscoveryDatabase::erase_endpoint(
237280
bool DiscoveryDatabase::exists_filtered_endpoint(
238281
const Guid endpoint_guid)
239282
{
283+
std::shared_lock<std::shared_timed_mutex> lock(mutex_);
240284
return entities_filter_.find(endpoint_guid) != entities_filter_.end();
241285
}
242286

@@ -255,9 +299,9 @@ Endpoint DiscoveryDatabase::get_endpoint(
255299
}
256300

257301
throw utils::InconsistencyException(
258-
utils::Formatter() <<
259-
"Error retrieving Endpoint with GUID " << endpoint_guid <<
260-
" from database. Endpoint entry not found.");
302+
utils::Formatter()
303+
<< "Error retrieving Endpoint with GUID " << endpoint_guid
304+
<< " from database. Endpoint entry not found.");
261305
}
262306

263307
return it->second;

‎ddspipe_core/test/unittest/dynamic/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,5 +13,6 @@
1313
# limitations under the License.
1414

1515
add_subdirectory(allowed_topic_list)
16+
add_subdirectory(discovery_database)
1617
# TODO uncomment when ready
1718
# add_subdirectory(participants_database)

‎ddspipe_core/test/unittest/dynamic/discovery_database/CMakeLists.txt‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
# Copyright 2021 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+
15+
set(TEST_NAME DiscoveryDatabaseTest)
16+
17+
set(TEST_SOURCES
18+
DiscoveryDatabaseTest.cpp
19+
)
20+
21+
file(
22+
GLOB_RECURSE LIBRARY_SOURCES
23+
"${PROJECT_SOURCE_DIR}/src/cpp/*.c*"
24+
"${PROJECT_SOURCE_DIR}/include/*.h*"
25+
"${PROJECT_SOURCE_DIR}/include/*.ipp"
26+
)
27+
28+
all_library_sources(
29+
"${TEST_SOURCES}"
30+
"${LIBRARY_SOURCES}"
31+
)
32+
33+
set(TEST_LIST
34+
inactive_update_erases_endpoint
35+
inactive_update_clears_filtered_endpoint
36+
erase_does_not_remove_filtered_only_endpoint
37+
stop_clears_stored_endpoints
38+
)
39+
40+
set(TEST_EXTRA_LIBRARIES
41+
fastcdr
42+
fastdds
43+
cpp_utils
44+
)
45+
46+
add_unittest_executable(
47+
"${TEST_NAME}"
48+
"${TEST_SOURCES}"
49+
"${TEST_LIST}"
50+
"${TEST_EXTRA_LIBRARIES}"
51+
)

0 commit comments

Comments
 (0)