Use EPOLLOUT in no-io_uring mode

.gitea/workflows/test.yml

@@ -324,6 +324,24 @@ jobs:
           echo ""
         done
 
+  test_no_uring:
+    runs-on: ubuntu-latest
+    needs: build
+    container: ${{env.TEST_IMAGE}}:${{github.sha}}
+    steps:
+    - name: Run test
+      id: test
+      timeout-minutes: 3
+      run: /root/vitastor/tests/test_no_uring.sh
+    - name: Print logs
+      if: always() && steps.test.outcome == 'failure'
+      run: |
+        for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do
+          echo "-------- $i --------"
+          cat $i
+          echo ""
+        done
+
   test_interrupted_rebalance:
     runs-on: ubuntu-latest
     needs: build

src/blockstore/blockstore.h

@@ -77,8 +77,11 @@ Output:
 - retval = number of bytes actually read/written or negative error number
   -EINVAL = invalid input parameters
   -ENOENT = requested object/version does not exist for reads
+  -EEXIST = requested version to write already exists
   -ENOSPC = no space left in the store for writes
-  -EDOM = checksum error.
+  -EAGAIN = temporary failure, operation should be retried (notably, writes may be retried
+    after stabilizing some journaled entries and freeing up journal space)
+  -EDOM = checksum error
 - version = the version actually read or written
 
 ## BS_OP_DELETE
@@ -90,7 +93,7 @@ Input:
 - version = requested version. Treated the same as with BS_OP_WRITE
 
 Output:
-- retval = 0 or negative error number (-EINVAL)
+- retval = 0 or negative error number (-EINVAL, -EAGAIN)
 - version = the version actually written (delete is initially written as an object version)
 
 ## BS_OP_SYNC
@@ -99,7 +102,7 @@ Make sure all previously issued modifications reach physical media.
 
 Input: Nothing except opcode
 Output:
-- retval = 0 or negative error number (-EINVAL)
+- retval = 0 or negative error number (-EINVAL, -EAGAIN)
 
 ## BS_OP_STABLE / BS_OP_ROLLBACK
 
@@ -129,7 +132,9 @@ Input:
 - max_oid = max inode/stripe or 0 to list all objects
 
 Output:
-- retval = total obj_ver_id count
+- retval = total obj_ver_id count or negative error number
+  -ENOMEM = could not allocate memory for the listing
+  -EAGAIN = pool sharding settings are incorrect, please reshard and retry
 - version = stable obj_ver_id count
 - buf = obj_ver_id array allocated by the blockstore. Stable versions come first.
   You must free it yourself after usage with free().

src/blockstore/blockstore_write.cpp

@@ -243,6 +243,12 @@ enospc:
         uint64_t loc = !op->len ? 0 : heap->find_free_buffer_area(op->len);
         if (loc == UINT64_MAX)
         {
+            if (!heap->get_to_compact_count())
+            {
+                op->retval = -EAGAIN;
+                FINISH_OP(op);
+                return 2;
+            }
             PRIV(op)->wait_for = WAIT_COMPACTION;
             PRIV(op)->wait_detail = heap->get_compacted_count();
             flusher->request_trim();

src/blockstore/v1/flush.cpp

@@ -76,6 +76,16 @@ size_t journal_flusher_t::get_queue_size()
     return flush_queue.size();
 }
 
+bool journal_flusher_t::may_advance()
+{
+    return flush_queue.size() || active_flushers > 0 || trim_possible;
+}
+
+bool journal_flusher_t::may_advance_except_one()
+{
+    return flush_queue.size() || active_flushers > 1 || trim_possible;
+}
+
 void journal_flusher_t::loop()
 {
     target_flusher_count = bs->write_iodepth*2;
@@ -190,6 +200,7 @@ void journal_flusher_t::request_trim()
 
 void journal_flusher_t::mark_trim_possible()
 {
+    trim_possible = true;
     if (trim_wanted > 0)
     {
         dequeuing = true;
@@ -652,9 +663,10 @@ resume_2:
         flusher->sync_to_repeat.erase(repeat_it);
     trim_journal:
         // Clear unused part of the journal every <journal_trim_interval> flushes
-        if (bs->journal_trim_interval && !((++flusher->journal_trim_counter) % bs->journal_trim_interval) ||
-            flusher->trim_wanted > 0)
+        if ((bs->journal_trim_interval && !((++flusher->journal_trim_counter) % bs->journal_trim_interval) ||
+            flusher->trim_wanted > 0) && flusher->trim_possible)
         {
+            flusher->trim_possible = false;
     resume_30:
     resume_31:
     resume_32:
@@ -1496,10 +1508,19 @@ bool journal_flusher_co::trim_journal(int wait_base)
                 printf("Journal flushed\n");
                 exit(0);
             }
+            bs->wakeup_wait_journal();
+        }
+        else if (!flusher->may_advance_except_one())
+        {
+            bs->wakeup_wait_journal();
         }
         flusher->journal_trim_counter = 0;
         flusher->trimming = false;
     }
+    else if (!flusher->may_advance_except_one())
+    {
+        bs->wakeup_wait_journal();
+    }
     return true;
 }
 

src/blockstore/v1/flush.h

@@ -94,6 +94,7 @@ public:
 class journal_flusher_t
 {
     int trim_wanted = 0;
+    bool trim_possible = false;
     bool dequeuing;
     int min_flusher_count, max_flusher_count, cur_flusher_count, target_flusher_count;
     int flusher_start_threshold;
@@ -117,12 +118,14 @@ class journal_flusher_t
 
     bool try_find_older(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
     bool try_find_other(std::map<obj_ver_id, dirty_entry>::iterator & dirty_end, obj_ver_id & cur);
+    bool may_advance_except_one();
 
 public:
     journal_flusher_t(blockstore_impl_t *bs);
     ~journal_flusher_t();
     void loop();
     bool is_trim_wanted() { return trim_wanted; }
+    bool may_advance();
     bool is_active();
     size_t get_queue_size();
     void mark_trim_possible();

src/blockstore/v1/impl.cpp

@@ -212,10 +212,6 @@ void blockstore_impl_t::loop()
                     // ring is full, stop submission
                     break;
                 }
-                else if (PRIV(op)->wait_for == WAIT_JOURNAL)
-                {
-                    PRIV(op)->wait_detail2 = (unstable_writes.size()+unstable_unsynced);
-                }
             }
         }
         if (op_idx != new_idx)
@@ -280,6 +276,12 @@ bool blockstore_impl_t::is_safe_to_stop()
     return true;
 }
 
