Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

protect subscriber_statistics_collectors_ with a mutex #1084

Merged
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -94,13 +94,16 @@ class SubscriptionTopicStatistics

/// Handle a message received by the subscription to collect statistics.
/**
* - This method acquires a lock to prevent race conditions to collectors list.
wjwwood marked this conversation as resolved.
Show resolved Hide resolved
*
* \param received_message the message received by the subscription
* \param now_nanoseconds current time in nanoseconds
*/
virtual void handle_message(
const CallbackMessageT & received_message,
const rclcpp::Time now_nanoseconds) const
{
std::lock_guard<std::mutex> lock(mutex_);
fujitatomoya marked this conversation as resolved.
Show resolved Hide resolved
for (const auto & collector : subscriber_statistics_collectors_) {
collector->OnMessageReceived(received_message, now_nanoseconds.nanoseconds());
}
Expand All @@ -116,33 +119,47 @@ class SubscriptionTopicStatistics
}

/// Publish a populated MetricsStatisticsMessage.
/**
* - This method acquires a lock to prevent race conditions to collectors list.
wjwwood marked this conversation as resolved.
Show resolved Hide resolved
*/
virtual void publish_message()
{
std::vector<MetricsMessage> msgs;
rclcpp::Time window_end{get_current_nanoseconds_since_epoch()};

for (auto & collector : subscriber_statistics_collectors_) {
const auto collected_stats = collector->GetStatisticsResults();

auto message = libstatistics_collector::collector::GenerateStatisticMessage(
node_name_,
collector->GetMetricName(),
collector->GetMetricUnit(),
window_start_,
window_end,
collected_stats);
publisher_->publish(message);
{
std::lock_guard<std::mutex> lock(mutex_);
for (auto & collector : subscriber_statistics_collectors_) {
const auto collected_stats = collector->GetStatisticsResults();

auto message = libstatistics_collector::collector::GenerateStatisticMessage(
node_name_,
collector->GetMetricName(),
collector->GetMetricUnit(),
window_start_,
window_end,
collected_stats);
msgs.push_back(message);
}
}

for (auto & msg : msgs) {
publisher_->publish(msg);
}
window_start_ = window_end;
}

protected:
/// Return a vector of all the currently collected data.
/**
* - This method acquires a lock to prevent race conditions to collectors list.
wjwwood marked this conversation as resolved.
Show resolved Hide resolved
*
* \return a vector of all the collected data
*/
std::vector<StatisticData> get_current_collector_data() const
{
std::vector<StatisticData> data;
std::lock_guard<std::mutex> lock(mutex_);
for (const auto & collector : subscriber_statistics_collectors_) {
data.push_back(collector->GetStatisticsResults());
}
Expand All @@ -151,23 +168,35 @@ class SubscriptionTopicStatistics

private:
/// Construct and start all collectors and set window_start_.
/**
* - This method acquires a lock to prevent race conditions to collectors list.
wjwwood marked this conversation as resolved.
Show resolved Hide resolved
*/
void bring_up()
{
auto received_message_period = std::make_unique<ReceivedMessagePeriod>();
received_message_period->Start();
subscriber_statistics_collectors_.emplace_back(std::move(received_message_period));
{
std::lock_guard<std::mutex> lock(mutex_);
subscriber_statistics_collectors_.emplace_back(std::move(received_message_period));
}

window_start_ = rclcpp::Time(get_current_nanoseconds_since_epoch());
}

/// Stop all collectors, clear measurements, stop publishing timer, and reset publisher.
/**
* - This method acquires a lock to prevent race conditions to collectors list.
wjwwood marked this conversation as resolved.
Show resolved Hide resolved
*/
void tear_down()
{
for (auto & collector : subscriber_statistics_collectors_) {
collector->Stop();
}
{
std::lock_guard<std::mutex> lock(mutex_);
for (auto & collector : subscriber_statistics_collectors_) {
collector->Stop();
}

subscriber_statistics_collectors_.clear();
subscriber_statistics_collectors_.clear();
}

if (publisher_timer_) {
publisher_timer_->cancel();
Expand All @@ -187,6 +216,8 @@ class SubscriptionTopicStatistics
return std::chrono::duration_cast<std::chrono::nanoseconds>(now.time_since_epoch()).count();
}

/// Mutex to protect the subsequence vectors
mutable std::mutex mutex_;
/// Collection of statistics collectors
std::vector<std::unique_ptr<TopicStatsCollector>> subscriber_statistics_collectors_{};
/// Node name used to generate topic statistics messages to be published
Expand Down