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

refactor(perception_rviz_plugin): apply thread pool to manage detached thread #1023

Merged
merged 1 commit into from
Nov 20, 2023
Merged
Show file tree
Hide file tree
Changes from all 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 @@ -24,9 +24,11 @@

#include <condition_variable>
#include <list>
#include <queue>
#include <set>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

namespace autoware
Expand All @@ -45,10 +47,31 @@ class AUTOWARE_AUTO_PERCEPTION_RVIZ_PLUGIN_PUBLIC PredictedObjectsDisplay
using PredictedObjects = autoware_auto_perception_msgs::msg::PredictedObjects;

PredictedObjectsDisplay();
~PredictedObjectsDisplay()
{
{
std::unique_lock<std::mutex> lock(queue_mutex);
should_terminate = true;
}
condition.notify_all();
for (std::thread & active_thread : threads) {
active_thread.join();
}
threads.clear();
}

private:
void processMessage(PredictedObjects::ConstSharedPtr msg) override;

void queueJob(std::function<void()> job)
{
{
std::unique_lock<std::mutex> lock(queue_mutex);
jobs.push(std::move(job));
}
condition.notify_one();
}

boost::uuids::uuid to_boost_uuid(const unique_identifier_msgs::msg::UUID & uuid_msg)
{
const std::string uuid_str = uuid_to_string(uuid_msg);
Expand Down Expand Up @@ -100,6 +123,8 @@ class AUTOWARE_AUTO_PERCEPTION_RVIZ_PLUGIN_PUBLIC PredictedObjectsDisplay
PredictedObjects::ConstSharedPtr msg);
void workerThread();

void messageProcessorThreadJob();

void update(float wall_dt, float ros_dt) override;

std::unordered_map<boost::uuids::uuid, int32_t, boost::hash<boost::uuids::uuid>> id_map;
Expand All @@ -108,6 +133,14 @@ class AUTOWARE_AUTO_PERCEPTION_RVIZ_PLUGIN_PUBLIC PredictedObjectsDisplay
int32_t marker_id = 0;
const int32_t PATH_ID_CONSTANT = 1e3;

// max_num_threads: number of threads created in the thread pool, hard-coded to be 1;
int max_num_threads;

bool should_terminate{false};
std::mutex queue_mutex;
std::vector<std::thread> threads;
std::queue<std::function<void()>> jobs;

PredictedObjects::ConstSharedPtr msg;
bool consumed{false};
std::mutex mutex;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,27 +25,44 @@ namespace object_detection
{
PredictedObjectsDisplay::PredictedObjectsDisplay() : ObjectPolygonDisplayBase("tracks")
{
std::thread worker(&PredictedObjectsDisplay::workerThread, this);
worker.detach();
max_num_threads = 1; // hard code the number of threads to be created

for (int ii = 0; ii < max_num_threads; ++ii) {
threads.emplace_back(std::thread(&PredictedObjectsDisplay::workerThread, this));
}
}

void PredictedObjectsDisplay::workerThread()
{
{ // A standard working thread that waiting for jobs
while (true) {
std::unique_lock<std::mutex> lock(mutex);
condition.wait(lock, [this] { return this->msg; });
std::function<void()> job;
{
std::unique_lock<std::mutex> lock(queue_mutex);
condition.wait(lock, [this] { return !jobs.empty() || should_terminate; });
if (should_terminate) {
return;
}
job = jobs.front();
jobs.pop();
}
job();
}
}

auto tmp_msg = this->msg;
this->msg.reset();
void PredictedObjectsDisplay::messageProcessorThreadJob()
{
// Receiving
std::unique_lock<std::mutex> lock(mutex);
auto tmp_msg = this->msg;
this->msg.reset();
lock.unlock();

lock.unlock();
auto tmp_markers = createMarkers(tmp_msg);

auto tmp_markers = createMarkers(tmp_msg);
lock.lock();
markers = tmp_markers;
lock.lock();
markers = tmp_markers;

consumed = true;
}
consumed = true;
}

std::vector<visualization_msgs::msg::Marker::SharedPtr> PredictedObjectsDisplay::createMarkers(
Expand Down Expand Up @@ -188,7 +205,7 @@ void PredictedObjectsDisplay::processMessage(PredictedObjects::ConstSharedPtr ms
std::unique_lock<std::mutex> lock(mutex);

this->msg = msg;
condition.notify_one();
queueJob(std::bind(&PredictedObjectsDisplay::messageProcessorThreadJob, this));
}

void PredictedObjectsDisplay::update(float wall_dt, float ros_dt)
Expand Down
Loading