+void blockstore_impl_t::wakeup_wait_journal()
+{
+    wait_journal_counter++;
+    ringloop->wakeup();
+}
+
 void blockstore_impl_t::check_wait(blockstore_op_t *op)
 {
     if (PRIV(op)->wait_for == WAIT_SQE)
@@ -296,12 +298,11 @@ void blockstore_impl_t::check_wait(blockstore_op_t *op)
     }
     else if (PRIV(op)->wait_for == WAIT_JOURNAL)
     {
-        if (journal.used_start == PRIV(op)->wait_detail &&
-            (unstable_writes.size()+unstable_unsynced) == PRIV(op)->wait_detail2)
+        if (wait_journal_counter == PRIV(op)->wait_detail)
         {
             // do not submit
 #ifdef BLOCKSTORE_DEBUG
-            printf("Still waiting to flush journal offset %08jx\n", PRIV(op)->wait_detail);
+            printf("Still waiting for flush journal counter %ju\n", PRIV(op)->wait_detail);
 #endif
             return;
         }

src/blockstore/v1/impl.h

@@ -93,7 +93,7 @@ struct blockstore_op_private_t
 {
     // Wait status
     int wait_for;
-    uint64_t wait_detail, wait_detail2;
+    uint64_t wait_detail;
     int pending_ops;
     int op_state;
 
@@ -215,6 +215,8 @@ class blockstore_impl_t: public blockstore_i
     blockstore_init_meta* metadata_init_reader;
     blockstore_init_journal* journal_init_reader;
 
+    uint64_t wait_journal_counter = 0;
+    void wakeup_wait_journal();
     void check_wait(blockstore_op_t *op);
     void init_op(blockstore_op_t *op);
 

src/blockstore/v1/init.cpp

@@ -1000,7 +1000,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         memcpy((uint8_t*)dyn+sizeof(int), dyn_from, dyn_size);
                     }
                     bs->dirty_db.emplace(ov, (dirty_entry){
-                        .state = (BS_ST_SMALL_WRITE | BS_ST_SYNCED),
+                        .state = (BS_ST_SMALL_WRITE | BS_ST_SYNCED | (je->type == JE_SMALL_WRITE_INSTANT ? BS_ST_INSTANT : 0)),
                         .flags = 0,
                         .location = location,
                         .offset = je->small_write.offset,
@@ -1084,7 +1084,7 @@ int blockstore_init_journal::handle_journal_part(void *buf, uint64_t done_pos, u
                         memcpy((uint8_t*)dyn+sizeof(int), dyn_from, dyn_size);
                     }
                     auto dirty_it = bs->dirty_db.emplace(ov, (dirty_entry){
-                        .state = (BS_ST_BIG_WRITE | BS_ST_SYNCED),
+                        .state = (BS_ST_BIG_WRITE | BS_ST_SYNCED | (je->type == JE_BIG_WRITE_INSTANT ? BS_ST_INSTANT : 0)),
                         .flags = 0,
                         .location = je->big_write.location,
                         .offset = je->big_write.offset,

src/blockstore/v1/journal.cpp

@@ -15,6 +15,7 @@ blockstore_journal_check_t::blockstore_journal_check_t(blockstore_impl_t *bs)
     first_sector = -1;
     next_in_pos = bs->journal.in_sector_pos;
     right_dir = next_pos >= bs->journal.used_start;
+    give_up = false;
 }
 
 // Check if we can write <required> entries of <size> bytes and <data_after> data bytes after them to the journal
@@ -119,13 +120,20 @@ int blockstore_journal_check_t::check_available(blockstore_op_t *op, int entries
         if (bs->log_level > 5)
         {
             printf(
-                "Ran out of journal space (used_start=%08jx, next_free=%08jx, dirty_start=%08jx)\n",
-                bs->journal.used_start, bs->journal.next_free, bs->journal.dirty_start
+                "Ran out of journal space (used_start=%08jx, next_free=%08jx, dirty_start=%08jx, queued %zu)\n",
+                bs->journal.used_start, bs->journal.next_free, bs->journal.dirty_start, bs->flusher->get_queue_size()
             );
         }
-        PRIV(op)->wait_for = WAIT_JOURNAL;
-        bs->flusher->request_trim();
-        PRIV(op)->wait_detail = bs->journal.used_start;
+        if (!bs->flusher->may_advance())
+        {
+            give_up = true;
+        }
+        else
+        {
+            PRIV(op)->wait_for = WAIT_JOURNAL;
+            PRIV(op)->wait_detail = bs->wait_journal_counter;
+            bs->flusher->request_trim();
+        }
         return 0;
     }
     return 1;

src/blockstore/v1/journal.h

@@ -70,6 +70,7 @@ struct blockstore_journal_check_t
     uint64_t next_pos, next_sector, next_in_pos;
     int sectors_to_write, first_sector;
     bool right_dir; // writing to the end or the beginning of the ring buffer
+    bool give_up;
 
     blockstore_journal_check_t(blockstore_impl_t *bs);
     int check_available(blockstore_op_t *op, int required, int size, int data_after);

src/blockstore/v1/rollback.cpp

@@ -63,6 +63,12 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, op->len, sizeof(journal_entry_rollback), 0))
     {
+        if (space_check.give_up)
+        {
+            op->retval = -EAGAIN;
+            FINISH_OP(op);
+            return 2;
+        }
         return 0;
     }
     // There is sufficient space. Check SQEs
@@ -162,7 +168,10 @@ void blockstore_impl_t::mark_rolled_back(const obj_ver_id & ov)
             if (unstab_it != unstable_writes.end())
             {
                 if (max_unstable == 0)
+                {
                     unstable_writes.erase(unstab_it);
+                    wakeup_wait_journal();
+                }
                 else
                     unstab_it->second = max_unstable;
             }

src/blockstore/v1/stable.cpp

@@ -362,6 +362,12 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, op->len, sizeof(journal_entry_stable), 0))
     {
+        if (space_check.give_up)
+        {
+            op->retval = -EAGAIN;
+            FINISH_OP(op);
+            return 2;
+        }
         return 0;
     }
     // There is sufficient space. Check SQEs
@@ -557,6 +563,7 @@ void blockstore_impl_t::mark_stable(obj_ver_id v, bool forget_dirty)
         unstab_it->second <= v.version)
     {
         unstable_writes.erase(unstab_it);
+        wakeup_wait_journal();
     }
 }
 

src/blockstore/v1/sync.cpp

@@ -79,6 +79,7 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
         // 2nd step: Data device is synced, prepare & write journal entries
         // Check space in the journal and journal memory buffers
         blockstore_journal_check_t space_check(this);
+        bool space_ok = true;
         if (dsk.csum_block_size)
         {
             // More complex check because all journal entries have different lengths
@@ -90,13 +91,34 @@ int blockstore_impl_t::continue_sync(blockstore_op_t *op)
                 uint64_t dyn_size = dsk.dirty_dyn_size(dirty_entry.offset, dirty_entry.len);
                 if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size, 0))
                 {
-                    return 0;
+                    space_ok = false;
+                    break;
                 }
             }
         }
         else if (!space_check.check_available(op, PRIV(op)->sync_big_writes.size(),
             sizeof(journal_entry_big_write) + dsk.clean_entry_bitmap_size, 0))
         {
+            space_ok = false;
+        }
+        if (!space_ok)
+        {
+            if (space_check.give_up)
+            {
+                // Return everything back
+                unsynced_big_write_count += PRIV(op)->sync_big_writes.size();
+                PRIV(op)->sync_big_writes.insert(PRIV(op)->sync_big_writes.end(),
+                    unsynced_big_writes.begin(), unsynced_big_writes.end());
+                PRIV(op)->sync_big_writes.swap(unsynced_big_writes);
+                PRIV(op)->sync_big_writes.clear();
+                PRIV(op)->sync_small_writes.insert(PRIV(op)->sync_small_writes.end(),
+                    unsynced_small_writes.begin(), unsynced_small_writes.end());
+                PRIV(op)->sync_small_writes.swap(unsynced_small_writes);
+                PRIV(op)->sync_small_writes.clear();
+                op->retval = -EAGAIN;
+                FINISH_OP(op);
+                return 2;
+            }
             return 0;
         }
         // Check SQEs. Don't bother about merging, submit each journal sector as a separate request
