Use linked list heap in blockstore code

WIP, still slower than the old version :-E

src/blockstore/blockstore_flush.cpp

@@ -62,6 +62,16 @@ journal_flusher_co::~journal_flusher_co()
     }
 }
 
+int journal_flusher_t::get_active()
+{
+    return active_flushers;
+}
+
+uint64_t journal_flusher_t::get_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);
@@ -107,8 +117,11 @@ void journal_flusher_t::loop()
             cur_flusher_count--;
         }
     }
+    int prev_active = active_flushers;
     for (int i = 0; is_active() && i < cur_flusher_count; i++)
         co[i].loop();
+    if (prev_active && !active_flushers && force_start > 0)
+        bs->ringloop->wakeup();
 }
 
 #define await_sqe(label) \
@@ -168,7 +181,7 @@ resume_1:
         goto resume_0;
     }
     compact_lsn = begin_wr->lsn;
-    assert(end_wr < (heap_write_t*)cur_obj->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;
     // "Lock" object for flushing
     repeat_it = flusher->sync_to_repeat.find(cur_oid);
@@ -194,7 +207,7 @@ resume_1:
     flusher->active_flushers++;
     // Scan versions to flush
     read_vec.clear();
-    for (auto wr = begin_wr; wr != end_wr; wr = wr->next(bs->heap))
+    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);
@@ -206,8 +219,7 @@ resume_1:
         overwrite_end = read_vec[read_vec.size()-1].offset + read_vec[read_vec.size()-1].len;
     }
     read_to_fill_incomplete = false;
-    if (bs->dsk.csum_block_size > bs->dsk.bitmap_granularity &&
+    if (bs->dsk.csum_block_size > bs->dsk.bitmap_granularity && end_wr->next())
-        end_wr < (heap_write_t*)cur_obj->next())
     {
         // Read original checksum blocks to calculate padded checksums if required
         fill_partial_checksum_blocks();
@@ -271,13 +283,12 @@ resume_12:
 resume_13:
     if (copy_count && !fsync_batch(false, 11))
         return false;
+    bs->heap->unlock_entry(cur_oid, cur_lsn);
     // Modify the metadata entry; don't write anything. Metadata block will be written on the next write
     calc_block_checksums();
     bs->heap->compact_object(cur_oid, cur_lsn, new_data_csums);
     // Done, free all buffers
     free_buffers();
-    // Unlock entry and free referenced block only after fsync
-    bs->heap->unlock_entry(cur_oid, cur_lsn);
 #ifdef BLOCKSTORE_DEBUG
     printf("Compacted %jx:%jx v%ju (%d writes)\n", cur_oid.inode, cur_oid.stripe, cur_version, copy_count);
 #endif
@@ -296,16 +307,16 @@ release_oid:
     repeat_it = flusher->sync_to_repeat.find(cur_oid);
     do_repeat = (repeat_it != flusher->sync_to_repeat.end() && repeat_it->second > cur_version);
     flusher->sync_to_repeat.erase(repeat_it);
+    flusher->active_flushers--;
     if (do_repeat)
     {
         // Flush the same object again
         goto resume_1;
     }
     // All done
-    flusher->active_flushers--;
+    flusher->compact_counter++;
     wait_state = 0;
     goto resume_0;
-    return true;
 }
 
 void journal_flusher_co::iterate_partial_overwrites(std::function<int(int, uint32_t, uint32_t)> cb)

src/blockstore/blockstore_flush.h

@@ -95,6 +95,7 @@ class journal_flusher_t
     friend class journal_flusher_co;
 
     int advance_lsn_counter = 0;
+    uint64_t compact_counter = 0;
 
     int active_flushers = 0;
     int syncing_flushers = 0;
@@ -105,6 +106,8 @@ public:
     journal_flusher_t(blockstore_impl_t *bs);
     ~journal_flusher_t();
     void loop();
+    int get_active();
+    uint64_t get_counter();
     bool is_active();
     void request_trim();
     void release_trim();

src/blockstore/blockstore_impl.cpp

