| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "backfill.h" |
| 4 | |
| 5 | struct backfill_request { |
| 6 | OBJECT_STATE_ID host_state_id; |
| 7 | RRDSET_ACQUIRED *rsa; |
| 8 | int32_t works; |
| 9 | int32_t successful; |
| 10 | int32_t failed; |
| 11 | backfill_callback_t cb; |
| 12 | struct backfill_request_data data; |
| 13 | }; |
| 14 | |
| 15 | struct backfill_dim_work { |
| 16 | RRDDIM_ACQUIRED *rda; |
| 17 | struct backfill_request *br; |
| 18 | }; |
| 19 | |
| 20 | DEFINE_JUDYL_TYPED(BACKFILL, struct backfill_dim_work *); |
| 21 | |
| 22 | static struct { |
| 23 | struct completion completion; |
| 24 | |
| 25 | SPINLOCK spinlock; |
| 26 | bool running; |
| 27 | Word_t id; |
| 28 | size_t queue_size; |
| 29 | BACKFILL_JudyLSet queue; |
| 30 | |
| 31 | size_t charts_added; |
| 32 | size_t callbacks_executed; |
| 33 | |
| 34 | ARAL *ar_br; |
| 35 | ARAL *ar_bdm; |
| 36 | |
| 37 | } backfill_globals = { |
| 38 | .spinlock = SPINLOCK_INITIALIZER, |
| 39 | .queue = { 0 }, |
| 40 | }; |
| 41 | |
| 42 | bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_request_data *data) { |
| 43 | bool rc = false; |
| 44 | size_t dimensions = dictionary_entries(st->rrddim_root_index); |
| 45 | if(!dimensions) |
| 46 | return rc; |
| 47 | |
| 48 | size_t added = 0; |
| 49 | struct backfill_dim_work **array = mallocz(dimensions * sizeof(*array)); |
| 50 | |
| 51 | if(backfill_globals.running) { |
| 52 | struct backfill_request *br = aral_callocz(backfill_globals.ar_br); |
| 53 | br->data = *data; |
| 54 | br->host_state_id = object_state_id(&st->rrdhost->state_id); |
| 55 | br->rsa = rrdset_find_and_acquire(st->rrdhost, string2str(st->id), true); |
| 56 | if(br->rsa) { |
| 57 | br->cb = cb; |
| 58 | |
| 59 | RRDDIM *rd; |
| 60 | rrddim_foreach_read(rd, st) { |
| 61 | if(added >= dimensions) |
| 62 | break; |
| 63 | |
| 64 | if (!rrddim_option_check(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS)) { |
| 65 | struct backfill_dim_work *bdm = aral_callocz(backfill_globals.ar_bdm); |
| 66 | bdm->rda = (RRDDIM_ACQUIRED *)dictionary_acquired_item_dup(st->rrddim_root_index, rd_dfe.item); |
| 67 | bdm->br = br; |
| 68 | br->works++; |
| 69 | array[added++] = bdm; |
| 70 | } |
| 71 | } |
| 72 | rrddim_foreach_done(rd); |
| 73 | } |
| 74 | |
| 75 | internal_fatal((size_t)br->works != added, "works and added are not the same"); |
| 76 | |
| 77 | if(added) { |
| 78 | spinlock_lock(&backfill_globals.spinlock); |
| 79 | |
| 80 | __atomic_add_fetch(&backfill_globals.charts_added, 1, __ATOMIC_RELAXED); |
| 81 | |
| 82 | for(size_t i = 0; i < added ;i++) { |
| 83 | backfill_globals.queue_size++; |
| 84 | BACKFILL_SET(&backfill_globals.queue, backfill_globals.id++, array[i]); |
| 85 | } |
| 86 | |
| 87 | spinlock_unlock(&backfill_globals.spinlock); |
| 88 | completion_mark_complete_a_job(&backfill_globals.completion); |
| 89 | |
| 90 | rc = true; |
| 91 | } |
| 92 | else { |
| 93 | // no dimensions added |
| 94 | rrdset_acquired_release(br->rsa); |
| 95 | aral_freez(backfill_globals.ar_br, br); |
| 96 | } |
| 97 | } |
| 98 | |
| 99 | freez(array); |
| 100 | return rc; |
| 101 | } |
| 102 | |
| 103 | bool backfill_execute(struct backfill_dim_work *bdm) { |
| 104 | RRDSET *st = rrdset_acquired_to_rrdset(bdm->br->rsa); |
| 105 | |
| 106 | if(!object_state_acquire(&st->rrdhost->state_id, bdm->br->host_state_id)) |
| 107 | return false; |
| 108 | |
| 109 | size_t success = 0; |
| 110 | RRDDIM *rd = rrddim_acquired_to_rrddim(bdm->rda); |
| 111 | |
| 112 | for (size_t tier = 1; tier < nd_profile.storage_tiers; tier++) |
| 113 | if (backfill_tier_from_smaller_tiers(rd, tier, now_realtime_sec())) |
| 114 | success++; |
| 115 | |
| 116 | if (success > 0) |
| 117 | rrddim_option_set(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS); |
| 118 | |
| 119 | object_state_release(&st->rrdhost->state_id); |
| 120 | return success > 0; |
| 121 | } |
| 122 | |
| 123 | static void backfill_dim_work_free(bool successful, struct backfill_dim_work *bdm) { |
| 124 | struct backfill_request *br = bdm->br; |
| 125 | |
| 126 | if(successful) |
| 127 | __atomic_add_fetch(&br->successful, 1, __ATOMIC_RELAXED); |
| 128 | else |
| 129 | __atomic_add_fetch(&br->failed, 1, __ATOMIC_RELAXED); |
| 130 | |
| 131 | int32_t works = __atomic_sub_fetch(&br->works, 1, __ATOMIC_RELAXED); |
| 132 | internal_fatal(works < 0, "negative backfill jobs"); |
| 133 | |
| 134 | if(works == 0) { |
| 135 | // we are the last dimension of the chart |
| 136 | |
| 137 | if(br->cb) { |
| 138 | __atomic_add_fetch(&backfill_globals.callbacks_executed, 1, __ATOMIC_RELAXED); |
| 139 | |
| 140 | br->cb( |
| 141 | __atomic_load_n(&br->successful, __ATOMIC_RELAXED), |
| 142 | __atomic_load_n(&br->failed, __ATOMIC_RELAXED), |
| 143 | &br->data); |
| 144 | } |
| 145 | |
| 146 | rrdset_acquired_release(br->rsa); |
| 147 | aral_freez(backfill_globals.ar_br, br); |
| 148 | } |
| 149 | |
| 150 | rrddim_acquired_release(bdm->rda); |
| 151 | aral_freez(backfill_globals.ar_bdm, bdm); |
| 152 | } |
| 153 | |
| 154 | #define LOG_WARNING_EVERY 10 |
| 155 | |
| 156 | void backfill_worker_thread(void *ptr) { |
| 157 | bool main_thread = (ptr == (void *)0x01); |
| 158 | size_t warning = LOG_WARNING_EVERY; |
| 159 | bool timeout = false; |
| 160 | |
| 161 | worker_register("BACKFILL"); |
| 162 | |
| 163 | worker_register_job_name(0, "get"); |
| 164 | worker_register_job_name(1, "backfill"); |
| 165 | worker_register_job_custom_metric(2, "backfill queue size", "dimensions", WORKER_METRIC_ABSOLUTE); |
| 166 | |
| 167 | size_t job_id = 0, queue_size = 0; |
| 168 | while(!nd_thread_signaled_to_cancel() && service_running(SERVICE_COLLECTORS|SERVICE_STREAMING)) { |
| 169 | worker_is_busy(0); |
| 170 | spinlock_lock(&backfill_globals.spinlock); |
| 171 | Word_t idx = 0; |
| 172 | struct backfill_dim_work *bdm = BACKFILL_FIRST(&backfill_globals.queue, &idx); |
| 173 | if(bdm) { |
| 174 | backfill_globals.queue_size--; |
| 175 | BACKFILL_DEL(&backfill_globals.queue, idx); |
| 176 | } |
| 177 | queue_size = backfill_globals.queue_size; |
| 178 | spinlock_unlock(&backfill_globals.spinlock); |
| 179 | |
| 180 | if(bdm) { |
| 181 | warning = LOG_WARNING_EVERY; |
| 182 | worker_is_busy(1); |
| 183 | bool success = backfill_execute(bdm); |
| 184 | backfill_dim_work_free(success, bdm); |
| 185 | continue; |
| 186 | } |
| 187 | else if(main_thread && timeout) { |
| 188 | size_t added = __atomic_load_n(&backfill_globals.charts_added, __ATOMIC_RELAXED); |
| 189 | size_t executed = __atomic_load_n(&backfill_globals.callbacks_executed, __ATOMIC_RELAXED); |
| 190 | |
| 191 | if(executed != added && --warning == 0) { |
| 192 | warning = LOG_WARNING_EVERY; |
| 193 | |
| 194 | nd_log(NDLS_DAEMON, NDLP_WARNING, |
| 195 | "BACKFILL: the queue is empty, but the commands executed %zu is not equal to the commands added %zu", |
| 196 | executed, added); |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | worker_set_metric(2, (NETDATA_DOUBLE)queue_size); |
| 201 | |
| 202 | worker_is_idle(); |
| 203 | size_t new_job_id = completion_wait_for_a_job_with_timeout(&backfill_globals.completion, job_id, 1000); |
| 204 | timeout = new_job_id == job_id; |
| 205 | job_id = new_job_id; |
| 206 | } |
| 207 | |
| 208 | worker_unregister(); |
| 209 | } |
| 210 | |
| 211 | void backfill_thread(void *ptr) { |
| 212 | struct netdata_static_thread *static_thread = ptr; |
| 213 | if(!static_thread) return; |
| 214 | |
| 215 | nd_thread_tag_set("BACKFILL[0]"); |
| 216 | |
| 217 | completion_init(&backfill_globals.completion); |
| 218 | BACKFILL_INIT(&backfill_globals.queue); |
| 219 | backfill_globals.ar_br = aral_by_size_acquire(sizeof(struct backfill_request)); |
| 220 | backfill_globals.ar_bdm = aral_by_size_acquire(sizeof(struct backfill_dim_work)); |
| 221 | |
| 222 | spinlock_lock(&backfill_globals.spinlock); |
| 223 | backfill_globals.running = true; |
| 224 | spinlock_unlock(&backfill_globals.spinlock); |
| 225 | |
| 226 | size_t threads = netdata_conf_cpus() / 2; |
| 227 | if(threads < 2) threads = 2; |
| 228 | if(threads > 16) threads = 16; |
| 229 | ND_THREAD *th[15]; |
| 230 | |
| 231 | for(size_t t = 0; t < threads - 1 ;t++) { |
| 232 | char tag[15]; |
| 233 | snprintfz(tag, sizeof(tag), "BACKFILL[%zu]", t + 1); |
| 234 | th[t] = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT, backfill_worker_thread, NULL); |
| 235 | } |
| 236 | |
| 237 | backfill_worker_thread((void *)0x01); |
| 238 | static_thread->enabled = NETDATA_MAIN_THREAD_EXITING; |
| 239 | |
| 240 | for(size_t t = 0; t < threads - 1 ;t++) { |
| 241 | nd_thread_signal_cancel(th[t]); |
| 242 | nd_thread_join(th[t]); |
| 243 | } |
| 244 | |
| 245 | // cleanup |
| 246 | spinlock_lock(&backfill_globals.spinlock); |
| 247 | backfill_globals.running = false; |
| 248 | Word_t idx = 0; |
| 249 | for(struct backfill_dim_work *bdm = BACKFILL_FIRST(&backfill_globals.queue, &idx); |
| 250 | bdm; |
| 251 | bdm = BACKFILL_NEXT(&backfill_globals.queue, &idx)) { |
| 252 | backfill_dim_work_free(false, bdm); |
| 253 | } |
| 254 | spinlock_unlock(&backfill_globals.spinlock); |
| 255 | |
| 256 | aral_by_size_release(backfill_globals.ar_br); |
| 257 | aral_by_size_release(backfill_globals.ar_bdm); |
| 258 | completion_destroy(&backfill_globals.completion); |
| 259 | |
| 260 | static_thread->enabled = NETDATA_MAIN_THREAD_EXITED; |
| 261 | } |