Use the same "inflight" queue to track compaction

src/blockstore/blockstore_flush.cpp

@@ -68,14 +68,14 @@ int journal_flusher_t::get_active()
     return active_flushers;
 }
 
-uint64_t journal_flusher_t::get_counter()
+uint64_t journal_flusher_t::get_compact_counter()
 {
     return compact_counter;
 }
 
 bool journal_flusher_t::is_active()
 {
-    return active_flushers > 0 || bs->heap->get_compact_queue_size() > (force_start > 0 ? 0 : bs->flusher_start_threshold);
+    return active_flushers > 0 || force_start > 0 || bs->heap->get_compact_queue_size() > bs->flusher_start_threshold;
 }
 
 void journal_flusher_t::request_trim()
@@ -158,10 +158,23 @@ bool journal_flusher_co::loop()
     else if (wait_state == 15) goto resume_15;
     else if (wait_state == 16) goto resume_16;
     else if (wait_state == 17) goto resume_17;
+    else if (wait_state == 18) goto resume_18;
+    else if (wait_state == 19) goto resume_19;
+    else if (wait_state == 20) goto resume_20;
+    else if (wait_state == 21) goto resume_21;
 resume_0:
     res = bs->heap->get_next_compact(cur_oid);
     if (res == ENOENT)
     {
+        if (co_id == 0 && flusher->force_start > 0)
+        {
+resume_18:
+resume_19:
+resume_20:
+resume_21:
+            if (!trim_lsn(18))
+                return false;
+        }
         cur_oid = {};
         wait_state = 0;
         return true;
@@ -203,7 +216,6 @@ resume_1:
     read_vec.clear();
     for (auto wr = begin_wr; wr != end_wr; wr = wr->next())
     {
-        min_compact_lsn = wr->lsn;
         bs->prepare_read(read_vec, cur_obj, wr, 0, bs->dsk.data_block_size);
     }
     overwrite_start = overwrite_end = 0;
@@ -277,19 +289,30 @@ 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...
     bs->heap->unlock_entry(cur_oid, copy_id);
-    // Modify the metadata entry; don't write anything. Metadata block will be written on the next write
+    // Mark the object compacted, but don't free and remove small_writes
+    // We'll free and remove them only when trimming
+    // The only thing we modify here are big_write block checksums if >4k block is used
+    cur_obj = bs->heap->read_entry(cur_oid, NULL);
+    if (!cur_obj)
+    {
+        // Abort compaction
+        goto release_oid;
+    }
     calc_block_checksums();
-    bs->heap->compact_object(cur_oid, compact_lsn, new_data_csums);
+    bs->heap->mark_object_compacted(cur_obj, compact_lsn);
     // Done, free all buffers
     free_buffers();
 #ifdef BLOCKSTORE_DEBUG
     printf("Compacted %jx:%jx v%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, cur_version, copy_count);
 #endif
+    flusher->compact_counter++;
+    flusher->active_flushers--;
     // Advance compacted_lsn every <journal_trim_interval> objects
-    bs->heap->set_compacted_lsn(min_compact_lsn);
-    if (bs->journal_trim_interval && !((++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;
 resume_14:
 resume_15:
 resume_16:
@@ -298,14 +321,12 @@ resume_17:
             return false;
     }
 release_oid:
-    flusher->active_flushers--;
     if (should_repeat)
     {
         // Flush the same object again
         goto resume_1;
     }
     // All done
-    flusher->compact_counter++;
     wait_state = 0;
     goto resume_0;
 }
@@ -479,7 +500,6 @@ void journal_flusher_co::calc_block_checksums()
     if (bs->dsk.csum_block_size <= bs->dsk.bitmap_granularity)
         return;
     new_data_csums = csum_buf + overwrite_start/bs->dsk.csum_block_size * (bs->dsk.data_csum_type & 0xFF);
-    cur_obj = bs->heap->read_locked_entry(cur_oid, copy_id);
     uint64_t block_offset = 0;
     uint32_t block_done = 0;
     uint32_t block_csum = 0;
@@ -667,6 +687,12 @@ 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+2) goto resume_2;
     else if (wait_state == wait_base+3) goto resume_3;
+    compact_lsn = bs->heap->get_compacted_lsn();
+    if (((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn == compact_lsn)
+    {
+        return true;
+    }
+    flusher->active_flushers++;
     if (!bs->disable_meta_fsync)
     {
         await_sqe(0);
@@ -681,7 +707,7 @@ resume_1:
             return false;
         }
     }
-    ((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn = bs->heap->get_compacted_lsn();
+    ((blockstore_meta_header_v3_t*)bs->meta_superblock)->compacted_lsn = compact_lsn;
     ((blockstore_meta_header_v3_t*)bs->meta_superblock)->set_crc32c();
     await_sqe(2);
     data->iov = (struct iovec){ bs->meta_superblock, (size_t)bs->dsk.meta_block_size };
@@ -694,6 +720,8 @@ resume_3:
         wait_state = wait_base+3;
         return false;
     }
-    flusher->advance_lsn_counter = 0;
+    bs->heap->mark_lsn_trimmed(compact_lsn);
+    flusher->compact_counter++;
+    flusher->active_flushers--;
     return true;
 }

src/blockstore/blockstore_flush.h

@@ -51,7 +51,6 @@ class journal_flusher_co
     object_id cur_oid;
     uint64_t copy_id;
     uint64_t compact_lsn;
-    uint64_t min_compact_lsn;
     uint64_t cur_version;
     heap_object_t *cur_obj;
     heap_write_t *begin_wr, *end_wr;
@@ -108,7 +107,7 @@ public:
     ~journal_flusher_t();
     void loop();
     int get_active();
-    uint64_t get_counter();
+    uint64_t get_compact_counter();
     bool is_active();
     void request_trim();
     void release_trim();

src/blockstore/blockstore_heap.cpp

@@ -61,7 +61,7 @@ bool heap_write_t::needs_recheck(blockstore_heap_t *heap)
 
 bool heap_write_t::needs_compact(uint64_t compacted_lsn)
 {
-    return lsn > compacted_lsn && (flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE) || flags == (BS_HEAP_INTENT_WRITE|BS_HEAP_STABLE));
+    return lsn > compacted_lsn && flags == (BS_HEAP_SMALL_WRITE|BS_HEAP_STABLE);
 }
 
 bool heap_write_t::is_compacted(uint64_t compacted_lsn)
@@ -585,9 +585,9 @@ skip_object:
                     // Mark data block as used
                     use_data(obj->inode, wr->location);
                 }
-                if (wr->needs_compact(this->compacted_lsn))
+                if (wr->lsn > this->compacted_lsn)
                 {
-                    tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn });
+                    tmp_compact_queue.push_back((tmp_compact_item_t){ .oid = oid, .lsn = wr->lsn, .compact = wr->needs_compact(0) });
                 }
                 else if (wr->is_compacted(this->compacted_lsn))
                 {
@@ -599,7 +599,7 @@ skip_object:
                     {
                         // We can't just collapse the object entry when csum_block_size is larger
                         // than bitmap_granularity, so we add the object into the compact queue
-                        tmp_compact_queue.push_back((heap_object_lsn_t){ .oid = oid, .lsn = wr->lsn });
+                        tmp_compact_queue.push_back((tmp_compact_item_t){ .oid = oid, .lsn = wr->lsn, .compact = wr->needs_compact(0) });
                     }
                 }
             }
