Cleanup code that writes extents to the database (#19596)
* Use local request and buffer * Cleanup structures/fields not needed (no callbacks to pass data) * Cleanup request before retrying
Stelios Fragkakis committed
Feb 10, 2025 at 16:43 UTC
87185ae42a6fcbca04fdcf95ff84ed795be1ea5d
3 files changed
+33
-49
src/database/engine/journalfile.c
+14
-17
@@ -4,51 +4,48 @@
4
/* Careful to always call this before creating a new journal file */
5
void journalfile_v1_extent_write(struct rrdengine_instance *ctx, struct rrdengine_datafile *datafile, WAL *wal)
6
{
7
- struct generic_io_descriptor *io_descr;
7
+ uv_fs_t request;
8
struct rrdengine_journalfile *journalfile = datafile->journalfile;
9
+ uv_buf_t iov;
10
10
- io_descr = &wal->io_descr;
11
- io_descr->ctx = ctx;
11
if (wal->size < wal->buf_size) {
12
/* simulate an empty transaction to skip the rest of the block */
13
*(uint8_t *) (wal->buf + wal->size) = STORE_PADDING;
14
}
16
- io_descr->buf = wal->buf;
17
- io_descr->bytes = wal->buf_size;
15
16
+ uint64_t journalfile_position;
17
spinlock_lock(&journalfile->unsafe.spinlock);
20
- io_descr->pos = journalfile->unsafe.pos;
18
+ journalfile_position = journalfile->unsafe.pos;
19
journalfile->unsafe.pos += wal->buf_size;
20
spinlock_unlock(&journalfile->unsafe.spinlock);
21
24
- io_descr->req.data = wal;
25
- io_descr->data = journalfile;
26
-
27
- io_descr->iov = uv_buf_init((void *)io_descr->buf, wal->buf_size);
22
+ iov = uv_buf_init(wal->buf, wal->buf_size);
23
24
int retries = 10;
25
int ret = -1;
26
while (ret == -1 && --retries) {
32
- ret = uv_fs_write(NULL, &io_descr->req, journalfile->file, &io_descr->iov, 1, (int64_t)io_descr->pos, NULL);
33
- if (ret == -1)
27
+ ret = uv_fs_write(NULL, &request, journalfile->file, &iov, 1, (int64_t)journalfile_position, NULL);
28
+ if (ret == -1) {
29
sleep_usec(300 * USEC_PER_MS);
30
+ uv_fs_req_cleanup(&request);
31
+ }
32
}
33
37
- bool jf_write_error = (ret == -1 || io_descr->req.result < 0);
34
+ bool jf_write_error = (ret == -1 || request.result < 0);
35
36
if (unlikely(jf_write_error)) {
37
ctx_io_error(ctx);
38
if (ret == -1)
39
netdata_log_error(
43
- "DBENGINE: %s: uv_fs_write: failed to store metadata in journalfile %u, offset %ld",
40
+ "DBENGINE: %s: uv_fs_write: failed to store metadata in journalfile %u, offset %"PRIu64,
41
__func__,
42
datafile->fileno,
46
- (int64_t)io_descr->pos);
43
+ journalfile_position);
44
else
48
- netdata_log_error("DBENGINE: %s: uv_fs_write: %s", __func__, uv_strerror((int)io_descr->req.result));
45
+ netdata_log_error("DBENGINE: %s: uv_fs_write: %s", __func__, uv_strerror((int)request.result));
46
}
47
51
- uv_fs_req_cleanup(&io_descr->req);
48
+ uv_fs_req_cleanup(&request);
49
ctx_current_disk_space_increase(ctx, wal->buf_size);
50
ctx_io_write_op_bytes(ctx, wal->buf_size);
51
src/database/engine/rrdengine.c
+16
-15
@@ -620,10 +620,8 @@ static void journalfile_extent_build(struct rrdengine_instance *ctx, struct exte
620
crc32set(jf_trailer->checksum, crc);
621
}
622
623
-static void extent_flushed_to_open_tp_worker(
624
- struct rrdengine_instance *ctx,
625
- struct extent_io_descriptor *xt_io_descr,
626
- bool have_error)
623
+static void
624
+extent_flush_to_open(struct rrdengine_instance *ctx, struct extent_io_descriptor *xt_io_descr, bool have_error)
625
{
626
worker_is_busy(UV_EVENT_DBENGINE_FLUSHED_TO_OPEN);
627
@@ -730,7 +728,8 @@ static struct rrdengine_datafile *get_datafile_to_write_extent(struct rrdengine_
728
/*
729
* Take a page list in a judy array and write them
730
*/
733
-static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_instance *ctx, struct page_descr_with_data *base)
731
+static struct extent_io_descriptor *
732
+datafile_extent_build(struct rrdengine_instance *ctx, struct page_descr_with_data *base, uv_buf_t *iov)
733
{
734
int ret;
735
unsigned i, count, size_bytes, pos, real_io_size;
@@ -760,7 +759,6 @@ static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_insta
759
}
760
761
xt_io_descr = extent_io_descriptor_get();
763
- xt_io_descr->ctx = ctx;
762
payload_offset = sizeof(*header) + count * sizeof(header->descr[0]);
763
max_compressed_size = dbengine_max_compressed_size(uncompressed_payload_length, compression_algorithm);
764
size_bytes = payload_offset + MAX(uncompressed_payload_length, max_compressed_size) + sizeof(*trailer);
@@ -846,14 +844,13 @@ static struct extent_io_descriptor *datafile_extent_build(struct rrdengine_insta
844
spinlock_unlock(&datafile->writers.spinlock);
845
846
xt_io_descr->bytes = size_bytes;
849
- xt_io_descr->uv_fs_request.data = xt_io_descr;
847
848
trailer = xt_io_descr->buf + size_bytes - sizeof(*trailer);
849
crc = crc32(0L, Z_NULL, 0);
850
crc = crc32(crc, xt_io_descr->buf, size_bytes - sizeof(*trailer));
851
crc32set(trailer->checksum, crc);
852
856
- xt_io_descr->iov = uv_buf_init((void *)xt_io_descr->buf, real_io_size);
853
+ *iov = uv_buf_init((void *)xt_io_descr->buf, real_io_size);
854
journalfile_extent_build(ctx, xt_io_descr);
855
856
ctx_last_flush_fileno_set(ctx, datafile->fileno);
@@ -875,23 +872,27 @@ static void *extent_write_tp_worker(
872
uv_work_t *req __maybe_unused)
873
{
874
worker_is_busy(UV_EVENT_DBENGINE_EXTENT_WRITE);
875
+ uv_buf_t iov;
876
struct page_descr_with_data *base = data;
879
- struct extent_io_descriptor *xt_io_descr = datafile_extent_build(ctx, base);
877
+ struct extent_io_descriptor *xt_io_descr = datafile_extent_build(ctx, base, &iov);
878
879
if (!xt_io_descr)
880
goto done;
881
882
struct rrdengine_datafile *datafile = xt_io_descr->datafile;
883
+ uv_fs_t request;
884
885
int retries = 10;
886
int ret = -1;
887
while (ret == -1 && --retries) {
889
- ret = uv_fs_write(NULL, &xt_io_descr->uv_fs_request, datafile->file, &xt_io_descr->iov, 1, (int64_t)xt_io_descr->pos, NULL);
890
- if (ret == -1)
888
+ ret = uv_fs_write(NULL, &request, datafile->file, &iov, 1, (int64_t)xt_io_descr->pos, NULL);
889
+ if (ret == -1) {
890
sleep_usec(300 * USEC_PER_MS);
891
+ uv_fs_req_cleanup(&request);
892
+ }
893
}
894
894
- bool df_write_error = (ret == -1 || xt_io_descr->uv_fs_request.result < 0);
895
+ bool df_write_error = (ret == -1 || request.result < 0);
896
897
if (unlikely(df_write_error)) {
898
ctx_io_error(ctx);
@@ -903,9 +904,9 @@ static void *extent_write_tp_worker(
904
(int64_t)xt_io_descr->pos);
905
else
906
netdata_log_error(
906
- "DBENGINE: %s: uv_fs_write: %s", __func__, uv_strerror((int)xt_io_descr->uv_fs_request.result));
907
+ "DBENGINE: %s: uv_fs_write: %s", __func__, uv_strerror((int)request.result));
908
}
908
- uv_fs_req_cleanup(&xt_io_descr->uv_fs_request);
909
+ uv_fs_req_cleanup(&request);
910
911
if (likely(!df_write_error)) {
912
journalfile_v1_extent_write(ctx, datafile, xt_io_descr->wal);
@@ -916,7 +917,7 @@ static void *extent_write_tp_worker(
917
datafile->writers.flushed_to_open_running++;
918
spinlock_unlock(&datafile->writers.spinlock);
919
919
- extent_flushed_to_open_tp_worker(ctx, xt_io_descr, df_write_error);
920
+ extent_flush_to_open(ctx, xt_io_descr, df_write_error);
921
922
if(ctx_is_available_for_queries(ctx))
923
rrdeng_enq_cmd(ctx, RRDENG_OPCODE_DATABASE_ROTATE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL);
src/database/engine/rrdengine.h
+3
-17
@@ -283,27 +283,14 @@ struct extent_io_data {
283
284
struct extent_io_descriptor {
285
struct rrdengine_instance *ctx;
286
- uv_fs_t uv_fs_request;
287
- uv_buf_t iov;
288
- uv_file file;
286
void *buf;
290
- struct wal *wal;
287
uint64_t pos;
292
- unsigned bytes;
288
unsigned descr_count;
289
+ unsigned bytes;
290
+ struct wal *wal;
291
+ uv_file file;
292
struct page_descr_with_data *descr_array[MAX_PAGES_PER_EXTENT];
293
struct rrdengine_datafile *datafile;
296
- struct extent_io_descriptor *next; /* multiple requests to be served by the same cached extent */
297
-};
298
-
299
-struct generic_io_descriptor {
300
- struct rrdengine_instance *ctx;
301
- uv_fs_t req;
302
- uv_buf_t iov;
303
- void *buf;
304
- void *data;
305
- uint64_t pos;
306
- unsigned bytes;
294
};
295
296
typedef struct wal {
@@ -311,7 +298,6 @@ typedef struct wal {
298
void *buf;
299
size_t size;
300
size_t buf_size;
314
- struct generic_io_descriptor io_descr;
301
302
struct {
303
struct wal *prev;