@@ -244,11 +244,11 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     }
     else if (PRIV(op)->wait_for == WAIT_COMPACTION)
     {
-        if (heap->get_compact_queue_size() >= PRIV(op)->wait_detail)
+        if (flusher->get_counter() <= PRIV(op)->wait_detail)
         {
             // do not submit
 #ifdef BLOCKSTORE_DEBUG
-            printf("Still waiting to reduce compaction queue size below %ju\n", PRIV(op)->wait_detail);
+            printf("Still waiting for more flushes\n");
 #endif
             return;
         }

src/blockstore/blockstore_read.cpp

@@ -19,7 +19,7 @@ int blockstore_impl_t::dequeue_read(blockstore_op_t *op)
     PRIV(op)->pending_ops = 0;
     auto & rv = PRIV(op)->read_vec;
     uint64_t result_version = 0;
-    for (heap_write_t *wr = obj->get_writes(); wr < (heap_write_t*)obj->next(); wr = wr->next(heap))
+    for (auto wr = obj->get_writes(); wr; wr = wr->next())
     {
         if (op->version < wr->version)
         {
@@ -362,7 +362,7 @@ int blockstore_impl_t::read_bitmap(object_id oid, uint64_t target_version, void
     heap_object_t *obj = heap->read_entry(oid, NULL);
     if (obj)
     {
-        for (heap_write_t *wr = obj->get_writes(); wr < (heap_write_t*)obj->next(); wr = wr->next(heap))
+        for (auto wr = obj->get_writes(); wr; wr = wr->next())
         {
             if (target_version < wr->version)
             {

src/blockstore/blockstore_write.cpp

@@ -28,6 +28,8 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, int retval)
             (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE))
         {
             // Mark operations to cancel them
+            if (PRIV(other_op)->op_state != 0 && PRIV(other_op)->op_state != 100)
+                write_iodepth--;
             PRIV(other_op)->op_state = 100;
             other_op->retval = retval;
         }
@@ -85,6 +87,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         prepare_meta_block_write(op, modified_block);
         PRIV(op)->pending_ops++;
         PRIV(op)->op_state = 5;
+        write_iodepth++;
     }
     // FIXME: Allow to do initial writes as buffered, not redirected
     // FIXME: Allow to do direct writes over holes
@@ -97,20 +100,18 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         if (loc == UINT64_MAX ||
             !obj && heap->get_block_for_new_object(tmp_block) != 0)
         {
-            auto queue_size = heap->get_compact_queue_size();
+            if (!heap->get_compact_queue_size() && !flusher->get_active())
-            if (!queue_size)
             {
                 // no space
                 cancel_all_writes(op, -ENOSPC);
                 return 2;
             }
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = queue_size;
+            PRIV(op)->wait_detail = flusher->get_counter();
             flusher->request_trim();
             return 0;
         }
         BS_SUBMIT_GET_SQE(sqe, data);
-        write_iodepth++;
         PRIV(op)->location = loc;
 #ifdef BLOCKSTORE_DEBUG
         printf(
@@ -141,6 +142,7 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         PRIV(op)->pending_ops = 1;
         unsynced_big_write_count++;
         PRIV(op)->op_state = 1;
+        write_iodepth++;
     }
     else
     {
@@ -150,13 +152,12 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         if (loc == UINT64_MAX)
         {
             PRIV(op)->wait_for = WAIT_COMPACTION;
-            PRIV(op)->wait_detail = heap->get_compact_queue_size();
+            PRIV(op)->wait_detail = flusher->get_counter();
             flusher->request_trim();
             return 0;
         }
         // There is sufficient space. Check SQE(s)
         BS_SUBMIT_CHECK_SQES(1 + (op->len > 0 ? 1 : 0));
-        write_iodepth++;
         uint8_t wr_buf[heap->get_max_write_entry_size()];
         heap_write_t *wr = (heap_write_t*)wr_buf;
         wr->version = op->version;
@@ -173,12 +174,15 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         int res = heap->post_write(op->oid, wr, &modified_block);
         if (res == ENOSPC)
         {
-            cancel_all_writes(op, -ENOSPC);
+            if (!heap->get_compact_queue_size() && !flusher->get_active())
-            return 2;
+            {
-        }
+                // no space
-        else if (res == EAGAIN)
+                cancel_all_writes(op, -ENOSPC);
-        {
+                return 2;
-            // Pause submission, wait for compaction
+            }
+            PRIV(op)->wait_for = WAIT_COMPACTION;
+            PRIV(op)->wait_detail = flusher->get_counter();
+            flusher->request_trim();
             return 0;
         }
         assert(res == 0);
@@ -202,11 +206,13 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         if (!PRIV(op)->pending_ops)
         {
             PRIV(op)->op_state = 6;
+            write_iodepth++;
             return continue_write(op);
         }
         else
         {
             PRIV(op)->op_state = 5;
+            write_iodepth++;
         }
     }
     return 1;
@@ -260,7 +266,8 @@ resume_4:
         int res = heap->post_write(op->oid, wr, &modified_block);
         if (res == ENOSPC)
         {
-            // wait for compaction
+            PRIV(op)->wait_for = WAIT_COMPACTION;
+            PRIV(op)->wait_detail = flusher->get_counter();
             return 1;
         }
         assert(res == 0);