Rewrite metadata_init to the same "goto-coroutine" style

blockstore_flush.cpp

@@ -8,7 +8,7 @@ journal_flusher_t::journal_flusher_t(int flusher_count, blockstore *bs)
     active_until_sync = 0;
     sync_required = true;
     sync_threshold = flusher_count == 1 ? 1 : flusher_count/2;
-    journal_trim_interval = sync_threshold;
+    journal_trim_interval = 1;//sync_threshold; //FIXME
     journal_trim_counter = 0;
     journal_superblock = (uint8_t*)memalign(512, 512);
     co = new journal_flusher_co[flusher_count];
@@ -43,12 +43,12 @@ journal_flusher_t::~journal_flusher_t()
 
 void journal_flusher_t::loop()
 {
-    if (!active_flushers && !flush_queue.size())
-    {
-        return;
-    }
     for (int i = 0; i < flusher_count; i++)
     {
+        if (!active_flushers && !flush_queue.size())
+        {
+            return;
+        }
         co[i].loop();
     }
 }

blockstore_init.cpp

@@ -22,55 +22,60 @@ void blockstore_init_meta::handle_event(ring_data_t *data)
 
 int blockstore_init_meta::loop()
 {
-    if (metadata_read >= bs->meta_len)
-    {
-        return 0;
-    }
+    if (wait_state == 1)
+        goto resume_1;
+    metadata_buffer = (uint8_t*)memalign(512, 2*bs->metadata_buf_size);
     if (!metadata_buffer)
+        throw std::bad_alloc();
+    while (1)
     {
-        metadata_buffer = (uint8_t*)memalign(512, 2*bs->metadata_buf_size);
-        if (!metadata_buffer)
-            throw std::bad_alloc();
-    }
-    if (!submitted)
-    {
-        struct io_uring_sqe *sqe = bs->get_sqe();
-        if (!sqe)
+    resume_1:
+        if (submitted)
         {
-            throw std::runtime_error("io_uring is full while trying to read metadata");
+            wait_state = 1;
+            return 1;
         }
-        struct ring_data_t *data = ((ring_data_t*)sqe->user_data);
-        data->iov = {
-            metadata_buffer + (prev == 1 ? bs->metadata_buf_size : 0),
-            bs->meta_len - metadata_read > bs->metadata_buf_size ? bs->metadata_buf_size : bs->meta_len - metadata_read,
-        };
-        data->callback = [this](ring_data_t *data) { handle_event(data); };
-        my_uring_prep_readv(sqe, bs->meta_fd, &data->iov, 1, bs->meta_offset + metadata_read);
-        bs->ringloop->submit();
-        submitted = (prev == 1 ? 2 : 1);
-        prev = submitted;
-    }
-    if (prev_done)
-    {
-        int count = 512 / sizeof(clean_disk_entry);
-        for (int sector = 0; sector < done_len; sector += 512)
+        if (metadata_read < bs->meta_len)
         {
-            clean_disk_entry *entries = (clean_disk_entry*)(metadata_buffer + (prev_done == 1 ? bs->metadata_buf_size : 0) + sector);
-            // handle <count> entries
-            handle_entries(entries, count, bs->block_order);
-            done_cnt += count;
+            sqe = bs->get_sqe();
+            if (!sqe)
+            {
+                throw std::runtime_error("io_uring is full while trying to read metadata");
+            }
+            data = ((ring_data_t*)sqe->user_data);
+            data->iov = {
+                metadata_buffer + (prev == 1 ? bs->metadata_buf_size : 0),
+                bs->meta_len - metadata_read > bs->metadata_buf_size ? bs->metadata_buf_size : bs->meta_len - metadata_read,
+            };
+            data->callback = [this](ring_data_t *data) { handle_event(data); };
+            my_uring_prep_readv(sqe, bs->meta_fd, &data->iov, 1, bs->meta_offset + metadata_read);
+            bs->ringloop->submit();
+            submitted = (prev == 1 ? 2 : 1);
+            prev = submitted;
+        }
+        if (prev_done)
+        {
+            int count = 512 / sizeof(clean_disk_entry);
+            for (int sector = 0; sector < done_len; sector += 512)
+            {
+                clean_disk_entry *entries = (clean_disk_entry*)(metadata_buffer + (prev_done == 1 ? bs->metadata_buf_size : 0) + sector);
+                // handle <count> entries
+                handle_entries(entries, count, bs->block_order);
+                done_cnt += count;
+            }
+            prev_done = 0;
+            done_len = 0;
+        }
+        if (!submitted)
+        {
+            break;
         }
-        prev_done = 0;
-        done_len = 0;
-    }
-    if (metadata_read >= bs->meta_len)
-    {
-        // metadata read finished
-        free(metadata_buffer);
-        metadata_buffer = NULL;
-        return 0;
     }
-    return 1;
+    // metadata read finished
+    printf("Metadata entries loaded: %d\n", entries_loaded);
+    free(metadata_buffer);
+    metadata_buffer = NULL;
+    return 0;
 }
 
 void blockstore_init_meta::handle_entries(struct clean_disk_entry* entries, int count, int block_order)
@@ -83,6 +88,7 @@ void blockstore_init_meta::handle_entries(struct clean_disk_entry* entries, int
             auto clean_it = bs->clean_db.find(entries[i].oid);
             if (clean_it == end || clean_it->second.version < entries[i].version)
             {
+                entries_loaded++;
                 if (clean_it != end)
                 {
                     // free the previous block

blockstore_init.h

@@ -3,9 +3,13 @@
 class blockstore_init_meta
 {
     blockstore *bs;
+    int wait_state = 0, wait_count = 0;
     uint8_t *metadata_buffer = NULL;
     uint64_t metadata_read = 0;
     int prev = 0, prev_done = 0, done_len = 0, submitted = 0, done_cnt = 0;
+    int entries_loaded = 0;
+    struct io_uring_sqe *sqe;
+    struct ring_data_t *data;
     void handle_entries(struct clean_disk_entry* entries, int count, int block_order);
     void handle_event(ring_data_t *data);
 public: