| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include <daemon/main.h> |
| 4 | #include "libuv_workers.h" |
| 5 | |
| 6 | static void register_libuv_worker_jobs_internal(void) { |
| 7 | signals_block_all_except_deadly(); |
| 8 | |
| 9 | worker_register("LIBUV"); |
| 10 | |
| 11 | // generic |
| 12 | worker_register_job_name(UV_EVENT_WORKER_INIT, "worker init"); |
| 13 | |
| 14 | // query related |
| 15 | worker_register_job_name(UV_EVENT_DBENGINE_QUERY, "query"); |
| 16 | worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_CACHE_LOOKUP, "extent cache"); |
| 17 | worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_MMAP, "extent mmap"); |
| 18 | worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_DECOMPRESSION, "extent decompression"); |
| 19 | worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_LOOKUP, "page lookup"); |
| 20 | worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_POPULATION, "page populate"); |
| 21 | worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_PAGE_ALLOCATION, "page allocate"); |
| 22 | // Weights |
| 23 | worker_register_job_name(UV_EVENT_WEIGHTS_CALCULATION, "weights calculation"); |
| 24 | |
| 25 | // flushing related |
| 26 | worker_register_job_name(UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE, "flush main"); |
| 27 | worker_register_job_name(UV_EVENT_DBENGINE_EXTENT_WRITE, "extent write"); |
| 28 | worker_register_job_name(UV_EVENT_DBENGINE_FLUSHED_TO_OPEN, "flushed to open"); |
| 29 | |
| 30 | // datafile full |
| 31 | worker_register_job_name(UV_EVENT_DBENGINE_JOURNAL_INDEX, "jv2 indexing"); |
| 32 | |
| 33 | // db rotation related |
| 34 | worker_register_job_name(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT, "datafile delete wait"); |
| 35 | worker_register_job_name(UV_EVENT_DBENGINE_DATAFILE_DELETE, "datafile deletion"); |
| 36 | worker_register_job_name(UV_EVENT_DBENGINE_FIND_ROTATED_METRICS, "find rotated metrics"); |
| 37 | worker_register_job_name(UV_EVENT_DBENGINE_FIND_REMAINING_RETENTION, "find remaining retention"); |
| 38 | worker_register_job_name(UV_EVENT_DBENGINE_POPULATE_MRG, "update retention"); |
| 39 | |
| 40 | // other dbengine events |
| 41 | worker_register_job_name(UV_EVENT_DBENGINE_EVICT_MAIN_CACHE, "evict main"); |
| 42 | worker_register_job_name(UV_EVENT_DBENGINE_EVICT_OPEN_CACHE, "evict open"); |
| 43 | worker_register_job_name(UV_EVENT_DBENGINE_EVICT_EXTENT_CACHE, "evict extent"); |
| 44 | worker_register_job_name(UV_EVENT_DBENGINE_BUFFERS_CLEANUP, "dbengine buffers cleanup"); |
| 45 | worker_register_job_name(UV_EVENT_DBENGINE_FLUSH_DIRTY, "dbengine flush dirty"); |
| 46 | worker_register_job_name(UV_EVENT_DBENGINE_QUIESCE, "dbengine quiesce"); |
| 47 | worker_register_job_name(UV_EVENT_DBENGINE_SHUTDOWN, "dbengine shutdown"); |
| 48 | worker_register_job_name(UV_EVENT_DBENGINE_MRG_LOAD, "jv2 mrg load"); |
| 49 | |
| 50 | // metadata |
| 51 | worker_register_job_name(UV_EVENT_HOST_CONTEXT_LOAD, "metadata load host context"); |
| 52 | worker_register_job_name(UV_EVENT_METADATA_STORE, "metadata store host"); |
| 53 | worker_register_job_name(UV_EVENT_METADATA_CLEANUP, "metadata cleanup"); |
| 54 | worker_register_job_name(UV_EVENT_METADATA_ML_LOAD, "metadata load ml models"); |
| 55 | worker_register_job_name(UV_EVENT_CTX_CLEANUP_SCHEDULE, "metadata ctx cleanup schedule"); |
| 56 | worker_register_job_name(UV_EVENT_CTX_CLEANUP, "metadata ctx cleanup"); |
| 57 | worker_register_job_name(UV_EVENT_STORE_ALERT_TRANSITIONS, "metadata store alert transitions"); |
| 58 | worker_register_job_name(UV_EVENT_STORE_SQL_STATEMENTS, "metadata store sql statements"); |
| 59 | worker_register_job_name(UV_EVENT_CHART_LABEL_CLEANUP, "metadata chart label cleanup"); |
| 60 | worker_register_job_name(UV_EVENT_HEALTH_LOG_CLEANUP, "alert transitions cleanup"); |
| 61 | worker_register_job_name(UV_EVENT_UUID_DELETION, "metadata dimension deletion"); |
| 62 | worker_register_job_name(UV_EVENT_DIMENSION_CLEANUP, "metadata dimension cleanup"); |
| 63 | worker_register_job_name(UV_EVENT_CHART_CLEANUP, "metadata chart cleanup"); |
| 64 | worker_register_job_name(UV_EVENT_STORE_HOST, "metadata store host"); |
| 65 | worker_register_job_name(UV_EVENT_STORE_CHART, "metadata store chart"); |
| 66 | worker_register_job_name(UV_EVENT_STORE_DIMENSION, "metadata store dimension"); |
| 67 | |
| 68 | // aclk_sync |
| 69 | worker_register_job_name(UV_EVENT_ACLK_NODE_INFO, "aclk host node info"); |
| 70 | worker_register_job_name(UV_EVENT_ACLK_ALERT_PUSH, "aclk alert push"); |
| 71 | worker_register_job_name(UV_EVENT_ACLK_QUERY_EXECUTE, "aclk query execute"); |
| 72 | // aclk |
| 73 | worker_register_job_name(UV_EVENT_CTX_STOP_STREAMING, "ctx stop streaming"); |
| 74 | worker_register_job_name(UV_EVENT_CTX_CHECKPOINT, "ctx version check"); |
| 75 | worker_register_job_name(UV_EVENT_ALARM_PROVIDE_CFG, "send alarm config"); |
| 76 | worker_register_job_name(UV_EVENT_ALARM_SNAPSHOT, "alert snapshot"); |
| 77 | worker_register_job_name(UV_EVENT_REGISTER_NODE, "register node"); |
| 78 | worker_register_job_name(UV_EVENT_UPDATE_NODE_COLLECTORS, "update collectors"); |
| 79 | worker_register_job_name(UV_EVENT_UPDATE_NODE_INFO, "send node info"); |
| 80 | worker_register_job_name(UV_EVENT_CTX_SEND_SNAPSHOT, "ctx send snapshot"); |
| 81 | worker_register_job_name(UV_EVENT_CTX_SEND_SNAPSHOT_UPD, "ctx send update"); |
| 82 | worker_register_job_name(UV_EVENT_NODE_STATE_UPDATE, "node state update"); |
| 83 | worker_register_job_name(UV_EVENT_SEND_NODE_INSTANCES, "send node instances"); |
| 84 | worker_register_job_name(UV_EVENT_ALERT_START_STREAMING, "alert start streaming"); |
| 85 | worker_register_job_name(UV_EVENT_ALERT_CHECKPOINT, "alert checkpoint"); |
| 86 | worker_register_job_name(UV_EVENT_CREATE_NODE_INSTANCE, "create node instance"); |
| 87 | worker_register_job_name(UV_EVENT_UNREGISTER_NODE, "unregister node locally"); |
| 88 | |
| 89 | // netdatacli |
| 90 | worker_register_job_name(UV_EVENT_SCHEDULE_CMD, "schedule command"); |
| 91 | |
| 92 | // maintenance |
| 93 | worker_register_job_name(UV_EVENT_CLEANUP_OBSOLETE_CHARTS, "cleanup obsolete charts"); |
| 94 | worker_register_job_name(UV_EVENT_ARCHIVE_CHART_DIMENSIONS, "archive chart dimensions"); |
| 95 | worker_register_job_name(UV_EVENT_ARCHIVE_DIMENSION, "archive dimension"); |
| 96 | worker_register_job_name(UV_EVENT_CLEANUP_ORPHAN_HOSTS, "cleanup orphan hosts"); |
| 97 | worker_register_job_name(UV_EVENT_CLEANUP_OBSOLETE_CHARTS_ON_HOSTS, "cleanup obsolete charts on all hosts"); |
| 98 | worker_register_job_name(UV_EVENT_FREE_HOST, "free host"); |
| 99 | worker_register_job_name(UV_EVENT_FREE_CHART, "free chart"); |
| 100 | worker_register_job_name(UV_EVENT_FREE_DIMENSION, "free dimension"); |
| 101 | |
| 102 | // make sure we have the right thread id |
| 103 | gettid_uncached(); |
| 104 | |
| 105 | static int workers = 0; |
| 106 | int worker_id = __atomic_add_fetch(&workers, 1, __ATOMIC_RELAXED); |
| 107 | |
| 108 | char buf[NETDATA_THREAD_TAG_MAX + 1]; |
| 109 | snprintfz(buf, NETDATA_THREAD_TAG_MAX, "UV_WORKER[%d]", worker_id); |
| 110 | uv_thread_set_name_np(buf); |
| 111 | } |
| 112 | |
| 113 | // Register workers |
| 114 | ALWAYS_INLINE |
| 115 | void register_libuv_worker_jobs() { |
| 116 | static __thread bool registered = false; |
| 117 | |
| 118 | if(likely(registered)) |
| 119 | return; |
| 120 | |
| 121 | registered = true; |
| 122 | register_libuv_worker_jobs_internal(); |
| 123 | } |
| 124 | |
| 125 | void libuv_close_callback(uv_handle_t *handle, void *data __maybe_unused) |
| 126 | { |
| 127 | // Only close handles that aren't already closing |
| 128 | if (!uv_is_closing(handle)) { |
| 129 | if (handle->type == UV_TIMER) { |
| 130 | uv_timer_stop((uv_timer_t *)handle); |
| 131 | } |
| 132 | uv_close(handle, NULL); |
| 133 | } |
| 134 | } |
| 135 | |
| 136 | // Initialize the worker pool |
| 137 | void init_worker_pool(WorkerPool *pool) { |
| 138 | for (int i = 0; i < MAX_ACTIVE_WORKERS; i++) { |
| 139 | pool->workers[i].allocated = false; |
| 140 | pool->free_stack[i] = i; // Fill the stack with indices |
| 141 | } |
| 142 | pool->top = MAX_ACTIVE_WORKERS; // All workers are initially free |
| 143 | } |
| 144 | |
| 145 | // Get a worker from the pool |
| 146 | // Needs to be called from the uv event loop thread |
| 147 | worker_data_t *get_worker(WorkerPool *pool) { |
| 148 | worker_data_t *worker; |
| 149 | if (pool->top == 0) { |
| 150 | worker = callocz(1, sizeof(worker_data_t)); |
| 151 | worker->allocated = true; // Mark as allocated |
| 152 | } else { |
| 153 | int index = pool->free_stack[--pool->top]; // Pop from stack |
| 154 | worker = &pool->workers[index]; |
| 155 | } |
| 156 | worker->request.data = worker; |
| 157 | return worker; |
| 158 | } |
| 159 | |
| 160 | // Return a worker for reuse |
| 161 | void return_worker(WorkerPool *pool, worker_data_t *worker) { |
| 162 | if (unlikely(worker->allocated)) { |
| 163 | freez(worker); |
| 164 | return; |
| 165 | } |
| 166 | |
| 167 | int index = (int) (worker - pool->workers); |
| 168 | if (index < 0 || index >= MAX_ACTIVE_WORKERS) { |
| 169 | return; // Invalid worker (should not happen) |
| 170 | } |
| 171 | pool->free_stack[pool->top++] = index; // Push index back to stack |
| 172 | } |
| 173 | |
| 174 | // Initialize the command pool |
| 175 | void init_cmd_pool(CmdPool *pool, int size) { |
| 176 | pool->buffer = mallocz(sizeof(cmd_data_t) * size); |
| 177 | |
| 178 | pool->size = size; |
| 179 | pool->head = 0; |
| 180 | pool->tail = 0; |
| 181 | pool->count = 0; |
| 182 | |
| 183 | fatal_assert(0 == netdata_mutex_init(&pool->lock)); |
| 184 | fatal_assert(0 == netdata_cond_init(&pool->not_full)); |
| 185 | } |
| 186 | |
| 187 | bool push_cmd(CmdPool *pool, const cmd_data_t *cmd, bool wait_on_full) |
| 188 | { |
| 189 | netdata_mutex_lock(&pool->lock); |
| 190 | |
| 191 | while (pool->count == pool->size) { |
| 192 | if (wait_on_full) |
| 193 | netdata_cond_wait(&pool->not_full, &pool->lock); |
| 194 | else { |
| 195 | netdata_mutex_unlock(&pool->lock); // No space, return |
| 196 | return false; |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | pool->buffer[pool->tail] = *cmd; |
| 201 | pool->tail = (pool->tail + 1) % pool->size; |
| 202 | pool->count++; |
| 203 | |
| 204 | netdata_mutex_unlock(&pool->lock); |
| 205 | return true; |
| 206 | } |
| 207 | |
| 208 | bool pop_cmd(CmdPool *pool, cmd_data_t *out_cmd) { |
| 209 | netdata_mutex_lock(&pool->lock); |
| 210 | if (pool->count == 0) { |
| 211 | netdata_mutex_unlock(&pool->lock); // No commands to pop |
| 212 | return false; |
| 213 | } |
| 214 | *out_cmd = pool->buffer[pool->head]; |
| 215 | pool->head = (pool->head + 1) % pool->size; |
| 216 | pool->count--; |
| 217 | |
| 218 | netdata_cond_signal(&pool->not_full); |
| 219 | netdata_mutex_unlock(&pool->lock); |
| 220 | return true; |
| 221 | } |
| 222 | |
| 223 | void release_cmd_pool(CmdPool *pool) { |
| 224 | if (pool->buffer) { |
| 225 | free(pool->buffer); |
| 226 | pool->buffer = NULL; |
| 227 | } |
| 228 | netdata_mutex_destroy(&pool->lock); |
| 229 | netdata_cond_destroy(&pool->not_full); |
| 230 | } |
| 231 | |
| 232 | /// Test |
| 233 | |
| 234 | typedef struct { |
| 235 | CmdPool *pool; |
| 236 | int total; |
| 237 | int failed; |
| 238 | } ThreadArgs; |
| 239 | |
| 240 | void push_thread(void *arg) { |
| 241 | ThreadArgs *args = (ThreadArgs *)arg; |
| 242 | CmdPool *pool = args->pool; |
| 243 | for (int i = 0; i < args->total; ++i) { |
| 244 | cmd_data_t cmd = { 0 }; |
| 245 | snprintf(cmd.data, sizeof(cmd.data), "cmd-%d", i); |
| 246 | push_cmd(pool, &cmd, true); |
| 247 | } |
| 248 | fprintf(stderr, "PUSHED: %d commands\n", args->total); |
| 249 | } |
| 250 | |
| 251 | void pop_thread(void *arg) { |
| 252 | ThreadArgs *args = (ThreadArgs *)arg; |
| 253 | CmdPool *pool = args->pool; |
| 254 | |
| 255 | cmd_data_t cmd; |
| 256 | for (int i = 0; i < args->total; ) { |
| 257 | bool got = pop_cmd(pool, &cmd); |
| 258 | if (got) { |
| 259 | char expected[64]; |
| 260 | snprintf(expected, sizeof(expected), "cmd-%d", i); |
| 261 | if (strcmp(cmd.data, expected) != 0) { |
| 262 | fprintf(stderr, "POPPED: %s --- EXPECTED %s FAILED\n", cmd.data, expected); |
| 263 | args->failed++; |
| 264 | } |
| 265 | i++; |
| 266 | } else { |
| 267 | uv_sleep(1); // avoid busy spin |
| 268 | } |
| 269 | } |
| 270 | fprintf(stderr, "POPPED: %d commands\n", args->total); |
| 271 | } |
| 272 | |
| 273 | int test_cmd_pool_fifo() |
| 274 | { |
| 275 | CmdPool pool; |
| 276 | |
| 277 | int pool_sizes[] = {32, 64, 128, 256}; |
| 278 | |
| 279 | for (size_t i = 0; i < sizeof(pool_sizes) / sizeof(pool_sizes[0]); ++i) { |
| 280 | int pool_size = pool_sizes[i]; |
| 281 | init_cmd_pool(&pool, pool_size); |
| 282 | |
| 283 | ThreadArgs args = {.pool = &pool, .total = 1000, .failed = 0}; |
| 284 | uv_thread_t producer, consumer; |
| 285 | fprintf(stderr, "Testing pool size %d\n", pool_size); |
| 286 | |
| 287 | uv_thread_create(&producer, push_thread, &args); |
| 288 | uv_thread_create(&consumer, pop_thread, &args); |
| 289 | |
| 290 | uv_thread_join(&producer); |
| 291 | uv_thread_join(&consumer); |
| 292 | |
| 293 | release_cmd_pool(&pool); |
| 294 | if (args.failed) { |
| 295 | fprintf(stderr, "Multithreaded FIFO test failed with %d errors.\n", args.failed); |
| 296 | return 1; |
| 297 | } |
| 298 | } |
| 299 | fprintf(stderr, "Multithreaded FIFO test passed.\n"); |
| 300 | return 0; |
| 301 | } |