[t:4552], make unpinning of node cheaper for reads

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

newbrt/brt-cachetable-wrappers.c

@@ -252,3 +252,16 @@ toku_unpin_brtnode(BRT brt, BRTNODE node)
     toku_unpin_brtnode_off_client_thread(brt->h, node);
 }
 
+void
+toku_unpin_brtnode_read_only(BRT brt, BRTNODE node)
+{
+    int r = toku_cachetable_unpin(
+        brt->h->cf,
+        node->thisnodename,
+        node->fullhash,
+        (enum cachetable_dirty) node->dirty,
+        make_invalid_pair_attr()
+        );
+    assert(r==0);
+}
+

newbrt/brt-cachetable-wrappers.h

@@ -122,6 +122,9 @@ toku_unpin_brtnode_off_client_thread(struct brt_header* h, BRTNODE node);
 void
 toku_unpin_brtnode(BRT brt, BRTNODE node);
 
+void
+toku_unpin_brtnode_read_only(BRT brt, BRTNODE node);
+
 C_END
 
 #endif

newbrt/brt-internal.h

@@ -438,6 +438,7 @@ struct brt {
 // FIXME needs toku prefix
 long brtnode_memory_size (BRTNODE node);
 PAIR_ATTR make_brtnode_pair_attr(BRTNODE node);
+PAIR_ATTR make_invalid_pair_attr(void);
 
 /* serialization code */
 void

newbrt/brt.c

@@ -499,11 +499,23 @@ PAIR_ATTR make_brtnode_pair_attr(BRTNODE node) {
      .nonleaf_size = (node->height > 0) ? size : 0, 
      .leaf_size = (node->height > 0) ? 0 : size, 
      .rollback_size = 0, 
-     .cache_pressure_size = cachepressure_size
+     .cache_pressure_size = cachepressure_size,
+     .is_valid = TRUE
     }; 
     return result; 
 }
 
+PAIR_ATTR make_invalid_pair_attr(void) {
+    PAIR_ATTR result={
+     .size = 0, 
+     .nonleaf_size = 0, 
+     .leaf_size = 0, 
+     .rollback_size = 0, 
+     .cache_pressure_size = 0,
+     .is_valid = FALSE
+    }; 
+    return result; 
+}
 
 
 // assign unique dictionary id
@@ -5227,7 +5239,7 @@ brt_search_child(BRT brt, BRTNODE node, int childnum, brt_search_t *search, BRT_
 #endif
 
         assert(next_unlockers.locked);
-        toku_unpin_brtnode(brt, childnode); // unpin the childnode before handling the reactive child (because that may make the childnode disappear.)
+        toku_unpin_brtnode_read_only(brt, childnode); // unpin the childnode before handling the reactive child (because that may make the childnode disappear.)
     } else {
         // try again.
 
@@ -5238,7 +5250,7 @@ brt_search_child(BRT brt, BRTNODE node, int childnum, brt_search_t *search, BRT_
         //  some piece of a node that it needed was not in memory. In this case,
         //  the node was not unpinned, so we unpin it here
         if (next_unlockers.locked) {
-            toku_unpin_brtnode(brt, childnode);
+            toku_unpin_brtnode_read_only(brt, childnode);
         }
     }
 
@@ -5554,7 +5566,7 @@ try_again:
             //  some piece of a node that it needed was not in memory. 
             //  In this case, the node was not unpinned, so we unpin it here
             if (unlockers.locked) {
-                toku_unpin_brtnode(brt, node);
+                toku_unpin_brtnode_read_only(brt, node);
             }
 	    goto try_again;
 	} else {
@@ -5563,7 +5575,7 @@ try_again:
     }
 
     assert(unlockers.locked);
-    toku_unpin_brtnode(brt, node);
+    toku_unpin_brtnode_read_only(brt, node);
 
 
     //Heaviside function (+direction) queries define only a lower or upper