@@ -193,6 +215,7 @@ void blockstore_impl_t::ack_sync(blockstore_op_t *op)
         {
             unstable_unsynced--;
             assert(unstable_unsynced >= 0);
+            wakeup_wait_journal();
         }
         dirty_it++;
         while (dirty_it != dirty_db.end() && dirty_it->first.oid == it->oid)
@@ -228,6 +251,7 @@ void blockstore_impl_t::ack_sync(blockstore_op_t *op)
             {
                 unstable_unsynced--;
                 assert(unstable_unsynced >= 0);
+                wakeup_wait_journal();
             }
         }
     }

src/blockstore/v1/write.cpp

@@ -39,6 +39,11 @@ bool blockstore_impl_t::enqueue_write(blockstore_op_t *op)
         dirty_it--; // segfaults when dirty_db is empty
         if (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
         {
+            if (op->opcode == BS_OP_WRITE_STABLE && !IS_INSTANT(dirty_it->second.state))
+            {
+                op->retval = -EINVAL;
+                return false;
+            }
             found = true;
             version = dirty_it->first.version + 1;
             deleted = IS_DELETE(dirty_it->second.state);
@@ -312,6 +317,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
             sizeof(journal_entry_big_write) + dsk.clean_dyn_size,
             (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
+            if (space_check.give_up)
+            {
+                cancel_all_writes(op, dirty_it, -EAGAIN);
+                return 2;
+            }
             return 0;
         }
         // Big (redirect) write
@@ -404,6 +414,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
                 sizeof(journal_entry_small_write) + dyn_size,
                 op->len + (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
+            if (space_check.give_up)
+            {
+                cancel_all_writes(op, dirty_it, -EAGAIN);
+                return 2;
+            }
             return 0;
         }
         // There is sufficient space. Check SQE(s)
@@ -559,6 +574,25 @@ resume_2:
         if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size,
             (unstable_writes.size()+unstable_unsynced+((dirty_it->second.state & BS_ST_INSTANT) ? 0 : 1))*journal.block_size))
         {
+            if (space_check.give_up)
+            {
+                auto uo_it = used_clean_objects.find(dirty_it->second.location);
+                if (uo_it != used_clean_objects.end())
+                    uo_it->second.was_freed = true;
+                else
+                    data_alloc->set(dirty_it->second.location / dsk.data_block_size, false);
+                if (!(dirty_it->second.state & BS_ST_INSTANT))
+                {
+                    unstable_unsynced--;
+                    assert(unstable_unsynced >= 0);
+                    wakeup_wait_journal();
+                }
+                if (immediate_commit != IMMEDIATE_ALL)
+                    unsynced_big_write_count--;
+                write_iodepth--;
+                cancel_all_writes(op, dirty_it, -EAGAIN);
+                return 2;
+            }
             return 0;
         }
         BS_SUBMIT_CHECK_SQES(1);
@@ -622,6 +656,7 @@ resume_4:
             {
                 unstable_unsynced--;
                 assert(unstable_unsynced >= 0);
+                wakeup_wait_journal();
             }
         }
         dirty_it->second.state = (dirty_it->second.state & ~BS_ST_WORKFLOW_MASK)
@@ -774,6 +809,11 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
     blockstore_journal_check_t space_check(this);
     if (!space_check.check_available(op, 1, sizeof(journal_entry_del), (unstable_writes.size()+unstable_unsynced)*journal.block_size))
     {
+        if (space_check.give_up)
+        {
+            cancel_all_writes(op, dirty_it, -EAGAIN);
+            return 2;
+        }
         return 0;
     }
     // Write current journal sector only if it's dirty and full, or in the immediate_commit mode

src/client/cluster_client.cpp

@@ -1250,9 +1250,11 @@ resume_2:
         return 1;
     }
     else if (op->retval != 0 && op->opcode != OSD_OP_SYNC && !(op->flags & OP_FLUSH_BUFFER) &&
-        op->retval != -EPIPE && (op->retval != -EIO || !client_eio_retry_interval) && (op->retval != -ENOSPC || !client_retry_enospc))
+        op->retval != -EPIPE && op->retval != -EAGAIN &&
+        (op->retval != -EIO || !client_eio_retry_interval) &&
+        (op->retval != -ENOSPC || !client_retry_enospc))
     {
-        // Fatal error (neither -EPIPE, -EIO nor -ENOSPC)
+        // Fatal error (neither -EPIPE, -EAGAIN, -EIO nor -ENOSPC)
         erase_op(op);
         return 1;
     }
@@ -1611,7 +1613,7 @@ resume_1:
     if (op->retval != 0)
     {
         wb->fsync_error();
-        if (op->retval == -EPIPE || op->retval == -EIO || op->retval == -ENOSPC)
+        if (op->retval == -EPIPE || op->retval == -EAGAIN || op->retval == -EIO || op->retval == -ENOSPC)
         {
             // Retry later
             op->parts.clear();
@@ -1662,9 +1664,6 @@ static inline void mem_or(void *res, const void *r2, unsigned int len)
     }
 }
 
