Release v0.5.4

- Fix a rare hang, more or less reproducible with very slow drives
- Fix a hang with the no_same_sector_overwrites mode

blockstore_flush.cpp

@@ -223,6 +223,7 @@ bool journal_flusher_co::loop()
 resume_0:
     if (!flusher->flush_queue.size() || !flusher->dequeuing)
     {
+stop_flusher:
         if (flusher->trim_wanted > 0 && flusher->journal_trim_counter > 0)
         {
             // Attempt forced trim
@@ -327,9 +328,7 @@ resume_0:
 #ifdef BLOCKSTORE_DEBUG
                     printf("No older flushes, stopping\n");
 #endif
-                    flusher->dequeuing = false;
+                    goto stop_flusher;
-                    wait_state = 0;
-                    return true;
                 }
             }
         }

blockstore_impl.cpp

@@ -287,7 +287,7 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     else if (PRIV(op)->wait_for == WAIT_JOURNAL_BUFFER)
     {
         int next = ((journal.cur_sector + 1) % journal.sector_count);
-        if (journal.sector_info[next].usage_count > 0 ||
+        if (journal.sector_info[next].flush_count > 0 ||
             journal.sector_info[next].dirty)
         {
             // do not submit

blockstore_journal.cpp

@@ -6,7 +6,7 @@
 blockstore_journal_check_t::blockstore_journal_check_t(blockstore_impl_t *bs)
 {
     this->bs = bs;
-    sectors_required = 0;
+    sectors_to_write = 0;
     next_pos = bs->journal.next_free;
     next_sector = bs->journal.cur_sector;
     first_sector = -1;
@@ -20,23 +20,26 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
     int required = entries_required;
     while (1)
     {
-        int fits = bs->journal.no_same_sector_overwrites && bs->journal.sector_info[next_sector].written
+        int fits = bs->journal.no_same_sector_overwrites && next_pos == bs->journal.next_free && bs->journal.sector_info[next_sector].written
             ? 0
             : (bs->journal.block_size - next_in_pos) / size;
         if (fits > 0)
         {
+            if (fits > required)
+            {
+                fits = required;
+            }
             if (first_sector == -1)
             {
                 first_sector = next_sector;
             }
             required -= fits;
             next_in_pos += fits * size;
-            sectors_required++;
+            sectors_to_write++;
         }
         else if (bs->journal.sector_info[next_sector].dirty)
         {
-            // sectors_required is more like "sectors to write"
+            sectors_to_write++;
-            sectors_required++;
         }
         if (required <= 0)
         {
@@ -59,7 +62,7 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
                 " is too small for a batch of "+std::to_string(entries_required)+" entries of "+std::to_string(size)+" bytes"
             );
         }
-        if (bs->journal.sector_info[next_sector].usage_count > 0 ||
+        if (bs->journal.sector_info[next_sector].flush_count > 0 ||
             bs->journal.sector_info[next_sector].dirty)
         {
             // No memory buffer available. Wait for it.
@@ -71,17 +74,18 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
                     dirty++;
                     used++;
                 }
-                if (bs->journal.sector_info[i].usage_count > 0)
+                if (bs->journal.sector_info[i].flush_count > 0)
                 {
                     used++;
                 }
             }
             // In fact, it's even more rare than "ran out of journal space", so print a warning
             printf(
-                "Ran out of journal sector buffers: %d/%lu buffers used (%d dirty), next buffer (%ld) is %s and flushed %lu times\n",
+                "Ran out of journal sector buffers: %d/%lu buffers used (%d dirty), next buffer (%ld)"
+                " is %s and flushed %lu times. Consider increasing \'journal_sector_buffer_count\'\n",
                 used, bs->journal.sector_count, dirty, next_sector,
                 bs->journal.sector_info[next_sector].dirty ? "dirty" : "not dirty",
-                bs->journal.sector_info[next_sector].usage_count
+                bs->journal.sector_info[next_sector].flush_count
             );
             PRIV(op)->wait_for = WAIT_JOURNAL_BUFFER;
             return 0;
@@ -100,11 +104,8 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
     {
         // No space in the journal. Wait until used_start changes.
         printf(
-            "Ran out of journal space (free space: %lu bytes, sectors to write: %d)\n",
+            "Ran out of journal space (used_start=%08lx, next_free=%08lx, dirty_start=%08lx)\n",
-            (bs->journal.next_free >= bs->journal.used_start
+            bs->journal.used_start, bs->journal.next_free, bs->journal.dirty_start
-                ? bs->journal.len-bs->journal.block_size - (bs->journal.next_free-bs->journal.used_start)
-                : bs->journal.used_start - bs->journal.next_free),
-            sectors_required
         );
         PRIV(op)->wait_for = WAIT_JOURNAL;
         bs->flusher->request_trim();
@@ -116,22 +117,21 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
 
 journal_entry* prefill_single_journal_entry(journal_t & journal, uint16_t type, uint32_t size)
 {
-    if (journal.block_size - journal.in_sector_pos < size ||
+    if (!journal.entry_fits(size))
-        journal.no_same_sector_overwrites && journal.sector_info[journal.cur_sector].written)
     {
         assert(!journal.sector_info[journal.cur_sector].dirty);
         // Move to the next journal sector
-        journal.sector_info[journal.cur_sector].written = false;
+        if (journal.sector_info[journal.cur_sector].flush_count > 0)
-        if (journal.sector_info[journal.cur_sector].usage_count > 0)
         {
             // Also select next sector buffer in memory
             journal.cur_sector = ((journal.cur_sector + 1) % journal.sector_count);
-            assert(!journal.sector_info[journal.cur_sector].usage_count);
+            assert(!journal.sector_info[journal.cur_sector].flush_count);
         }
         else
         {
             journal.dirty_start = journal.next_free;
         }
+        journal.sector_info[journal.cur_sector].written = false;
         journal.sector_info[journal.cur_sector].offset = journal.next_free;
         journal.in_sector_pos = 0;
         journal.next_free = (journal.next_free+journal.block_size) < journal.len ? journal.next_free + journal.block_size : journal.block_size;
@@ -157,7 +157,7 @@ void prepare_journal_sector_write(journal_t & journal, int cur_sector, io_uring_
 {
     journal.sector_info[cur_sector].dirty = false;
     journal.sector_info[cur_sector].written = true;
-    journal.sector_info[cur_sector].usage_count++;
+    journal.sector_info[cur_sector].flush_count++;
     ring_data_t *data = ((ring_data_t*)sqe->user_data);
     data->iov = (struct iovec){
         (journal.inmemory

blockstore_journal.h

@@ -133,7 +133,7 @@ inline uint32_t je_crc32(journal_entry *je)
 struct journal_sector_info_t
 {
     uint64_t offset;
-    uint64_t usage_count;
+    uint64_t flush_count;
     bool written;
     bool dirty;
 };
@@ -170,13 +170,18 @@ struct journal_t
     ~journal_t();
     bool trim();
     uint64_t get_trim_pos();
+    inline bool entry_fits(int size)
+    {
+        return !(block_size - in_sector_pos < size ||
+            no_same_sector_overwrites && sector_info[cur_sector].written);
+    }
 };
 
 struct blockstore_journal_check_t
 {
     blockstore_impl_t *bs;
     uint64_t next_pos, next_sector, next_in_pos;
-    int sectors_required, first_sector;
+    int sectors_to_write, first_sector;
     bool right_dir; // writing to the end or the beginning of the ring buffer
 
     blockstore_journal_check_t(blockstore_impl_t *bs);

blockstore_rollback.cpp

@@ -75,44 +75,35 @@ skip_ov:
         return 0;
     }
     // There is sufficient space. Get SQEs
-    struct io_uring_sqe *sqe[space_check.sectors_required];
+    struct io_uring_sqe *sqe[space_check.sectors_to_write];
-    for (i = 0; i < space_check.sectors_required; i++)
+    for (i = 0; i < space_check.sectors_to_write; i++)
     {
         BS_SUBMIT_GET_SQE_DECL(sqe[i]);
     }
     // Prepare and submit journal entries
     auto cb = [this, op](ring_data_t *data) { handle_rollback_event(data, op); };
     int s = 0, cur_sector = -1;
-    if ((journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_rollback) &&
-        journal.sector_info[journal.cur_sector].dirty)
-    {
-        if (cur_sector == -1)
-            PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-        cur_sector = journal.cur_sector;
-        prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
-    }
     for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
     {
+        if (!journal.entry_fits(sizeof(journal_entry_rollback)) &&
+            journal.sector_info[journal.cur_sector].dirty)
+        {
+            if (cur_sector == -1)
+                PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
+            prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
+            cur_sector = journal.cur_sector;
+        }
         journal_entry_rollback *je = (journal_entry_rollback*)
             prefill_single_journal_entry(journal, JE_ROLLBACK, sizeof(journal_entry_rollback));
-        journal.sector_info[journal.cur_sector].dirty = false;
         je->oid = v->oid;
         je->version = v->version;
         je->crc32 = je_crc32((journal_entry*)je);
         journal.crc32_last = je->crc32;
-        if (cur_sector != journal.cur_sector)
-        {
-            // Write previous sector. We should write the sector only after filling it,
-            // because otherwise we'll write a lot more sectors in the "no_same_sector_overwrite" mode
-            if (cur_sector != -1)
-                prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
-            else
-                PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-            cur_sector = journal.cur_sector;
-        }
     }
-    if (cur_sector != -1)
+    prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
-        prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
+    assert(s == space_check.sectors_to_write);
+    if (cur_sector == -1)
+        PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
     PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
     PRIV(op)->pending_ops = s;
     PRIV(op)->op_state = 1;

blockstore_stable.cpp

@@ -98,45 +98,36 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
         return 0;
     }
     // There is sufficient space. Get SQEs
-    struct io_uring_sqe *sqe[space_check.sectors_required];
+    struct io_uring_sqe *sqe[space_check.sectors_to_write];
-    for (i = 0; i < space_check.sectors_required; i++)
+    for (i = 0; i < space_check.sectors_to_write; i++)
     {
         BS_SUBMIT_GET_SQE_DECL(sqe[i]);
     }
     // Prepare and submit journal entries
     auto cb = [this, op](ring_data_t *data) { handle_stable_event(data, op); };
     int s = 0, cur_sector = -1;
-    if ((journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_stable) &&
-        journal.sector_info[journal.cur_sector].dirty)
-    {
-        if (cur_sector == -1)
-            PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-        cur_sector = journal.cur_sector;
-        prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
-    }
     for (i = 0, v = (obj_ver_id*)op->buf; i < op->len; i++, v++)
     {
         // FIXME: Only stabilize versions that aren't stable yet
+        if (!journal.entry_fits(sizeof(journal_entry_stable)) &&
+            journal.sector_info[journal.cur_sector].dirty)
+        {
+            if (cur_sector == -1)
+                PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
+            prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
+            cur_sector = journal.cur_sector;
+        }
         journal_entry_stable *je = (journal_entry_stable*)
             prefill_single_journal_entry(journal, JE_STABLE, sizeof(journal_entry_stable));
-        journal.sector_info[journal.cur_sector].dirty = false;
         je->oid = v->oid;
         je->version = v->version;
         je->crc32 = je_crc32((journal_entry*)je);
         journal.crc32_last = je->crc32;
-        if (cur_sector != journal.cur_sector)
-        {
-            // Write previous sector. We should write the sector only after filling it,
-            // because otherwise we'll write a lot more sectors in the "no_same_sector_overwrite" mode
-            if (cur_sector != -1)
-                prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
-            else
-                PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-            cur_sector = journal.cur_sector;
-        }
     }
-    if (cur_sector != -1)
+    prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
-        prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
+    assert(s == space_check.sectors_to_write);
+    if (cur_sector == -1)
+        PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
     PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
     PRIV(op)->pending_ops = s;
     PRIV(op)->op_state = 1;

blockstore_sync.cpp

@@ -112,30 +112,29 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
             return 0;
         }
         // Get SQEs. Don't bother about merging, submit each journal sector as a separate request
-        struct io_uring_sqe *sqe[space_check.sectors_required];
+        struct io_uring_sqe *sqe[space_check.sectors_to_write];
-        for (int i = 0; i < space_check.sectors_required; i++)
+        for (int i = 0; i < space_check.sectors_to_write; i++)
         {
             BS_SUBMIT_GET_SQE_DECL(sqe[i]);
         }
         // Prepare and submit journal entries
         auto it = PRIV(op)->sync_big_writes.begin();
         int s = 0, cur_sector = -1;
-        if ((journal_block_size - journal.in_sector_pos) < sizeof(journal_entry_big_write) &&
-            journal.sector_info[journal.cur_sector].dirty)
-        {
-            if (cur_sector == -1)
-                PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-            cur_sector = journal.cur_sector;
-            prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
-        }
         while (it != PRIV(op)->sync_big_writes.end())
         {
+            if (!journal.entry_fits(sizeof(journal_entry_big_write)) &&
+                journal.sector_info[journal.cur_sector].dirty)
+            {
+                if (cur_sector == -1)
+                    PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
+                prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
+                cur_sector = journal.cur_sector;
+            }
             journal_entry_big_write *je = (journal_entry_big_write*)prefill_single_journal_entry(
                 journal, (dirty_db[*it].state & BS_ST_INSTANT) ? JE_BIG_WRITE_INSTANT : JE_BIG_WRITE,
                 sizeof(journal_entry_big_write)
             );
             dirty_db[*it].journal_sector = journal.sector_info[journal.cur_sector].offset;
-            journal.sector_info[journal.cur_sector].dirty = false;
             journal.used_sectors[journal.sector_info[journal.cur_sector].offset]++;
 #ifdef BLOCKSTORE_DEBUG
             printf(
@@ -152,19 +151,11 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
             je->crc32 = je_crc32((journal_entry*)je);
             journal.crc32_last = je->crc32;
             it++;
-            if (cur_sector != journal.cur_sector)
-            {
-                // Write previous sector. We should write the sector only after filling it,
-                // because otherwise we'll write a lot more sectors in the "no_same_sector_overwrite" mode
-                if (cur_sector != -1)
-                    prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
-                else
-                    PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
-                cur_sector = journal.cur_sector;
-            }
         }
-        if (cur_sector != -1)
+        prepare_journal_sector_write(journal, journal.cur_sector, sqe[s++], cb);
-            prepare_journal_sector_write(journal, cur_sector, sqe[s++], cb);
+        assert(s == space_check.sectors_to_write);
+        if (cur_sector == -1)
+            PRIV(op)->min_flushed_journal_sector = 1 + journal.cur_sector;
         PRIV(op)->max_flushed_journal_sector = 1 + journal.cur_sector;
         PRIV(op)->pending_ops = s;
         PRIV(op)->op_state = SYNC_JOURNAL_WRITE_SENT;

blockstore_write.cpp

@@ -377,7 +377,6 @@ resume_2:
         sizeof(journal_entry_big_write)
     );
     dirty_it->second.journal_sector = journal.sector_info[journal.cur_sector].offset;
-    journal.sector_info[journal.cur_sector].dirty = false;
     journal.used_sectors[journal.sector_info[journal.cur_sector].offset]++;
 #ifdef BLOCKSTORE_DEBUG
     printf(
@@ -469,8 +468,8 @@ void blockstore_impl_t::release_journal_sectors(blockstore_op_t *op)
         uint64_t s = PRIV(op)->min_flushed_journal_sector;
         while (1)
         {
-            journal.sector_info[s-1].usage_count--;
+            journal.sector_info[s-1].flush_count--;
-            if (s != (1+journal.cur_sector) && journal.sector_info[s-1].usage_count == 0)
+            if (s != (1+journal.cur_sector) && journal.sector_info[s-1].flush_count == 0)
             {
                 // We know for sure that we won't write into this sector anymore
                 uint64_t new_ds = journal.sector_info[s-1].offset + journal.block_size;

cluster_client.cpp

@@ -473,7 +473,7 @@ void cluster_client_t::slice_rw(cluster_op_t *op)
     // Primary OSDs still operate individual stripes, but their size is multiplied by PG minsize in case of EC
     auto & pool_cfg = st_cli.pool_config[INODE_POOL(op->inode)];
     uint64_t pg_block_size = bs_block_size * (
-        pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_minsize
+        pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pool_cfg.pg_size-pool_cfg.parity_chunks
     );
     uint64_t first_stripe = (op->offset / pg_block_size) * pg_block_size;
     uint64_t last_stripe = ((op->offset + op->len + pg_block_size - 1) / pg_block_size - 1) * pg_block_size;

debian/changelog

@@ -1,3 +1,9 @@
+vitastor (0.5.4-1) unstable; urgency=medium
+
+  * Bugfixes
+
+ -- Vitaliy Filippov <vitalif@yourcmc.ru>  Tue, 02 Feb 2021 23:01:24 +0300
+
 vitastor (0.5.1-1) unstable; urgency=medium
 
   * Add jerasure support

debian/vitastor.Dockerfile

@@ -45,10 +45,10 @@ RUN set -e -x; \
     mkdir -p /root/build/vitastor-$REL; \
     rm -rf /root/build/vitastor-$REL/*; \
     cd /root/build/vitastor-$REL; \
-    cp -r /root/vitastor vitastor-0.5.1; \
+    cp -r /root/vitastor vitastor-0.5.4; \
-    ln -s /root/build/qemu-$REL/qemu-*/ vitastor-0.5.1/qemu; \
+    ln -s /root/build/qemu-$REL/qemu-*/ vitastor-0.5.4/qemu; \
-    ln -s /root/fio-build/fio-*/ vitastor-0.5.1/fio; \
+    ln -s /root/fio-build/fio-*/ vitastor-0.5.4/fio; \
-    cd vitastor-0.5.1; \
+    cd vitastor-0.5.4; \
     FIO=$(head -n1 fio/debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     QEMU=$(head -n1 qemu/debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     sh copy-qemu-includes.sh; \
@@ -64,8 +64,8 @@ RUN set -e -x; \
     echo "dep:fio=$FIO" > debian/substvars; \
     echo "dep:qemu=$QEMU" >> debian/substvars; \
     cd /root/build/vitastor-$REL; \
-    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_0.5.1.orig.tar.xz vitastor-0.5.1; \
+    tar --sort=name --mtime='2020-01-01' --owner=0 --group=0 --exclude=debian -cJf vitastor_0.5.4.orig.tar.xz vitastor-0.5.4; \
-    cd vitastor-0.5.1; \
+    cd vitastor-0.5.4; \
     V=$(head -n1 debian/changelog | perl -pe 's/^.*\((.*?)\).*$/$1/'); \
     DEBFULLNAME="Vitaliy Filippov <vitalif@yourcmc.ru>" dch -D $REL -v "$V""$REL" "Rebuild for $REL"; \
     DEB_BUILD_OPTIONS=nocheck dpkg-buildpackage --jobs=auto -sa; \

fio_cluster.cpp

@@ -93,7 +93,7 @@ static struct fio_option options[] = {
     {
         .name   = "cluster_log_level",
         .lname  = "cluster log level",
-        .type   = FIO_OPT_BOOL,
+        .type   = FIO_OPT_INT,
         .off1   = offsetof(struct sec_options, cluster_log),
         .help   = "Set log level for the Vitastor client",
         .def    = "0",
@@ -145,9 +145,7 @@ static void sec_cleanup(struct thread_data *td)
         delete bsd->cli;
         delete bsd->epmgr;
         delete bsd->ringloop;
-        bsd->cli = NULL;
+        delete bsd;
-        bsd->epmgr = NULL;
-        bsd->ringloop = NULL;
     }
 }
 

fio_sec_osd.cpp

@@ -140,6 +140,7 @@ static void sec_cleanup(struct thread_data *td)
     if (bsd)
     {
         close(bsd->connect_fd);
+        delete bsd;
     }
 }
 
@@ -312,6 +313,7 @@ static int sec_getevents(struct thread_data *td, unsigned int min, unsigned int
             exit(1);
         }
         io_u* io = it->second;
+        bsd->queue.erase(it);
         if (io->ddir == DDIR_READ)
         {
             if (reply.hdr.retval != io->xfer_buflen)

messenger.cpp

@@ -30,7 +30,7 @@ osd_messenger_t::~osd_messenger_t()
 {
     while (clients.size() > 0)
     {
-        stop_client(clients.begin()->first);
+        stop_client(clients.begin()->first, true);
     }
 }
 
@@ -111,7 +111,7 @@ void osd_messenger_t::try_connect_peer_addr(osd_num_t peer_osd, const char *peer
         timeout_id = tfd->set_timer(1000*peer_connect_timeout, false, [this, peer_fd](int timer_id)
         {
             osd_num_t peer_osd = clients.at(peer_fd)->osd_num;
-            stop_client(peer_fd);
+            stop_client(peer_fd, true);
             on_connect_peer(peer_osd, -EIO);
             return;
         });
@@ -149,7 +149,7 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
     }
     if (result != 0)
     {
-        stop_client(peer_fd);
+        stop_client(peer_fd, true);
         on_connect_peer(peer_osd, -result);
         return;
     }
@@ -171,7 +171,7 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
     {
         // Stop client
         printf("[OSD %lu] client %d disconnected\n", this->osd_num, peer_fd);
-        stop_client(peer_fd);
+        stop_client(peer_fd, true);
     }
     else if (epoll_events & EPOLLIN)
     {
@@ -309,7 +309,7 @@ void osd_messenger_t::cancel_op(osd_op_t *op)
     }
 }
 
-void osd_messenger_t::stop_client(int peer_fd)
+void osd_messenger_t::stop_client(int peer_fd, bool force)
 {
     assert(peer_fd != 0);
     auto it = clients.find(peer_fd);
@@ -334,6 +334,10 @@ void osd_messenger_t::stop_client(int peer_fd)
                 printf("[OSD %lu] Stopping client %d (regular client)\n", osd_num, peer_fd);
         }
     }
+    else if (!force)
+    {
+        return;
+    }
     cl->peer_state = PEER_STOPPED;
     clients.erase(it);
     tfd->set_fd_handler(peer_fd, false, NULL);

messenger.h

@@ -275,7 +275,7 @@ struct osd_messenger_t
 
 public:
     void connect_peer(uint64_t osd_num, json11::Json peer_state);
-    void stop_client(int peer_fd);
+    void stop_client(int peer_fd, bool force = false);
     void outbox_push(osd_op_t *cur_op);
     std::function<void(osd_op_t*)> exec_op;
     std::function<void(osd_num_t)> repeer_pgs;

mon/mon.js

@@ -253,7 +253,10 @@ class Mon
         const res = await this.etcd_call('/kv/txn', { success: [
             { requestRange: { key: b64(this.etcd_prefix+'/config/global') } }
         ] }, this.etcd_start_timeout, -1);
-        this.parse_kv(res.responses[0].response_range.kvs[0]);
+        if (res.responses[0].response_range.kvs)
+        {
+            this.parse_kv(res.responses[0].response_range.kvs[0]);
+        }
         this.check_config();
     }
 

mon/simple-offsets.js

@@ -4,6 +4,7 @@
 // Simple tool to calculate journal and metadata offsets for a single device
 // Will be replaced by smarter tools in the future
 
+const fs = require('fs').promises;
 const child_process = require('child_process');
 
 async function run()
@@ -15,6 +16,7 @@ async function run()
         device_block_size: 4096,
         journal_offset: 0,
         device_size: 0,
+        format: 'text',
     };
     for (let i = 2; i < process.argv.length; i++)
     {
@@ -24,7 +26,22 @@ async function run()
             i++;
         }
     }
-    const device_size = Number(options.device_size || await system("blockdev --getsize64 "+options.device));
+    if (!options.device)
+    {
+        process.stderr.write('USAGE: nodejs '+process.argv[1]+' --device /dev/sdXXX\n');
+        process.exit(1);
+    }
+    options.device_size = Number(options.device_size);
+    let device_size = options.device_size;
+    if (!device_size)
+    {
+        const st = await fs.stat(options.device);
+        options.device_block_size = st.blksize;
+        if (st.isBlockDevice())
+            device_size = Number(await system("/sbin/blockdev --getsize64 "+options.device))
+        else
+            device_size = st.size;
+    }
     if (!device_size)
     {
         process.stderr.write('Failed to get device size\n');
@@ -32,25 +49,45 @@ async function run()
     }
     options.journal_offset = Math.ceil(options.journal_offset/options.device_block_size)*options.device_block_size;
     const meta_offset = options.journal_offset + Math.ceil(options.journal_size/options.device_block_size)*options.device_block_size;
-    const entries_per_block = Math.floor(options.device_block_size / (24 + options.object_size/options.bitmap_granularity/8));
+    const entries_per_block = Math.floor(options.device_block_size / (24 + 2*options.object_size/options.bitmap_granularity/8));
     const object_count = Math.floor((device_size-meta_offset)/options.object_size);
     const meta_size = Math.ceil(object_count / entries_per_block) * options.device_block_size;
     const data_offset = meta_offset + meta_size;
     const meta_size_fmt = (meta_size > 1024*1024*1024 ? Math.round(meta_size/1024/1024/1024*100)/100+" GB"
         : Math.round(meta_size/1024/1024*100)/100+" MB");
-    process.stdout.write(
+    if (options.format == 'text' || options.format == 'options')
-        `Metadata size: ${meta_size_fmt}\n`+
+    {
-        `Options for the OSD:\n`+
+        if (options.format == 'text')
-        `    --journal_offset ${options.journal_offset}\n`+
+        {
-        `    --meta_offset ${meta_offset}\n`+
+            process.stderr.write(
-        `    --data_offset ${data_offset}\n`+
+                `Metadata size: ${meta_size_fmt}\n`+
-        (options.device_size ? `    --data_size ${device_size-data_offset}\n` : '')
+                `Options for the OSD:\n`
-    );
+            );
+        }
+        process.stdout.write(
+            `    --data_device ${options.device}\n`+
+            `    --journal_offset ${options.journal_offset}\n`+
+            `    --meta_offset ${meta_offset}\n`+
+            `    --data_offset ${data_offset}\n`+
+            (options.device_size ? `    --data_size ${device_size-data_offset}\n` : '')
+        );
+    }
+    else if (options.format == 'env')
+    {
+        process.stdout.write(
+            `journal_offset=${options.journal_offset}\n`+
+            `meta_offset=${meta_offset}\n`+
+            `data_offset=${data_offset}\n`+
+            `data_size=${device_size-data_offset}\n`
+        );
+    }
+    else
+        process.stdout.write('Unknown format: '+options.format);
 }
 
 function system(cmd)
 {
-    return new Promise((ok, no) => child_process.exec(cmd, { maxBuffer: 64*1024*1024 }, (err, stdout, stderr) => (err ? no(err) : ok(stdout))));
+    return new Promise((ok, no) => child_process.exec(cmd, { maxBuffer: 64*1024*1024 }, (err, stdout, stderr) => (err ? no(err.message) : ok(stdout))));
 }
 
-run().catch(console.error);
+run().catch(err => { console.error(err); process.exit(1); });

msgr_receive.cpp

@@ -9,6 +9,10 @@ void osd_messenger_t::read_requests()
     {
         int peer_fd = read_ready_clients[i];
         osd_client_t *cl = clients[peer_fd];
+        if (cl->read_msg.msg_iovlen)
+        {
+            continue;
+        }
         if (cl->read_remaining < receive_buffer_size)
         {
             cl->read_iov.iov_base = cl->in_buf;
@@ -29,6 +33,7 @@ void osd_messenger_t::read_requests()
             io_uring_sqe* sqe = ringloop->get_sqe();
             if (!sqe)
             {
+                cl->read_msg.msg_iovlen = 0;
                 read_ready_clients.erase(read_ready_clients.begin(), read_ready_clients.begin() + i);
                 return;
             }
@@ -52,6 +57,7 @@ void osd_messenger_t::read_requests()
 bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
 {
     bool ret = false;
+    cl->read_msg.msg_iovlen = 0;
     cl->refs--;
     if (cl->peer_state == PEER_STOPPED)
     {
@@ -160,8 +166,14 @@ bool osd_messenger_t::handle_finished_read(osd_client_t *cl)
     {
         if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC)
             return handle_reply_hdr(cl);
-        else
+        else if (cl->read_op->req.hdr.magic == SECONDARY_OSD_OP_MAGIC)
             handle_op_hdr(cl);
+        else
+        {
+            printf("Received garbage: magic=%lx id=%lu opcode=%lx from %d\n", cl->read_op->req.hdr.magic, cl->read_op->req.hdr.id, cl->read_op->req.hdr.opcode, cl->peer_fd);
+            stop_client(cl->peer_fd);
+            return false;
+        }
     }
     else if (cl->read_state == CL_READ_DATA)
     {

msgr_send.cpp

@@ -46,7 +46,8 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
         to_send_list.push_back((iovec){ .iov_base = cur_op->req.buf, .iov_len = OSD_PACKET_SIZE });
         cl->sent_ops[cur_op->req.hdr.id] = cur_op;
     }
-    // Pre-defined send_lists
+    to_outbox.push_back(NULL);
+    // Operation data
     if ((cur_op->op_type == OSD_OP_IN
         ? (cur_op->req.hdr.opcode == OSD_OP_READ ||
         cur_op->req.hdr.opcode == OSD_OP_SEC_READ ||
@@ -58,17 +59,17 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
         cur_op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
         cur_op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK)) && cur_op->iov.count > 0)
     {
-        to_outbox.push_back(NULL);
         for (int i = 0; i < cur_op->iov.count; i++)
         {
             assert(cur_op->iov.buf[i].iov_base);
             to_send_list.push_back(cur_op->iov.buf[i]);
-            to_outbox.push_back(i == cur_op->iov.count-1 ? cur_op : NULL);
+            to_outbox.push_back(NULL);
         }
     }
-    else
+    if (cur_op->op_type == OSD_OP_IN)
     {
-        to_outbox.push_back(cur_op);
+        // To free it later
+        to_outbox[to_outbox.size()-1] = cur_op;
     }
     if (!ringloop)
     {
@@ -92,6 +93,10 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
 void osd_messenger_t::measure_exec(osd_op_t *cur_op)
 {
     // Measure execution latency
+    if (cur_op->req.hdr.opcode > OSD_OP_MAX)
+    {
+        return;
+    }
     timespec tv_end;
     clock_gettime(CLOCK_REALTIME, &tv_end);
     stats.op_stat_count[cur_op->req.hdr.opcode]++;
@@ -198,11 +203,8 @@ void osd_messenger_t::handle_send(int result, osd_client_t *cl)
             {
                 if (cl->outbox[done])
                 {
-                    // Operation fully sent
+                    // Reply fully sent
-                    if (cl->outbox[done]->op_type == OSD_OP_IN)
+                    delete cl->outbox[done];
-                    {
-                        delete cl->outbox[done];
-                    }
                 }
                 result -= iov.iov_len;
                 done++;

osd_cluster.cpp

@@ -384,6 +384,7 @@ void osd_t::create_osd_state()
         {
             st_cli.load_pgs();
         }
+        report_statistics();
     });
 }
 

osd_primary.cpp

@@ -489,7 +489,11 @@ resume_7:
         }
         // Remember PG as dirty to drop the connection when PG goes offline
         // (this is required because of the "lazy sync")
-        c_cli.clients[cur_op->peer_fd]->dirty_pgs.insert({ .pool_id = pg.pool_id, .pg_num = pg.pg_num });
+        auto cl_it = c_cli.clients.find(cur_op->peer_fd);
+        if (cl_it != c_cli.clients.end())
+        {
+            cl_it->second->dirty_pgs.insert({ .pool_id = pg.pool_id, .pg_num = pg.pg_num });
+        }
         dirty_pgs.insert({ .pool_id = pg.pool_id, .pg_num = pg.pg_num });
     }
     return true;

osd_primary_subops.cpp

@@ -295,7 +295,7 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
             uint64_t version = subop->reply.sec_rw.version;
 #ifdef OSD_DEBUG
             uint64_t peer_osd = c_cli.clients.find(subop->peer_fd) != c_cli.clients.end()
-                ? c_cli.clients[subop->peer_fd].osd_num : osd_num;
+                ? c_cli.clients[subop->peer_fd]->osd_num : osd_num;
             printf("subop %lu from osd %lu: version = %lu\n", opcode, peer_osd, version);
 #endif
             if (op_data->fact_ver != 0 && op_data->fact_ver != version)

ringloop.cpp

@@ -77,7 +77,10 @@ void ring_loop_t::loop()
             dl.callback(&dl);
         }
         else
+        {
+            printf("Warning: empty callback in SQE\n");
             free_ring_data[free_ring_data_ptr++] = d - ring_datas;
+        }
         io_uring_cqe_seen(&ring, cqe);
     }
     while (get_sqe_queue.size() > 0)

ringloop.h

@@ -142,7 +142,10 @@ public:
             return NULL;
         struct io_uring_sqe* sqe = io_uring_get_sqe(&ring);
         if (sqe)
+        {
+            *sqe = { 0 };
             io_uring_sqe_set_data(sqe, ring_datas + free_ring_data[--free_ring_data_ptr]);
+        }
         return sqe;
     }
     inline int wait_sqe(std::function<void()> cb)

rpm/build-tarball.sh

@@ -48,4 +48,4 @@ FIO=`rpm -qi fio | perl -e 'while(<>) { /^Epoch[\s:]+(\S+)/ && print "$1:"; /^Ve
 QEMU=`rpm -qi qemu qemu-kvm | perl -e 'while(<>) { /^Epoch[\s:]+(\S+)/ && print "$1:"; /^Version[\s:]+(\S+)/ && print $1; /^Release[\s:]+(\S+)/ && print "-$1"; }'`
 perl -i -pe 's/(Requires:\s*fio)([^\n]+)?/$1 = '$FIO'/' $VITASTOR/rpm/vitastor-el$EL.spec
 perl -i -pe 's/(Requires:\s*qemu(?:-kvm)?)([^\n]+)?/$1 = '$QEMU'/' $VITASTOR/rpm/vitastor-el$EL.spec
-tar --transform 's#^#vitastor-0.5.1/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.5.1$(rpm --eval '%dist').tar.gz *
+tar --transform 's#^#vitastor-0.5.4/#' --exclude 'rpm/*.rpm' -czf $VITASTOR/../vitastor-0.5.4$(rpm --eval '%dist').tar.gz *

rpm/vitastor-el7.Dockerfile

@@ -37,7 +37,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-0.5.1.el7.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.5.4.el7.tar.gz ~/rpmbuild/SOURCES; \
     cp vitastor-el7.spec ~/rpmbuild/SPECS/vitastor.spec; \
     cd ~/rpmbuild/SPECS/; \
     rpmbuild -ba vitastor.spec; \

rpm/vitastor-el7.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        0.5.1
+Version:        0.5.4
 Release:        2%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.0
 URL:            https://vitastor.io/
-Source0:        vitastor-0.5.1.el7.tar.gz
+Source0:        vitastor-0.5.4.el7.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

rpm/vitastor-el8.Dockerfile

@@ -35,7 +35,7 @@ ADD . /root/vitastor
 RUN set -e; \
     cd /root/vitastor/rpm; \
     sh build-tarball.sh; \
-    cp /root/vitastor-0.5.1.el8.tar.gz ~/rpmbuild/SOURCES; \
+    cp /root/vitastor-0.5.4.el8.tar.gz ~/rpmbuild/SOURCES; \
     cp vitastor-el8.spec ~/rpmbuild/SPECS/vitastor.spec; \
     cd ~/rpmbuild/SPECS/; \
     rpmbuild -ba vitastor.spec; \

rpm/vitastor-el8.spec

@@ -1,11 +1,11 @@
 Name:           vitastor
-Version:        0.5.1
+Version:        0.5.4
 Release:        2%{?dist}
 Summary:        Vitastor, a fast software-defined clustered block storage
 
 License:        Vitastor Network Public License 1.0
 URL:            https://vitastor.io/
-Source0:        vitastor-0.5.1.el8.tar.gz
+Source0:        vitastor-0.5.4.el8.tar.gz
 
 BuildRequires:  liburing-devel >= 0.6
 BuildRequires:  gperftools-devel

run_tests.sh

@@ -0,0 +1,77 @@
+#!/bin/bash
+
+if [ ! "$BASH_VERSION" ] ; then
+    echo "Use bash to run this script ($0)" 1>&2
+    exit 1
+fi
+
+format_error()
+{
+    echo $(echo -n -e "\033[1;31m")$1$(echo -n -e "\033[m")
+    $ETCDCTL get --prefix /vitastor > ./testdata/etcd-dump.txt
+    exit 1
+}
+format_green()
+{
+    echo $(echo -n -e "\033[1;32m")$1$(echo -n -e "\033[m")
+}
+
+set -e -x
+
+trap 'kill -9 $(jobs -p)' EXIT
+
+ETCD=${ETCD:-etcd}
+ETCD_PORT=${ETCD_PORT:-12379}
+
+rm -rf ./testdata
+mkdir -p ./testdata
+dd if=/dev/zero of=./testdata/test_osd1.bin bs=1024 count=1 seek=$((1024*1024-1))
+dd if=/dev/zero of=./testdata/test_osd2.bin bs=1024 count=1 seek=$((1024*1024-1))
+dd if=/dev/zero of=./testdata/test_osd3.bin bs=1024 count=1 seek=$((1024*1024-1))
+
+$ETCD -name etcd_test --data-dir ./testdata/etcd \
+    --advertise-client-urls http://127.0.0.1:$ETCD_PORT --listen-client-urls http://127.0.0.1:$ETCD_PORT \
+    --initial-advertise-peer-urls http://127.0.0.1:$((ETCD_PORT+1)) --listen-peer-urls http://127.0.0.1:$((ETCD_PORT+1)) \
+    --max-txn-ops=100000 --auto-compaction-retention=10 --auto-compaction-mode=revision &>./testdata/etcd.log &
+ETCD_PID=$!
+ETCD_URL=127.0.0.1:$ETCD_PORT/v3
+ETCDCTL="${ETCD}ctl --endpoints=http://$ETCD_URL"
+
+./osd --osd_num 1 --bind_address 127.0.0.1 --etcd_address $ETCD_URL $(node mon/simple-offsets.js --format options --device ./testdata/test_osd1.bin 2>/dev/null) &>./testdata/osd1.log &
+OSD1_PID=$!
+./osd --osd_num 2 --bind_address 127.0.0.1 --etcd_address $ETCD_URL $(node mon/simple-offsets.js --format options --device ./testdata/test_osd2.bin 2>/dev/null) &>./testdata/osd2.log &
+OSD2_PID=$!
+./osd --osd_num 3 --bind_address 127.0.0.1 --etcd_address $ETCD_URL $(node mon/simple-offsets.js --format options --device ./testdata/test_osd3.bin 2>/dev/null) &>./testdata/osd3.log &
+OSD3_PID=$!
+
+cd mon
+npm install
+cd ..
+node mon/mon-main.js --etcd_url http://$ETCD_URL --etcd_prefix "/vitastor" &>./testdata/mon.log &
+MON_PID=$!
+
+$ETCDCTL put /vitastor/config/pools '{"1":{"name":"testpool","scheme":"xor","pg_size":3,"pg_minsize":2,"parity_chunks":1,"pg_count":1,"failure_domain":"osd"}}'
+
+sleep 2
+
+if ! ($ETCDCTL get /vitastor/config/pgs --print-value-only | jq -s -e '(. | length) != 0 and (.[0].items["1"]["1"].osd_set | sort) == ["1","2","3"]'); then
+    format_error "FAILED: 1 PG NOT CONFIGURED"
+fi
+
+if ! ($ETCDCTL get /vitastor/pg/state/1/1 --print-value-only | jq -s -e '(. | length) != 0 and .[0].state == ["active"]'); then
+    format_error "FAILED: 1 PG NOT UP"
+fi
+
+echo leak:fio >> testdata/lsan-suppress.txt
+echo leak:tcmalloc >> testdata/lsan-suppress.txt
+echo leak:ceph >> testdata/lsan-suppress.txt
+echo leak:librbd >> testdata/lsan-suppress.txt
+echo leak:_M_mutate >> testdata/lsan-suppress.txt
+echo leak:_M_assign >> testdata/lsan-suppress.txt
+#LSAN_OPTIONS=suppressions=`pwd`/testdata/lsan-suppress.txt LD_PRELOAD=libasan.so.5 \
+#    fio -thread -name=test -ioengine=./libfio_sec_osd.so -bs=4k -fsync=128 `$ETCDCTL get /vitastor/osd/state/1 --print-value-only | jq -r '"-host="+.addresses[0]+" -port="+(.port|tostring)'` -rw=write -size=32M
+
+LSAN_OPTIONS=suppressions=`pwd`/testdata/lsan-suppress.txt LD_PRELOAD=libasan.so.5 \
+    fio -thread -name=test -ioengine=./libfio_cluster.so -bs=4M -direct=1 -iodepth=1 -fsync=1 -rw=write -etcd=$ETCD_URL -pool=1 -inode=1 -size=1G -cluster_log_level=10
+
+format_green OK