src: replace duplicate loop hook regs

Replace blocked and unblocked loop hook entries when the same callback
is registered again.

These hooks were stored in append-only TSList containers, so dynamic
reconfiguration in agents like gRPC and ZMQ kept stale registrations
alive. That caused duplicate callback delivery and let old blocked loop
thresholds continue affecting detection.

Add TSList::replace_if() and use it in blocked and unblocked loop
hook registration. When the callback function pointer matches an
existing entry, overwrite it instead of appending a new one.

For blocked loop hooks, recompute min_blocked_threshold_ after each
registration so a replaced threshold immediately becomes effective.

PR-URL: #444
Reviewed-By: Rafael Gonzaga <rafael.nunu@hotmail.com>
Reviewed-By: EHortua <55801532+EHortua@users.noreply.github.com>

Author: santigimeno

src/nsolid/nsolid_api.cc

@@ -17,6 +17,7 @@
 #include "node_url.h"
 #include "v8-fast-api-calls.h"
 
+#include <algorithm>
 #include <cmath>
 
 #if defined(__linux__)
@@ -967,10 +968,12 @@ void EnvList::OnBlockedLoopHook(
     void* data,
     internal::on_block_loop_hook_proxy_sig proxy,
     internal::deleter_sig deleter) {
-  blocked_hooks_list_.push_back(
+  blocked_hooks_list_.replace_if(
+    [proxy](const BlockedLoopStor& stor) {
+      return stor.cb == proxy;
+    },
     { threshold, proxy, nsolid::internal::user_data(data, deleter) });
-  if (threshold < min_blocked_threshold_)
-    min_blocked_threshold_ = threshold;
+  refresh_min_blocked_threshold();
 }
 
 void EnvList::OnUnblockedLoopHook(
@@ -979,7 +982,10 @@ void EnvList::OnUnblockedLoopHook(
     internal::deleter_sig deleter) {
   // Using BlockedLoopStor because it's easier than duplicating a bunch of code,
   // but that means some value needs to be passed in for threshold.
-  unblocked_hooks_list_.push_back(
+  unblocked_hooks_list_.replace_if(
+    [proxy](const BlockedLoopStor& stor) {
+      return stor.cb == proxy;
+    },
     { 0, proxy, nsolid::internal::user_data(data, deleter) });
 }
 
@@ -1010,6 +1016,17 @@ nlohmann::json EnvList::CurrentConfigJSON() {
 }
 
 
+void EnvList::refresh_min_blocked_threshold() {
+  uint64_t min_threshold = UINT64_MAX;
+
+  blocked_hooks_list_.for_each([&min_threshold](const BlockedLoopStor& stor) {
+    min_threshold = std::min(min_threshold, stor.threshold);
+  });
+
+  min_blocked_threshold_ = min_threshold;
+}
+
+
 void EnvList::AddEnv(Environment* env) {
   SharedEnvInst envinst_sp = EnvInst::Create(env);
 

src/nsolid/nsolid_api.h

@@ -633,6 +633,7 @@ class EnvList {
   void fill_trace_id_q();
 
   void update_continuous_profiler(bool enabled, uint64_t interval);
+  void refresh_min_blocked_threshold();
 
 #ifdef __POSIX__
   static void signal_handler_(int signum, siginfo_t* info, void* ucontext);

src/nsolid/thread_safe.h

@@ -142,6 +142,18 @@ struct TSList {
     list_.push_back(std::move(data));
     return --list_.end();
   }
+  template <typename Match>
+  inline bool replace_if(Match match, DataType&& data) {
+    nsuv::ns_mutex::scoped_lock lock(lock_);
+    for (auto it = list_.begin(); it != list_.end(); ++it) {
+      if (!match(*it))
+        continue;
+      *it = std::move(data);
+      return true;
+    }
+    list_.push_back(std::move(data));
+    return false;
+  }
   inline void for_each(std::function<void(const DataType&)> fn) {
     nsuv::ns_mutex::scoped_lock lock(lock_);
     std::for_each(list_.begin(), list_.end(), fn);

test/agents/test-grpc-blocked-loop.mjs

@@ -143,7 +143,43 @@ function checkUnblockedLoopData(blocked, metadata, agentId, threadId, bInfo) {
 const tests = [];
 
 tests.push({
-  name: 'should work in the main thread',
+  name: 'should work in the main thread with default threshold of 200ms',
+  test: async (getEnv) => {
+    return new Promise((resolve) => {
+      let times = 0;
+      const grpcServer = new GRPCServer();
+      grpcServer.start(mustSucceed(async (port) => {
+        grpcServer.on('loop_blocked', mustCall(async (data) => {
+          checkBlockedLoopData(data.msg, data.metadata, agentId, threadId);
+        }, 2));
+
+        grpcServer.on('loop_unblocked', mustCall(async (data) => {
+          checkUnblockedLoopData(data.msg, data.metadata, agentId, threadId);
+          if (++times === 2) {
+            await child.shutdown(0);
+            grpcServer.close();
+            resolve();
+          } else {
+            await child.block(0, 400);
+          }
+        }, 2));
+
+        const env = getEnv(port);
+
+        const opts = {
+          stdio: ['inherit', 'inherit', 'inherit', 'ipc'],
+          env,
+        };
+        const child = new TestClient([], opts);
+        const agentId = await child.id();
+        await child.block(0, 400);
+      }));
+    });
+  },
+});
+
+tests.push({
+  name: 'should work in the main thread with different threshold',
   test: async (getEnv) => {
     return new Promise((resolve) => {
       const grpcServer = new GRPCServer();
@@ -163,18 +199,62 @@ tests.push({
 
         const opts = {
           stdio: ['inherit', 'inherit', 'inherit', 'ipc'],
-          env,
+          env: {
+            ...env,
+            NSOLID_BLOCKED_LOOP_THRESHOLD: '1000',
+          },
         };
         const child = new TestClient([], opts);
         const agentId = await child.id();
         await child.block(0, 400);
+        setTimeout(() => {
+          child.block(0, 1500);
+        }, 500);
+      }));
+    });
+  },
+});
+
+tests.push({
+  name: 'should work for workers with default threshold of 200ms',
+  test: async (getEnv) => {
+    return new Promise((resolve) => {
+      let times = 0;
+      const grpcServer = new GRPCServer();
+      grpcServer.start(mustSucceed(async (port) => {
+        grpcServer.on('loop_blocked', mustCall(async (data) => {
+          checkBlockedLoopData(data.msg, data.metadata, agentId, wid);
+        }, 2));
+
+        grpcServer.on('loop_unblocked', mustCall(async (data) => {
+          checkUnblockedLoopData(data.msg, data.metadata, agentId, wid);
+          if (++times === 2) {
+            await child.shutdown(0);
+            grpcServer.close();
+            resolve();
+          } else {
+            await child.block(wid, 800);
+          }
+        }, 2));
+
+        const env = getEnv(port);
+
+        const opts = {
+          stdio: ['inherit', 'inherit', 'inherit', 'ipc'],
+          env,
+        };
+        const child = new TestClient([ '-w', 1 ], opts);
+        const agentId = await child.id();
+        const workers = await child.workers();
+        const wid = workers[0];
+        await child.block(wid, 400);
       }));
     });
   },
 });
 
 tests.push({
-  name: 'should work for workers',
+  name: 'should work for workers with different threshold',
   test: async (getEnv) => {
     return new Promise((resolve) => {
       const grpcServer = new GRPCServer();
@@ -194,13 +274,19 @@ tests.push({
 
         const opts = {
           stdio: ['inherit', 'inherit', 'inherit', 'ipc'],
-          env,
+          env: {
+            ...env,
+            NSOLID_BLOCKED_LOOP_THRESHOLD: '1000',
+          },
         };
         const child = new TestClient([ '-w', 1 ], opts);
         const agentId = await child.id();
         const workers = await child.workers();
         const wid = workers[0];
         await child.block(wid, 400);
+        setTimeout(() => {
+          child.block(wid, 1500);
+        }, 500);
       }));
     });
   },