Block the background thread when no handles are added

Currently, if parameter service is not enabled and no handles are added
into the wait set, the background thread will loop without doing
anything, which causes a high CPU load.

This patch implements that when there is no handles are attached to the
current node, the background thread will be blocked by a semaphore and
the main thread will signal it later when new handle has been created.

Fix #752

Author: Minggang Wang

lib/node_options.js

@@ -23,11 +23,18 @@ class NodeOptions {
   /**
    * Create a new instance with default property values.
    * @constructor
+   * @param {boolean} [startParameterServices=true]
+   * @param {array} [parameterOverrides=[]]
+   * @param {boolean} [automaticallyDeclareParametersFromOverrides=false]
    */
-  constructor() {
-    this._startParameterServices = true;
-    this._parameterOverrides = [];
-    this._automaticallyDeclareParametersFromOverrides = false;
+  constructor(
+    startParameterServices = true,
+    parameterOverrides = [],
+    automaticallyDeclareParametersFromOverrides = false
+  ) {
+    this._startParameterServices = startParameterServices;
+    this._parameterOverrides = parameterOverrides;
+    this._automaticallyDeclareParametersFromOverrides = automaticallyDeclareParametersFromOverrides;
   }
 
   /**

src/executor.cpp

@@ -45,7 +45,8 @@ Executor::Executor(HandleManager* handle_manager, Delegate* delegate)
     : async_(nullptr),
       handle_manager_(handle_manager),
       delegate_(delegate),
-      context_(nullptr) {
+      context_(nullptr),
+      main_thread_(uv_thread_self()) {
   running_.store(false);
 }
 
@@ -65,7 +66,7 @@ void Executor::Start(rcl_context_t* context, int32_t time_out) {
     // Mark flag before creating thread
     // Make sure thread can run
     running_.store(true);
-    uv_thread_create(&thread_, Executor::Run, this);
+    uv_thread_create(&background_thread_, Executor::Run, this);
   }
 }
 
@@ -95,7 +96,8 @@ void Executor::Stop() {
     // Stop thread first, and then uv_close
     // Make sure async_ is not used anymore
     running_.store(false);
-    uv_thread_join(&thread_);
+    handle_manager_->StopWaitingHandles();
+    uv_thread_join(&background_thread_);
 
     if (uv_is_active(reinterpret_cast<uv_handle_t*>(async_))) {
       static bool handle_closed = false;
@@ -114,6 +116,11 @@ void Executor::Stop() {
   }
 }
 
+bool Executor::IsMainThread() {
+  uv_thread_t this_thread = uv_thread_self();
+  return uv_thread_equal(&main_thread_, &this_thread) != 0;
+}
+
 void Executor::DoWork(uv_async_t* handle) {
   Executor* executor = reinterpret_cast<Executor*>(handle->data);
   executor->ExecuteReadyHandles();
@@ -158,12 +165,16 @@ void Executor::Run(void* arg) {
 
 RclResult Executor::WaitForReadyCallbacks(rcl_wait_set_t* wait_set,
                                           int32_t time_out) {
+  // Wait the handles on the background thread if there is none.
+  if (handle_manager_->sum() == 0 && !IsMainThread())
+    handle_manager_->WaitForHandles();
+
   if (handle_manager_->is_synchronizing())
     handle_manager_->WaitForSynchronizing();
+
   {
     ScopedReadWriteLock read_lock(handle_manager_->handle_rwlock(),
                                   ScopedReadWriteLock::LockType::kRead);
-    if (handle_manager_->is_empty()) return RclResult(RCL_RET_OK, "" /* msg */);
 
     size_t num_subscriptions = 0u;
     size_t num_guard_conditions = 0u;

src/executor.hpp

@@ -46,6 +46,7 @@ class Executor {
   void Stop();
   void SpinOnce(rcl_context_t* context, int32_t time_out);
   int32_t time_out() { return time_out_; }
+  bool IsMainThread();
 
   static void DoWork(uv_async_t* handle);
   static void Run(void* arg);
@@ -58,7 +59,12 @@ class Executor {
   void ExecuteReadyHandles();
 
   uv_async_t* async_;
-  uv_thread_t thread_;
+
+  // The v8 main thread.
+  uv_thread_t main_thread_;
+
+  // Sub thread used to query the ready handles.
+  uv_thread_t background_thread_;
 
   HandleManager* handle_manager_;
   Delegate* delegate_;

src/handle_manager.cpp

@@ -27,15 +27,15 @@ HandleManager::HandleManager() {
   is_synchronizing_.store(false);
   uv_rwlock_init(&sync_handles_rwlock_);
   uv_rwlock_init(&ready_handles_rwlock_);
-  uv_mutex_init(&mutex_);
-  uv_sem_init(&sem_, 0);
+  uv_sem_init(&sync_handle_sem_, 0);
+  uv_sem_init(&wait_handle_sem_, 0);
 }
 
 HandleManager::~HandleManager() {
-  uv_mutex_destroy(&mutex_);
   uv_rwlock_destroy(&sync_handles_rwlock_);
   uv_rwlock_destroy(&ready_handles_rwlock_);
-  uv_sem_destroy(&sem_);
+  uv_sem_destroy(&sync_handle_sem_);
+  uv_sem_destroy(&wait_handle_sem_);
 }
 
 void HandleManager::SynchronizeHandles(const v8::Local<v8::Object> node) {
@@ -54,38 +54,44 @@ void HandleManager::SynchronizeHandles(const v8::Local<v8::Object> node) {
       Nan::Get(node, Nan::New("_actionClients").ToLocalChecked());
   Nan::MaybeLocal<v8::Value> action_servers =
       Nan::Get(node, Nan::New("_actionServers").ToLocalChecked());
-  is_synchronizing_.store(true);
 
+  uint32_t sum = 0;
+  is_synchronizing_.store(true);
   {
     ScopedReadWriteLock scoped_lock(&sync_handles_rwlock_,
                                     ScopedReadWriteLock::LockType::kWrite);
     ClearHandles();
-
-    SynchronizeHandlesByType(
+    sum += SynchronizeHandlesByType(
         Nan::To<v8::Object>(timers.ToLocalChecked()).ToLocalChecked(),
         &timers_);
-    SynchronizeHandlesByType(
+    sum += SynchronizeHandlesByType(
         Nan::To<v8::Object>(subscriptions.ToLocalChecked()).ToLocalChecked(),
         &subscriptions_);
-    SynchronizeHandlesByType(
+    sum += SynchronizeHandlesByType(
         Nan::To<v8::Object>(clients.ToLocalChecked()).ToLocalChecked(),
         &clients_);
-    SynchronizeHandlesByType(
+    sum += SynchronizeHandlesByType(
         Nan::To<v8::Object>(services.ToLocalChecked()).ToLocalChecked(),
         &services_);
-    SynchronizeHandlesByType(
+    sum += SynchronizeHandlesByType(
         Nan::To<v8::Object>(guard_conditions.ToLocalChecked()).ToLocalChecked(),
         &guard_conditions_);
-    SynchronizeHandlesByType(
+    sum += SynchronizeHandlesByType(
         Nan::To<v8::Object>(action_clients.ToLocalChecked()).ToLocalChecked(),
         &action_clients_);
-    SynchronizeHandlesByType(
+    sum += SynchronizeHandlesByType(
         Nan::To<v8::Object>(action_servers.ToLocalChecked()).ToLocalChecked(),
         &action_servers_);
   }
-
   is_synchronizing_.store(false);
-  uv_sem_post(&sem_);
+
+  // Signals that the synchronization has finished.
+  uv_sem_post(&sync_handle_sem_);
+
+  // Wakeup the backgroud thread if the sum was zero, but now it is greater than
+  // zero.
+  uint32_t sum_was = sum_.exchange(sum);
+  if (sum_was == 0 && sum_ > 0) uv_sem_post(&wait_handle_sem_);
 
   RCLNODEJS_DEBUG(
       "Add %lu timers, %lu subscriptions, %lu clients, %lu services, %lu "
@@ -94,6 +100,12 @@ void HandleManager::SynchronizeHandles(const v8::Local<v8::Object> node) {
       guard_conditions_.size());
 }
 
+void HandleManager::WaitForSynchronizing() { uv_sem_wait(&sync_handle_sem_); }
+
+void HandleManager::WaitForHandles() { uv_sem_wait(&wait_handle_sem_); }
+
+void HandleManager::StopWaitingHandles() { uv_sem_post(&wait_handle_sem_); }
+
 void HandleManager::ClearHandles() {
   timers_.clear();
   clients_.clear();
@@ -252,7 +264,7 @@ uint32_t HandleManager::ready_handles_count() {
   return ready_handles_.size();
 }
 
-void HandleManager::SynchronizeHandlesByType(
+uint32_t HandleManager::SynchronizeHandlesByType(
     const v8::Local<v8::Object>& typeObject,
     std::vector<rclnodejs::RclHandle*>* vec) {
   Nan::HandleScope scope;
@@ -276,6 +288,7 @@ void HandleManager::SynchronizeHandlesByType(
       vec->push_back(rcl_handle);
     }
   }
+  return vec->size();
 }
 
 template <typename T>

src/handle_manager.hpp

@@ -54,7 +54,13 @@ class HandleManager {
   ~HandleManager();
 
   void SynchronizeHandles(const v8::Local<v8::Object> node);
-  void WaitForSynchronizing() { uv_sem_wait(&sem_); }
+  void WaitForSynchronizing();
+
+  // Waits the handles to be attached from the background thread.
+  void WaitForHandles();
+  // Signals to stop waiting the handles from the main thread.
+  void StopWaitingHandles();
+
   void ClearHandles();
 
   rcl_ret_t AddHandlesToWaitSet(rcl_wait_set_t* wait_set);
@@ -70,21 +76,21 @@ class HandleManager {
   uint32_t guard_condition_count() const { return guard_conditions_.size(); }
   uv_rwlock_t* handle_rwlock() { return &sync_handles_rwlock_; }
 
-  std::vector<rclnodejs::RclHandle*> TakeReadyHandles();
-
   uint32_t ready_handles_count();
 
+  // Takes the ownership of the handles that are ready to be taken.
+  std::vector<rclnodejs::RclHandle*> TakeReadyHandles();
+
+  // Thread safe function to get if the handles are being synchronized.
   bool is_synchronizing() const { return is_synchronizing_.load(); }
-  bool is_empty() const {
-    return subscriptions_.size() == 0 && services_.size() == 0 &&
-           clients_.size() == 0 && timers_.size() == 0 &&
-           guard_conditions_.size() == 0 && action_clients_.size() == 0 &&
-           action_servers_.size() == 0;
-  }
+
+  // Thread safe function to get the sum of the handles.
+  uint32_t sum() const { return sum_; }
 
  protected:
-  void SynchronizeHandlesByType(const v8::Local<v8::Object>& typeObject,
-                                std::vector<rclnodejs::RclHandle*>* vec);
+  // Synchronize the handles from `typeObject`.
+  uint32_t SynchronizeHandlesByType(const v8::Local<v8::Object>& typeObject,
+                                    std::vector<rclnodejs::RclHandle*>* vec);
   template <typename T>
   void CollectReadyHandlesByType(
       const T** struct_ptr, size_t size,
@@ -104,11 +110,24 @@ class HandleManager {
   std::vector<rclnodejs::RclHandle*> action_clients_;
   std::vector<rclnodejs::RclHandle*> ready_handles_;
 
-  uv_mutex_t mutex_;
+  // Protects the handles.
   uv_rwlock_t sync_handles_rwlock_;
   uv_rwlock_t ready_handles_rwlock_;
-  uv_sem_t sem_;
+
+  // Used to wait for handles added from the background thread.
+  uv_sem_t wait_handle_sem_;
+
+  // Used to wait from the background thread when the handles are being
+  // synchronized on the main thread.
+  uv_sem_t sync_handle_sem_;
+
+  // Atomically written from the main thread only, but read from background
+  // thread.
   std::atomic_bool is_synchronizing_;
+
+  // Sum of the handles attached to the current node, atomically written from
+  // the main thread.
+  std::atomic_uint32_t sum_;
 };
 
 }  // namespace rclnodejs

test/test-node.js

@@ -18,7 +18,9 @@ const IsClose = require('is-close');
 const assert = require('assert');
 const rclnodejs = require('../index.js');
 const assertUtils = require('./utils.js');
+const { NodeOptions } = require('../index.js');
 const assertThrowsError = assertUtils.assertThrowsError;
+const Context = require('../lib/context.js');
 
 describe('rclnodejs node test suite', function () {
   this.timeout(60 * 1000);
@@ -462,3 +464,45 @@ describe('topic & serviceName getter/setter', function () {
     node.destroy();
   });
 });
+
+describe('Test the node with no handles attached when initializing', function () {
+  this.timeout(60 * 1000);
+
+  before(function () {
+    return rclnodejs.init();
+  });
+
+  after(function () {
+    rclnodejs.shutdown();
+  });
+
+  it('Publish a topic after initialization', function (done) {
+    // Init the node with no parameter services.
+    const node = rclnodejs.createNode(
+      'publisher',
+      '/topic_getter',
+      Context.defaultContext(),
+      new NodeOptions(false, [], false)
+    );
+    const str = 'hello world';
+    rclnodejs.spin(node);
+
+    setTimeout(() => {
+      const publisher = node.createPublisher('std_msgs/msg/String', 'chatter');
+      publisher.publish(str);
+    }, 200);
+
+    setTimeout(() => {
+      // The backgroud thread should get waken up when the subscription is attached.
+      const subscription = node.createSubscription(
+        'std_msgs/msg/String',
+        'chatter',
+        (msg) => {
+          assert.deepStrictEqual(msg.data, str);
+          node.destroy();
+          done();
+        }
+      );
+    }, 100);
+  });
+});