Fix another possible reenterability issue in cluster_client

Non-reproducible in QEMU/FIO, only caught during K/V DB debugging

src/cluster_client.cpp

@@ -169,46 +169,48 @@ void cluster_client_t::calc_wait(cluster_op_t *op)
 
 void cluster_client_t::inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *next, int inc)
 {
-    if (opcode == OSD_OP_WRITE)
+    if (opcode != OSD_OP_WRITE && opcode != OSD_OP_SYNC)
     {
+        return;
+    }
+    cluster_op_t *bh_ops_local[32], **bh_ops = bh_ops_local;
+    int bh_op_count = 0, bh_op_max = 32;
     while (next)
     {
         auto n2 = next->next;
-            if (next->opcode == OSD_OP_SYNC && (!(flags & OP_IMMEDIATE_COMMIT) || enable_writeback) ||
+        if (opcode == OSD_OP_WRITE
+            ? (next->opcode == OSD_OP_SYNC && (!(flags & OP_IMMEDIATE_COMMIT) || enable_writeback) ||
                 next->opcode == OSD_OP_WRITE && (flags & OP_FLUSH_BUFFER) && !(next->flags & OP_FLUSH_BUFFER))
+            : (next->opcode == OSD_OP_SYNC || next->opcode == OSD_OP_WRITE))
         {
             next->prev_wait += inc;
             assert(next->prev_wait >= 0);
             if (!next->prev_wait)
             {
+                // Kind of std::vector with local "small vector optimisation"
+                if (bh_op_count >= bh_op_max)
+                {
+                    bh_op_max *= 2;
+                    cluster_op_t **n = (cluster_op_t**)malloc_or_die(sizeof(cluster_op_t*) * bh_op_max);
+                    memcpy(n, bh_ops, sizeof(cluster_op_t*) * bh_op_count);
+                    bh_ops = n;
+                }
+                bh_ops[bh_op_count++] = next;
+            }
+        }
+        next = n2;
+    }
+    for (int i = 0; i < bh_op_count; i++)
+    {
+        cluster_op_t *next = bh_ops[i];
         if (next->opcode == OSD_OP_SYNC)
             continue_sync(next);
         else
             continue_rw(next);
     }
-            }
+    if (bh_ops != bh_ops_local)
-            next = n2;
-        }
-    }
-    else if (opcode == OSD_OP_SYNC)
     {
-        while (next)
+        free(bh_ops);
-        {
-            auto n2 = next->next;
-            if (next->opcode == OSD_OP_SYNC || next->opcode == OSD_OP_WRITE)
-            {
-                next->prev_wait += inc;
-                assert(next->prev_wait >= 0);
-                if (!next->prev_wait)
-                {
-                    if (next->opcode == OSD_OP_SYNC)
-                        continue_sync(next);
-                    else
-                        continue_rw(next);
-                }
-            }
-            next = n2;
-        }
     }
 }
 

tests/test_heal.sh

@@ -7,7 +7,7 @@ if [[ "$SCHEME" = "ec" ]]; then
     PG_DATA_SIZE=${PG_DATA_SIZE:-2}
     PG_MINSIZE=${PG_MINSIZE:-3}
 fi
-OSD_COUNT=7
+OSD_COUNT=${OSD_COUNT:-7}
 PG_COUNT=32
 . `dirname $0`/run_3osds.sh
 check_qemu
@@ -29,7 +29,7 @@ kill_osds()
     kill -9 $OSD1_PID
     $ETCDCTL del /vitastor/osd/state/1
 
-    for i in 2 3 4 5 6 7; do
+    for i in $(seq 2 $OSD_COUNT); do
         sleep 15
         echo Killing OSD $i and starting OSD $((i-1))
         p=OSD${i}_PID
@@ -40,8 +40,8 @@ kill_osds()
     done
 
     sleep 5
-    echo Starting OSD 7
+    echo Starting OSD $OSD_COUNT
-    start_osd 7
+    start_osd $OSD_COUNT
 
     sleep 5
 }