-// Error priority: others > EPERM > EIO > ENOSPC > ETIMEDOUT > EPIPE
-#define ERR_PRIO(e) (((e) == -EPERM ? 5 : ((e) == -EIO ? 4 : ((e) == -ENOSPC ? 3 : ((e) == -ETIMEDOUT ? 2 : ((e) == -EPIPE ? 1 : (!(e) ? 0 : 10)))))))
-
 void cluster_client_t::handle_op_part(cluster_op_part_t *part)
 {
     cluster_op_t *op = part->parent;
@@ -1676,7 +1675,7 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         if (ERR_PRIO(part->op.reply.hdr.retval) > ERR_PRIO(op->retval))
             op->retval = part->op.reply.hdr.retval;
         uint64_t stop_client_id = 0;
-        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM)
+        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM && op->retval != -EAGAIN)
         {
             stop_client_id = part->op.client_id;
             if (op->retval != -EPIPE || log_level > 0)
@@ -1698,11 +1697,11 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         // So do all these things after modifying operation state, otherwise we may hit reenterability bugs
         // FIXME postpone such things to set_immediate here to avoid bugs
         // Set op->retry_after to retry operation after a short pause (not immediately)
-        if (!op->retry_after && (op->retval == -EPIPE ||
+        if (!op->retry_after && (op->retval == -EPIPE || op->retval == -EAGAIN ||
             op->retval == -EIO && client_eio_retry_interval ||
             op->retval == -ENOSPC && client_retry_enospc))
         {
-            op->retry_after = op->retval != -EPIPE ? client_eio_retry_interval : client_retry_interval;
+            op->retry_after = op->retval != -EPIPE && op->retval != -EAGAIN ? client_eio_retry_interval : client_retry_interval;
         }
         reset_retry_timer(op->retry_after);
         if (stop_client_id)

src/client/cluster_client_list.cpp

@@ -370,8 +370,7 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
             fprintf(stderr, "Failed to get PG %u/%u object list from OSD %ju (retval=%jd), skipping\n",
                 cur_list->pg->lst->pool_id, cur_list->pg->pg_num, cur_list->osd_num, op->reply.hdr.retval);
             if (!cur_list->pg->errcode ||
-                cur_list->pg->errcode == -EPIPE ||
-                op->reply.hdr.retval != -EPIPE)
+                ERR_PRIO(op->reply.hdr.retval) > ERR_PRIO(cur_list->pg->errcode))
             {
                 cur_list->pg->errcode = op->reply.hdr.retval;
             }
@@ -434,7 +433,7 @@ void cluster_client_t::finish_list_pg(inode_list_pg_t *pg, bool retry_epipe)
     auto lst = pg->lst;
     if (pg->inflight_ops == 0)
     {
-        if (pg->errcode == -EPIPE && retry_epipe)
+        if ((pg->errcode == -EPIPE || pg->errcode == -EAGAIN) && retry_epipe)
         {
             // Retry listing after <client_retry_interval> ms on EPIPE
             pg->state = LIST_PG_WAIT_RETRY;

src/client/messenger.cpp

@@ -504,18 +504,21 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd)
     int one = 1;
     setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
     cl->peer_state = PEER_CONNECTED;
-    tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
+    // EPOLLOUT is only required without io_uring, to continue partially sent messages
+    tfd->set_fd_handler(peer_fd, use_sync_send_recv, [this](int peer_fd, int epoll_events)
     {
         handle_peer_epoll(peer_fd, epoll_events);
     });
     // Check OSD number
-    init_tls_client(cl);
+    if (!init_tls_client(cl))
+    {
+        return;
+    }
     check_peer_config(cl);
 }
 
 void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
 {
-    // Mark client as ready (i.e. some data is available)
     auto cl = clients_by_fd.at(peer_fd);
     if (epoll_events & EPOLLRDHUP)
     {
@@ -525,8 +528,9 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
             fprintf(stderr, "[OSD %ju] client %ju disconnected\n", this->osd_num, cl->client_id);
         }
         stop_client(cl->client_id);
+        return;
     }
-    else if (epoll_events & EPOLLIN)
+    if (epoll_events & EPOLLIN)
     {
         // Mark client as ready (i.e. some data is available)
         cl->read_ready++;
@@ -539,6 +543,23 @@ void osd_messenger_t::handle_peer_epoll(int peer_fd, int epoll_events)
                 read_requests();
         }
     }
+    if (epoll_events & EPOLLOUT)
+    {
+        // The socket became writable again - continue the interrupted use_sync_send_recv
+        // read_requests() above may have stopped the client, so look it up again
+        auto cl_it = clients.find(cl->client_id);
+        if (cl_it == clients.end())
+        {
+            return;
+        }
+        cl = cl_it->second;
+        cl->write_ready = true;
+        // send_list is only left non-empty when we're waiting for EPOLLOUT
+        if (cl->send_list.size() && cl->peer_state != PEER_RDMA)
+        {
+            try_send(cl);
+        }
+    }
 }
 
 void osd_messenger_t::on_connect_peer(osd_num_t peer_osd, int errcode, uint64_t client_id)
@@ -770,9 +791,12 @@ void osd_messenger_t::accept_connections(int listen_fd)
         cl->peer_fd = peer_fd;
         cl->peer_state = PEER_CONNECTED;
         cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
-        init_tls_client(cl);
-        // Add FD to epoll
-        tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
+        if (!init_tls_client(cl))
+        {
+            continue;
+        }
+        // Add FD to epoll. EPOLLOUT is only required without io_uring
+        tfd->set_fd_handler(peer_fd, use_sync_send_recv, [this](int peer_fd, int epoll_events)
         {
             handle_peer_epoll(peer_fd, epoll_events);
         });

src/client/messenger.h

@@ -147,6 +147,8 @@ struct osd_client_t
     size_t write_op_pos = 0;
     msghdr write_msg = { 0 };
     int write_state = 0;
+    // Socket is writable. Only tracked with use_sync_send_recv, io_uring doesn't need it
+    bool write_ready = true;
     std::vector<iovec> send_list;
     size_t send_list_size = 0;
     std::deque<osd_op_t*> send_free_ops;
@@ -248,7 +250,7 @@ protected:
 
     void init_tls();
     void destroy_tls();
-    void init_tls_client(osd_client_t *cl);
+    bool init_tls_client(osd_client_t *cl);
     bool do_tls_handshake(osd_client_t *cl, bool from_recv = false);
     bool derive_aes_keys(osd_client_t *cl, bool update_my, bool update_peer);
 
@@ -343,7 +345,8 @@ protected:
     void cancel_op(osd_op_t *op);
 
     bool try_send(osd_client_t *cl);
-    void handle_send(int result, bool prev, bool more, osd_client_t *cl);
+    bool wakeup_send(osd_client_t *cl);
+    bool handle_send(int result, bool prev, bool more, osd_client_t *cl);
     static void post_send_free(osd_op_t *op);
     bool op_write_to(osd_client_t *cl, msgr_op_writer_t & wr);
     void next_write_op(osd_client_t *cl);
@@ -370,7 +373,7 @@ protected:
     void op_decrypt_free(osd_client_t* cl);
 
 #ifdef WITH_RDMA
-    void try_send_rdma(osd_client_t *cl);
+    bool try_send_rdma(osd_client_t *cl);
     bool init_recv_rdma(osd_client_t *cl);
     void handle_rdma_events(msgr_rdma_context_t *rdma_context);
     msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl);

src/client/msgr_encrypt.cpp

@@ -616,7 +616,7 @@ void osd_messenger_t::init_tls()
     }
 }
 
-void osd_messenger_t::init_tls_client(osd_client_t *cl)
+bool osd_messenger_t::init_tls_client(osd_client_t *cl)
 {
     if (gcm_enabled)
     {
@@ -625,13 +625,13 @@ void osd_messenger_t::init_tls_client(osd_client_t *cl)
         cl->hs->init(cl->is_incoming);
         if (cl->hs->out_size())
         {
-            if (cl->write_state == 0)
+            if (!wakeup_send(cl))
             {
-                cl->write_state = CL_WRITE_READY;
-                write_ready_clients.push_back(cl->client_id);
+                return false;
             }
         }
     }
+    return true;
 }
 
 void osd_messenger_t::destroy_tls()

src/client/msgr_rdma.cpp

@@ -571,7 +571,7 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
     cl->rdma_conn->cur_send++;
 }
 
-void osd_messenger_t::try_send_rdma(osd_client_t *cl)
+bool osd_messenger_t::try_send_rdma(osd_client_t *cl)
 {
     auto rc = cl->rdma_conn;
     if (!rc->send_out_size)
@@ -602,7 +602,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
         if (cl->io_error)
         {
             stop_client(cl->client_id);
-            return;
+            return false;
         }
         if (copied > 0)
         {
@@ -622,6 +622,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
             cl->send_free_ops.push_back(NULL); // end marker
         }
     }
+    return true;
 }
 
 static void try_recv_rdma_wr(osd_client_t *cl, void *buf)
