@cryptotaxi247 / netdata-1 / commits / 29271f575

Use mmap to read an extent from a datafile (#13834)

* Use MMAP to read an extent from the datafile. Fallback to uv_fs_read if that fails * Schedule the fallback using uv_fs_read in the MMAP callback * Check the return value of mmap correctly * Add unlikely

Stelios Fragkakis committed Oct 24, 2022 at 10:20 UTC 29271f575a65ed5cd13265413bff3b32398bd538

2 files changed +70 -24

database/engine/rrdengine.c

+66 -24

@@ -272,11 +272,9 @@ static void fill_page_with_nulls(void *page, uint32_t page_length, uint8_t type)

272 }

273 }

274

275 -void read_extent_cb(uv_fs_t* req)

275 +static void do_extent_processing (struct rrdengine_worker_config *wc, struct extent_io_descriptor *xt_io_descr, bool read_failed)

276 {

277 - struct rrdengine_worker_config* wc = req->loop->data;

277 struct rrdengine_instance *ctx = wc->ctx;

279 - struct extent_io_descriptor *xt_io_descr;

278 struct rrdeng_page_descr *descr;

279 struct page_cache_descr *pg_cache_descr;

280 int ret;

@@ -289,21 +287,20 @@ void read_extent_cb(uv_fs_t* req)

287 struct rrdeng_df_extent_trailer *trailer;

288 uLong crc;

289

292 - xt_io_descr = req->data;

290 header = xt_io_descr->buf;

291 payload_length = header->payload_length;

292 count = header->number_of_pages;

293 payload_offset = sizeof(*header) + sizeof(header->descr[0]) * count;

294 trailer = xt_io_descr->buf + xt_io_descr->bytes - sizeof(*trailer);

295

299 - if (req->result < 0) {

296 + if (unlikely(read_failed)) {

297 struct rrdengine_datafile *datafile = xt_io_descr->descr_array[0]->extent->datafile;

298

299 ++ctx->stats.io_errors;

300 rrd_stat_atomic_add(&global_io_errors, 1);

301 have_read_error = 1;

305 - error("%s: uv_fs_read - %s - extent at offset %"PRIu64"(%u) in datafile %u-%u.", __func__,

306 - uv_strerror((int)req->result), xt_io_descr->pos, xt_io_descr->bytes, datafile->tier, datafile->fileno);

302 + error("%s: uv_fs_read - extent at offset %"PRIu64"(%u) in datafile %u-%u.", __func__, xt_io_descr->pos,

303 + xt_io_descr->bytes, datafile->tier, datafile->fileno);

304 goto after_crc_check;

305 }

306 crc = crc32(0L, Z_NULL, 0);

@@ -421,11 +418,67 @@ after_crc_check:

418 }

419 if (xt_io_descr->completion)

420 completion_mark_complete(xt_io_descr->completion);

421 +}

422 +

423 +static void read_extent_cb(uv_fs_t *req)

424 +{

425 + struct rrdengine_worker_config *wc = req->loop->data;

426 + struct extent_io_descriptor *xt_io_descr;

427 +

428 + xt_io_descr = req->data;

429 + do_extent_processing(wc, xt_io_descr, req->result < 0);

430 uv_fs_req_cleanup(req);

431 posix_memfree(xt_io_descr->buf);

432 freez(xt_io_descr);

433 }

434

435 +static void read_mmap_extent_cb(uv_work_t *req, int status __maybe_unused)

