@cryptotaxi247 / netdata-1 / commits / 7279dd092

DBENGINE v2 - improvements part 3 (#14269)

* reduce journal v2 shared memory using madvise() - not integrated yet * working attempt to minimize dbengine shared memory * never call willneed - let the kernel decide which parts of each file are really needed * journal files get MADV_RANDOM * dont call MADV_DONTNEED too frequently * madvise() is always called with the journal unlocked but referenced * call madvise() even less frequently * added chart for monitoring database events * turn batch mode on under critical conditions * max size to evict is 1/4 of the max * fix max size to evict calculation * use dbengine_page/extent_alloc/free to pages and extents allocations, tracking also the size of these allocations at free time * fix calculation for batch evictions * allow main and open cache to have as many evictors as needed * control inline evictors for each cache; report different levels of cache pressure on every cache evaluation * more inline evictors for extent cache * bypass max inline evictors above critical level * current cache usage has to be taken * re-arrange items in journafile * updated docs - work in progress * more docs work * more docs work * Map / unmap journal file * draw.io diagram for dbengine operations * updated dbengine diagram * updated docs * journal files v2 now get mapped and unmapped as needed * unmap journal v2 immediately when getting retention * mmap and munmap do not block queries evaluating journal files v2 * have only one unmap function Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Jan 17, 2023 at 19:35 UTC 7279dd092c23fbafcd7edb8ef7f3f79e1a0e5ecb
20 files changed +822 -305
collectors/proc.plugin/proc_net_netstat.c
+1 -1
@@ -356,7 +356,7 @@ static void do_proc_net_snmp6(int update_every) {
356 if (unlikely(!ff_snmp6))
357 return;
358
359 - size_t lines, l, words;
359 + size_t lines, l;
360
361 lines = procfile_lines(ff_snmp6);
362
daemon/global_statistics.c
+46
@@ -2076,6 +2076,52 @@ static void dbengine2_statistics_charts(void) {
2076 rrdset_done(st_query_pages_from_disk);
2077 }
2078
2079 + {
2080 + static RRDSET *st_events = NULL;
2081 + static RRDDIM *rd_journal_v2_mapped = NULL;
2082 + static RRDDIM *rd_journal_v2_unmapped = NULL;
2083 + static RRDDIM *rd_datafile_creation = NULL;
2084 + static RRDDIM *rd_datafile_deletion = NULL;
2085 + static RRDDIM *rd_datafile_deletion_spin = NULL;
2086 + static RRDDIM *rd_jv2_indexing = NULL;
2087 + static RRDDIM *rd_retention = NULL;
2088 +
2089 + if (unlikely(!st_events)) {
2090 + st_events = rrdset_create_localhost(
2091 + "netdata",
2092 + "dbengine_events",
2093 + NULL,
2094 + "dbengine query router",
2095 + NULL,
2096 + "Netdata Database Events",
2097 + "events/s",
2098 + "netdata",
2099 + "stats",
2100 + priority,
2101 + localhost->rrd_update_every,
2102 + RRDSET_TYPE_LINE);
2103 +
2104 + rd_journal_v2_mapped = rrddim_add(st_events, "journal v2 mapped", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2105 + rd_journal_v2_unmapped = rrddim_add(st_events, "journal v2 unmapped", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2106 + rd_datafile_creation = rrddim_add(st_events, "datafile creation", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2107 + rd_datafile_deletion = rrddim_add(st_events, "datafile deletion", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2108 + rd_datafile_deletion_spin = rrddim_add(st_events, "datafile deletion spin", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2109 + rd_jv2_indexing = rrddim_add(st_events, "journal v2 indexing", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2110 + rd_retention = rrddim_add(st_events, "retention", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
2111 + }
2112 + priority++;
2113 +
2114 + rrddim_set_by_pointer(st_events, rd_journal_v2_mapped, (collected_number)cache_efficiency_stats.journal_v2_mapped);
2115 + rrddim_set_by_pointer(st_events, rd_journal_v2_unmapped, (collected_number)cache_efficiency_stats.journal_v2_unmapped);
2116 + rrddim_set_by_pointer(st_events, rd_datafile_creation, (collected_number)cache_efficiency_stats.datafile_creation_started);
2117 + rrddim_set_by_pointer(st_events, rd_datafile_deletion, (collected_number)cache_efficiency_stats.datafile_deletion_started);
2118 + rrddim_set_by_pointer(st_events, rd_datafile_deletion_spin, (collected_number)cache_efficiency_stats.datafile_deletion_spin);
2119 + rrddim_set_by_pointer(st_events, rd_jv2_indexing, (collected_number)cache_efficiency_stats.journal_v2_indexing_started);
2120 + rrddim_set_by_pointer(st_events, rd_retention, (collected_number)cache_efficiency_stats.metrics_retention_started);
2121 +
2122 + rrdset_done(st_events);
2123 + }
2124 +
2125 {
2126 static RRDSET *st_prep_timings = NULL;
2127 static RRDDIM *rd_routing = NULL;
database/engine/README.md
+231 -60
@@ -4,45 +4,119 @@ description: "Netdata's highly-efficient database engine use both RAM and disk f
4 custom_edit_url: https://github.com/netdata/netdata/edit/master/database/engine/README.md
5 -->
6
7 -# Database engine
7 +# DBENGINE
8
9 -The Database Engine works like a traditional time series database. Unlike other [database modes](/database/README.md),
10 -the amount of historical metrics stored is based on the amount of disk space you allocate and the effective compression
11 -ratio, not a fixed number of metrics collected.
9 +DBENGINE is the time-series database of Netdata.
10
13 -## Tiering
11 +## Design
12
15 -Tiering is a mechanism of providing multiple tiers of data with
16 -different [granularity on metrics](/docs/store/distributed-data-architecture.md#granularity-of-metrics).
13 +### Data Points
14
18 -For Netdata Agents with version `netdata-1.35.0.138.nightly` and greater, `dbengine` supports Tiering, allowing almost
19 -unlimited retention of data.
15 +**Data points** represent the collected values of metrics.
16
17 +A **data point** has:
18
22 -### Metric size
19 +1. A **value**, the data collected for a metric. There is a special **value** to indicate that the collector failed to collect a valid value, and thus the data point is a **gap**.
20 +2. A **timestamp**, the time it has been collected.
21 +3. A **duration**, the time between this and the previous data collection.
22 +4. A flag which is set when machine-learning categorized the collected value as **anomalous** (an outlier based on the trained models).
23
24 -Every Tier down samples the exact lower tier (lower tiers have greater resolution). You can have up to 5
25 -Tiers **[0. . 4]** of data (including the Tier 0, which has the highest resolution)
24 +Using the **timestamp** and **duration**, Netdata calculates for each point its **start time**, **end time** and **update every**.
25
27 -Tier 0 is the default that was always available in `dbengine` mode. Tier 1 is the first level of aggregation, Tier 2 is
28 -the second, and so on.
26 +For incremental metrics (counters), Netdata interpolates the collected values to align them to the expected **end time** at the microsecond level, absorbing data collection micro-latencies.
27
30 -Metrics on all tiers except of the _Tier 0_ also store the following five additional values for every point for accurate
31 -representation:
28 +When data points are stored in higher tiers (time aggregations - see [Tiers](#Tiers) below), each data point has:
29
33 -1. The `sum` of the points aggregated
34 -2. The `min` of the points aggregated
35 -3. The `max` of the points aggregated
36 -4. The `count` of the points aggregated (could be constant, but it may not be due to gaps in data collection)
37 -5. The `anomaly_count` of the points aggregated (how many of the aggregated points found anomalous)
30 +1. The **sum** of the original values that have been aggregated,
31 +2. The **count** of all the original values aggregated,
32 +3. The **minimum** value among them,
33 +4. The **maximum** value among them,
34 +5. Their **anomaly rate**, i.e. the count of values that were detected as outliers based on the currently trained models for the metric,
35 +6. A **timestamp**, which is the equal to the **end time** of the last point aggregated,
36 +7. A **duration**, which is the duration between the **first time** of the first point aggregated to the **end time** of the last point aggregated.
37
39 -Among `min`, `max` and `sum`, the correct value is chosen based on the user query. `average` is calculated on the fly at
40 -query time.
38 +This design allows Netdata to accurately know the **average**, **minimum**, **maximum** and **anomaly rate** values even when using higher tiers to satisfy a query.
39
42 -### Tiering in a nutshell
40 +### Pages
41 +Data points are organized into **pages**, i.e. segments of contiguous data collections of the same metric.
42
44 -The `dbengine` is capable of retaining metrics for years. To further understand the `dbengine` tiering mechanism let's
45 -explore the following configuration.
43 +Each page:
44 +
45 +1. Contains contiguous **data points** of a single metric.
46 +2. Contains **data points** having the same **update every**. If a metric changes **update every** on the fly, the page is flushed and a new one with the new **update every** is created. If a data collection is missed, a **gap point** is inserted into the page, so that the data points in a page remain contiguous.
47 +3. Has a **start time**, which is equivalent to the **end time** of the first data point stored into it,
48 +4. Has an **end time**, which is equal to the **end time** of the last data point stored into it,
49 +5. Has an **update every**, common for all points in the page.
50 +
51 +A **page** is a simple array of values. Each slot in the array has a **timestamp** implied by its position in the array, and each value stored represents the **data point** for that time, for the metric the page belongs to.
52 +
53 +This simple fixed step page design allows Netdata to collect several millions of points per second and pack all the values in a compact form with minimal metadata overhead.
54 +
55 +#### Hot Pages
56 +
57 +While a metric is collected, there is one **hot page** in memory for each of the configured tiers. Values collected for a metric are appended to its **hot page** until that page becomes full.
58 +
59 +#### Dirty Pages
60 +
61 +Once a **hot page** is full, it becomes a **dirty page**, and it is scheduled for immediate **flushing** (saving) to disk.
62 +
63 +#### Clean Pages
64 +
65 +Flushed (saved) pages are **clean pages**, i.e. read-only pages that reside primarily on disk, and are loaded on demand to satisfy data queries.
66 +
67 +#### Pages Configuration
68 +
69 +Pages are configured like this:
70 +
71 +| Attribute | Tier0 | Tier1 | Tier2 |
72 +|---------------------------------------------------------------------------------------|:-------------------------------------:|:---------------------------------------------------------------:|:---------------------------------------------------------------:|
73 +| Point Size in Memory, in Bytes | 4 | 16 | 16 |
74 +| Point Size on Disk, in Bytes<br/><small>after LZ4 compression, on the average</small> | 1 | 4 | 4 |
75 +| Page Size in Bytes | 4096<br/><small>2048 in 32bit</small> | 2048<br/><small>1024 in 32bit</small> | 384<br/><small>192 in 32bit</small> |
76 +| Collections per Point | 1 | 60x Tier0<br/><small>configurable in<br/>`netdata.conf`</small> | 60x Tier1<br/><small>configurable in<br/>`netdata.conf`</small> |
77 +| Points per Page | 1024<br/><small>512 in 32bit</small> | 128<br/><small>64 in 32bit</small> | 24<br/><small>12 in 32bit</small> |
78 +
79 +### Files
80 +
81 +To minimize the amount of data written to disk and the amount of storage required for storing metrics, Netdata aggregates up to 64 **dirty pages** of independent metrics, packs them all together into one bigger buffer, compresses this buffer with LZ4 (about 75% savings on the average) and commits a transaction to the disk files.
82 +
83 +#### Extents
84 +
85 +This collection of 64 pages that is packed and compressed together is called an **extent**. Netdata tries to store together, in the same **extent**, metrics that are meant to be "close". Dimensions of the same chart are such. They are usually queried together, so it is beneficial to have them in the same **extent** to read all of them at once at query time.
86 +
87 +#### Datafiles
88 +
89 +Multiple **extents** are appended to **datafiles** (filename suffix `.ndf`), until these **datafiles** become full. The size of each **datafile** is determined automatically by Netdata. The minimum for each **datafile** is 4MB and the maximum 512MB. Depending on the amount of disk space configured for each tier, Netdata will decide a **datafile** size trying to maintain about 50 datafiles for the whole database, within the limits mentioned (4MB min, 512MB max per file). The maximum number of datafiles supported is 65536, and therefore the maximum database size (per tier) that Netdata can support is 32TB.
90 +
91 +#### Journal Files
92 +
93 +Each **datafile** has two **journal files** with metadata related to the stored data in the **datafile**.
94 +
95 +- **journal file v1**, with filename suffix `.njf`, holds information about the transactions in its **datafile** and provides the ability to recover as much data as possible, in case either the datafile or the journal files get corrupted. This journal file has a maximum transaction size of 4KB, so in case data are corrupted on disk transactions of 4KB are lost. Each transaction holds the metadata of one **extent** (this is why DBENGINE supports up to 64 pages per extent).
96 +
97 +- **journal file v2**, with filename suffix `.njfv2`, which is a disk-based index for all the **pages** and **extents**. This file is memory mapped at runtime and is consulted to find where the data of a metric are in the datafile. This journal file is automatically re-created from **journal file v1** if it is missing. It is safe to delete these files (when Netdata does not run). Netdata will re-create them on the next run. Journal files v2 are supported in Netdata Agents with version `netdata-1.37.0-115-nightly`. Older versions maintain the journal index in memory.
98 +
99 +#### Database Rotation
100 +
101 +Database rotation is achieved by deleting the oldest **datafile** (and its journals) and creating a new one (with its journals).
102 +
103 +Data on disk are append-only. There is no way to delete, add, or update data in the middle of the database. If data are not useful for whatever reason, Netdata can be instructed to ignore these data. They will eventually be deleted from disk when the database is rotated. New data are always appended.
104 +
105 +#### Tiers
106 +
107 +Tiers are supported in Netdata Agents with version `netdata-1.35.0.138.nightly` and greater.
108 +
109 +**datafiles** and **journal files** are organized in **tiers**. All tiers share the same metrics and same collected values.
110 +
111 +- **tier 0** is the high resolution tier that stores the collected data at the frequency they are collected.
112 +- **tier 1** by default aggregates 60 values of **tier 0**.
113 +- **tier 2** by default aggregates 60 values of **tier 1**, or 3600 values of **tier 0**.
114 +
115 +Updating the higher **tiers** is automated, and it happens in real-time while data are being collected for **tier 0**.
116 +
117 +When the Netdata Agent starts, during the first data collection of each metric, higher tiers are automatically **backfilled** with data from lower tiers, so that the aggregation they provide will be accurate.
118 +
119 +3 tiers are enabled by default in Netdata, with the following configuration:
120
121 ```
122 [db]
@@ -51,46 +125,151 @@ explore the following configuration.
125 # per second data collection
126 update every = 1
127
54 - # enables Tier 1 and Tier 2, Tier 0 is always enabled in dbengine mode
128 + # number of tiers used (1 to 5, 3 being default)
129 storage tiers = 3
130
57 - # Tier 0, per second data for a week
58 - dbengine multihost disk space MB = 1100
131 + # Tier 0, per second data
132 + dbengine multihost disk space MB = 256
133
60 - # Tier 1, per minute data for a month
61 - dbengine tier 1 multihost disk space MB = 330
134 + # Tier 1, per minute data
135 + dbengine tier 1 multihost disk space MB = 128
136 +
137 + # Tier 2, per hour data
138 + dbengine tier 2 multihost disk space MB = 64
139 +```
140 +
141 +The exact retention that can be achieved by each tier depends on the number of metrics collected. The more the metrics, the smaller the retention that will fit in a given size. The general rule is that Netdata needs about **1 byte per data point on disk for tier 0**, and **4 bytes per data point on disk for tier 1 and above**.
142 +
143 +So, for 1000 metrics collected per second and 256 MB for tier 0, Netdata will store about:
144 +
145 +```
146 +256MB on disk / 1 byte per point / 1000 metrics => 256k points per metric / 86400 seconds per day = about 3 days
147 +```
148 +
149 +At tier 1 (per minute):
150 +
151 +```
152 +128MB on disk / 4 bytes per point / 1000 metrics => 32k points per metric / (24 hours * 60 minutes) = about 22 days
153 +```
154 +
155 +At tier 2 (per hour):
156 +
157 +```
158 +64MB on disk / 4 bytes per point / 1000 metrics => 16k points per metric / 24 hours per day = about 2 years
159 +```
160 +
161 +Of course double the metrics, half the retention. There are more factors that affect retention. The number of ephemeral metrics (i.e. metrics that are collected for part of the time). The number of metrics that are usually constant over time (affecting compression efficiency). The number of restarts a Netdata Agents gets through time (because it has to break pages prematurely, increasing the metadata overhead). But the actual numbers should not deviate significantly from the above.
162 +
163 +### Data Loss
164 +
165 +Until **hot pages** and **dirty pages** are **flushed** to disk they are at risk (e.g. due to a crash, or
166 +power failure), as they are stored only in memory.
167 +
168 +The supported way of ensuring high data availability is the use of Netdata Parents to stream the data in real-time to
169 +multiple other Netdata agents.
170 +
171 +## Memory Requirements
172 +
173 +DBENGINE memory is related to the number of metrics concurrently being collected, the retention of the metrics on disk in relation with the queries running, and the number of metrics for which retention is maintained.
174 +
175 +### Memory for concurrently collected metrics
176 +
177 +DBENGINE is automatically sized to use memory according to this equation:
178 +
179 +```
180 +memory in KiB = METRICS x (TIERS - 1) x 4KiB x 2 + 32768 KiB
181 +```
182 +
183 +Where:
184 +- `METRICS`: the maximum number of concurrently collected metrics (dimensions) from the time the agent started.
185 +- `TIERS`: the number of storage tiers configured, by default 3 ( `-1` when using 3+ tiers)
186 +- `x 2`, to accommodate room for flushing data to disk
187 +- `x 4KiB`, the data segment size of each metric
188 +- `+ 32768 KiB`, 32 MB for operational caches
189 +
190 +So, for 2000 metrics (dimensions) in 3 storage tiers:
191 +
192 +```
193 +memory for 2k metrics = 2000 x (3 - 1) x 4 KiB x 2 + 32768 KiB = 64 MiB
194 +```
195 +
196 +For 100k concurrently collected metrics in 3 storage tiers:
197 +
198 +```
199 +memory for 100k metrics = 100000 x (3 - 1) x 4 KiB x 2 + 32768 KiB = 1.6 GiB
200 +```
201 +
202 +#### Exceptions
203 +
204 +Netdata has several protection mechanisms to prevent the use of more memory (than the above), by incrementally fetching data from disk and aggressively evicting old data to make room for new data, but still memory may grow beyond the above limit under the following conditions:
205 +
206 +1. The number of pages concurrently used in queries do not fit the in the above size. This can happen when multiple queries of unreasonably long time-frames run on lower, higher resolution, tiers. The Netdata query planner attempts to avoid such situations by gradually loading pages, but still under extreme conditions the system may use more memory to satisfy these queries.
207 +
208 +2. The disks that host Netdata files are extremely slow for the workload required by the database so that data cannot be flushed to disk quickly to free memory. Netdata will automatically spawn more flushing workers in an attempt to parallelize and speed up flushing, but still if the disks cannot write the data quickly enough, they will remain in memory until they are written to disk.
209 +
210 +### Caches
211 +
212 +DBENGINE stores metric data to disk. To achieve high performance even under severe stress, it uses several layers of caches.
213 +
214 +#### Main Cache
215 +
216 +Stores page data. It is the primary storage of hot and dirty pages (before they are saved to disk), and its clean queue is the LRU cache for speeding up queries.
217 +
218 +The entire DBENGINE is designed to use the hot queue size (the currently collected metrics) as the key for sizing all its memory consumption. We call this feature **memory ballooning**. More collected metrics, bigger main cache and vice versa.
219 +
220 +In the equation:
221
63 - # Tier 2, per hour data for a year
64 - dbengine tier 2 multihost disk space MB = 67
222 ```
223 +memory in KiB = METRICS x (TIERS - 1) x 4KiB x 2 + 32768 KiB
224 +```
225 +
226 +the part `METRICS x (TIERS - 1) x 4KiB` is an estimate for the max hot size of the main cache. Tier 0 pages are 4KiB, but tier 1 pages are 2 KiB and tier 2 pages are 384 bytes. So a single metric in 3 tiers uses 4096 + 2048 + 384 = 6528 bytes. The equation estimates 8192 per metric, which includes cache internal structures and leaves some spare.
227 +
228 +Then `x 2` is the worst case estimate for the dirty queue. If all collected metrics (hot) become available for saving at once, to avoid stopping data collection all their pages will become dirty and new hot pages will be created instantly. To save memory, when Netdata starts, DBENGINE allocates randomly smaller pages for metrics, to spread their completion evenly across time.
229 +
230 +The memory we saved with the above is used to improve the LRU cache. So, although we reserved 32MiB for the LRU, in bigger setups (Netdata Parents) the LRU grows a lot more, within the limits of the equation.
231 +
232 +In practice, the main cache sizes itself with `hot x 1.5` instead of `host x 2`. The reason is that 5% of main cache is reserved for expanding open cache, 5% for expanding extent cache and we need room for the extensive buffers that are allocated in these setups. When the main cache exceeds `hot x 1.5` it enters a mode of critical evictions, and aggresively frees pages from the LRU to maintain a healthy memory footprint within its design limits.
233 +
234 +#### Open Cache
235
67 -For 2000 metrics, collected every second and retained for a week, Tier 0 needs: 1 byte x 2000 metrics x 3600 secs per
68 -hour x 24 hours per day x 7 days per week = 1100MB.
236 +Stores metadata about on disk pages. Not the data itself. Only metadata about the location of the data on disk.
237
70 -By setting `dbengine multihost disk space MB` to `1100`, this node will start maintaining about a week of data. But pay
71 -attention to the number of metrics. If you have more than 2000 metrics on a node, or you need more that a week of high
72 -resolution metrics, you may need to adjust this setting accordingly.
238 +Its primary use is to index information about the open datafile, the one that still accepts new pages. Once that datafile becomes full, all the hot pages of the open cache are indexed in journal v2 files.
239
74 -Tier 1 is by default sampling the data every **60 points of Tier 0**. In our case, Tier 0 is per second, if we want to
75 -transform this information in terms of time then the Tier 1 "resolution" is per minute.
240 +The clean queue is an LRU for reducing the journal v2 scans during quering.
241
77 -Tier 1 needs four times more storage per point compared to Tier 0. So, for 2000 metrics, with per minute resolution,
78 -retained for a month, Tier 1 needs: 4 bytes x 2000 metrics x 60 minutes per hour x 24 hours per day x 30 days per month
79 -= 330MB.
242 +Open cache uses memory ballooning too, like the main cache, based on its own hot pages. Open cache hot size is mainly controlled by the size of the open datafile. This is why on netdata versions with journal files v2, we decreased the maximum datafile size from 1GB to 512MB and we increased the target number of datafiles from 20 to 50.
243
81 -Tier 2 is by default sampling data every 3600 points of Tier 0 (60 of Tier 1, which is the previous exact Tier). Again
82 -in term of "time" (Tier 0 is per second), then Tier 2 is per hour.
244 +On bigger setups open cache will get a bigger LRU by automatically sizing it (the whole open cache) to 5% to the size of (the whole) main cache.
245
84 -The storage requirements are the same to Tier 1.
246 +#### Extent Cache
247 +
248 +Caches compressed **extent** data, to avoid reading too repeatedly the same data from disks.
249 +
250 +
251 +### Shared Memory
252 +
253 +Journal v2 indexes are mapped into memory. Netdata attempts to minimize shared memory use by instructing the kernel about the use of these files, or even unmounting them when they are not needed.
254 +
255 +The time-ranges of the queries running control the amount of shared memory required.
256 +
257 +## Metrics Registry
258 +
259 +DBENGINE uses 150 bytes of memory for every metric for which retention is maintained but is not currently being collected.
260 +
261 +---
262 +
263 +--- OLD DOCS BELOW THIS POINT ---
264 +
265 +---
266
86 -For 2000 metrics, with per hour resolution, retained for a year, Tier 2 needs: 4 bytes x 2000 metrics x 24 hours per day
87 -x 365 days per year = 67MB.
267
268 ## Legacy configuration
269
270 ### v1.35.1 and prior
271
93 -These versions of the Agent do not support [Tiering](#Tiering). You could change the metric retention for the parent and
272 +These versions of the Agent do not support [Tiers](#Tiers). You could change the metric retention for the parent and
273 all of its children only with the `dbengine multihost disk space MB` setting. This setting accounts the space allocation
274 for the parent node and all of its children.
275
@@ -105,15 +284,9 @@ the `[db]` section of your `netdata.conf`.
284
285 ### v1.23.2 and prior
286
108 -_For Netdata Agents earlier than v1.23.2_, the Agent on the parent node uses one dbengine instance for itself, and
109 -another instance for every child node it receives metrics from. If you had four streaming nodes, you would have five
110 -instances in total (`1 parent + 4 child nodes = 5 instances`).
287 +_For Netdata Agents earlier than v1.23.2_, the Agent on the parent node uses one dbengine instance for itself, and another instance for every child node it receives metrics from. If you had four streaming nodes, you would have five instances in total (`1 parent + 4 child nodes = 5 instances`).
288
112 -The Agent allocates resources for each instance separately using the `dbengine disk space MB` (**deprecated**) setting.
113 -If
114 -`dbengine disk space MB`(**deprecated**) is set to the default `256`, each instance is given 256 MiB in disk space,
115 -which means the total disk space required to store all instances is,
116 -roughly, `256 MiB * 1 parent * 4 child nodes = 1280 MiB`.
289 +The Agent allocates resources for each instance separately using the `dbengine disk space MB` (**deprecated**) setting. If `dbengine disk space MB`(**deprecated**) is set to the default `256`, each instance is given 256 MiB in disk space, which means the total disk space required to store all instances is, roughly, `256 MiB * 1 parent * 4 child nodes = 1280 MiB`.
290
291 #### Backward compatibility
292
@@ -208,7 +381,7 @@ You can apply the settings by running `sysctl -p` or by rebooting.
381
382 ## Files
383
211 -With the DB engine mode the metric data are stored in database files. These files are organized in pairs, the datafiles
384 +With the DB engine mode the metric data are stored in database files. These files are organized in pairs, the datafiles
385 and their corresponding journalfiles, e.g.:
386
387 ```sh
@@ -298,5 +471,3 @@ An interesting observation to make is that the CPU-bound run (16 GiB page cache)
471 and generate a read load of 1.7M/sec, whereas in the CPU-bound scenario the read load is 70 times higher at 118M/sec.
472 Consequently, there is a significant degree of interference by the reader threads, that slow down the writer threads.
473 This is also possible because the interference effects are greater than the SSD impact on data generation throughput.
301 -
302 -
database/engine/cache.c
+42 -33
@@ -310,50 +310,45 @@ static inline size_t cache_usage_per1000(PGC *cache, size_t *size_to_evict) {
310
311 current_cache_size = __atomic_load_n(&cache->stats.size, __ATOMIC_RELAXED);
312
313 - per1000 = (size_t)((unsigned long long)current_cache_size * 1000UL / (unsigned long long)wanted_cache_size);
313 + per1000 = (size_t)((unsigned long long)current_cache_size * 1000ULL / (unsigned long long)wanted_cache_size);
314
315 __atomic_store_n(&cache->usage.per1000, per1000, __ATOMIC_RELAXED);
316 __atomic_store_n(&cache->stats.wanted_cache_size, wanted_cache_size, __ATOMIC_RELAXED);
317 __atomic_store_n(&cache->stats.current_cache_size, current_cache_size, __ATOMIC_RELAXED);
318
319 + netdata_spinlock_unlock(&cache->usage.spinlock);
320 +
321 if(size_to_evict) {
320 - size_t target = wanted_cache_size / 1000 * cache->config.evict_low_threshold_per1000;
322 + size_t target = (size_t)((unsigned long long)wanted_cache_size * (unsigned long long)cache->config.evict_low_threshold_per1000 / 1000ULL);
323 if(current_cache_size > target)
324 *size_to_evict = current_cache_size - target;
325 else
326 *size_to_evict = 0;
327 }
328
327 - netdata_spinlock_unlock(&cache->usage.spinlock);
328 - return per1000;
329 -}
330 -
331 -static inline bool cache_under_severe_pressure(PGC *cache) {
332 - if(unlikely(cache_usage_per1000(cache, NULL) >= cache->config.severe_pressure_per1000)) {
329 + if(per1000 >= cache->config.severe_pressure_per1000)
330 __atomic_add_fetch(&cache->stats.events_cache_under_severe_pressure, 1, __ATOMIC_RELAXED);
334 - return true;
335 - }
336 -
337 - return false;
338 -}
331
340 -static inline bool cache_needs_space_aggressively(PGC *cache) {
341 - if(unlikely(cache_usage_per1000(cache, NULL) >= cache->config.aggressive_evict_per1000)) {
332 + else if(per1000 >= cache->config.aggressive_evict_per1000)
333 __atomic_add_fetch(&cache->stats.events_cache_needs_space_aggressively, 1, __ATOMIC_RELAXED);
343 - return true;
344 - }
334
346 - return false;
335 + return per1000;
336 +}
337 +
338 +static inline bool cache_pressure(PGC *cache, size_t limit) {
339 + return (cache_usage_per1000(cache, NULL) >= limit);
340 }
341
349 -#define cache_above_healthy_limit(cache) (cache_usage_per1000(cache, NULL) >= (cache)->config.healthy_size_per1000)
342 +#define cache_under_severe_pressure(cache) cache_pressure(cache, (cache)->config.severe_pressure_per1000)
343 +#define cache_needs_space_aggressively(cache) cache_pressure(cache, (cache)->config.aggressive_evict_per1000)
344 +#define cache_above_healthy_limit(cache) cache_pressure(cache, (cache)->config.healthy_size_per1000)
345
346 typedef bool (*evict_filter)(PGC_PAGE *page, void *data);
347 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);
348 #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)
349
350 static inline void evict_on_clean_page_added(PGC *cache __maybe_unused) {
356 - if((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_under_severe_pressure(cache)) {
351 + if((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_needs_space_aggressively(cache)) {
352 evict_pages(cache,
353 cache->config.max_skip_pages_per_inline_eviction,
354 cache->config.max_pages_per_inline_eviction,
@@ -362,7 +357,7 @@ static inline void evict_on_clean_page_added(PGC *cache __maybe_unused) {
357 }
358
359 static inline void evict_on_page_release_when_permitted(PGC *cache __maybe_unused) {
365 - if (unlikely((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_under_severe_pressure(cache))) {
360 + if ((cache->config.options & PGC_OPTIONS_EVICT_PAGES_INLINE) || cache_under_severe_pressure(cache)) {
361 evict_pages(cache,
362 cache->config.max_skip_pages_per_inline_eviction,
363 cache->config.max_pages_per_inline_eviction,
@@ -956,12 +951,14 @@ static bool make_acquired_page_clean_and_evict_or_page_release(PGC *cache, PGC_P
951
952 // returns true, when there is more work to do
953 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) {
959 - if(!all_of_them && !cache_above_healthy_limit(cache))
954 + size_t per1000 = cache_usage_per1000(cache, NULL);
955 +
956 + if(!all_of_them && per1000 < cache->config.healthy_size_per1000)
957 // don't bother - not enough to do anything
958 return false;
959
960 size_t workers_running = __atomic_add_fetch(&cache->stats.workers_evict, 1, __ATOMIC_RELAXED);
964 - if(!wait && !all_of_them && workers_running > cache->config.max_workers_evict_inline) {
961 + if(!wait && !all_of_them && workers_running > cache->config.max_workers_evict_inline && per1000 < cache->config.severe_pressure_per1000) {
962 __atomic_sub_fetch(&cache->stats.workers_evict, 1, __ATOMIC_RELAXED);
963 return false;
964 }
@@ -979,7 +976,6 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
976 else if(unlikely(max_evict < 2))
977 max_evict = 2;
978
982 - PGC_PAGE *pages_to_evict = NULL;
979 size_t total_pages_evicted = 0;
980 size_t total_pages_skipped = 0;
981 bool stopped_before_finishing = false;
@@ -989,12 +985,20 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
985 if(++spins > 1)
986 __atomic_add_fetch(&cache->stats.evict_spins, 1, __ATOMIC_RELAXED);
987
988 + bool batch;
989 size_t max_size_to_evict = 0;
993 - if (all_of_them)
990 + if (unlikely(all_of_them)) {
991 max_size_to_evict = SIZE_MAX;
995 -
996 - else
997 - cache_usage_per1000(cache, &max_size_to_evict);
992 + batch = true;
993 + }
994 + else if(unlikely(wait)) {
995 + per1000 = cache_usage_per1000(cache, &max_size_to_evict);
996 + batch = (wait && per1000 > cache->config.severe_pressure_per1000) ? true : false;
997 + }
998 + else {
999 + batch = false;
1000 + max_size_to_evict = (cache_above_healthy_limit(cache)) ? 1 : 0;
1001 + }
1002
1003 if (!max_size_to_evict)
1004 break;
@@ -1017,7 +1021,8 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
1021 pgc_ll_lock(cache, &cache->clean);
1022
1023 // find a page to evict
1020 - pages_to_evict = NULL;
1024 + PGC_PAGE *pages_to_evict = NULL;
1025 + size_t pages_to_evict_size = 0;
1026 for(PGC_PAGE *page = cache->clean.base, *next = NULL, *first_page_we_relocated = NULL; page ; page = next) {
1027 next = page->link.next;
1028
@@ -1046,7 +1051,9 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
1051
1052 DOUBLE_LINKED_LIST_APPEND_UNSAFE(pages_to_evict, page, link.prev, link.next);
1053
1049 - if(unlikely(all_of_them))
1054 + pages_to_evict_size += page->assumed_size;
1055 +
1056 + if(unlikely(all_of_them || (batch && pages_to_evict_size < max_size_to_evict)))
1057 // get more pages
1058 ;
1059 else
@@ -1698,7 +1705,8 @@ void free_all_unreferenced_clean_pages(PGC *cache) {
1705
1706 PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_cb,
1707 size_t max_dirty_pages_per_flush, save_dirty_page_callback pgc_save_dirty_cb,
1701 - size_t max_pages_per_inline_eviction, size_t max_skip_pages_per_inline_eviction,
1708 + size_t max_pages_per_inline_eviction, size_t max_inline_evictors,
1709 + size_t max_skip_pages_per_inline_eviction,
1710 size_t max_flushes_inline,
1711 PGC_OPTIONS options, size_t partitions, size_t additional_bytes_per_page) {
1712
@@ -1723,7 +1731,7 @@ PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_cb,
1731 cache->config.partitions = partitions < 1 ? (size_t)get_system_cpus() : partitions;
1732 cache->config.additional_bytes_per_page = additional_bytes_per_page;
1733
1726 - cache->config.max_workers_evict_inline = 10;
1734 + cache->config.max_workers_evict_inline = max_inline_evictors;
1735 cache->config.severe_pressure_per1000 = 1010;
1736 cache->config.aggressive_evict_per1000 = 990;
1737 cache->config.healthy_size_per1000 = 980;
@@ -1980,6 +1988,7 @@ struct pgc_statistics pgc_get_statistics(PGC *cache) {
1988 }
1989
1990 void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_fileno, uint8_t type, migrate_to_v2_callback cb, void *data) {
1991 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.journal_v2_indexing_started, 1, __ATOMIC_RELAXED);
1992 __atomic_add_fetch(&cache->stats.workers_jv2_flush, 1, __ATOMIC_RELAXED);
1993
1994 pgc_ll_lock(cache, &cache->hot);
@@ -2571,7 +2580,7 @@ void unittest_stress_test(void) {
2580 int pgc_unittest(void) {
2581 PGC *cache = pgc_create(32 * 1024 * 1024, unittest_free_clean_page_callback,
2582 64, unittest_save_dirty_page_callback,
2574 - 10, 1000, 10,
2583 + 10, 10, 1000, 10,
2584 PGC_OPTIONS_DEFAULT, 1, 11);
2585
2586 // FIXME - unit tests
database/engine/cache.h
+2 -1
@@ -164,7 +164,8 @@ typedef void (*save_dirty_page_callback)(PGC *cache, PGC_ENTRY *entries_array, P
164 // create a cache
165 PGC *pgc_create(size_t clean_size_bytes, free_clean_page_callback pgc_free_clean_cb,
166 size_t max_dirty_pages_per_flush, save_dirty_page_callback pgc_save_dirty_cb,
167 - size_t max_pages_per_inline_eviction, size_t max_skip_pages_per_inline_eviction,
167 + size_t max_pages_per_inline_eviction, size_t max_inline_evictors,
168 + size_t max_skip_pages_per_inline_eviction,
169 size_t max_flushes_inline,
170 PGC_OPTIONS options, size_t partitions, size_t additional_bytes_per_page);
171
database/engine/datafile.c
+19 -23
@@ -447,10 +447,8 @@ static int scan_data_files(struct rrdengine_instance *ctx)
447 if (0 != ret) {
448 must_delete_pair = 1;
449 }
450 - journalfile = mallocz(sizeof(*journalfile));
451 - datafile->journalfile = journalfile;
452 - journalfile_init(journalfile, datafile);
453 - ret = load_journal_file(ctx, journalfile, datafile);
450 + journalfile = journalfile_alloc_and_init(datafile);
451 + ret = journalfile_load(ctx, journalfile, datafile);
452 if (0 != ret) {
453 if (!must_delete_pair) /* If datafile is still open close it */
454 close_data_file(datafile);
@@ -460,9 +458,9 @@ static int scan_data_files(struct rrdengine_instance *ctx)
458 char path[RRDENG_PATH_MAX];
459
460 error("DBENGINE: deleting invalid data and journal file pair.");
463 - ret = unlink_journal_file(journalfile);
461 + ret = journalfile_unlink(journalfile);
462 if (!ret) {
465 - generate_journalfilepath(datafile, path, sizeof(path));
463 + journalfile_generate_path(datafile, path, sizeof(path));
464 info("DBENGINE: deleted journal file \"%s\".", path);
465 }
466 ret = unlink_data_file(datafile);
@@ -488,6 +486,8 @@ static int scan_data_files(struct rrdengine_instance *ctx)
486 /* Creates a datafile and a journalfile pair */
487 int create_new_datafile_pair(struct rrdengine_instance *ctx)
488 {
489 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_creation_started, 1, __ATOMIC_RELAXED);
490 +
491 struct rrdengine_datafile *datafile;
492 struct rrdengine_journalfile *journalfile;
493 unsigned fileno = __atomic_load_n(&ctx->last_fileno, __ATOMIC_RELAXED) + 1;
@@ -497,23 +497,20 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx)
497 info("DBENGINE: creating new data and journal files in path %s", ctx->dbfiles_path);
498 datafile = datafile_alloc_and_init(ctx, 1, fileno);
499 ret = create_data_file(datafile);
500 - if (!ret) {
501 - generate_datafilepath(datafile, path, sizeof(path));
502 - info("DBENGINE: created data file \"%s\".", path);
503 - } else {
500 + if(ret)
501 goto error_after_datafile;
505 - }
502
507 - journalfile = mallocz(sizeof(*journalfile));
508 - datafile->journalfile = journalfile;
509 - journalfile_init(journalfile, datafile);
510 - ret = create_journal_file(journalfile, datafile);
511 - if (!ret) {
512 - generate_journalfilepath(datafile, path, sizeof(path));
513 - info("DBENGINE: created journal file \"%s\".", path);
514 - } else {
503 + generate_datafilepath(datafile, path, sizeof(path));
504 + info("DBENGINE: created data file \"%s\".", path);
505 +
506 + journalfile = journalfile_alloc_and_init(datafile);
507 + ret = journalfile_create(journalfile, datafile);
508 + if (ret)
509 goto error_after_journalfile;
516 - }
510 +
511 + journalfile_generate_path(datafile, path, sizeof(path));
512 + info("DBENGINE: created journal file \"%s\".", path);
513 +
514 datafile_list_insert(ctx, datafile);
515 ctx->disk_space += datafile->pos + journalfile->pos;
516
@@ -524,10 +521,9 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx)
521 error_after_journalfile:
522 destroy_data_file_unsafe(datafile);
523 freez(journalfile);
524 +
525 error_after_datafile:
526 freez(datafile);
529 -
530 - uv_rwlock_wrunlock(&ctx->datafiles.rwlock);
527 return ret;
528 }
529
@@ -587,7 +583,7 @@ void finalize_data_files(struct rrdengine_instance *ctx)
583 }
584 } while(!available);
585
590 - close_journal_file(journalfile, datafile);
586 + journalfile_close(journalfile, datafile);
587 close_data_file(datafile);
588 datafile_list_delete_unsafe(ctx, datafile);
589 netdata_spinlock_unlock(&datafile->writers.spinlock);
database/engine/dbengine-diagram.xml new
+1
@@ -0,0 +1 @@
1 +<mxfile host="app.diagrams.net" modified="2023-01-16T23:29:24.274Z" agent="5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36" etag="IzytOgui5y4srcr9Zrcm" version="20.5.1" type="device"><diagram name="Page-1" id="90a13364-a465-7bf4-72fc-28e22215d7a0">7V1rc5tI1v41rsp8MEVz52PsOJvsxkkmydTu7Je3EGpZTJDQIOTY++vf7oYG+oJoIUDIlqcmthA00H3O0+d+rszb1dM/0mCzvE/mML4y9PnTlfnuyjBM3zc1G/2Bjz3nx4DuG/mRhzSaF8eqA9+j/0F6YnF0F83hljkxS5I4izbswTBZr2GYMceCNE1+sactkpi96yZ4gMKB72EQ06P0DfDxf0fzbFkcB45fffEBRg/L4uae4eRfzILw50Oa7NbFHdfJGubfrAI6TPGW22UwT37VDpl3V+ZtmiRZ/tfq6RbGeG7pnNHrsmf6oFfmzTJbxegDQH+Sr983XAxULkbvlcJ1Vr9d03jL53+l9tvPH39EN3fr2R/2l88f/rq2i1l4DOIdvQt/WzhHc118JLNj3pD5gnhgHX3a/oRZuCw+JGm2TB6SdRB/SpJN8ax/wSx7Lsgm2GUJ+yblvJLBsjT5Wa4hfudFss6Ka4FHzgjS7C2mmuqByLH3URwXo8D1nJ4RxsF2G4X5weIUQD5l6fN/8AdNLz//ia/XHMOlB949lSPiT8/Fp3mwXZL3J1c+Rdl/yIV28elPehH6uxoBf3hmXvM2iZOUzLJ5S37ojb7CNFrBDKbF6fmi4JXg2KBafUovyS4N4b4lL1YYzdgDzPacaLkljSPsgAl6nvQZXZjCOMiiR/ZJgoKDH8rzyku/JhF6RkMvAAcYfsEXBdr4ns6Okb9CcVmdovmRfEfX9OoHUUd9XGB5Njtw/srCwOiP2gtWhwgDHcJMYGrMxJDpfs5yhuAshq1aWKrkojoPlRwl5yL8fl+DDPHJmtzO0MFe3urGRu/v9J/Jp//+988/7ufO82fvK7iPrm3/xGwE3Drt65bNEL/j6kq0L2FPljstW407+2Ii07sw0R4m0ly7Ex8Zk2UkxxyGP/Q2Qj6AI9iRbMPV7HGZwheY4n2EJFZD/4rk2u2V4cRo1m9mKcMnzt87LJoSCr3eEhJ9i1fW2jyRJaTfo78eit/BCjHIzXq2xb+iNTr7PiC/boNwCaWM+CmYIYWCYZ8gjh4wHYWIGLAIc/MI0yxC4vrb4otVNJ/HhG0heq5gRsbDdLjBM0jm1L65st9x7GURRs8QWSRrRjDC48MnhlIKVaMYuxLG6yS6B4BEeiyGv9Y1RBAWQxDXxT2PJNlrC7DDOhyVJYvFFh5LYFImRIrS1GB3fEWgxFIVcUWqAdC/VWSXNnDtgqVUqa0LJdITPUOK4ANIKR6LnUDvKu17jqHpnut6hqubvuca7LjAsjQX3c12TNOxHGANJfrLJ15B9Gd55dcyyuD3TUDW81cabFhW4GGPIbdGAhFwsBHGDINfl+Lzr8p64tFjy5rhxOS30ToRMNN6MAYZh85hF4hhUEUEnTqCKWDHEayqyKnUxNYfpx63SKawSB+SrGdBZAZD9LjYELjDT3R6oUNc5v0UrCqONLInEjIs3+1HqgCssGJw/DucUCHj54I+pORRmWKvw3xbxFQSraMsCmIpoUhIDz3prCbQluTIHplHj+0nRfwBTBZHEXbxpHq4SzEpxc/SmzNC+AxG6wfMCgh0EPC1nv8rQsiG6Q2dHmb4Cn0eZEFtevK3YN8MHRbeFh1jpoljw5YNrIRVfjfDkht+mZrks1hAJwxlMtHc9Wd6Jecft+XZrHgNLHHLA65kyzN4kaU3NDVkLDIhqVsmiQy6K5qKu2Lv4utRyyjuiTWgy9nnNufHJD0QqMQjR4NQ8SglMOiLNFmhX6sddtjFsPoiX65tI3YMiAg2/k+GCA756QcRWH9BuTPW8YCaVet4APgttD/panLGzxPjAZV+28Vkd1KIAESD3bsozZ5lgnLOTdvgkWzWaDUQ/0Xbn+ck91Ky7UHuNSzT6Unu5dh7PLnXnTYX87Y0RS6Wu7npN3Ij1wicT21eU+F8fXCtR4okZ6H3ZMsAD4TIoVTu+YcIiFsjP+NXgCYBaz9tSg8BzRkZNN4RQxkDpJNWfRZGg+rjzBy7J0HHNjlr3wRUn4mLOi0gWTxOhZClh6B74MMBuMq4PirArns/jhS83DOFX9ne2y/83sYwWA8Iv9sNGf4oPy6L4xz8Ikra4t/JmkNIOQqH2PWrYHkifuIVXCXpcwu25m94Ksid29CbWzLI9YyZ2ZtuCaYHuaJSIszzpJ1UtnNyJ5UpE+/2zuE5O6lMQ3EXMO1J7QKm6I7FJi0EBFDEsdzbJFHJz8z7VJLmlL1P5mhaOKXdUdxPx4sB9MA9zALOYdPJTDyMEoe+rx5QL2xW/CPkmlfBb1juINJEm/jwFwIjBJL0mkdEarfiVaVKqC/QCmGRBtZupqgphnGyhVRFJGpjhAD7SfYuRKipPRp+MuNq2hrlKZxpjiR+ZGTxxpS5YSakUY7vTLNVd+9p6XCmJSwk1WCU+LTg0fPZuCnhXszndbIdyYLKbGm9K/M87o9z6Qibf+tGWzf17ol3YbdfXcLHR++1jRPX/zTtXc0pygqnsD5PQVaQhbsPZyacnnowltlxL2Zw9shG3t/3+H9sK29PCue7EAqD6NswWNcttReb5HQY8czcQIPknWhlTl/nzJNDsma7axTK+SXU+DJ0fomts2mvNhgs5XvvhOwL/TtYuPAaAPZdbnV5T1zcaYtgIZVA9mwV34NHjJxcbEGOq++k4UhxHG22mPZbcJChV15F4nAQkh8ZDpqO6ZvzfnDQsVkctCS+mbK2CRP5N5hfwVLwK0wJBxWNF1182zVc1Flg1AynBRv3FM8Y1oLij4R3XO0Mr2s2nWObmuci+PQt17A9n1X2fcPUTMt0fd20HUvXjVFB1RK9N+fECy89NKR/a6GcRH1g42oHvZC7Z/ma7zs4dNuwbM81OXIHmu84vg8sAHTdB+OSu9EuQ/yTVWg+YntJVxWQbu01Rw7RvIhHBwZpuCTmRhoaos+CbWGaQTfVmu9wtPGEe8nSe0RMSKtgsykeg8gkNNBEf4M4NSXf5Id+0xolIw5DOsc6yBS4AHoLqSXFCT04W/D6YA9SjJjCK0lgkIZHAGcwdU4WYjKUPCxjit7l4oLVmkzFnNeTe6Qov3gQybk5JGQSMnUJ1SeUqSdnW3gJpYT2Ch/yOkLdZQ3LUhTA6YjDF7TgywrxIsnQ0sKlFBBFQGkpoFZvq9VbstLLLAU0vSqG4wPlgRbZEfDN6j3AsgHfTKPBRzASvtHypPskyN93kLzrMsD6wUOwUQzDf7PNy4qQyLGAPEuyuCqCyPiTU/j3LsoVi02hMRFNhD9vTYIAF6SCcK5J8WesctwkcfS/TQE4FZHSbhICuyCl4Zis0k1z+I9FSoer+uqMBZSUdyWUqqzaOI3hFtstEZ6a9fnSI8zsyoeYBNT1Y9Cuq3TK7sfn1fUR8iNVonvM+hfqOUrN/5ZEVRnMC2o1hzz1RUqGAil92cALKR1FSs7JSQkAGSxNWH6z+pPf1PKP2oUyVS+3MZLXR1A6BTv4wEIZACJAfU02O/RWYriNIP5UdmZiI0tylAmnokYqSkMVY/UjDllcfG0/0hAnZNFSFCNIQ82REH1tYabCFsYZYL+Hwfqyk3XZybyT72T2xPM5zskQMbzZAPga66t1+YwB5bYe4lglMY5lhGhW7QQjxDaLyIATNUbohfUB34Y/u74Tn5VhoreEmSENE1yQAR+ZOKAFV6ZNnhQ4p+X36tZBo0sDjSN0ENXIs9F0EN1lBQRbsTGMGIvjOppT+2Gh3nKF5hpDg70oubIeNanTv2/nGmt7OT0Ud3OuldAzaeca4JxrVBUbAZonVxpwWtBsdMHm/fGQvSOzqnWIxvUNbx3i0nr8rpK3zyXq+cM10JNzhxiwcwIk/ufEssg7IjFFmkkjMdcdpkTmEZC4OVLx6MBEQVcsMncKlfEoc9Rg0bdf0+QxmsNSGSUTsy20UolNN0WwjhiSVGmhJcSziKi0GUZZYuvNDGkVl/J0HK6p/8jHeJPuMPsVDQw2aZSkUZ6L9Nu4xjm8zx1ej7zJbNe40xxgnOO7KBmSxM3SEDdOqXLHmJokc6aJm5pum4yi6ZK8+z35UehDjylOkj6mJxVnLF6a4YPVlaUZQ9PR3LqG7fiuoXOpJEA3LM3WfcfyfdvRDZqxOpKo4x1oo0HrisDwG5naZH1XffPCOOrqIEtNd7J3VDuN+GMF3nm6zxCoM7aPVxcDP+8jtFyyAtyFPH564VhRGvb6tEvogGu2fKTBuOw/MpbsC/Qzc631GCMyAmSUvRQnUk8N6KI7KwipQ4ln7c2ZsXZFzL14gwxa0oHao4/jbSrUcOrzaEGqjozT+1dzJaOI51OP6bdkl1V5eocm5NEjyvWKWqJKrNptZUWAhujAtYThz1LRpg7dQvFd0T2XduEKHoJoTQok1QPJr4qyqvochrnajtTdEA8AnzKIWcvQ17CqjUTqrMVJMCeHCgW8yPAlJ6DL0oC0BdvkqY4KbQMv4TP027I0+Z5S5WXaLVNaiQ9t6E9DP7NA0EHiZ1h9AuitBUQknbwVypX0p48rZwRRZB9cMbH4KvxdHb/CSI47rorjiEo33ZMijJe44Dvts3o28k/J5pMOhnHYYa3R5B9ZWOrrBkFVF2sdAocrKacc6DIe3nHGdqcz3jlcTK0LbM3iSH9oyBMDDl4A5FGmnjLkAYcbdTTzjjO5uhenCJzWdY/FPd/wuyDfqMKfamzJaGBo82kBfAMXdTDkc5j4kYZGQtHh/wKQsM9iFoMJfzZrLh8xyoMqGKNEeahmYrfZSNBKZhyYMraLAh7rBpDikDpFyiwv3AbQXF+qlVZ7qGUmMZ/YEvMJDyK9WU9cWRmUoShHNfH6QjmdegSOTDlGu/g1rTaLZXovnUPqUqmHF0nncLDabe7eQoK9Nmi4o/by3pu/9N9SoXzWA3q5zna1RixFL1dd0qyVa6YUEBlitUGyS17/c5GkK/FctvFC4VBIYTC/NFzAnMWJPrYlctbIHRe8MUWinF4vW9twuDzu3ubJ9jZ5sPP7vHxvNydvUW21GIqA2zspuL0pnKRMq8laQYqqnaSY4Hrm9X8B31dpCgWAfRFdwjrrVytHfdH8cVhAxoQNVvm+MaCRvvxGJZK4kZRazU2eakQTFWlbzU0FaeoaMB3O/knjh480Jhi672l+9WMwd/FpEGWLUUkIvrsezSThNUfkKMuLh++/nzC3pYfBcWtADB60kBS3GxhGC5ISQhm4IeljPlwzmYmVxAYesLhCFj4QERrt6xKEHiwsxDsz70CPYMsQz4F5of1h5NEACBgA5GKRhbDxBgN9X3Dmiw5GGt4WNZYzueLyLilk5JZ4qqSejRXe680Kr2uWK42UODrlnUsgG60YCd2g92kMn779Id+HGncJhB8bSDg92c3bdw11tY3bKJy55zrWlRBVuCA/0g3kLdLxe5Lwfd4qLtk9ZFl/w0n3zbLLZTEPXEzbPfViKhS5uixmk9eBK1tjn5w1RbGOjegmPm4igvO7cWkUfl2qdyMZtOrNvmrMmq/aKq1WakG3OcWZKtJdRYAR9vjm8tWNKq3cX5M+zN4YuBwPLghg4F4n5A9b/40smN69AgOc7zYxEgtJwdq7BhX1G/x7B7fZgWbLPXpyXy/e9H79Ti9/l3uYEjPsahdn0Yb0hNsgHsqzU1g1/8XOATZT5i+ccsQhvDIHm3QnxBbq71mSYpN2u5k5eoJzCnZ1aARt+2NrmZjD7Reux6Uky8pOyPwPw9mXAe1FemQu4wvcz2qxj75fhjoW0duGddUS+9hvJKOvGsmonizLqa7HRja6nEpsKbYOH3NjBUCkdjydNUf+65LZGnIaWmj7GLtff5TuqRI6a0EqyNPsxxBk8MXiDif7sigRM85oJRR9MaPnrlR0GL9xgwNCYJ3GovLlkVqvg0IKqJdilobEsLm6TaJBEu5WhOLa5IJZzryfZtJgl6EjEYDHhYGbEk9zGXXASAL+YArvAa1m962teOQDJLFSxBb9jNj/SsiFbjaENIUjlRCEV0jdZE0JZfX0gEhiqQUYH7eG9tdutfm4LtDyZo5XBeAMrTleMBPHJ5AzP8AA39O222hkEUebD6UIKk+sRk8wjyBjbJmR/64kxpZek6strvSwIaE+aRRVCXYDyKEdwqgU9NSDXcHN1Q1q2KRc2uD426ZJCLfbJD383mqBl0cXX3iTV10QK+q3OanyCMpK2SiLHi4hh/fN9QovZRP2l00wTQlny8omDKlhih7N2v5fltCgNlWy28u3+YnFLI4iKTg+57OS2QxkfWQGi1kE4MCi25U29GLr7HFlt3WvrS4GW7tSZiuoClrSVPH29MkRtK7ROuA4fPHs7u1vDLQgrusZrm76Ho0rKduIWJqL5HHbMU3HcoDJlZIavBOcLHwo38SWgO5fn2FWxOViTWwWbInRff0Qresx4NX59GAt+P/Hksjf5B880BIhBYZf+AjxQ87hFgNhvWcObiyX5ibQ4vC7m7vP//j4+U6r3VSSuKGUy7HnUokwgpH/ugDxt+SxF3IB8G3e+Ic++Ta3RkfrMN4VZue/iyTVEH2bpbswf0F8FkKkjJxzW+xL+QVMnB0pzhzvtsvyROLS1eFjFBY3zMOryQ5W3B5B2xXNGAm2z+twmSbrZLdVmcdjJmP02acJwEWx6QhNAL5glJfcxfyRWEiTKeXhDx//8UEqJ5N7IZ5In7N8iWli0CxOwp/bUnVF5BPHsCAe/Q2uvI2/xC4SLH7kl+bB8ynMHWX5qUVVMrLZ/B8iQBis8MnmO0SLjKArPv2e9/n85dv920/Nb7TbksrgefpWbmzBfBAxKs1Bd/z05d/Nt2NfOZ+GJQxi/Ordbndz9/0Hxrz37798+9F84xXiOgKKiF6CdF2rtSq9ITooks3JmbE3qRfpiWFOXHb16QeRqK6NphyNBBH/IiZizxIJxXB9hABMxe40F0t7EYh9LlJEouFYshwNczDbhSMzXTRYKHEpwhhSQ1OL4klsTlSM1TXLQIqiZRi24egWcEi0DjmlmHJdMy3TRoIScDwTmBbwuSXptxA/q0MzRFHznfp9OU8N29O4sp9AtzRahqKu4BqmBkyRAvpQcZfP/0rtt58//ohu7tazP+wvnz/8dS2Ga3z/8RbDlP7h7tvdccYHPtLeBpi6DwjTujF0vbyCqmSyFg0NKyJZtz2L5OByzswiyWyLhoQ/+wjQl66OKGC/1tXxNN+sRbr73rQWSuaAuMCoAKNH0YCvmbZe+2F5FQCgUUtAnQxMXwP+iIAKxB31A9FXOXJoW3icRYWflUQBSeSTpiR2iWvGd+e668pWmEtwPxJAga9RTxwVcGyRMWVBQn3UjpCvhli65XOSrnA+8MDrQZOAJevhwsCBupTjaIJwPxvaBNfDENbjE8KMgRcDbWBG2LAYi8XcdxzZYsydmWP3xRwmF/J/+pUQ8zdu4FYSAnG3WCSpGFXU9xJ5IWxaIsQuM/kSzTzbsnvaXoDpTm2JxKyMr2UrNH6Zft/BnSTAcRp+J7l36bjlApYSvMliAYZbscm12z5Fuc2DOqFVviLDrDc/A7hj11W35mecg6ebW4l2hay7lfZIfP17kfgtw+K1F+XwPGHzUewK28FTJJ+hybU6PiO2sADLFUSnOwOuGMi3aiItw+6LL4SxRucMBcPBS+WMKkjA81kS162DSbwrn52ENYyBWMNuEGDPbsMwRCPKq2GLozcM13E5djqTHcMchi0s29Ysr8l0bPGWwPNhEllZ4lfHJMDjavsXaoMSq9TD0zq3djqUS/aajdt7L1sDcYm1l0sUK/RLxuWsKqWDeywuES2Or5BLTN99EVyiWoFrMC5x9nJJZ4HL4kKlrbH3kjPrAjwMlzi54nG4yDUhDjFOzSH7pS3FNn/trbgErX9oDrm00ESE7gHz7DlEtdLtifaQzvqIsIco9mfqjUNE58criXzwKDjQBHhJBbBRPYeGmBj/asIePHdqiyF6P15HzINvWHxE5ekXQwypfOVhD747wVUS3SKXyIeqXJHdjnCjhj2YF3P9MbIx0vaPk44HtuDnNR37F4Q9y9N8p0kQ9vikF2VBGI/rsX5dl9c7B5aFzYtt/hiOsJigB1zEzn09XOEPxBX+6bniYos/hiucV8wVPt4r3P65Ao/rOafliovt/Riu8EiNs3q4g/l6uMIfiCv803PFxd5+tjrFMEKRazYRun+EUORr1D5HuYY3wA9N6JecgQuhM/KIa/VP6L6FCN07LaGLTolXSOjANhhC14DZ1uJzIsSuUtd6oBBObNEZBP6t08P/JTcGw7/BlSEExoUrVKT/QfYK/+R7hXWg/RStSZQ9fyMTmqynWauzYoM6rwCWVwpuqhhFZxiFZZK2EM6TiTGGaVn1xqQ+l2XVNcalqvXbNNLQdClaMN/HEmd+S4NSWlSs9/6kym1HWSrYw4ONbshrXC6J6zhbzM7xHUqBbNheG0rIX/nMTHFWfxij3BjlFHAisH3XiroImGytVkPItvlx0bcA+9iRxmS7SijVG/GJFq/bZL3dxfKWPmcPLw1SUTH8cO2PLUc27AjgcmaGnqHAZTowopplcSiMoHEZGKEhx2MBiWhoqboWvEQkafCuvFwkkdkMuKrD9+geUYgLO3+DD9GWvMkhzVWUKqEr94hp6sBCByrLG98HaI7R/6TWOx47Wl+v4CohCxohIHyqvYasKLKswcyesshDv88V1zmGFKPP286s6Aod/D7RQc+oy5uf5gW146bXYJ56gS4y9F/LKFy2PO6p5jmFmGpIdfCiynmBRGXx+scgigm2cO82xfkn/YPyTgbF2yyiGLaRitrcj048eQ31/G3y1gLaRIlob53z8wCbEZ//oNYML+vVL88/9vMf1EKv/9f7iOW+TZo8RnPS6StaL3A+Fe0SMUt2GdsupPclk/cI6aOrQ/+zlWN9Fq0gUhTRVsC25onnJOGG2ae5DbrDLMpbcpzh/OT/YmWtZZZur4r+JC99uiIixpUzstvM8/57C9KAdx0+t8kTynNwpdJUhdOg++zASHP2mvPy9mXzNWrW6rlF11znsrIpY71lH61yytRT500pvSUXWbIuYxM2mwl+v14c5O5BxcDYFn0dkvBVbXQKsa/FarWW/8qNGr0b/a4NYNVNdbrPEXjXnudcG+yyPsVYnsjJVY5kiO6M6uWpc8VxbjEFXpGUphiVVwQDOTC0jvzBFy9W9Nj1xR626Kg/WEDxWgSUGT3wjvRX1H9Hp5Wt4StFiJo0cRu7PUrSTFFwqS7+hnusUBsRYyCqzgmw6LSGuBF3QEikJjFWZ6UwDOKQ9jVOYYbjabBqwwNKHEebLebUFqmHwQLBqcRspkwTNq4nEfmRyUOmY/rmvB+ZBwCuRB3QJY3ZDEMi9YDBGgrZl0iEPZEIJehzUYPKNe1qkF9tAIOLQp6qLJSjV/8Rg4Zpc5SuWI1RUm2I+DibPKB4YNYFqrSL9Ib+F1e7CvccGE5IuQechnusk3MPt0/4nft6A5MJHwDe/nGH5pZmdzIm0qOMObk1a71bzRrEH9E7WrQnJr680keKowMQC8X4z1WA/WNtYxFXE352/JD7BK/8HRusO5XEQ/v3hXGym7dLQC2mnrmn6655dUC7xbe2rlsVSx3Z68jTfLaIjCkpIlOqA4zMM1gZGbu5n/zRdEhF5EaCqOhL7h6uzkw2sOySfqEvKX2ZFouUsgpFcuIaTqCW2Ut6Iq6iRXnZllxPg/VDM42JX6C1Twu8K0JdLlQlK1XG7b/uyalK2j+7J6paBU91ikBEJhJEHpgzn8H1A+kdj31CkcQMwKOa8VTuoyWQLRNJGbwr48Y07ve7d09Cio7lOp57CCmawDJtaxiAk1GirAQb4KsoKFAi+pgmmDIqmQ9N1vI+meOozbv/Bw==</diagram></mxfile>
\ No newline at end of file
database/engine/journalfile.c
+313 -96
@@ -52,7 +52,7 @@ static void update_metric_retention_and_granularity_by_uuid(
52 mrg_metric_release(main_mrg, metric);
53 }
54
55 -static void flush_transaction_buffer_cb(uv_fs_t* req)
55 +static void wal_flush_transaction_buffer_cb(uv_fs_t* req)
56 {
57 worker_is_busy(RRDENG_FLUSH_TRANSACTION_BUFFER_CB);
58
@@ -99,7 +99,7 @@ void wal_flush_transaction_buffer(struct rrdengine_instance *ctx, struct rrdengi
99
100 io_descr->iov = uv_buf_init((void *)io_descr->buf, wal->buf_size);
101 ret = uv_fs_write(loop, &io_descr->req, journalfile->file, &io_descr->iov, 1,
102 - journalfile->pos, flush_transaction_buffer_cb);
102 + journalfile->pos, wal_flush_transaction_buffer_cb);
103 fatal_assert(-1 != ret);
104 journalfile->pos += wal->buf_size;
105 ctx->disk_space += wal->buf_size;
@@ -107,26 +107,256 @@ void wal_flush_transaction_buffer(struct rrdengine_instance *ctx, struct rrdengi
107 ++ctx->stats.io_write_requests;
108 }
109
110 -void generate_journalfilepath_v2(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
110 +void journalfile_v2_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
111 {
112 (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION_V2,
113 datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
114 }
115
116 -void generate_journalfilepath(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
116 +void journalfile_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen)
117 {
118 (void) snprintfz(str, maxlen, "%s/" WALFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL WALFILE_EXTENSION,
119 datafile->ctx->dbfiles_path, datafile->tier, datafile->fileno);
120 }
121
122 -void journalfile_init(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
122 +static struct journal_v2_header *journalfile_v2_mounted_data_get(struct rrdengine_journalfile *journalfile, size_t *data_size) {
123 + struct journal_v2_header *j2_header = NULL;
124 +
125 + netdata_spinlock_lock(&journalfile->mmap.spinlock);
126 +
127 + if(!journalfile->mmap.data) {
128 + journalfile->mmap.data = mmap(NULL, journalfile->mmap.size, PROT_READ, MAP_SHARED, journalfile->mmap.fd, 0);
129 + if (journalfile->mmap.data == MAP_FAILED) {
130 + internal_fatal(true, "DBENGINE: failed to re-mmap() journal file v2");
131 + close(journalfile->mmap.fd);
132 + journalfile->mmap.fd = -1;
133 + journalfile->mmap.data = NULL;
134 + journalfile->mmap.size = 0;
135 +
136 + netdata_spinlock_lock(&journalfile->v2.spinlock);
137 + journalfile->v2.flags &= ~(JOURNALFILE_FLAG_IS_AVAILABLE | JOURNALFILE_FLAG_IS_MOUNTED);
138 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
139 +
140 + ++journalfile->datafile->ctx->stats.fs_errors;
141 + rrd_stat_atomic_add(&global_fs_errors, 1);
142 + }
143 + else {
144 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.journal_v2_mapped, 1, __ATOMIC_RELAXED);
145 +
146 + madvise_dontfork(journalfile->mmap.data, journalfile->mmap.size);
147 + madvise_dontdump(journalfile->mmap.data, journalfile->mmap.size);
148 + madvise_random(journalfile->mmap.data, journalfile->mmap.size);
149 + madvise_dontneed(journalfile->mmap.data, journalfile->mmap.size);
150 +
151 + netdata_spinlock_lock(&journalfile->v2.spinlock);
152 + journalfile->v2.flags |= JOURNALFILE_FLAG_IS_AVAILABLE | JOURNALFILE_FLAG_IS_MOUNTED;
153 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
154 + }
155 + }
156 +
157 + if(journalfile->mmap.data) {
158 + j2_header = journalfile->mmap.data;
159 +
160 + if (data_size)
161 + *data_size = journalfile->mmap.size;
162 + }
163 +
164 + netdata_spinlock_unlock(&journalfile->mmap.spinlock);
165 +
166 + return j2_header;
167 +}
168 +
169 +static bool journalfile_v2_mounted_data_unmount(struct rrdengine_journalfile *journalfile, bool have_locks) {
170 + bool unmounted = false;
171 +
172 + if(!have_locks) {
173 + netdata_spinlock_lock(&journalfile->mmap.spinlock);
174 + netdata_spinlock_lock(&journalfile->v2.spinlock);
175 + }
176 +
177 + if(!journalfile->v2.refcount && journalfile->mmap.data) {
178 + if (munmap(journalfile->mmap.data, journalfile->mmap.size)) {
179 + char path[RRDENG_PATH_MAX];
180 + journalfile_v2_generate_path(journalfile->datafile, path, sizeof(path));
181 + error("DBENGINE: failed to unmap index file '%s'", path);
182 + internal_fatal(true, "DBENGINE: failed to unmap file '%s'", path);
183 + ++journalfile->datafile->ctx->stats.fs_errors;
184 + rrd_stat_atomic_add(&global_fs_errors, 1);
185 + }
186 + else {
187 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.journal_v2_unmapped, 1, __ATOMIC_RELAXED);
188 + journalfile->mmap.data = NULL;
189 + journalfile->v2.flags &= ~JOURNALFILE_FLAG_IS_MOUNTED;
190 + }
191 +
192 + unmounted = true;
193 + }
194 +
195 + if(!have_locks) {
196 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
197 + netdata_spinlock_unlock(&journalfile->mmap.spinlock);
198 + }
199 +
200 + return unmounted;
201 +}
202 +
203 +struct journal_v2_header *journalfile_v2_data_acquire(struct rrdengine_journalfile *journalfile, size_t *data_size, time_t wanted_first_time_s, time_t wanted_last_time_s) {
204 + netdata_spinlock_lock(&journalfile->v2.spinlock);
205 +
206 + bool has_data = (journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE);
207 + bool is_mounted = (journalfile->v2.flags & JOURNALFILE_FLAG_IS_MOUNTED);
208 + bool do_we_need_it = false;
209 + bool unmount = false;
210 +
211 + if(has_data) {
212 + if (!wanted_first_time_s || !wanted_last_time_s ||
213 + is_page_in_time_range(journalfile->v2.first_time_s, journalfile->v2.last_time_s,
214 + wanted_first_time_s, wanted_last_time_s) == PAGE_IS_IN_RANGE) {
215 +
216 + journalfile->v2.refcount++;
217 +
218 + do_we_need_it = true;
219 + journalfile->v2.not_needed_counter = 0;
220 +
221 + if (!wanted_first_time_s && !wanted_last_time_s && !is_mounted)
222 + journalfile->v2.flags |= JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION;
223 + else
224 + journalfile->v2.flags &= ~JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION;
225 +
226 + }
227 + else if (is_mounted) {
228 + // this journal has data, but it does not match our query
229 +
230 + if (!journalfile->v2.refcount) {
231 + // this journal has no references
232 +
233 + if (!journalfile->v2.not_needed_counter)
234 + journalfile->v2.not_needed_since_s = now_monotonic_sec();
235 +
236 + if ((++journalfile->v2.not_needed_counter) % 100 == 0) {
237 + // at least 100 times it has been evaluated since last use
238 +
239 + if (now_monotonic_sec() - journalfile->v2.not_needed_since_s >= 120)
240 + // 2 minutes have passed since last use
241 + unmount = true;
242 + }
243 + }
244 + }
245 + }
246 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
247 +
248 + if(do_we_need_it)
249 + return journalfile_v2_mounted_data_get(journalfile, data_size);
250 +
251 + else if(unmount)
252 + journalfile_v2_mounted_data_unmount(journalfile, false);
253 +
254 + return NULL;
255 +}
256 +
257 +void journalfile_v2_data_release(struct rrdengine_journalfile *journalfile) {
258 + netdata_spinlock_lock(&journalfile->v2.spinlock);
259 +
260 + internal_fatal(!journalfile->mmap.data, "trying to release a journalfile without data");
261 + internal_fatal(journalfile->v2.refcount < 1, "trying to release a non-acquired journalfile");
262 +
263 + bool unmount = false;
264 +
265 + journalfile->v2.refcount--;
266 +
267 + if(journalfile->v2.refcount == 0) {
268 + journalfile->v2.not_needed_counter = 0;
269 +
270 + if(journalfile->v2.flags & JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION)
271 + unmount = true;
272 + }
273 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
274 +
275 + if(unmount)
276 + journalfile_v2_mounted_data_unmount(journalfile, false);
277 +}
278 +
279 +bool journalfile_v2_data_available(struct rrdengine_journalfile *journalfile) {
280 +
281 + netdata_spinlock_lock(&journalfile->v2.spinlock);
282 + bool has_data = (journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE);
283 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
284 +
285 + return has_data;
286 +}
287 +
288 +size_t journalfile_v2_data_size_get(struct rrdengine_journalfile *journalfile) {
289 +
290 + netdata_spinlock_lock(&journalfile->mmap.spinlock);
291 + size_t data_size = journalfile->mmap.size;
292 + netdata_spinlock_unlock(&journalfile->mmap.spinlock);
293 +
294 + return data_size;
295 +}
296 +
297 +void journalfile_v2_data_set(struct rrdengine_journalfile *journalfile, int fd, void *journal_data, uint32_t journal_data_size) {
298 + netdata_spinlock_lock(&journalfile->mmap.spinlock);
299 + netdata_spinlock_lock(&journalfile->v2.spinlock);
300 +
301 + internal_fatal(journalfile->mmap.fd != -1, "DBENGINE JOURNALFILE: trying to re-set journal fd");
302 + internal_fatal(journalfile->mmap.data, "DBENGINE JOURNALFILE: trying to re-set journal_data");
303 + internal_fatal(journalfile->v2.refcount, "DBENGINE JOURNALFILE: trying to re-set journal_data of referenced journalfile");
304 +
305 + journalfile->mmap.fd = fd;
306 + journalfile->mmap.data = journal_data;
307 + journalfile->mmap.size = journal_data_size;
308 + journalfile->v2.not_needed_since_s = now_monotonic_sec();
309 + journalfile->v2.flags |= JOURNALFILE_FLAG_IS_AVAILABLE | JOURNALFILE_FLAG_IS_MOUNTED;
310 +
311 + struct journal_v2_header *j2_header = journalfile->mmap.data;
312 + journalfile->v2.first_time_s = (time_t)(j2_header->start_time_ut / USEC_PER_SEC);
313 + journalfile->v2.last_time_s = (time_t)(j2_header->end_time_ut / USEC_PER_SEC);
314 +
315 + journalfile_v2_mounted_data_unmount(journalfile, true);
316 +
317 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
318 + netdata_spinlock_unlock(&journalfile->mmap.spinlock);
319 +}
320 +
321 +static void journalfile_v2_data_unmap_permanently(struct rrdengine_journalfile *journalfile) {
322 + bool has_references = false;
323 +
324 + do {
325 + if (has_references)
326 + sleep_usec(10 * USEC_PER_MS);
327 +
328 + netdata_spinlock_lock(&journalfile->mmap.spinlock);
329 + netdata_spinlock_lock(&journalfile->v2.spinlock);
330 +
331 + if(journalfile_v2_mounted_data_unmount(journalfile, true)) {
332 + close(journalfile->mmap.fd);
333 + journalfile->mmap.fd = -1;
334 + journalfile->mmap.data = NULL;
335 + journalfile->mmap.size = 0;
336 + journalfile->v2.first_time_s = 0;
337 + journalfile->v2.last_time_s = 0;
338 + journalfile->v2.flags = 0;
339 + }
340 + else {
341 + has_references = true;
342 + internal_error(true, "DBENGINE JOURNALFILE: waiting for journalfile to be available to unmap...");
343 + }
344 +
345 + netdata_spinlock_unlock(&journalfile->v2.spinlock);
346 + netdata_spinlock_unlock(&journalfile->mmap.spinlock);
347 +
348 + } while(has_references);
349 +}
350 +
351 +struct rrdengine_journalfile *journalfile_alloc_and_init(struct rrdengine_datafile *datafile)
352 {
124 - journalfile->file = (uv_file)0;
125 - journalfile->pos = 0;
353 + struct rrdengine_journalfile *journalfile = callocz(1, sizeof(struct rrdengine_journalfile));
354 journalfile->datafile = datafile;
127 - SET_JOURNAL_DATA(journalfile, 0);
128 - SET_JOURNAL_DATA_SIZE(journalfile, 0);
129 - journalfile->data = NULL;
355 + netdata_spinlock_init(&journalfile->mmap.spinlock);
356 + netdata_spinlock_init(&journalfile->v2.spinlock);
357 + journalfile->mmap.fd = -1;
358 + datafile->journalfile = journalfile;
359 + return journalfile;
360 }
361
362 static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file)
@@ -137,7 +367,7 @@ static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file)
367 uv_fs_t req;
368 ret = uv_fs_close(NULL, &req, file, NULL);
369 if (ret < 0) {
140 - generate_journalfilepath(datafile, path, sizeof(path));
370 + journalfile_generate_path(datafile, path, sizeof(path));
371 error("DBENGINE: uv_fs_close(%s): %s", path, uv_strerror(ret));
372 ++datafile->ctx->stats.fs_errors;
373 rrd_stat_atomic_add(&global_fs_errors, 1);
@@ -146,30 +376,17 @@ static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file)
376 return ret;
377 }
378
149 -int close_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
379 +int journalfile_close(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
380 {
151 - struct rrdengine_instance *ctx = datafile->ctx;
152 - char path[RRDENG_PATH_MAX];
153 -
154 - void *journal_data = GET_JOURNAL_DATA(journalfile);
155 - size_t journal_data_size = GET_JOURNAL_DATA_SIZE(journalfile);
156 -
157 - if (likely(journal_data)) {
158 - if (munmap(journal_data, journal_data_size)) {
159 - generate_journalfilepath_v2(datafile, path, sizeof(path));
160 - error("DBENGINE: failed to unmap journal index file for %s", path);
161 - ++ctx->stats.fs_errors;
162 - rrd_stat_atomic_add(&global_fs_errors, 1);
163 - }
164 - SET_JOURNAL_DATA(journalfile, 0);
165 - SET_JOURNAL_DATA_SIZE(journalfile, 0);
381 + if(journalfile_v2_data_available(journalfile)) {
382 + journalfile_v2_data_unmap_permanently(journalfile);
383 return 0;
384 }
385
386 return close_uv_file(datafile, journalfile->file);
387 }
388
172 -int unlink_journal_file(struct rrdengine_journalfile *journalfile)
389 +int journalfile_unlink(struct rrdengine_journalfile *journalfile)
390 {
391 struct rrdengine_datafile *datafile = journalfile->datafile;
392 struct rrdengine_instance *ctx = datafile->ctx;
@@ -177,7 +394,7 @@ int unlink_journal_file(struct rrdengine_journalfile *journalfile)
394 int ret;
395 char path[RRDENG_PATH_MAX];
396
180 - generate_journalfilepath(datafile, path, sizeof(path));
397 + journalfile_generate_path(datafile, path, sizeof(path));
398
399 ret = uv_fs_unlink(NULL, &req, path, NULL);
400 if (ret < 0) {
@@ -192,7 +409,7 @@ int unlink_journal_file(struct rrdengine_journalfile *journalfile)
409 return ret;
410 }
411
195 -int destroy_journal_file_unsafe(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
412 +int journalfile_destroy_unsafe(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
413 {
414 struct rrdengine_instance *ctx = datafile->ctx;
415 uv_fs_t req;
@@ -200,8 +417,8 @@ int destroy_journal_file_unsafe(struct rrdengine_journalfile *journalfile, struc
417 char path[RRDENG_PATH_MAX];
418 char path_v2[RRDENG_PATH_MAX];
419
203 - generate_journalfilepath(datafile, path, sizeof(path));
204 - generate_journalfilepath_v2(datafile, path_v2, sizeof(path));
420 + journalfile_generate_path(datafile, path, sizeof(path));
421 + journalfile_v2_generate_path(datafile, path_v2, sizeof(path));
422
423 if (journalfile->file) {
424 ret = uv_fs_ftruncate(NULL, &req, journalfile->file, 0, NULL);
@@ -234,19 +451,13 @@ int destroy_journal_file_unsafe(struct rrdengine_journalfile *journalfile, struc
451 ++ctx->stats.journalfile_deletions;
452 ++ctx->stats.journalfile_deletions;
453
237 - void *journal_data = GET_JOURNAL_DATA(journalfile);
238 - size_t journal_data_size = GET_JOURNAL_DATA_SIZE(journalfile);
239 -
240 - if (journal_data) {
241 - if (munmap(journal_data, journal_data_size)) {
242 - error("DBENGINE: failed to unmap index file %s", path_v2);
243 - }
244 - }
454 + if(journalfile_v2_data_available(journalfile))
455 + journalfile_v2_data_unmap_permanently(journalfile);
456
457 return ret;
458 }
459
249 -int create_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
460 +int journalfile_create(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
461 {
462 struct rrdengine_instance *ctx = datafile->ctx;
463 uv_fs_t req;
@@ -256,7 +467,7 @@ int create_journal_file(struct rrdengine_journalfile *journalfile, struct rrdeng
467 uv_buf_t iov;
468 char path[RRDENG_PATH_MAX];
469
259 - generate_journalfilepath(datafile, path, sizeof(path));
470 + journalfile_generate_path(datafile, path, sizeof(path));
471 fd = open_file_direct_io(path, O_CREAT | O_RDWR | O_TRUNC, &file);
472 if (fd < 0) {
473 ++ctx->stats.fs_errors;
@@ -286,7 +497,7 @@ int create_journal_file(struct rrdengine_journalfile *journalfile, struct rrdeng
497 uv_fs_req_cleanup(&req);
498 posix_memfree(superblock);
499 if (ret < 0) {
289 - destroy_journal_file_unsafe(journalfile, datafile);
500 + journalfile_destroy_unsafe(journalfile, datafile);
501 return ret;
502 }
503
@@ -297,7 +508,7 @@ int create_journal_file(struct rrdengine_journalfile *journalfile, struct rrdeng
508 return 0;
509 }
510
300 -static int check_journal_file_superblock(uv_file file)
511 +static int journalfile_check_superblock(uv_file file)
512 {
513 int ret;
514 struct rrdeng_jf_sb *superblock;
@@ -331,7 +542,7 @@ static int check_journal_file_superblock(uv_file file)
542 return ret;
543 }
544
334 -static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, void *buf, unsigned max_size)
545 +static void journalfile_restore_extent_metadata(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, void *buf, unsigned max_size)
546 {
547 static BITMAP256 page_error_map;
548 unsigned i, count, payload_length, descr_size;
@@ -407,8 +618,8 @@ static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrden
618 * Sets id to the current transaction id or to 0 if unknown.
619 * Returns size of transaction record or 0 for unknown size.
620 */
410 -static unsigned replay_transaction(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
411 - void *buf, uint64_t *id, unsigned max_size)
621 +static unsigned journalfile_replay_transaction(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
622 + void *buf, uint64_t *id, unsigned max_size)
623 {
624 unsigned payload_length, size_bytes;
625 int ret;
@@ -446,7 +657,7 @@ static unsigned replay_transaction(struct rrdengine_instance *ctx, struct rrdeng
657 switch (jf_header->type) {
658 case STORE_DATA:
659 debug(D_RRDENGINE, "Replaying transaction %"PRIu64"", jf_header->id);
449 - restore_extent_metadata(ctx, journalfile, buf + sizeof(*jf_header), payload_length);
660 + journalfile_restore_extent_metadata(ctx, journalfile, buf + sizeof(*jf_header), payload_length);
661 break;
662 default:
663 error("DBENGINE: unknown transaction type, skipping record.");
@@ -463,7 +674,7 @@ static unsigned replay_transaction(struct rrdengine_instance *ctx, struct rrdeng
674 * Page cache must already be initialized.
675 * Returns the maximum transaction id it discovered.
676 */
466 -static uint64_t iterate_transactions(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile)
677 +static uint64_t journalfile_iterate_transactions(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile)
678 {
679 uv_file file;
680 uint64_t file_size;//, data_file_size;
@@ -507,7 +718,7 @@ static uint64_t iterate_transactions(struct rrdengine_instance *ctx, struct rrde
718 unsigned max_size;
719
720 max_size = pos + size_bytes - pos_i;
510 - ret = replay_transaction(ctx, journalfile, buf + pos_i, &id, max_size);
721 + ret = journalfile_replay_transaction(ctx, journalfile, buf + pos_i, &id, max_size);
722 if (!ret) /* TODO: support transactions bigger than 4K */
723 /* unknown transaction size, move on to the next block */
724 pos_i = ALIGN_BYTES_FLOOR(pos_i + RRDENG_BLOCK_SIZE);
@@ -525,7 +736,7 @@ skip_file:
736 }
737
738 // Checks that the extent list checksum is valid
528 -static int check_journal_v2_extent_list (void *data_start, size_t file_size)
739 +static int journalfile_check_v2_extent_list (void *data_start, size_t file_size)
740 {
741 UNUSED(file_size);
742 uLong crc;
@@ -545,7 +756,7 @@ static int check_journal_v2_extent_list (void *data_start, size_t file_size)
756 }
757
758 // Checks that the metric list (UUIDs) checksum is valid
548 -static int check_journal_v2_metric_list(void *data_start, size_t file_size)
759 +static int journalfile_check_v2_metric_list(void *data_start, size_t file_size)
760 {
761 UNUSED(file_size);
762 uLong crc;
@@ -570,7 +781,7 @@ static int check_journal_v2_metric_list(void *data_start, size_t file_size)
781 // 2 Force rebuild
782 // 3 skip
783
573 -static int check_journal_v2_file(void *data_start, size_t file_size, uint32_t original_size)
784 +static int journalfile_v2_validate(void *data_start, size_t file_size, uint32_t original_size)
785 {
786 int rc;
787 uLong crc;
@@ -605,10 +816,10 @@ static int check_journal_v2_file(void *data_start, size_t file_size, uint32_t or
816 return 1;
817 }
818
608 - rc = check_journal_v2_extent_list(data_start, file_size);
819 + rc = journalfile_check_v2_extent_list(data_start, file_size);
820 if (rc) return 1;
821
611 - rc = check_journal_v2_metric_list(data_start, file_size);
822 + rc = journalfile_check_v2_metric_list(data_start, file_size);
823 if (rc) return 1;
824
825 if (!db_engine_journal_check)
@@ -667,7 +878,7 @@ static int check_journal_v2_file(void *data_start, size_t file_size, uint32_t or
878 return 0;
879 }
880
670 -int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
881 +int journalfile_v2_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile)
882 {
883 int ret, fd;
884 uint64_t file_size;
@@ -675,12 +886,12 @@ int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journa
886 struct stat statbuf;
887 uint32_t original_file_size = 0;
888
678 - generate_journalfilepath(datafile, path, sizeof(path));
889 + journalfile_generate_path(datafile, path, sizeof(path));
890 ret = stat(path, &statbuf);
891 if (!ret)
892 original_file_size = (uint32_t)statbuf.st_size;
893
683 - generate_journalfilepath_v2(datafile, path, sizeof(path));
894 + journalfile_v2_generate_path(datafile, path, sizeof(path));
895
896 fd = open(path, O_RDONLY);
897 if (fd < 0) {
@@ -713,10 +924,9 @@ int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journa
924 close(fd);
925 return 1;
926 }
716 - close(fd);
927
928 info("DBENGINE: checking integrity of '%s'", path);
719 - int rc = check_journal_v2_file(data_start, file_size, original_file_size);
929 + int rc = journalfile_v2_validate(data_start, file_size, original_file_size);
930 if (unlikely(rc)) {
931 if (rc == 2)
932 error_report("File %s needs to be rebuilt", path);
@@ -728,6 +938,7 @@ int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journa
938 if (unlikely(munmap(data_start, file_size)))
939 error("DBENGINE: failed to unmap '%s'", path);
940
941 + close(fd);
942 return rc;
943 }
944
@@ -738,6 +949,7 @@ int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journa
949 if (unlikely(munmap(data_start, file_size)))
950 error("DBENGINE: failed to unmap '%s'", path);
951
952 + close(fd);
953 return 1;
954 }
955
@@ -746,10 +958,6 @@ int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journa
958
959 struct journal_metric_list *metric = (struct journal_metric_list *) (data_start + j2_header->metric_offset);
960
749 - // Initialize the journal file to be able to access the data
750 - SET_JOURNAL_DATA(journalfile, data_start);
751 - SET_JOURNAL_DATA_SIZE(journalfile, file_size);
752 -
961 time_t header_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
962
963 time_t now_s = now_realtime_sec();
@@ -771,6 +979,9 @@ int load_journal_file_v2(struct rrdengine_instance *ctx, struct rrdengine_journa
979 info("DBENGINE: journal file '%s' loaded (size:%"PRIu64") with %u metrics in %d ms", path, file_size, entries,
980 (int) ((now_realtime_usec() - start_loading) / USEC_PER_MS));
981
982 + // Initialize the journal file to be able to access the data
983 + journalfile_v2_data_set(journalfile, fd, data_start, file_size);
984 +
985 // File is OK load it
986 return 0;
987 }
@@ -779,7 +990,7 @@ struct journal_metric_list_to_sort {
990 struct jv2_metrics_info *metric_info;
991 };
992
782 -static int journal_metric_compare (const void *item1, const void *item2)
993 +static int journalfile_metric_compare (const void *item1, const void *item2)
994 {
995 const struct jv2_metrics_info *metric1 = ((struct journal_metric_list_to_sort *) item1)->metric_info;
996 const struct jv2_metrics_info *metric2 = ((struct journal_metric_list_to_sort *) item2)->metric_info;
@@ -789,7 +1000,7 @@ static int journal_metric_compare (const void *item1, const void *item2)
1000
1001
1002 // Write list of extents for the journalfile
792 -void *journal_v2_write_extent_list(Pvoid_t JudyL_extents_pos, void *data)
1003 +void *journalfile_v2_write_extent_list(Pvoid_t JudyL_extents_pos, void *data)
1004 {
1005 Pvoid_t *PValue;
1006 struct journal_extent_list *j2_extent_base = (void *) data;
@@ -810,7 +1021,7 @@ void *journal_v2_write_extent_list(Pvoid_t JudyL_extents_pos, void *data)
1021 return j2_extent_base + count;
1022 }
1023
813 -static int verify_journal_space(struct journal_v2_header *j2_header, void *data, uint32_t bytes)
1024 +static int journalfile_verify_space(struct journal_v2_header *j2_header, void *data, uint32_t bytes)
1025 {
1026 if ((unsigned long)(((uint8_t *) data - (uint8_t *) j2_header->data) + bytes) > (j2_header->total_file_size - sizeof(struct journal_v2_block_trailer)))
1027 return 1;
@@ -818,11 +1029,11 @@ static int verify_journal_space(struct journal_v2_header *j2_header, void *data,
1029 return 0;
1030 }
1031
821 -void *journal_v2_write_metric_page(struct journal_v2_header *j2_header, void *data, struct jv2_metrics_info *metric_info, uint32_t pages_offset)
1032 +void *journalfile_v2_write_metric_page(struct journal_v2_header *j2_header, void *data, struct jv2_metrics_info *metric_info, uint32_t pages_offset)
1033 {
1034 struct journal_metric_list *metric = (void *) data;
1035
825 - if (verify_journal_space(j2_header, data, sizeof(*metric)))
1036 + if (journalfile_verify_space(j2_header, data, sizeof(*metric)))
1037 return NULL;
1038
1039 uuid_copy(metric->uuid, *metric_info->uuid);
@@ -834,7 +1045,7 @@ void *journal_v2_write_metric_page(struct journal_v2_header *j2_header, void *da
1045 return ++metric;
1046 }
1047
837 -void *journal_v2_write_data_page_header(struct journal_v2_header *j2_header __maybe_unused, void *data, struct jv2_metrics_info *metric_info, uint32_t uuid_offset)
1048 +void *journalfile_v2_write_data_page_header(struct journal_v2_header *j2_header __maybe_unused, void *data, struct jv2_metrics_info *metric_info, uint32_t uuid_offset)
1049 {
1050 struct journal_page_header *data_page_header = (void *) data;
1051 uLong crc;
@@ -849,7 +1060,7 @@ void *journal_v2_write_data_page_header(struct journal_v2_header *j2_header __ma
1060 return ++data_page_header;
1061 }
1062
852 -void *journal_v2_write_data_page_trailer(struct journal_v2_header *j2_header __maybe_unused, void *data, void *page_header)
1063 +void *journalfile_v2_write_data_page_trailer(struct journal_v2_header *j2_header __maybe_unused, void *data, void *page_header)
1064 {
1065 struct journal_page_header *data_page_header = (void *) page_header;
1066 struct journal_v2_block_trailer *journal_trailer = (void *) data;
@@ -861,11 +1072,11 @@ void *journal_v2_write_data_page_trailer(struct journal_v2_header *j2_header __m
1072 return ++journal_trailer;
1073 }
1074
864 -void *journal_v2_write_data_page(struct journal_v2_header *j2_header, void *data, struct jv2_page_info *page_info)
1075 +void *journalfile_v2_write_data_page(struct journal_v2_header *j2_header, void *data, struct jv2_page_info *page_info)
1076 {
1077 struct journal_page_list *data_page = data;
1078
868 - if (verify_journal_space(j2_header, data, sizeof(*data_page)))
1079 + if (journalfile_verify_space(j2_header, data, sizeof(*data_page)))
1080 return NULL;
1081
1082 struct extent_io_data *ei = page_info->custom_data;
@@ -882,7 +1093,7 @@ void *journal_v2_write_data_page(struct journal_v2_header *j2_header, void *data
1093 }
1094
1095 // Must be recorded in metric_info->entries
885 -void *journal_v2_write_descriptors(struct journal_v2_header *j2_header, void *data, struct jv2_metrics_info *metric_info)
1096 +void *journalfile_v2_write_descriptors(struct journal_v2_header *j2_header, void *data, struct jv2_metrics_info *metric_info)
1097 {
1098 Pvoid_t *PValue;
1099
@@ -897,7 +1108,7 @@ void *journal_v2_write_descriptors(struct journal_v2_header *j2_header, void *da
1108 while ((PValue = JudyLFirstThenNext(JudyL_array, &index_time, &first))) {
1109 page_info = *PValue;
1110 // Write one descriptor and return the next data page location
900 - data_page = journal_v2_write_data_page(j2_header, (void *)data_page, page_info);
1111 + data_page = journalfile_v2_write_data_page(j2_header, (void *) data_page, page_info);
1112 if (NULL == data_page)
1113 break;
1114 }
@@ -910,9 +1121,9 @@ void *journal_v2_write_descriptors(struct journal_v2_header *j2_header, void *da
1121 // startup : if the migration is done during agent startup
1122 // this will allow us to optimize certain things
1123
913 -void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_unused, uint8_t type __maybe_unused,
914 - Pvoid_t JudyL_metrics, Pvoid_t JudyL_extents_pos,
915 - size_t number_of_extents, size_t number_of_metrics, size_t number_of_pages, void *user_data)
1124 +void journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_unused, uint8_t type __maybe_unused,
1125 + Pvoid_t JudyL_metrics, Pvoid_t JudyL_extents_pos,
1126 + size_t number_of_extents, size_t number_of_metrics, size_t number_of_pages, void *user_data)
1127 {
1128 char path[RRDENG_PATH_MAX];
1129 Pvoid_t *PValue;
@@ -923,7 +1134,7 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1134 time_t max_time_s = 0;
1135 struct jv2_metrics_info *metric_info;
1136
926 - generate_journalfilepath_v2(datafile, path, sizeof(path));
1137 + journalfile_v2_generate_path(datafile, path, sizeof(path));
1138
1139 info("DBENGINE: indexing file '%s': extents %zu, metrics %zu, pages %zu",
1140 path,
@@ -961,7 +1172,8 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1172 uint32_t trailer_offset = total_file_size;
1173 total_file_size += sizeof(struct journal_v2_block_trailer);
1174
964 - uint8_t *data_start = netdata_mmap(path, total_file_size, MAP_SHARED, 0, false);
1175 + int fd_v2;
1176 + uint8_t *data_start = netdata_mmap(path, total_file_size, MAP_SHARED, 0, false, &fd_v2);
1177 uint8_t *data = data_start;
1178
1179 memset(data_start, 0, extent_offset);
@@ -987,7 +1199,7 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1199
1200 struct journal_v2_block_trailer *journal_v2_trailer;
1201
990 - data = journal_v2_write_extent_list(JudyL_extents_pos, data_start + extent_offset);
1202 + data = journalfile_v2_write_extent_list(JudyL_extents_pos, data_start + extent_offset);
1203 internal_error(true, "DBENGINE: write extent list so far %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1204
1205 fatal_assert(data == data_start + extent_offset_trailer);
@@ -1025,7 +1237,7 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1237 j2_header.start_time_ut = min_time_s * USEC_PER_SEC;
1238 j2_header.end_time_ut = max_time_s * USEC_PER_SEC;
1239
1028 - qsort(&uuid_list[0], number_of_metrics, sizeof(struct journal_metric_list_to_sort), journal_metric_compare);
1240 + qsort(&uuid_list[0], number_of_metrics, sizeof(struct journal_metric_list_to_sort), journalfile_metric_compare);
1241 internal_error(true, "DBENGINE: traverse and qsort UUID %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1242
1243 uint32_t resize_file_to = total_file_size;
@@ -1037,7 +1249,7 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1249 uint32_t uuid_offset = data - data_start;
1250
1251 // Write the UUID we are processing
1040 - data = (void *) journal_v2_write_metric_page(&j2_header, data, metric_info, pages_offset);
1252 + data = (void *) journalfile_v2_write_metric_page(&j2_header, data, metric_info, pages_offset);
1253 if (unlikely(!data))
1254 break;
1255
@@ -1049,15 +1261,17 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1261 // Keep the page_list_header, to be used for migration when where agent is running
1262 metric_info->page_list_header = pages_offset;
1263 // Write page header
1052 - void *metric_page = journal_v2_write_data_page_header(&j2_header, data_start + pages_offset, metric_info, uuid_offset);
1264 + void *metric_page = journalfile_v2_write_data_page_header(&j2_header, data_start + pages_offset, metric_info,
1265 + uuid_offset);
1266
1267 // Start writing descr @ time
1055 - void *page_trailer = journal_v2_write_descriptors(&j2_header, metric_page, metric_info);
1268 + void *page_trailer = journalfile_v2_write_descriptors(&j2_header, metric_page, metric_info);
1269 if (unlikely(!page_trailer))
1270 break;
1271
1272 // Trailer (checksum)
1060 - uint8_t *next_page_address = journal_v2_write_data_page_trailer(&j2_header, page_trailer, data_start + pages_offset);
1273 + uint8_t *next_page_address = journalfile_v2_write_data_page_trailer(&j2_header, page_trailer,
1274 + data_start + pages_offset);
1275
1276 // Calculate start of the pages start for next descriptor
1277 pages_offset += (metric_info->number_of_pages * (sizeof(struct journal_page_list)) + sizeof(struct journal_page_header) + sizeof(struct journal_v2_block_trailer));
@@ -1094,8 +1308,8 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1308
1309 info("DBENGINE: migrated journal file '%s', file size %zu", path, total_file_size);
1310
1097 - SET_JOURNAL_DATA(journalfile, data_start);
1098 - SET_JOURNAL_DATA_SIZE(journalfile, total_file_size);
1311 + // msync(data_start, total_file_size, MS_SYNC);
1312 + journalfile_v2_data_set(journalfile, fd_v2, data_start, total_file_size);
1313
1314 internal_error(true, "DBENGINE: ACTIVATING NEW INDEX JNL %llu", (now_realtime_usec() - start_loading) / USEC_PER_MS);
1315 ctx->disk_space += total_file_size;
@@ -1127,8 +1341,8 @@ void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_
1341 ctx->disk_space += sizeof(struct journal_v2_header);
1342 }
1343
1130 -int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
1131 - struct rrdengine_datafile *datafile)
1344 +int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
1345 + struct rrdengine_datafile *datafile)
1346 {
1347 uv_fs_t req;
1348 uv_file file;
@@ -1138,11 +1352,13 @@ int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfi
1352
1353 // Do not try to load the latest file (always rebuild and live migrate)
1354 if (datafile->fileno != ctx->last_fileno) {
1141 - if (!load_journal_file_v2(ctx, journalfile, datafile))
1355 + if (!journalfile_v2_load(ctx, journalfile, datafile)) {
1356 +// unmap_journal_file(journalfile);
1357 return 0;
1358 + }
1359 }
1360
1145 - generate_journalfilepath(datafile, path, sizeof(path));
1361 + journalfile_generate_path(datafile, path, sizeof(path));
1362
1363 // If it is not the last file, open read only
1364 fd = open_file_direct_io(path, O_RDWR, &file);
@@ -1157,7 +1373,7 @@ int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfi
1373 goto error;
1374 file_size = ALIGN_BYTES_FLOOR(file_size);
1375
1160 - ret = check_journal_file_superblock(file);
1376 + ret = journalfile_check_superblock(file);
1377 if (ret) {
1378 info("DBENGINE: invalid journal file '%s' ; superblock check failed.", path);
1379 goto error;
@@ -1168,10 +1384,10 @@ int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfi
1384 journalfile->file = file;
1385 journalfile->pos = file_size;
1386
1171 - journalfile->data = netdata_mmap(path, file_size, MAP_SHARED, 0, !(datafile->fileno == ctx->last_fileno));
1387 + journalfile->data = netdata_mmap(path, file_size, MAP_SHARED, 0, !(datafile->fileno == ctx->last_fileno), NULL);
1388 info("DBENGINE: loading journal file '%s' using %s.", path, journalfile->data?"MMAP":"uv_fs_read");
1389
1174 - max_id = iterate_transactions(ctx, journalfile);
1390 + max_id = journalfile_iterate_transactions(ctx, journalfile);
1391
1392 ctx->commit_log.transaction_id = MAX(ctx->commit_log.transaction_id, max_id + 1);
1393
@@ -1185,7 +1401,8 @@ int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfi
1401 return 0;
1402 }
1403
1188 - pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->page_type, do_migrate_to_v2_callback, (void *) datafile->journalfile);
1404 + pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->page_type,
1405 + journalfile_migrate_to_v2_callback, (void *) datafile->journalfile);
1406
1407 if (is_last_file)
1408 ctx->create_new_datafile_pair = true;
database/engine/journalfile.h
+42 -14
@@ -17,13 +17,34 @@ struct rrdengine_journalfile;
17
18 #define is_descr_journal_v2(descr) ((descr)->extent_entry != NULL)
19
20 +typedef enum __attribute__ ((__packed__)) {
21 + JOURNALFILE_FLAG_IS_AVAILABLE = (1 << 0),
22 + JOURNALFILE_FLAG_IS_MOUNTED = (1 << 1),
23 + JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION = (1 << 2),
24 +} JOURNALFILE_FLAGS;
25 +
26 /* only one event loop is supported for now */
27 struct rrdengine_journalfile {
28 + struct {
29 + SPINLOCK spinlock;
30 + void *data; // MMAPed file of journal v2
31 + uint32_t size; // Total file size mapped
32 + int fd;
33 + } mmap;
34 +
35 + struct {
36 + SPINLOCK spinlock;
37 + JOURNALFILE_FLAGS flags;
38 + int32_t refcount;
39 + time_t first_time_s;
40 + time_t last_time_s;
41 + size_t not_needed_counter;
42 + time_t not_needed_since_s;
43 + } v2;
44 +
45 uv_file file;
46 uint64_t pos;
47 void *data;
25 - void *journal_data; // MMAPed file of journal v2
26 - uint32_t journal_data_size; // Total file size mapped
48 struct rrdengine_datafile *datafile;
49 };
50
@@ -110,20 +131,27 @@ struct transaction_commit_log {
131
132 struct wal;
133
113 -void generate_journalfilepath(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
114 -void generate_journalfilepath_v2(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
115 -void journalfile_init(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
134 +void journalfile_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
135 +void journalfile_v2_generate_path(struct rrdengine_datafile *datafile, char *str, size_t maxlen);
136 +struct rrdengine_journalfile *journalfile_alloc_and_init(struct rrdengine_datafile *datafile);
137 void wal_flush_transaction_buffer(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, struct wal *wal, uv_loop_t *loop);
117 -int close_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
118 -int unlink_journal_file(struct rrdengine_journalfile *journalfile);
119 -int destroy_journal_file_unsafe(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
120 -int create_journal_file(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
121 -int load_journal_file(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
122 - struct rrdengine_datafile *datafile);
138 +int journalfile_close(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
139 +int journalfile_unlink(struct rrdengine_journalfile *journalfile);
140 +int journalfile_destroy_unsafe(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
141 +int journalfile_create(struct rrdengine_journalfile *journalfile, struct rrdengine_datafile *datafile);
142 +int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
143 + struct rrdengine_datafile *datafile);
144 void init_commit_log(struct rrdengine_instance *ctx);
145
125 -void do_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_unused, uint8_t type __maybe_unused,
126 - Pvoid_t JudyL_metrics, Pvoid_t JudyL_extents_pos,
127 - size_t number_of_extents, size_t number_of_metrics, size_t number_of_pages, void *user_data);
146 +void journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno __maybe_unused, uint8_t type __maybe_unused,
147 + Pvoid_t JudyL_metrics, Pvoid_t JudyL_extents_pos,
148 + size_t number_of_extents, size_t number_of_metrics, size_t number_of_pages, void *user_data);
149 +
150 +
151 +bool journalfile_v2_data_available(struct rrdengine_journalfile *journalfile);
152 +size_t journalfile_v2_data_size_get(struct rrdengine_journalfile *journalfile);
153 +void journalfile_v2_data_set(struct rrdengine_journalfile *journalfile, int fd, void *journal_data, uint32_t journal_data_size);
154 +struct journal_v2_header *journalfile_v2_data_acquire(struct rrdengine_journalfile *journalfile, size_t *data_size, time_t wanted_first_time_s, time_t wanted_last_time_s);
155 +void journalfile_v2_data_release(struct rrdengine_journalfile *journalfile);
156
157 #endif /* NETDATA_JOURNALFILE_H */
\ No newline at end of file
database/engine/pagecache.c
+25 -29
@@ -12,7 +12,7 @@ struct rrdeng_cache_efficiency_stats rrdeng_cache_efficiency_stats = {};
12 static void main_cache_free_clean_page_callback(PGC *cache __maybe_unused, PGC_ENTRY entry __maybe_unused)
13 {
14 // Release storage associated with the page
15 - freez(entry.data);
15 + dbengine_page_free(entry.data, entry.size);
16 }
17
18 static void main_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PGC_ENTRY *entries_array __maybe_unused, PGC_PAGE **pages_array __maybe_unused, size_t entries __maybe_unused)
@@ -73,7 +73,7 @@ static void open_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PGC_
73
74 static void extent_cache_free_clean_page_callback(PGC *cache __maybe_unused, PGC_ENTRY entry __maybe_unused)
75 {
76 - freez(entry.data);
76 + dbengine_extent_free(entry.data, entry.size);
77 }
78
79 static void extent_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PGC_ENTRY *entries_array __maybe_unused, PGC_PAGE **pages_array __maybe_unused, size_t entries __maybe_unused)
@@ -81,13 +81,7 @@ static void extent_cache_flush_dirty_page_callback(PGC *cache __maybe_unused, PG
81 ;
82 }
83
84 -typedef enum {
85 - PAGE_IS_IN_THE_PAST = -1,
86 - PAGE_IS_IN_RANGE = 0,
87 - PAGE_IS_IN_THE_FUTURE = 1,
88 -} TIME_RANGE_COMPARE;
89 -
90 -static inline TIME_RANGE_COMPARE is_page_in_time_range(time_t page_first_time_s, time_t page_last_time_s, time_t wanted_start_time_s, time_t wanted_end_time_s) {
84 +inline TIME_RANGE_COMPARE is_page_in_time_range(time_t page_first_time_s, time_t page_last_time_s, time_t wanted_start_time_s, time_t wanted_end_time_s) {
85 // page_first_time_s <= wanted_end_time_s && page_last_time_s >= wanted_start_time_s
86
87 if(page_last_time_s < wanted_start_time_s)
@@ -465,8 +459,8 @@ static size_t list_has_time_gaps(
459 return gaps;
460 }
461
468 -typedef void (*page_found_callback)(PGC_PAGE *page, void *data);
469 -size_t get_page_list_from_journal_v2(struct rrdengine_instance *ctx, METRIC *metric, usec_t start_time_ut, usec_t end_time_ut, page_found_callback callback, void *callback_data) {
462 +typedef void (*page_found_callback_t)(PGC_PAGE *page, void *data);
463 +static size_t get_page_list_from_journal_v2(struct rrdengine_instance *ctx, METRIC *metric, usec_t start_time_ut, usec_t end_time_ut, page_found_callback_t callback, void *callback_data) {
464 uuid_t *uuid = mrg_metric_uuid(main_mrg, metric);
465 Word_t metric_id = mrg_metric_id(main_mrg, metric);
466
@@ -478,32 +472,29 @@ size_t get_page_list_from_journal_v2(struct rrdengine_instance *ctx, METRIC *met
472 uv_rwlock_rdlock(&ctx->datafiles.rwlock);
473 struct rrdengine_datafile *datafile;
474 for(datafile = ctx->datafiles.first; datafile ; datafile = datafile->next) {
481 - struct journal_v2_header *journal_header = (struct journal_v2_header *) GET_JOURNAL_DATA(datafile->journalfile);
482 -
483 - if (!journal_header)
475 + struct journal_v2_header *j2_header = journalfile_v2_data_acquire(datafile->journalfile, NULL,
476 + wanted_start_time_s,
477 + wanted_end_time_s);
478 + if (unlikely(!j2_header))
479 continue;
480
486 - time_t journal_start_time_s = (time_t)(journal_header->start_time_ut / USEC_PER_SEC);
487 - time_t journal_end_time_s = (time_t)(journal_header->end_time_ut / USEC_PER_SEC);
488 -
489 - // is the datafile within our time-range?
490 - TIME_RANGE_COMPARE jrc = is_page_in_time_range(journal_start_time_s, journal_end_time_s, wanted_start_time_s, wanted_end_time_s);
491 - if(jrc != PAGE_IS_IN_RANGE)
492 - continue;
481 + time_t journal_start_time_s = (time_t)(j2_header->start_time_ut / USEC_PER_SEC);
482
483 // the datafile possibly contains useful data for this query
484
496 - size_t journal_metric_count = (size_t)journal_header->metric_count;
497 - struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) journal_header + journal_header->metric_offset);
485 + size_t journal_metric_count = (size_t)j2_header->metric_count;
486 + struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
487 struct journal_metric_list *uuid_entry = bsearch(uuid,uuid_list,journal_metric_count,sizeof(*uuid_list), journal_metric_uuid_compare);
488
500 - if (unlikely(!uuid_entry))
489 + if (unlikely(!uuid_entry)) {
490 // our UUID is not in this datafile
491 + journalfile_v2_data_release(datafile->journalfile);
492 continue;
493 + }
494
504 - struct journal_page_header *page_list_header = (struct journal_page_header *) ((uint8_t *) journal_header + uuid_entry->page_offset);
495 + struct journal_page_header *page_list_header = (struct journal_page_header *) ((uint8_t *) j2_header + uuid_entry->page_offset);
496 struct journal_page_list *page_list = (struct journal_page_list *)((uint8_t *) page_list_header + sizeof(*page_list_header));
506 - struct journal_extent_list *extent_list = (void *)((uint8_t *)journal_header + journal_header->extent_offset);
497 + struct journal_extent_list *extent_list = (void *)((uint8_t *)j2_header + j2_header->extent_offset);
498 uint32_t uuid_page_entries = page_list_header->entries;
499
500 for (uint32_t index = 0; index < uuid_page_entries; index++) {
@@ -555,6 +546,8 @@ size_t get_page_list_from_journal_v2(struct rrdengine_instance *ctx, METRIC *met
546 pages_found++;
547 }
548 }
549 +
550 + journalfile_v2_data_release(datafile->journalfile);
551 }
552 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
553
@@ -1034,7 +1027,8 @@ void init_page_cache(void)
1027 main_cache_free_clean_page_callback,
1028 (size_t) rrdeng_pages_per_extent,
1029 main_cache_flush_dirty_page_callback,
1037 - 20, //
1030 + 10,
1031 + 10240, // if there are that many threads, evict so many at once!
1032 1000, //
1033 5, // don't delay too much other threads
1034 PGC_OPTIONS_AUTOSCALE, // AUTOSCALE = 2x max hot pages
@@ -1047,7 +1041,8 @@ void init_page_cache(void)
1041 open_cache_free_clean_page_callback,
1042 1,
1043 open_cache_flush_dirty_page_callback,
1050 - 20, //
1044 + 10,
1045 + 10240, // if there are that many threads, evict that many at once!
1046 1000, //
1047 3, // don't delay too much other threads
1048 PGC_OPTIONS_AUTOSCALE | PGC_OPTIONS_EVICT_PAGES_INLINE | PGC_OPTIONS_FLUSH_PAGES_INLINE,
@@ -1061,7 +1056,8 @@ void init_page_cache(void)
1056 extent_cache_free_clean_page_callback,
1057 1,
1058 extent_cache_flush_dirty_page_callback,
1064 - 5, //
1059 + 5,
1060 + 10, // it will lose up to that extents at once!
1061 100, //
1062 2, // don't delay too much other threads
1063 PGC_OPTIONS_AUTOSCALE | PGC_OPTIONS_EVICT_PAGES_INLINE | PGC_OPTIONS_FLUSH_PAGES_INLINE,
database/engine/pdc.c
+4 -4
@@ -1078,7 +1078,7 @@ static bool epdl_populate_pages_from_extent_data(
1078 if(worker)
1079 worker_is_busy(UV_EVENT_PAGE_POPULATION);
1080
1081 - void *page_data = dbengine_page_alloc(ctx, vd.page_length);
1081 + void *page_data = dbengine_page_alloc(vd.page_length);
1082
1083 if (unlikely(!vd.data_on_disk_valid)) {
1084 fill_page_with_nulls(page_data, vd.page_length, vd.type);
@@ -1120,7 +1120,7 @@ static bool epdl_populate_pages_from_extent_data(
1120 bool added = true;
1121 PGC_PAGE *page = pgc_page_add_and_acquire(main_cache, page_entry, &added);
1122 if (false == added) {
1123 - dbengine_page_free(page_data);
1123 + dbengine_page_free(page_data, vd.page_length);
1124 stats_cache_hit_while_inserting++;
1125 stats_data_from_main_cache++;
1126 }
@@ -1227,7 +1227,7 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1227 if(mmap_data != MAP_FAILED) {
1228 extent_compressed_data = mmap_data + (epdl->extent_offset - map_start);
1229
1230 - void *copied_extent_compressed_data = mallocz(epdl->extent_size);
1230 + void *copied_extent_compressed_data = dbengine_extent_alloc(epdl->extent_size);
1231 memcpy(copied_extent_compressed_data, extent_compressed_data, epdl->extent_size);
1232
1233 int ret = munmap(mmap_data, length);
@@ -1249,7 +1249,7 @@ void epdl_find_extent_and_populate_pages(struct rrdengine_instance *ctx, EPDL *e
1249 }, &added);
1250
1251 if (!added) {
1252 - freez(copied_extent_compressed_data);
1252 + dbengine_extent_free(copied_extent_compressed_data, epdl->extent_size);
1253 internal_fatal(epdl->extent_size != pgc_page_data_size(extent_cache, extent_cache_page),
1254 "DBENGINE: cache size does not match the expected size");
1255 }
database/engine/rrdengine.c
+33 -17
@@ -733,15 +733,24 @@ static inline struct rrdeng_cmd rrdeng_deq_cmd(void) {
733
734 // ----------------------------------------------------------------------------
735
736 -void *dbengine_page_alloc(struct rrdengine_instance *ctx __maybe_unused, size_t size) {
736 +void *dbengine_page_alloc(size_t size) {
737 void *page = mallocz(size);
738 return page;
739 }
740
741 -void dbengine_page_free(void *page) {
741 +void dbengine_page_free(void *page, size_t size __maybe_unused) {
742 freez(page);
743 }
744
745 +void *dbengine_extent_alloc(size_t size) {
746 + void *extent = mallocz(size);
747 + return extent;
748 +}
749 +
750 +void dbengine_extent_free(void *extent, size_t size __maybe_unused) {
751 + freez(extent);
752 +}
753 +
754 static void commit_data_extent(struct rrdengine_instance *ctx, struct extent_io_descriptor *xt_io_descr) {
755 unsigned count, payload_length, descr_size, size_bytes;
756 void *buf;
@@ -1059,15 +1068,15 @@ void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_dataf
1068 unsigned v2_count = 0;
1069 unsigned journalfile_count = 0;
1070 while (datafile) {
1062 - struct journal_v2_header *journal_header = (struct journal_v2_header *) GET_JOURNAL_DATA(datafile->journalfile);
1063 - if (!journal_header || !datafile->users.available) {
1071 + struct journal_v2_header *j2_header = journalfile_v2_data_acquire(datafile->journalfile, NULL, 0, 0);
1072 + if (!j2_header) {
1073 datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1074 continue;
1075 }
1076
1068 - time_t journal_start_time_s = (time_t) (journal_header->start_time_ut / USEC_PER_SEC);
1069 - size_t journal_metric_count = (size_t)journal_header->metric_count;
1070 - struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) journal_header + journal_header->metric_offset);
1077 + time_t journal_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC);
1078 + size_t journal_metric_count = (size_t)j2_header->metric_count;
1079 + struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
1080
1081 Word_t index = 0;
1082 bool first_then_next = true;
@@ -1087,6 +1096,7 @@ void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_dataf
1096 v2_count++;
1097 }
1098 journalfile_count++;
1099 + journalfile_v2_data_release(datafile->journalfile);
1100 datafile = datafile_release_and_acquire_next_for_retention(ctx, datafile);
1101 }
1102
@@ -1117,19 +1127,21 @@ void find_uuid_first_time(struct rrdengine_instance *ctx, struct rrdengine_dataf
1127 }
1128
1129 static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile_to_delete, struct rrdengine_datafile *first_datafile_remaining, bool worker) {
1130 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.metrics_retention_started, 1, __ATOMIC_RELAXED);
1131 +
1132 if(worker)
1133 worker_is_busy(UV_EVENT_ANALYZE_V2);
1134
1123 - struct rrdengine_journalfile *journal_file = datafile_to_delete->journalfile;
1124 - struct journal_v2_header *journal_header = (struct journal_v2_header *)GET_JOURNAL_DATA(journal_file);
1125 - struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) journal_header + journal_header->metric_offset);
1135 + struct rrdengine_journalfile *journalfile = datafile_to_delete->journalfile;
1136 + struct journal_v2_header *j2_header = journalfile_v2_data_acquire(journalfile, NULL, 0, 0);
1137 + struct journal_metric_list *uuid_list = (struct journal_metric_list *)((uint8_t *) j2_header + j2_header->metric_offset);
1138
1139 Pvoid_t metric_first_time_JudyL = (Pvoid_t) NULL;
1140 Pvoid_t *PValue;
1141
1142 unsigned count = 0;
1143 struct uuid_first_time_s *uuid_first_t_entry;
1132 - for (uint32_t index = 0; index < journal_header->metric_count; ++index) {
1144 + for (uint32_t index = 0; index < j2_header->metric_count; ++index) {
1145 METRIC *metric = mrg_metric_get_and_acquire(main_mrg, &uuid_list[index].uuid, (Word_t) ctx);
1146 if (!metric)
1147 continue;
@@ -1146,6 +1158,7 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
1158 count++;
1159 }
1160 }
1161 + journalfile_v2_data_release(journalfile);
1162
1163 info("DBENGINE: recalculating retention for %u metrics", count);
1164
@@ -1198,10 +1211,12 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1211 "it is in use currently by %u users.",
1212 ctx->dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno, datafile->users.lockers);
1213
1214 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_spin, 1, __ATOMIC_RELAXED);
1215 sleep_usec(1 * USEC_PER_SEC);
1216 }
1217 }
1218
1219 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_started, 1, __ATOMIC_RELAXED);
1220 info("DBENGINE: deleting data file '%s/"
1221 DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION
1222 "'.",
@@ -1220,15 +1235,15 @@ static void datafile_delete(struct rrdengine_instance *ctx, struct rrdengine_dat
1235 journal_file = datafile->journalfile;
1236 datafile_bytes = datafile->pos;
1237 journal_file_bytes = journal_file->pos;
1223 - deleted_bytes = GET_JOURNAL_DATA_SIZE(journal_file);
1238 + deleted_bytes = journalfile_v2_data_size_get(journal_file);
1239
1240 info("DBENGINE: deleting data and journal files to maintain disk quota");
1241 datafile_list_delete_unsafe(ctx, datafile);
1227 - ret = destroy_journal_file_unsafe(journal_file, datafile);
1242 + ret = journalfile_destroy_unsafe(journal_file, datafile);
1243 if (!ret) {
1229 - generate_journalfilepath(datafile, path, sizeof(path));
1244 + journalfile_generate_path(datafile, path, sizeof(path));
1245 info("DBENGINE: deleted journal file \"%s\".", path);
1231 - generate_journalfilepath_v2(datafile, path, sizeof(path));
1246 + journalfile_v2_generate_path(datafile, path, sizeof(path));
1247 info("DBENGINE: deleted journal file \"%s\".", path);
1248 deleted_bytes += journal_file_bytes;
1249 }
@@ -1377,9 +1392,10 @@ static void journal_v2_indexing_tp_worker(struct rrdengine_instance *ctx __maybe
1392 if(!available)
1393 continue;
1394
1380 - if (unlikely(!GET_JOURNAL_DATA(datafile->journalfile))) {
1395 + if (unlikely(!journalfile_v2_data_available(datafile->journalfile))) {
1396 info("DBENGINE: journal file %u is ready to be indexed", datafile->fileno);
1382 - pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->page_type, do_migrate_to_v2_callback, (void *) datafile->journalfile);
1397 + pgc_open_cache_to_journal_v2(open_cache, (Word_t) ctx, (int) datafile->fileno, ctx->page_type,
1398 + journalfile_migrate_to_v2_callback, (void *) datafile->journalfile);
1399 count++;
1400 }
1401
database/engine/rrdengine.h
+13 -7
@@ -31,11 +31,6 @@ struct rrdeng_cmd;
31
32 #define MAX_PAGES_PER_EXTENT (64) /* TODO: can go higher only when journal supports bigger than 4KiB transactions */
33
34 -#define GET_JOURNAL_DATA(x) __atomic_load_n(&(x)->journal_data, __ATOMIC_ACQUIRE)
35 -#define GET_JOURNAL_DATA_SIZE(x) __atomic_load_n(&(x)->journal_data_size, __ATOMIC_ACQUIRE)
36 -#define SET_JOURNAL_DATA(x, y) __atomic_store_n(&(x)->journal_data, (y), __ATOMIC_RELEASE)
37 -#define SET_JOURNAL_DATA_SIZE(x, y) __atomic_store_n(&(x)->journal_data_size, (y), __ATOMIC_RELEASE)
38 -
34 #define RRDENG_FILE_NUMBER_SCAN_TMPL "%1u-%10u"
35 #define RRDENG_FILE_NUMBER_PRINT_TMPL "%1.1u-%10.10u"
36
@@ -397,8 +392,11 @@ struct rrdengine_instance {
392
393 #define ctx_is_available_for_queries(ctx) (__atomic_load_n(&(ctx)->quiesce, __ATOMIC_RELAXED) == NO_QUIESCE)
394
400 -void *dbengine_page_alloc(struct rrdengine_instance *ctx, size_t size);
401 -void dbengine_page_free(void *page);
395 +void *dbengine_page_alloc(size_t size);
396 +void dbengine_page_free(void *page, size_t size);
397 +
398 +void *dbengine_extent_alloc(size_t size);
399 +void dbengine_extent_free(void *extent, size_t size);
400
401 int init_rrd_files(struct rrdengine_instance *ctx);
402 void finalize_rrd_files(struct rrdengine_instance *ctx);
@@ -446,4 +444,12 @@ typedef struct validated_page_descriptor {
444
445 VALIDATED_PAGE_DESCRIPTOR validate_extent_page_descr(const struct rrdeng_extent_page_descr *descr, time_t now_s, time_t overwrite_zero_update_every_s, bool have_read_error);
446
447 +typedef enum {
448 + PAGE_IS_IN_THE_PAST = -1,
449 + PAGE_IS_IN_RANGE = 0,
450 + PAGE_IS_IN_THE_FUTURE = 1,
451 +} TIME_RANGE_COMPARE;
452 +
453 +TIME_RANGE_COMPARE is_page_in_time_range(time_t page_first_time_s, time_t page_last_time_s, time_t wanted_start_time_s, time_t wanted_end_time_s);
454 +
455 #endif /* NETDATA_RRDENGINE_H */
database/engine/rrdengineapi.c
+9 -11
@@ -290,7 +290,7 @@ time_t point_in_time_s = (time_t)(point_in_time_ut / USEC_PER_SEC);
290 memcpy(pgc_page_data(page), data, PAGE_POINT_CTX_SIZE_BYTES(ctx));
291
292 // free data
293 - dbengine_page_free(page_entry.data);
293 + dbengine_page_free(page_entry.data, data_size);
294
295 handle->page_entries_max = pgc_page_data_size(main_cache, page) / PAGE_POINT_CTX_SIZE_BYTES(ctx);
296 }
@@ -338,7 +338,7 @@ static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle,
338 }
339
340 *data_size = size;
341 - return dbengine_page_alloc(ctx, size);
341 + return dbengine_page_alloc(size);
342 }
343
344 static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection_handle,
@@ -964,11 +964,11 @@ void rrdeng_prepare_exit(struct rrdengine_instance *ctx) {
964
965 static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_SIZE_STATS *stats)
966 {
967 - void *data_start = GET_JOURNAL_DATA(datafile->journalfile);
968 - if (unlikely(!data_start))
969 - return;
967 + struct journal_v2_header *j2_header = journalfile_v2_data_acquire(datafile->journalfile, NULL, 0, 0);
968 + void *data_start = (void *)j2_header;
969
971 - struct journal_v2_header *j2_header = (void *) data_start;
970 + if(unlikely(!j2_header))
971 + return;
972
973 stats->extents += j2_header->extent_count;
974
@@ -1026,6 +1026,8 @@ static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_S
1026 }
1027 metric++;
1028 }
1029 +
1030 + journalfile_v2_data_release(datafile->journalfile);
1031 }
1032
1033 RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) {
@@ -1034,11 +1036,7 @@ RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) {
1036 uv_rwlock_rdlock(&ctx->datafiles.rwlock);
1037 for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next) {
1038 stats.datafiles++;
1037 -
1038 - if (GET_JOURNAL_DATA(df->journalfile)) {
1039 - // FIXME: Rework statistics based only on V2
1040 - populate_v2_statistics(df, &stats);
1041 - }
1039 + populate_v2_statistics(df, &stats);
1040 }
1041 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);
1042
database/engine/rrdengineapi.h
+9
@@ -190,6 +190,15 @@ struct rrdeng_cache_efficiency_stats {
190 size_t pages_invalid_size_skipped;
191 size_t pages_invalid_update_every_fixed;
192 size_t pages_invalid_entries_fixed;
193 +
194 + // database events
195 + size_t journal_v2_mapped;
196 + size_t journal_v2_unmapped;
197 + size_t datafile_creation_started;
198 + size_t datafile_deletion_started;
199 + size_t datafile_deletion_spin;
200 + size_t journal_v2_indexing_started;
201 + size_t metrics_retention_started;
202 };
203
204 struct rrdeng_buffer_sizes {
database/rrddim.c
+2 -2
@@ -64,7 +64,7 @@ static void rrddim_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
64 size_t entries = st->entries;
65 if(!entries) entries = 5;
66
67 - rd->db = netdata_mmap(NULL, entries * sizeof(storage_number), MAP_PRIVATE, 1, false);
67 + rd->db = netdata_mmap(NULL, entries * sizeof(storage_number), MAP_PRIVATE, 1, false, NULL);
68 if(!rd->db) {
69 info("Failed to use memory mode ram for chart '%s', dimension '%s', falling back to alloc", rrdset_name(st), rrddim_name(rd));
70 ctr->memory_mode = RRD_MEMORY_MODE_ALLOC;
@@ -696,7 +696,7 @@ bool rrddim_memory_load_or_create_map_save(RRDSET *st, RRDDIM *rd, RRD_MEMORY_MO
696 snprintfz(fullfilename, FILENAME_MAX, "%s/%s.db", st->cache_dir, filename);
697
698 rd_on_file = (struct rrddim_map_save_v019 *)netdata_mmap(
699 - fullfilename, size, ((memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE), 1, false);
699 + fullfilename, size, ((memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE), 1, false, NULL);
700
701 if(unlikely(!rd_on_file)) return false;
702
database/rrdset.c
+1 -1
@@ -2095,7 +2095,7 @@ bool rrdset_memory_load_or_create_map_save(RRDSET *st, RRD_MEMORY_MODE memory_mo
2095
2096 unsigned long size = sizeof(struct rrdset_map_save_v019);
2097 struct rrdset_map_save_v019 *st_on_file = (struct rrdset_map_save_v019 *)netdata_mmap(
2098 - fullfilename, size, ((memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE), 0, false);
2098 + fullfilename, size, ((memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE), 0, false, NULL);
2099
2100 if(!st_on_file) return false;
2101
libnetdata/arrayalloc/arrayalloc.c
+1 -1
@@ -209,7 +209,7 @@ static void arrayalloc_add_page(ARAL *ar TRACE_ALLOCATIONS_FUNCTION_DEFINITION_P
209 char filename[FILENAME_MAX + 1];
210 snprintfz(filename, FILENAME_MAX, "%s/array_alloc.mmap/%s.%zu", *ar->cache_dir, ar->filename, ar->internal.file_number);
211 page->filename = strdupz(filename);
212 - page->data = netdata_mmap(page->filename, page->size, MAP_SHARED, 0, false);
212 + page->data = netdata_mmap(page->filename, page->size, MAP_SHARED, 0, false, NULL);
213 if (unlikely(!page->data))
214 fatal("Cannot allocate arrayalloc buffer of size %zu on filename '%s'", page->size, page->filename);
215 }
libnetdata/libnetdata.c
+25 -4
@@ -1184,6 +1184,14 @@ inline int madvise_sequential(void *mem, size_t len) {
1184 return ret;
1185 }
1186
1187 +inline int madvise_random(void *mem, size_t len) {
1188 + static int logger = 1;
1189 + int ret = madvise(mem, len, MADV_RANDOM);
1190 +
1191 + if (ret != 0 && logger-- > 0) error("madvise(MADV_RANDOM) failed.");
1192 + return ret;
1193 +}
1194 +
1195 inline int madvise_dontfork(void *mem, size_t len) {
1196 static int logger = 1;
1197 int ret = madvise(mem, len, MADV_DONTFORK);
@@ -1200,6 +1208,14 @@ inline int madvise_willneed(void *mem, size_t len) {
1208 return ret;
1209 }
1210
1211 +inline int madvise_dontneed(void *mem, size_t len) {
1212 + static int logger = 1;
1213 + int ret = madvise(mem, len, MADV_DONTNEED);
1214 +
1215 + if (ret != 0 && logger-- > 0) error("madvise(MADV_DONTNEED) failed.");
1216 + return ret;
1217 +}
1218 +
1219 inline int madvise_dontdump(void *mem __maybe_unused, size_t len __maybe_unused) {
1220 #if __linux__
1221 static int logger = 1;
@@ -1224,7 +1240,7 @@ inline int madvise_mergeable(void *mem __maybe_unused, size_t len __maybe_unused
1240 #endif
1241 }
1242
1227 -void *netdata_mmap(const char *filename, size_t size, int flags, int ksm, bool read_only)
1243 +void *netdata_mmap(const char *filename, size_t size, int flags, int ksm, bool read_only, int *open_fd)
1244 {
1245 // info("netdata_mmap('%s', %zu", filename, size);
1246
@@ -1281,15 +1297,20 @@ void *netdata_mmap(const char *filename, size_t size, int flags, int ksm, bool r
1297 else info("Cannot seek to beginning of file '%s'.", filename);
1298 }
1299
1284 - madvise_sequential(mem, size);
1300 + // madvise_sequential(mem, size);
1301 madvise_dontfork(mem, size);
1302 madvise_dontdump(mem, size);
1287 - if(flags & MAP_SHARED) madvise_willneed(mem, size);
1303 + // if(flags & MAP_SHARED) madvise_willneed(mem, size);
1304 if(ksm) madvise_mergeable(mem, size);
1305 }
1306
1307 cleanup:
1292 - if(fd != -1) close(fd);
1308 + if(fd != -1) {
1309 + if (open_fd)
1310 + *open_fd = fd;
1311 + else
1312 + close(fd);
1313 + }
1314 if(mem == MAP_FAILED) return NULL;
1315 errno = 0;
1316 return mem;
libnetdata/libnetdata.h
+3 -1
@@ -308,8 +308,10 @@ char *trim(char *s); // remove leading and trailing spaces; may return NULL
308 char *trim_all(char *buffer); // like trim(), but also remove duplicate spaces inside the string; may return NULL
309
310 int madvise_sequential(void *mem, size_t len);
311 +int madvise_random(void *mem, size_t len);
312 int madvise_dontfork(void *mem, size_t len);
313 int madvise_willneed(void *mem, size_t len);
314 +int madvise_dontneed(void *mem, size_t len);
315 int madvise_dontdump(void *mem, size_t len);
316 int madvise_mergeable(void *mem, size_t len);
317
@@ -347,7 +349,7 @@ void posix_memfree(void *ptr);
349 void json_escape_string(char *dst, const char *src, size_t size);
350 void json_fix_string(char *s);
351
350 -void *netdata_mmap(const char *filename, size_t size, int flags, int ksm, bool read_only);
352 +void *netdata_mmap(const char *filename, size_t size, int flags, int ksm, bool read_only, int *open_fd);
353 int netdata_munmap(void *ptr, size_t size);
354 int memory_file_save(const char *filename, void *mem, size_t size);
355