thebestvibedev/tromcho.net
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fsync data on trim_lsn, not when writing compacted data

Browse Source
This commit is contained in:
Vitaliy Filippov
2025-11-23 19:08:23 +03:00
parent 908ce74500
commit 2afb16cd19
1 changed files with 30 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/blockstore/blockstore_flush.cpp
+30 -23
View File
@@ -161,18 +161,18 @@ bool journal_flusher_co::loop()
else if (wait_state == 18) goto resume_18; else if (wait_state == 18) goto resume_18;
else if (wait_state == 19) goto resume_19; else if (wait_state == 19) goto resume_19;
else if (wait_state == 20) goto resume_20; else if (wait_state == 20) goto resume_20;
else if (wait_state == 21) goto resume_21;
resume_0: resume_0:
res = bs->heap->get_next_compact(cur_oid); res = bs->heap->get_next_compact(cur_oid);
if (res == ENOENT) if (res == ENOENT)
{ {
if (co_id == 0 && flusher->force_start > 0) if (co_id == 0 && flusher->force_start > 0)
{ {
resume_16:
resume_17:
resume_18: resume_18:
resume_19: resume_19:
resume_20: resume_20:
resume_21: if (!trim_lsn(16))
if (!trim_lsn(18))
return false; return false;
} }
cur_oid = {}; cur_oid = {};
@@ -209,7 +209,7 @@ resume_1:
assert(!end_wr->next() && end_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE)); assert(!end_wr->next() && end_wr->flags == (BS_HEAP_BIG_WRITE|BS_HEAP_STABLE));
clean_loc = end_wr->location; clean_loc = end_wr->location;
#ifdef BLOCKSTORE_DEBUG #ifdef BLOCKSTORE_DEBUG
printf("Flushing %jx:%jx v%ju .. v%ju\n", cur_oid.inode, cur_oid.stripe, end_wr->version, begin_wr->version); printf("Compacting %jx:%jx l%ju .. l%ju (last l%ju)\n", cur_oid.inode, cur_oid.stripe, end_wr->lsn, begin_wr->lsn, compact_lsn);
#endif #endif
flusher->active_flushers++; flusher->active_flushers++;
// Scan versions to flush // Scan versions to flush
@@ -283,12 +283,6 @@ resume_10:
wait_state = 10; wait_state = 10;
return false; return false;
} }
// Sync data before modifying metadata
resume_11:
resume_12:
resume_13:
if (copy_count && !fsync_batch(false, 11))
return false;
// Lock is only needed to prevent freeing the big_write because we overwrite it... // Lock is only needed to prevent freeing the big_write because we overwrite it...
bs->heap->unlock_entry(cur_oid, copy_id); bs->heap->unlock_entry(cur_oid, copy_id);
// Mark the object compacted, but don't free and remove small_writes // Mark the object compacted, but don't free and remove small_writes
@@ -305,7 +299,7 @@ resume_13:
// Done, free all buffers // Done, free all buffers
free_buffers(); free_buffers();
#ifdef BLOCKSTORE_DEBUG #ifdef BLOCKSTORE_DEBUG
printf("Compacted %jx:%jx v%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, cur_version, copy_count); printf("Compacted %jx:%jx l%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, compact_lsn, copy_count);
#endif #endif
flusher->compact_counter++; flusher->compact_counter++;
flusher->active_flushers--; flusher->active_flushers--;
@@ -313,11 +307,12 @@ resume_13:
if (co_id == 0 && !((++flusher->advance_lsn_counter) % bs->journal_trim_interval)) if (co_id == 0 && !((++flusher->advance_lsn_counter) % bs->journal_trim_interval))
{ {
flusher->advance_lsn_counter = 0; flusher->advance_lsn_counter = 0;
resume_11:
resume_12:
resume_13:
resume_14: resume_14:
resume_15: resume_15:
resume_16: if (!trim_lsn(11))
resume_17:
if (!trim_lsn(14))
return false; return false;
} }
release_oid: release_oid:
@@ -687,12 +682,14 @@ bool journal_flusher_co::trim_lsn(int wait_base)
else if (wait_state == wait_base+1) goto resume_1; else if (wait_state == wait_base+1) goto resume_1;
else if (wait_state == wait_base+2) goto resume_2; else if (wait_state == wait_base+2) goto resume_2;
else if (wait_state == wait_base+3) goto resume_3; else if (wait_state == wait_base+3) goto resume_3;
else if (wait_state == wait_base+4) goto resume_4;
compact_lsn = bs->heap->get_compacted_lsn(); compact_lsn = bs->heap->get_compacted_lsn();
if (((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn == compact_lsn) if (((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn == compact_lsn)
{ {
return true; return true;
} }
flusher->active_flushers++; flusher->active_flushers++;
assert(!wait_count);
if (!bs->disable_meta_fsync) if (!bs->disable_meta_fsync)
{ {
await_sqe(0); await_sqe(0);
@@ -700,24 +697,34 @@ bool journal_flusher_co::trim_lsn(int wait_base)
data->callback = simple_callback_w; data->callback = simple_callback_w;
io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC); io_uring_prep_fsync(sqe, bs->dsk.meta_fd, IORING_FSYNC_DATASYNC);
wait_count++; wait_count++;
resume_1: }
if (wait_count > 0) if (!bs->disable_data_fsync && bs->dsk.data_fd != bs->dsk.meta_fd)
{ {
wait_state = wait_base+1; await_sqe(1);
return false; data->iov = { 0 };
} data->callback = simple_callback_w;
io_uring_prep_fsync(sqe, bs->dsk.data_fd, IORING_FSYNC_DATASYNC);
wait_count++;
}
resume_2:
if (wait_count > 0)
{
wait_state = wait_base+2;
return false;
} }
((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn = compact_lsn; ((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn = compact_lsn;
((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c(); ((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c();
await_sqe(2); await_sqe(3);
data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size }; data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size };
data->callback = simple_callback_w; data->callback = simple_callback_w;
io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset); io_uring_prep_writev(sqe, bs->dsk.meta_fd, &data->iov, 1, bs->dsk.meta_offset);
// Update superblock with datasync
sqe->rw_flags = RWF_DSYNC;
wait_count++; wait_count++;
resume_3: resume_4:
if (wait_count > 0) if (wait_count > 0)
{ {
wait_state = wait_base+3; wait_state = wait_base+4;
return false; return false;
} }
bs->heap->mark_lsn_trimmed(compact_lsn); bs->heap->mark_lsn_trimmed(compact_lsn);