@@ -750,6 +751,7 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
             }
         }
     } while (event_count > 0);
+    handle_immediate_ops();
 }
 
 void osd_messenger_t::destroy_rdma_conn(msgr_rdma_connection_t *rdma_conn)

src/client/msgr_rdmacm.cpp

@@ -503,7 +503,6 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
     cl->osd_num = peer_osd;
     cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
     cl->rdma_conn = rc;
-    init_tls_client(cl);
     clients[conn->client_id] = cl;
     if (conn->timeout_id >= 0)
     {
@@ -517,6 +516,10 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
         fprintf(stderr, "Successfully connected with OSD %ju using RDMA-CM\n", peer_osd);
     }
     // Add initial receive request(s)
+    if (!init_tls_client(cl))
+    {
+        return;
+    }
     init_recv_rdma(cl);
     if (peer_osd)
     {

src/client/msgr_receive.cpp

@@ -458,8 +458,7 @@ void osd_messenger_t::read_requests()
             {
                 result = -errno;
             }
-            // like set_immediate
-            tfd->set_timer_us(0, false, [this, result, cl](int){ handle_read(result, cl); });
+            handle_read(result, cl);
         }
     }
     read_ready_clients.clear();
@@ -490,8 +489,10 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl)
         }
         stop_client(cl->client_id);
 out_wakeup:
-        if (set_immediate_ops.size())
+        if (ringloop && set_immediate_ops.size())
+        {
             ringloop->wakeup();
+        }
         return;
     }
     bool full_read = false;
@@ -501,7 +502,9 @@ out_wakeup:
         {
             full_read = result >= cl->read_iov.iov_len;
             if (!handle_read_buffer(cl, cl->in_buf, result))
+            {
                 goto out_wakeup;
+            }
         }
         else
         {
@@ -597,13 +600,16 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size
             bufsize -= done;
             if (cl->hs->out_size())
             {
-                if (cl->write_state == 0)
+                if (!wakeup_send(cl))
                 {
-                    cl->write_state = CL_WRITE_READY;
-                    write_ready_clients.push_back(cl->client_id);
+                    return false;
                 }
             }
-            if (cl->hs->done() && !cl->hs->out_size())
+            if (!cl->hs)
+            {
+                // Deleted by wakeup_send with use_sync_send_recv
+            }
+            else if (cl->hs->done() && !cl->hs->out_size())
             {
                 // Delete hs when done and nothing to send
                 delete cl->hs;

src/client/msgr_send.cpp

@@ -17,6 +17,10 @@
 
 #define GCM_TMP_BUF_SIZE 4096
 
+// How many times to retry sendmsg() in place before falling back to EPOLLOUT.
+// Only used with use_sync_send_recv - io_uring waits for the socket by itself
+#define MSGR_SEND_RETRIES 4
+
 class msgr_op_writer_t
 {
 public:
@@ -467,36 +471,60 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
         return;
     }
 #endif
-    if (!ringloop)
+    try_send(cl);
+}
+
+bool osd_messenger_t::wakeup_send(osd_client_t *cl)
+{
+    if (cl->peer_state == PEER_RDMA)
     {
-        // FIXME: It's worse because it doesn't allow batching
-        while (cl->write_op || cl->write_ops.size())
-        {
-            try_send(cl);
-        }
+        return try_send_rdma(cl);
     }
-    else
+    if (use_sync_send_recv)
     {
-        if (!try_send(cl) && cl->write_state == 0)
-        {
-            cl->write_state = CL_WRITE_READY;
-            write_ready_clients.push_back(cl->client_id);
-        }
-        ringloop->wakeup();
+        // FIXME: Batch send/recv without io_uring too.
+        return try_send(cl);
     }
+    if (cl->write_state == 0)
+    {
+        cl->write_state = CL_WRITE_READY;
+        write_ready_clients.push_back(cl->client_id);
+    }
+    ringloop->wakeup();
+    return true;
 }
 
 bool osd_messenger_t::try_send(osd_client_t *cl)
 {
+    assert(cl->peer_state != PEER_RDMA);
+restart:
     if (cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
     {
         return true;
     }
-    if (cl->write_msg.msg_iovlen > 0 || !ringloop->space_left() && !use_sync_send_recv)
+    if (cl->write_msg.msg_iovlen > 0)
     {
-        return false;
+        return true;
+    }
+    if (!use_sync_send_recv)
+    {
+        if (!ringloop->space_left())
+        {
+            if (cl->write_state == 0)
+            {
+                cl->write_state = CL_WRITE_READY;
+                write_ready_clients.push_back(cl->client_id);
+            }
+            ringloop->wakeup();
+            return true;
+        }
+    }
+    else if (!cl->write_ready)
+    {
+        // Socket buffer is full, the unsent tail is still in send_list.
+        // Do not copy more data into it, just wait for EPOLLOUT
+        return true;
     }
-    assert(cl->peer_state != PEER_RDMA);
     if (cl->hs)
     {
         // Send handshake message
@@ -523,11 +551,10 @@ copy_ops:
     if (cl->io_error)
     {
         stop_client(cl->client_id);
-        return true;
+        return false;
     }
     if (!cl->send_list.size())
     {
-        cl->write_state = 0;
         return true;
     }
     // get_op_writer_t already guarantees that it's under IOV_MAX
@@ -575,16 +602,29 @@ copy_ops:
     }
     else
     {
-        cl->write_msg.msg_iov = cl->send_list.data();
-        cl->write_msg.msg_iovlen = cl->send_list.size();
-        cl->refs++;
-        int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL);
-        if (result < 0)
+        // Try to first send with a busy loop, then postpone to the EPOLLOUT handler
+        int send_retries = MSGR_SEND_RETRIES;
+        while (cl->send_list.size())
         {
-            result = -errno;
+            cl->write_msg.msg_iov = cl->send_list.data();
+            cl->write_msg.msg_iovlen = cl->send_list.size();
+            cl->refs++;
+            int result = sendmsg(cl->peer_fd, &cl->write_msg, MSG_NOSIGNAL);
+            if (result < 0)
+            {
+                result = -errno;
+            }
+            if (!handle_send(result, false, false, cl))
+            {
+                return false;
+            }
+            if (cl->send_list.size() && --send_retries <= 0)
+            {
+                cl->write_ready = false;
+                return true;
+            }
         }
-        // like set_immediate
-        tfd->set_timer_us(0, false, [this, result, cl](int){ handle_send(result, false, false, cl); });
+        goto restart;
     }
     return true;
 }
@@ -664,21 +704,42 @@ void osd_messenger_t::send_replies()
     {
         uint64_t client_id = write_ready_clients[i];
         auto cl_it = clients.find(client_id);
-        if (cl_it != clients.end() && cl_it->second->peer_state != PEER_RDMA && !try_send(cl_it->second))
+        if (cl_it != clients.end() && cl_it->second->peer_state != PEER_RDMA)
         {
-            write_ready_clients.erase(write_ready_clients.begin(), write_ready_clients.begin() + i);
-            return;
+            if (!ringloop->space_left())
+            {
+                write_ready_clients.erase(write_ready_clients.begin(), write_ready_clients.begin() + i);
+                return;
+            }
+            cl_it->second->write_state = 0;
+            try_send(cl_it->second);
         }
     }
     write_ready_clients.clear();
 }
 