436 +{

437 + struct rrdengine_worker_config *wc = req->loop->data;

438 + struct rrdengine_instance *ctx = wc->ctx;

439 + struct extent_io_descriptor *xt_io_descr;

440 + xt_io_descr = req->data;

441 +

442 + if (likely(xt_io_descr->map_base)) {

443 + do_extent_processing(wc, xt_io_descr, false);

444 + munmap(xt_io_descr->map_base, xt_io_descr->map_length);

445 + freez(xt_io_descr);

446 + return;

447 + }

448 +

449 + // MMAP failed, so do uv_fs_read

450 + int ret = posix_memalign((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(xt_io_descr->bytes));

451 + if (unlikely(ret)) {

452 + fatal("posix_memalign:%s", strerror(ret));

453 + }

454 + unsigned real_io_size = ALIGN_BYTES_CEILING( xt_io_descr->bytes);

455 + xt_io_descr->iov = uv_buf_init((void *)xt_io_descr->buf, real_io_size);

456 + xt_io_descr->req.data = xt_io_descr;

457 + ret = uv_fs_read(req->loop, &xt_io_descr->req, xt_io_descr->file, &xt_io_descr->iov, 1, (unsigned) xt_io_descr->pos, read_extent_cb);

458 + fatal_assert(-1 != ret);

459 + ctx->stats.io_read_bytes += real_io_size;

460 + ctx->stats.io_read_extent_bytes += real_io_size;

461 +}

462 +

463 +static void do_mmap_read_extent(uv_work_t *req)

464 +{

465 + struct extent_io_descriptor *xt_io_descr = (struct extent_io_descriptor * )req->data;

466 + struct rrdengine_worker_config *wc = req->loop->data;

467 + struct rrdengine_instance *ctx = wc->ctx;

468 +

469 + off_t map_start = ALIGN_BYTES_FLOOR(xt_io_descr->pos);

470 + size_t length = ALIGN_BYTES_CEILING(xt_io_descr->pos + xt_io_descr->bytes) - map_start;

471 + unsigned real_io_size = xt_io_descr->bytes;

472 +

473 + void *data = mmap(NULL, length, PROT_READ, MAP_SHARED, xt_io_descr->file, map_start);

474 + if (likely(data != MAP_FAILED)) {

475 + xt_io_descr->map_base = data;

476 + xt_io_descr->map_length = length;

477 + xt_io_descr->buf = data + (xt_io_descr->pos - map_start);

478 + ctx->stats.io_read_bytes += real_io_size;

479 + ctx->stats.io_read_extent_bytes += real_io_size;

480 + }

481 +}

482

483 static void do_read_extent(struct rrdengine_worker_config* wc,

484 struct rrdeng_page_descr **descr,

@@ -435,8 +488,7 @@ static void do_read_extent(struct rrdengine_worker_config* wc,

488 struct rrdengine_instance *ctx = wc->ctx;

489 struct page_cache_descr *pg_cache_descr;

490 int ret;

438 - unsigned i, size_bytes, pos, real_io_size;

439 -// uint32_t payload_length;

491 + unsigned i, size_bytes, pos;

492 struct extent_io_descriptor *xt_io_descr;

493 struct rrdengine_datafile *datafile;

494 struct extent_info *extent = descr[0]->extent;

@@ -452,18 +504,17 @@ static void do_read_extent(struct rrdengine_worker_config* wc,

504 rrdeng_page_descr_mutex_lock(ctx, descr[i]);

505 pg_cache_descr = descr[i]->pg_cache_descr;

506 pg_cache_descr->flags |= RRD_PAGE_READ_PENDING;

455 -// payload_length = descr[i]->page_length;

507 rrdeng_page_descr_mutex_unlock(ctx, descr[i]);

457 -

508 xt_io_descr->descr_array[i] = descr[i];

509 }

510 xt_io_descr->descr_count = count;

511 + xt_io_descr->file = datafile->file;

512 xt_io_descr->bytes = size_bytes;

513 xt_io_descr->pos = pos;

463 - xt_io_descr->req.data = xt_io_descr;

514 + xt_io_descr->req_worker.data = xt_io_descr;

515 xt_io_descr->completion = NULL;

465 - /* xt_io_descr->descr_commit_idx_array[0] */

516 xt_io_descr->release_descr = release_descr;

517 + xt_io_descr->buf = NULL;

518

519 xt_is_cached = !lookup_in_xt_cache(wc, extent, &xt_idx);

520 if (xt_is_cached) {

@@ -483,19 +534,10 @@ static void do_read_extent(struct rrdengine_worker_config* wc,

534 }

535 }

536

486 - ret = posix_memalign((void *)&xt_io_descr->buf, RRDFILE_ALIGNMENT, ALIGN_BYTES_CEILING(size_bytes));

487 - if (unlikely(ret)) {

488 - fatal("posix_memalign:%s", strerror(ret));

489 - /* freez(xt_io_descr);

490 - return;*/

491 - }

492 - real_io_size = ALIGN_BYTES_CEILING(size_bytes);

493 - xt_io_descr->iov = uv_buf_init((void *)xt_io_descr->buf, real_io_size);

494 - ret = uv_fs_read(wc->loop, &xt_io_descr->req, datafile->file, &xt_io_descr->iov, 1, pos, read_extent_cb);

537 + ret = uv_queue_work(wc->loop, &xt_io_descr->req_worker, do_mmap_read_extent, read_mmap_extent_cb);

538 fatal_assert(-1 != ret);

496 - ctx->stats.io_read_bytes += real_io_size;

539 +

540 ++ctx->stats.io_read_requests;

498 - ctx->stats.io_read_extent_bytes += real_io_size;

541 ++ctx->stats.io_read_extents;

542 ctx->stats.pg_cache_backfills += count;

543 }

database/engine/rrdengine.h

@@ -105,8 +105,12 @@ struct rrdeng_cmdqueue {

105

106 struct extent_io_descriptor {

107 uv_fs_t req;

108 + uv_work_t req_worker;

109 uv_buf_t iov;

110 + uv_file file;

111 void *buf;

112 + void *map_base;

113 + size_t map_length;

114 uint64_t pos;

115 unsigned bytes;

116 struct completion *completion;