cache.c · cryptotaxi247/netdata-1 · siGit Code & Deploy

Code Commits Issues Pulls Sessions CI/CD

master

c 3,192 lines 118 KB

Raw

1	// SPDX-License-Identifier: GPL-3.0-or-later
2	#include "cache.h"
3
4	/* STATES AND TRANSITIONS
5	*
6	* entry \| entry
7	* v v
8	* HOT -> DIRTY --> CLEAN --> EVICT
9	* v \| v
10	* flush \| evict
11	* v \| v
12	* save \| free
13	* callback \| callback
14	*
15	*/
16
17	// to use ARAL uncomment the following line:
18	#if !defined(FSANITIZE_ADDRESS)
19	#define PGC_WITH_ARAL 1
20	#endif
21
22	#define PGC_QUEUE_LOCK_AS_WAITING_QUEUE 1
23
24	typedef enum __attribute__ ((__packed__)) {
25	// mutually exclusive flags
26	PGC_PAGE_CLEAN = (1 << 0), // none of the following
27	PGC_PAGE_DIRTY = (1 << 1), // contains unsaved data
28	PGC_PAGE_HOT = (1 << 2), // currently being collected
29
30	// flags related to various actions on each page
31	PGC_PAGE_IS_BEING_DELETED = (1 << 3),
32	PGC_PAGE_IS_BEING_MIGRATED_TO_V2 = (1 << 4),
33	PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES = (1 << 5),
34	PGC_PAGE_HAS_BEEN_ACCESSED = (1 << 6),
35	} PGC_PAGE_FLAGS;
36
37	#define page_flag_check(page, flag) (__atomic_load_n(&((page)->flags), __ATOMIC_ACQUIRE) & (flag))
38	#define page_flag_set(page, flag) __atomic_or_fetch(&((page)->flags), flag, __ATOMIC_RELEASE)
39	#define page_flag_clear(page, flag) __atomic_and_fetch(&((page)->flags), ~(flag), __ATOMIC_RELEASE)
40
41	#define page_get_status_flags(page) page_flag_check(page, PGC_PAGE_HOT \| PGC_PAGE_DIRTY \| PGC_PAGE_CLEAN)
42	#define is_page_hot(page) (page_get_status_flags(page) == PGC_PAGE_HOT)
43	#define is_page_dirty(page) (page_get_status_flags(page) == PGC_PAGE_DIRTY)
44	#define is_page_clean(page) (page_get_status_flags(page) == PGC_PAGE_CLEAN)
45
46	struct pgc_page {
47	// indexing data
48	Word_t section;
49	Word_t metric_id;
50	time_t start_time_s;
51	time_t end_time_s;
52	uint32_t update_every_s;
53	uint32_t assumed_size;
54
55	REFCOUNT refcount;
56	uint16_t accesses; // counts the number of accesses on this page
57	PGC_PAGE_FLAGS flags;
58	SPINLOCK transition_spinlock; // when the page changes between HOT, DIRTY, CLEAN, we have to get this lock
59
60	struct {
61	struct pgc_page *next;
62	struct pgc_page *prev;
63	} link;
64
65	void *data;
66	uint8_t custom_data[];
67
68	// IMPORTANT!
69	// THIS STRUCTURE NEEDS TO BE INITIALIZED BY HAND!
70	};
71
72	struct pgc_queue {
73	#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
74	WAITQ wq;
75	#else
76	SPINLOCK spinlock;
77	#endif
78	union {
79	PGC_PAGE *base;
80	Pvoid_t sections_judy;
81	};
82	PGC_PAGE_FLAGS flags;
83	size_t version;
84	size_t last_version_checked;
85	bool linked_list_in_sections_judy; // when true, we use 'sections_judy', otherwise we use 'base'
86	struct pgc_queue_statistics *stats;
87	};
88
89	struct pgc {
90	struct {
91	char name[PGC_NAME_MAX + 1];
92	bool stats; // enable extended statistics
93	bool use_all_ram;
94
95	size_t partitions;
96	int64_t clean_size;
97	size_t max_dirty_pages_per_call;
98	size_t max_pages_per_inline_eviction;
99	size_t max_skip_pages_per_inline_eviction;
100	size_t max_flushes_inline;
101	size_t max_workers_evict_inline;
102	size_t additional_bytes_per_page;
103	int64_t out_of_memory_protection_bytes;
104	free_clean_page_callback pgc_free_clean_cb;
105	save_dirty_page_callback pgc_save_dirty_cb;
106	save_dirty_init_callback pgc_save_init_cb;
107	PGC_OPTIONS options;
108
109	ssize_t severe_pressure_per1000;
110	ssize_t aggressive_evict_per1000;
111	ssize_t healthy_size_per1000;
112	ssize_t evict_low_threshold_per1000;
113
114	dynamic_target_cache_size_callback dynamic_target_size_cb;
115	nominal_page_size_callback nominal_page_size_cb;
116	} config;
117
118	struct {
119	ND_THREAD *thread; // the thread
120	struct completion completion; // signal the thread to wake up
121	} evictor;
122
123	struct pgc_index {
124	RW_SPINLOCK rw_spinlock;
125	Pvoid_t sections_judy;
126	#ifdef PGC_WITH_ARAL
127	ARAL *aral;
128	#endif
129	} *index;
130
131	struct {
132	SPINLOCK spinlock;
133	ssize_t per1000;
134	} usage;
135
136	struct pgc_queue clean; // LRU is applied here to free memory from the cache
137	struct pgc_queue dirty; // in the dirty list, pages are ordered the way they were marked dirty
138	struct pgc_queue hot; // in the hot list, pages are order the way they were marked hot
139	struct pgc_statistics stats; // statistics
140
141	#ifdef NETDATA_PGC_POINTER_CHECK
142	netdata_mutex_t global_pointer_registry_mutex;
143	Pvoid_t global_pointer_registry;
144	#endif
145	};
146
147	// ----------------------------------------------------------------------------
148	// validate each pointer is indexed once - internal checks only
149
150	static inline void pointer_index_init(PGC *cache __maybe_unused) {
151	#ifdef NETDATA_PGC_POINTER_CHECK
152	netdata_mutex_init(&cache->global_pointer_registry_mutex);
153	#else
154	;
155	#endif
156	}
157
158	static inline void pointer_destroy_index(PGC *cache __maybe_unused) {
159	#ifdef NETDATA_PGC_POINTER_CHECK
160	netdata_mutex_lock(&cache->global_pointer_registry_mutex);
161	JudyHSFreeArray(&cache->global_pointer_registry, PJE0);
162	netdata_mutex_unlock(&cache->global_pointer_registry_mutex);
163	#else
164	;
165	#endif
166	}
167	static inline void pointer_add(PGC cache __maybe_unused, PGC_PAGE page __maybe_unused) {
168	#ifdef NETDATA_PGC_POINTER_CHECK
169	netdata_mutex_lock(&cache->global_pointer_registry_mutex);
170	Pvoid_t PValue = JudyHSIns(&cache->global_pointer_registry, &page, sizeof(void ), PJE0);
171	if(*PValue != NULL)
172	fatal("pointer already exists in registry");
173	*PValue = page;
174	netdata_mutex_unlock(&cache->global_pointer_registry_mutex);
175	#else
176	;
177	#endif
178	}
179
180	static inline void pointer_check(PGC cache __maybe_unused, PGC_PAGE page __maybe_unused) {
181	#ifdef NETDATA_PGC_POINTER_CHECK
182	netdata_mutex_lock(&cache->global_pointer_registry_mutex);
183	Pvoid_t PValue = JudyHSGet(cache->global_pointer_registry, &page, sizeof(void ));
184	if(PValue == NULL)
185	fatal("pointer is not found in registry");
186	netdata_mutex_unlock(&cache->global_pointer_registry_mutex);
187	#else
188	;
189	#endif
190	}
191
192	static inline void pointer_del(PGC cache __maybe_unused, PGC_PAGE page __maybe_unused) {
193	#ifdef NETDATA_PGC_POINTER_CHECK
194	netdata_mutex_lock(&cache->global_pointer_registry_mutex);
195	int ret = JudyHSDel(&cache->global_pointer_registry, &page, sizeof(void *), PJE0);
196	if(!ret)
197	fatal("pointer to be deleted does not exist in registry");
198	netdata_mutex_unlock(&cache->global_pointer_registry_mutex);
199	#else
200	;
201	#endif
202	}
203
204	// ----------------------------------------------------------------------------
205	// helpers
206
207	static inline size_t page_assumed_size(PGC *cache, size_t size) {
208	return size + (sizeof(PGC_PAGE) + cache->config.additional_bytes_per_page + sizeof(Word_t) * 3);
209	}
210
211	static inline size_t page_size_from_assumed_size(PGC *cache, size_t assumed_size) {
212	return assumed_size - (sizeof(PGC_PAGE) + cache->config.additional_bytes_per_page + sizeof(Word_t) * 3);
213	}
214
215	// ----------------------------------------------------------------------------
216	// locking
217
218	static inline size_t pgc_indexing_partition(PGC *cache, Word_t metric_id) {
219	static __thread PGC *last_cache = NULL;
220	static __thread size_t last_partitions = 0;
221	static __thread Word_t last_metric_id = 0;
222	static __thread size_t last_partition = 0;
223
224	if(cache == last_cache &&
225	cache->config.partitions == last_partitions &&
226	(metric_id == last_metric_id \|\| cache->config.partitions == 1))
227	return last_partition;
228
229	last_cache = cache;
230	last_partitions = cache->config.partitions;
231	last_metric_id = metric_id;
232	last_partition = indexing_partition(metric_id, cache->config.partitions);
233
234	return last_partition;
235	}
236
237	#define pgc_index_read_lock(cache, partition) rw_spinlock_read_lock(&(cache)->index[partition].rw_spinlock)
238	#define pgc_index_read_unlock(cache, partition) rw_spinlock_read_unlock(&(cache)->index[partition].rw_spinlock)
239	#define pgc_index_write_lock(cache, partition) rw_spinlock_write_lock(&(cache)->index[partition].rw_spinlock)
240	#define pgc_index_write_unlock(cache, partition) rw_spinlock_write_unlock(&(cache)->index[partition].rw_spinlock)
241	#define pgc_index_trywrite_lock(cache, partition, force) ({ \
242	bool _result; \
243	if (force) { \
244	rw_spinlock_write_lock(&(cache)->index[partition].rw_spinlock); \
245	_result = true; \
246	} else \
247	_result = rw_spinlock_trywrite_lock(&(cache)->index[partition].rw_spinlock); \
248	_result; \
249	})
250
251	#define PGC_QUEUE_LOCK_PRIO_COLLECTORS WAITQ_PRIO_URGENT
252	#define PGC_QUEUE_LOCK_PRIO_EVICTORS WAITQ_PRIO_HIGH
253	#define PGC_QUEUE_LOCK_PRIO_FLUSHERS WAITQ_PRIO_NORMAL
254	#define PGC_QUEUE_LOCK_PRIO_LOW WAITQ_PRIO_LOW
255
256	#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
257	#define pgc_queue_trylock(cache, ll, prio) waitq_try_acquire(&((ll)->wq), prio)
258	#define pgc_queue_lock(cache, ll, prio) waitq_acquire(&((ll)->wq), prio)
259	#define pgc_queue_unlock(cache, ll) waitq_release(&((ll)->wq))
260	#else
261	#define pgc_queue_trylock(cache, ll, prio) spinlock_trylock(&((ll)->spinlock))
262	#define pgc_queue_lock(cache, ll, prio) spinlock_lock(&((ll)->spinlock))
263	#define pgc_queue_unlock(cache, ll) spinlock_unlock(&((ll)->spinlock))
264	#endif
265
266	#define page_transition_trylock(cache, page) spinlock_trylock(&(page)->transition_spinlock)
267	#define page_transition_lock(cache, page) spinlock_lock(&(page)->transition_spinlock)
268	#define page_transition_unlock(cache, page) spinlock_unlock(&(page)->transition_spinlock)
269
270	// ----------------------------------------------------------------------------
271	// size histogram
272
273	static void pgc_size_histogram_init(struct pgc_size_histogram *h) {
274	// the histogram needs to be all-inclusive for the possible sizes
275	// so, we start from 0, and the last value is SIZE_MAX.
276
277	size_t values[PGC_SIZE_HISTOGRAM_ENTRIES] = {
278	0, 32, 64, 128, 256, 512, 1024, 2048,
279	4096, 8192, 16384, 32768, 65536, 128 * 1024, SIZE_MAX
280	};
281
282	size_t last_value = 0;
283	for(size_t i = 0; i < PGC_SIZE_HISTOGRAM_ENTRIES; i++) {
284	if(i > 0 && values[i] == 0)
285	fatal("only the first value in the array can be zero");
286
287	if(i > 0 && values[i] <= last_value)
288	fatal("the values need to be sorted");
289
290	h->array[i].upto = values[i];
291	last_value = values[i];
292	}
293	}
294
295	static inline size_t pgc_size_histogram_slot(struct pgc_size_histogram *h, size_t size) {
296	if(size <= h->array[0].upto)
297	return 0;
298
299	if(size >= h->array[_countof(h->array) - 1].upto)
300	return _countof(h->array) - 1;
301
302	// binary search for the right size
303	size_t low = 0, high = _countof(h->array) - 1;
304	while (low < high) {
305	size_t mid = low + (high - low) / 2;
306	if (size < h->array[mid].upto)
307	high = mid;
308	else
309	low = mid + 1;
310	}
311	return low - 1;
312	}
313
314	static inline void pgc_size_histogram_add(PGC cache, struct pgc_size_histogram h, PGC_PAGE *page) {
315	size_t size;
316	if(cache->config.nominal_page_size_cb)
317	size = cache->config.nominal_page_size_cb(page->data);
318	else
319	size = page_size_from_assumed_size(cache, page->assumed_size);
320
321	size_t slot = pgc_size_histogram_slot(h, size);
322	internal_fatal(slot >= _countof(h->array), "hey!");
323
324	__atomic_add_fetch(&h->array[slot].count, 1, __ATOMIC_RELAXED);
325	}
326
327	static inline void pgc_size_histogram_del(PGC cache, struct pgc_size_histogram h, PGC_PAGE *page) {
328	size_t size;
329	if(cache->config.nominal_page_size_cb)
330	size = cache->config.nominal_page_size_cb(page->data);
331	else
332	size = page_size_from_assumed_size(cache, page->assumed_size);
333
334	size_t slot = pgc_size_histogram_slot(h, size);
335	internal_fatal(slot >= _countof(h->array), "hey!");
336
337	__atomic_sub_fetch(&h->array[slot].count, 1, __ATOMIC_RELAXED);
338	}
339
340	// ----------------------------------------------------------------------------
341	// evictions control
342
343	ALWAYS_INLINE
344	static int64_t pgc_threshold(ssize_t threshold, int64_t wanted, int64_t current, int64_t clean) {
345	if(current < clean)
346	current = clean;
347
348	if(wanted < current - clean)
349	wanted = current - clean;
350
351	int64_t ret = wanted * threshold / 1000LL;
352	if(ret < current - clean)
353	ret = current - clean;
354
355	return ret;
356	}
357
358	ALWAYS_INLINE
359	static int64_t pgc_wanted_size(const int64_t hot, const int64_t hot_max, const int64_t dirty_max, const int64_t index) {
360	// our promise to users
361	const int64_t max_size1 = MAX(hot_max, hot) * 2;
362
363	// protection against slow flushing
364	const int64_t max_size2 = hot_max + MAX(dirty_max * 2, hot_max * 2 / 3) + index;
365
366	// the final wanted cache size
367	return MIN(max_size1, max_size2);
368	}
369
370	static ssize_t cache_usage_per1000(PGC cache, int64_t size_to_evict) {
371
372	if(size_to_evict)
373	spinlock_lock(&cache->usage.spinlock);
374
375	else if(!spinlock_trylock(&cache->usage.spinlock))
376	return __atomic_load_n(&cache->usage.per1000, __ATOMIC_RELAXED);
377
378	int64_t wanted_cache_size;
379
380	const int64_t dirty = __atomic_load_n(&cache->dirty.stats->size, __ATOMIC_RELAXED);
381	const int64_t hot = __atomic_load_n(&cache->hot.stats->size, __ATOMIC_RELAXED);
382	const int64_t clean = __atomic_load_n(&cache->clean.stats->size, __ATOMIC_RELAXED);
383	const int64_t evicting = __atomic_load_n(&cache->stats.evicting_size, __ATOMIC_RELAXED);
384	const int64_t flushing = __atomic_load_n(&cache->stats.flushing_size, __ATOMIC_RELAXED);
385	const int64_t current_cache_size = __atomic_load_n(&cache->stats.size, __ATOMIC_RELAXED);
386	const int64_t all_pages_size = hot + dirty + clean + evicting + flushing;
387	const int64_t index = current_cache_size > all_pages_size ? current_cache_size - all_pages_size : 0;
388	const int64_t referenced_size = __atomic_load_n(&cache->stats.referenced_size, __ATOMIC_RELAXED);
389
390	if(cache->config.options & PGC_OPTIONS_AUTOSCALE) {
391	const int64_t dirty_max = __atomic_load_n(&cache->dirty.stats->max_size, __ATOMIC_RELAXED);
392	const int64_t hot_max = __atomic_load_n(&cache->hot.stats->max_size, __ATOMIC_RELAXED);
393
394	if(cache->config.dynamic_target_size_cb) {
395	wanted_cache_size = pgc_wanted_size(hot, hot, dirty, index);
396
397	const int64_t wanted_cache_size_cb = cache->config.dynamic_target_size_cb();
398	if(wanted_cache_size_cb > wanted_cache_size)
399	wanted_cache_size = wanted_cache_size_cb;
400	}
401	else
402	wanted_cache_size = pgc_wanted_size(hot, hot_max, dirty_max, index);
403
404	if (wanted_cache_size < hot + dirty + index + cache->config.clean_size)
405	wanted_cache_size = hot + dirty + index + cache->config.clean_size;
406	}
407	else
408	wanted_cache_size = hot + dirty + index + cache->config.clean_size;
409
410	// calculate the absolute minimum we can go
411	const int64_t min_cache_size1 = (referenced_size > hot ? referenced_size : hot) + dirty + index;
412	const int64_t min_cache_size2 = (current_cache_size > clean) ? current_cache_size - clean : min_cache_size1;
413	const int64_t min_cache_size = MAX(min_cache_size1, min_cache_size2);
414
415	if(cache->config.out_of_memory_protection_bytes) {
416	// out of memory protection
417	OS_SYSTEM_MEMORY sm = os_system_memory(false);
418	if(OS_SYSTEM_MEMORY_OK(sm)) {
419	// when the total exists, ram_available_bytes is also right
420
421	const int64_t ram_available_bytes = (int64_t)sm.ram_available_bytes;
422
423	const int64_t min_available = cache->config.out_of_memory_protection_bytes;
424	if (ram_available_bytes < min_available) {
425	// we must shrink
426	int64_t must_lose = min_available - ram_available_bytes;
427
428	if(current_cache_size > must_lose)
429	wanted_cache_size = current_cache_size - must_lose;
430	else
431	wanted_cache_size = min_cache_size;
432	}
433	else if(cache->config.use_all_ram) {
434	// we can grow
435	wanted_cache_size = current_cache_size + (ram_available_bytes - min_available);
436	}
437	}
438	}
439
440	// never go below our minimum
441	if(unlikely(wanted_cache_size < min_cache_size))
442	wanted_cache_size = min_cache_size;
443
444	// protection for the case the cache is totally empty
445	if(unlikely(wanted_cache_size < 65536))
446	wanted_cache_size = 65536;
447
448	const ssize_t per1000 = (ssize_t)(current_cache_size * 1000LL / wanted_cache_size);
449	__atomic_store_n(&cache->usage.per1000, per1000, __ATOMIC_RELAXED);
450	__atomic_store_n(&cache->stats.wanted_cache_size, wanted_cache_size, __ATOMIC_RELAXED);
451	__atomic_store_n(&cache->stats.current_cache_size, current_cache_size, __ATOMIC_RELAXED);
452
453	int64_t healthy_target = pgc_threshold(cache->config.healthy_size_per1000, wanted_cache_size, current_cache_size, clean);
454	if(current_cache_size > healthy_target) {
455	int64_t low_watermark_target = pgc_threshold(cache->config.evict_low_threshold_per1000, wanted_cache_size, current_cache_size, clean);
456
457	int64_t size_to_evict_now = current_cache_size - low_watermark_target;
458	if(size_to_evict_now > clean)
459	size_to_evict_now = clean;
460
461	if(size_to_evict)
462	*size_to_evict = size_to_evict_now;
463
464	bool signal = false;
465	if(per1000 >= cache->config.severe_pressure_per1000) {
466	__atomic_add_fetch(&cache->stats.events_cache_under_severe_pressure, 1, __ATOMIC_RELAXED);
467	signal = true;
468	}
469	else if(per1000 >= cache->config.aggressive_evict_per1000) {
470	__atomic_add_fetch(&cache->stats.events_cache_needs_space_aggressively, 1, __ATOMIC_RELAXED);
471	signal = true;
472	}
473
474	if(signal) {
475	completion_mark_complete_a_job(&cache->evictor.completion);
476	p2_add_fetch(&cache->stats.p2_waste_evict_thread_signals, 1);
477	}
478	}
479
480	spinlock_unlock(&cache->usage.spinlock);
481
482	return per1000;
483	}
484
485	static inline bool cache_pressure(PGC *cache, ssize_t limit) {
486	return (cache_usage_per1000(cache, NULL) >= limit);
487	}
488
489	#define cache_under_severe_pressure(cache) cache_pressure(cache, (cache)->config.severe_pressure_per1000)
490	#define cache_needs_space_aggressively(cache) cache_pressure(cache, (cache)->config.aggressive_evict_per1000)
491	#define cache_above_healthy_limit(cache) cache_pressure(cache, (cache)->config.healthy_size_per1000)
492
493	typedef bool (evict_filter)(PGC_PAGE page, void *data);
494	static bool evict_pages_with_filter(PGC cache, size_t max_skip, size_t max_evict, bool wait, bool all_of_them, evict_filter filter, void data);
495	#define evict_pages(cache, max_skip, max_evict, wait, all_of_them) evict_pages_with_filter(cache, max_skip, max_evict, wait, all_of_them, NULL, NULL)
496
497	static inline bool flushing_critical(PGC *cache);
498	static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wait, bool all_of_them);
499
500	static ALWAYS_INLINE void evict_pages_inline(PGC *cache, bool on_release) {
501	const ssize_t per1000 = cache_usage_per1000(cache, NULL);
502
503	if(!(cache->config.options & PGC_OPTIONS_EVICT_PAGES_NO_INLINE)) {
504	if (per1000 > cache->config.aggressive_evict_per1000 && !on_release) {
505	// the threads that add pages, turn into evictors when the cache needs evictions aggressively
506	p2_add_fetch(&cache->stats.p2_waste_evictions_inline_on_add, 1);
507	evict_pages(cache,
508	cache->config.max_skip_pages_per_inline_eviction,
509	cache->config.max_pages_per_inline_eviction,
510	false, false);
511	}
512	else if (per1000 > cache->config.severe_pressure_per1000 && on_release) {
513	// the threads that are releasing pages, turn into evictors when the cache is critical
514	p2_add_fetch(&cache->stats.p2_waste_evictions_inline_on_release, 1);
515
516	evict_pages(cache,
517	cache->config.max_skip_pages_per_inline_eviction,
518	cache->config.max_pages_per_inline_eviction,
519	false, false);
520	}
521	}
522	}
523
524	static ALWAYS_INLINE void evict_on_clean_page_added(PGC *cache) {
525	evict_pages_inline(cache, false);
526	}
527
528	static ALWAYS_INLINE void evict_on_page_release_when_permitted(PGC *cache) {
529	evict_pages_inline(cache, true);
530	}
531
532	static ALWAYS_INLINE void flush_inline(PGC *cache, bool on_release) {
533	if(!(cache->config.options & PGC_OPTIONS_FLUSH_PAGES_NO_INLINE) && flushing_critical(cache)) {
534	if (on_release)
535	p2_add_fetch(&cache->stats.p2_waste_flush_on_release, 1);
536	else
537	p2_add_fetch(&cache->stats.p2_waste_flush_on_add, 1);
538
539	flush_pages(cache, cache->config.max_flushes_inline, PGC_SECTION_ALL, false, false);
540	}
541	}
542
543	static ALWAYS_INLINE void flush_on_page_add(PGC *cache) {
544	flush_inline(cache, false);
545	}
546
547	static ALWAYS_INLINE void flush_on_page_hot_release(PGC *cache) {
548	flush_inline(cache, true);
549	}
550
551
552	// ----------------------------------------------------------------------------
553	// flushing control
554
555	static inline bool flushing_critical(PGC *cache) {
556	if(unlikely(__atomic_load_n(&cache->dirty.stats->size, __ATOMIC_RELAXED) > __atomic_load_n(&cache->hot.stats->max_size, __ATOMIC_RELAXED))) {
557	__atomic_add_fetch(&cache->stats.events_flush_critical, 1, __ATOMIC_RELAXED);
558	return true;
559	}
560
561	return false;
562	}
563
564	// ----------------------------------------------------------------------------
565	// Linked list management
566
567	static inline void atomic_set_max_size_t(size_t *max, size_t desired) {
568	size_t expected;
569
570	expected = __atomic_load_n(max, __ATOMIC_RELAXED);
571
572	do {
573
574	if(expected >= desired)
575	return;
576
577	} while(!__atomic_compare_exchange_n(max, &expected, desired,
578	false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
579	}
580
581	static inline void atomic_set_max_int64_t(int64_t *max, int64_t desired) {
582	int64_t expected;
583
584	expected = __atomic_load_n(max, __ATOMIC_RELAXED);
585
586	do {
587
588	if(expected >= desired)
589	return;
590
591	} while(!__atomic_compare_exchange_n(max, &expected, desired,
592	false, __ATOMIC_RELAXED, __ATOMIC_RELAXED));
593	}
594
595	struct section_pages {
596	SPINLOCK migration_to_v2_spinlock;
597	size_t entries;
598	size_t size;
599	PGC_PAGE *base;
600	};
601
602	static struct aral_statistics pgc_aral_statistics = { 0 };
603
604	static ARAL *pgc_sections_aral = NULL;
605
606	static void pgc_section_pages_static_aral_init(void) {
607	static SPINLOCK spinlock = SPINLOCK_INITIALIZER;
608
609	spinlock_lock(&spinlock);
610
611	if(!pgc_sections_aral) {
612	pgc_sections_aral = aral_create(
613	"pgc-sections", sizeof(struct section_pages), 0, 0, &pgc_aral_statistics,
614	NULL, NULL, false, false, false);
615
616	pulse_aral_register_statistics(&pgc_aral_statistics, "pgc");
617	}
618
619	spinlock_unlock(&spinlock);
620	}
621
622	static ALWAYS_INLINE void pgc_stats_queue_judy_change(PGC cache, struct pgc_queue ll, int64_t delta) {
623	__atomic_add_fetch(&ll->stats->size, delta, __ATOMIC_RELAXED);
624	__atomic_add_fetch(&cache->stats.size, delta, __ATOMIC_RELAXED);
625	}
626
627	static ALWAYS_INLINE void pgc_stats_index_judy_change(PGC *cache, int64_t delta) {
628	__atomic_add_fetch(&cache->stats.size, delta, __ATOMIC_RELAXED);
629	}
630
631	static ALWAYS_INLINE void pgc_queue_add(PGC cache __maybe_unused, struct pgc_queue q, PGC_PAGE *page, bool having_lock, WAITQ_PRIORITY prio __maybe_unused) {
632	if(!having_lock)
633	pgc_queue_lock(cache, q, prio);
634
635	internal_fatal(page_get_status_flags(page) != 0,
636	"DBENGINE CACHE: invalid page flags, the page has %d, but it is should be %d",
637	page_get_status_flags(page),
638	0);
639
640	if(q->linked_list_in_sections_judy) {
641	// HOT and DIRTY pages end up here.
642
643	JudyAllocThreadPulseReset();
644	int64_t mem_delta = 0;
645
646	Pvoid_t *section_pages_pptr = JudyLIns(&q->sections_judy, page->section, PJE0);
647	if(section_pages_pptr == NULL \|\| section_pages_pptr == PJERR)
648	fatal("DBENGINE CACHE: JudyLIns(q->sections_judy, 0x%lx) failed, q->sections_judy = %p, result = %p",
649	(long unsigned)page->section, q->sections_judy, section_pages_pptr);
650
651	struct section_pages sp = section_pages_pptr;
652	if(!sp) {
653	// sp = callocz(1, sizeof(struct section_pages));
654	sp = aral_mallocz(pgc_sections_aral);
655	memset(sp, 0, sizeof(struct section_pages));
656
657	*section_pages_pptr = sp;
658
659	mem_delta += sizeof(struct section_pages);
660	}
661
662	mem_delta += JudyAllocThreadPulseGetAndReset();
663	pgc_stats_queue_judy_change(cache, q, mem_delta);
664
665	sp->entries++;
666	sp->size += page->assumed_size;
667	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(sp->base, page, link.prev, link.next);
668
669	if((sp->entries % cache->config.max_dirty_pages_per_call) == 0)
670	q->version++;
671	}
672	else {
673	// CLEAN pages end up here.
674	// - New pages created as CLEAN, always have 1 access.
675	// - DIRTY pages made CLEAN, depending on their accesses may be appended (accesses > 0) or prepended (accesses = 0).
676
677	if(page->accesses \|\| page_flag_check(page, PGC_PAGE_HAS_BEEN_ACCESSED \| PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES) == PGC_PAGE_HAS_BEEN_ACCESSED) {
678	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(q->base, page, link.prev, link.next);
679	page_flag_clear(page, PGC_PAGE_HAS_BEEN_ACCESSED);
680	}
681	else
682	DOUBLE_LINKED_LIST_PREPEND_ITEM_UNSAFE(q->base, page, link.prev, link.next);
683
684	q->version++;
685	}
686
687	page_flag_set(page, q->flags);
688
689	if(!having_lock)
690	pgc_queue_unlock(cache, q);
691
692	size_t entries = __atomic_add_fetch(&q->stats->entries, 1, __ATOMIC_RELAXED);
693	int64_t size = __atomic_add_fetch(&q->stats->size, page->assumed_size, __ATOMIC_RELAXED);
694	__atomic_add_fetch(&q->stats->added_entries, 1, __ATOMIC_RELAXED);
695	__atomic_add_fetch(&q->stats->added_size, page->assumed_size, __ATOMIC_RELAXED);
696
697	atomic_set_max_size_t(&q->stats->max_entries, entries);
698	atomic_set_max_int64_t(&q->stats->max_size, size);
699
700	if(cache->config.stats)
701	pgc_size_histogram_add(cache, &q->stats->size_histogram, page);
702	}
703
704	static ALWAYS_INLINE void pgc_queue_del(PGC cache __maybe_unused, struct pgc_queue q, PGC_PAGE *page, bool having_lock,
705	WAITQ_PRIORITY prio __maybe_unused) {
706	if(cache->config.stats)
707	pgc_size_histogram_del(cache, &q->stats->size_histogram, page);
708
709	__atomic_sub_fetch(&q->stats->entries, 1, __ATOMIC_RELAXED);
710	__atomic_sub_fetch(&q->stats->size, page->assumed_size, __ATOMIC_RELAXED);
711	__atomic_add_fetch(&q->stats->removed_entries, 1, __ATOMIC_RELAXED);
712	__atomic_add_fetch(&q->stats->removed_size, page->assumed_size, __ATOMIC_RELAXED);
713
714	if(!having_lock)
715	pgc_queue_lock(cache, q, prio);
716
717	internal_fatal(page_get_status_flags(page) != q->flags,
718	"DBENGINE CACHE: invalid page flags, the page has %d, but it is should be %d",
719	page_get_status_flags(page),
720	q->flags);
721
722	struct section_pages *sp_to_free = NULL;
723
724	if(q->linked_list_in_sections_judy) {
725	Pvoid_t *section_pages_pptr = JudyLGet(q->sections_judy, page->section, PJE0);
726	if(section_pages_pptr == NULL \|\| section_pages_pptr == PJERR)
727	fatal("DBENGINE CACHE: JudyLGet(q->sections_judy, 0x%lx) failed, q->sections_judy = %p",
728	(long unsigned)page->section, q->sections_judy);
729
730	struct section_pages sp = section_pages_pptr;
731	sp->entries--;
732	sp->size -= page->assumed_size;
733	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(sp->base, page, link.prev, link.next);
734
735	if(!sp->base) {
736	JudyAllocThreadPulseReset();
737	int64_t mem_delta = 0;
738
739	int rc = JudyLDel(&q->sections_judy, page->section, PJE0);
740
741	if(!rc)
742	fatal("DBENGINE CACHE: cannot delete section from Judy LL");
743
744	sp_to_free = sp;
745
746	mem_delta -= sizeof(struct section_pages);
747	mem_delta += JudyAllocThreadPulseGetAndReset();
748
749	pgc_stats_queue_judy_change(cache, q, mem_delta);
750	}
751	}
752	else {
753	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(q->base, page, link.prev, link.next);
754	q->version++;
755	}
756
757	// Clear the queue flag only after the unlink, while still under the queue
758	// lock. This guarantees that "flag cleared" implies "page already unlinked",
759	// so a lockless reader observing "no longer on this queue" cannot race
760	// with an in-progress unlink. The reciprocal (flag set implies on the
761	// list) is intentionally not provided; readers that act on link pointers
762	// must re-validate under the queue lock, as commit 2733e6fc60 (#21793)
763	// does in page_has_been_accessed.
764	page_flag_clear(page, q->flags);
765
766	if(!having_lock)
767	pgc_queue_unlock(cache, q);
768
769	if (sp_to_free)
770	aral_freez(pgc_sections_aral, sp_to_free);
771	}
772
773	static ALWAYS_INLINE void page_has_been_accessed(PGC cache, PGC_PAGE page) {
774	PGC_PAGE_FLAGS flags = page_flag_check(page, PGC_PAGE_CLEAN \| PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES);
775
776	if (!(flags & PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES)) {
777	__atomic_add_fetch(&page->accesses, 1, __ATOMIC_RELAXED);
778
779	if (flags & PGC_PAGE_CLEAN) {
780	if(pgc_queue_trylock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS)) {
781	// The status check above is lockless. Re-validate under the clean lock to avoid
782	// touching clean-list pointers after the page moved to another queue.
783	if(is_page_clean(page)) {
784	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
785	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
786	page_flag_clear(page, PGC_PAGE_HAS_BEEN_ACCESSED);
787	}
788	else {
789	// Expected concurrent transition: page may move clean -> dirty/hot
790	// between lockless flag read and acquiring the clean queue lock.
791	page_flag_set(page, PGC_PAGE_HAS_BEEN_ACCESSED);
792	}
793
794	pgc_queue_unlock(cache, &cache->clean);
795	}
796	else
797	page_flag_set(page, PGC_PAGE_HAS_BEEN_ACCESSED);
798	}
799	}
800	}
801
802
803	// ----------------------------------------------------------------------------
804	// state transitions
805
806	static ALWAYS_INLINE void page_set_clean(PGC cache, PGC_PAGE page, bool having_transition_lock, bool having_clean_lock, WAITQ_PRIORITY prio) {
807	if(!having_transition_lock)
808	page_transition_lock(cache, page);
809
810	PGC_PAGE_FLAGS flags = page_get_status_flags(page);
811
812	if(flags & PGC_PAGE_CLEAN) {
813	if(!having_transition_lock)
814	page_transition_unlock(cache, page);
815	return;
816	}
817
818	if(flags & PGC_PAGE_HOT)
819	pgc_queue_del(cache, &cache->hot, page, false, prio);
820
821	if(flags & PGC_PAGE_DIRTY)
822	pgc_queue_del(cache, &cache->dirty, page, false, prio);
823
824	// first add to linked list, the set the flag (required for move_page_last())
825	pgc_queue_add(cache, &cache->clean, page, having_clean_lock, prio);
826
827	if(!having_transition_lock)
828	page_transition_unlock(cache, page);
829	}
830
831	static ALWAYS_INLINE void page_set_dirty(PGC cache, PGC_PAGE page, bool having_hot_lock, WAITQ_PRIORITY prio) {
832	if(!having_hot_lock)
833	// to avoid deadlocks, we have to get the hot lock before the page transition
834	// since this is what all_hot_to_dirty() does
835	pgc_queue_lock(cache, &cache->hot, prio);
836
837	page_transition_lock(cache, page);
838
839	PGC_PAGE_FLAGS flags = page_get_status_flags(page);
840
841	if(flags & PGC_PAGE_DIRTY) {
842	page_transition_unlock(cache, page);
843
844	if(!having_hot_lock)
845	// we don't need the hot lock anymore
846	pgc_queue_unlock(cache, &cache->hot);
847
848	return;
849	}
850
851	__atomic_add_fetch(&cache->stats.hot2dirty_entries, 1, __ATOMIC_RELAXED);
852	__atomic_add_fetch(&cache->stats.hot2dirty_size, page->assumed_size, __ATOMIC_RELAXED);
853
854	if(likely(flags & PGC_PAGE_HOT))
855	pgc_queue_del(cache, &cache->hot, page, true, prio);
856
857	if(!having_hot_lock)
858	// we don't need the hot lock anymore
859	pgc_queue_unlock(cache, &cache->hot);
860
861	if(unlikely(flags & PGC_PAGE_CLEAN))
862	pgc_queue_del(cache, &cache->clean, page, false, prio);
863
864	// first add to linked list, the set the flag (required for move_page_last())
865	pgc_queue_add(cache, &cache->dirty, page, false, prio);
866
867	__atomic_sub_fetch(&cache->stats.hot2dirty_entries, 1, __ATOMIC_RELAXED);
868	__atomic_sub_fetch(&cache->stats.hot2dirty_size, page->assumed_size, __ATOMIC_RELAXED);
869
870	page_transition_unlock(cache, page);
871	}
872
873	static ALWAYS_INLINE void page_set_hot(PGC cache, PGC_PAGE page, WAITQ_PRIORITY prio) {
874	page_transition_lock(cache, page);
875
876	PGC_PAGE_FLAGS flags = page_get_status_flags(page);
877
878	if(flags & PGC_PAGE_HOT) {
879	page_transition_unlock(cache, page);
880	return;
881	}
882
883	if(flags & PGC_PAGE_DIRTY)
884	pgc_queue_del(cache, &cache->dirty, page, false, prio);
885
886	if(flags & PGC_PAGE_CLEAN)
887	pgc_queue_del(cache, &cache->clean, page, false, prio);
888
889	// first add to linked list, the set the flag (required for move_page_last())
890	pgc_queue_add(cache, &cache->hot, page, false, prio);
891
892	page_transition_unlock(cache, page);
893	}
894
895
896	// ----------------------------------------------------------------------------
897	// Referencing
898
899	static ALWAYS_INLINE size_t PGC_REFERENCED_PAGES(PGC *cache) {
900	return __atomic_load_n(&cache->stats.referenced_entries, __ATOMIC_RELAXED);
901	}
902
903	static ALWAYS_INLINE void PGC_REFERENCED_PAGES_PLUS1(PGC cache, PGC_PAGE page) {
904	__atomic_add_fetch(&cache->stats.referenced_entries, 1, __ATOMIC_RELAXED);
905	__atomic_add_fetch(&cache->stats.referenced_size, page->assumed_size, __ATOMIC_RELAXED);
906	}
907
908	static ALWAYS_INLINE void PGC_REFERENCED_PAGES_MINUS1(PGC *cache, int64_t assumed_size) {
909	__atomic_sub_fetch(&cache->stats.referenced_entries, 1, __ATOMIC_RELAXED);
910	__atomic_sub_fetch(&cache->stats.referenced_size, assumed_size, __ATOMIC_RELAXED);
911	}
912
913	// If the page is not already acquired,
914	// YOU HAVE TO HAVE THE QUEUE (hot, dirty, clean - the page is in), LOCKED!
915	// If you don't have it locked, NOTHING PREVENTS THIS PAGE FROM VANISHING WHILE THIS IS CALLED!
916	static ALWAYS_INLINE bool page_acquire(PGC cache, PGC_PAGE page) {
917	__atomic_add_fetch(&cache->stats.acquires, 1, __ATOMIC_RELAXED);
918
919	REFCOUNT rc = refcount_acquire_advanced(&page->refcount);
920	if(REFCOUNT_ACQUIRED(rc)) {
921	if(rc == 1)
922	PGC_REFERENCED_PAGES_PLUS1(cache, page);
923
924	return true;
925	}
926
927	return false;
928	}
929
930	static ALWAYS_INLINE void page_release(PGC cache, PGC_PAGE page, bool evict_if_necessary) {
931	__atomic_add_fetch(&cache->stats.releases, 1, __ATOMIC_RELAXED);
932
933	int64_t assumed_size = page->assumed_size; // take the size before we release it
934
935	if(refcount_release(&page->refcount) == 0) {
936	PGC_REFERENCED_PAGES_MINUS1(cache, assumed_size);
937
938	if(evict_if_necessary)
939	evict_on_page_release_when_permitted(cache);
940	}
941	}
942
943	static ALWAYS_INLINE bool non_acquired_page_get_for_deletion___while_having_clean_locked(PGC cache __maybe_unused, PGC_PAGE page) {
944	__atomic_add_fetch(&cache->stats.acquires_for_deletion, 1, __ATOMIC_RELAXED);
945
946	internal_fatal(!is_page_clean(page),
947	"DBENGINE CACHE: only clean pages can be deleted");
948
949	if(refcount_acquire_for_deletion(&page->refcount)) {
950	// we can delete this page
951	internal_fatal(page_flag_check(page, PGC_PAGE_IS_BEING_DELETED),
952	"DBENGINE CACHE: page is already being deleted");
953
954	page_flag_set(page, PGC_PAGE_IS_BEING_DELETED);
955
956	return true;
957	}
958
959	return false;
960	}
961
962	static ALWAYS_INLINE bool acquired_page_get_for_deletion_or_release_it(PGC cache __maybe_unused, PGC_PAGE page) {
963	__atomic_add_fetch(&cache->stats.acquires_for_deletion, 1, __ATOMIC_RELAXED);
964
965	int64_t assumed_size = page->assumed_size; // take the size before we release it
966
967	if(refcount_release_and_acquire_for_deletion(&page->refcount)) {
968	PGC_REFERENCED_PAGES_MINUS1(cache, assumed_size);
969
970	// we can delete this page
971	internal_fatal(page_flag_check(page, PGC_PAGE_IS_BEING_DELETED),
972	"DBENGINE CACHE: page is already being deleted");
973
974	page_flag_set(page, PGC_PAGE_IS_BEING_DELETED);
975
976	return true;
977	}
978
979	return false;
980	}
981
982
983	// ----------------------------------------------------------------------------
984	// Indexing
985
986	static inline void free_this_page(PGC cache, PGC_PAGE page, size_t partition __maybe_unused) {
987	size_t size = page_size_from_assumed_size(cache, page->assumed_size);
988
989	// call the callback to free the user supplied memory
990	cache->config.pgc_free_clean_cb(cache, (PGC_ENTRY){
991	.section = page->section,
992	.metric_id = page->metric_id,
993	.start_time_s = page->start_time_s,
994	.end_time_s = __atomic_load_n(&page->end_time_s, __ATOMIC_RELAXED),
995	.update_every_s = page->update_every_s,
996	.size = size,
997	.hot = (is_page_hot(page)) ? true : false,
998	.data = page->data,
999	.custom_data = (cache->config.additional_bytes_per_page) ? page->custom_data : NULL,
1000	});
1001
1002	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_FREE_CB);
1003
1004	// update statistics
1005	__atomic_add_fetch(&cache->stats.removed_entries, 1, __ATOMIC_RELAXED);
1006	__atomic_add_fetch(&cache->stats.removed_size, page->assumed_size, __ATOMIC_RELAXED);
1007
1008	__atomic_sub_fetch(&cache->stats.entries, 1, __ATOMIC_RELAXED);
1009	__atomic_sub_fetch(&cache->stats.size, page->assumed_size, __ATOMIC_RELAXED);
1010
1011	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_FREE_ATOMICS2);
1012
1013	// free our memory
1014	#ifdef PGC_WITH_ARAL
1015	aral_freez(cache->index[partition].aral, page);
1016	#else
1017	freez(page);
1018	#endif
1019
1020	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_FREE_ARAL);
1021	}
1022
1023	static void remove_this_page_from_index_unsafe(PGC cache, PGC_PAGE page, size_t partition) {
1024	// remove it from the Judy arrays
1025
1026	pointer_check(cache, page);
1027
1028	internal_fatal(page_flag_check(page, PGC_PAGE_HOT \| PGC_PAGE_DIRTY \| PGC_PAGE_CLEAN),
1029	"DBENGINE CACHE: page to be removed from the cache is still in the linked-list");
1030
1031	internal_fatal(!page_flag_check(page, PGC_PAGE_IS_BEING_DELETED),
1032	"DBENGINE CACHE: page to be removed from the index, is not marked for deletion");
1033
1034	internal_fatal(partition != pgc_indexing_partition(cache, page->metric_id),
1035	"DBENGINE CACHE: attempted to remove this page from the wrong partition of the cache");
1036
1037	Pvoid_t *metrics_judy_pptr = JudyLGet(cache->index[partition].sections_judy, page->section, PJE0);
1038	if(unlikely(!metrics_judy_pptr))
1039	fatal("DBENGINE CACHE: section '%p' should exist, but it does not.", (void *)page->section);
1040
1041	Pvoid_t pages_judy_pptr = JudyLGet(metrics_judy_pptr, page->metric_id, PJE0);
1042	if(unlikely(!pages_judy_pptr))
1043	fatal("DBENGINE CACHE: metric '%p' in section '%p' should exist, but it does not.",
1044	(void )page->metric_id, (void )page->section);
1045
1046	Pvoid_t page_ptr = JudyLGet(pages_judy_pptr, page->start_time_s, PJE0);
1047	if(unlikely(!page_ptr))
1048	fatal("DBENGINE CACHE: page with start time '%ld' of metric '%p' in section '%p' should exist, but it does not.",
1049	page->start_time_s, (void )page->metric_id, (void )page->section);
1050
1051	PGC_PAGE found_page = page_ptr;
1052	if(unlikely(found_page != page))
1053	fatal("DBENGINE CACHE: page with start time '%ld' of metric '%p' in section '%p' should exist, "
1054	"but the index returned a different address (expected %p, got %p).",
1055	page->start_time_s, (void )page->metric_id, (void )page->section,
1056	page, found_page);
1057
1058	JudyAllocThreadPulseReset();
1059
1060	if(unlikely(!JudyLDel(pages_judy_pptr, page->start_time_s, PJE0)))
1061	fatal("DBENGINE CACHE: page with start time '%ld' of metric '%p' in section '%p' exists, but cannot be deleted.",
1062	page->start_time_s, (void )page->metric_id, (void )page->section);
1063
1064	if(!*pages_judy_pptr && !JudyLDel(metrics_judy_pptr, page->metric_id, PJE0))
1065	fatal("DBENGINE CACHE: metric '%p' in section '%p' exists and is empty, but cannot be deleted.",
1066	(void )page->metric_id, (void )page->section);
1067
1068	if(!*metrics_judy_pptr && !JudyLDel(&cache->index[partition].sections_judy, page->section, PJE0))
1069	fatal("DBENGINE CACHE: section '%p' exists and is empty, but cannot be deleted.", (void *)page->section);
1070
1071	pgc_stats_index_judy_change(cache, JudyAllocThreadPulseGetAndReset());
1072
1073	pointer_del(cache, page);
1074	}
1075
1076	static inline void remove_and_free_page_not_in_any_queue_and_acquired_for_deletion(PGC cache, PGC_PAGE page) {
1077	size_t partition = pgc_indexing_partition(cache, page->metric_id);
1078	pgc_index_write_lock(cache, partition);
1079	remove_this_page_from_index_unsafe(cache, page, partition);
1080	pgc_index_write_unlock(cache, partition);
1081	free_this_page(cache, page, partition);
1082	}
1083
1084	static inline bool make_acquired_page_clean_and_evict_or_page_release(PGC cache, PGC_PAGE page) {
1085	pointer_check(cache, page);
1086
1087	WAITQ_PRIORITY prio = is_page_clean(page) ? PGC_QUEUE_LOCK_PRIO_EVICTORS : PGC_QUEUE_LOCK_PRIO_COLLECTORS;
1088
1089	page_transition_lock(cache, page);
1090	pgc_queue_lock(cache, &cache->clean, prio);
1091
1092	// make it clean - it does not have any accesses, so it will be prepended
1093	page_set_clean(cache, page, true, true, prio);
1094
1095	if(!acquired_page_get_for_deletion_or_release_it(cache, page)) {
1096	pgc_queue_unlock(cache, &cache->clean);
1097	page_transition_unlock(cache, page);
1098	return false;
1099	}
1100
1101	// remove it from the linked list
1102	pgc_queue_del(cache, &cache->clean, page, true, prio);
1103	pgc_queue_unlock(cache, &cache->clean);
1104	page_transition_unlock(cache, page);
1105
1106	remove_and_free_page_not_in_any_queue_and_acquired_for_deletion(cache, page);
1107
1108	return true;
1109	}
1110
1111	// returns true, when there is potentially more work to do
1112	static bool evict_pages_with_filter(PGC cache, size_t max_skip, size_t max_evict, bool wait, bool all_of_them, evict_filter filter, void data) {
1113	ssize_t per1000 = cache_usage_per1000(cache, NULL);
1114
1115	if(!all_of_them && per1000 < cache->config.healthy_size_per1000)
1116	// don't bother - not enough to do anything
1117	return false;
1118
1119	bool under_sever_pressure = per1000 >= cache->config.severe_pressure_per1000;
1120	size_t workers_running = __atomic_add_fetch(&cache->stats.p0_workers_evict, 1, __ATOMIC_RELAXED);
1121	if(!wait && !all_of_them && workers_running > cache->config.max_workers_evict_inline && !under_sever_pressure) {
1122	__atomic_sub_fetch(&cache->stats.p0_workers_evict, 1, __ATOMIC_RELAXED);
1123	return false;
1124	}
1125
1126	internal_fatal(cache->clean.linked_list_in_sections_judy,
1127	"wrong clean pages configuration - clean pages need to have a linked list, not a judy array");
1128
1129	if(unlikely(!max_skip))
1130	max_skip = SIZE_MAX;
1131	else if(unlikely(max_skip < 2))
1132	max_skip = 2;
1133
1134	if(unlikely(!max_evict))
1135	max_evict = SIZE_MAX;
1136	else if(unlikely(max_evict < 2))
1137	max_evict = 2;
1138
1139	size_t this_loop_evicted = 0;
1140	size_t total_pages_evicted = 0;
1141	size_t total_pages_relocated = 0;
1142	bool stopped_before_finishing = false;
1143	size_t spins = 0;
1144	size_t max_pages_to_evict = 0;
1145
1146	do {
1147	int64_t max_size_to_evict = 0;
1148	if (unlikely(all_of_them)) {
1149	// evict them all
1150	max_size_to_evict = SIZE_MAX;
1151	max_pages_to_evict = SIZE_MAX;
1152	under_sever_pressure = true;
1153	}
1154	else if(unlikely(wait)) {
1155	// evict as many as necessary for the cache to go at the predefined threshold
1156	per1000 = cache_usage_per1000(cache, &max_size_to_evict);
1157	if(per1000 >= cache->config.severe_pressure_per1000) {
1158	under_sever_pressure = true;
1159	max_pages_to_evict = max_pages_to_evict ? max_pages_to_evict * 2 : 16;
1160	if(max_pages_to_evict > 64)
1161	max_pages_to_evict = 64;
1162	}
1163	else if(per1000 >= cache->config.aggressive_evict_per1000) {
1164	under_sever_pressure = false;
1165	max_pages_to_evict = max_pages_to_evict ? max_pages_to_evict * 2 : 4;
1166	if(max_pages_to_evict > 16)
1167	max_pages_to_evict = 16;
1168	}
1169	else {
1170	under_sever_pressure = false;
1171	max_pages_to_evict = 1;
1172	}
1173	}
1174	else {
1175	// this is an adder, so evict just 1 page
1176	max_size_to_evict = (cache_above_healthy_limit(cache)) ? 1 : 0;
1177	max_pages_to_evict = 1;
1178	}
1179
1180	if (!max_size_to_evict \|\| !max_pages_to_evict)
1181	break;
1182
1183	// check if we have to stop
1184	if(total_pages_evicted >= max_evict && !all_of_them) {
1185	stopped_before_finishing = true;
1186	break;
1187	}
1188
1189	if(++spins > 1 && !this_loop_evicted)
1190	p2_add_fetch(&cache->stats.p2_waste_evict_useless_spins, 1);
1191
1192	this_loop_evicted = 0;
1193
1194	timing_dbengine_evict_init();
1195
1196	if(!all_of_them && !wait) {
1197	if(!pgc_queue_trylock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS)) {
1198	stopped_before_finishing = true;
1199	goto premature_exit;
1200	}
1201
1202	// at this point we have the clean lock
1203	}
1204	else
1205	pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS);
1206
1207	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_LOCK);
1208
1209	// find a page to evict
1210	PGC_PAGE *pages_to_evict = NULL;
1211	int64_t pages_to_evict_size = 0;
1212	size_t pages_to_evict_count = 0;
1213	for(PGC_PAGE page = cache->clean.base, next = NULL, *first_page_we_relocated = NULL; page ; page = next) {
1214	next = page->link.next;
1215
1216	if(unlikely(page == first_page_we_relocated))
1217	// we did a complete loop on all pages
1218	break;
1219
1220	if(unlikely(page_flag_check(page, PGC_PAGE_HAS_BEEN_ACCESSED \| PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES) == PGC_PAGE_HAS_BEEN_ACCESSED)) {
1221	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
1222	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
1223	page_flag_clear(page, PGC_PAGE_HAS_BEEN_ACCESSED);
1224	continue;
1225	}
1226
1227	if(unlikely(filter && !filter(page, data)))
1228	continue;
1229
1230	if(non_acquired_page_get_for_deletion___while_having_clean_locked(cache, page)) {
1231	// we can delete this page
1232
1233	// remove it from the clean list
1234	pgc_queue_del(cache, &cache->clean, page, true, PGC_QUEUE_LOCK_PRIO_EVICTORS);
1235
1236	__atomic_add_fetch(&cache->stats.evicting_entries, 1, __ATOMIC_RELAXED);
1237	__atomic_add_fetch(&cache->stats.evicting_size, page->assumed_size, __ATOMIC_RELAXED);
1238
1239	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(pages_to_evict, page, link.prev, link.next);
1240
1241	pages_to_evict_size += page->assumed_size;
1242	pages_to_evict_count++;
1243
1244	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_SELECT_PAGE);
1245
1246	if((pages_to_evict_count < max_pages_to_evict && pages_to_evict_size < max_size_to_evict) \|\| all_of_them)
1247	// get more pages
1248	;
1249	else
1250	// one page at a time
1251	break;
1252	}
1253	else {
1254	// we can't delete this page
1255
1256	if(!first_page_we_relocated)
1257	first_page_we_relocated = page;
1258
1259	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
1260	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
1261
1262	total_pages_relocated++;
1263
1264	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_RELOCATE_PAGE);
1265
1266	// check if we have to stop
1267	if(unlikely(total_pages_relocated >= max_skip && !all_of_them)) {
1268	stopped_before_finishing = true;
1269	break;
1270	}
1271	}
1272	}
1273	pgc_queue_unlock(cache, &cache->clean);
1274
1275	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_SELECT);
1276
1277	if(likely(pages_to_evict)) {
1278	// remove them from the index
1279
1280	if(unlikely(pages_to_evict->link.next)) {
1281	// we have many pages, let's minimize the index locks we are going to get
1282
1283	PGC_PAGE *pages_per_partition[cache->config.partitions];
1284	memset(pages_per_partition, 0, sizeof(PGC_PAGE ) cache->config.partitions);
1285
1286	bool partitions_done[cache->config.partitions];
1287	memset(partitions_done, 0, sizeof(bool) * cache->config.partitions);
1288
1289	// sort them by partition
1290	for (PGC_PAGE page = pages_to_evict, next = NULL; page; page = next) {
1291	next = page->link.next;
1292
1293	size_t partition = pgc_indexing_partition(cache, page->metric_id);
1294	DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(pages_to_evict, page, link.prev, link.next);
1295	DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(pages_per_partition[partition], page, link.prev, link.next);
1296	}
1297
1298	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_SORT);
1299
1300	// remove them from the index
1301	size_t remaining_partitions = cache->config.partitions;
1302	size_t last_remaining_partitions = remaining_partitions + 1;
1303	while(remaining_partitions) {
1304	bool force = remaining_partitions == last_remaining_partitions;
1305	last_remaining_partitions = remaining_partitions;
1306	remaining_partitions = 0;
1307
1308	for (size_t partition = 0; partition < cache->config.partitions; partition++) {
1309	if (!pages_per_partition[partition] \|\| partitions_done[partition])
1310	continue;
1311
1312	if(pgc_index_trywrite_lock(cache, partition, force)) {
1313	partitions_done[partition] = true;
1314
1315	for (PGC_PAGE *page = pages_per_partition[partition]; page; page = page->link.next)
1316	remove_this_page_from_index_unsafe(cache, page, partition);
1317
1318	pgc_index_write_unlock(cache, partition);
1319
1320	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_DEINDEX_PAGE);
1321	}
1322	else
1323	remaining_partitions++;
1324	}
1325	}
1326
1327	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_DEINDEX);
1328
1329	// free them
1330	for (size_t partition = 0; partition < cache->config.partitions; partition++) {
1331	if (!pages_per_partition[partition]) continue;
1332
1333	for (PGC_PAGE page = pages_per_partition[partition], next = NULL; page; page = next) {
1334	next = page->link.next;
1335
1336	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_FREE_LOOP);
1337
1338	int64_t page_size = page->assumed_size;
1339	free_this_page(cache, page, partition);
1340
1341	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_FREE_PAGE);
1342
1343	__atomic_sub_fetch(&cache->stats.evicting_entries, 1, __ATOMIC_RELAXED);
1344	__atomic_sub_fetch(&cache->stats.evicting_size, page_size, __ATOMIC_RELAXED);
1345
1346	total_pages_evicted++;
1347	this_loop_evicted++;
1348
1349	timing_dbengine_evict_step(TIMING_STEP_DBENGINE_EVICT_FREE_ATOMICS);
1350	}
1351	}
1352
1353	timing_dbengine_evict_report();
1354	}
1355	else {
1356	// just one page to be evicted
1357	PGC_PAGE *page = pages_to_evict;
1358
1359	int64_t page_size = page->assumed_size;
1360
1361	size_t partition = pgc_indexing_partition(cache, page->metric_id);
1362	pgc_index_write_lock(cache, partition);
1363	remove_this_page_from_index_unsafe(cache, page, partition);
1364	pgc_index_write_unlock(cache, partition);
1365	free_this_page(cache, page, partition);
1366
1367	__atomic_sub_fetch(&cache->stats.evicting_entries, 1, __ATOMIC_RELAXED);
1368	__atomic_sub_fetch(&cache->stats.evicting_size, page_size, __ATOMIC_RELAXED);
1369
1370	total_pages_evicted++;
1371	this_loop_evicted++;
1372	}
1373	}
1374	else
1375	break;
1376
1377	} while(all_of_them \|\| (total_pages_evicted < max_evict && total_pages_relocated < max_skip));
1378
1379	if(all_of_them && !filter) {
1380	pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS);
1381	size_t entries = __atomic_load_n(&cache->clean.stats->entries, __ATOMIC_RELAXED);
1382	if(entries) {
1383	nd_log_limit_static_global_var(erl, 1, 0);
1384	nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE,
1385	"DBENGINE CACHE: cannot free all clean pages, %zu are still in the clean queue",
1386	entries);
1387	}
1388	pgc_queue_unlock(cache, &cache->clean);
1389	}
1390
1391	premature_exit:
1392	if(unlikely(total_pages_relocated))
1393	p2_add_fetch(&cache->stats.p2_waste_evict_relocated, total_pages_relocated);
1394
1395	__atomic_sub_fetch(&cache->stats.p0_workers_evict, 1, __ATOMIC_RELAXED);
1396
1397	return stopped_before_finishing;
1398	}
1399
1400	static PGC_PAGE pgc_page_add(PGC cache, PGC_ENTRY entry, bool added) {
1401	internal_fatal(entry->start_time_s < 0 \|\| entry->end_time_s < 0,
1402	"DBENGINE CACHE: timestamps are negative");
1403
1404	// Clamp before the values are copied into the new PGC_PAGE and used as
1405	// the Judy index key, so the struct and the index agree.
1406	if(unlikely(entry->start_time_s < 0))
1407	entry->start_time_s = 0;
1408
1409	if(unlikely(entry->end_time_s < 0))
1410	entry->end_time_s = 0;
1411
1412	p2_add_fetch(&cache->stats.p2_workers_add, 1);
1413
1414	size_t partition = pgc_indexing_partition(cache, entry->metric_id);
1415
1416	#ifdef PGC_WITH_ARAL
1417	PGC_PAGE *allocation = aral_mallocz(cache->index[partition].aral);
1418	#else
1419	PGC_PAGE *allocation = mallocz(sizeof(PGC_PAGE) + cache->config.additional_bytes_per_page);
1420	#endif
1421
1422	allocation->refcount = 1;
1423	allocation->accesses = (entry->hot) ? 0 : 1;
1424	allocation->flags = 0;
1425	allocation->section = entry->section;
1426	allocation->metric_id = entry->metric_id;
1427	allocation->start_time_s = entry->start_time_s;
1428	allocation->end_time_s = entry->end_time_s,
1429	allocation->update_every_s = entry->update_every_s,
1430	allocation->data = entry->data;
1431	allocation->assumed_size = page_assumed_size(cache, entry->size);
1432	spinlock_init(&allocation->transition_spinlock);
1433	allocation->link.prev = NULL;
1434	allocation->link.next = NULL;
1435
1436	if(cache->config.additional_bytes_per_page) {
1437	if(entry->custom_data)
1438	memcpy(allocation->custom_data, entry->custom_data, cache->config.additional_bytes_per_page);
1439	else
1440	memset(allocation->custom_data, 0, cache->config.additional_bytes_per_page);
1441	}
1442
1443	PGC_PAGE *page;
1444	size_t spins = 0;
1445
1446	do {
1447	spins++;
1448
1449	pgc_index_write_lock(cache, partition);
1450
1451	JudyAllocThreadPulseReset();
1452
1453	Pvoid_t *metrics_judy_pptr = JudyLIns(&cache->index[partition].sections_judy, entry->section, PJE0);
1454	if(unlikely(!metrics_judy_pptr \|\| metrics_judy_pptr == PJERR))
1455	fatal("DBENGINE CACHE: JudyLIns(sections_judy, 0x%lx) failed, sections_judy = %p, result = %p",
1456	(long unsigned)entry->section, cache->index[partition].sections_judy, metrics_judy_pptr);
1457
1458	Pvoid_t *pages_judy_pptr = JudyLIns(metrics_judy_pptr, entry->metric_id, PJE0);
1459	if(unlikely(!pages_judy_pptr \|\| pages_judy_pptr == PJERR))
1460	fatal("DBENGINE CACHE: JudyLIns(metrics_judy, 0x%lx) failed, metrics_judy = %p, result = %p",
1461	(long unsigned)entry->metric_id, metrics_judy_pptr, pages_judy_pptr);
1462
1463	Pvoid_t *page_ptr = JudyLIns(pages_judy_pptr, entry->start_time_s, PJE0);
1464	if(unlikely(!page_ptr \|\| page_ptr == PJERR))
1465	fatal("DBENGINE CACHE: JudyLIns(pages_judy, %ld) failed, pages_judy = %p, result = %p",
1466	(long)entry->start_time_s, pages_judy_pptr, page_ptr);
1467
1468	pgc_stats_index_judy_change(cache, JudyAllocThreadPulseGetAndReset());
1469
1470	page = *page_ptr;
1471
1472	if (likely(!page)) {
1473	// consume it
1474	page = allocation;
1475	allocation = NULL;
1476
1477	// put it in the index
1478	*page_ptr = page;
1479	pointer_add(cache, page);
1480	pgc_index_write_unlock(cache, partition);
1481
1482	if (entry->hot)
1483	page_set_hot(cache, page, PGC_QUEUE_LOCK_PRIO_COLLECTORS);
1484	else
1485	page_set_clean(cache, page, false, false, PGC_QUEUE_LOCK_PRIO_EVICTORS);
1486
1487	PGC_REFERENCED_PAGES_PLUS1(cache, page);
1488
1489	// update statistics
1490	__atomic_add_fetch(&cache->stats.added_entries, 1, __ATOMIC_RELAXED);
1491	__atomic_add_fetch(&cache->stats.added_size, page->assumed_size, __ATOMIC_RELAXED);
1492
1493	__atomic_add_fetch(&cache->stats.entries, 1, __ATOMIC_RELAXED);
1494	__atomic_add_fetch(&cache->stats.size, page->assumed_size, __ATOMIC_RELAXED);
1495
1496	if(added)
1497	*added = true;
1498	}
1499	else {
1500	if (!page_acquire(cache, page))
1501	page = NULL;
1502
1503	else if(added)
1504	*added = false;
1505
1506	pgc_index_write_unlock(cache, partition);
1507
1508	if(unlikely(!page)) {
1509	// now that we don't have the lock,
1510	// give it some time for the old page to go away
1511	tinysleep();
1512	}
1513	}
1514
1515	} while(!page);
1516
1517	if(allocation) {
1518	#ifdef PGC_WITH_ARAL
1519	aral_freez(cache->index[partition].aral, allocation);
1520	#else
1521	freez(allocation);
1522	#endif
1523	}
1524
1525	if(spins > 1)
1526	p2_add_fetch(&cache->stats.p2_waste_insert_spins, spins - 1);
1527
1528	p2_sub_fetch(&cache->stats.p2_workers_add, 1);
1529
1530	if(!entry->hot)
1531	evict_on_clean_page_added(cache);
1532
1533	flush_on_page_add(cache);
1534
1535	return page;
1536	}
1537
1538	static ALWAYS_INLINE PGC_PAGE page_find_and_acquire_exact_unsafe(PGC cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1539	Pvoid_t page_ptr = JudyLGet(pages_judy_pptr, start_time_s, PJE0);
1540	if(!page_ptr)
1541	return NULL;
1542
1543	if (unlikely(page_ptr == PJERR))
1544	fatal("DBENGINE CACHE: corrupted page in pages judy array");
1545
1546	PGC_PAGE page = page_ptr;
1547	if(page && page_acquire(cache, page))
1548	// we have our page acquired
1549	return page;
1550
1551	return NULL;
1552	}
1553
1554	static ALWAYS_INLINE PGC_PAGE page_find_and_acquire_first_unsafe(PGC cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1555	Word_t time = start_time_s;
1556	for(Pvoid_t page_ptr = JudyLFirst(pages_judy_pptr, &time, PJE0);
1557	page_ptr ;
1558	page_ptr = JudyLNext(*pages_judy_pptr, &time, PJE0)) {
1559
1560	if (unlikely(page_ptr == PJERR))
1561	fatal("DBENGINE CACHE: corrupted page in pages judy array");
1562
1563	PGC_PAGE page = page_ptr;
1564	if(page && page_acquire(cache, page))
1565	// we have our page acquired
1566	return page;
1567	}
1568
1569	return NULL;
1570	}
1571
1572	static ALWAYS_INLINE PGC_PAGE page_find_and_acquire_next_unsafe(PGC cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1573	Word_t time = start_time_s;
1574	for(Pvoid_t page_ptr = JudyLNext(pages_judy_pptr, &time, PJE0);
1575	page_ptr ;
1576	page_ptr = JudyLNext(*pages_judy_pptr, &time, PJE0)) {
1577
1578	if (unlikely(page_ptr == PJERR))
1579	fatal("DBENGINE CACHE: corrupted page in pages judy array");
1580
1581	PGC_PAGE page = page_ptr;
1582	if(page && page_acquire(cache, page))
1583	// we have our page acquired
1584	return page;
1585	}
1586
1587	return NULL;
1588	}
1589
1590	static ALWAYS_INLINE PGC_PAGE page_find_and_acquire_last_unsafe(PGC cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1591	Word_t time = start_time_s;
1592	for(Pvoid_t page_ptr = JudyLLast(pages_judy_pptr, &time, PJE0);
1593	page_ptr ;
1594	page_ptr = JudyLPrev(*pages_judy_pptr, &time, PJE0)) {
1595
1596	if (unlikely(page_ptr == PJERR))
1597	fatal("DBENGINE CACHE: corrupted page in pages judy array");
1598
1599	PGC_PAGE page = page_ptr;
1600	if(page && page_acquire(cache, page))
1601	// we have our page acquired
1602	return page;
1603	}
1604
1605	return NULL;
1606	}
1607
1608	static ALWAYS_INLINE PGC_PAGE page_find_and_acquire_prev_unsafe(PGC cache, Pvoid_t *pages_judy_pptr, time_t start_time_s) {
1609	Word_t time = start_time_s;
1610	for(Pvoid_t page_ptr = JudyLPrev(pages_judy_pptr, &time, PJE0);
1611	page_ptr ;
1612	page_ptr = JudyLPrev(*pages_judy_pptr, &time, PJE0)) {
1613
1614	if (unlikely(page_ptr == PJERR))
1615	fatal("DBENGINE CACHE: corrupted page in pages judy array");
1616
1617	PGC_PAGE page = page_ptr;
1618	if(page && page_acquire(cache, page))
1619	// we have our page acquired
1620	return page;
1621	}
1622
1623	return NULL;
1624	}
1625
1626	static ALWAYS_INLINE PGC_PAGE page_find_and_acquire_once(PGC cache, Word_t section, Word_t metric_id, time_t start_time_s, PGC_SEARCH method) {
1627	PGC_PAGE *page = NULL;
1628	size_t partition = pgc_indexing_partition(cache, metric_id);
1629
1630	pgc_index_read_lock(cache, partition);
1631
1632	Pvoid_t *metrics_judy_pptr = JudyLGet(cache->index[partition].sections_judy, section, PJE0);
1633	if(unlikely(metrics_judy_pptr == PJERR))
1634	fatal("DBENGINE CACHE: corrupted sections judy array");
1635
1636	if(unlikely(!metrics_judy_pptr)) {
1637	// section does not exist
1638	goto cleanup;
1639	}
1640
1641	Pvoid_t pages_judy_pptr = JudyLGet(metrics_judy_pptr, metric_id, PJE0);
1642	if(unlikely(pages_judy_pptr == PJERR))
1643	fatal("DBENGINE CACHE: corrupted pages judy array");
1644
1645	if(unlikely(!pages_judy_pptr)) {
1646	// metric does not exist
1647	goto cleanup;
1648	}
1649
1650	switch(method) {
1651	default:
1652	case PGC_SEARCH_CLOSEST: {
1653	page = page_find_and_acquire_exact_unsafe(cache, pages_judy_pptr, start_time_s);
1654	if(!page) {
1655	page = page_find_and_acquire_prev_unsafe(cache, pages_judy_pptr, start_time_s);
1656	if(page && start_time_s > page->end_time_s) {
1657	// found a page starting before our timestamp
1658	// but our timestamp is not included in it
1659	page_release(cache, page, false);
1660	page = NULL;
1661	}
1662
1663	if(!page)
1664	page = page_find_and_acquire_next_unsafe(cache, pages_judy_pptr, start_time_s);
1665	}
1666	}
1667	break;
1668
1669	case PGC_SEARCH_EXACT:
1670	page = page_find_and_acquire_exact_unsafe(cache, pages_judy_pptr, start_time_s);
1671	break;
1672
1673	case PGC_SEARCH_FIRST:
1674	page = page_find_and_acquire_first_unsafe(cache, pages_judy_pptr, start_time_s);
1675	break;
1676
1677	case PGC_SEARCH_NEXT:
1678	page = page_find_and_acquire_next_unsafe(cache, pages_judy_pptr, start_time_s);
1679	break;
1680
1681	case PGC_SEARCH_LAST:
1682	page = page_find_and_acquire_last_unsafe(cache, pages_judy_pptr, start_time_s);
1683	break;
1684
1685	case PGC_SEARCH_PREV:
1686	page = page_find_and_acquire_prev_unsafe(cache, pages_judy_pptr, start_time_s);
1687	break;
1688	}
1689
1690	#ifdef NETDATA_PGC_POINTER_CHECK
1691	if(page)
1692	pointer_check(cache, page);
1693	#endif
1694
1695	cleanup:
1696	pgc_index_read_unlock(cache, partition);
1697	return page;
1698	}
1699
1700	static void all_hot_pages_to_dirty(PGC *cache, Word_t section) {
1701	pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_COLLECTORS);
1702
1703	bool first = true;
1704	Word_t last_section = (section == PGC_SECTION_ALL) ? 0 : section;
1705	Pvoid_t *section_pages_pptr;
1706	while ((section_pages_pptr = JudyLFirstThenNext(cache->hot.sections_judy, &last_section, &first))) {
1707	if(section != PGC_SECTION_ALL && last_section != section)
1708	break;
1709
1710	struct section_pages sp = section_pages_pptr;
1711
1712	PGC_PAGE *page = sp->base;
1713	while(page) {
1714	PGC_PAGE *next = page->link.next;
1715
1716	if(page_acquire(cache, page)) {
1717	page_set_dirty(cache, page, true, PGC_QUEUE_LOCK_PRIO_COLLECTORS);
1718	page_release(cache, page, false);
1719	// page ptr may be invalid now
1720	}
1721
1722	page = next;
1723	}
1724	}
1725	pgc_queue_unlock(cache, &cache->hot);
1726	}
1727
1728	// returns true when there is more work to do
1729	static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wait, bool all_of_them) {
1730	internal_fatal(!cache->dirty.linked_list_in_sections_judy,
1731	"wrong dirty pages configuration - dirty pages need to have a judy array, not a linked list");
1732
1733	if(!all_of_them && !wait) {
1734	// we have been called from a data collection thread
1735	// let's not waste its time...
1736
1737	if(!pgc_queue_trylock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS)) {
1738	// we would block, so give up...
1739	return false;
1740	}
1741
1742	// we got the lock at this point
1743	}
1744	else
1745	pgc_queue_lock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS);
1746
1747	size_t optimal_flush_size = cache->config.max_dirty_pages_per_call;
1748	size_t dirty_version_at_entry = cache->dirty.version;
1749	size_t entries = __atomic_load_n(&cache->dirty.stats->entries, __ATOMIC_RELAXED);
1750	if(!all_of_them && (entries < optimal_flush_size \|\| cache->dirty.last_version_checked == dirty_version_at_entry)) {
1751	pgc_queue_unlock(cache, &cache->dirty);
1752	return false;
1753	}
1754
1755	p2_add_fetch(&cache->stats.p2_workers_flush, 1);
1756
1757	bool have_dirty_lock = true;
1758
1759	if(all_of_them \|\| !max_flushes)
1760	max_flushes = SIZE_MAX;
1761
1762	Word_t last_section = (section == PGC_SECTION_ALL) ? 0 : section;
1763	size_t flushes_so_far = 0;
1764	Pvoid_t *section_pages_pptr;
1765	bool stopped_before_finishing = false;
1766	bool first = true;
1767
1768	while (have_dirty_lock && (section_pages_pptr = JudyLFirstThenNext(cache->dirty.sections_judy, &last_section, &first))) {
1769	if(section != PGC_SECTION_ALL && last_section != section)
1770	break;
1771
1772	struct section_pages sp = section_pages_pptr;
1773	if(!all_of_them && sp->entries < optimal_flush_size)
1774	continue;
1775
1776	if(!all_of_them && flushes_so_far > max_flushes) {
1777	stopped_before_finishing = true;
1778	break;
1779	}
1780
1781	PGC_ENTRY array[optimal_flush_size];
1782	PGC_PAGE *pages[optimal_flush_size];
1783
1784	size_t pages_added = 0,
1785	pages_removed_dirty = 0,
1786	pages_cancelled = 0,
1787	pages_made_clean = 0;
1788
1789	int64_t pages_added_size = 0,
1790	pages_removed_dirty_size = 0,
1791	pages_cancelled_size = 0,
1792	pages_made_clean_size = 0;
1793
1794	PGC_PAGE *page = sp->base;
1795	while (page && pages_added < optimal_flush_size) {
1796	PGC_PAGE *next = page->link.next;
1797
1798	internal_fatal(page_get_status_flags(page) != PGC_PAGE_DIRTY,
1799	"DBENGINE CACHE: page should be in the dirty list before saved");
1800
1801	if (page_acquire(cache, page)) {
1802	internal_fatal(page_get_status_flags(page) != PGC_PAGE_DIRTY,
1803	"DBENGINE CACHE: page should be in the dirty list before saved");
1804
1805	internal_fatal(page->section != last_section,
1806	"DBENGINE CACHE: dirty page is not in the right section (tier)");
1807
1808	if(!page_transition_trylock(cache, page)) {
1809	page_release(cache, page, false);
1810	// page ptr may be invalid now
1811	}
1812	else {
1813	pages[pages_added] = page;
1814	array[pages_added] = (PGC_ENTRY) {
1815	.section = page->section,
1816	.metric_id = page->metric_id,
1817	.start_time_s = page->start_time_s,
1818	.end_time_s = __atomic_load_n(&page->end_time_s, __ATOMIC_RELAXED),
1819	.update_every_s = page->update_every_s,
1820	.size = page_size_from_assumed_size(cache, page->assumed_size),
1821	.data = page->data,
1822	.custom_data = (cache->config.additional_bytes_per_page) ? page->custom_data : NULL,
1823	.hot = false,
1824	};
1825
1826	pages_added_size += page->assumed_size;
1827	pages_added++;
1828	}
1829	}
1830
1831	page = next;
1832	}
1833
1834	// do we have enough to save?
1835	if(all_of_them \|\| pages_added == optimal_flush_size) {
1836	// we should do it
1837
1838	for (size_t i = 0; i < pages_added; i++) {
1839	PGC_PAGE *tpg = pages[i];
1840
1841	internal_fatal(page_get_status_flags(tpg) != PGC_PAGE_DIRTY,
1842	"DBENGINE CACHE: page should be in the dirty list before saved");
1843
1844	__atomic_add_fetch(&cache->stats.flushing_entries, 1, __ATOMIC_RELAXED);
1845	__atomic_add_fetch(&cache->stats.flushing_size, tpg->assumed_size, __ATOMIC_RELAXED);
1846
1847	// remove it from the dirty list
1848	pgc_queue_del(cache, &cache->dirty, tpg, true, PGC_QUEUE_LOCK_PRIO_FLUSHERS);
1849
1850	pages_removed_dirty_size += tpg->assumed_size;
1851	pages_removed_dirty++;
1852	}
1853
1854	// next time, repeat the same section (tier)
1855	first = true;
1856	}
1857	else {
1858	// we can't do it
1859
1860	for (size_t i = 0; i < pages_added; i++) {
1861	PGC_PAGE *tpg = pages[i];
1862
1863	internal_fatal(page_get_status_flags(tpg) != PGC_PAGE_DIRTY,
1864	"DBENGINE CACHE: page should be in the dirty list before saved");
1865
1866	pages_cancelled_size += tpg->assumed_size;
1867	pages_cancelled++;
1868
1869	page_transition_unlock(cache, tpg);
1870	page_release(cache, tpg, false);
1871	// page ptr may be invalid now
1872	}
1873
1874	p2_add_fetch(&cache->stats.p2_waste_flushes_cancelled, pages_cancelled);
1875	p2_add_fetch(&cache->stats.flushes_cancelled_size, pages_cancelled_size);
1876
1877	internal_fatal(pages_added != pages_cancelled \|\| pages_added_size != pages_cancelled_size,
1878	"DBENGINE CACHE: flushing cancel pages mismatch");
1879
1880	// next time, continue to the next section (tier)
1881	first = false;
1882	continue;
1883	}
1884
1885	if(cache->config.pgc_save_init_cb)
1886	cache->config.pgc_save_init_cb(cache, last_section);
1887
1888	pgc_queue_unlock(cache, &cache->dirty);
1889	have_dirty_lock = false;
1890
1891	// call the callback to save them
1892	// it may take some time, so let's release the lock
1893	if(cache->config.pgc_save_dirty_cb)
1894	cache->config.pgc_save_dirty_cb(cache, array, pages, pages_added);
1895
1896	flushes_so_far++;
1897
1898	__atomic_add_fetch(&cache->stats.flushes_completed, pages_added, __ATOMIC_RELAXED);
1899	__atomic_add_fetch(&cache->stats.flushes_completed_size, pages_added_size, __ATOMIC_RELAXED);
1900
1901	size_t pages_to_evict = 0; (void)pages_to_evict;
1902	for (size_t i = 0; i < pages_added; i++) {
1903	PGC_PAGE *tpg = pages[i];
1904
1905	internal_fatal(page_get_status_flags(tpg) != 0,
1906	"DBENGINE CACHE: page should not be in any list while it is being saved");
1907
1908	__atomic_sub_fetch(&cache->stats.flushing_entries, 1, __ATOMIC_RELAXED);
1909	__atomic_sub_fetch(&cache->stats.flushing_size, tpg->assumed_size, __ATOMIC_RELAXED);
1910
1911	pages_made_clean_size += tpg->assumed_size;
1912	pages_made_clean++;
1913
1914	if(!tpg->accesses)
1915	pages_to_evict++;
1916
1917	page_set_clean(cache, tpg, true, false, PGC_QUEUE_LOCK_PRIO_FLUSHERS);
1918	page_transition_unlock(cache, tpg);
1919	page_release(cache, tpg, false);
1920	// tpg ptr may be invalid now
1921	}
1922
1923	internal_fatal(pages_added != pages_made_clean \|\| pages_added != pages_removed_dirty \|\|
1924	pages_added_size != pages_made_clean_size \|\| pages_added_size != pages_removed_dirty_size
1925	, "DBENGINE CACHE: flushing pages mismatch");
1926
1927	if(!all_of_them && !wait) {
1928	if(pgc_queue_trylock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS))
1929	have_dirty_lock = true;
1930
1931	else {
1932	stopped_before_finishing = true;
1933	have_dirty_lock = false;
1934	}
1935	}
1936	else {
1937	pgc_queue_lock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS);
1938	have_dirty_lock = true;
1939	}
1940	}
1941
1942	if(have_dirty_lock) {
1943	if(!stopped_before_finishing && dirty_version_at_entry > cache->dirty.last_version_checked)
1944	cache->dirty.last_version_checked = dirty_version_at_entry;
1945
1946	pgc_queue_unlock(cache, &cache->dirty);
1947	}
1948
1949	p2_sub_fetch(&cache->stats.p2_workers_flush, 1);
1950
1951	return stopped_before_finishing;
1952	}
1953
1954	void free_all_unreferenced_clean_pages(PGC *cache) {
1955	evict_pages(cache, 0, 0, true, true);
1956	}
1957
1958	static void pgc_evict_thread(void *ptr) {
1959	static usec_t last_malloc_release_ut = 0;
1960
1961	PGC *cache = ptr;
1962
1963	worker_register("PGCEVICT");
1964	worker_register_job_name(0, "signaled");
1965	worker_register_job_name(1, "scheduled");
1966	worker_register_job_name(2, "cleanup");
1967
1968	unsigned job_id = 0;
1969
1970	while (true) {
1971	worker_is_idle();
1972	unsigned new_job_id = completion_wait_for_a_job_with_timeout(
1973	&cache->evictor.completion, job_id, 1000);
1974
1975	worker_is_busy(new_job_id > job_id ? 1 : 0);
1976	job_id = new_job_id;
1977
1978	if (nd_thread_signaled_to_cancel())
1979	break;
1980
1981	int64_t size_to_evict = 0;
1982	bool system_cleanup = false;
1983	if(cache_usage_per1000(cache, &size_to_evict) > cache->config.aggressive_evict_per1000)
1984	system_cleanup = true;
1985
1986	evict_pages(cache, 0, 0, true, false);
1987
1988	if(system_cleanup) {
1989	usec_t now_ut = now_monotonic_usec();
1990
1991	if(__atomic_load_n(&last_malloc_release_ut, __ATOMIC_RELAXED) + USEC_PER_SEC <= now_ut) {
1992	__atomic_store_n(&last_malloc_release_ut, now_ut, __ATOMIC_RELAXED);
1993	worker_is_busy(2);
1994	mallocz_release_as_much_memory_to_the_system();
1995	}
1996	}
1997	}
1998
1999	worker_unregister();
2000	}
2001
2002	// ----------------------------------------------------------------------------
2003	// public API
2004
2005	PGC pgc_create(const char name,
2006	size_t clean_size_bytes,
2007	free_clean_page_callback pgc_free_cb,
2008	size_t max_dirty_pages_per_flush,
2009	save_dirty_init_callback pgc_save_init_cb,
2010	save_dirty_page_callback pgc_save_dirty_cb,
2011	size_t max_pages_per_inline_eviction,
2012	size_t max_inline_evictors,
2013	size_t max_skip_pages_per_inline_eviction,
2014	size_t max_flushes_inline,
2015	PGC_OPTIONS options,
2016	size_t partitions,
2017	size_t additional_bytes_per_page) {
2018
2019	if(max_pages_per_inline_eviction < 1)
2020	max_pages_per_inline_eviction = 1;
2021
2022	if(max_dirty_pages_per_flush < 1)
2023	max_dirty_pages_per_flush = 1;
2024
2025	if(max_flushes_inline * max_dirty_pages_per_flush < 2)
2026	max_flushes_inline = 2;
2027
2028	PGC *cache = callocz(1, sizeof(PGC));
2029	strncpyz(cache->config.name, name, PGC_NAME_MAX);
2030
2031	cache->config.options = options;
2032	cache->config.additional_bytes_per_page = additional_bytes_per_page;
2033	cache->config.stats = pulse_enabled;
2034
2035	// flushing
2036	cache->config.max_flushes_inline = (max_flushes_inline == 0) ? 2 : max_flushes_inline;
2037	cache->config.max_dirty_pages_per_call = max_dirty_pages_per_flush;
2038	cache->config.pgc_save_init_cb = pgc_save_init_cb;
2039	cache->config.pgc_save_dirty_cb = pgc_save_dirty_cb;
2040
2041	// eviction strategy
2042	cache->config.clean_size = (clean_size_bytes < 1 * 1024 * 1024) ? 1 * 1024 * 1024 : (int64_t)clean_size_bytes;
2043	cache->config.pgc_free_clean_cb = pgc_free_cb;
2044	cache->config.max_workers_evict_inline = max_inline_evictors;
2045	cache->config.max_pages_per_inline_eviction = max_pages_per_inline_eviction;
2046	cache->config.max_skip_pages_per_inline_eviction = (max_skip_pages_per_inline_eviction < 2) ? 2 : max_skip_pages_per_inline_eviction;
2047	cache->config.severe_pressure_per1000 = 1010; // INLINE: use releasers to evict pages (up to max_pages_per_inline_eviction)
2048	cache->config.aggressive_evict_per1000 = 990; // INLINE: use adders to evict pages (up to max_pages_per_inline_eviction)
2049	cache->config.healthy_size_per1000 = 980; // no evictions happen below this threshold
2050	cache->config.evict_low_threshold_per1000 = 970; // when evicting, bring the size down to this threshold
2051	// the eviction thread is signaled ONLY if we run out of memory
2052	// otherwise, it runs by itself every 100ms
2053
2054	// use all ram and protection from out of memory
2055	cache->config.use_all_ram = dbengine_use_all_ram_for_caches;
2056	cache->config.out_of_memory_protection_bytes = (int64_t)dbengine_out_of_memory_protection;
2057
2058	// partitions
2059	if(partitions == 0) partitions = netdata_conf_cpus() * 2;
2060	if(partitions <= 4) partitions = 4;
2061	if(partitions > 256) partitions = 256;
2062	cache->config.partitions = partitions;
2063	cache->index = callocz(cache->config.partitions, sizeof(struct pgc_index));
2064
2065	pgc_section_pages_static_aral_init();
2066
2067	for(size_t part = 0; part < cache->config.partitions ; part++) {
2068	rw_spinlock_init(&cache->index[part].rw_spinlock);
2069	#ifdef PGC_WITH_ARAL
2070	{
2071	char buf[100];
2072	snprintfz(buf, sizeof(buf), "%s", name);
2073	cache->index[part].aral = aral_create(
2074	buf,
2075	sizeof(PGC_PAGE) + cache->config.additional_bytes_per_page,
2076	0,
2077	0,
2078	&pgc_aral_statistics,
2079	NULL, NULL,
2080	false, false, false);
2081	}
2082	#endif
2083	}
2084
2085
2086	#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
2087	waitq_init(&cache->hot.wq);
2088	waitq_init(&cache->dirty.wq);
2089	waitq_init(&cache->clean.wq);
2090	#else
2091	spinlock_init(&cache->hot.spinlock);
2092	spinlock_init(&cache->dirty.spinlock);
2093	spinlock_init(&cache->clean.spinlock);
2094	#endif
2095
2096	cache->hot.flags = PGC_PAGE_HOT;
2097	cache->hot.linked_list_in_sections_judy = true;
2098	cache->hot.stats = &cache->stats.queues[PGC_QUEUE_HOT];
2099
2100	cache->dirty.flags = PGC_PAGE_DIRTY;
2101	cache->dirty.linked_list_in_sections_judy = true;
2102	cache->dirty.stats = &cache->stats.queues[PGC_QUEUE_DIRTY];
2103
2104	cache->clean.flags = PGC_PAGE_CLEAN;
2105	cache->clean.linked_list_in_sections_judy = false;
2106	cache->clean.stats = &cache->stats.queues[PGC_QUEUE_CLEAN];
2107
2108	pointer_index_init(cache);
2109	pgc_size_histogram_init(&cache->hot.stats->size_histogram);
2110	pgc_size_histogram_init(&cache->dirty.stats->size_histogram);
2111	pgc_size_histogram_init(&cache->clean.stats->size_histogram);
2112
2113	// last create the eviction thread
2114	{
2115	completion_init(&cache->evictor.completion);
2116	cache->evictor.thread = nd_thread_create(name, NETDATA_THREAD_OPTION_DEFAULT, pgc_evict_thread, cache);
2117	}
2118
2119	return cache;
2120	}
2121
2122	struct aral_statistics *pgc_aral_stats(void) {
2123	return &pgc_aral_statistics;
2124	}
2125
2126	void pgc_flush_dirty_pages(PGC *cache, Word_t section) {
2127	flush_pages(cache, 0, section, true, true);
2128	}
2129
2130	void pgc_flush_all_hot_and_dirty_pages(PGC *cache, Word_t section) {
2131	all_hot_pages_to_dirty(cache, section);
2132
2133	// save all dirty pages to make them clean
2134	flush_pages(cache, 0, section, true, true);
2135	}
2136
2137	void pgc_destroy(PGC *cache, bool flush) {
2138	if(!cache)
2139	return;
2140
2141	if(!flush) {
2142	cache->config.pgc_save_init_cb = NULL;
2143	cache->config.pgc_save_dirty_cb = NULL;
2144	}
2145
2146	// convert all hot pages to dirty
2147	all_hot_pages_to_dirty(cache, PGC_SECTION_ALL);
2148
2149	// save all dirty pages to make them clean
2150	flush_pages(cache, 0, PGC_SECTION_ALL, true, true);
2151
2152	// free all unreferenced clean pages
2153	free_all_unreferenced_clean_pages(cache);
2154
2155	// stop the eviction thread
2156	nd_thread_signal_cancel(cache->evictor.thread);
2157	completion_mark_complete_a_job(&cache->evictor.completion);
2158	nd_thread_join(cache->evictor.thread);
2159	completion_destroy(&cache->evictor.completion);
2160
2161	if(PGC_REFERENCED_PAGES(cache))
2162	netdata_log_error("DBENGINE CACHE: there are %zu referenced cache pages - leaving the cache allocated", PGC_REFERENCED_PAGES(cache));
2163	else {
2164	pointer_destroy_index(cache);
2165
2166	for(size_t part = 0; part < cache->config.partitions ;part++) {
2167	// netdata_rwlock_destroy(&cache->index[part].rw_spinlock);
2168	#ifdef PGC_WITH_ARAL
2169	aral_destroy(cache->index[part].aral);
2170	#endif
2171	}
2172
2173	#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
2174	waitq_destroy(&cache->hot.wq);
2175	waitq_destroy(&cache->dirty.wq);
2176	waitq_destroy(&cache->clean.wq);
2177	#endif
2178	freez(cache->index);
2179	freez(cache);
2180	}
2181	}
2182
2183	ALWAYS_INLINE PGC_PAGE pgc_page_add_and_acquire(PGC cache, PGC_ENTRY entry, bool *added) {
2184	return pgc_page_add(cache, &entry, added);
2185	}
2186
2187	ALWAYS_INLINE PGC_PAGE pgc_page_dup(PGC cache, PGC_PAGE *page) {
2188	if(!page_acquire(cache, page))
2189	fatal("DBENGINE CACHE: tried to dup a page that is not acquired!");
2190
2191	return page;
2192	}
2193
2194	ALWAYS_INLINE void pgc_page_release(PGC cache, PGC_PAGE page) {
2195	page_release(cache, page, is_page_clean(page));
2196	}
2197
2198	ALWAYS_INLINE void pgc_page_hot_to_dirty_and_release(PGC cache, PGC_PAGE page, bool never_flush) {
2199	p2_add_fetch(&cache->stats.p2_workers_hot2dirty, 1);
2200
2201	//#ifdef NETDATA_INTERNAL_CHECKS
2202	// page_transition_lock(cache, page);
2203	// internal_fatal(!is_page_hot(page), "DBENGINE CACHE: called %s() but page is not hot", __FUNCTION__ );
2204	// page_transition_unlock(cache, page);
2205	//#endif
2206
2207	// make page dirty
2208	page_set_dirty(cache, page, false, PGC_QUEUE_LOCK_PRIO_COLLECTORS);
2209
2210	// release the page
2211	page_release(cache, page, true);
2212	// page ptr may be invalid now
2213
2214	p2_sub_fetch(&cache->stats.p2_workers_hot2dirty, 1);
2215
2216	// flush, if we have to
2217	if(!never_flush)
2218	flush_on_page_hot_release(cache);
2219	}
2220
2221	bool pgc_page_to_clean_evict_or_release(PGC cache, PGC_PAGE page) {
2222	bool ret;
2223
2224	p2_add_fetch(&cache->stats.p2_workers_hot2dirty, 1);
2225
2226	// prevent accesses from increasing the accesses counter
2227	page_flag_set(page, PGC_PAGE_HAS_NO_DATA_IGNORE_ACCESSES);
2228
2229	// zero the accesses counter
2230	__atomic_store_n(&page->accesses, 0, __ATOMIC_RELEASE);
2231
2232	// if there are no other references to it, evict it immediately
2233	if(make_acquired_page_clean_and_evict_or_page_release(cache, page)) {
2234	__atomic_add_fetch(&cache->stats.hot_empty_pages_evicted_immediately, 1, __ATOMIC_RELAXED);
2235	ret = true;
2236	}
2237	else {
2238	__atomic_add_fetch(&cache->stats.hot_empty_pages_evicted_later, 1, __ATOMIC_RELAXED);
2239	ret = false;
2240	}
2241
2242	p2_sub_fetch(&cache->stats.p2_workers_hot2dirty, 1);
2243
2244	return ret;
2245	}
2246
2247	Word_t pgc_page_section(PGC_PAGE *page) {
2248	return page->section;
2249	}
2250
2251	Word_t pgc_page_metric(PGC_PAGE *page) {
2252	return page->metric_id;
2253	}
2254
2255	time_t pgc_page_start_time_s(PGC_PAGE *page) {
2256	return page->start_time_s;
2257	}
2258
2259	time_t pgc_page_end_time_s(PGC_PAGE *page) {
2260	return page->end_time_s;
2261	}
2262
2263	uint32_t pgc_page_update_every_s(PGC_PAGE *page) {
2264	return page->update_every_s;
2265	}
2266
2267	uint32_t pgc_page_fix_update_every(PGC_PAGE *page, uint32_t update_every_s) {
2268	if(page->update_every_s == 0)
2269	page->update_every_s = update_every_s;
2270
2271	return page->update_every_s;
2272	}
2273
2274	time_t pgc_page_fix_end_time_s(PGC_PAGE *page, time_t end_time_s) {
2275	page->end_time_s = end_time_s;
2276	return page->end_time_s;
2277	}
2278
2279	void pgc_page_data(PGC_PAGE page) {
2280	return page->data;
2281	}
2282
2283	void pgc_page_custom_data(PGC cache, PGC_PAGE *page) {
2284	if(cache->config.additional_bytes_per_page)
2285	return page->custom_data;
2286
2287	return NULL;
2288	}
2289
2290	size_t pgc_page_data_size(PGC cache, PGC_PAGE page) {
2291	return page_size_from_assumed_size(cache, page->assumed_size);
2292	}
2293
2294	bool pgc_is_page_hot(PGC_PAGE *page) {
2295	return is_page_hot(page);
2296	}
2297
2298	bool pgc_is_page_dirty(PGC_PAGE *page) {
2299	return is_page_dirty(page);
2300	}
2301
2302	bool pgc_is_page_clean(PGC_PAGE *page) {
2303	return is_page_clean(page);
2304	}
2305
2306	void pgc_reset_hot_max(PGC *cache) {
2307	size_t entries = __atomic_load_n(&cache->hot.stats->entries, __ATOMIC_RELAXED);
2308	int64_t size = __atomic_load_n(&cache->hot.stats->size, __ATOMIC_RELAXED);
2309
2310	__atomic_store_n(&cache->hot.stats->max_entries, entries, __ATOMIC_RELAXED);
2311	__atomic_store_n(&cache->hot.stats->max_size, size, __ATOMIC_RELAXED);
2312
2313	int64_t size_to_evict = 0;
2314	cache_usage_per1000(cache, &size_to_evict);
2315	evict_pages(cache, 0, 0, true, false);
2316	}
2317
2318	void pgc_set_dynamic_target_cache_size_callback(PGC *cache, dynamic_target_cache_size_callback callback) {
2319	cache->config.dynamic_target_size_cb = callback;
2320	cache->config.out_of_memory_protection_bytes = 0;
2321	cache->config.use_all_ram = false;
2322
2323	int64_t size_to_evict = 0;
2324	cache_usage_per1000(cache, &size_to_evict);
2325	evict_pages(cache, 0, 0, true, false);
2326	}
2327
2328	void pgc_set_nominal_page_size_callback(PGC *cache, nominal_page_size_callback callback) {
2329	cache->config.nominal_page_size_cb = callback;
2330	}
2331
2332	int64_t pgc_get_current_cache_size(PGC *cache) {
2333	return __atomic_load_n(&cache->stats.current_cache_size, __ATOMIC_RELAXED);
2334	}
2335
2336	int64_t pgc_get_wanted_cache_size(PGC *cache) {
2337	return __atomic_load_n(&cache->stats.wanted_cache_size, __ATOMIC_RELAXED);
2338	}
2339
2340	bool pgc_evict_pages(PGC *cache, size_t max_skip, size_t max_evict) {
2341	bool under_pressure = cache_needs_space_aggressively(cache);
2342	return evict_pages(cache,
2343	under_pressure ? 0 : max_skip,
2344	under_pressure ? 0 : max_evict,
2345	true, false);
2346	}
2347
2348	bool pgc_flush_pages(PGC *cache) {
2349	return flush_pages(cache, 0, PGC_SECTION_ALL, true, false);
2350	}
2351
2352	void pgc_page_hot_set_end_time_s(PGC cache __maybe_unused, PGC_PAGE page, time_t end_time_s, size_t additional_bytes) {
2353	internal_fatal(!is_page_hot(page) && !exit_initiated_get(),
2354	"DBENGINE CACHE: end_time_s update on non-hot page");
2355
2356	internal_fatal(end_time_s < __atomic_load_n(&page->end_time_s, __ATOMIC_RELAXED),
2357	"DBENGINE CACHE: end_time_s is not bigger than existing");
2358
2359	__atomic_store_n(&page->end_time_s, end_time_s, __ATOMIC_RELAXED);
2360
2361	if(additional_bytes) {
2362	page_transition_lock(cache, page);
2363
2364	struct pgc_queue_statistics *queue_stats = NULL;
2365	if(page->flags & PGC_PAGE_HOT)
2366	queue_stats = cache->hot.stats;
2367	else if(page->flags & PGC_PAGE_DIRTY)
2368	queue_stats = cache->dirty.stats;
2369	else if(page->flags & PGC_PAGE_CLEAN)
2370	queue_stats = cache->clean.stats;
2371
2372	if(queue_stats && cache->config.stats)
2373	pgc_size_histogram_del(cache, &queue_stats->size_histogram, page);
2374
2375	int64_t old_assumed_size = page->assumed_size;
2376
2377	size_t old_size = page_size_from_assumed_size(cache, old_assumed_size);
2378	size_t size = old_size + additional_bytes;
2379	page->assumed_size = page_assumed_size(cache, size);
2380
2381	int64_t delta = page->assumed_size - old_assumed_size;
2382	__atomic_add_fetch(&cache->stats.size, delta, __ATOMIC_RELAXED);
2383	__atomic_add_fetch(&cache->stats.added_size, delta, __ATOMIC_RELAXED);
2384	__atomic_add_fetch(&cache->stats.referenced_size, delta, __ATOMIC_RELAXED);
2385
2386	if(queue_stats) {
2387	__atomic_add_fetch(&queue_stats->size, delta, __ATOMIC_RELAXED);
2388	__atomic_add_fetch(&queue_stats->added_size, delta, __ATOMIC_RELAXED);
2389
2390	if(cache->config.stats)
2391	pgc_size_histogram_add(cache, &queue_stats->size_histogram, page);
2392	}
2393
2394	page_transition_unlock(cache, page);
2395	}
2396
2397	#ifdef PGC_COUNT_POINTS_COLLECTED
2398	__atomic_add_fetch(&cache->stats.points_collected, 1, __ATOMIC_RELAXED);
2399	#endif
2400	}
2401
2402	PGC_PAGE pgc_page_get_and_acquire(PGC cache, Word_t section, Word_t metric_id, time_t start_time_s, PGC_SEARCH method) {
2403	PGC_PAGE *page = NULL;
2404
2405	p2_add_fetch(&cache->stats.p2_workers_search, 1);
2406
2407	size_t stats_hit_ptr, stats_miss_ptr;
2408
2409	if(method == PGC_SEARCH_CLOSEST) {
2410	__atomic_add_fetch(&cache->stats.searches_closest, 1, __ATOMIC_RELAXED);
2411	stats_hit_ptr = &cache->stats.searches_closest_hits;
2412	stats_miss_ptr = &cache->stats.searches_closest_misses;
2413	}
2414	else {
2415	__atomic_add_fetch(&cache->stats.searches_exact, 1, __ATOMIC_RELAXED);
2416	stats_hit_ptr = &cache->stats.searches_exact_hits;
2417	stats_miss_ptr = &cache->stats.searches_exact_misses;
2418	}
2419
2420	page = page_find_and_acquire_once(cache, section, metric_id, start_time_s, method);
2421	if(page) {
2422	__atomic_add_fetch(stats_hit_ptr, 1, __ATOMIC_RELAXED);
2423	page_has_been_accessed(cache, page);
2424	}
2425	else
2426	__atomic_add_fetch(stats_miss_ptr, 1, __ATOMIC_RELAXED);
2427
2428	p2_sub_fetch(&cache->stats.p2_workers_search, 1);
2429
2430	return page;
2431	}
2432
2433	struct pgc_statistics pgc_get_statistics(PGC *cache) {
2434	// FIXME - get the statistics atomically
2435	return cache->stats;
2436	}
2437
2438	size_t pgc_hot_and_dirty_entries(PGC *cache) {
2439	size_t entries = 0;
2440
2441	entries += __atomic_load_n(&cache->hot.stats->entries, __ATOMIC_RELAXED);
2442	entries += __atomic_load_n(&cache->dirty.stats->entries, __ATOMIC_RELAXED);
2443	entries += __atomic_load_n(&cache->stats.flushing_entries, __ATOMIC_RELAXED);
2444	entries += __atomic_load_n(&cache->stats.hot2dirty_entries, __ATOMIC_RELAXED);
2445
2446	return entries;
2447	}
2448
2449	void pgc_open_cache_to_journal_v2(
2450	PGC *cache,
2451	Word_t section,
2452	unsigned datafile_fileno,
2453	uint8_t type,
2454	migrate_to_v2_callback cb,
2455	void *data,
2456	bool startup)
2457	{
2458	__atomic_add_fetch(&rrdeng_cache_efficiency_stats.journal_v2_indexing_started, 1, __ATOMIC_RELAXED);
2459	p2_add_fetch(&cache->stats.p2_workers_jv2_flush, 1);
2460
2461	pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_LOW);
2462
2463	Pvoid_t JudyL_metrics = NULL;
2464	Pvoid_t JudyL_extents_pos = NULL;
2465
2466	size_t count_of_unique_extents = 0;
2467	size_t count_of_unique_metrics = 0;
2468	size_t count_of_unique_pages = 0;
2469
2470	size_t master_extent_index_id = 0;
2471
2472	Pvoid_t *section_pages_pptr = JudyLGet(cache->hot.sections_judy, section, PJE0);
2473	if(!section_pages_pptr) {
2474	pgc_queue_unlock(cache, &cache->hot);
2475	return;
2476	}
2477
2478	struct section_pages sp = section_pages_pptr;
2479	if(!spinlock_trylock(&sp->migration_to_v2_spinlock)) {
2480	netdata_log_info("DBENGINE: migration to journal v2 for datafile %u (section %" PRIu64 ") is postponed, another jv2 indexer is already running for this section",
2481	datafile_fileno, (uint64_t)section);
2482	pgc_queue_unlock(cache, &cache->hot);
2483	return;
2484	}
2485
2486	ARAL *ar_mi = aral_by_size_acquire(sizeof(struct jv2_metrics_info));
2487	ARAL *ar_pi = aral_by_size_acquire(sizeof(struct jv2_page_info));
2488	ARAL *ar_ei = aral_by_size_acquire(sizeof(struct jv2_extents_info));
2489
2490	for(PGC_PAGE *page = sp->base; page ; page = page->link.next) {
2491	struct extent_io_data xio = (struct extent_io_data )page->custom_data;
2492	if(xio->fileno != datafile_fileno) continue;
2493
2494	if(page_flag_check(page, PGC_PAGE_IS_BEING_MIGRATED_TO_V2)) {
2495	internal_fatal(true, "Migration to journal v2: page has already been migrated to v2");
2496	continue;
2497	}
2498
2499	if(!page_transition_trylock(cache, page)) {
2500	internal_fatal(true, "Migration to journal v2: cannot get page transition lock");
2501	continue;
2502	}
2503
2504	if(!page_acquire(cache, page)) {
2505	internal_fatal(true, "Migration to journal v2: cannot acquire page for migration to v2");
2506	page_transition_unlock(cache, page);
2507	continue;
2508	}
2509
2510	METRIC metric = mrg_metric_dup(main_mrg, (METRIC )page->metric_id);
2511	if(!metric) {
2512	// metric has been deleted, skip this page
2513	page_transition_unlock(cache, page);
2514	page_release(cache, page, false);
2515	continue;
2516	}
2517
2518	// Check UUID validity early, before any JudyL modifications
2519	nd_uuid_t *uuid = mrg_metric_uuid(main_mrg, metric);
2520	if (unlikely(!uuid)) {
2521	mrg_metric_release(main_mrg, metric);
2522	page_transition_unlock(cache, page);
2523	page_release(cache, page, false);
2524	continue;
2525	}
2526
2527	page_flag_set(page, PGC_PAGE_IS_BEING_MIGRATED_TO_V2);
2528
2529	pgc_queue_unlock(cache, &cache->hot);
2530
2531	// update the extents JudyL
2532
2533	size_t current_extent_index_id;
2534	Pvoid_t *PValue = JudyLIns(&JudyL_extents_pos, xio->block, PJE0);
2535	if(!PValue \|\| PValue == PJERR)
2536	fatal("CACHE: JudyLIns(JudyL_extents_pos, %" PRIu64 ") failed, JudyL_extents_pos = %p, result = %p",
2537	BLOCK_TO_OFFSET(xio->block), JudyL_extents_pos, PValue);
2538
2539	struct jv2_extents_info *ei;
2540	if(!*PValue) {
2541	ei = aral_mallocz(ar_ei); // callocz(1, sizeof(struct jv2_extents_info));
2542	ei->block = xio->block;
2543	ei->bytes = xio->bytes;
2544	ei->number_of_pages = 1;
2545	ei->index = master_extent_index_id++;
2546	*PValue = ei;
2547
2548	count_of_unique_extents++;
2549	}
2550	else {
2551	ei = *PValue;
2552	ei->number_of_pages++;
2553	}
2554
2555	current_extent_index_id = ei->index;
2556
2557	// update the metrics JudyL
2558
2559	PValue = JudyLIns(&JudyL_metrics, page->metric_id, PJE0);
2560	if(!PValue \|\| PValue == PJERR)
2561	fatal("CACHE: JudyLIns(JudyL_metrics, 0x%lx) failed, JudyL_metrics = %p, result = %p",
2562	(long unsigned)page->metric_id, JudyL_metrics, PValue);
2563
2564	struct jv2_metrics_info *mi;
2565	if(!*PValue) {
2566	mi = aral_mallocz(ar_mi);
2567	mi->metric = metric;
2568	mi->uuid = uuid;
2569	mi->first_time_s = page->start_time_s;
2570	mi->last_time_s = page->end_time_s;
2571	mi->number_of_pages = 1;
2572	mi->page_list_header = 0;
2573	mi->JudyL_pages_by_start_time = NULL;
2574	*PValue = mi;
2575
2576	count_of_unique_metrics++;
2577	}
2578	else {
2579	mi = *PValue;
2580	mi->number_of_pages++;
2581	mrg_metric_release(main_mrg, metric);
2582	if(page->start_time_s < mi->first_time_s)
2583	mi->first_time_s = page->start_time_s;
2584	if(page->end_time_s > mi->last_time_s)
2585	mi->last_time_s = page->end_time_s;
2586	}
2587
2588	PValue = JudyLIns(&mi->JudyL_pages_by_start_time, page->start_time_s, PJE0);
2589	if(!PValue \|\| PValue == PJERR)
2590	fatal("CACHE: JudyLIns(JudyL_pages_by_start_time, %ld) failed, JudyL_pages_by_start_time = %p, result = %p",
2591	(long)page->start_time_s, mi->JudyL_pages_by_start_time, PValue);
2592
2593	if(!*PValue) {
2594	struct jv2_page_info *pi = aral_mallocz(ar_pi);
2595	pi->start_time_s = page->start_time_s;
2596	pi->end_time_s = page->end_time_s;
2597	pi->update_every_s = page->update_every_s;
2598	pi->page_length = page_size_from_assumed_size(cache, page->assumed_size);
2599	pi->page = page;
2600	pi->extent_index = current_extent_index_id;
2601	pi->custom_data = (cache->config.additional_bytes_per_page) ? page->custom_data : NULL;
2602	*PValue = pi;
2603
2604	count_of_unique_pages++;
2605	}
2606	else {
2607	// impossible situation
2608	internal_fatal(true, "Page is already in JudyL metric pages");
2609	page_flag_clear(page, PGC_PAGE_IS_BEING_MIGRATED_TO_V2);
2610	page_transition_unlock(cache, page);
2611	page_release(cache, page, false);
2612	}
2613
2614	if (likely(false == startup))
2615	yield_the_processor(); // do not lock too aggressively
2616	pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_LOW);
2617	}
2618
2619	spinlock_unlock(&sp->migration_to_v2_spinlock);
2620	pgc_queue_unlock(cache, &cache->hot);
2621
2622	// callback
2623	bool success = cb(section, datafile_fileno, type, JudyL_metrics, JudyL_extents_pos, count_of_unique_extents, count_of_unique_metrics, count_of_unique_pages, data);
2624
2625	{
2626	Pvoid_t *PValue1;
2627	bool metric_id_first = true;
2628	Word_t metric_id = 0;
2629	while ((PValue1 = JudyLFirstThenNext(JudyL_metrics, &metric_id, &metric_id_first))) {
2630	struct jv2_metrics_info mi = PValue1;
2631
2632	Pvoid_t *PValue2;
2633	bool start_time_first = true;
2634	Word_t start_time = 0;
2635	while ((PValue2 = JudyLFirstThenNext(mi->JudyL_pages_by_start_time, &start_time, &start_time_first))) {
2636	struct jv2_page_info pi = PValue2;
2637
2638	if (likely(false == startup))
2639	yield_the_processor(); // do not lock too aggressively
2640	if (likely(success))
2641	page_set_clean(cache, pi->page, true, false, PGC_QUEUE_LOCK_PRIO_LOW);
2642	else
2643	page_flag_clear(pi->page, PGC_PAGE_IS_BEING_MIGRATED_TO_V2);
2644
2645	page_transition_unlock(cache, pi->page);
2646	page_release(cache, pi->page, success);
2647	// before balance-parents:
2648	// page_transition_unlock(cache, pi->page);
2649	// pgc_page_hot_to_dirty_and_release(cache, pi->page, true);
2650
2651	// old test - don't enable:
2652	// make_acquired_page_clean_and_evict_or_page_release(cache, pi->page);
2653	aral_freez(ar_pi, pi);
2654	}
2655
2656	JudyLFreeArray(&mi->JudyL_pages_by_start_time, PJE0);
2657	mrg_metric_release(main_mrg, mi->metric);
2658	aral_freez(ar_mi, mi);
2659	}
2660	JudyLFreeArray(&JudyL_metrics, PJE0);
2661	}
2662
2663	{
2664	Pvoid_t *PValue;
2665	bool extent_pos_first = true;
2666	Word_t extent_pos = 0;
2667	while ((PValue = JudyLFirstThenNext(JudyL_extents_pos, &extent_pos, &extent_pos_first))) {
2668	struct jv2_extents_info ei = PValue;
2669	aral_freez(ar_ei, ei);
2670	}
2671	JudyLFreeArray(&JudyL_extents_pos, PJE0);
2672	}
2673
2674	aral_by_size_release(ar_ei);
2675	aral_by_size_release(ar_pi);
2676	aral_by_size_release(ar_mi);
2677
2678	p2_sub_fetch(&cache->stats.p2_workers_jv2_flush, 1);
2679
2680	// balance-parents: do not flush, there is nothing dirty
2681	// flush_pages(cache, cache->config.max_flushes_inline, PGC_SECTION_ALL, false, false);
2682	}
2683
2684	static bool match_page_data(PGC_PAGE page, void data) {
2685	return (page->data == data);
2686	}
2687
2688	void pgc_open_evict_clean_pages_of_datafile(PGC cache, struct rrdengine_datafile datafile) {
2689	evict_pages_with_filter(cache, 0, 0, true, true, match_page_data, datafile);
2690	}
2691
2692	size_t pgc_count_clean_pages_having_data_ptr(PGC cache, Word_t section, void ptr) {
2693	size_t found = 0;
2694
2695	pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_LOW);
2696	for(PGC_PAGE *page = cache->clean.base; page ;page = page->link.next)
2697	found += (page->data == ptr && page->section == section) ? 1 : 0;
2698	pgc_queue_unlock(cache, &cache->clean);
2699
2700	return found;
2701	}
2702
2703	size_t pgc_count_hot_pages_having_data_ptr(PGC cache, Word_t section, void ptr) {
2704	size_t found = 0;
2705
2706	pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_LOW);
2707	Pvoid_t *section_pages_pptr = JudyLGet(cache->hot.sections_judy, section, PJE0);
2708	if(section_pages_pptr) {
2709	struct section_pages sp = section_pages_pptr;
2710	for(PGC_PAGE *page = sp->base; page ;page = page->link.next)
2711	found += (page->data == ptr) ? 1 : 0;
2712	}
2713	pgc_queue_unlock(cache, &cache->hot);
2714
2715	return found;
2716	}
2717
2718	// ----------------------------------------------------------------------------
2719	// unittest
2720
2721	static void unittest_free_clean_page_callback(PGC *cache __maybe_unused, PGC_ENTRY entry __maybe_unused) {
2722	;
2723	}
2724
2725	static void unittest_save_dirty_page_callback(PGC cache __maybe_unused, PGC_ENTRY entries_array __maybe_unused, PGC_PAGE **pages_array __maybe_unused, size_t entries __maybe_unused) {
2726	;
2727	}
2728
2729	#ifdef PGC_STRESS_TEST
2730
2731	struct {
2732	bool stop;
2733	PGC *cache;
2734	PGC_PAGE **metrics;
2735	size_t clean_metrics;
2736	size_t hot_metrics;
2737	time_t first_time_t;
2738	time_t last_time_t;
2739	size_t cache_size;
2740	size_t query_threads;
2741	size_t collect_threads;
2742	size_t partitions;
2743	size_t points_per_page;
2744	time_t time_per_collection_ut;
2745	time_t time_per_query_ut;
2746	time_t time_per_flush_ut;
2747	PGC_OPTIONS options;
2748	char rand_statebufs[1024];
2749	struct random_data *random_data;
2750	} pgc_uts = {
2751	.stop = false,
2752	.metrics = NULL,
2753	.clean_metrics = 100000,
2754	.hot_metrics = 1000000,
2755	.first_time_t = 100000000,
2756	.last_time_t = 0,
2757	.cache_size = 0, // get the default (8MB)
2758	.collect_threads = 16,
2759	.query_threads = 16,
2760	.partitions = 0, // get the default (system cpus)
2761	.options = PGC_OPTIONS_AUTOSCALE,/* PGC_OPTIONS_FLUSH_PAGES_INLINE \| PGC_OPTIONS_EVICT_PAGES_INLINE,*/
2762	.points_per_page = 10,
2763	.time_per_collection_ut = 1000000,
2764	.time_per_query_ut = 250,
2765	.time_per_flush_ut = 100,
2766	.rand_statebufs = {},
2767	.random_data = NULL,
2768	};
2769
2770	void unittest_stress_test_collector(void ptr) {
2771	size_t id = ((size_t )ptr);
2772
2773	size_t metric_start = pgc_uts.clean_metrics;
2774	size_t metric_end = pgc_uts.clean_metrics + pgc_uts.hot_metrics;
2775	size_t number_of_metrics = metric_end - metric_start;
2776	size_t per_collector_metrics = number_of_metrics / pgc_uts.collect_threads;
2777	metric_start = metric_start + per_collector_metrics * id + 1;
2778	metric_end = metric_start + per_collector_metrics - 1;
2779
2780	time_t start_time_t = pgc_uts.first_time_t + 1;
2781
2782	heartbeat_t hb;
2783	heartbeat_init(&hb, pgc_uts.time_per_collection_ut);
2784
2785	while(!__atomic_load_n(&pgc_uts.stop, __ATOMIC_RELAXED)) {
2786	// netdata_log_info("COLLECTOR %zu: collecting metrics %zu to %zu, from %ld to %lu", id, metric_start, metric_end, start_time_t, start_time_t + pgc_uts.points_per_page);
2787
2788	for (size_t i = metric_start; i < metric_end; i++) {
2789	bool added;
2790
2791	pgc_uts.metrics[i] = pgc_page_add_and_acquire(pgc_uts.cache, (PGC_ENTRY) {
2792	.section = 1,
2793	.metric_id = i,
2794	.start_time_t = start_time_t,
2795	.end_time_t = start_time_t,
2796	.update_every = 1,
2797	.size = 4096,
2798	.data = NULL,
2799	.hot = true,
2800	}, &added);
2801
2802	if(!pgc_is_page_hot(pgc_uts.metrics[i]) \|\| !added) {
2803	pgc_page_release(pgc_uts.cache, pgc_uts.metrics[i]);
2804	pgc_uts.metrics[i] = NULL;
2805	}
2806	}
2807
2808	time_t end_time_t = start_time_t + (time_t)pgc_uts.points_per_page;
2809	while(++start_time_t <= end_time_t && !__atomic_load_n(&pgc_uts.stop, __ATOMIC_RELAXED)) {
2810	heartbeat_next(&hb);
2811
2812	for (size_t i = metric_start; i < metric_end; i++) {
2813	if(pgc_uts.metrics[i])
2814	pgc_page_hot_set_end_time_t(pgc_uts.cache, pgc_uts.metrics[i], start_time_t);
2815	}
2816
2817	__atomic_store_n(&pgc_uts.last_time_t, start_time_t, __ATOMIC_RELAXED);
2818	}
2819
2820	for (size_t i = metric_start; i < metric_end; i++) {
2821	if (pgc_uts.metrics[i]) {
2822	if(i % 10 == 0)
2823	pgc_page_to_clean_evict_or_release(pgc_uts.cache, pgc_uts.metrics[i]);
2824	else
2825	pgc_page_hot_to_dirty_and_release(pgc_uts.cache, pgc_uts.metrics[i], false);
2826	}
2827	}
2828	}
2829
2830	return ptr;
2831	}
2832
2833	void unittest_stress_test_queries(void ptr) {
2834	size_t id = ((size_t )ptr);
2835	struct random_data *random_data = &pgc_uts.random_data[id];
2836
2837	size_t start = 0;
2838	size_t end = pgc_uts.clean_metrics + pgc_uts.hot_metrics;
2839
2840	while(!__atomic_load_n(&pgc_uts.stop, __ATOMIC_RELAXED)) {
2841	int32_t random_number;
2842	random_r(random_data, &random_number);
2843
2844	size_t metric_id = random_number % (end - start);
2845	time_t start_time_t = pgc_uts.first_time_t;
2846	time_t end_time_t = __atomic_load_n(&pgc_uts.last_time_t, __ATOMIC_RELAXED);
2847	if(end_time_t <= start_time_t)
2848	end_time_t = start_time_t + 1;
2849	size_t pages = (end_time_t - start_time_t) / pgc_uts.points_per_page + 1;
2850
2851	PGC_PAGE *array[pages];
2852	for(size_t i = 0; i < pages ;i++)
2853	array[i] = NULL;
2854
2855	// find the pages the cache has
2856	for(size_t i = 0; i < pages ;i++) {
2857	time_t page_start_time = start_time_t + (time_t)(i * pgc_uts.points_per_page);
2858	array[i] = pgc_page_get_and_acquire(pgc_uts.cache, 1, metric_id,
2859	page_start_time, (i < pages - 1)?PGC_SEARCH_EXACT:PGC_SEARCH_CLOSEST);
2860	}
2861
2862	// load the rest of the pages
2863	for(size_t i = 0; i < pages ;i++) {
2864	if(array[i]) continue;
2865
2866	time_t page_start_time = start_time_t + (time_t)(i * pgc_uts.points_per_page);
2867	array[i] = pgc_page_add_and_acquire(pgc_uts.cache, (PGC_ENTRY) {
2868	.section = 1,
2869	.metric_id = metric_id,
2870	.start_time_t = page_start_time,
2871	.end_time_t = page_start_time + (time_t)pgc_uts.points_per_page,
2872	.update_every = 1,
2873	.size = 4096,
2874	.data = NULL,
2875	.hot = false,
2876	}, NULL);
2877	}
2878
2879	// do the query
2880	// ...
2881	struct timespec work_duration = {.tv_sec = 0, .tv_nsec = pgc_uts.time_per_query_ut * NSEC_PER_USEC };
2882	nanosleep(&work_duration, NULL);
2883
2884	// release the pages
2885	for(size_t i = 0; i < pages ;i++) {
2886	if(!array[i]) continue;
2887	pgc_page_release(pgc_uts.cache, array[i]);
2888	array[i] = NULL;
2889	}
2890	}
2891
2892	return ptr;
2893	}
2894
2895	void unittest_stress_test_service(void ptr) {
2896	heartbeat_t hb;
2897	heartbeat_init(&hb, USEC_PER_SEC);
2898	while(!__atomic_load_n(&pgc_uts.stop, __ATOMIC_RELAXED)) {
2899	heartbeat_next(&hb);
2900
2901	pgc_flush_pages(pgc_uts.cache, 1000);
2902	pgc_evict_pages(pgc_uts.cache, 0, 0);
2903	}
2904	return ptr;
2905	}
2906
2907	static void unittest_stress_test_save_dirty_page_callback(PGC cache __maybe_unused, PGC_ENTRY entries_array __maybe_unused, PGC_PAGE **pages_array __maybe_unused, size_t entries __maybe_unused) {
2908	// netdata_log_info("SAVE %zu pages", entries);
2909	if(!pgc_uts.stop) {
2910	usec_t t = pgc_uts.time_per_flush_ut;
2911
2912	if(t > 0) {
2913	struct timespec work_duration = {
2914	.tv_sec = t / USEC_PER_SEC,
2915	.tv_nsec = (long) ((t % USEC_PER_SEC) * NSEC_PER_USEC)
2916	};
2917
2918	nanosleep(&work_duration, NULL);
2919	}
2920	}
2921	}
2922
2923	void unittest_stress_test(void) {
2924	pgc_uts.cache = pgc_create(pgc_uts.cache_size * 1024 * 1024,
2925	unittest_free_clean_page_callback,
2926	64, unittest_stress_test_save_dirty_page_callback,
2927	1000, 10000, 1,
2928	pgc_uts.options, pgc_uts.partitions, 0);
2929
2930	pgc_uts.metrics = callocz(pgc_uts.clean_metrics + pgc_uts.hot_metrics, sizeof(PGC_PAGE *));
2931
2932	pthread_t service_thread;
2933	nd_thread_create(&service_thread, "SERVICE",
2934	NETDATA_THREAD_OPTION_DONT_LOG,
2935	unittest_stress_test_service, NULL);
2936
2937	pthread_t collect_threads[pgc_uts.collect_threads];
2938	size_t collect_thread_ids[pgc_uts.collect_threads];
2939	for(size_t i = 0; i < pgc_uts.collect_threads ;i++) {
2940	collect_thread_ids[i] = i;
2941	char buffer[100 + 1];
2942	snprintfz(buffer, sizeof(buffer) - 1, "COLLECT_%zu", i);
2943	nd_thread_create(&collect_threads[i], buffer,
2944	NETDATA_THREAD_OPTION_DONT_LOG,
2945	unittest_stress_test_collector, &collect_thread_ids[i]);
2946	}
2947
2948	pthread_t queries_threads[pgc_uts.query_threads];
2949	size_t query_thread_ids[pgc_uts.query_threads];
2950	pgc_uts.random_data = callocz(pgc_uts.query_threads, sizeof(struct random_data));
2951	for(size_t i = 0; i < pgc_uts.query_threads ;i++) {
2952	query_thread_ids[i] = i;
2953	char buffer[100 + 1];
2954	snprintfz(buffer, sizeof(buffer) - 1, "QUERY_%zu", i);
2955	initstate_r(1, pgc_uts.rand_statebufs, 1024, &pgc_uts.random_data[i]);
2956	nd_thread_create(&queries_threads[i], buffer,
2957	NETDATA_THREAD_OPTION_DONT_LOG,
2958	unittest_stress_test_queries, &query_thread_ids[i]);
2959	}
2960
2961	heartbeat_t hb;
2962	heartbeat_init(&hb, USEC_PER_SEC);
2963
2964	struct {
2965	size_t entries;
2966	size_t added;
2967	size_t deleted;
2968	size_t referenced;
2969
2970	size_t hot_entries;
2971	size_t hot_added;
2972	size_t hot_deleted;
2973
2974	size_t dirty_entries;
2975	size_t dirty_added;
2976	size_t dirty_deleted;
2977
2978	size_t clean_entries;
2979	size_t clean_added;
2980	size_t clean_deleted;
2981
2982	size_t searches_exact;
2983	size_t searches_exact_hits;
2984	size_t searches_closest;
2985	size_t searches_closest_hits;
2986
2987	size_t collections;
2988
2989	size_t events_cache_under_severe_pressure;
2990	size_t events_cache_needs_space_90;
2991	size_t events_flush_critical;
2992	} stats = {}, old_stats = {};
2993
2994	for(int i = 0; i < 86400 ;i++) {
2995	heartbeat_next(&hb);
2996
2997	old_stats = stats;
2998	stats.entries = __atomic_load_n(&pgc_uts.cache->stats.entries, __ATOMIC_RELAXED);
2999	stats.added = __atomic_load_n(&pgc_uts.cache->stats.added_entries, __ATOMIC_RELAXED);
3000	stats.deleted = __atomic_load_n(&pgc_uts.cache->stats.removed_entries, __ATOMIC_RELAXED);
3001	stats.referenced = __atomic_load_n(&pgc_uts.cache->stats.referenced_entries, __ATOMIC_RELAXED);
3002
3003	stats.hot_entries = __atomic_load_n(&pgc_uts.cache->hot.stats->entries, __ATOMIC_RELAXED);
3004	stats.hot_added = __atomic_load_n(&pgc_uts.cache->hot.stats->added_entries, __ATOMIC_RELAXED);
3005	stats.hot_deleted = __atomic_load_n(&pgc_uts.cache->hot.stats->removed_entries, __ATOMIC_RELAXED);
3006
3007	stats.dirty_entries = __atomic_load_n(&pgc_uts.cache->dirty.stats->entries, __ATOMIC_RELAXED);
3008	stats.dirty_added = __atomic_load_n(&pgc_uts.cache->dirty.stats->added_entries, __ATOMIC_RELAXED);
3009	stats.dirty_deleted = __atomic_load_n(&pgc_uts.cache->dirty.stats->removed_entries, __ATOMIC_RELAXED);
3010
3011	stats.clean_entries = __atomic_load_n(&pgc_uts.cache->clean.stats->entries, __ATOMIC_RELAXED);
3012	stats.clean_added = __atomic_load_n(&pgc_uts.cache->clean.stats->added_entries, __ATOMIC_RELAXED);
3013	stats.clean_deleted = __atomic_load_n(&pgc_uts.cache->clean.stats->removed_entries, __ATOMIC_RELAXED);
3014
3015	stats.searches_exact = __atomic_load_n(&pgc_uts.cache->stats.searches_exact, __ATOMIC_RELAXED);
3016	stats.searches_exact_hits = __atomic_load_n(&pgc_uts.cache->stats.searches_exact_hits, __ATOMIC_RELAXED);
3017
3018	stats.searches_closest = __atomic_load_n(&pgc_uts.cache->stats.searches_closest, __ATOMIC_RELAXED);
3019	stats.searches_closest_hits = __atomic_load_n(&pgc_uts.cache->stats.searches_closest_hits, __ATOMIC_RELAXED);
3020
3021	stats.events_cache_under_severe_pressure = __atomic_load_n(&pgc_uts.cache->stats.events_cache_under_severe_pressure, __ATOMIC_RELAXED);
3022	stats.events_cache_needs_space_90 = __atomic_load_n(&pgc_uts.cache->stats.events_cache_needs_space_aggressively, __ATOMIC_RELAXED);
3023	stats.events_flush_critical = __atomic_load_n(&pgc_uts.cache->stats.events_flush_critical, __ATOMIC_RELAXED);
3024
3025	size_t searches_exact = stats.searches_exact - old_stats.searches_exact;
3026	size_t searches_closest = stats.searches_closest - old_stats.searches_closest;
3027
3028	size_t hit_exact = stats.searches_exact_hits - old_stats.searches_exact_hits;
3029	size_t hit_closest = stats.searches_closest_hits - old_stats.searches_closest_hits;
3030
3031	double hit_exact_pc = (searches_exact > 0) ? (double)hit_exact * 100.0 / (double)searches_exact : 0.0;
3032	double hit_closest_pc = (searches_closest > 0) ? (double)hit_closest * 100.0 / (double)searches_closest : 0.0;
3033
3034	#ifdef PGC_COUNT_POINTS_COLLECTED
3035	stats.collections = __atomic_load_n(&pgc_uts.cache->stats.points_collected, __ATOMIC_RELAXED);
3036	#endif
3037
3038	char *cache_status = "N";
3039	if(stats.events_cache_under_severe_pressure > old_stats.events_cache_under_severe_pressure)
3040	cache_status = "F";
3041	else if(stats.events_cache_needs_space_90 > old_stats.events_cache_needs_space_90)
3042	cache_status = "f";
3043
3044	char *flushing_status = "N";
3045	if(stats.events_flush_critical > old_stats.events_flush_critical)
3046	flushing_status = "F";
3047
3048	netdata_log_info("PGS %5zuk +%4zuk/-%4zuk "
3049	"\| RF %5zuk "
3050	"\| HOT %5zuk +%4zuk -%4zuk "
3051	"\| DRT %s %5zuk +%4zuk -%4zuk "
3052	"\| CLN %s %5zuk +%4zuk -%4zuk "
3053	"\| SRCH %4zuk %4zuk, HIT %4.1f%% %4.1f%% "
3054	#ifdef PGC_COUNT_POINTS_COLLECTED
3055	"\| CLCT %8.4f Mps"
3056	#endif
3057	, stats.entries / 1000
3058	, (stats.added - old_stats.added) / 1000, (stats.deleted - old_stats.deleted) / 1000
3059	, stats.referenced / 1000
3060	, stats.hot_entries / 1000, (stats.hot_added - old_stats.hot_added) / 1000, (stats.hot_deleted - old_stats.hot_deleted) / 1000
3061	, flushing_status
3062	, stats.dirty_entries / 1000
3063	, (stats.dirty_added - old_stats.dirty_added) / 1000, (stats.dirty_deleted - old_stats.dirty_deleted) / 1000
3064	, cache_status
3065	, stats.clean_entries / 1000
3066	, (stats.clean_added - old_stats.clean_added) / 1000, (stats.clean_deleted - old_stats.clean_deleted) / 1000
3067	, searches_exact / 1000, searches_closest / 1000
3068	, hit_exact_pc, hit_closest_pc
3069	#ifdef PGC_COUNT_POINTS_COLLECTED
3070	, (double)(stats.collections - old_stats.collections) / 1000.0 / 1000.0
3071	#endif
3072	);
3073	}
3074	netdata_log_info("Waiting for threads to stop...");
3075	__atomic_store_n(&pgc_uts.stop, true, __ATOMIC_RELAXED);
3076
3077	nd_thread_join(service_thread, NULL);
3078
3079	for(size_t i = 0; i < pgc_uts.collect_threads ;i++)
3080	nd_thread_join(collect_threads[i],NULL);
3081
3082	for(size_t i = 0; i < pgc_uts.query_threads ;i++)
3083	nd_thread_join(queries_threads[i],NULL);
3084
3085	pgc_destroy(pgc_uts.cache);
3086
3087	freez(pgc_uts.metrics);
3088	freez(pgc_uts.random_data);
3089	}
3090	#endif
3091
3092	int pgc_unittest(void) {
3093	PGC *cache = pgc_create("test",
3094	32 * 1024 * 1024, unittest_free_clean_page_callback,
3095	64, NULL, unittest_save_dirty_page_callback,
3096	10, 10, 1000, 10,
3097	PGC_OPTIONS_DEFAULT, 1, 11);
3098
3099	// FIXME - unit tests
3100	// - add clean page
3101	// - add clean page again (should not add it)
3102	// - release page (should decrement counters)
3103	// - add hot page
3104	// - add hot page again (should not add it)
3105	// - turn hot page to dirty, with and without a reference counter to it
3106	// - dirty pages are saved once there are enough of them
3107	// - find page exact
3108	// - find page (should return last)
3109	// - find page (should return next)
3110	// - page cache full (should evict)
3111	// - on destroy, turn hot pages to dirty and save them
3112
3113	PGC_PAGE *page1 = pgc_page_add_and_acquire(cache, (PGC_ENTRY){
3114	.section = 1,
3115	.metric_id = 10,
3116	.start_time_s = 100,
3117	.end_time_s = 1000,
3118	.size = 4096,
3119	.data = NULL,
3120	.hot = false,
3121	.custom_data = (uint8_t *)"0123456789",
3122	}, NULL);
3123
3124	if(strcmp(pgc_page_custom_data(cache, page1), "0123456789") != 0)
3125	fatal("custom data do not work");
3126
3127	memcpy(pgc_page_custom_data(cache, page1), "ABCDEFGHIJ", 11);
3128	if(strcmp(pgc_page_custom_data(cache, page1), "ABCDEFGHIJ") != 0)
3129	fatal("custom data do not work");
3130
3131	pgc_page_release(cache, page1);
3132
3133	PGC_PAGE *page2 = pgc_page_add_and_acquire(cache, (PGC_ENTRY){
3134	.section = 2,
3135	.metric_id = 10,
3136	.start_time_s = 1001,
3137	.end_time_s = 2000,
3138	.size = 4096,
3139	.data = NULL,
3140	.hot = true,
3141	}, NULL);
3142
3143	pgc_page_hot_set_end_time_s(cache, page2, 2001, 0);
3144	pgc_page_hot_to_dirty_and_release(cache, page2, false);
3145
3146	PGC_PAGE *page3 = pgc_page_add_and_acquire(cache, (PGC_ENTRY){
3147	.section = 3,
3148	.metric_id = 10,
3149	.start_time_s = 1001,
3150	.end_time_s = 2000,
3151	.size = 4096,
3152	.data = NULL,
3153	.hot = true,
3154	}, NULL);
3155
3156	pgc_page_hot_set_end_time_s(cache, page3, 2001, 0);
3157	pgc_page_hot_to_dirty_and_release(cache, page3, false);
3158
3159	pgc_destroy(cache, true);
3160
3161	{
3162	PGC *cache_a = pgc_create("partition-cache-a",
3163	32 * 1024 * 1024, unittest_free_clean_page_callback,
3164	64, NULL, unittest_save_dirty_page_callback,
3165	10, 10, 1000, 10,
3166	PGC_OPTIONS_DEFAULT, 4, 0);
3167	PGC *cache_b = pgc_create("partition-cache-b",
3168	32 * 1024 * 1024, unittest_free_clean_page_callback,
3169	64, NULL, unittest_save_dirty_page_callback,
3170	10, 10, 1000, 10,
3171	PGC_OPTIONS_DEFAULT, 5, 0);
3172
3173	Word_t metric_id = 5;
3174	size_t partition_a = pgc_indexing_partition(cache_a, metric_id);
3175	size_t partition_b = pgc_indexing_partition(cache_b, metric_id);
3176
3177	if(partition_a != indexing_partition(metric_id, cache_a->config.partitions))
3178	fatal("pgc_indexing_partition() returned the wrong partition for cache_a");
3179
3180	if(partition_b != indexing_partition(metric_id, cache_b->config.partitions))
3181	fatal("pgc_indexing_partition() returned the wrong partition for cache_b");
3182
3183	pgc_destroy(cache_a, true);
3184	pgc_destroy(cache_b, true);
3185	}
3186
3187	#ifdef PGC_STRESS_TEST
3188	unittest_stress_test();
3189	#endif
3190
3191	return 0;
3192	}