rrdengine.c · cryptotaxi247/netdata

Code Commits Issues Pulls Sessions CI/CD

master

c 2,744 lines 104 KB

Raw

1	// SPDX-License-Identifier: GPL-3.0-or-later
2
3	#include "rrdengine.h"
4	#include "pdc.h"
5	#include "dbengine-compression.h"
6
7	struct rrdeng_global_stats global_stats = { 0 };
8
9	unsigned rrdeng_pages_per_extent = DEFAULT_PAGES_PER_EXTENT;
10
11	#if WORKER_UTILIZATION_MAX_JOB_TYPES < (RRDENG_OPCODE_MAX + 2)
12	#error Please increase WORKER_UTILIZATION_MAX_JOB_TYPES to at least (RRDENG_MAX_OPCODE + 2)
13	#endif
14
15	struct rrdeng_cmd {
16	struct rrdengine_instance *ctx;
17	enum rrdeng_opcode opcode;
18	void *data;
19	struct completion *completion;
20	enum storage_priority priority;
21	dequeue_callback_t dequeue_cb;
22
23	struct {
24	struct rrdeng_cmd *prev;
25	struct rrdeng_cmd *next;
26	} queue;
27	};
28
29	static inline struct rrdeng_cmd rrdeng_deq_cmd(bool from_worker);
30	static inline void worker_dispatch_extent_read(struct rrdeng_cmd cmd, bool from_worker);
31	static inline void worker_dispatch_query_prep(struct rrdeng_cmd cmd, bool from_worker);
32
33	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;
43	bool shutdown;
44
45	size_t flushes_running;
46	size_t evict_main_running;
47	size_t evict_open_running;
48	size_t evict_extent_running;
49	size_t cleanup_running;
50
51	struct {
52	ARAL *ar;
53
54	struct {
55	SPINLOCK spinlock;
56
57	size_t waiting;
58	struct rrdeng_cmd *waiting_items_by_priority[STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE];
59	size_t executed_by_priority[STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE];
60	} unsafe;
61	} cmd_queue;
62
63	struct {
64	ARAL *ar;
65
66	struct {
67	size_t dispatched;
68	size_t executing;
69	} atomics;
70	} work_cmd;
71
72	struct {
73	ARAL *ar;
74	} handles;
75
76	struct {
77	ARAL *ar;
78	} descriptors;
79
80	struct {
81	ARAL *ar;
82	} xt_io_descr;
83
84	} rrdeng_main = {
85	.thread = 0,
86	.loop = {},
87	.async = {},
88	.timer = {},
89	.retention_timer = {},
90	.flushes_running = 0,
91	.evict_main_running = 0,
92	.cleanup_running = 0,
93
94	.cmd_queue = {
95	.unsafe = {
96	.spinlock = SPINLOCK_INITIALIZER,
97	},
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));
140
141	/* Magic numbers must fit in the super-blocks */
142	BUILD_BUG_ON(sizeof(RRDENG_DF_MAGIC) - 1 > RRDENG_MAGIC_SZ);
143	BUILD_BUG_ON(sizeof(RRDENG_JF_MAGIC) - 1 > RRDENG_MAGIC_SZ);
144
145	/* Version strings must fit in the super-blocks */
146	BUILD_BUG_ON(sizeof(RRDENG_DF_VER) - 1 > RRDENG_VER_SZ);
147	BUILD_BUG_ON(sizeof(RRDENG_JF_VER) - 1 > RRDENG_VER_SZ);
148
149	/* Data file super-block cannot be larger than RRDENG_BLOCK_SIZE */
150	BUILD_BUG_ON(RRDENG_DF_SB_PADDING_SZ < 0);
151
152	BUILD_BUG_ON(sizeof(nd_uuid_t) != UUID_SZ); /* check UUID size */
153
154	/* page count must fit in 8 bits */
155	BUILD_BUG_ON(MAX_PAGES_PER_EXTENT > 255);
156	}
157
158	// ----------------------------------------------------------------------------
159	// work request cache
160
161	typedef void (work_cb)(struct rrdengine_instance ctx, void data, struct completion completion, uv_work_t req);
162	typedef void (after_work_cb)(struct rrdengine_instance ctx, void data, struct completion completion, uv_work_t* req, int status);
163
164	struct rrdeng_work {
165	uv_work_t req;
166
167	struct rrdengine_instance *ctx;
168	void *data;
169	struct completion *completion;
170
171	work_cb work_cb;
172	after_work_cb after_work_cb;
173	enum rrdeng_opcode opcode;
174	};
175
176	static void work_request_init(void) {
177	rrdeng_main.work_cmd.ar = aral_create(
178	"dbengine-work-cmd",
179	sizeof(struct rrdeng_work),
180	0,
181	0,
182	NULL,
183	NULL, NULL, false, false, true
184	);
185
186	pulse_aral_register(rrdeng_main.work_cmd.ar, "workers");
187	}
188
189	enum LIBUV_WORKERS_STATUS {
190	LIBUV_WORKERS_RELAXED,
191	LIBUV_WORKERS_STRESSED,
192	LIBUV_WORKERS_CRITICAL,
193	};
194
195	static inline enum LIBUV_WORKERS_STATUS work_request_full(void) {
196	size_t dispatched = __atomic_load_n(&rrdeng_main.work_cmd.atomics.dispatched, __ATOMIC_RELAXED);
197
198	if(dispatched >= (size_t)(libuv_worker_threads))
199	return LIBUV_WORKERS_CRITICAL;
200
201	else if(dispatched >= (size_t)(libuv_worker_threads - RESERVED_LIBUV_WORKER_THREADS))
202	return LIBUV_WORKERS_STRESSED;
203
204	return LIBUV_WORKERS_RELAXED;
205	}
206
207	// This needs to be called from event loop thread only (callback)
208	static inline void check_and_schedule_db_rotation(struct rrdengine_instance *ctx)
209	{
210	internal_fatal(rrdeng_main.tid != gettid_cached(), "check_and_schedule_db_rotation() can only be run from the event loop thread");
211
212	if (__atomic_load_n(&ctx->atomic.needs_indexing, __ATOMIC_RELAXED)) {
213	if (ctx->datafiles.pending_index == false) {
214	ctx->datafiles.pending_index = true;
215	rrdeng_enq_cmd(ctx, RRDENG_OPCODE_JOURNAL_INDEX, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
216	}
217	}
218
219	if (ctx->datafiles.pending_rotate) {
220	nd_log_daemon(NDLP_DEBUG, "DBENGINE: tier %d is already pending rotation", ctx->config.tier);
221	return;
222	}
223
224	if(rrdeng_ctx_tier_cap_exceeded(ctx)) {
225	ctx->datafiles.pending_rotate = true;
226	rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
227	}
228	}
229
230	static inline void work_done(struct rrdeng_work *work_request) {
231	aral_freez(rrdeng_main.work_cmd.ar, work_request);
232	}
233
234	static void work_standard_worker(uv_work_t *req) {
235	__atomic_add_fetch(&rrdeng_main.work_cmd.atomics.executing, 1, __ATOMIC_RELAXED);
236
237	register_libuv_worker_jobs();
238	worker_is_busy(UV_EVENT_WORKER_INIT);
239
240	struct rrdeng_work *work_request = req->data;
241
242	work_request->data = work_request->work_cb(work_request->ctx, work_request->data, work_request->completion, req);
243	worker_is_idle();
244
245	if(work_request->opcode == RRDENG_OPCODE_EXTENT_READ \|\| work_request->opcode == RRDENG_OPCODE_QUERY) {
246	internal_fatal(work_request->after_work_cb != NULL, "DBENGINE: opcodes with a callback should not boosted");
247
248	while(1) {
249	struct rrdeng_cmd cmd = rrdeng_deq_cmd(true);
250	if (cmd.opcode == RRDENG_OPCODE_NOOP)
251	break;
252
253	worker_is_busy(UV_EVENT_WORKER_INIT);
254	switch (cmd.opcode) {
255	case RRDENG_OPCODE_EXTENT_READ:
256	worker_dispatch_extent_read(cmd, true);
257	break;
258
259	case RRDENG_OPCODE_QUERY:
260	worker_dispatch_query_prep(cmd, true);
261	break;
262
263	default:
264	fatal("DBENGINE: Opcode should not be executed synchronously");
265	break;
266	}
267	worker_is_idle();
268	}
269	}
270
271	__atomic_sub_fetch(&rrdeng_main.work_cmd.atomics.dispatched, 1, __ATOMIC_RELAXED);
272	__atomic_sub_fetch(&rrdeng_main.work_cmd.atomics.executing, 1, __ATOMIC_RELAXED);
273	}
274
275	static void after_work_standard_callback(uv_work_t* req, int status) {
276	struct rrdeng_work *work_request = req->data;
277
278	worker_is_busy(RRDENG_OPCODE_MAX + work_request->opcode);
279
280	if(work_request->after_work_cb)
281	work_request->after_work_cb(work_request->ctx, work_request->data, work_request->completion, req, status);
282
283	work_done(work_request);
284
285	worker_is_idle();
286	}
287
288	static bool work_dispatch(struct rrdengine_instance ctx, void data, struct completion *completion, enum rrdeng_opcode opcode, work_cb do_work_cb, after_work_cb do_after_work_cb) {
289	struct rrdeng_work *work_request = NULL;
290
291	internal_fatal(rrdeng_main.tid != gettid_cached(), "work_dispatch() can only be run from the event loop thread");
292
293	work_request = aral_mallocz(rrdeng_main.work_cmd.ar);
294	memset(work_request, 0, sizeof(struct rrdeng_work));
295	work_request->req.data = work_request;
296	work_request->ctx = ctx;
297	work_request->data = data;
298	work_request->completion = completion;
299	work_request->work_cb = do_work_cb;
300	work_request->after_work_cb = do_after_work_cb;
301	work_request->opcode = opcode;
302
303	if(uv_queue_work(&rrdeng_main.loop, &work_request->req, work_standard_worker, after_work_standard_callback)) {
304	internal_fatal(true, "DBENGINE: cannot queue work");
305	work_done(work_request);
306	return false;
307	}
308
309	__atomic_add_fetch(&rrdeng_main.work_cmd.atomics.dispatched, 1, __ATOMIC_RELAXED);
310
311	return true;
312	}
313
314	// ----------------------------------------------------------------------------
315	// page descriptor cache
316
317	void page_descriptors_init(void) {
318	rrdeng_main.descriptors.ar = aral_create(
319	"dbengine-descriptors",
320	sizeof(struct page_descr_with_data),
321	0,
322	0,
323	NULL,
324	NULL, NULL, false, false, true);
325
326	pulse_aral_register(rrdeng_main.descriptors.ar, "descriptors");
327	}
328
329	struct page_descr_with_data *page_descriptor_get(void) {
330	struct page_descr_with_data *descr = aral_mallocz(rrdeng_main.descriptors.ar);
331	memset(descr, 0, sizeof(struct page_descr_with_data));
332	return descr;
333	}
334
335	static inline void page_descriptor_release(struct page_descr_with_data *descr) {
336	uuidmap_free(descr->uuid_id);
337	aral_freez(rrdeng_main.descriptors.ar, descr);
338	}
339
340	// ----------------------------------------------------------------------------
341	// extent io descriptor cache
342
343	static void extent_io_descriptor_init(void) {
344	rrdeng_main.xt_io_descr.ar = aral_create(
345	"dbengine-extent-io",
346	sizeof(struct extent_io_descriptor),
347	0,
348	0,
349	NULL,
350	NULL, NULL, false, false, true
351	);
352
353	pulse_aral_register(rrdeng_main.xt_io_descr.ar, "extent io");
354	}
355
356	static struct extent_io_descriptor *extent_io_descriptor_get(void) {
357	struct extent_io_descriptor *xt_io_descr = aral_mallocz(rrdeng_main.xt_io_descr.ar);
358	memset(xt_io_descr, 0, sizeof(struct extent_io_descriptor));
359	return xt_io_descr;
360	}
361
362	static inline void extent_io_descriptor_release(struct extent_io_descriptor *xt_io_descr) {
363	aral_freez(rrdeng_main.xt_io_descr.ar, xt_io_descr);
364	}
365
366	// ----------------------------------------------------------------------------
367	// query handle cache
368
369	void rrdeng_query_handle_init(void) {
370	rrdeng_main.handles.ar = aral_create(
371	"dbengine-query-handles",
372	sizeof(struct rrdeng_query_handle),
373	0,
374	0,
375	NULL,
376	NULL, NULL, false, false, true);
377
378	pulse_aral_register(rrdeng_main.handles.ar, "query handles");
379	}
380
381	ALWAYS_INLINE struct rrdeng_query_handle *rrdeng_query_handle_get(void) {
382	struct rrdeng_query_handle *handle = aral_mallocz(rrdeng_main.handles.ar);
383	memset(handle, 0, sizeof(struct rrdeng_query_handle));
384	return handle;
385	}
386
387	ALWAYS_INLINE void rrdeng_query_handle_release(struct rrdeng_query_handle *handle) {
388	aral_freez(rrdeng_main.handles.ar, handle);
389	}
390
391	// ----------------------------------------------------------------------------
392	// WAL cache
393
394	static struct {
395	struct {
396	SPINLOCK spinlock;
397	WAL *available_items;
398	size_t available;
399	} protected;
400
401	struct {
402	size_t allocated;
403	} atomics;
404	} wal_globals = {
405	.protected = {
406	.spinlock = SPINLOCK_INITIALIZER,
407	.available_items = NULL,
408	.available = 0,
409	},
410	.atomics = {
411	.allocated = 0,
412	},
413	};
414
415	static void wal_cleanup1(void) {
416	WAL *wal = NULL;
417
418	if(!spinlock_trylock(&wal_globals.protected.spinlock))
419	return;
420
421	if(wal_globals.protected.available_items && wal_globals.protected.available > nd_profile.storage_tiers) {
422	wal = wal_globals.protected.available_items;
423	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wal_globals.protected.available_items, wal, cache.prev, cache.next);
424	wal_globals.protected.available--;
425	}
426
427	spinlock_unlock(&wal_globals.protected.spinlock);
428
429	if(wal) {
430	posix_memalign_freez(wal->buf);
431	freez(wal);
432	__atomic_sub_fetch(&wal_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
433	}
434	}
435
436	WAL wal_get(struct rrdengine_instance ctx, unsigned size) {
437	if(!size \|\| size > RRDENG_BLOCK_SIZE)
438	fatal("DBENGINE: invalid WAL size requested");
439
440	WAL *wal = NULL;
441
442	spinlock_lock(&wal_globals.protected.spinlock);
443
444	if(likely(wal_globals.protected.available_items)) {
445	wal = wal_globals.protected.available_items;
446	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wal_globals.protected.available_items, wal, cache.prev, cache.next);
447	wal_globals.protected.available--;
448	}
449
450	uint64_t transaction_id = __atomic_fetch_add(&ctx->atomic.transaction_id, 1, __ATOMIC_RELAXED);
451	spinlock_unlock(&wal_globals.protected.spinlock);
452
453	if(unlikely(!wal)) {
454	wal = mallocz(sizeof(WAL));
455	wal->buf_size = RRDENG_BLOCK_SIZE;
456	(void)posix_memalignz((void *)&wal->buf, RRDFILE_ALIGNMENT, wal->buf_size);
457	__atomic_add_fetch(&wal_globals.atomics.allocated, 1, __ATOMIC_RELAXED);
458	}
459
460	// these need to survive
461	unsigned buf_size = wal->buf_size;
462	void *buf = wal->buf;
463
464	memset(wal, 0, sizeof(WAL));
465
466	// put them back
467	wal->buf_size = buf_size;
468	wal->buf = buf;
469
470	memset(wal->buf, 0, wal->buf_size);
471
472	wal->transaction_id = transaction_id;
473	wal->size = size;
474
475	return wal;
476	}
477
478	void wal_release(WAL *wal) {
479	if(unlikely(!wal)) return;
480
481	spinlock_lock(&wal_globals.protected.spinlock);
482	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(wal_globals.protected.available_items, wal, cache.prev, cache.next);
483	wal_globals.protected.available++;
484	spinlock_unlock(&wal_globals.protected.spinlock);
485	}
486
487	// ----------------------------------------------------------------------------
488	// command queue cache
489
490	static void rrdeng_cmd_queue_init(void) {
491	rrdeng_main.cmd_queue.ar = aral_create("dbengine-opcodes",
492	sizeof(struct rrdeng_cmd),
493	0,
494	0,
495	NULL,
496	NULL, NULL, false, false, true);
497
498	pulse_aral_register(rrdeng_main.cmd_queue.ar, "opcodes");
499	}
500
501	static inline STORAGE_PRIORITY rrdeng_enq_cmd_map_opcode_to_priority(enum rrdeng_opcode opcode, STORAGE_PRIORITY priority) {
502	if(unlikely(priority >= STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE))
503	priority = STORAGE_PRIORITY_BEST_EFFORT;
504
505	switch(opcode) {
506	case RRDENG_OPCODE_QUERY:
507	priority = STORAGE_PRIORITY_INTERNAL_QUERY_PREP;
508	break;
509
510	default:
511	break;
512	}
513
514	return priority;
515	}
516
517	ALWAYS_INLINE void rrdeng_enqueue_epdl_cmd(struct rrdeng_cmd *cmd) {
518	epdl_cmd_queued(cmd->data, cmd);
519	}
520
521	ALWAYS_INLINE void rrdeng_dequeue_epdl_cmd(struct rrdeng_cmd *cmd) {
522	epdl_cmd_dequeued(cmd->data);
523	}
524
525	ALWAYS_INLINE void rrdeng_req_cmd(requeue_callback_t get_cmd_cb, void *data, STORAGE_PRIORITY priority) {
526	spinlock_lock(&rrdeng_main.cmd_queue.unsafe.spinlock);
527
528	struct rrdeng_cmd *cmd = get_cmd_cb(data);
529	if(cmd) {
530	priority = rrdeng_enq_cmd_map_opcode_to_priority(cmd->opcode, priority);
531
532	if (cmd->priority > priority) {
533	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[cmd->priority], cmd, queue.prev, queue.next);
534	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority], cmd, queue.prev, queue.next);
535	cmd->priority = priority;
536	}
537	}
538
539	spinlock_unlock(&rrdeng_main.cmd_queue.unsafe.spinlock);
540	}
541
542	ALWAYS_INLINE void rrdeng_enq_cmd(struct rrdengine_instance ctx, enum rrdeng_opcode opcode, void data, struct completion *completion,
543	enum storage_priority priority, enqueue_callback_t enqueue_cb, dequeue_callback_t dequeue_cb) {
544
545	priority = rrdeng_enq_cmd_map_opcode_to_priority(opcode, priority);
546
547	struct rrdeng_cmd *cmd = aral_mallocz(rrdeng_main.cmd_queue.ar);
548	memset(cmd, 0, sizeof(struct rrdeng_cmd));
549	cmd->ctx = ctx;
550	cmd->opcode = opcode;
551	cmd->data = data;
552	cmd->completion = completion;
553	cmd->priority = priority;
554	cmd->dequeue_cb = dequeue_cb;
555
556	spinlock_lock(&rrdeng_main.cmd_queue.unsafe.spinlock);
557	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority], cmd, queue.prev, queue.next);
558	rrdeng_main.cmd_queue.unsafe.waiting++;
559	if(enqueue_cb)
560	enqueue_cb(cmd);
561	spinlock_unlock(&rrdeng_main.cmd_queue.unsafe.spinlock);
562
563	rrdeng_async_wakeup();
564	}
565
566	static inline bool rrdeng_cmd_has_waiting_opcodes_in_lower_priorities(STORAGE_PRIORITY priority, STORAGE_PRIORITY max_priority) {
567	for(; priority <= max_priority ; priority++)
568	if(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority])
569	return true;
570
571	return false;
572	}
573
574	#define opcode_empty (struct rrdeng_cmd) { \
575	.ctx = NULL, \
576	.opcode = RRDENG_OPCODE_NOOP, \
577	.priority = STORAGE_PRIORITY_BEST_EFFORT, \
578	.completion = NULL, \
579	.data = NULL, \
580	}
581
582	static inline struct rrdeng_cmd rrdeng_deq_cmd(bool from_worker) {
583	struct rrdeng_cmd *cmd = NULL;
584	enum LIBUV_WORKERS_STATUS status = work_request_full();
585	STORAGE_PRIORITY min_priority, max_priority;
586
587	if(unlikely(from_worker)) {
588	if(status == LIBUV_WORKERS_CRITICAL)
589	return opcode_empty;
590
591	min_priority = STORAGE_PRIORITY_INTERNAL_QUERY_PREP;
592	max_priority = STORAGE_PRIORITY_BEST_EFFORT;
593	}
594	else {
595	min_priority = STORAGE_PRIORITY_INTERNAL_DBENGINE;
596	max_priority = (status != LIBUV_WORKERS_RELAXED) ? STORAGE_PRIORITY_INTERNAL_DBENGINE : STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE - 1;
597	}
598
599	// find an opcode to execute from the queue
600	spinlock_lock(&rrdeng_main.cmd_queue.unsafe.spinlock);
601	for(STORAGE_PRIORITY priority = min_priority; priority <= max_priority ; priority++) {
602	cmd = rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority];
603	if(cmd) {
604
605	// avoid starvation of lower priorities
606	if(unlikely(priority >= STORAGE_PRIORITY_HIGH &&
607	priority < STORAGE_PRIORITY_BEST_EFFORT &&
608	++rrdeng_main.cmd_queue.unsafe.executed_by_priority[priority] % 50 == 0 &&
609	rrdeng_cmd_has_waiting_opcodes_in_lower_priorities(priority + 1, max_priority))) {
610	// let the others run 2% of the requests
611	cmd = NULL;
612	continue;
613	}
614
615	// remove it from the queue
616	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(rrdeng_main.cmd_queue.unsafe.waiting_items_by_priority[priority], cmd, queue.prev, queue.next);
617	rrdeng_main.cmd_queue.unsafe.waiting--;
618	break;
619	}
620	}
621
622	if(cmd && cmd->dequeue_cb) {
623	cmd->dequeue_cb(cmd);
624	cmd->dequeue_cb = NULL;
625	}
626
627	spinlock_unlock(&rrdeng_main.cmd_queue.unsafe.spinlock);
628
629	struct rrdeng_cmd ret;
630	if(cmd) {
631	// copy it, to return it
632	ret = *cmd;
633
634	aral_freez(rrdeng_main.cmd_queue.ar, cmd);
635	}
636	else
637	ret = opcode_empty;
638
639	return ret;
640	}
641
642
643	// ----------------------------------------------------------------------------
644
645	static void journalfile_extent_build(struct rrdengine_instance ctx, struct extent_io_descriptor xt_io_descr) {
646	unsigned count, payload_length, descr_size, size_bytes;
647	void *buf;
648	/* persistent structures */
649	struct rrdeng_df_extent_header *df_header;
650	struct rrdeng_jf_transaction_header *jf_header;
651	struct rrdeng_jf_store_data *jf_metric_data;
652	struct rrdeng_jf_transaction_trailer *jf_trailer;
653	uLong crc;
654
655	df_header = xt_io_descr->buf;
656	count = df_header->number_of_pages;
657	fatal_assert(count <= MAX_PAGES_PER_EXTENT);
658	descr_size = sizeof(jf_metric_data->descr) count;
659	payload_length = sizeof(*jf_metric_data) + descr_size;
660	size_bytes = sizeof(jf_header) + payload_length + sizeof(jf_trailer);
661
662	xt_io_descr->wal = wal_get(ctx, size_bytes);
663	buf = xt_io_descr->wal->buf;
664
665	jf_header = buf;
666	jf_header->type = STORE_DATA;
667	jf_header->reserved = 0;
668	jf_header->id = xt_io_descr->wal->transaction_id;
669	jf_header->payload_length = payload_length;
670
671	jf_metric_data = buf + sizeof(*jf_header);
672	jf_metric_data->extent_offset = xt_io_descr->pos;
673	jf_metric_data->extent_size = xt_io_descr->bytes;
674	jf_metric_data->number_of_pages = count;
675	memcpy(jf_metric_data->descr, df_header->descr, descr_size);
676
677	jf_trailer = buf + sizeof(*jf_header) + payload_length;
678	crc = crc32(0L, Z_NULL, 0);
679	crc = crc32(crc, buf, sizeof(*jf_header) + payload_length);
680	crc32set(jf_trailer->checksum, crc);
681	}
682
683	static void
684	extent_flush_to_open(struct rrdengine_instance ctx, struct extent_io_descriptor xt_io_descr, bool have_error)
685	{
686	worker_is_busy(UV_EVENT_DBENGINE_FLUSHED_TO_OPEN);
687
688	struct page_descr_with_data *descr;
689	struct rrdengine_datafile *datafile;
690	unsigned i;
691
692	datafile = xt_io_descr->datafile;
693
694	bool still_running = ctx_is_available_for_queries(ctx);
695	if (likely(still_running && !have_error))
696	internal_fatal(!rrdeng_valid_extent_disk_size(xt_io_descr->bytes),
697	"DBENGINE: flushed extent has invalid size %u",
698	xt_io_descr->bytes);
699
700	usec_t max_end_time_ut = 0;
701	for (i = 0 ; i < xt_io_descr->descr_count ; ++i) {
702	descr = xt_io_descr->descr_array[i];
703
704	if (descr->end_time_ut > max_end_time_ut)
705	max_end_time_ut = descr->end_time_ut;
706
707	if (likely(still_running && !have_error)) {
708	pgc_open_add_hot_page(
709	(Word_t)ctx,
710	descr->metric_id,
711	(time_t)(descr->start_time_ut / USEC_PER_SEC),
712	(time_t)(descr->end_time_ut / USEC_PER_SEC),
713	descr->update_every_s,
714	datafile,
715	xt_io_descr->pos,
716	xt_io_descr->bytes);
717	}
718
719	page_descriptor_release(descr);
720	}
721
722	if (!have_error) {
723	if (max_end_time_ut > 0) {
724	time_t new_last_time_s = (time_t)(max_end_time_ut / USEC_PER_SEC);
725
726	// Atomically update to keep the maximum
727	spinlock_lock(&datafile->journalfile->data_spinlock);
728	if (new_last_time_s > datafile->journalfile->v2.last_time_s)
729	datafile->journalfile->v2.last_time_s = new_last_time_s;
730	spinlock_unlock(&datafile->journalfile->data_spinlock);
731	}
732	}
733
734	posix_memalign_freez(xt_io_descr->buf);
735	extent_io_descriptor_release(xt_io_descr);
736
737	spinlock_lock(&datafile->writers.spinlock);
738	datafile->writers.flushed_to_open_running--;
739	spinlock_unlock(&datafile->writers.spinlock);
740
741	if(datafile->fileno != ctx_last_fileno_get(ctx) && still_running)
742	__atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);
743
744	worker_is_idle();
745	}
746
747	// Main event loop callback
748
749	static bool datafile_is_full(struct rrdengine_instance ctx, struct rrdengine_datafile datafile, uint64_t extent_size) {
750	bool ret = false;
751
752	spinlock_lock(&datafile->writers.spinlock);
753
754	#ifdef OS_WINDOWS
755	time_t now = now_realtime_sec();
756	if (now - datafile->writers.last_sync_time > 60) {
757	sync_uv_file_data(datafile->file);
758	sync_uv_file_data(datafile->journalfile->file);
759	datafile->writers.last_sync_time = now_realtime_sec();
760	}
761	#endif
762
763	// Check if adding this extent would exceed the target size
764	if(datafile->pos + extent_size > rrdeng_target_data_file_size(ctx))
765	ret = true;
766
767	spinlock_unlock(&datafile->writers.spinlock);
768
769	return ret;
770	}
771
772	size_t datafile_count(struct rrdengine_instance *ctx, bool with_lock)
773	{
774	size_t count = 0;
775
776	if (!with_lock)
777	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
778
779	count = JudyLCount(ctx->datafiles.JudyL, 0, -1, PJE0);
780
781	if (!with_lock)
782	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
783
784	return count;
785	}
786
787	struct rrdengine_datafile *
788	get_next_datafile(struct rrdengine_datafile this_datafile, struct rrdengine_instance ctx, bool with_lock)
789	{
790	struct rrdengine_datafile *datafile = NULL;
791
792	ctx = this_datafile ? this_datafile->ctx : ctx;
793	if (!ctx)
794	return NULL;
795
796	if (!with_lock)
797	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
798
799	Word_t Index = this_datafile ? this_datafile->fileno : 0;
800	Pvoid_t *Pvalue;
801
802	Pvalue = JudyLNext(ctx->datafiles.JudyL, &Index, PJE0);
803
804	if (Pvalue)
805	datafile = *Pvalue;
806
807	if (!with_lock)
808	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
809
810	return datafile;
811	}
812
813	static struct rrdengine_datafile get_ctx_datafile_first_or_last(struct rrdengine_instance ctx, bool first, bool with_lock)
814	{
815	struct rrdengine_datafile *datafile = NULL;
816
817	if (!with_lock)
818	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
819
820	Word_t Index = 0;
821	Pvoid_t *Pvalue;
822
823	if (first)
824	Pvalue = JudyLFirst(ctx->datafiles.JudyL, &Index, PJE0);
825	else {
826	Index = -1;
827	Pvalue = JudyLLast(ctx->datafiles.JudyL, &Index, PJE0);
828	}
829
830	if (Pvalue)
831	datafile = *Pvalue;
832
833	if (!with_lock)
834	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
835
836	return datafile;
837	}
838
839	struct rrdengine_datafile get_first_ctx_datafile(struct rrdengine_instance ctx, bool with_lock) {
840	return get_ctx_datafile_first_or_last(ctx, true, with_lock);
841	}
842
843	struct rrdengine_datafile get_last_ctx_datafile(struct rrdengine_instance ctx, bool with_lock) {
844	return get_ctx_datafile_first_or_last(ctx, false, with_lock);
845	}
846
847	static netdata_mutex_t mutex;
848
849	static void __attribute__((constructor)) init_mutex(void) {
850	netdata_mutex_init(&mutex);
851	}
852
853	static void __attribute__((destructor)) destroy_mutex(void) {
854	netdata_mutex_destroy(&mutex);
855	}
856
857	static struct rrdengine_datafile get_datafile_to_write_extent(struct rrdengine_instance ctx, uint64_t extent_size) {
858	struct rrdengine_datafile *datafile;
859
860	// Acquire the mutex at the beginning to make the entire check-and-act atomic
861	// This prevents the race condition where multiple threads pass the "is full" check
862	// before any of them increments the position, causing files to grow beyond limits
863	netdata_mutex_lock(&mutex);
864
865	// get the latest datafile
866	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
867
868	datafile = get_last_ctx_datafile(ctx, true);
869	// become a writer on this datafile, to prevent it from vanishing
870	spinlock_lock(&datafile->writers.spinlock);
871	datafile->writers.running++;
872	spinlock_unlock(&datafile->writers.spinlock);
873	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
874
875	if(datafile_is_full(ctx, datafile, extent_size)) {
876	// remember the datafile we have become writers to
877	struct rrdengine_datafile *old_datafile = datafile;
878
879	// Create a new datafile - since we hold the mutex, no other thread can interfere
880	if(create_new_datafile_pair(ctx) == 0)
881	__atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);
882
883	// get the new datafile
884	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
885	datafile = get_last_ctx_datafile(ctx, true);
886	// become a writer on this datafile, to prevent it from vanishing
887	spinlock_lock(&datafile->writers.spinlock);
888	datafile->writers.running++;
889	spinlock_unlock(&datafile->writers.spinlock);
890	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
891
892	// release the writers on the old datafile
893	spinlock_lock(&old_datafile->writers.spinlock);
894	old_datafile->writers.running--;
895	#ifdef OS_WINDOWS
896	sync_uv_file_data(old_datafile->file);
897	sync_uv_file_data(old_datafile->journalfile->file);
898	datafile->writers.last_sync_time = now_realtime_sec();
899	old_datafile->writers.last_sync_time = now_realtime_sec();
900	#endif
901	spinlock_unlock(&old_datafile->writers.spinlock);
902	}
903
904	netdata_mutex_unlock(&mutex);
905
906	return datafile;
907	}
908
909	/*
910	* Take a page list in a judy array and write them
911	*/
912	static struct extent_io_descriptor *
913	datafile_extent_build(struct rrdengine_instance ctx, struct page_descr_with_data base, uv_buf_t *iov)
914	{
915	unsigned i;
916	uint32_t real_io_size, size_bytes, count, pos;
917	uint32_t uncompressed_payload_length, max_compressed_size, payload_offset;
918	struct page_descr_with_data descr, eligible_pages[MAX_PAGES_PER_EXTENT];
919	struct extent_io_descriptor *xt_io_descr;
920	Word_t Index;
921	uint8_t compression_algorithm = ctx->config.global_compress_alg;
922	struct rrdengine_datafile *datafile;
923	/* persistent structures */
924	struct rrdeng_df_extent_header *header;
925	struct rrdeng_df_extent_trailer *trailer;
926	uLong crc;
927
928	for(descr = base, Index = 0, count = 0, uncompressed_payload_length = 0;
929	descr && count != rrdeng_pages_per_extent;
930	descr = descr->link.next, Index++) {
931
932	uncompressed_payload_length += descr->page_length;
933	eligible_pages[count++] = descr;
934
935	}
936
937	if (!count) {
938	__atomic_sub_fetch(&ctx->atomic.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
939	return NULL;
940	}
941
942	xt_io_descr = extent_io_descriptor_get();
943	payload_offset = sizeof(header) + count sizeof(header->descr[0]);
944	max_compressed_size = dbengine_max_compressed_size(uncompressed_payload_length, compression_algorithm);
945	size_bytes = payload_offset + MAX(uncompressed_payload_length, max_compressed_size) + sizeof(*trailer);
946	(void)posix_memalignz((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(size_bytes));
947	memset(xt_io_descr->buf, 0, ALIGN_BYTES_CEILING(size_bytes));
948	(void) memcpy(xt_io_descr->descr_array, eligible_pages, sizeof(struct page_descr_with_data ) count);
949	xt_io_descr->descr_count = count;
950
951	pos = 0;
952	header = xt_io_descr->buf;
953	header->number_of_pages = count;
954	pos += sizeof(*header);
955
956	for (i = 0 ; i < count ; ++i) {
957	descr = xt_io_descr->descr_array[i];
958	header->descr[i].type = descr->type;
959	uuid_copy((nd_uuid_t )header->descr[i].uuid, *uuidmap_uuid_ptr(descr->uuid_id));
960	header->descr[i].page_length = descr->page_length;
961	header->descr[i].start_time_ut = descr->start_time_ut;
962
963	switch (descr->type) {
964	case RRDENG_PAGE_TYPE_ARRAY_32BIT:
965	case RRDENG_PAGE_TYPE_ARRAY_TIER1:
966	header->descr[i].end_time_ut = descr->end_time_ut;
967	break;
968	case RRDENG_PAGE_TYPE_GORILLA_32BIT:
969	header->descr[i].gorilla.delta_time_s = (uint32_t) ((descr->end_time_ut - descr->start_time_ut) / USEC_PER_SEC);
970	header->descr[i].gorilla.entries = pgd_slots_used(descr->pgd);
971	break;
972	default:
973	fatal("Unknown page type: %uc", descr->type);
974	}
975
976	pos += sizeof(header->descr[i]);
977	}
978
979	// build the extent payload
980	for (i = 0 ; i < count ; ++i) {
981	descr = xt_io_descr->descr_array[i];
982	pgd_copy_to_extent(descr->pgd, xt_io_descr->buf + pos, descr->page_length);
983	pos += descr->page_length;
984	}
985
986	// compress the payload
987	size_t compressed_size =
988	(int)dbengine_compress(xt_io_descr->buf + payload_offset,
989	uncompressed_payload_length,
990	compression_algorithm);
991
992	internal_fatal(compressed_size > max_compressed_size, "DBENGINE: compression returned more data than the max allowed");
993	internal_fatal(compressed_size > uncompressed_payload_length, "DBENGINE: compression returned more data than the uncompressed extent");
994
995	if(compressed_size) {
996	header->compression_algorithm = compression_algorithm;
997	header->payload_length = compressed_size;
998	}
999	else {
1000	// compression failed, or generated bigger pages
1001	// so it didn't touch our uncompressed buffer
1002	header->compression_algorithm = RRDENG_COMPRESSION_NONE;
1003	header->payload_length = compressed_size = uncompressed_payload_length;
1004	}
1005
1006	// set the correct size
1007	size_bytes = payload_offset + compressed_size + sizeof(*trailer);
1008
1009	if(compression_algorithm != RRDENG_COMPRESSION_NONE) {
1010	__atomic_add_fetch(&ctx->stats.before_compress_bytes, uncompressed_payload_length, __ATOMIC_RELAXED);
1011	__atomic_add_fetch(&ctx->stats.after_compress_bytes, compressed_size, __ATOMIC_RELAXED);
1012	}
1013
1014	real_io_size = ALIGN_BYTES_CEILING(size_bytes);
1015
1016	// Pass the extent size so the check can determine if this extent will fit
1017	datafile = get_datafile_to_write_extent(ctx, real_io_size);
1018	spinlock_lock(&datafile->writers.spinlock);
1019	xt_io_descr->datafile = datafile;
1020	xt_io_descr->pos = datafile->pos;
1021	datafile->pos += real_io_size;
1022	spinlock_unlock(&datafile->writers.spinlock);
1023
1024	xt_io_descr->bytes = size_bytes;
1025
1026	trailer = xt_io_descr->buf + size_bytes - sizeof(*trailer);
1027	crc = crc32(0L, Z_NULL, 0);
1028	crc = crc32(crc, xt_io_descr->buf, size_bytes - sizeof(*trailer));
1029	crc32set(trailer->checksum, crc);
1030
1031	iov = uv_buf_init((void )xt_io_descr->buf, real_io_size);
1032	journalfile_extent_build(ctx, xt_io_descr);
1033
1034	ctx_last_flush_fileno_set(ctx, datafile->fileno);
1035	xt_io_descr->real_io_size = real_io_size;
1036
1037	return xt_io_descr;
1038	}
1039
1040
1041	static void after_weights_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t uv_work_req __maybe_unused, int status __maybe_unused)
1042	{
1043	;
1044	}
1045
1046	static void *weights_worker(
1047	struct rrdengine_instance *ctx __maybe_unused,
1048	void *data,
1049	struct completion *completion,
1050	uv_work_t *req __maybe_unused)
1051	{
1052	worker_is_busy(UV_EVENT_WEIGHTS_CALCULATION);
1053	query_weights_worker_thread(data);
1054	completion_mark_complete(completion);
1055	worker_is_idle();
1056	return NULL;
1057	}
1058
1059	static void after_extent_write(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t uv_work_req __maybe_unused, int status __maybe_unused)
1060	{
1061	check_and_schedule_db_rotation(ctx);
1062	}
1063
1064	static void *extent_write_tp_worker(
1065	struct rrdengine_instance *ctx,
1066	void *data,
1067	struct completion *completion __maybe_unused,
1068	uv_work_t *req __maybe_unused)
1069	{
1070	worker_is_busy(UV_EVENT_DBENGINE_EXTENT_WRITE);
1071	uv_buf_t iov;
1072	struct page_descr_with_data *base = data;
1073	struct extent_io_descriptor *xt_io_descr = datafile_extent_build(ctx, base, &iov);
1074
1075	if (!xt_io_descr)
1076	goto done;
1077
1078	struct rrdengine_datafile *datafile = xt_io_descr->datafile;
1079	uv_fs_t request;
1080
1081	int retries = 10;
1082	int ret = -1;
1083	while (ret < 0 && --retries) {
1084	ret = uv_fs_write(NULL, &request, datafile->file, &iov, 1, (int64_t)xt_io_descr->pos, NULL);
1085	uv_fs_req_cleanup(&request);
1086	if (ret < 0) {
1087	if (ret == -ENOSPC \|\| ret == -EBADF \|\| ret == -EACCES \|\| ret == -EROFS \|\| ret == -EINVAL)
1088	break;
1089	sleep_usec(300 * USEC_PER_MS);
1090	}
1091	}
1092
1093	if (unlikely(ret < 0))
1094	ctx_io_error(ctx);
1095	else {
1096	ctx_current_disk_space_increase(ctx, xt_io_descr->real_io_size);
1097	ctx_io_write_op_bytes(ctx, xt_io_descr->real_io_size);
1098	ret = journalfile_v1_extent_write(ctx, datafile, xt_io_descr->wal);
1099	}
1100
1101	if (ret < 0) {
1102	nd_log_limit_static_global_var(dbengine_erl, 10, 0);
1103	nd_log_limit(&dbengine_erl, NDLS_DAEMON, NDLP_ERR, "DBENGINE: Tier %d, %s", ctx->config.tier, uv_strerror(ret));
1104	}
1105
1106	spinlock_lock(&datafile->writers.spinlock);
1107	datafile->writers.running--;
1108	datafile->writers.flushed_to_open_running++;
1109	spinlock_unlock(&datafile->writers.spinlock);
1110
1111	extent_flush_to_open(ctx, xt_io_descr, ret < 0);
1112
1113	done:
1114	__atomic_sub_fetch(&ctx->atomic.extents_currently_being_flushed, 1, __ATOMIC_RELAXED);
1115	completion_mark_complete(completion);
1116	worker_is_idle();
1117	return NULL;
1118	}
1119
1120	static void after_database_rotate(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
1121	__atomic_store_n(&ctx->atomic.now_deleting_files, false, __ATOMIC_RELAXED);
1122
1123	check_and_schedule_db_rotation(ctx);
1124	}
1125
1126	struct uuid_first_time_s {
1127	nd_uuid_t *uuid;
1128	time_t first_time_s;
1129	METRIC *metric;
1130	size_t pages_found;
1131	size_t df_matched;
1132	size_t df_index_oldest;
1133	};
1134
1135	struct rrdengine_datafile datafile_release_and_acquire_next_for_retention(struct rrdengine_instance ctx, struct rrdengine_datafile *datafile) {
1136
1137	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
1138
1139	struct rrdengine_datafile *next_datafile = get_next_datafile(datafile, NULL, true);
1140
1141	while(next_datafile && !datafile_acquire(next_datafile, DATAFILE_ACQUIRE_RETENTION))
1142	next_datafile = get_next_datafile(next_datafile, NULL, true);
1143
1144	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
1145
1146	datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);
1147
1148	return next_datafile;
1149	}
1150
1151	static time_t find_uuid_first_time(
1152	struct rrdengine_instance *ctx,
1153	struct rrdengine_datafile *datafile,
1154	struct uuid_first_time_s *uuid_first_entry_list,
1155	size_t count)
1156	{
1157	time_t global_first_time_s = LONG_MAX;
1158
1159	// acquire the datafile to work with it
1160	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
1161	while(datafile && !datafile_acquire(datafile, DATAFILE_ACQUIRE_RETENTION))
1162	datafile = get_next_datafile(datafile, NULL, true);
1163
1164	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
1165
1166	if (unlikely(!datafile))
1167	return global_first_time_s;
1168
1169	unsigned journalfile_count = 0;
1170	size_t binary_match = 0;
1171	size_t not_matching_bsearches = 0;
1172
1173	bool agent_shutdown = false;
1174	while (datafile) {
1175	size_t journal_v2_file_size = 0;
1176	struct journal_v2_header *j2_header = journalfile_v2_data_acquire_with_hint(
1177	datafile->journalfile, &journal_v2_file_size, 0, 0, JOURNALFILE_V2_ACCESS_RANDOM);
1178	if (!j2_header) {
1179	datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1180	continue;
1181	}
1182
1183	bool any_matching = false;
1184	bool journal_access_failed = false;
1185
1186	char file_path[RRDENG_PATH_MAX];
1187	journalfile_v2_generate_path(datafile, file_path, sizeof(file_path));
1188	PROTECTED_ACCESS_SETUP(datafile->journalfile->mmap.data, datafile->journalfile->mmap.size, file_path, "read");
1189	if (no_signal_received) {
1190	time_t journal_start_time_s = (time_t)(j2_header->start_time_ut / USEC_PER_SEC);
1191
1192	if (journal_start_time_s < global_first_time_s)
1193	global_first_time_s = journal_start_time_s;
1194
1195	size_t metric_offset = j2_header->metric_offset;
1196	size_t journal_metric_count = j2_header->metric_count;
1197	size_t metric_list_size;
1198	if (__builtin_mul_overflow(journal_metric_count, sizeof(struct journal_metric_list), &metric_list_size)) {
1199	nd_log_daemon(NDLP_ERR,
1200	"DBENGINE: metric list size overflow in journalfile \"%s\" "
1201	"(metric_count=%zu, entry_size=%zu), skipping it",
1202	file_path, journal_metric_count, sizeof(struct journal_metric_list));
1203	journal_access_failed = true;
1204	goto release_journal;
1205	}
1206	if (metric_offset > journal_v2_file_size \|\|
1207	metric_list_size > journal_v2_file_size - metric_offset) {
1208	nd_log_daemon(NDLP_ERR,
1209	"DBENGINE: metric list exceeds journal file size in journalfile \"%s\" "
1210	"(metric_offset=%zu, list_size=%zu, file_size=%zu), skipping it",
1211	file_path, metric_offset, metric_list_size, journal_v2_file_size);
1212	journal_access_failed = true;
1213	goto release_journal;
1214	}
1215
1216	struct journal_metric_list *uuid_list =
1217	(struct journal_metric_list )((uint8_t )j2_header + metric_offset);
1218	struct uuid_first_time_s *uuid_original_entry;
1219
1220	size_t journal_search_start = 0; // Start of remaining search space
1221	any_matching = false;
1222	for (size_t index = 0; index < count; ++index) {
1223	uuid_original_entry = &uuid_first_entry_list[index];
1224
1225	if (uuid_original_entry->df_matched > 3 \|\| uuid_original_entry->pages_found > 5)
1226	continue;
1227
1228	any_matching = true;
1229	if (journal_search_start >= journal_metric_count)
1230	break;
1231
1232	struct journal_metric_list *live_entry = &uuid_list[journal_search_start];
1233	// Check if we avoid bsearch
1234	if (journal_metric_uuid_compare(uuid_original_entry->uuid, live_entry->uuid) != 0) {
1235	live_entry = bsearch(
1236	uuid_original_entry->uuid,
1237	uuid_list + journal_search_start,
1238	journal_metric_count - journal_search_start,
1239	sizeof(*uuid_list),
1240	journal_metric_uuid_compare);
1241
1242	if (!live_entry) {
1243	not_matching_bsearches++;
1244	continue;
1245	}
1246	}
1247
1248	size_t found_index = live_entry - uuid_list;
1249	journal_search_start = found_index + 1; // Next search starts after this match
1250
1251	if (journal_search_start >= journal_metric_count)
1252	break;
1253
1254	uuid_original_entry->pages_found += live_entry->entries;
1255	uuid_original_entry->df_matched++;
1256
1257	time_t old_first_time_s = uuid_original_entry->first_time_s;
1258	time_t first_time_s = live_entry->delta_start_s + journal_start_time_s;
1259	uuid_original_entry->first_time_s = MIN(uuid_original_entry->first_time_s, first_time_s);
1260
1261	if (uuid_original_entry->first_time_s != old_first_time_s)
1262	uuid_original_entry->df_index_oldest = uuid_original_entry->df_matched;
1263
1264	binary_match++;
1265
1266	if (unlikely(!ctx_is_available_for_queries(ctx))) {
1267	agent_shutdown = true;
1268	break;
1269	}
1270	}
1271	} else {
1272	journal_access_failed = true;
1273	}
1274
1275	release_journal:
1276	journalfile_v2_data_release(datafile->journalfile);
1277
1278	if (agent_shutdown) {
1279	datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);
1280	break;
1281	}
1282
1283	journalfile_count++;
1284	datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1285	if (journal_access_failed)
1286	continue;
1287
1288	if (!any_matching) {
1289	if (datafile)
1290	datafile_release(datafile, DATAFILE_ACQUIRE_RETENTION);
1291	break;
1292	}
1293	}
1294
1295	if (agent_shutdown)
1296	return global_first_time_s;
1297
1298	// Let's scan the open cache for almost exact match
1299	size_t open_cache_count = 0;
1300
1301	size_t df_index[10] = { 0 };
1302	size_t without_metric = 0;
1303	size_t open_cache_gave_first_time_s = 0;
1304	size_t metric_count = 0;
1305	size_t without_retention = 0;
1306	size_t not_needed_bsearches = 0;
1307
1308	for (size_t index = 0; index < count; ++index) {
1309	struct uuid_first_time_s *uuid_first_t_entry = &uuid_first_entry_list[index];
1310
1311	metric_count++;
1312
1313	size_t idx = uuid_first_t_entry->df_index_oldest;
1314	if(idx >= 10)
1315	idx = 9;
1316
1317	df_index[idx]++;
1318
1319	not_needed_bsearches += uuid_first_t_entry->df_matched - uuid_first_t_entry->df_index_oldest;
1320
1321	if (unlikely(!uuid_first_t_entry->metric)) {
1322	without_metric++;
1323	continue;
1324	}
1325
1326	PGC_PAGE *page = pgc_page_get_and_acquire(
1327	open_cache, (Word_t)ctx,
1328	(Word_t)uuid_first_t_entry->metric, 0,
1329	PGC_SEARCH_FIRST);
1330
1331	if (page) {
1332	time_t old_first_time_s = uuid_first_t_entry->first_time_s;
1333
1334	time_t first_time_s = pgc_page_start_time_s(page);
1335	uuid_first_t_entry->first_time_s = MIN(uuid_first_t_entry->first_time_s, first_time_s);
1336	pgc_page_release(open_cache, page);
1337	open_cache_count++;
1338
1339	if(uuid_first_t_entry->first_time_s != old_first_time_s) {
1340	open_cache_gave_first_time_s++;
1341	}
1342	}
1343	else {
1344	if(!uuid_first_t_entry->df_index_oldest)
1345	without_retention++;
1346	}
1347	}
1348	internal_error(true,
1349	"DBENGINE: analyzed the retention of %zu rotated metrics of tier %d, "
1350	"did %zu jv2 matching binary searches (%zu not matching, %zu overflown) in %u journal files, "
1351	"%zu metrics with entries in open cache, "
1352	"metrics first time found per datafile index ([not in jv2]:%zu, [1]:%zu, [2]:%zu, [3]:%zu, [4]:%zu, [5]:%zu, [6]:%zu, [7]:%zu, [8]:%zu, [bigger]: %zu), "
1353	"open cache found first time %zu, "
1354	"metrics without any remaining retention %zu, "
1355	"metrics not in MRG %zu",
1356	metric_count,
1357	ctx->config.tier,
1358	binary_match,
1359	not_matching_bsearches,
1360	not_needed_bsearches,
1361	journalfile_count,
1362	open_cache_count,
1363	df_index[0], df_index[1], df_index[2], df_index[3], df_index[4], df_index[5], df_index[6], df_index[7], df_index[8], df_index[9],
1364	open_cache_gave_first_time_s,
1365	without_retention,
1366	without_metric
1367	);
1368
1369	return global_first_time_s;
1370	}
1371
1372	static void update_metrics_first_time_s(struct rrdengine_instance ctx, struct rrdengine_datafile datafile_to_delete, struct rrdengine_datafile *first_datafile_remaining, bool worker) {
1373	time_t global_first_time_s = LONG_MAX;
1374
1375	if(worker)
1376	worker_is_busy(UV_EVENT_DBENGINE_FIND_ROTATED_METRICS);
1377
1378	struct rrdengine_journalfile *journalfile = datafile_to_delete->journalfile;
1379	struct journal_v2_header *j2_header = journalfile_v2_data_acquire_with_hint(
1380	journalfile, NULL, 0, 0, JOURNALFILE_V2_ACCESS_SEQUENTIAL_DIRECTORY);
1381
1382	if (unlikely(!j2_header)) {
1383	if (worker)
1384	worker_is_idle();
1385	return;
1386	}
1387
1388	__atomic_add_fetch(&rrdeng_cache_efficiency_stats.metrics_retention_started, 1, __ATOMIC_RELAXED);
1389
1390	char file_path[RRDENG_PATH_MAX];
1391	journalfile_v2_generate_path(datafile_to_delete, file_path, sizeof(file_path));
1392
1393	struct uuid_first_time_s *uuid_first_t_entry;
1394	// PROTECTED_ACCESS_SETUP below uses sigsetjmp/siglongjmp (see
1395	// src/daemon/protected-access.h). Per C11 7.13.2.1, non-volatile locals
1396	// that are modified between setjmp and longjmp have indeterminate values
1397	// on the recovery path. uuid_first_entry_list / count / added are all
1398	// mutated inside the protected region and then read afterwards (the
1399	// unconditional log line and the journal_access_failed cleanup loop), so
1400	// they must be volatile to keep the recovery path well-defined.
1401	// journal_access_failed is also marked volatile defensively: although it
1402	// is only assigned on a path that does not subsequently SIGBUS, future
1403	// edits could break that invariant, and the bool is read once on cleanup.
1404	struct uuid_first_time_s * volatile uuid_first_entry_list = NULL;
1405	volatile size_t count = 0;
1406	volatile size_t added = 0;
1407	volatile bool journal_access_failed = false;
1408
1409	// Protect the mmap walk: reading j2_header->metric_offset, metric_count,
1410	// and the per-metric uuid_list[] entries can SIGBUS if the underlying v2
1411	// file has any unreadable page (truncated, sparse hole, transient I/O
1412	// error). Without a protected region active the process aborts. Same
1413	// pattern as find_uuid_first_time() added by commit 26b26ac25a (#22310);
1414	// this caller was missed at the time.
1415	// Scope the protected frame tightly to the mmap walk only. The
1416	// PROTECTED_ACCESS_AUTO_CLEANUP() guard inside PROTECTED_ACCESS_SETUP
1417	// declares a __attribute__((cleanup)) local; when this inner block exits
1418	// (normally or via the SIGBUS recovery else-branch falling through), the
1419	// cleanup runs and the protected-access depth drops back to its prior
1420	// value. Without this scoping, the frame would stay live through the
1421	// post-walk log, the data_release, the cleanup loop, the nested
1422	// find_uuid_first_time() (which registers its own frame), and the final
1423	// cleanup -- masking unrelated faults that might land in the mmap range
1424	// and inflating nesting depth unnecessarily.
1425	{
1426	PROTECTED_ACCESS_SETUP(journalfile->mmap.data, journalfile->mmap.size, file_path, "mrg-retention");
1427	if(no_signal_received) {
1428	size_t journal_v2_file_size = journalfile->mmap.size;
1429	size_t metric_offset = j2_header->metric_offset;
1430	count = j2_header->metric_count;
1431	size_t metric_list_size;
1432	size_t entry_list_size;
1433	// Also check count * sizeof(uuid_first_time_s) -- the allocation below
1434	// sizes the working array by count, and on 32-bit builds count can
1435	// pass the metric_list_size bound while still overflowing the entry
1436	// list multiplication (struct uuid_first_time_s is larger than
1437	// struct journal_metric_list).
1438	if (__builtin_mul_overflow(count, sizeof(struct journal_metric_list), &metric_list_size) \|\|
1439	__builtin_mul_overflow(count, sizeof(struct uuid_first_time_s), &entry_list_size) \|\|
1440	metric_offset > journal_v2_file_size \|\|
1441	metric_list_size > journal_v2_file_size - metric_offset) {
1442	nd_log_daemon(NDLP_ERR,
1443	"DBENGINE: metric list exceeds journal file size in journalfile \"%s\" "
1444	"(metric_offset=%zu, list_size=%zu, file_size=%zu), skipping retention update",
1445	file_path, metric_offset, metric_list_size, journal_v2_file_size);
1446	journal_access_failed = true;
1447	}
1448	else {
1449	struct journal_metric_list uuid_list = (struct journal_metric_list )((uint8_t *) j2_header + metric_offset);
1450	uuid_first_entry_list = callocz(count, sizeof(struct uuid_first_time_s));
1451
1452	for (size_t index = 0; index < count; ++index) {
1453	// Copy uuid out of the mmap onto the stack BEFORE calling mrg.
1454	// If a backing page is unreadable, uuid_copy SIGBUSes here and
1455	// the protected region recovers cleanly; the mrg call then
1456	// never executes. If we passed &uuid_list[index].uuid into
1457	// mrg, a SIGBUS could fire INSIDE mrg while it holds internal
1458	// locks, and siglongjmp would skip mrg's unlock paths.
1459	nd_uuid_t local_uuid;
1460	uuid_copy(local_uuid, uuid_list[index].uuid);
1461
1462	METRIC *metric = mrg_metric_get_and_acquire_by_uuid(main_mrg, &local_uuid, (Word_t)ctx);
1463	if (!metric)
1464	continue;
1465
1466	uuid_first_entry_list[added].metric = metric;
1467	uuid_first_entry_list[added].first_time_s = LONG_MAX;
1468	uuid_first_entry_list[added].df_matched = 0;
1469	uuid_first_entry_list[added].df_index_oldest = 0;
1470	uuid_first_entry_list[added].uuid = mrg_metric_uuid(main_mrg, metric);
1471	added++;
1472	}
1473	}
1474	}
1475	else {
1476	// SIGBUS/SIGSEGV inside the mmap walk -- bail cleanly. The
1477	// PROTECTED_ACCESS_SETUP macro already rate-limits the error log.
1478	journal_access_failed = true;
1479	}
1480	}
1481
1482	netdata_log_info(
1483	"DBENGINE: tier %d: recalculating retention for %zu metrics starting with datafile %u",
1484	ctx->config.tier,
1485	count,
1486	first_datafile_remaining ? first_datafile_remaining->fileno : 0);
1487
1488	journalfile_v2_data_release(journalfile);
1489
1490	if (unlikely(journal_access_failed)) {
1491	// Release any partially-acquired metrics; uuid_first_entry_list may
1492	// be NULL (signal received before callocz).
1493	for (size_t index = 0; index < added; ++index)
1494	mrg_metric_release(main_mrg, uuid_first_entry_list[index].metric);
1495	goto done;
1496	}
1497
1498	// Update the first time / last time for all metrics we plan to delete
1499
1500	if(worker)
1501	worker_is_busy(UV_EVENT_DBENGINE_FIND_REMAINING_RETENTION);
1502
1503	global_first_time_s = find_uuid_first_time(ctx, first_datafile_remaining, uuid_first_entry_list, added);
1504
1505	if (!ctx_is_available_for_queries(ctx)) {
1506	for (size_t index = 0; index < added; ++index) {
1507	uuid_first_t_entry = &uuid_first_entry_list[index];
1508	mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1509	}
1510	goto done;
1511	}
1512
1513	if(worker)
1514	worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
1515
1516	netdata_log_info("DBENGINE: tier %d: updating metrics registry retention for %zu metrics", ctx->config.tier, added);
1517
1518	size_t deleted_metrics = 0, zero_retention_referenced = 0, zero_disk_retention = 0, zero_disk_but_live = 0;
1519	for (size_t index = 0; index < added; ++index) {
1520	uuid_first_t_entry = &uuid_first_entry_list[index];
1521
1522	if (!ctx_is_available_for_queries(ctx)) {
1523	mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1524	continue;
1525	}
1526
1527	if (likely(uuid_first_t_entry->first_time_s != LONG_MAX)) {
1528
1529	time_t old_first_time_s = mrg_metric_get_first_time_s(main_mrg, uuid_first_t_entry->metric);
1530
1531	bool changed = mrg_metric_set_first_time_s_if_bigger(main_mrg, uuid_first_t_entry->metric, uuid_first_t_entry->first_time_s);
1532	if (changed) {
1533	uint32_t update_every_s = mrg_metric_get_update_every_s(main_mrg, uuid_first_t_entry->metric);
1534	if (update_every_s && old_first_time_s && uuid_first_t_entry->first_time_s > old_first_time_s) {
1535	uint64_t remove_samples = (uuid_first_t_entry->first_time_s - old_first_time_s) / update_every_s;
1536	__atomic_sub_fetch(&ctx->atomic.samples, remove_samples, __ATOMIC_RELAXED);
1537	}
1538	}
1539	mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1540	}
1541	else {
1542	zero_disk_retention++;
1543
1544	// there is no retention for this metric
1545	bool has_retention = mrg_metric_has_zero_disk_retention(main_mrg, uuid_first_t_entry->metric);
1546	if (!has_retention) {
1547	time_t first_time_s = mrg_metric_get_first_time_s(main_mrg, uuid_first_t_entry->metric);
1548	time_t last_time_s = mrg_metric_get_latest_time_s(main_mrg, uuid_first_t_entry->metric);
1549	time_t update_every_s = mrg_metric_get_update_every_s(main_mrg, uuid_first_t_entry->metric);
1550	if (update_every_s && first_time_s && last_time_s) {
1551	uint64_t remove_samples = (first_time_s - last_time_s) / update_every_s;
1552	__atomic_sub_fetch(&ctx->atomic.samples, remove_samples, __ATOMIC_RELAXED);
1553	}
1554
1555	bool deleted = mrg_metric_release_and_delete(main_mrg, uuid_first_t_entry->metric);
1556	if(deleted)
1557	deleted_metrics++;
1558	else
1559	zero_retention_referenced++;
1560	}
1561	else {
1562	zero_disk_but_live++;
1563	mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
1564	}
1565	}
1566	}
1567
1568	if (!ctx_is_available_for_queries(ctx))
1569	goto done;
1570
1571	internal_error(zero_disk_retention,
1572	"DBENGINE: tier %d: deleted %zu metrics from metrics registry; %zu still had zero retention but were referenced "
1573	"(out of %zu total zero on-disk retention metrics, of which %zu have main cache retention)",
1574	ctx->config.tier, deleted_metrics, zero_retention_referenced, zero_disk_retention, zero_disk_but_live);
1575
1576	if(global_first_time_s != LONG_MAX)
1577	__atomic_store_n(&ctx->atomic.first_time_s, global_first_time_s, __ATOMIC_RELAXED);
1578
1579	done:
1580	freez(uuid_first_entry_list);
1581
1582	if(worker)
1583	worker_is_idle();
1584	}
1585
1586	void datafile_delete(
1587	struct rrdengine_instance *ctx,
1588	struct rrdengine_datafile *datafile,
1589	bool update_retention,
1590	bool disk_time,
1591	bool worker)
1592	{
1593	unsigned tier = ctx->config.tier;
1594	unsigned fileno = datafile->fileno;
1595
1596	if(worker)
1597	worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT);
1598
1599	bool datafile_got_for_deletion = datafile_acquire_for_deletion(datafile, false);
1600	size_t attempts = 0;
1601
1602	while (!datafile_got_for_deletion) {
1603	if(worker)
1604	worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE_WAIT);
1605
1606	datafile_got_for_deletion = datafile_acquire_for_deletion(datafile, false);
1607
1608	if (!datafile_got_for_deletion) {
1609	if(++attempts >= 30) {
1610	// pending_deletion is already set, blocking new acquires.
1611	// Bail out and let the next rotation cycle retry - lockers
1612	// will drain over time since no new ones can be added.
1613	netdata_log_error("DBENGINE: tier %u: " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL
1614	" could not be acquired for deletion after %zu attempts (%u lockers remain)"
1615	" - will retry on next rotation",
1616	tier, datafile->tier, fileno, attempts, datafile->users.lockers);
1617
1618	if(worker)
1619	worker_is_idle();
1620
1621	return;
1622	}
1623
1624	netdata_log_info("DBENGINE: tier %u: waiting for " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL
1625	" to be available for deletion, in use by %u users.",
1626	tier, datafile->tier, fileno, datafile->users.lockers);
1627
1628	__atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_spin, 1, __ATOMIC_RELAXED);
1629	sleep_usec(1 * USEC_PER_SEC);
1630	}
1631	}
1632
1633	if (update_retention)
1634	update_metrics_first_time_s(ctx, datafile, get_next_datafile(datafile, NULL, false), worker);
1635
1636	// if (!ctx_is_available_for_queries(ctx)) {
1637	// // agent is shutting down, we cannot continue
1638	// if(worker)
1639	// worker_is_idle();
1640	// return;
1641	// }
1642
1643	__atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_started, 1, __ATOMIC_RELAXED);
1644	netdata_log_info("DBENGINE: tier %u: deleting " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL " to maintain %s.",
1645	tier, datafile->tier, fileno, disk_time ? "disk quota" : "time retention");
1646
1647	if(worker)
1648	worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE);
1649
1650	struct rrdengine_journalfile *journal_file;
1651	size_t deleted_bytes, journal_file_bytes, datafile_bytes;
1652	uint8_t deleted_journal_files = 0;
1653	uint8_t expected_journal_files = JOURNALFILE_DELETED_V1;
1654	bool deleted_datafile = false;
1655	unsigned datafile_tier = datafile->tier;
1656	int ret;
1657
1658	netdata_rwlock_wrlock(&ctx->datafiles.rwlock);
1659	datafile_list_delete_unsafe(ctx, datafile);
1660	netdata_rwlock_wrunlock(&ctx->datafiles.rwlock);
1661
1662	journal_file = datafile->journalfile;
1663	datafile_bytes = datafile->pos;
1664	journal_file_bytes = journalfile_current_size(journal_file);
1665	size_t journal_v2_bytes = journalfile_v2_data_size_get(journal_file);
1666	if (journalfile_v2_data_available(journal_file))
1667	expected_journal_files \|= JOURNALFILE_DELETED_V2;
1668	deleted_bytes = 0;
1669
1670	// This will delete journalfile_v2 and journalfile_v1 (returns bitmask of JOURNALFILE_DELETED_V1/V2)
1671	deleted_journal_files = journalfile_destroy_unsafe(journal_file, datafile);
1672	if (deleted_journal_files & JOURNALFILE_DELETED_V1)
1673	deleted_bytes += journal_file_bytes;
1674	if (deleted_journal_files & JOURNALFILE_DELETED_V2)
1675	deleted_bytes += journal_v2_bytes;
1676	// This will delete the datafile
1677	ret = destroy_data_file_unsafe(datafile);
1678	if (!ret) {
1679	deleted_datafile = true;
1680	deleted_bytes += datafile_bytes;
1681	}
1682
1683	cleanup_datafile_epdl_structures(datafile);
1684
1685	memset(journal_file, 0, sizeof(*journal_file));
1686	memset(datafile, 0, sizeof(*datafile));
1687
1688	freez(journal_file);
1689	freez(datafile);
1690
1691	ctx_current_disk_space_decrease(ctx, deleted_bytes);
1692	char size_for_humans[128];
1693	size_snprintf(size_for_humans, sizeof(size_for_humans), deleted_bytes, "B", false);
1694
1695	bool del_ndf = deleted_datafile;
1696	bool del_njf = deleted_journal_files & JOURNALFILE_DELETED_V1;
1697	bool del_njfv2 = deleted_journal_files & JOURNALFILE_DELETED_V2;
1698	bool exp_njf = expected_journal_files & JOURNALFILE_DELETED_V1;
1699	bool exp_njfv2 = expected_journal_files & JOURNALFILE_DELETED_V2;
1700
1701	if (del_ndf && del_njf && del_njfv2)
1702	netdata_log_info("DBENGINE: tier %u: deleted " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL " (.ndf, .njf, .njfv2), reclaimed %s.",
1703	tier, datafile_tier, fileno, size_for_humans);
1704	else if (del_ndf && del_njf && !exp_njfv2)
1705	netdata_log_info("DBENGINE: tier %u: deleted " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL " (.ndf, .njf), reclaimed %s.",
1706	tier, datafile_tier, fileno, size_for_humans);
1707	else if (del_ndf \|\| del_njf \|\| del_njfv2) {
1708	BUFFER *removed = buffer_create(0, NULL);
1709	BUFFER *failed = buffer_create(0, NULL);
1710	const char *sep;
1711
1712	sep = "";
1713	if (del_ndf) { buffer_strcat(removed, sep); buffer_strcat(removed, ".ndf"); sep = ", "; }
1714	if (del_njf) { buffer_strcat(removed, sep); buffer_strcat(removed, ".njf"); sep = ", "; }
1715	if (del_njfv2) { buffer_strcat(removed, sep); buffer_strcat(removed, ".njfv2"); }
1716
1717	sep = "";
1718	if (!del_ndf) { buffer_strcat(failed, sep); buffer_strcat(failed, ".ndf"); sep = ", "; }
1719	if (exp_njf && !del_njf) { buffer_strcat(failed, sep); buffer_strcat(failed, ".njf"); sep = ", "; }
1720	if (exp_njfv2 && !del_njfv2) { buffer_strcat(failed, sep); buffer_strcat(failed, ".njfv2"); }
1721
1722	if(buffer_strlen(failed))
1723	netdata_log_error("DBENGINE: tier %u: partial delete of " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL
1724	" - removed: %s, failed: %s, reclaimed %s.",
1725	tier, datafile_tier, fileno,
1726	buffer_tostring(removed), buffer_tostring(failed), size_for_humans);
1727	else
1728	netdata_log_info("DBENGINE: tier %u: deleted " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL " (%s), reclaimed %s.",
1729	tier, datafile_tier, fileno, buffer_tostring(removed), size_for_humans);
1730	buffer_free(removed);
1731	buffer_free(failed);
1732	}
1733	else
1734	netdata_log_error("DBENGINE: tier %u: failed to delete " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL " to maintain %s.",
1735	tier, datafile_tier, fileno, disk_time ? "disk quota" : "time retention");
1736	}
1737
1738	static void database_rotate_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1739
1740	struct rrdengine_datafile *datafile = get_first_ctx_datafile(ctx, false);
1741	datafile_delete(ctx, datafile, ctx_is_available_for_queries(ctx), true, true);
1742
1743	rrdcontext_db_rotation();
1744
1745	return data;
1746	}
1747
1748	static void after_flush_all_hot_and_dirty_pages_of_section(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
1749	;
1750	}
1751
1752	static void flush_all_hot_and_dirty_pages_of_section_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1753	worker_is_busy(UV_EVENT_DBENGINE_QUIESCE);
1754	pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx);
1755
1756	for(size_t i = 0; i < pgc_max_flushers() ; i++)
1757	rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_MAIN, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1758
1759	return data;
1760	}
1761
1762	static void after_flush_dirty_pages_of_section(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
1763	;
1764	}
1765
1766	static void flush_dirty_pages_of_section_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1767	worker_is_busy(UV_EVENT_DBENGINE_FLUSH_DIRTY);
1768	pgc_flush_dirty_pages(main_cache, (Word_t)ctx);
1769
1770	for(size_t i = 0; i < pgc_max_flushers() ; i++)
1771	rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_MAIN, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1772
1773	return data;
1774	}
1775
1776	struct mrg_load_thread {
1777	uv_sem_t *sem;
1778	struct rrdengine_datafile *datafile;
1779	size_t *total;
1780	size_t *populated_datafiles;
1781	};
1782
1783	void journalfile_v2_populate_retention_to_mrg_worker(void *arg)
1784	{
1785	struct mrg_load_thread *mlt = arg;
1786	struct rrdengine_instance *ctx = mlt->datafile->ctx;
1787
1788	journalfile_v2_populate_retention_to_mrg(ctx, mlt->datafile->journalfile);
1789
1790	uv_sem_post(mlt->sem);
1791	}
1792
1793	static void *tier_mrg_load(
1794	struct rrdengine_instance *ctx __maybe_unused,
1795	void *data,
1796	struct completion *completion __maybe_unused,
1797	uv_work_t *req __maybe_unused)
1798	{
1799	worker_is_busy(UV_EVENT_DBENGINE_MRG_LOAD);
1800	struct mrg_load_thread *mlt = data;
1801	journalfile_v2_populate_retention_to_mrg_worker(mlt);
1802	mlt->datafile->populate_mrg.populated = true;
1803	spinlock_unlock(&mlt->datafile->populate_mrg.spinlock);
1804
1805	__atomic_add_fetch(mlt->populated_datafiles, 1, __ATOMIC_RELAXED);
1806	__atomic_sub_fetch(mlt->total, 1, __ATOMIC_RELEASE);
1807	freez(mlt);
1808	worker_is_idle();
1809	return NULL;
1810	}
1811
1812
1813	static void after_populate_mrg(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
1814	if (completion)
1815	completion_mark_complete(completion);
1816	}
1817
1818	static void *populate_mrg_tp_worker(
1819	struct rrdengine_instance *ctx,
1820	void *data,
1821	struct completion *completion __maybe_unused,
1822	uv_work_t *uv_work_req __maybe_unused)
1823	{
1824	worker_is_busy(UV_EVENT_DBENGINE_POPULATE_MRG);
1825
1826	struct mrg_load_thread *mlt = data;
1827	int tier = ctx->config.tier;
1828
1829	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
1830
1831	size_t total_datafiles = 0;
1832	size_t populated_datafiles = 0;
1833	struct rrdengine_datafile *df = NULL;
1834	while ((df = get_next_datafile(df, ctx, true))) {
1835	total_datafiles++;
1836	if (df->populate_mrg.populated)
1837	populated_datafiles++;
1838	}
1839	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
1840
1841	if (total_datafiles == 0) {
1842	nd_log_daemon(NDLP_WARNING, "DBENGINE: tier %d: no datafiles to populate MRG", tier);
1843	worker_is_idle();
1844	return data;
1845	}
1846
1847	size_t total = 0;
1848	Word_t last_index = 0;
1849	bool resume_scan = false;
1850	do {
1851	struct rrdengine_datafile *datafile = NULL;
1852
1853	// find a datafile to work on
1854	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
1855	Pvoid_t *Pvalue = NULL;
1856	Word_t Index = resume_scan ? last_index : 0;
1857	bool first_then_next = !resume_scan;
1858	while((Pvalue = JudyLFirstThenNext(ctx->datafiles.JudyL, &Index, &first_then_next))) {
1859	datafile = *Pvalue;
1860	if(!spinlock_trylock(&datafile->populate_mrg.spinlock)) {
1861	datafile = NULL;
1862	continue;
1863	}
1864
1865	if(datafile->populate_mrg.populated) {
1866	spinlock_unlock(&datafile->populate_mrg.spinlock);
1867	datafile = NULL;
1868	continue;
1869	}
1870	break;
1871	}
1872	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
1873
1874	if(!datafile)
1875	break;
1876
1877	// resume next scan from current position
1878	last_index = Index;
1879	resume_scan = true;
1880
1881	uv_sem_wait(mlt->sem);
1882	struct mrg_load_thread *local_mlt = callocz(1, sizeof(struct mrg_load_thread));
1883	local_mlt->datafile = datafile;
1884	local_mlt->sem = mlt->sem;
1885	local_mlt->total = &total;
1886	local_mlt->populated_datafiles = &populated_datafiles;
1887	__atomic_add_fetch(local_mlt->total, 1, __ATOMIC_RELAXED);
1888	rrdeng_enq_cmd(ctx, RRDENG_OPCODE_MRG_LOAD, local_mlt, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
1889	{
1890	nd_log_limit_static_thread_var(erl, 10, 0);
1891	size_t completed = __atomic_load_n(&populated_datafiles, __ATOMIC_RELAXED);
1892	nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO,
1893	"DBENGINE: tier %d: MRG population completed: %.2f%% (%zu/%zu)",
1894	tier, (completed * 100.0) / total_datafiles, completed, total_datafiles);
1895	}
1896	} while(1);
1897
1898	// We've queued all datafiles. Now wait for all worker threads to complete.
1899	size_t pending;
1900	do {
1901	pending = __atomic_load_n(&total, __ATOMIC_ACQUIRE);
1902	if (pending) {
1903	nd_log_limit_static_thread_var(erl, 10, 0);
1904	size_t completed = __atomic_load_n(&populated_datafiles, __ATOMIC_RELAXED);
1905	nd_log_limit(&erl, NDLS_DAEMON, NDLP_INFO,
1906	"DBENGINE: tier %d: MRG population completed: %.2f%% (%zu/%zu), waiting for %zu workers",
1907	tier, (completed * 100.0) / total_datafiles, completed, total_datafiles, pending);
1908	sleep_usec(10 * USEC_PER_MS);
1909	}
1910	} while (pending > 0);
1911
1912	worker_is_idle();
1913	return data;
1914	}
1915
1916	static void after_ctx_shutdown(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
1917	;
1918	}
1919
1920	static void ctx_shutdown_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1921	worker_is_busy(UV_EVENT_DBENGINE_SHUTDOWN);
1922
1923	bool logged = false;
1924	while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED) \|\|
1925	__atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED)) {
1926	if(!logged) {
1927	logged = true;
1928	netdata_log_info("DBENGINE: waiting for %zu inflight queries to finish to shutdown tier %d...",
1929	__atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED), ctx->config.tier);
1930	}
1931	sleep_usec(1 * USEC_PER_MS);
1932	}
1933
1934	completion_mark_complete(completion);
1935
1936	return data;
1937	}
1938
1939	static void cache_flush_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
1940	if (!main_cache)
1941	return data;
1942
1943	worker_is_busy(UV_EVENT_DBENGINE_FLUSH_MAIN_CACHE);
1944	while (pgc_flush_pages(main_cache))
1945	yield_the_processor();
1946
1947	return data;
1948	}
1949
1950	static void cache_evict_main_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *req __maybe_unused) {
1951	if (!main_cache)
1952	return data;
1953
1954	worker_is_busy(UV_EVENT_DBENGINE_EVICT_MAIN_CACHE);
1955	while (pgc_evict_pages(main_cache, 0, 0))
1956	yield_the_processor();
1957
1958	return data;
1959	}
1960
1961	static void cache_evict_open_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *req __maybe_unused) {
1962	if (!open_cache)
1963	return data;
1964
1965	worker_is_busy(UV_EVENT_DBENGINE_EVICT_OPEN_CACHE);
1966	while (pgc_evict_pages(open_cache, 0, 0))
1967	yield_the_processor();
1968
1969	return data;
1970	}
1971
1972	static void cache_evict_extent_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *req __maybe_unused) {
1973	if (!extent_cache)
1974	return data;
1975
1976	worker_is_busy(UV_EVENT_DBENGINE_EVICT_EXTENT_CACHE);
1977	while (pgc_evict_pages(extent_cache, 0, 0))
1978	yield_the_processor();
1979
1980	return data;
1981	}
1982
1983	static void query_prep_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *req __maybe_unused) {
1984	PDC *pdc = data;
1985	rrdeng_prep_query(pdc, true);
1986	return data;
1987	}
1988
1989	uint64_t rrdeng_target_data_file_size(struct rrdengine_instance *ctx) {
1990	uint64_t target_size = ctx->config.max_disk_space ? ctx->config.max_disk_space / TARGET_DATAFILES : MAX_DATAFILE_SIZE;
1991	target_size = MIN(target_size, MAX_DATAFILE_SIZE);
1992	target_size = MAX(target_size, MIN_DATAFILE_SIZE);
1993	return target_size;
1994	}
1995
1996	/* return 0 on success */
1997	int init_rrd_files(struct rrdengine_instance *ctx)
1998	{
1999	return init_data_files(ctx);
2000	}
2001
2002	void finalize_rrd_files(struct rrdengine_instance *ctx)
2003	{
2004	return finalize_data_files(ctx);
2005	}
2006
2007	#if defined(OS_WINDOWS)
2008	uint64_t last_async_callback;
2009
2010	void async_cb(uv_async_t *handle)
2011	{
2012	last_async_callback = uv_hrtime();
2013
2014	netdata_log_debug(D_RRDENGINE, "%s called, active=%d.", __func__, uv_is_active((uv_handle_t *)handle));
2015	}
2016
2017	static void async_closed_cb(uv_handle_t *handle)
2018	{
2019	struct rrdeng_main *main = handle->data;
2020
2021	int ret = uv_async_init(handle->loop, &main->async, async_cb);
2022	if (ret)
2023	netdata_log_error("DBENGINE: reinitializing uv_async_init(): %s", uv_strerror(ret));
2024	__atomic_store_n(&main->async_ready, true, __ATOMIC_RELEASE);
2025	}
2026	#else
2027	void async_cb(uv_async_t *handle __maybe_unused)
2028	{
2029	netdata_log_debug(D_RRDENGINE, "%s called, active=%d.", __func__, uv_is_active((uv_handle_t *)handle));
2030	}
2031	#endif
2032
2033	#define TIMER_PERIOD_MS (1000)
2034
2035	static void extent_read_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
2036	EPDL *epdl = data;
2037	epdl_find_extent_and_populate_pages(ctx, epdl, true);
2038	return data;
2039	}
2040
2041	static NOT_INLINE_HOT void epdl_populate_pages_asynchronously(struct rrdengine_instance ctx, EPDL epdl, STORAGE_PRIORITY priority) {
2042	rrdeng_enq_cmd(ctx, RRDENG_OPCODE_EXTENT_READ, epdl, NULL, priority,
2043	rrdeng_enqueue_epdl_cmd, rrdeng_dequeue_epdl_cmd);
2044	}
2045
2046	NOT_INLINE_HOT void pdc_route_asynchronously(struct rrdengine_instance ctx, struct page_details_control pdc) {
2047	pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_asynchronously, epdl_populate_pages_asynchronously);
2048	}
2049
2050	NOT_INLINE_HOT void epdl_populate_pages_synchronously(struct rrdengine_instance ctx, EPDL epdl, enum storage_priority priority __maybe_unused) {
2051	epdl_find_extent_and_populate_pages(ctx, epdl, false);
2052	}
2053
2054	NOT_INLINE_HOT void pdc_route_synchronously(struct rrdengine_instance ctx, struct page_details_control pdc) {
2055	pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_synchronously, epdl_populate_pages_synchronously);
2056	}
2057
2058	NOT_INLINE_HOT void pdc_route_synchronously_first(struct rrdengine_instance ctx, struct page_details_control pdc) {
2059	pdc_to_epdl_router(ctx, pdc, epdl_populate_pages_synchronously, epdl_populate_pages_asynchronously);
2060	}
2061
2062	static struct rrdengine_datafile release_and_aquire_next_datafile_for_indexing(struct rrdengine_instance ctx, struct rrdengine_datafile *release_datafile)
2063	{
2064	struct rrdengine_datafile *datafile = NULL;
2065
2066	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
2067	if (release_datafile) {
2068	datafile = get_next_datafile(release_datafile, NULL, true);
2069	datafile_release(release_datafile, DATAFILE_ACQUIRE_INDEXING);
2070	}
2071	else
2072	datafile = get_first_ctx_datafile(ctx, true);
2073
2074	while (datafile && datafile->fileno != ctx_last_fileno_get(ctx) && datafile->fileno != ctx_last_flush_fileno_get(ctx)) {
2075	if(journalfile_v2_data_available(datafile->journalfile)) {
2076	datafile = get_next_datafile(datafile, NULL, true);
2077	continue;
2078	}
2079
2080	int retries = 5;
2081	bool locked = false;
2082	while (retries-- > 0) {
2083	locked = datafile_acquire(datafile, DATAFILE_ACQUIRE_INDEXING);
2084	if (locked)
2085	break;
2086	sleep_usec(200 * USEC_PER_MS);
2087	}
2088	if (locked) {
2089	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
2090	return datafile;
2091	}
2092	nd_log_daemon(NDLP_INFO, "DBENGINE: tier %d: " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL " cannot be locked for indexing after retries; skipping",
2093	ctx->config.tier, datafile->tier, datafile->fileno);
2094	datafile = get_next_datafile(datafile, NULL, true);
2095	}
2096	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
2097	return NULL;
2098	}
2099
2100
2101	static void journal_v2_indexing_tp_worker(struct rrdengine_instance ctx, void data, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
2102	unsigned count = 0;
2103
2104	if (unlikely(!ctx_is_available_for_queries(ctx)))
2105	return data;
2106
2107	worker_is_busy(UV_EVENT_DBENGINE_JOURNAL_INDEX);
2108	struct rrdengine_datafile *datafile = NULL;
2109
2110	bool index_once = false;
2111	while ((datafile = release_and_aquire_next_datafile_for_indexing(ctx, datafile))) {
2112
2113	spinlock_lock(&datafile->writers.spinlock);
2114	bool available = (datafile->writers.running \|\| datafile->writers.flushed_to_open_running) ? false : true;
2115	spinlock_unlock(&datafile->writers.spinlock);
2116
2117	if(!available) {
2118	nd_log_daemon(NDLP_NOTICE,
2119	"DBENGINE: tier %d: " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL
2120	" needs to be indexed, but it has writers working on it - skipping it for now",
2121	ctx->config.tier, datafile->tier, datafile->fileno);
2122	continue;
2123	}
2124
2125	if (index_once && unlikely(rrdeng_ctx_tier_cap_exceeded(ctx))) {
2126	nd_log_daemon(
2127	NDLP_INFO, "DBENGINE: tier %d: reached quota limit, stopping journal indexing", ctx->config.tier);
2128	__atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);
2129	datafile_release(datafile, DATAFILE_ACQUIRE_INDEXING);
2130	break;
2131	}
2132	nd_log_daemon(NDLP_INFO, "DBENGINE: tier %d: " DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL " is ready to be indexed",
2133	ctx->config.tier, datafile->tier, datafile->fileno);
2134
2135	pgc_open_cache_to_journal_v2(
2136	open_cache,
2137	(Word_t)ctx,
2138	(int)datafile->fileno,
2139	ctx->config.page_type,
2140	journalfile_migrate_to_v2_callback,
2141	(void *)datafile->journalfile,
2142	false);
2143
2144	index_once = true;
2145
2146	count++;
2147
2148	// check if we are shutting down
2149	if (unlikely(!ctx_is_available_for_queries(ctx))) {
2150	datafile_release(datafile, DATAFILE_ACQUIRE_INDEXING);
2151	break;
2152	}
2153	}
2154
2155	errno_clear();
2156	if(count)
2157	nd_log(NDLS_DAEMON, NDLP_DEBUG,
2158	"DBENGINE: tier %d: journal indexing done; %u files processed",
2159	ctx->config.tier, count);
2160
2161	worker_is_idle();
2162
2163	return data;
2164	}
2165
2166	static void after_do_cache_flush(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
2167	rrdeng_main.flushes_running--;
2168	}
2169
2170	static void after_do_main_cache_evict(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
2171	rrdeng_main.evict_main_running--;
2172	}
2173
2174	static void after_do_open_cache_evict(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
2175	rrdeng_main.evict_open_running--;
2176	}
2177
2178	static void after_do_extent_cache_evict(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
2179	rrdeng_main.evict_extent_running--;
2180	}
2181
2182	static void after_journal_v2_indexing(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
2183	__atomic_store_n(&ctx->atomic.migration_to_v2_running, false, __ATOMIC_RELAXED);
2184
2185	check_and_schedule_db_rotation(ctx);
2186	}
2187
2188	struct rrdeng_buffer_sizes rrdeng_pulse_memory_sizes(void) {
2189	return (struct rrdeng_buffer_sizes) {
2190	.as = {
2191	[RRDENG_MEM_PGC] = pgc_aral_stats(),
2192	[RRDENG_MEM_PGD] = pgd_aral_stats(),
2193	[RRDENG_MEM_MRG] = mrg_aral_stats(),
2194	[RRDENG_MEM_PDC] = pdc_aral_stats(),
2195	[RRDENG_MEM_EPDL] = epdl_aral_stats(),
2196	[RRDENG_MEM_DEOL] = deol_aral_stats(),
2197	[RRDENG_MEM_PD] = pd_aral_stats(),
2198	[RRDENG_MEM_EPDL_EXTENT] = epdl_extent_aral_stats(),
2199	[RRDENG_MEM_OPCODES] = aral_get_statistics(rrdeng_main.cmd_queue.ar),
2200	[RRDENG_MEM_HANDLES] = aral_get_statistics(rrdeng_main.handles.ar),
2201	[RRDENG_MEM_DESCRIPTORS] = aral_get_statistics(rrdeng_main.descriptors.ar),
2202	[RRDENG_MEM_WORKERS] = aral_get_statistics(rrdeng_main.work_cmd.ar),
2203	[RRDENG_MEM_XT_IO] = aral_get_statistics(rrdeng_main.xt_io_descr.ar),
2204	},
2205	.wal = __atomic_load_n(&wal_globals.atomics.allocated, __ATOMIC_RELAXED) * (sizeof(WAL) + RRDENG_BLOCK_SIZE),
2206	.xt_buf = extent_buffer_cache_size(),
2207	};
2208	}
2209
2210	static void after_cleanup(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t req __maybe_unused, int status __maybe_unused) {
2211	rrdeng_main.cleanup_running--;
2212	}
2213
2214	static void cleanup_tp_worker(struct rrdengine_instance ctx __maybe_unused, void data __maybe_unused, struct completion completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {
2215	worker_is_busy(UV_EVENT_DBENGINE_BUFFERS_CLEANUP);
2216
2217	wal_cleanup1();
2218	extent_buffer_cleanup1();
2219
2220	{
2221	static time_t last_run_s = 0;
2222	time_t now_s = now_monotonic_sec();
2223	if(now_s - last_run_s >= 10) {
2224	last_run_s = now_s;
2225	journalfile_v2_data_unmount_cleanup(now_s);
2226	}
2227	}
2228
2229	return data;
2230	}
2231
2232	uint64_t rrdeng_get_used_disk_space(struct rrdengine_instance *ctx, bool having_lock)
2233	{
2234	uint64_t active_space = 0;
2235
2236	if (!having_lock)
2237	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
2238
2239	struct rrdengine_datafile *first_datafile = get_first_ctx_datafile(ctx, true);
2240	struct rrdengine_datafile *last_datafile = get_last_ctx_datafile(ctx, true);
2241
2242	if (first_datafile && last_datafile)
2243	active_space = last_datafile->pos;
2244
2245	if (!having_lock)
2246	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
2247
2248	// calculate the estimated disk space based on the expected final size of the datafile
2249	// We cant know the final v1/v2 journal size -- we let the current v1 size be part of the calculation by not
2250	// including it in the active_space
2251	uint64_t estimated_disk_space = ctx_current_disk_space_get(ctx) + rrdeng_target_data_file_size(ctx) - active_space;
2252
2253	uint64_t database_space = get_total_database_space();
2254	uint64_t adjusted_database_space = database_space * ctx->config.disk_percentage / 100 ;
2255	estimated_disk_space += adjusted_database_space;
2256
2257	return estimated_disk_space;
2258	}
2259
2260	// Check if disk or retention time cap reached
2261	bool rrdeng_ctx_tier_cap_exceeded(struct rrdengine_instance *ctx)
2262	{
2263	bool trigger_time_retention = false;
2264	uint64_t estimated_disk_space = 0;
2265
2266	netdata_rwlock_rdlock(&ctx->datafiles.rwlock);
2267	struct rrdengine_datafile *first_datafile = get_first_ctx_datafile(ctx, true);
2268
2269	if (!first_datafile \|\| datafile_count(ctx, true) < 2) {
2270	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
2271	return false;
2272	}
2273
2274	if (ctx->config.max_retention_s) {
2275	time_t last_time_s = first_datafile->journalfile->v2.last_time_s;
2276	if (!last_time_s)
2277	last_time_s = first_datafile->journalfile->v2.first_time_s;
2278
2279	time_t cutoff_before_time_s = now_realtime_sec() - ctx->config.max_retention_s;
2280	trigger_time_retention = (last_time_s && last_time_s <= cutoff_before_time_s);
2281	}
2282
2283	// avoid disk calculation if we will trigger time retention
2284	// calculate estimated disk space only if we have a disk cap
2285	if (false == trigger_time_retention && ctx->config.max_disk_space)
2286	estimated_disk_space = rrdeng_get_used_disk_space(ctx, true);
2287
2288	netdata_rwlock_rdunlock(&ctx->datafiles.rwlock);
2289
2290	if (trigger_time_retention) {
2291	__atomic_store_n(&ctx->datafiles.disk_time, false, __ATOMIC_RELAXED);
2292	return true;
2293	}
2294
2295	if (ctx->config.max_disk_space && estimated_disk_space > ctx->config.max_disk_space) {
2296	__atomic_store_n(&ctx->datafiles.disk_time, true, __ATOMIC_RELAXED);
2297	return true;
2298	}
2299
2300	return false;
2301	}
2302
2303	static void retention_timer_cb(uv_timer_t *handle __maybe_unused)
2304	{
2305	if (!localhost)
2306	return;
2307
2308	worker_is_busy(RRDENG_RETENTION_TIMER_CB);
2309
2310	for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++) {
2311	STORAGE_ENGINE *eng = localhost->db[tier].eng;
2312	if (!eng \|\| eng->seb != STORAGE_ENGINE_BACKEND_DBENGINE)
2313	continue;
2314	check_and_schedule_db_rotation(multidb_ctx[tier]);
2315	}
2316
2317	worker_is_idle();
2318	}
2319
2320	static void timer_per_sec_cb(uv_timer_t *handle __maybe_unused)
2321	{
2322	worker_is_busy(RRDENG_TIMER_CB);
2323
2324	worker_set_metric(RRDENG_OPCODES_WAITING, (NETDATA_DOUBLE)rrdeng_main.cmd_queue.unsafe.waiting);
2325	worker_set_metric(RRDENG_WORKS_DISPATCHED, (NETDATA_DOUBLE)__atomic_load_n(&rrdeng_main.work_cmd.atomics.dispatched, __ATOMIC_RELAXED));
2326	worker_set_metric(RRDENG_WORKS_EXECUTING, (NETDATA_DOUBLE)__atomic_load_n(&rrdeng_main.work_cmd.atomics.executing, __ATOMIC_RELAXED));
2327
2328	rrdeng_enq_cmd(NULL, RRDENG_OPCODE_FLUSH_MAIN, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
2329	rrdeng_enq_cmd(NULL, RRDENG_OPCODE_CLEANUP, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
2330
2331	worker_is_idle();
2332	}
2333
2334	static void dbengine_initialize_structures(void) {
2335	pgd_init_arals();
2336	pgc_and_mrg_initialize();
2337
2338	pdc_init();
2339	page_details_init();
2340	epdl_init();
2341	deol_init();
2342	epdl_extent_init();
2343	rrdeng_cmd_queue_init();
2344	work_request_init();
2345	rrdeng_query_handle_init();
2346	page_descriptors_init();
2347	extent_buffer_init();
2348	extent_io_descriptor_init();
2349	}
2350
2351	bool rrdeng_dbengine_spawn(struct rrdengine_instance *ctx __maybe_unused) {
2352	static bool spawned = false;
2353	static SPINLOCK spinlock = SPINLOCK_INITIALIZER;
2354
2355	spinlock_lock(&spinlock);
2356
2357	if(!spawned) {
2358	int ret;
2359
2360	ret = uv_loop_init(&rrdeng_main.loop);
2361	if (ret) {
2362	netdata_log_error("DBENGINE: uv_loop_init(): %s", uv_strerror(ret));
2363	return false;
2364	}
2365	rrdeng_main.loop.data = &rrdeng_main;
2366
2367	ret = uv_async_init(&rrdeng_main.loop, &rrdeng_main.async, async_cb);
2368	if (ret) {
2369	netdata_log_error("DBENGINE: uv_async_init(): %s", uv_strerror(ret));
2370	fatal_assert(0 == uv_loop_close(&rrdeng_main.loop));
2371	return false;
2372	}
2373	rrdeng_main.async.data = &rrdeng_main;
2374	#if defined(OS_WINDOWS)
2375	rrdeng_main.async_ready = true;
2376	#endif
2377
2378	ret = uv_timer_init(&rrdeng_main.loop, &rrdeng_main.timer);
2379	if (ret) {
2380	netdata_log_error("DBENGINE: uv_timer_init(): %s", uv_strerror(ret));
2381	uv_close((uv_handle_t *)&rrdeng_main.async, NULL);
2382	fatal_assert(0 == uv_loop_close(&rrdeng_main.loop));
2383	return false;
2384	}
2385
2386	ret = uv_timer_init(&rrdeng_main.loop, &rrdeng_main.retention_timer);
2387	if (ret) {
2388	netdata_log_error("DBENGINE: uv_timer_init(): %s", uv_strerror(ret));
2389	uv_close((uv_handle_t *)&rrdeng_main.async, NULL);
2390	fatal_assert(0 == uv_loop_close(&rrdeng_main.loop));
2391	return false;
2392	}
2393
2394	rrdeng_main.timer.data = &rrdeng_main;
2395	rrdeng_main.retention_timer.data = &rrdeng_main;
2396
2397	dbengine_initialize_structures();
2398
2399	int retries = 0;
2400	rrdeng_main.thread = nd_thread_create("DBEV", NETDATA_THREAD_OPTION_DEFAULT, dbengine_event_loop, &rrdeng_main);
2401
2402	fatal_assert(0 != rrdeng_main.thread);
2403
2404	if (retries)
2405	nd_log_daemon(NDLP_WARNING, "DBENGINE thread was created after %d attempts", retries);
2406
2407	spawned = true;
2408	}
2409
2410	spinlock_unlock(&spinlock);
2411	return true;
2412	}
2413
2414	static inline void worker_dispatch_extent_read(struct rrdeng_cmd cmd, bool from_worker) {
2415	struct rrdengine_instance *ctx = cmd.ctx;
2416	EPDL *epdl = cmd.data;
2417
2418	if(from_worker)
2419	epdl_find_extent_and_populate_pages(ctx, epdl, true);
2420	else
2421	work_dispatch(ctx, epdl, NULL, cmd.opcode, extent_read_tp_worker, NULL);
2422	}
2423
2424	static inline void worker_dispatch_query_prep(struct rrdeng_cmd cmd, bool from_worker) {
2425	struct rrdengine_instance *ctx = cmd.ctx;
2426	PDC *pdc = cmd.data;
2427
2428	if(from_worker)
2429	rrdeng_prep_query(pdc, true);
2430	else
2431	work_dispatch(ctx, pdc, NULL, cmd.opcode, query_prep_tp_worker, NULL);
2432	}
2433
2434	uint64_t rrdeng_get_directory_free_bytes_space(struct rrdengine_instance *ctx)
2435	{
2436	uint64_t free_bytes = 0;
2437	OS_SYSTEM_DISK_SPACE space = os_disk_space(ctx->config.dbfiles_path);
2438	free_bytes = OS_SYSTEM_DISK_SPACE_OK(space) ? space.free_bytes : 0;
2439	return (free_bytes - (free_bytes * 5 / 100));
2440	}
2441
2442	void rrdeng_calculate_tier_disk_space_percentage(void)
2443	{
2444	uint64_t tier_space[RRD_STORAGE_TIERS];
2445
2446	if (!localhost)
2447	return;
2448
2449	uint64_t total_diskspace = 0;
2450	for(size_t tier = 0; tier < nd_profile.storage_tiers;tier++) {
2451	STORAGE_ENGINE *eng = localhost->db[tier].eng;
2452	if (!eng \|\| eng->seb != STORAGE_ENGINE_BACKEND_DBENGINE) {
2453	tier_space[tier] = 0;
2454	continue;
2455	}
2456	uint64_t tier_disk_space = multidb_ctx[tier]->config.max_disk_space ?
2457	multidb_ctx[tier]->config.max_disk_space :
2458	rrdeng_get_directory_free_bytes_space(multidb_ctx[tier]);
2459	total_diskspace += tier_disk_space;
2460	tier_space[tier] = tier_disk_space;
2461	}
2462
2463	if (total_diskspace) {
2464	for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++) {
2465	multidb_ctx[tier]->config.disk_percentage = (100 * tier_space[tier] / total_diskspace);
2466	}
2467	}
2468	}
2469
2470	#define NOT_DELETING_FILES(ctx) \
2471	(!__atomic_load_n(&(ctx)->atomic.now_deleting_files, __ATOMIC_RELAXED))
2472
2473	#define NOT_INDEXING_FILES(ctx) \
2474	(!__atomic_load_n(&(ctx)->atomic.migration_to_v2_running, __ATOMIC_RELAXED))
2475
2476	void dbengine_event_loop(void* arg) {
2477	sanity_check();
2478	uv_thread_set_name_np("DBENGINE");
2479	service_register(NULL, NULL, NULL);
2480
2481	worker_register("DBENGINE");
2482
2483	// opcode jobs
2484	worker_register_job_name(RRDENG_OPCODE_NOOP, "noop");
2485
2486	worker_register_job_name(RRDENG_OPCODE_QUERY, "query");
2487	worker_register_job_name(RRDENG_OPCODE_EXTENT_WRITE, "extent write");
2488	worker_register_job_name(RRDENG_OPCODE_EXTENT_READ, "extent read");
2489	worker_register_job_name(RRDENG_OPCODE_DATABASE_ROTATE, "db rotate");
2490	worker_register_job_name(RRDENG_OPCODE_JOURNAL_INDEX, "journal index");
2491	worker_register_job_name(RRDENG_OPCODE_FLUSH_MAIN, "flush init");
2492	worker_register_job_name(RRDENG_OPCODE_EVICT_MAIN, "evict init");
2493	worker_register_job_name(RRDENG_OPCODE_CTX_SHUTDOWN, "ctx shutdown");
2494	worker_register_job_name(RRDENG_OPCODE_CTX_FLUSH_DIRTY, "ctx flush dirty");
2495	worker_register_job_name(RRDENG_OPCODE_CTX_FLUSH_HOT_DIRTY, "ctx flush all");
2496	worker_register_job_name(RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce");
2497	worker_register_job_name(RRDENG_OPCODE_SHUTDOWN_EVLOOP, "dbengine shutdown");
2498	worker_register_job_name(RRDENG_OPCODE_PARALLEL_WEIGHT, "parallel weight");
2499	worker_register_job_name(RRDENG_OPCODE_MRG_LOAD, "mrg tier load");
2500
2501
2502	worker_register_job_name(RRDENG_OPCODE_MAX, "get opcode");
2503
2504	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_QUERY, "query cb");
2505	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EXTENT_WRITE, "extent write cb");
2506	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EXTENT_READ, "extent read cb");
2507	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_DATABASE_ROTATE, "db rotate cb");
2508	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_JOURNAL_INDEX, "journal index cb");
2509	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_FLUSH_MAIN, "flush init cb");
2510	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_EVICT_MAIN, "evict init cb");
2511	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_SHUTDOWN, "ctx shutdown cb");
2512	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_FLUSH_DIRTY, "ctx flush dirty cb");
2513	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_CTX_QUIESCE, "ctx quiesce cb");
2514	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_PARALLEL_WEIGHT, "parallel weight cb");
2515	worker_register_job_name(RRDENG_OPCODE_MAX + RRDENG_OPCODE_MRG_LOAD, "mrg tier load cb");
2516
2517	// special jobs
2518	worker_register_job_name(RRDENG_RETENTION_TIMER_CB, "retention timer");
2519	worker_register_job_name(RRDENG_TIMER_CB, "timer");
2520
2521	worker_register_job_custom_metric(RRDENG_OPCODES_WAITING, "opcodes waiting", "opcodes", WORKER_METRIC_ABSOLUTE);
2522	worker_register_job_custom_metric(RRDENG_WORKS_DISPATCHED, "works dispatched", "works", WORKER_METRIC_ABSOLUTE);
2523	worker_register_job_custom_metric(RRDENG_WORKS_EXECUTING, "works executing", "works", WORKER_METRIC_ABSOLUTE);
2524
2525	struct rrdeng_main *main = arg;
2526	enum rrdeng_opcode opcode;
2527	struct rrdeng_cmd cmd;
2528	main->tid = gettid_cached();
2529
2530	fatal_assert(0 == uv_timer_start(&main->timer, timer_per_sec_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));
2531	fatal_assert(0 == uv_timer_start(&main->retention_timer, retention_timer_cb, TIMER_PERIOD_MS * 60, TIMER_PERIOD_MS * 60));
2532
2533	bool shutdown = false;
2534	size_t cpus = netdata_conf_cpus();
2535	uv_sem_t sem;
2536	uv_sem_init(&sem, (unsigned int) cpus);
2537	struct mrg_load_thread mlt = callocz(cpus, sizeof(mlt));
2538	for (size_t i = 0; i < cpus; i++) {
2539	mlt[i].sem = &sem;
2540	}
2541
2542	#if defined(OS_WINDOWS)
2543	last_async_callback = uv_hrtime();
2544	#endif
2545
2546	while (likely(!shutdown)) {
2547	worker_is_idle();
2548	uv_run(&main->loop, UV_RUN_ONCE);
2549
2550	do {
2551	worker_is_busy(RRDENG_OPCODE_MAX);
2552	cmd = rrdeng_deq_cmd(RRDENG_OPCODE_NOOP);
2553	opcode = cmd.opcode;
2554
2555	worker_is_busy(opcode);
2556
2557	switch (opcode) {
2558	case RRDENG_OPCODE_MRG_LOAD:
2559	work_dispatch(NULL, cmd.data, cmd.completion, cmd.opcode, tier_mrg_load, NULL);
2560	break;
2561
2562	case RRDENG_OPCODE_PARALLEL_WEIGHT:;
2563
2564	work_dispatch(NULL, cmd.data, cmd.completion, cmd.opcode, weights_worker, after_weights_worker);
2565	break;
2566
2567	case RRDENG_OPCODE_EXTENT_READ:
2568	worker_dispatch_extent_read(cmd, false);
2569	break;
2570
2571	case RRDENG_OPCODE_QUERY:;
2572	#if defined(OS_WINDOWS)
2573	static int max_timeout_count = 0;
2574	if (uv_hrtime() - last_async_callback > 1000UL * NSEC_PER_MSEC) {
2575	if (++max_timeout_count > 30) {
2576	netdata_log_error("DBENGINE: async callback timeout detected, re-initializing the async handle");
2577	__atomic_store_n(&main->async_ready, false, __ATOMIC_RELEASE);
2578	uv_close((uv_handle_t *)&main->async, async_closed_cb);
2579	max_timeout_count = 0;
2580	} else
2581	netdata_log_error("DBENGINE: async callback timeout detected count = %d", max_timeout_count);
2582	}
2583	#endif
2584	worker_dispatch_query_prep(cmd, false);
2585	break;
2586
2587	case RRDENG_OPCODE_EXTENT_WRITE: {
2588	struct rrdengine_instance *ctx = cmd.ctx;
2589	struct page_descr_with_data *base = cmd.data;
2590	struct completion *completion = cmd.completion; // optional
2591	work_dispatch(ctx, base, completion, opcode, extent_write_tp_worker, after_extent_write);
2592	break;
2593	}
2594
2595	case RRDENG_OPCODE_FLUSH_MAIN: {
2596	if(rrdeng_main.flushes_running < pgc_max_flushers()) {
2597	rrdeng_main.flushes_running++;
2598	work_dispatch(NULL, NULL, NULL, opcode, cache_flush_tp_worker, after_do_cache_flush);
2599	}
2600	break;
2601	}
2602
2603	case RRDENG_OPCODE_EVICT_MAIN: {
2604	if(rrdeng_main.evict_main_running < pgc_max_evictors()) {
2605	rrdeng_main.evict_main_running++;
2606	work_dispatch(NULL, NULL, NULL, opcode, cache_evict_main_tp_worker, after_do_main_cache_evict);
2607	}
2608	break;
2609	}
2610
2611	case RRDENG_OPCODE_EVICT_OPEN: {
2612	if(rrdeng_main.evict_open_running < pgc_max_evictors()) {
2613	rrdeng_main.evict_open_running++;
2614	work_dispatch(NULL, NULL, NULL, opcode, cache_evict_open_tp_worker, after_do_open_cache_evict);
2615	}
2616	break;
2617	}
2618
2619	case RRDENG_OPCODE_EVICT_EXTENT: {
2620	if(rrdeng_main.evict_extent_running < pgc_max_evictors()) {
2621	rrdeng_main.evict_extent_running++;
2622	work_dispatch(NULL, NULL, NULL, opcode, cache_evict_extent_tp_worker, after_do_extent_cache_evict);
2623	}
2624	break;
2625	}
2626
2627	case RRDENG_OPCODE_CLEANUP: {
2628	if(!rrdeng_main.cleanup_running) {
2629	rrdeng_main.cleanup_running++;
2630	work_dispatch(NULL, NULL, NULL, opcode, cleanup_tp_worker, after_cleanup);
2631	}
2632	break;
2633	}
2634
2635	case RRDENG_OPCODE_JOURNAL_INDEX: {
2636	struct rrdengine_instance *ctx = cmd.ctx;
2637	struct rrdengine_datafile *datafile = cmd.data;
2638	// We no longer have an indexing command pending
2639	ctx->datafiles.pending_index = false;
2640	if (NOT_INDEXING_FILES(ctx) && ctx_is_available_for_queries(ctx)) {
2641	__atomic_store_n(&ctx->atomic.migration_to_v2_running, true, __ATOMIC_RELAXED);
2642	__atomic_store_n(&ctx->atomic.needs_indexing, false, __ATOMIC_RELAXED);
2643	work_dispatch(ctx, datafile, NULL, opcode, journal_v2_indexing_tp_worker, after_journal_v2_indexing);
2644	}
2645	break;
2646	}
2647
2648	case RRDENG_OPCODE_DATABASE_ROTATE: {
2649	struct rrdengine_instance *ctx = cmd.ctx;
2650	ctx->datafiles.pending_rotate = false;
2651	if (NOT_DELETING_FILES(ctx) && datafile_count(ctx, false) > 2 &&
2652	rrdeng_ctx_tier_cap_exceeded(ctx)) {
2653	__atomic_store_n(&ctx->atomic.now_deleting_files, true, __ATOMIC_RELAXED);
2654	work_dispatch(ctx, NULL, NULL, opcode, database_rotate_tp_worker, after_database_rotate);
2655	}
2656	break;
2657	}
2658
2659	case RRDENG_OPCODE_CTX_POPULATE_MRG: {
2660	struct rrdengine_instance *ctx = cmd.ctx;
2661	struct completion *completion = cmd.completion;
2662	work_dispatch(ctx, mlt, completion, opcode, populate_mrg_tp_worker, after_populate_mrg);
2663	break;
2664	}
2665
2666	case RRDENG_OPCODE_CTX_FLUSH_DIRTY: {
2667	struct rrdengine_instance *ctx = cmd.ctx;
2668	work_dispatch(ctx, NULL, NULL, opcode,
2669	flush_dirty_pages_of_section_tp_worker,
2670	after_flush_dirty_pages_of_section);
2671	break;
2672	}
2673
2674	case RRDENG_OPCODE_CTX_FLUSH_HOT_DIRTY: {
2675	struct rrdengine_instance *ctx = cmd.ctx;
2676	work_dispatch(ctx, NULL, NULL, opcode,
2677	flush_all_hot_and_dirty_pages_of_section_tp_worker,
2678	after_flush_all_hot_and_dirty_pages_of_section);
2679	break;
2680	}
2681
2682	case RRDENG_OPCODE_CTX_QUIESCE: {
2683	// a ctx will shutdown shortly
2684	struct rrdengine_instance *ctx = cmd.ctx;
2685	nd_log_daemon(NDLP_INFO, "DBENGINE: Tier %d is shutting down — query processing disabled", ctx->config.tier);
2686	__atomic_store_n(&ctx->quiesce.enabled, true, __ATOMIC_RELEASE);
2687	break;
2688	}
2689
2690	case RRDENG_OPCODE_CTX_SHUTDOWN: {
2691	// a ctx is shutting down
2692	struct rrdengine_instance *ctx = cmd.ctx;
2693	struct completion *completion = cmd.completion;
2694	work_dispatch(ctx, NULL, completion, opcode, ctx_shutdown_tp_worker, after_ctx_shutdown);
2695	break;
2696	}
2697
2698	case RRDENG_OPCODE_SHUTDOWN_EVLOOP: {
2699	uv_close((uv_handle_t *)&main->async, NULL);
2700
2701	(void) uv_timer_stop(&main->timer);
2702	uv_close((uv_handle_t *)&main->timer, NULL);
2703
2704	(void) uv_timer_stop(&main->retention_timer);
2705	uv_close((uv_handle_t *)&main->retention_timer, NULL);
2706	shutdown = true;
2707	break;
2708	}
2709
2710	case RRDENG_OPCODE_NOOP: {
2711	/* the command queue was empty, do nothing */
2712	break;
2713	}
2714
2715	// not opcodes
2716	case RRDENG_OPCODE_MAX:
2717	default: {
2718	internal_fatal(true, "DBENGINE: unknown opcode");
2719	break;
2720	}
2721	}
2722	if (opcode != RRDENG_OPCODE_NOOP)
2723	uv_run(&main->loop, UV_RUN_NOWAIT);
2724
2725	} while (opcode != RRDENG_OPCODE_NOOP);
2726	}
2727	freez(mlt);
2728	uv_sem_destroy(&sem);
2729
2730	nd_log(NDLS_DAEMON, NDLP_DEBUG, "Shutting down dbengine thread");
2731	(void) uv_loop_close(&main->loop);
2732	worker_unregister();
2733	}
2734
2735	void dbengine_shutdown()
2736	{
2737	rrdeng_enq_cmd(NULL, RRDENG_OPCODE_SHUTDOWN_EVLOOP, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
2738
2739	int rc = nd_thread_join(rrdeng_main.thread);
2740	if (rc)
2741	nd_log_daemon(NDLP_ERR, "DBENGINE: Failed to join thread, error %s", uv_err_name(rc));
2742	else
2743	nd_log_daemon(NDLP_INFO, "DBENGINE: thread shutdown completed");
2744	}