K/V control prints (for debug only) O:-)

src/kv_db.cpp

@@ -781,11 +781,13 @@ static void invalidate(kv_db_t *db, uint64_t offset, uint64_t version)
                 // because otherwise, if we keep `updating` flag more than just during
                 // the write_block() operation, we may end up modifying an outdated
                 // version of the block in memory
+                printf("mark invalidated %lu\n", b_it->first);
                 b_it->second.invalidated = true;
                 b_it++;
             }
             else
             {
+                printf("drop %lu\n", b_it->first);
                 auto blk = &b_it->second;
                 del_block_level(db, blk);
                 db->block_cache.erase(b_it++);
@@ -891,6 +893,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
     op->offset = offset;
     op->len = db->kv_block_size;
     op->iov.push_back(malloc_or_die(op->len), op->len);
+    printf("reading %lu\n", op->offset);
     op->callback = [=](cluster_op_t *op)
     {
         if (op->retval != op->len)
@@ -905,6 +908,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
             // read may start BEFORE update and end AFTER update, in this case known will be > returned version
             (db->known_versions[op->offset/db->ino_block_size] >= op->version && !blk_it->second.invalidated || blk_it->second.updating > 0))
         {
+            printf("actual %lu v%lu (known v%lu)\n", op->offset, op->version, db->known_versions[op->offset/db->ino_block_size]);
             auto blk = &db->block_cache.at(op->offset);
             if (blk->updating > 0 && recheck_policy == KV_RECHECK_WAIT)
             {
@@ -927,6 +931,7 @@ static void get_block(kv_db_t *db, uint64_t offset, int cur_level, int recheck_p
                 del_block_level(db, blk);
                 *blk = {};
             }
+            printf("parse %lu v%lu\n", op->offset, op->version);
             int err = blk->parse(op->offset, (uint8_t*)op->iov.buf[0].iov_base, op->len);
             if (err == 0)
             {
@@ -1105,10 +1110,14 @@ int kv_op_t::handle_block(int res, int refresh, bool stop_on_split)
     if (stop_on_split && (blk->type == KV_LEAF_SPLIT || blk->type == KV_INT_SPLIT) &&
         (prev_key_lt == "" || prev_key_lt > blk->right_half))
     {
+        printf("find %s: %lu r%d t%d %s..%s..%s v%lu -> echild\n", key.c_str(), blk->offset, refresh, blk->type,
+            blk->key_ge.c_str(), blk->right_half.c_str(), blk->key_lt.c_str(), path.size() > 0 ? path[path.size()-1].version : 0);
         return -ECHILD;
     }
     else if ((blk->type == KV_INT_SPLIT || blk->type == KV_LEAF_SPLIT) && key >= blk->right_half)
     {
+        printf("find %s: %lu r%d t%d %s..%s..%s v%lu -> right %lu\n", key.c_str(), blk->offset, refresh, blk->type,
+            blk->key_ge.c_str(), blk->right_half.c_str(), blk->key_lt.c_str(), path.size() > 0 ? path[path.size()-1].version : 0, blk->right_half_block);
         cur_block = blk->right_half_block;
         if (opcode != KV_GET && opcode != KV_GET_CACHED)
         {
@@ -1122,6 +1131,8 @@ int kv_op_t::handle_block(int res, int refresh, bool stop_on_split)
     }
     else if (blk->type == KV_LEAF || blk->type == KV_LEAF_SPLIT)
     {
+        printf("find %s: %lu r%d t%d %s..%s..%s v%lu -> leaf\n", key.c_str(), blk->offset, refresh, blk->type,
+            blk->key_ge.c_str(), blk->right_half.c_str(), blk->key_lt.c_str(), path.size() > 0 ? path[path.size()-1].version : 0);
         return 0;
     }
     else
@@ -1151,6 +1162,9 @@ int kv_op_t::handle_block(int res, int refresh, bool stop_on_split)
         {
             path.push_back((kv_path_t){ .offset = cur_block });
         }
+        printf("find %s: %lu r%d t%d %s..%s..%s v%lu -> down %s..%s %lu\n", key.c_str(), blk->offset, refresh, blk->type,
+            blk->key_ge.c_str(), blk->right_half.c_str(), blk->key_lt.c_str(),
+            path.size() > 0 ? path[path.size()-2].version : 0, prev_key_ge.c_str(), prev_key_lt.c_str(), cur_block);
         return -EAGAIN;
     }
     return 0;
@@ -1227,6 +1241,7 @@ static void write_block(kv_db_t *db, kv_block_t *blk, std::function<void(int)> c
         int res = op->retval == op->len ? 0 : (op->retval > 0 ? -EIO : op->retval);
         if (res == 0)
         {
+            printf("written %lu v%lu\n", blk->offset, op->version);
             blk->invalidated = false;
             db->known_versions[blk->offset/db->ino_block_size] = op->version;
             auto b_it = db->continue_write.find(blk->offset/db->ino_block_size);
@@ -1308,12 +1323,14 @@ static void place_again(kv_db_t *db, kv_block_t *blk, std::function<void(int, kv
 
 static void write_new_block(kv_db_t *db, kv_block_t *blk, std::function<void(int, kv_block_t *)> cb)
 {
+    printf("write_new_block %lu %s..%s\n", blk->offset, blk->key_ge.c_str(), blk->key_lt.c_str());
     write_block(db, blk, [=](int res)
     {
         db->stop_writing_new(blk->offset);
         if (res == -EINTR)
         {
             // CAS failure => re-read, then, if not zero, find position again and retry
+            printf("CAS write_new_block %lu\n", blk->offset);
             if (!(blk->offset % db->ino_block_size))
             {
                 // Any failed write of the first block within an inode block
@@ -1373,6 +1390,7 @@ static void write_new_block(kv_db_t *db, kv_block_t *blk, std::function<void(int
                 // confirms that it's now "locked" by us
                 db->confirm_allocation_block(blk->offset);
             }
+            printf("OK write_new_block %lu\n", blk->offset);
             cb(0, blk);
         }
     });
@@ -1486,10 +1504,12 @@ void kv_op_t::create_root()
     blk->set_data_size();
     add_block_level(db, blk);
     blk->updating++;
+    printf("create_root\n");
     write_block(db, blk, [=](int res)
     {
         if (res == -EINTR)
         {
+            printf("create_root CAS\n");
             db->clear_allocation_block(blk->offset);
             auto blk_offset = blk->offset;
             del_block_level(db, blk);
@@ -1499,6 +1519,7 @@ void kv_op_t::create_root()
         }
         else
         {
+            printf("create_root OK\n");
             db->stop_writing_new(blk->offset);
             db->confirm_allocation_block(blk->offset);
             db->stop_updating(blk);
@@ -1521,6 +1542,7 @@ void kv_op_t::resume_split()
         return;
     }
     auto blk = &db->block_cache.at(cur_block);
+    printf("resume_split block %lu to %lu - key %s\n", blk->offset, path[path.size()-2].offset, blk->right_half.c_str());
     update_block(
         path.size()-2, false, blk->right_half,
         std::string((char*)&blk->right_half_block, sizeof(blk->right_half_block)),
@@ -1549,6 +1571,7 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
     }
     auto blk = &blk_it->second;
     auto block_ver = path[path_pos].version;
+    printf("update_block %lu u=%d exp/act %lu/%lu\n", blk->offset, blk->updating, db->known_versions[blk->offset/db->ino_block_size], block_ver);
     if (blk->updating)
     {
         // Wait if block is being modified
@@ -1635,6 +1658,7 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
         {
             if (res < 0)
             {
+                printf("write_simple CAS %lu %s..%s +%s\n", blk->offset, blk->key_ge.c_str(), blk->key_lt.c_str(), key.c_str());
                 auto blk_offset = blk->offset;
                 del_block_level(db, blk);
                 db->block_cache.erase(blk_offset);
@@ -1642,6 +1666,7 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
             }
             else
             {
+                printf("write_simple OK %lu %s..%s +%s\n", blk->offset, blk->key_ge.c_str(), blk->key_lt.c_str(), key.c_str());
                 blk->apply_change();
                 db->stop_updating(blk);
             }
@@ -1707,6 +1732,7 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
                 {
                     new_root->invalidated = true;
                 }
+                printf("split_root %lu\n", blk->offset);
                 write_block(db, new_root, [=](int write_res)
                 {
                     if (write_res < 0)
@@ -1763,6 +1789,7 @@ void kv_op_t::update_block(int path_pos, bool is_delete, const std::string & key
                 blk->change_key = key;
                 blk->change_value = value;
             }
+            printf("split_nonroot %lu %s..%s..%s\n", blk->offset, blk->key_ge.c_str(), separator.c_str(), blk->key_lt.c_str());
             write_block(db, blk, [=](int write_res)
             {
                 if (write_res < 0)