@@ -6060,7 +6072,7 @@ toku_brt_keyrange_internal (BRT brt, BRTNODE node,
 		*greater += rows_per_child * (node->n_children - child_number - 1);
 
 		assert(unlockers->locked);
-		toku_unpin_brtnode(brt, childnode);
+		toku_unpin_brtnode_read_only(brt, childnode);
 	    }
 	}
     }
@@ -6119,7 +6131,7 @@ toku_brt_keyrange (BRT brt, DBT *key, u_int64_t *less_p, u_int64_t *equal_p, u_i
 	    }
 	}
 	assert(unlockers.locked);
-	toku_unpin_brtnode(brt, node);
+	toku_unpin_brtnode_read_only(brt, node);
 	*less_p    = less;
 	*equal_p   = equal;
 	*greater_p = greater;

newbrt/brttypes.h

@@ -19,6 +19,16 @@
 extern "C" {
 #endif
 
+#include <stdbool.h>
+#ifndef TRUE
+// In the future, use the stdbool bool and constants (true false), rather than BOOL, TRUE, and FALSE.
+#define TRUE true
+#define FALSE false
+typedef bool BOOL;
+#endif
+
+
+
 typedef struct brt *BRT;
 typedef struct brtnode *BRTNODE;
 typedef struct brtnode_leaf_basement_node *BASEMENTNODE;
@@ -56,6 +66,7 @@ typedef struct pair_attr_s {
     long leaf_size; // size if PAIR is a leaf node, 0 otherwise, used only for engine status
     long rollback_size; // size of PAIR is a rollback node, 0 otherwise, used only for engine status
     long cache_pressure_size; // amount PAIR contributes to cache pressure, is sum of buffer sizes and workdone counts
+    BOOL is_valid;
 } PAIR_ATTR;
 
 static inline PAIR_ATTR make_pair_attr(long size) { 
@@ -65,10 +76,11 @@ static inline PAIR_ATTR make_pair_attr(long size) {
         .nonleaf_size = 0, 
         .leaf_size = 0, 
         .rollback_size = 0, 
-        .cache_pressure_size = 0 
+        .cache_pressure_size = 0,
+        .is_valid = TRUE
     }; 
 #else
-    PAIR_ATTR result = {size, 0, 0, 0, 0};
+    PAIR_ATTR result = {size, 0, 0, 0, 0, TRUE};
 #endif
     return result; 
 }
@@ -107,14 +119,6 @@ typedef struct {
     FILENUM  *filenums;
 } FILENUMS;
 
-#include <stdbool.h>
-#ifndef TRUE
-// In the future, use the stdbool bool and constants (true false), rather than BOOL, TRUE, and FALSE.
-#define TRUE true
-#define FALSE false
-typedef bool BOOL;
-#endif
-
 typedef struct tokulogger *TOKULOGGER;
 #define NULL_LOGGER ((TOKULOGGER)0)
 typedef struct tokutxn    *TOKUTXN;

newbrt/cachetable.c

@@ -172,7 +172,8 @@ static PAIR_ATTR const zero_attr = {
     .nonleaf_size = 0, 
     .leaf_size = 0, 
     .rollback_size = 0, 
-    .cache_pressure_size = 0
+    .cache_pressure_size = 0,
+    .is_valid = TRUE
 };
 
 static void maybe_flush_some (CACHETABLE ct, long size, BOOL ct_locked);
@@ -1276,6 +1277,7 @@ pending_pairs_remove (CACHETABLE ct, PAIR p) {
 
 static void
 cachetable_remove_pair_attr (CACHETABLE ct, PAIR_ATTR attr) {
+    assert(attr.is_valid);
     ct->size_current -= attr.size;
     ct->size_nonleaf -= attr.nonleaf_size;
     ct->size_leaf -= attr.leaf_size;
@@ -1286,6 +1288,7 @@ cachetable_remove_pair_attr (CACHETABLE ct, PAIR_ATTR attr) {
 
 static void
 cachetable_add_pair_attr(CACHETABLE ct, PAIR_ATTR attr) {
+    assert(attr.is_valid);
     ct->size_current += attr.size;
     if (ct->size_current > ct->size_max) {
         ct->size_max = ct->size_current;
@@ -2432,10 +2435,12 @@ cachetable_unpin_internal(CACHEFILE cachefile, CACHEKEY key, u_int32_t fullhash,
             assert(!p->cq);
             nb_mutex_write_unlock(&p->nb_mutex);
 	    if (dirty) p->dirty = CACHETABLE_DIRTY;
+            if (attr.is_valid) {
                 PAIR_ATTR old_attr = p->attr;
                 PAIR_ATTR new_attr = attr;
                 cachetable_change_pair_attr(ct, old_attr, new_attr);
                 p->attr = attr;
+            }
 	    WHEN_TRACE_CT(printf("[count=%lld]\n", p->pinned));
 	    {
                 if (flush) {

newbrt/rollback.c

@@ -460,7 +460,8 @@ static inline PAIR_ATTR make_rollback_pair_attr(long size) {
      .nonleaf_size = 0, 
      .leaf_size = 0, 
      .rollback_size = size, 
-     .cache_pressure_size = 0 
+     .cache_pressure_size = 0,
+     .is_valid = TRUE
     }; 
     return result; 
 }

newbrt/tests/cachetable-cleaner-thread-attrs-accumulate.c

@@ -14,14 +14,14 @@ toku_pthread_mutex_t attr_mutex;
 #define STATUS_VALUE(x) ct_status.status[x].value.num
 
 const PAIR_ATTR attrs[] = {
-    { .size = 20, .nonleaf_size = 13, .leaf_size = 900, .rollback_size = 123, .cache_pressure_size = 403 },
-    { .size = 21, .nonleaf_size = 16, .leaf_size = 910, .rollback_size = 113, .cache_pressure_size = 401 },
-    { .size = 22, .nonleaf_size = 17, .leaf_size = 940, .rollback_size = 133, .cache_pressure_size = 402 },
-    { .size = 23, .nonleaf_size = 18, .leaf_size = 931, .rollback_size = 153, .cache_pressure_size = 404 },
-    { .size = 25, .nonleaf_size = 19, .leaf_size = 903, .rollback_size = 173, .cache_pressure_size = 413 },
-    { .size = 26, .nonleaf_size = 10, .leaf_size = 903, .rollback_size = 193, .cache_pressure_size = 423 },
-    { .size = 20, .nonleaf_size = 11, .leaf_size = 902, .rollback_size = 103, .cache_pressure_size = 433 },
-    { .size = 29, .nonleaf_size = 12, .leaf_size = 909, .rollback_size = 113, .cache_pressure_size = 443 }
+    { .size = 20, .nonleaf_size = 13, .leaf_size = 900, .rollback_size = 123, .cache_pressure_size = 403, .is_valid = TRUE },
+    { .size = 21, .nonleaf_size = 16, .leaf_size = 910, .rollback_size = 113, .cache_pressure_size = 401, .is_valid = TRUE },
+    { .size = 22, .nonleaf_size = 17, .leaf_size = 940, .rollback_size = 133, .cache_pressure_size = 402, .is_valid = TRUE },
+    { .size = 23, .nonleaf_size = 18, .leaf_size = 931, .rollback_size = 153, .cache_pressure_size = 404, .is_valid = TRUE },
+    { .size = 25, .nonleaf_size = 19, .leaf_size = 903, .rollback_size = 173, .cache_pressure_size = 413, .is_valid = TRUE },
+    { .size = 26, .nonleaf_size = 10, .leaf_size = 903, .rollback_size = 193, .cache_pressure_size = 423, .is_valid = TRUE },
+    { .size = 20, .nonleaf_size = 11, .leaf_size = 902, .rollback_size = 103, .cache_pressure_size = 433, .is_valid = TRUE },
+    { .size = 29, .nonleaf_size = 12, .leaf_size = 909, .rollback_size = 113, .cache_pressure_size = 443, .is_valid = TRUE }
 };
 const int n_pairs = (sizeof attrs) / (sizeof attrs[0]);