Rework dbengine async wakeup on windows (#21003)
Stelios Fragkakis committed
Sep 19, 2025 at 21:20 UTC
e147fcbdb51c3465776b978befe8e5265247f9f0
1 file changed
+85
-3
src/database/engine/rrdengine.c
+85
-3
@@ -34,6 +34,9 @@ struct rrdeng_main {
34
ND_THREAD *thread;
35
uv_loop_t loop;
36
uv_async_t async;
37
+#if defined(OS_WINDOWS)
38
+ bool async_ready;
39
+#endif
40
uv_timer_t timer;
41
uv_timer_t retention_timer;
42
pid_t tid;
@@ -95,6 +98,42 @@ struct rrdeng_main {
98
}
99
};
100
101
+#if defined(OS_WINDOWS)
102
+netdata_mutex_t rrdeng_async_mutex;
103
+
104
+__attribute__((constructor)) void initialize_rrdeng_async_mutex(void)
105
+{
106
+ netdata_mutex_init(&rrdeng_async_mutex);
107
+}
108
+
109
+__attribute__((destructor)) void destroy_rrdeng_async_mutex(void)
110
+{
111
+ netdata_mutex_destroy(&rrdeng_async_mutex);
112
+}
113
+
114
+void rrdeng_async_wakeup()
115
+{
116
+
117
+ if (__atomic_load_n(&rrdeng_main.async_ready, __ATOMIC_RELAXED)) {
118
+ netdata_mutex_lock(&rrdeng_async_mutex);
119
+ int rc = uv_async_send(&rrdeng_main.async);
120
+ if (rc)
121
+ nd_log_daemon(NDLP_ERR,"DBENGINE: wakeup async error = %d", rc);
122
+
123
+ netdata_mutex_unlock(&rrdeng_async_mutex);
124
+ } else {
125
+ nd_log_daemon(NDLP_WARNING,"DBENGINE: wakeup async handler is being reset");
126
+ }
127
+}
128
+#else
129
+void rrdeng_async_wakeup()
130
+{
131
+ int rc = uv_async_send(&rrdeng_main.async);
132
+ if (rc)
133
+ nd_log_daemon(NDLP_ERR,"DBENGINE: wakeup async error = %d", rc);
134
+}
135
+#endif
136
+
137
static void sanity_check(void)
138
{
139
BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < (RRDENG_OPCODE_MAX + 2));
@@ -233,7 +272,7 @@ static void work_standard_worker(uv_work_t *req) {
272
__atomic_sub_fetch(&rrdeng_main.work_cmd.atomics.executing, 1, __ATOMIC_RELAXED);
273
274
// signal the event loop a worker is available
236
- fatal_assert(0 == uv_async_send(&rrdeng_main.async));
275
+ rrdeng_async_wakeup();
276
}
277
278
static void after_work_standard_callback(uv_work_t* req, int status) {
@@ -523,7 +562,7 @@ ALWAYS_INLINE void rrdeng_enq_cmd(struct rrdengine_instance *ctx, enum rrdeng_op
562
enqueue_cb(cmd);
563
spinlock_unlock(&rrdeng_main.cmd_queue.unsafe.spinlock);
564
526
- fatal_assert(0 == uv_async_send(&rrdeng_main.async));
565
+ rrdeng_async_wakeup();
566
}
567
568
static inline bool rrdeng_cmd_has_waiting_opcodes_in_lower_priorities(STORAGE_PRIORITY priority, STORAGE_PRIORITY max_priority) {
@@ -1818,12 +1857,36 @@ void finalize_rrd_files(struct rrdengine_instance *ctx)
1857
return finalize_data_files(ctx);
1858
}
1859
1860
+#if defined(OS_WINDOWS)
1861
+uint64_t last_async_callback;
1862
+
1863
+void async_cb(uv_async_t *handle)
1864
+{
1865
+ uv_stop(handle->loop);
1866
+ uv_update_time(handle->loop);
1867
+
1868
+ last_async_callback = uv_hrtime();
1869
+
1870
+ netdata_log_debug(D_RRDENGINE, "%s called, active=%d.", __func__, uv_is_active((uv_handle_t *)handle));
1871
+}
1872
+
1873
+static void async_closed_cb(uv_handle_t *handle)
1874
+{
1875
+ struct rrdeng_main *main = handle->data;
1876
+
1877
+ int ret = uv_async_init(handle->loop, &main->async, async_cb);
1878
+ if (ret)
1879
+ netdata_log_error("DBENGINE: reinitializing uv_async_init(): %s", uv_strerror(ret));
1880
+ __atomic_store_n(&main->async_ready, true, __ATOMIC_RELEASE);
1881
+}
1882
+#else
1883
void async_cb(uv_async_t *handle)
1884
{
1885
uv_stop(handle->loop);
1886
uv_update_time(handle->loop);
1887
netdata_log_debug(D_RRDENGINE, "%s called, active=%d.", __func__, uv_is_active((uv_handle_t *)handle));
1888
}
1889
+#endif
1890
1891
#define TIMER_PERIOD_MS (1000)
1892
@@ -2175,6 +2238,9 @@ bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx __maybe_unused) {
2238
return false;
2239
}
2240
rrdeng_main.async.data = &rrdeng_main;
2241
+#if defined(OS_WINDOWS)
2242
+ rrdeng_main.async_ready = true;
2243
+#endif
2244
2245
ret = uv_timer_init(&rrdeng_main.loop, &rrdeng_main.timer);
2246
if (ret) {
@@ -2340,6 +2406,10 @@ void dbengine_event_loop(void* arg) {
2406
mlt[i].finished = false;
2407
}
2408
2409
+#if defined(OS_WINDOWS)
2410
+ last_async_callback = uv_hrtime();
2411
+#endif
2412
+
2413
while (likely(!shutdown)) {
2414
worker_is_idle();
2415
uv_run(&main->loop, UV_RUN_DEFAULT);
@@ -2365,7 +2435,19 @@ void dbengine_event_loop(void* arg) {
2435
worker_dispatch_extent_read(cmd, false);
2436
break;
2437
2368
- case RRDENG_OPCODE_QUERY:
2438
+ case RRDENG_OPCODE_QUERY:;
2439
+#if defined(OS_WINDOWS)
2440
+ static int max_timeout_count = 0;
2441
+ if (uv_hrtime() - last_async_callback > 1000UL * NSEC_PER_MSEC) {
2442
+ if (++max_timeout_count > 30) {
2443
+ netdata_log_error("DBENGINE: async callback timeout detected, re-initializing the async handle");
2444
+ __atomic_store_n(&main->async_ready, false, __ATOMIC_RELEASE);
2445
+ uv_close((uv_handle_t *)&main->async, async_closed_cb);
2446
+ max_timeout_count = 0;
2447
+ } else
2448
+ netdata_log_error("DBENGINE: async callback timeout detected count = %d", max_timeout_count);
2449
+ }
2450
+#endif
2451
worker_dispatch_query_prep(cmd, false);
2452
break;
2453