refs #5367 Fix verify to work properly with marked messages

First it evaluates using a write lock on a node (before moving messages)
and then doing it again after moving messages
Does appropriate checks before and after

git-svn-id: file:///svn/toku/tokudb@46949 c7de825b-a66e-492c-adef-691d508d4ae1

ft/ft-cachetable-wrappers.cc

@@ -204,7 +204,7 @@ toku_pin_ftnode_batched(
 }
 
 void
-toku_pin_ftnode_off_client_thread(
+toku_pin_ftnode_off_client_thread_and_maybe_move_messages(
     FT h,
     BLOCKNUM blocknum,
     uint32_t fullhash,
@@ -212,10 +212,11 @@ toku_pin_ftnode_off_client_thread(
     bool may_modify_node,
     uint32_t num_dependent_nodes,
     FTNODE* dependent_nodes,
-    FTNODE *node_p)
+    FTNODE *node_p,
+    bool move_messages)
 {
     toku_cachetable_begin_batched_pin(h->cf);
-    toku_pin_ftnode_off_client_thread_batched(
+    toku_pin_ftnode_off_client_thread_batched_and_maybe_move_messages(
         h,
         blocknum,
         fullhash,
@@ -223,13 +224,14 @@ toku_pin_ftnode_off_client_thread(
         may_modify_node,
         num_dependent_nodes,
         dependent_nodes,
-        node_p
+        node_p,
+        move_messages
         );
     toku_cachetable_end_batched_pin(h->cf);
 }
 
 void
-toku_pin_ftnode_off_client_thread_batched(
+toku_pin_ftnode_off_client_thread(
     FT h,
     BLOCKNUM blocknum,
     uint32_t fullhash,
@@ -238,6 +240,22 @@ toku_pin_ftnode_off_client_thread_batched(
     uint32_t num_dependent_nodes,
     FTNODE* dependent_nodes,
     FTNODE *node_p)
+{
+    toku_pin_ftnode_off_client_thread_and_maybe_move_messages(
+            h, blocknum, fullhash, bfe, may_modify_node, num_dependent_nodes, dependent_nodes, node_p, true);
+}
+
+void
+toku_pin_ftnode_off_client_thread_batched_and_maybe_move_messages(
+    FT h,
+    BLOCKNUM blocknum,
+    uint32_t fullhash,
+    FTNODE_FETCH_EXTRA bfe,
+    bool may_modify_node,
+    uint32_t num_dependent_nodes,
+    FTNODE* dependent_nodes,
+    FTNODE *node_p,
+    bool move_messages)
 {
     void *node_v;
     CACHEFILE dependent_cf[num_dependent_nodes];
@@ -271,12 +289,27 @@ toku_pin_ftnode_off_client_thread_batched(
         );
     assert(r==0);
     FTNODE node = (FTNODE) node_v;
-    if (may_modify_node && node->height > 0) {
+    if (may_modify_node && node->height > 0 && move_messages) {
         toku_move_ftnode_messages_to_stale(h, node);
     }
     *node_p = node;
 }
 
+void
+toku_pin_ftnode_off_client_thread_batched(
+    FT h,
+    BLOCKNUM blocknum,
+    uint32_t fullhash,
+    FTNODE_FETCH_EXTRA bfe,
+    bool may_modify_node,
+    uint32_t num_dependent_nodes,
+    FTNODE* dependent_nodes,
+    FTNODE *node_p)
+{
+    toku_pin_ftnode_off_client_thread_batched_and_maybe_move_messages(
+            h, blocknum, fullhash, bfe, may_modify_node, num_dependent_nodes, dependent_nodes, node_p, true);
+}
+
 int toku_maybe_pin_ftnode_clean(FT ft, BLOCKNUM blocknum, uint32_t fullhash, FTNODE *nodep, bool may_modify_node) {
     void *node_v;
     int r = toku_cachetable_maybe_get_and_pin_clean(ft->cf, blocknum, fullhash, &node_v);

ft/ft-cachetable-wrappers.h

@@ -114,12 +114,42 @@ toku_pin_ftnode_off_client_thread(
     FTNODE *node_p
     );
 
+void
+toku_pin_ftnode_off_client_thread_and_maybe_move_messages(
+    FT h,
+    BLOCKNUM blocknum,
+    uint32_t fullhash,
+    FTNODE_FETCH_EXTRA bfe,
+    bool may_modify_node,
+    uint32_t num_dependent_nodes,
+    FTNODE* dependent_nodes,
+    FTNODE *node_p,
+    bool move_messages
+    );
+
 /**
  * This function may return a pinned ftnode to the caller, if pinning is cheap.
  * If the node is already locked, or is pending a checkpoint, the node is not pinned and -1 is returned.
  */
 int toku_maybe_pin_ftnode_clean(FT ft, BLOCKNUM blocknum, uint32_t fullhash, FTNODE *nodep, bool may_modify_node);
 
+/**
+ * Batched version of toku_pin_ftnode_off_client_thread, see cachetable
+ * batched API for more details.
+ */
+void
+toku_pin_ftnode_off_client_thread_batched_and_maybe_move_messages(
+    FT h,
+    BLOCKNUM blocknum,
+    uint32_t fullhash,
+    FTNODE_FETCH_EXTRA bfe,
+    bool may_modify_node,
+    uint32_t num_dependent_nodes,
+    FTNODE* dependent_nodes,
+    FTNODE *node_p,
+    bool move_messages
+    );
+
 /**
  * Batched version of toku_pin_ftnode_off_client_thread, see cachetable
  * batched API for more details.