-void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
+static void eat_send_list(osd_client_t *cl, size_t n)
+{
+    int i = 0;
+    while (i < cl->send_list.size() && n >= cl->send_list[i].iov_len)
+    {
+        n -= cl->send_list[i].iov_len;
+        i++;
+    }
+    if (i > 0)
+        cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+i);
+    if (cl->send_list.size())
+    {
+        cl->send_list[0].iov_base = (uint8_t*)cl->send_list[0].iov_base + n;
+        cl->send_list[0].iov_len -= n;
+    }
+}
+
+bool osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t *cl)
 {
     if (!prev)
     {
         cl->write_msg.msg_iovlen = 0;
-        cl->send_list.clear();
     }
     if (!more)
     {
@@ -690,14 +751,15 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         {
             destroy_client(cl);
         }
-        return;
+        // FIXME: Return error codes instead of true/false (?)
+        return false;
     }
     if (result < 0 && result != -EAGAIN && result != -EINTR)
     {
         // this is a client socket, so don't panic. just disconnect it
         fprintf(stderr, "Client %ju socket write error: %d (%s). Disconnecting client\n", cl->client_id, -result, strerror(-result));
         stop_client(cl->client_id);
-        return;
+        return false;
     }
     if (result >= 0)
     {
@@ -709,14 +771,26 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
                 post_send_free(cl->zc_free_list[i]);
             if (i > 0)
                 cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1);
-            return;
+            return true;
         }
-        if (cl->send_list_size > result)
+        if (cl->send_list_size <= result)
+        {
+            cl->send_list.clear();
+            cl->send_list_size = 0;
+        }
+        else if (!use_sync_send_recv)
         {
             fprintf(stderr, "Client %ju socket write error: expected to send "
                 "%zu bytes with MSG_WAITALL but sent %u. Disconnecting client\n", cl->client_id, cl->send_list_size, result);
             stop_client(cl->client_id);
-            return;
+            return false;
+        }
+        else
+        {
+            // Partially sent
+            eat_send_list(cl, result);
+            cl->send_list_size -= result;
+            return true;
         }
         for (auto op: cl->send_free_ops)
         {
@@ -728,9 +802,6 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         if (more)
             cl->zc_free_list.push_back(NULL); // end marker
         cl->send_free_ops.clear();
-        cl->write_state = 0;
-        if (cl->write_op || cl->write_ops.size())
-            cl->write_state = CL_WRITE_READY;
         if ((cl->proto_csum_status & MSGR_CSUM_NEG) && !cl->write_op && !cl->write_ops.size())
         {
             // Checksums negotiated, enable
@@ -750,10 +821,16 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         }
 #endif
     }
-    if (cl->write_state != 0)
+    if ((cl->write_op || cl->write_ops.size() || cl->hs && cl->hs->out_size()) && !use_sync_send_recv)
     {
-        write_ready_clients.push_back(cl->client_id);
+        if (cl->write_state == 0)
+        {
+            cl->write_state = CL_WRITE_READY;
+            write_ready_clients.push_back(cl->client_id);
+        }
+        ringloop->wakeup();
     }
+    return true;
 }
 
 static inline bool op_has_data_for_ssl(osd_op_t *op)

src/client/object_id.h

@@ -111,4 +111,15 @@ namespace std
             return seed;
         }
     };
+
+    template<> struct hash<pool_pg_num_t>
+    {
+        inline size_t operator()(const pool_pg_num_t &s) const
+        {
+            size_t seed = 0;
+            seed ^= ((size_t)s.pool_id + 0xc6a4a7935bd1e995 + (seed << 6) + (seed >> 2));
+            seed ^= ((size_t)s.pg_num + 0xc6a4a7935bd1e995 + (seed << 6) + (seed >> 2));
+            return seed;
+        }
+    };
 }

src/client/osd_ops.h

@@ -57,6 +57,9 @@
 #define OSD_SEC_LOCK_PG 1
 #define OSD_SEC_UNLOCK_PG 2
 
