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

Assign new LSN on stabilize

Browse Source
This commit is contained in:
Vitaliy Filippov
2025-12-02 01:52:12 +03:00
parent 59b7b2e0c3
commit b44c3a7971
3 changed files with 38 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/blockstore/blockstore_heap.cpp
+26 -24
View File
@@ -1264,7 +1264,7 @@ int blockstore_heap_t::add_object(object_id oid, heap_write_t *wr, uint32_t *mod
new_wr->size = wr_size;
new_wr->lsn = ++next_lsn;
wr->lsn = new_wr->lsn;
push_inflight_lsn(new_wr->lsn, oid, wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
push_inflight_lsn(oid, new_wr);
if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
{
uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
@@ -1430,7 +1430,7 @@ int blockstore_heap_t::update_object(uint32_t block_num, heap_object_t *obj, hea
new_wr->size = wr_size;
new_wr->lsn = ++next_lsn;
wr->lsn = new_wr->lsn;
push_inflight_lsn(new_wr->lsn, oid, wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
push_inflight_lsn(oid, new_wr);
if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
{
uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
@@ -1455,7 +1455,7 @@ int blockstore_heap_t::post_write(object_id oid, heap_write_t *wr, uint32_t *mod
return update_object(block_num, obj, wr, modified_block);
}
int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *before_compact_lsn, uint64_t *to_compact_lsn)
int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *new_lsn, uint64_t *new_to_lsn)
{
uint32_t block_num = 0;
heap_object_t *obj = read_entry(oid, &block_num);
@@ -1469,21 +1469,21 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
heap_write_t *unstable_wr = NULL;
heap_write_t *unstable_big_wr = NULL;
heap_write_t *wr = obj->get_writes();
heap_write_t *pre_wr = NULL;
if (wr->version < version)
{
// No such version
return ENOENT;
}
uint64_t stab_count = 0;
for (; wr; wr = wr->next())
{
if ((wr->flags & BS_HEAP_STABLE))
{
pre_wr = wr;
break;
}
else if (wr->version <= version)
{
stab_count++;
unstable_wr = wr;
if (!unstable_big_wr &&
((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE ||
@@ -1502,10 +1502,6 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
{
*modified_block = block_num;
}
if (before_compact_lsn && pre_wr && pre_wr->needs_compact(0))
{
*before_compact_lsn = pre_wr->lsn;
}
// Save a copy of the object
if (unstable_big_wr && unstable_big_wr->next())
{
@@ -1515,22 +1511,25 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
add_used_space(block_num, -free_writes(unstable_big_wr->next(), NULL));
unstable_big_wr->next_pos = 0;
}
// Set the stability flag
uint64_t to_compact = 0;
// Set the stability flag and assign new LSNs
if (new_lsn)
{
*new_lsn = next_lsn+1;
}
if (new_to_lsn)
{
*new_to_lsn = next_lsn+stab_count;
}
next_lsn += stab_count;
uint64_t last_lsn = next_lsn;
for (wr = obj->get_writes(); wr; wr = wr->next())
{
if (!(wr->flags & BS_HEAP_STABLE) && wr->version <= version)
{
wr->flags |= BS_HEAP_STABLE;
wr->lsn = last_lsn--;
push_inflight_lsn(oid, wr);
}
if (wr->needs_compact(0) && to_compact == 0)
{
to_compact = wr->lsn;
}
}
if (to_compact_lsn)
{
*to_compact_lsn = to_compact;
}
obj->crc32c = obj->calc_crc32c();
return 0;
@@ -1932,17 +1931,20 @@ void blockstore_heap_t::set_fail_on_warn(bool fail)
fail_on_warn = fail;
}
void blockstore_heap_t::push_inflight_lsn(uint64_t lsn, object_id oid, uint64_t flags)
void blockstore_heap_t::push_inflight_lsn(object_id oid, heap_write_t *wr)
{
if (!inflight_lsn.size())
uint64_t next_inf = first_inflight_lsn + inflight_lsn.size();
uint64_t flags = wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0;
if (wr->lsn == next_inf)
{
first_inflight_lsn = lsn;
inflight_lsn.push_back((heap_inflight_lsn_t){ .oid = oid, .flags = flags });
}
else
{
assert(lsn == first_inflight_lsn+inflight_lsn.size());
if (wr->lsn > next_inf)
inflight_lsn.resize(wr->lsn-first_inflight_lsn+1);
inflight_lsn[wr->lsn-first_inflight_lsn] = (heap_inflight_lsn_t){ .oid = oid, .flags = flags };
}
inflight_lsn.push_back((heap_inflight_lsn_t){ .oid = oid, .flags = flags });
}
void blockstore_heap_t::complete_lsn(uint64_t lsn)
src/blockstore/blockstore_heap.h
+3 -3
View File
@@ -155,7 +155,7 @@ class blockstore_heap_t
uint64_t data_used_space = 0;
std::deque<heap_inflight_lsn_t> inflight_lsn;
uint64_t first_inflight_lsn = 0;
uint64_t first_inflight_lsn = 1;
uint64_t completed_lsn = 0;
std::vector<heap_object_lsn_t> tmp_compact_queue;
@@ -180,6 +180,7 @@ class blockstore_heap_t
void erase_block_index(inode_t inode, uint64_t stripe);
void free_object_space(inode_t inode, heap_write_t *from, heap_write_t *to, int mode = 0);
void add_used_space(uint32_t block_num, int32_t used_delta);
void push_inflight_lsn(object_id oid, heap_write_t *wr);
public:
blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level = 0);
@@ -220,7 +221,7 @@ public:
int post_write(object_id oid, heap_write_t *wr, uint32_t *modified_block);
// stabilize an unstable object version
// return 0 if OK, ENOENT if not exists
int post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *before_compact_lsn, uint64_t *to_compact_lsn);
int post_stabilize(object_id oid, uint64_t version, uint32_t *modified_block, uint64_t *new_lsn, uint64_t *new_to_lsn);
// rollback an unstable object version
// return 0 if OK, ENOENT if not exists, EBUSY if already stable
int post_rollback(object_id oid, uint64_t version, uint32_t *modified_block);
@@ -242,7 +243,6 @@ public:
int get_block_for_new_object(uint32_t & out_block_num);
// inflight write tracking
void push_inflight_lsn(uint64_t lsn, object_id oid, uint64_t flags);
void complete_lsn(uint64_t lsn);
uint64_t get_completed_lsn();
void add_to_compact_queue(object_id oid);
src/test/test_heap.cpp
+9 -9
View File
@@ -419,20 +419,20 @@ void test_compact(bool csum, bool stable)
res = heap.get_next_compact(compact_oid);
assert(res == ENOENT);
uint64_t to_lsn = 0, before_lsn = 0;
res = heap.post_stabilize({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, 3, NULL, &before_lsn, &to_lsn);
uint64_t new_lsn = 0, new_to_lsn = 0;
res = heap.post_stabilize({ .inode = INODE_WITH_POOL(1, 2), .stripe = 0 }, 3, NULL, &new_lsn, &new_to_lsn);
assert(res == ENOENT);
res = heap.post_stabilize(oid, 5, NULL, &before_lsn, &to_lsn);
res = heap.post_stabilize(oid, 5, NULL, &new_lsn, &new_to_lsn);
assert(res == ENOENT);
res = heap.post_stabilize(oid, 1, &mblock, &before_lsn, &to_lsn);
res = heap.post_stabilize(oid, 1, &mblock, &new_lsn, &new_to_lsn);
assert(res == 0);
assert(before_lsn == 0);
assert(to_lsn == 0);
res = heap.post_stabilize(oid, 3, &mblock, &before_lsn, &to_lsn);
assert(new_lsn == 0);
assert(new_to_lsn == 0);
res = heap.post_stabilize(oid, 3, &mblock, &new_lsn, &new_to_lsn);
assert(res == 0);
assert(mblock == 0);
assert(before_lsn == 0);
assert(to_lsn == 2);
assert(new_lsn == 4);
assert(new_to_lsn == 4);
assert(check_used_space(heap, dsk, 0));
assert(heap.get_meta_block_used_space(0) == 2*old_size + wr_size);
}