@@ -643,13 +643,28 @@ skip_object:
 
 void blockstore_heap_t::finish_load()
 {
-    std::sort(tmp_compact_queue.begin(), tmp_compact_queue.end(), [this](const heap_object_lsn_t & a, const heap_object_lsn_t & b)
+    completed_lsn = first_inflight_lsn = next_lsn+1;
+    if (!tmp_compact_queue.size())
+    {
+        return;
+    }
+    std::sort(tmp_compact_queue.begin(), tmp_compact_queue.end(), [this](const tmp_compact_item_t & a, const tmp_compact_item_t & b)
     {
         return a.lsn < b.lsn;
     });
+    first_inflight_lsn = tmp_compact_queue[0].lsn;
+    if (compacted_lsn < tmp_compact_queue[0].lsn-1)
+    {
+        compacted_lsn = tmp_compact_queue[0].lsn-1;
+    }
     for (auto & e: tmp_compact_queue)
     {
-        compact_queue.push_back(e.oid);
+        push_inflight_lsn(e.oid, e.lsn, HEAP_INFLIGHT_DONE | (e.compact ? HEAP_INFLIGHT_COMPACTABLE : 0));
+    }
+    while (compacted_lsn+1-first_inflight_lsn < inflight_lsn.size() &&
+        !(inflight_lsn[compacted_lsn+1-first_inflight_lsn].flags & HEAP_INFLIGHT_COMPACTABLE))
+    {
+        compacted_lsn++;
     }
     tmp_compact_queue.clear();
 }
@@ -758,7 +773,7 @@ bool blockstore_heap_t::recheck_small_writes(std::function<void(bool is_data, ui
                 uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, wr->len);
                 if (log_level > 5)
                 {
-                    fprintf(stderr, "Notice: rechecking %u bytes at %ju in %s area\n", wr->len, loc, is_intent ? "data" : "buffer");
+                    fprintf(stderr, "Notice: rechecking %u bytes at %ju in %s area (lsn %lu)\n", wr->len, loc, is_intent ? "data" : "buffer", wr->lsn);
                 }
                 recheck_cb(is_intent, loc, wr->len, buf, [this, oid, lsn = wr->lsn, buf]()
                 {
@@ -1264,7 +1279,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(oid, new_wr);
+    push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
     if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
     {
         uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
@@ -1430,7 +1445,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(oid, new_wr);
+    push_inflight_lsn(oid, new_wr->lsn, new_wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
     if ((wr->flags & BS_HEAP_TYPE) == BS_HEAP_BIG_WRITE)
     {
         uint8_t *int_bitmap = new_wr->get_int_bitmap(this);
@@ -1528,7 +1543,7 @@ int blockstore_heap_t::post_stabilize(object_id oid, uint64_t version, uint32_t
         {
             wr->flags |= BS_HEAP_STABLE;
             wr->lsn = last_lsn--;
-            push_inflight_lsn(oid, wr);
+            push_inflight_lsn(oid, wr->lsn, wr->needs_compact(0) ? HEAP_INFLIGHT_COMPACTABLE : 0);
         }
     }
     obj->crc32c = obj->calc_crc32c();
@@ -1608,10 +1623,24 @@ int blockstore_heap_t::post_delete(object_id oid, uint32_t *modified_block)
 
 int blockstore_heap_t::get_next_compact(object_id & oid)
 {
-    while (compact_queue.size())
+    if (next_compact_lsn < first_inflight_lsn)
     {
-        oid = compact_queue.front();
-        compact_queue.pop_front();
+        next_compact_lsn = first_inflight_lsn;
+    }
+    while (next_compact_lsn-first_inflight_lsn < inflight_lsn.size())
+    {
+        auto & item = inflight_lsn[next_compact_lsn-first_inflight_lsn];
+        if (!(item.flags & HEAP_INFLIGHT_COMPACTABLE))
+        {
+            next_compact_lsn++;
+            continue;
+        }
+        if (!(item.flags & HEAP_INFLIGHT_DONE))
+        {
+            break;
+        }
+        next_compact_lsn++;
+        oid = item.oid;
         return 0;
     }
     return ENOENT;
@@ -1918,7 +1947,12 @@ uint32_t blockstore_heap_t::get_meta_nearfull_blocks()
 
 uint32_t blockstore_heap_t::get_compact_queue_size()
 {
-    return compact_queue.size();
+    return to_compact_count;
+}
+
+uint32_t blockstore_heap_t::get_inflight_queue_size()
+{
+    return inflight_lsn.size();
 }
 
 uint32_t blockstore_heap_t::get_max_write_entry_size()
@@ -1931,39 +1965,95 @@ void blockstore_heap_t::set_fail_on_warn(bool fail)
     fail_on_warn = fail;
 }
 
-void blockstore_heap_t::push_inflight_lsn(object_id oid, heap_write_t *wr)
+void blockstore_heap_t::push_inflight_lsn(object_id oid, uint64_t lsn, uint64_t flags)
 {
     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)
+    if (flags & HEAP_INFLIGHT_COMPACTABLE)
+    {
+        to_compact_count++;
+    }
+    if (lsn == next_inf)
     {
         inflight_lsn.push_back((heap_inflight_lsn_t){ .oid = oid, .flags = flags });
     }
     else
     {
-        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 };
+        if (lsn > next_inf)
+        {
+            inflight_lsn.resize(lsn-first_inflight_lsn+1, (heap_inflight_lsn_t){ .flags = HEAP_INFLIGHT_DONE });
+        }
+        inflight_lsn[lsn-first_inflight_lsn] = (heap_inflight_lsn_t){ .oid = oid, .flags = flags };
     }
 }
 
-void blockstore_heap_t::complete_lsn(uint64_t lsn)
+void blockstore_heap_t::mark_lsn_completed(uint64_t lsn)
 {
     assert(lsn >= first_inflight_lsn && lsn < first_inflight_lsn+inflight_lsn.size());
-    assert(!(inflight_lsn[lsn - first_inflight_lsn].flags & HEAP_INFLIGHT_DONE));
-    inflight_lsn[lsn - first_inflight_lsn].flags |= HEAP_INFLIGHT_DONE;
-    if (lsn == first_inflight_lsn)
+    auto & item = inflight_lsn[lsn - first_inflight_lsn];
+    assert(!(item.flags & HEAP_INFLIGHT_DONE));
+    item.flags |= HEAP_INFLIGHT_DONE;
+    if (lsn == compacted_lsn+1 && !(item.flags & HEAP_INFLIGHT_COMPACTABLE))
     {
-        while (inflight_lsn.size() && (inflight_lsn[0].flags & HEAP_INFLIGHT_DONE))
+        assert(compacted_lsn+1 >= first_inflight_lsn);
+        while (compacted_lsn+1-first_inflight_lsn < inflight_lsn.size() &&
+            (inflight_lsn[compacted_lsn+1-first_inflight_lsn].flags == HEAP_INFLIGHT_DONE))
         {
-            if ((inflight_lsn[0].flags & HEAP_INFLIGHT_COMPACTABLE))
-            {
-                compact_queue.push_back(inflight_lsn[0].oid);
-            }
-            inflight_lsn.pop_front();
-            first_inflight_lsn++;
+            compacted_lsn++;
         }
-        completed_lsn = first_inflight_lsn-1;
+        assert(inflight_lsn[compacted_lsn-first_inflight_lsn].flags == HEAP_INFLIGHT_DONE);
+    }
+    if (lsn > completed_lsn)
+    {
+        assert(completed_lsn+1 >= first_inflight_lsn);
+        while (completed_lsn+1-first_inflight_lsn < inflight_lsn.size() &&
+            (inflight_lsn[completed_lsn+1-first_inflight_lsn].flags & HEAP_INFLIGHT_DONE))
+        {
+            completed_lsn++;
+        }
+    }
+}
+
+void blockstore_heap_t::mark_lsn_compacted(uint64_t lsn)
+{
+    assert(lsn >= first_inflight_lsn && lsn < first_inflight_lsn+inflight_lsn.size());
+    auto & item = inflight_lsn[lsn - first_inflight_lsn];
+    assert(item.flags & HEAP_INFLIGHT_DONE);
+    if (!(item.flags & HEAP_INFLIGHT_COMPACTABLE))
+        return;
+    item.flags -= HEAP_INFLIGHT_COMPACTABLE;
+    to_compact_count--;
+    if (lsn == compacted_lsn+1)
+    {
+        assert(completed_lsn+1 >= first_inflight_lsn);
+        while (compacted_lsn+1-first_inflight_lsn < inflight_lsn.size() &&
+            (inflight_lsn[compacted_lsn+1-first_inflight_lsn].flags == HEAP_INFLIGHT_DONE))
+        {
+            compacted_lsn++;
+        }
+    }
+}
+
+void blockstore_heap_t::mark_object_compacted(heap_object_t *obj, uint64_t max_lsn)
+{
+    for (auto wr = obj->get_writes(); wr; wr = wr->next())
+    {
+        if (wr->is_compacted(max_lsn))
+        {
+            mark_lsn_compacted(wr->lsn);
+        }
+    }
+}
+
+void blockstore_heap_t::mark_lsn_trimmed(uint64_t lsn)
+{
+    assert(lsn >= first_inflight_lsn && lsn < first_inflight_lsn+inflight_lsn.size());
+    while (first_inflight_lsn <= lsn && inflight_lsn.size() > 0)
+    {
+        assert(inflight_lsn[0].flags == HEAP_INFLIGHT_DONE);
+        // FIXME don't touch unflushable lsns
+        compact_object(inflight_lsn[0].oid, lsn, NULL); // FIXME from prev trimmed lsn
+        inflight_lsn.pop_front();
+        first_inflight_lsn++;
     }
 }
 
@@ -1971,8 +2061,3 @@ uint64_t blockstore_heap_t::get_completed_lsn()
 {
     return completed_lsn;
 }
-
-void blockstore_heap_t::add_to_compact_queue(object_id oid)
-{
-    compact_queue.push_back(oid);
-}

src/blockstore/blockstore_heap.h

@@ -84,6 +84,13 @@ inline bool operator < (const heap_object_lsn_t & a, const heap_object_lsn_t & b
     return a.oid < b.oid || a.oid == b.oid && a.lsn < b.lsn;
 }
 
+struct tmp_compact_item_t
+{
+    object_id oid;
+    uint64_t lsn;
+    bool compact;
+};
+
 struct heap_object_mvcc_t
 {
     uint32_t readers = 0;
@@ -136,11 +143,9 @@ class blockstore_heap_t
     const int meta_alloc_buckets = 4;
 
     uint64_t next_lsn = 0;
-    uint64_t compacted_lsn = 0;
     std::map<pool_id_t, pool_shard_settings_t> pool_shard_settings;
     // PG => inode => stripe => block number
     std::map<uint64_t, std::map<inode_t, btree::btree_map<uint64_t, uint64_t>>> block_index;
-    std::deque<object_id> compact_queue;
     std::vector<heap_block_info_t> block_info;
     allocator_t *data_alloc = NULL;
     allocator_t *meta_allocs[4] = {};
@@ -154,11 +159,15 @@ class blockstore_heap_t
     uint64_t buffer_area_used_space = 0;
     uint64_t data_used_space = 0;
 
+    // LSN queue: inflight (writing) -> completed [-> fsynced] -> compactable -> compacted [-> fsynced] -> trimmed and removed
     std::deque<heap_inflight_lsn_t> inflight_lsn;
-    uint64_t first_inflight_lsn = 1;
+    uint32_t to_compact_count = 0;
+    uint64_t first_inflight_lsn = 0;
     uint64_t completed_lsn = 0;
+    uint64_t compacted_lsn = 0;
+    uint64_t next_compact_lsn = 0;
 
-    std::vector<heap_object_lsn_t> tmp_compact_queue;
+    std::vector<tmp_compact_item_t> tmp_compact_queue;
     std::deque<object_id> recheck_queue;
     int recheck_in_progress = 0;
     bool in_recheck = false;
@@ -180,7 +189,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);
+    void push_inflight_lsn(object_id oid, uint64_t lsn, uint64_t flags);
 
 public:
     blockstore_heap_t(blockstore_disk_t *dsk, uint8_t *buffer_area, int log_level = 0);
@@ -243,9 +252,11 @@ public:
     int get_block_for_new_object(uint32_t & out_block_num);
 
     // inflight write tracking
-    void complete_lsn(uint64_t lsn);
+    void mark_lsn_completed(uint64_t lsn);
+    void mark_lsn_compacted(uint64_t lsn);
+    void mark_object_compacted(heap_object_t *obj, uint64_t max_lsn);
+    void mark_lsn_trimmed(uint64_t lsn);
     uint64_t get_completed_lsn();
-    void add_to_compact_queue(object_id oid);
 
     // data device block allocator functions
     uint64_t find_free_data();
@@ -269,6 +280,7 @@ public:
     uint64_t get_meta_total_space();
     uint64_t get_meta_used_space();
     uint32_t get_meta_nearfull_blocks();
+    uint32_t get_inflight_queue_size();
     uint32_t get_compact_queue_size();
 
     // get maximum size for a temporary heap_write_t buffer

src/blockstore/blockstore_impl.cpp

@@ -244,7 +244,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     }
     else if (PRIV(op)->wait_for == WAIT_COMPACTION)
     {
-        if (flusher->get_counter() <= PRIV(op)->wait_detail)
+        if (flusher->get_compact_counter() <= PRIV(op)->wait_detail)
         {
             // do not submit
 #ifdef BLOCKSTORE_DEBUG

src/blockstore/blockstore_init.cpp

@@ -117,7 +117,7 @@ resume_1:
     }
     else
     {
-        blockstore_meta_header_v2_t *hdr = (blockstore_meta_header_v2_t *)bs->meta_superblock;
+        blockstore_meta_header_v3_t *hdr = (blockstore_meta_header_v3_t *)bs->meta_superblock;
         if (hdr->zero != 0 || hdr->magic != BLOCKSTORE_META_MAGIC_V1 || hdr->version < BLOCKSTORE_META_FORMAT_V1)
         {
             printf(
@@ -153,6 +153,7 @@ resume_1:
             );
             exit(1);
         }
+        bs->heap->set_compacted_lsn(hdr->compacted_lsn);
     }
     if (bs->dsk.inmemory_journal)
     {

src/blockstore/blockstore_stable.cpp

@@ -86,7 +86,7 @@ resume_4:
     // FIXME: Also mark as fsynced
     for (uint64_t lsn = priv->lsn; lsn <= priv->to_lsn; lsn++)
     {
-        heap->complete_lsn(lsn);
+        heap->mark_lsn_completed(lsn);
     }
     // Done. Don't touch op->retval - if anything resulted in ENOENT, return it as is
     FINISH_OP(op);

src/blockstore/blockstore_write.cpp

@@ -103,14 +103,14 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         if (loc == UINT64_MAX ||
             !obj && heap->get_block_for_new_object(tmp_block) != 0)
         {
-            if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active())
+            if (!heap->get_inflight_queue_size())
             {
                 // no space
                 cancel_all_writes(op, -ENOSPC);
                 return 2;
             }
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_counter();
+            PRIV(op)->wait_detail = flusher->get_compact_counter();
             flusher->request_trim();
             return 0;
         }
@@ -179,14 +179,14 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         int res = heap->post_write(op->oid, wr, &modified_block);
         if (res == ENOSPC)
         {
-            if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active())
+            if (!heap->get_inflight_queue_size())
             {
                 // no space
                 cancel_all_writes(op, -ENOSPC);
                 return 2;
             }
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_counter();
+            PRIV(op)->wait_detail = flusher->get_compact_counter();
             flusher->request_trim();
             return 0;
         }
@@ -205,7 +205,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         if (loc == UINT64_MAX)
         {
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_counter();
+            PRIV(op)->wait_detail = flusher->get_compact_counter();
             flusher->request_trim();
             return 0;
         }
@@ -226,14 +226,14 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         int res = heap->post_write(op->oid, wr, &modified_block);
         if (res == ENOSPC)
         {
-            if (!heap->get_compact_queue_size() && !write_iodepth && !flusher->get_active())
+            if (!heap->get_inflight_queue_size())
             {
                 // no space
                 cancel_all_writes(op, -ENOSPC);
                 return 2;
             }
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_counter();
+            PRIV(op)->wait_detail = flusher->get_compact_counter();
             flusher->request_trim();
             return 0;
         }
@@ -315,7 +315,7 @@ resume_4:
         if (res == ENOSPC)
         {
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = flusher->get_counter();
+            PRIV(op)->wait_detail = flusher->get_compact_counter();
             return 1;
         }
         assert(res == 0);
@@ -364,7 +364,7 @@ resume_6:
 resume_8:
     // Acknowledge write
     op->retval = op->len;
-    heap->complete_lsn(PRIV(op)->lsn);
+    heap->mark_lsn_completed(PRIV(op)->lsn);
     write_iodepth--;
     FINISH_OP(op);
     return 2;

src/test/test_heap.cpp

@@ -403,6 +403,7 @@ void test_compact(bool csum, bool stable)
     _test_small_write(heap, dsk, 1, 0, 3, 8192, 4096, 16384, stable);
     obj = heap.read_entry(oid, NULL);
     uint64_t wr_size = obj->get_writes()->get_size(&heap);
+    assert(obj->get_writes()->lsn == 2);
     assert(check_used_space(heap, dsk, 0));
     assert(heap.get_meta_block_used_space(0) == old_size + wr_size);
 
@@ -412,6 +413,10 @@ void test_compact(bool csum, bool stable)
     assert(obj);
     assert(count_writes(obj) == 2);
 
+    heap.mark_lsn_completed(1);
+    heap.mark_lsn_completed(2);
+    heap.mark_lsn_completed(3);
+
     uint32_t mblock;
     object_id compact_oid = {};
     if (!stable)
@@ -435,9 +440,9 @@ void test_compact(bool csum, bool stable)
         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);
+        heap.mark_lsn_completed(4);
     }
 
-    heap.add_to_compact_queue(oid);
     assert(heap.get_compact_queue_size() == 1);
     res = heap.get_next_compact(compact_oid);
     assert(res == 0);
@@ -447,15 +452,18 @@ void test_compact(bool csum, bool stable)
     obj = heap.read_entry(oid, NULL);
     assert(obj);
     assert(count_writes(obj) == 2);
-    heap.get_compact_range(obj, UINT64_MAX, &compact_begin, &compact_end);
+    heap.get_compact_range(obj, 4, &compact_begin, &compact_end);
     assert(compact_begin == obj->get_writes());
     assert(compact_end == obj->get_writes()->next());
 
+    heap.mark_object_compacted(obj, 4);
+    assert(heap.get_compacted_lsn() == (stable ? 3 : 4));
+    assert(heap.get_compact_queue_size() == 0);
+
     res = heap.compact_object((object_id){ .inode = INODE_WITH_POOL(1, 3), .stripe = 0 }, compact_begin->lsn, NULL);
     assert(res == ENOENT);
 
-    res = heap.compact_object(compact_oid, compact_begin->lsn, NULL);
-    assert(res == 0);
+    heap.mark_lsn_trimmed((stable ? 3 : 4));
     assert(check_used_space(heap, dsk, 0));
     assert(heap.get_meta_block_used_space(0) == 2*old_size);