+// Error priority: others > EPERM > EIO > EDOM > ENOSPC > EPIPE > EAGAIN
+#define ERR_PRIO(e) (((e) == -EPERM ? 6 : ((e) == -EIO ? 5 : ((e) == -EDOM ? 4 : ((e) == -ENOSPC ? 3 : ((e) == -EPIPE ? 2 : ((e) == -EAGAIN ? 1 : (!(e) ? 0 : 10))))))))
+
 // common request and reply headers
 struct __attribute__((__packed__)) osd_op_header_t
 {

src/osd/osd_flush.cpp

@@ -81,18 +81,9 @@ void osd_t::handle_flush_op(bool rollback, pool_id_t pool_id, pg_num_t pg_num, p
     if (retval != 0)
     {
         fb->flush_errors++;
-        if (peer_osd == this->osd_num)
+        printf("Error while doing flush on OSD %ju: %d (%s)\n", peer_osd, retval, strerror(-retval));
+        if (peer_osd != this->osd_num && retval != -EAGAIN)
         {
-            throw std::runtime_error(
-                std::string(rollback
-                    ? "Error while doing local rollback operation: "
-                    : "Error while doing local stabilize operation: "
-                ) + strerror(-retval)
-            );
-        }
-        else
-        {
-            printf("Error while doing flush on OSD %ju: %d (%s)\n", osd_num, retval, strerror(-retval));
             auto peer_it = msgr.osd_peers.find(peer_osd);
             if (peer_it != msgr.osd_peers.end())
             {

src/osd/osd_primary_subops.cpp

@@ -311,11 +311,10 @@ void osd_t::handle_primary_bs_subop(osd_op_t *subop)
     blockstore_op_t *bs_op = subop->bs_op;
     int64_t expected = bs_op->opcode == BS_OP_READ || bs_op->opcode == BS_OP_WRITE
         || bs_op->opcode == BS_OP_WRITE_STABLE ? bs_op->len : 0;
-    if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ &&
-        (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE ||
-        bs_op->retval != -ENOSPC))
+    if (bs_op->retval != expected && bs_op->opcode != BS_OP_READ && bs_op->retval != -EAGAIN &&
+        (bs_op->opcode != BS_OP_WRITE && bs_op->opcode != BS_OP_WRITE_STABLE || bs_op->retval != -ENOSPC))
     {
-        // die on any error except ENOSPC during write
+        // die on any error except ENOSPC or EAGAIN during write
         if (bs_op->opcode == BS_OP_WRITE || bs_op->opcode == BS_OP_WRITE_STABLE)
         {
             printf(
@@ -428,7 +427,7 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
 #endif
         if (!(op_data->flags & OP_DATA_SKIP_VER_CHECK))
         {
-            if (op_data->fact_ver != version && (op_data->errors > 0 || op_data->done > 0) &&
+            if (op_data->fact_ver != version && op_data->done > 0 &&
                 (version != 0 && op_data->fact_ver != 0 || !(op_data->flags & OP_DATA_ENOENT_OK)))
             {
                 fprintf(
@@ -473,16 +472,12 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
             );
         }
         subop->rmw_buf = NULL;
-        // Error priority: ENOSPC > others > EIO > EDOM > EPIPE
-        if (op_data->errcode == 0 ||
-            retval == -ENOSPC && op_data->errcode != -ENOSPC ||
-            retval == -EIO && (op_data->errcode == -EDOM || op_data->errcode == -EPIPE) ||
-            retval == -EDOM && (op_data->errcode == -EPIPE) ||
-            retval != -EIO && retval != -EDOM && retval != -EPIPE)
+        // Error priority: ENOSPC > others > EIO > EDOM > EPIPE > EAGAIN
+        if (op_data->errcode == 0 || ERR_PRIO(retval) > ERR_PRIO(op_data->errcode))
         {
             op_data->errcode = retval;
         }
-        if (subop->client_id && retval != -EDOM && retval != -ERANGE &&
+        if (subop->client_id && retval != -EDOM && retval != -ERANGE && retval != -EAGAIN &&
             (retval != -ENOSPC || opcode != OSD_OP_SEC_WRITE && opcode != OSD_OP_SEC_WRITE_STABLE) &&
             (retval != -EIO || opcode != OSD_OP_SEC_READ))
         {
@@ -490,6 +485,32 @@ void osd_t::handle_primary_subop(osd_op_t *subop, osd_op_t *cur_op)
             msgr.stop_client(subop->client_id);
             op_data->drops++;
         }
+        else if (opcode == OSD_OP_SEC_STABILIZE || opcode == OSD_OP_SEC_ROLLBACK)
+        {
+            // STABILIZE/ROLLBACK should never fail, but if they do, we repeer the PGs
+            assert(subop->iov.count == 1);
+            obj_ver_id *vers = (obj_ver_id*)subop->iov.buf[0].iov_base;
+            size_t n = subop->iov.buf[0].iov_len/sizeof(obj_ver_id);
+            robin_hood::unordered_flat_set<pool_pg_num_t> to_repeer;
+            for (size_t i = 0; i < n; i++)
+            {
+                to_repeer.insert((pool_pg_num_t){
+                    .pool_id = INODE_POOL(vers[n].oid.inode),
+                    .pg_num = map_to_pg(vers[n].oid),
+                });
+            }
+            ringloop->set_immediate([this, to_repeer = std::move(to_repeer)]()
+            {
+                for (auto & ppg: to_repeer)
+                {
+                    auto pg_it = pgs.find(ppg);
+                    if (pg_it != pgs.end() && (pg_it->second.state & PG_ACTIVE))
+                    {
+                        repeer_pg(pg_it->second);
+                    }
+                }
+            });
+        }
         // Increase op_data->errors after stop_client to prevent >= n_subops running twice
         op_data->errors++;
     }
@@ -703,6 +724,7 @@ void osd_t::submit_primary_stab_subops(osd_op_t *cur_op)
         {
             clock_gettime(CLOCK_REALTIME, &subops[i].tv_begin);
             subops[i].op_type = (uint64_t)cur_op;
+            subops[i].iov.push_back(op_data->unstable_writes + stab_osd.start, stab_osd.len * sizeof(obj_ver_id));
             subops[i].bs_op = new blockstore_op_t((blockstore_op_t){
                 .opcode = BS_OP_STABLE,
                 .callback = [subop = &subops[i], this](blockstore_op_t *bs_subop)
@@ -724,7 +746,7 @@ void osd_t::submit_primary_stab_subops(osd_op_t *cur_op)
                     .magic = SECONDARY_OSD_OP_MAGIC,
                     .opcode = OSD_OP_SEC_STABILIZE,
                 },
-                .len = (uint64_t)(stab_osd.len * sizeof(obj_ver_id)),
+                .len = (uint64_t)stab_osd.len * sizeof(obj_ver_id),
                 .flags = cur_op->client_id == SELF_CLIENT && cur_op->req.hdr.opcode != OSD_OP_SCRUB ? OSD_OP_RECOVERY_RELATED : 0,
             } };
             subops[i].iov.push_back(op_data->unstable_writes + stab_osd.start, stab_osd.len * sizeof(obj_ver_id));
@@ -778,6 +800,7 @@ void osd_t::submit_primary_rollback_subops(osd_op_t *cur_op, const uint64_t* osd
             {
                 clock_gettime(CLOCK_REALTIME, &subop->tv_begin);
                 subop->op_type = (uint64_t)cur_op;
+                subop->iov.push_back(op_data->unstable_writes + i, sizeof(obj_ver_id));
                 subop->bs_op = new blockstore_op_t((blockstore_op_t){
                     .opcode = BS_OP_ROLLBACK,
                     .callback = [subop, this](blockstore_op_t *bs_subop)

src/osd/osd_primary_write.cpp

@@ -258,9 +258,21 @@ resume_4:
         return;
     }
 resume_5:
+    if (!(pg.state & PG_ACTIVE))
+    {
+        if (op_data->subops)
+        {
+            delete[] op_data->subops;
+            op_data->subops = NULL;
+        }
+        pg.ver_override.erase(op_data->oid);
+        deref_object_state(pg, &op_data->object_state, true);
+        pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
+        return;
+    }
     if (op_data->errors > 0)
     {
-        // Handle ENOSPC/EDOM/ERANGE/EIO. If some subops fail, but others succeed,
+        // Handle ENOSPC/EAGAIN/EDOM/ERANGE/EIO. If some subops fail, but others succeed,
         // next writes to the same object will also fail because they'll try
         // to overwrite the same version number which will result in EEXIST.
         // To fix it, we should mark the object as degraded for replicas,
@@ -269,6 +281,11 @@ resume_5:
         {
             if (pg.scheme != POOL_SCHEME_REPLICATED)
             {
+                if (op_data->subops)
+                {
+                    delete[] op_data->subops;
+                    op_data->subops = NULL;
+                }
                 submit_primary_rollback_subops(cur_op, pg.cur_set.data());
 resume_11:
                 if (op_data->n_subops > 0)
@@ -277,21 +294,17 @@ resume_11:
                     return;
                 }
 resume_12:
-                // Ignore ROLLBACK errors - submit_primary_subops will drop the connection if it fails
-                delete[] op_data->unstable_writes;
-                op_data->unstable_writes = NULL;
+                if (op_data->unstable_writes)
+                {
+                    delete[] op_data->unstable_writes;
+                    op_data->unstable_writes = NULL;
+                }
+                // Ignore ROLLBACK errors - submit_primary_subops will repeer the PG if it fails
             }
             else
             {
                 pg.ver_override.erase(op_data->oid);
                 mark_partial_write(pg, cur_op);
-                if (op_data->subops)
-                {
-                    delete[] op_data->subops;
-                    op_data->subops = NULL;
-                }
-                pg_cancel_write_queue(pg, cur_op, op_data->oid, op_data->errcode);
-                return;
             }
         }
         if (op_data->subops)
@@ -304,13 +317,6 @@ resume_12:
         cur_op->reply.hdr.retval = op_data->errcode;
         goto continue_others;
     }
-    if (!(pg.state & PG_ACTIVE))
-    {
-        pg.ver_override.erase(op_data->oid);
-        deref_object_state(pg, &op_data->object_state, true);
-        pg_cancel_write_queue(pg, cur_op, op_data->oid, -EPIPE);
-        return;
-    }
     if (pg.scheme != POOL_SCHEME_REPLICATED)
     {
         // Remove version override just after the write, but before stabilizing
@@ -555,6 +561,7 @@ resume_6:
                 return false;
             }
 resume_7:
+            // Ignore STABILIZE errors - submit_primary_subops will repeer the PG if it fails
             // FIXME: Free those in the destructor?
             delete op_data->unstable_write_osds;
             delete[] op_data->unstable_writes;

src/test/mock/messenger.cpp

@@ -73,6 +73,11 @@ void osd_messenger_t::accept_connections(int listen_fd)
 {
 }
 
+bool osd_messenger_t::wakeup_send(osd_client_t *cl)
+{
+    return true;
+}
+
 #ifdef WITH_RDMA
 json11::Json osd_messenger_t::connect_rdma(uint64_t client_id, std::string rdma_address, uint64_t client_max_msg)
 {

src/test/test_blockstore.cpp

@@ -137,12 +137,13 @@ struct bs_test_t
         }
     }
 
-    void exec_op(blockstore_op_t *op)
+    void exec_op(blockstore_op_t *op, bool print = true)
     {
         bool done = false;
         op->callback = [&](blockstore_op_t *op)
         {
-            printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
+            if (print)
+                printf("op opcode=%ju completed retval=%jd\n", op->opcode, op->retval);
             done = true;
         };
         bs->enqueue_op(op);
@@ -976,6 +977,62 @@ static void test_list_limit()
     free(op.buf);
 }
 
+static void test_write_no_space_eagain()
+{
+    printf("\n-- test_write_no_space_eagain\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    // 1 MB journal (data_offset 32 MB - journal_offset)
+    test.config["journal_offset"] = std::to_string(31*1024*1024);
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(buf, 0xAA, 4096);
+
+    // fill the journal with unstable writes
+    int i = 0, blocked_retval = 0;
+    auto do_write = [&]()
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_WRITE;
+        op.oid = { .inode = 1, .stripe = (uint64_t)(i % 4) * 0x20000 };
+        op.version = 1 + i / 4;
+        op.offset = 0;
+        op.len = 4096;
+        op.buf = buf;
+        test.exec_op(&op, false);
+        if (op.retval != (int)op.len)
+            blocked_retval = op.retval;
+    };
+    for (i = 0; i < 100000 && !blocked_retval; i++)
+        do_write();
+    printf("write #%d refused with retval=%d\n", i, blocked_retval);
+    assert(blocked_retval == -EAGAIN);
+
+    // check that a stabilize operation and one more write can proceed
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_STABLE;
+        op.len = 1;
+        op.buf = (uint8_t*)malloc_or_die(1 * sizeof(obj_ver_id));
+        ((obj_ver_id*)op.buf)[0] = {
+            .oid = { .inode = 1, .stripe = 0 },
+            .version = 2,
+        };
+        test.exec_op(&op);
+        assert(op.retval == 0);
+        free(op.buf);
+    }
+
+    blocked_retval = 0;
+    do_write();
+    assert(!blocked_retval);
+
+    free(buf);
+}
+
 // FIXME Add a simple intent_write / big_intent test
 
 int main(int narg, char *args[])
@@ -996,5 +1053,6 @@ int main(int narg, char *args[])
     test_fsync_almost_batch_big();
     test_fsync_batch_big();
     test_list_limit();
+    test_write_no_space_eagain();
     return 0;
 }

src/test/test_blockstore_v1.cpp

@@ -883,6 +883,75 @@ static void test_meta_tail_reload()
     printf("blockstore re-initialized with an overflowing entry\n");
 }
 
+static void test_write_no_space_eagain()
+{
+    printf("\n-- test_write_no_space_eagain\n");
+
+    bs_test_t test;
+    test.default_cfg();
+    // 1 MB journal (data_offset 32 MB - journal_offset), so it fills in a few hundred writes
+    test.config["journal_offset"] = std::to_string(31*1024*1024);
+    test.init();
+    printf("blockstore initialized\n");
+
+    uint8_t *buf = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, 4096);
+    memset(buf, 0xAA, 4096);
+
+    // fill the journal with unstable writes
+    int i = 0, blocked_retval = 0;
+    auto do_write = [&]()
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_WRITE;
+        op.oid = { .inode = 1, .stripe = (uint64_t)(i % 4) * 0x20000 };
+        op.version = 1 + i / 4;
+        op.offset = 0;
+        op.len = 4096;
+        op.buf = buf;
+        test.exec_op(&op);
+        if (op.retval != (int)op.len)
+            blocked_retval = op.retval;
+    };
+    for (i = 0; i < 100000 && !blocked_retval; i++)
+        do_write();
+    printf("write #%d refused with retval=%d\n", i, blocked_retval);
+    assert(blocked_retval == -EAGAIN);
+
+    // check that a stabilize operation and one more write can proceed
+    // journal space_check is simplified so it requires slightly more space than actually needed
+    {
+        blockstore_op_t op;
+        op.opcode = BS_OP_STABLE;
+        op.len = 4;
+        op.buf = (uint8_t*)malloc_or_die(4 * sizeof(obj_ver_id));
+        ((obj_ver_id*)op.buf)[0] = {
+            .oid = { .inode = 1, .stripe = 0 },
+            .version = 2,
+        };
+        ((obj_ver_id*)op.buf)[1] = {
+            .oid = { .inode = 1, .stripe = 0x20000 },
+            .version = 2,
+        };
+        ((obj_ver_id*)op.buf)[2] = {
+            .oid = { .inode = 1, .stripe = 0x40000 },
+            .version = 1,
+        };
+        ((obj_ver_id*)op.buf)[3] = {
+            .oid = { .inode = 1, .stripe = 0x60000 },
+            .version = 1,
+        };
+        test.exec_op(&op);
+        assert(op.retval == 0);
+        free(op.buf);
+    }
+
+    blocked_retval = 0;
+    do_write();
+    assert(!blocked_retval);
+
+    free(buf);
+}
+
 // Check null dereference in read_bitmap over DELETE
 static void test_read_bitmap_of_deleted_object()
 {
@@ -1138,6 +1207,7 @@ int main(int narg, char *args[])
     test_read_retry_on_ring_full_1M_csum4k_clean();
     test_read_free_clean_loc_used();
     test_meta_tail_reload();
+    test_write_no_space_eagain();
     test_read_bitmap_of_deleted_object();
     test_used_blocks_replay_stable_over_small_write();
     test_used_blocks_replay_delete_over_unstable_big_write();

tests/run_tests.sh

@@ -27,6 +27,8 @@ ETCD_SCHEME=https ./test_snapshot.sh
 
 ./test_img_online_resize.sh
 
+./test_no_uring.sh
+
 ./test_interrupted_rebalance.sh
 IMMEDIATE_COMMIT=1 ./test_interrupted_rebalance.sh
 SCHEME=ec ./test_interrupted_rebalance.sh

tests/test_no_uring.sh

@@ -0,0 +1,8 @@
+#!/bin/bash -ex
+
+. `dirname $0`/run_3osds.sh
+
+$VITASTOR_FIO -bs=128k -direct=1 -numjobs=1 -iodepth=4 \
+    -rw=write -pool=1 -inode=1 -size=128M -runtime=10 -no_io_uring=1
+
+format_green OK