@cryptotaxi247 / netdata / commits / a2e24cd93

Rework datafiles (#20581)

* Datafiles in judyL * Use CLOSE_FILE macro for clarity * Fix compilation error

Stelios Fragkakis committed Jul 7, 2025 at 21:28 UTC a2e24cd939ff34b172b292859606382d179d0d48

6 files changed +189 -109

src/database/engine/datafile.c

+33 -55

@@ -6,7 +6,12 @@ void datafile_list_insert(struct rrdengine_instance *ctx, struct rrdengine_dataf

6 if(!having_lock)

7 uv_rwlock_wrlock(&ctx->datafiles.rwlock);

9 - DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(ctx->datafiles.first, datafile, prev, next);

9 + Pvoid_t *Pvalue = JudyLIns(&ctx->datafiles.JudyL, (Word_t ) datafile->fileno, PJE0);

10 + if(!Pvalue || Pvalue == PJERR)

11 + fatal("DBENGINE: cannot insert datafile %u of tier %d into the datafiles list",

12 + datafile->fileno, ctx->config.tier);

13 +

14 + *Pvalue = datafile;

16 if(!having_lock)

17 uv_rwlock_wrunlock(&ctx->datafiles.rwlock);

@@ -14,7 +19,7 @@ void datafile_list_insert(struct rrdengine_instance *ctx, struct rrdengine_dataf

20 void datafile_list_delete_unsafe(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile)

21 {

17 - DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(ctx->datafiles.first, datafile, prev, next);

22 + (void) JudyLDel(&ctx->datafiles.JudyL, (Word_t)datafile->fileno, PJE0);

23 }

@@ -174,19 +179,10 @@ void generate_datafilepath(struct rrdengine_datafile *datafile, char *str, size_

179 int close_data_file(struct rrdengine_datafile *datafile)

180 {

181 struct rrdengine_instance *ctx = datafile_ctx(datafile);

177 - uv_fs_t req;

182 int ret;

183 char path[RRDENG_PATH_MAX];

180 -

184 generate_datafilepath(datafile, path, sizeof(path));

182 -

183 - ret = uv_fs_close(NULL, &req, datafile->file, NULL);

184 - if (ret < 0) {

185 - netdata_log_error("DBENGINE: uv_fs_close(%s): %s", path, uv_strerror(ret));

186 - ctx_fs_error(ctx);

187 - }

188 - uv_fs_req_cleanup(&req);

189 -

185 + CLOSE_FILE(ctx, path, datafile->file, ret);

186 return ret;

187 }

188

@@ -208,34 +204,13 @@ int unlink_data_file(struct rrdengine_datafile *datafile)

204 int destroy_data_file_unsafe(struct rrdengine_datafile *datafile)

205 {

206 struct rrdengine_instance *ctx = datafile_ctx(datafile);

211 - uv_fs_t req;

207 int ret;

208 char path[RRDENG_PATH_MAX];

209

210 generate_datafilepath(datafile, path, sizeof(path));

211

217 - ret = uv_fs_ftruncate(NULL, &req, datafile->file, 0, NULL);

218 - if (ret < 0) {

219 - netdata_log_error("DBENGINE: uv_fs_ftruncate(%s): %s", path, uv_strerror(ret));

220 - ctx_fs_error(ctx);

221 - }

222 - uv_fs_req_cleanup(&req);

223 -

224 - ret = uv_fs_close(NULL, &req, datafile->file, NULL);

225 - if (ret < 0) {

226 - netdata_log_error("DBENGINE: uv_fs_close(%s): %s", path, uv_strerror(ret));

227 - ctx_fs_error(ctx);

228 - }

229 - uv_fs_req_cleanup(&req);

230 -

231 - ret = uv_fs_unlink(NULL, &req, path, NULL);

232 - if (ret < 0) {

233 - netdata_log_error("DBENGINE: uv_fs_fsunlink(%s): %s", path, uv_strerror(ret));

234 - ctx_fs_error(ctx);

235 - }

236 - uv_fs_req_cleanup(&req);

237 -

238 - __atomic_add_fetch(&ctx->stats.datafile_deletions, 1, __ATOMIC_RELAXED);

212 + CLOSE_FILE(ctx, path, datafile->file, ret);

213 + ret = unlink_data_file(datafile);

214

215 return ret;

216 }

@@ -280,7 +255,7 @@ int create_data_file(struct rrdengine_datafile *datafile)

255

256 posix_memalign_freez(superblock);

257 if (ret < 0) {

283 - destroy_data_file_unsafe(datafile);

258 + (void) destroy_data_file_unsafe(datafile);

259 ctx_io_error(ctx);

260 nd_log_limit_static_global_var(dbengine_erl, 10, 0);

261 nd_log_limit(&dbengine_erl, NDLS_DAEMON, NDLP_ERR, "DBENGINE: Failed to create datafile %s", path);

@@ -329,7 +304,6 @@ static int check_data_file_superblock(uv_file file)

304 static int load_data_file(struct rrdengine_datafile *datafile)

305 {

306 struct rrdengine_instance *ctx = datafile_ctx(datafile);

332 - uv_fs_t req;

307 uv_file file;

308 int ret, fd, error;

309 uint64_t file_size;

@@ -346,12 +320,12 @@ static int load_data_file(struct rrdengine_datafile *datafile)

320

321 ret = check_file_properties(file, &file_size, sizeof(struct rrdeng_df_sb));

322 if (ret)

349 - goto error;

323 + goto err_exit;

324 file_size = ALIGN_BYTES_CEILING(file_size);

325

326 ret = check_data_file_superblock(file);

327 if (ret)

354 - goto error;

328 + goto err_exit;

329

330 ctx_io_read_op_bytes(ctx, sizeof(struct rrdeng_df_sb));

331

@@ -362,14 +336,9 @@ static int load_data_file(struct rrdengine_datafile *datafile)

336

337 return 0;

338

365 - error:

339 +err_exit:

340 error = ret;

367 - ret = uv_fs_close(NULL, &req, file, NULL);

368 - if (ret < 0) {

369 - netdata_log_error("DBENGINE: uv_fs_close(%s): %s", path, uv_strerror(ret));

370 - ctx_fs_error(ctx);

371 - }

372 - uv_fs_req_cleanup(&req);

341 + CLOSE_FILE(ctx, path, file, ret);

342 return error;

343 }

344

@@ -593,7 +562,7 @@ int create_new_datafile_pair(struct rrdengine_instance *ctx, bool having_lock)

562 return 0;

563

564 error_after_journalfile:

596 - destroy_data_file_unsafe(datafile);

565 + (void) destroy_data_file_unsafe(datafile);

566 freez(journalfile);

567

568 error_after_datafile:

@@ -625,8 +594,14 @@ int init_data_files(struct rrdengine_instance *ctx)

594 if (ctx->loading.create_new_datafile_pair)

595 create_new_datafile_pair(ctx, false);

596

628 - while(rrdeng_ctx_tier_cap_exceeded(ctx))

629 - datafile_delete(ctx, ctx->datafiles.first, false, true, false);

597 + while(rrdeng_ctx_tier_cap_exceeded(ctx)) {

598 + Word_t Index = 0;

599 + Pvoid_t *PValue = JudyLFirst(ctx->datafiles.JudyL, &Index, PJE0);

600 + if (PValue && *PValue) {

601 + struct rrdengine_datafile *datafile = *PValue;

602 + datafile_delete(ctx, datafile, false, true, false);

603 + }

604 + }

605 }

606

607 pgc_reset_hot_max(open_cache);

@@ -654,7 +629,7 @@ void finalize_data_files(struct rrdengine_instance *ctx)

629 {

630 bool logged = false;

631

657 - if (!ctx->datafiles.first)

632 + if (!ctx->datafiles.JudyL)

633 return;

634

635 while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED)) {

@@ -665,13 +640,17 @@ void finalize_data_files(struct rrdengine_instance *ctx)

640 sleep_usec(100 * USEC_PER_MS);

641 }

642

668 - do {

669 - struct rrdengine_datafile *datafile = ctx->datafiles.first;

643 + bool first_then_next = true;

644 + Pvoid_t *PValue;

645 + Word_t Index = 0;

646 +

647 + while ((PValue = JudyLFirstThenNext(ctx->datafiles.JudyL, &Index, &first_then_next))) {

648 + struct rrdengine_datafile *datafile = *PValue;

649 struct rrdengine_journalfile *journalfile = datafile->journalfile;

650

651 logged = false;

652 size_t iterations = 10;

674 - while(!datafile_acquire_for_deletion(datafile, true) && datafile != ctx->datafiles.first->prev && --iterations > 0) {

653 + while(!datafile_acquire_for_deletion(datafile, true) && --iterations > 0) {

654 if(!logged) {

655 netdata_log_info("Waiting to acquire data file %u of tier %d to close it...", datafile->fileno, ctx->config.tier);

656 logged = true;

@@ -711,6 +690,5 @@ void finalize_data_files(struct rrdengine_instance *ctx)

690

691 freez(journalfile);

692 freez(datafile);

714 -

715 - } while(ctx->datafiles.first);

693 + }

694 }

src/database/engine/datafile.h

-2

@@ -59,8 +59,6 @@ struct rrdengine_datafile {

59 uv_rwlock_t extent_rwlock;

60 struct rrdengine_instance *ctx;

61 struct rrdengine_journalfile *journalfile;

62 - struct rrdengine_datafile *prev;

63 - struct rrdengine_datafile *next;

63 struct {

64 SPINLOCK spinlock;

src/database/engine/journalfile.c

+15 -16

@@ -296,7 +296,16 @@ void journalfile_v2_data_unmount_cleanup(time_t now_s) {

296 if(uv_rwlock_tryrdlock(&ctx->datafiles.rwlock) != 0)

297 continue;

298

299 - for (datafile = ctx->datafiles.first; datafile; datafile = datafile->next) {

299 + bool first_then_next = true;

300 + Pvoid_t *Pvalue = NULL;

301 + Word_t Index = 0;

302 +

303 + while((Pvalue = JudyLFirstThenNext(ctx->datafiles.JudyL, &Index, &first_then_next))) {

304 +

305 + datafile = *Pvalue;

306 + if (!datafile)

307 + continue;

308 +

309 struct rrdengine_journalfile *journalfile = datafile->journalfile;

310

311 if(!spinlock_trylock(&journalfile->data_spinlock))

@@ -477,14 +486,10 @@ static int close_uv_file(struct rrdengine_datafile *datafile, uv_file file)

486 int ret;

487 char path[RRDENG_PATH_MAX];

488

480 - uv_fs_t req;

481 - ret = uv_fs_close(NULL, &req, file, NULL);

482 - if (ret < 0) {

483 - journalfile_v1_generate_path(datafile, path, sizeof(path));

484 - netdata_log_error("DBENGINE: uv_fs_close(\"%s\"): %s", path, uv_strerror(ret));

485 - ctx_fs_error(datafile_ctx(datafile));

486 - }

487 - uv_fs_req_cleanup(&req);

489 + struct rrdengine_instance *ctx = datafile_ctx(datafile);

490 + journalfile_v1_generate_path(datafile, path, sizeof(path));

491 +

492 + CLOSE_FILE(ctx, path, file, ret);

493 return ret;

494 }

495

@@ -1524,7 +1529,6 @@ bool journalfile_migrate_to_v2_callback(Word_t section, unsigned datafile_fileno

1529 int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,

1530 struct rrdengine_datafile *datafile)

1531 {

1527 - uv_fs_t req;

1532 uv_file file;

1533 int ret, fd, error;

1534 uint64_t file_size, max_id;

@@ -1600,11 +1604,6 @@ int journalfile_load(struct rrdengine_instance *ctx, struct rrdengine_journalfil

1604 return 0;

1605

1606 cleanup:

1603 - ret = uv_fs_close(NULL, &req, file, NULL);

1604 - if (ret < 0) {

1605 - netdata_log_error("DBENGINE: uv_fs_close(\"%s\"): %s", path, uv_strerror(ret));

1606 - ctx_fs_error(ctx);

1607 - }

1608 - uv_fs_req_cleanup(&req);

1607 + CLOSE_FILE(ctx, path, file, ret);

1608 return error;

1609 }

src/database/engine/rrdengine.c

+118 -28

@@ -698,12 +698,89 @@ static bool datafile_is_full(struct rrdengine_instance *ctx, struct rrdengine_da

698 return ret;

699 }

700

701 +size_t datafile_count(struct rrdengine_instance *ctx, bool with_lock)

702 +{

703 + size_t count = 0;

704 +

705 + if (!with_lock)

706 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);

707 +

708 + count = JudyLCount(ctx->datafiles.JudyL, 0, -1, PJE0);

709 +

710 + if (!with_lock)

711 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

712 +

713 + return count;

714 +}

715 +

716 +struct rrdengine_datafile *

717 +get_next_datafile(struct rrdengine_datafile *this_datafile, struct rrdengine_instance *ctx, bool with_lock)

718 +{

719 + struct rrdengine_datafile *datafile = NULL;

720 +

721 + ctx = this_datafile ? this_datafile->ctx : ctx;

722 + if (!ctx)

723 + return NULL;

724 +

725 + if (!with_lock)

726 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);

727 +

728 + Word_t Index = this_datafile ? this_datafile->fileno : 0;

729 + Pvoid_t *Pvalue;

730 +

731 + Pvalue = JudyLNext(ctx->datafiles.JudyL, &Index, PJE0);

732 +

733 + if (Pvalue)

734 + datafile = *Pvalue;

735 +

736 + if (!with_lock)

737 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

738 +

739 + return datafile;

740 +}

741 +

742 +static struct rrdengine_datafile *get_ctx_datafile_first_or_last(struct rrdengine_instance *ctx, bool first, bool with_lock)

743 +{

744 + struct rrdengine_datafile *datafile = NULL;

745 +

746 + if (!with_lock)

747 + uv_rwlock_rdlock(&ctx->datafiles.rwlock);

748 +

749 + Word_t Index = 0;

750 + Pvoid_t *Pvalue;

751 +

752 + if (first)

753 + Pvalue = JudyLFirst(ctx->datafiles.JudyL, &Index, PJE0);

754 + else {

755 + Index = -1;

756 + Pvalue = JudyLLast(ctx->datafiles.JudyL, &Index, PJE0);

757 + }

758 +

759 + if (Pvalue)

760 + datafile = *Pvalue;

761 +

762 + if (!with_lock)

763 + uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

764 +

765 + return datafile;

766 +}

767 +

768 +struct rrdengine_datafile *get_first_ctx_datafile(struct rrdengine_instance *ctx, bool with_lock) {

769 + return get_ctx_datafile_first_or_last(ctx, true, with_lock);

770 +}

771 +

772 +struct rrdengine_datafile *get_last_ctx_datafile(struct rrdengine_instance *ctx, bool with_lock) {

773 + return get_ctx_datafile_first_or_last(ctx, false, with_lock);

774 +}

775 +

776 +

777 static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_instance *ctx) {

778 struct rrdengine_datafile *datafile;

779

780 // get the latest datafile

781 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

706 - datafile = ctx->datafiles.first->prev;

782 +

783 + datafile = get_last_ctx_datafile(ctx, true);

784 // become a writer on this datafile, to prevent it from vanishing

785 spinlock_lock(&datafile->writers.spinlock);

786 datafile->writers.running++;

@@ -719,9 +796,7 @@ static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_

796 netdata_mutex_lock(&mutex);

797

798 // take the latest datafile again - without this, multiple threads may create multiple files

722 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

723 - datafile = ctx->datafiles.first->prev;

724 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

799 + datafile = get_last_ctx_datafile(ctx, false);

800

801 if(datafile_is_full(ctx, datafile) && create_new_datafile_pair(ctx, true) == 0)

802 __atomic_store_n(&ctx->atomic.needs_indexing, true, __ATOMIC_RELAXED);

@@ -730,7 +805,7 @@ static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_

805

806 // get the new latest datafile again, like above

807 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

733 - datafile = ctx->datafiles.first->prev;

808 + datafile = get_last_ctx_datafile(ctx, true);

809 // become a writer on this datafile, to prevent it from vanishing

810 spinlock_lock(&datafile->writers.spinlock);

811 datafile->writers.running++;

@@ -958,10 +1033,10 @@ struct rrdengine_datafile *datafile_release_and_acquire_next_for_retention(struc

1033

1034 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1035

961 - struct rrdengine_datafile *next_datafile = datafile->next;

1036 + struct rrdengine_datafile *next_datafile = get_next_datafile(datafile, NULL, true);

1037

1038 while(next_datafile && !datafile_acquire(next_datafile, DATAFILE_ACQUIRE_RETENTION))

964 - next_datafile = next_datafile->next;

1039 + next_datafile = get_next_datafile(next_datafile, NULL, true);

1040

1041 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1042

@@ -981,7 +1056,8 @@ static time_t find_uuid_first_time(

1056 // acquire the datafile to work with it

1057 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1058 while(datafile && !datafile_acquire(datafile, DATAFILE_ACQUIRE_RETENTION))

984 - datafile = datafile->next;

1059 + datafile = get_next_datafile(datafile, NULL, true);

1060 +

1061 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1062

1063 if (unlikely(!datafile))

@@ -1313,7 +1389,7 @@ void datafile_delete(

1389 DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION

1390 "' to be available for deletion, "

1391 "it is in use currently by %u users.",

1316 - ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno, datafile->users.lockers);

1392 + ctx->config.dbfiles_path, datafile->tier, datafile->fileno, datafile->users.lockers);

1393

1394 __atomic_add_fetch(&rrdeng_cache_efficiency_stats.datafile_deletion_spin, 1, __ATOMIC_RELAXED);

1395 sleep_usec(1 * USEC_PER_SEC);

@@ -1323,10 +1399,10 @@ void datafile_delete(

1399 netdata_log_info("DBENGINE: acquired data file \"%s/"

1400 DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION

1401 "\" for deletion.",

1326 - ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);

1402 + ctx->config.dbfiles_path, datafile->tier, datafile->fileno);

1403

1404 if (update_retention)

1329 - update_metrics_first_time_s(ctx, datafile, datafile->next, worker);

1405 + update_metrics_first_time_s(ctx, datafile, get_next_datafile(datafile, NULL, true), worker);

1406

1407 // if (!ctx_is_available_for_queries(ctx)) {

1408 // // agent is shutting down, we cannot continue

@@ -1339,7 +1415,7 @@ void datafile_delete(

1415 netdata_log_info("DBENGINE: deleting data file \"%s/"

1416 DATAFILE_PREFIX RRDENG_FILE_NUMBER_PRINT_TMPL DATAFILE_EXTENSION

1417 "\".",

1342 - ctx->config.dbfiles_path, ctx->datafiles.first->tier, ctx->datafiles.first->fileno);

1418 + ctx->config.dbfiles_path, datafile->tier, datafile->fileno);

1419

1420 if(worker)

1421 worker_is_busy(UV_EVENT_DBENGINE_DATAFILE_DELETE);

@@ -1391,7 +1467,9 @@ void datafile_delete(

1467 }

1468

1469 static void *database_rotate_tp_worker(struct rrdengine_instance *ctx __maybe_unused, void *data __maybe_unused, struct completion *completion __maybe_unused, uv_work_t *uv_work_req __maybe_unused) {

1394 - datafile_delete(ctx, ctx->datafiles.first, ctx_is_available_for_queries(ctx), true, true);

1470 +

1471 + struct rrdengine_datafile *datafile = get_first_ctx_datafile(ctx, false);

1472 + datafile_delete(ctx, datafile, ctx_is_available_for_queries(ctx), true, true);

1473

1474 rrdcontext_db_rotation();

1475

@@ -1487,9 +1565,11 @@ static void *populate_mrg_tp_worker(

1565 int rc;

1566

1567 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1568 +

1569 size_t total_datafiles = 0;

1570 size_t populated_datafiles = 0;

1492 - for (struct rrdengine_datafile *df = ctx->datafiles.first; df; df = df->next) {

1571 + struct rrdengine_datafile *df = NULL;

1572 + while ((df = get_next_datafile(df, ctx, true))) {

1573 total_datafiles++;

1574 if (df->populate_mrg.populated)

1575 populated_datafiles++;

@@ -1507,16 +1587,21 @@ static void *populate_mrg_tp_worker(

1587

1588 // find a datafile to work on

1589 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1510 - for(datafile = ctx->datafiles.first; datafile; datafile = datafile->next) {

1511 - if(!spinlock_trylock(&datafile->populate_mrg.spinlock))

1590 + bool first_then_next = true;

1591 + Pvoid_t *Pvalue = NULL;

1592 + Word_t Index = 0;

1593 + while((Pvalue = JudyLFirstThenNext(ctx->datafiles.JudyL, &Index, &first_then_next))) {

1594 + datafile = *Pvalue;

1595 + if(!spinlock_trylock(&datafile->populate_mrg.spinlock)) {

1596 + datafile = NULL;

1597 continue;

1598 + }

1599

1600 if(datafile->populate_mrg.populated) {

1601 spinlock_unlock(&datafile->populate_mrg.spinlock);

1602 + datafile = NULL;

1603 continue;

1604 }

1518 -

1519 - // we have the spinlock and it is not populated

1605 break;

1606 }

1607 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

@@ -1704,7 +1789,7 @@ time_t get_datafile_end_time(struct rrdengine_instance *ctx)

1789 time_t last_time_s = 0;

1790

1791 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1707 - struct rrdengine_datafile *datafile = ctx->datafiles.first;

1792 + struct rrdengine_datafile *datafile = get_last_ctx_datafile(ctx, true);

1793

1794 if (datafile) {

1795 last_time_s = datafile->journalfile->v2.last_time_s;

@@ -1770,15 +1855,15 @@ static struct rrdengine_datafile *release_and_aquire_next_datafile_for_indexing(

1855

1856 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1857 if (release_datafile) {

1773 - datafile = release_datafile->next;

1858 + datafile = get_next_datafile(release_datafile, NULL, true);

1859 datafile_release(release_datafile, DATAFILE_ACQUIRE_INDEXING);

1860 }

1861 else

1777 - datafile = ctx->datafiles.first;

1862 + datafile = get_first_ctx_datafile(ctx, true);

1863

1864 while (datafile && datafile->fileno != ctx_last_fileno_get(ctx) && datafile->fileno != ctx_last_flush_fileno_get(ctx)) {

1865 if(journalfile_v2_data_available(datafile->journalfile)) {

1781 - datafile = datafile->next;

1866 + datafile = get_next_datafile(datafile, NULL, true);

1867 continue;

1868 }

1869

@@ -1795,7 +1880,7 @@ static struct rrdengine_datafile *release_and_aquire_next_datafile_for_indexing(

1880 return datafile;

1881 }

1882 nd_log_daemon(NDLP_INFO, "DBENGINE: Datafile %u CANNOT be locked for indexing after retries; skipping", datafile->fileno);

1798 - datafile = datafile->next;

1883 + datafile = get_next_datafile(datafile, NULL, true);

1884 }

1885 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1886 return NULL;

@@ -1942,8 +2027,11 @@ uint64_t rrdeng_get_used_disk_space(struct rrdengine_instance *ctx, bool having_

2027 if (!having_lock)

2028 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

2029

1945 - if (ctx->datafiles.first && ctx->datafiles.first->prev)

1946 - active_space = ctx->datafiles.first->prev->pos;

2030 + struct rrdengine_datafile *first_datafile = get_first_ctx_datafile(ctx, true);

2031 + struct rrdengine_datafile *last_datafile = get_last_ctx_datafile(ctx, true);

2032 +

2033 + if (first_datafile && last_datafile)

2034 + active_space = last_datafile->pos;

2035

2036 if (!having_lock)

2037 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

@@ -1978,7 +2066,9 @@ static time_t get_tier_retention(struct rrdengine_instance *ctx)

2066 bool rrdeng_ctx_tier_cap_exceeded(struct rrdengine_instance *ctx)

2067 {

2068 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1981 - if (!ctx->datafiles.first || !ctx->datafiles.first->next) {

2069 + struct rrdengine_datafile *first_datafile = get_first_ctx_datafile(ctx, true);

2070 +

2071 + if (!first_datafile || datafile_count(ctx, true) < 2) {

2072 uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

2073 return false;

2074 }

@@ -2337,8 +2427,8 @@ void *dbengine_event_loop(void* arg) {

2427 case RRDENG_OPCODE_DATABASE_ROTATE: {

2428 struct rrdengine_instance *ctx = cmd.ctx;

2429 ctx->datafiles.pending_rotate = false;

2340 - if (NOT_DELETING_FILES(ctx) && ctx->datafiles.first->next != NULL &&

2341 - ctx->datafiles.first->next->next != NULL && rrdeng_ctx_tier_cap_exceeded(ctx)) {

2430 + if (NOT_DELETING_FILES(ctx) && datafile_count(ctx, false) > 2 &&

2431 + rrdeng_ctx_tier_cap_exceeded(ctx)) {

2432 __atomic_store_n(&ctx->atomic.now_deleting_files, true, __ATOMIC_RELAXED);

2433 work_dispatch(ctx, NULL, NULL, opcode, database_rotate_tp_worker, after_database_rotate);

2434 }

src/database/engine/rrdengine.h

+19 -2

@@ -35,6 +35,17 @@ extern unsigned rrdeng_pages_per_extent;

35 uv_fs_req_cleanup(&(_req)); \

36 } while (0)

38 +#define CLOSE_FILE(ctx, path, file, ret_var) \

39 + do { \

40 + uv_fs_t _req; \

41 + (ret_var) = uv_fs_close(NULL, &(_req), (file), NULL); \

42 + if ((ret_var) < 0) { \

43 + netdata_log_error("DBENGINE: uv_fs_close(\"%s\"): %s", (path), uv_strerror(ret_var)); \

44 + ctx_fs_error(ctx); \

45 + } \

46 + uv_fs_req_cleanup(&(_req)); \

47 + } while (0)

48 +

49 /* Forward declarations */

50 struct rrdengine_instance;

51 struct rrdeng_cmd;

@@ -386,11 +397,11 @@ struct rrdengine_instance {

397 TIER_CONFIG_PROTOTYPE config;

398

399 struct {

389 - uv_rwlock_t rwlock; // the linked list of datafiles is protected by this lock

400 + uv_rwlock_t rwlock; // the JudyL of datafiles is protected by this lock

401 bool disk_time; // true: delete for disk quota, false: delete for retention

402 bool pending_rotate; // Change from event loop

403 bool pending_index; // Change from event loop

393 - struct rrdengine_datafile *first; // oldest - the newest with ->first->prev

404 + Pvoid_t JudyL; // the datafiles, indexed by fileno

405 } datafiles;

406

407 struct {

@@ -570,5 +581,11 @@ uint64_t rrdeng_get_used_disk_space(struct rrdengine_instance *ctx, bool having_

581 void rrdeng_calculate_tier_disk_space_percentage(void);

582 uint64_t rrdeng_get_directory_free_bytes_space(struct rrdengine_instance *ctx);

583 void dbengine_shutdown();

584 +size_t datafile_count(struct rrdengine_instance *ctx, bool with_lock);

585 +struct rrdengine_datafile *get_first_ctx_datafile(struct rrdengine_instance *ctx, bool with_lock);

586 +struct rrdengine_datafile *get_last_ctx_datafile(struct rrdengine_instance *ctx, bool with_lock);

587 +struct rrdengine_datafile *

588 +get_next_datafile(struct rrdengine_datafile *this_datafile, struct rrdengine_instance *ctx, bool with_lock);

589 +

590

591 #endif /* NETDATA_RRDENGINE_H */

src/database/engine/rrdengineapi.c

+4 -6

@@ -1108,11 +1108,7 @@ void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long

1108

1109 static void rrdeng_populate_mrg(struct rrdengine_instance *ctx)

1110 {

1111 - uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1112 - size_t datafiles = 0;

1113 - for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next)

1114 - datafiles++;

1115 - uv_rwlock_rdunlock(&ctx->datafiles.rwlock);

1111 + size_t datafiles = datafile_count(ctx, false);

1112

1113 ssize_t cpus = (ssize_t)netdata_conf_cpus();

1114 if(cpus < 1)

@@ -1353,7 +1349,9 @@ RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) {

1349 RRDENG_SIZE_STATS stats = { 0 };

1350

1351 uv_rwlock_rdlock(&ctx->datafiles.rwlock);

1356 - for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next) {

1352 + struct rrdengine_datafile *df = NULL;

1353 +

1354 + while ((df = get_next_datafile(df, ctx, true))) {

1355 stats.datafiles++;

1356 populate_v2_statistics(df, &stats);

1357 }