[t:4887] move stats changing stuff to brt_header.h and make the loader update stats

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

newbrt/brt-internal.h

@@ -50,13 +50,6 @@ enum { BUFFER_HEADER_SIZE = (4 // height//
 			     + TREE_FANOUT * 8 // children
 			     ) };
 
-typedef struct {
-    int64_t numrows;           // delta versions in basements could be negative
-    int64_t numbytes;
-} STAT64INFO_S, *STAT64INFO;
-
-static const STAT64INFO_S ZEROSTATS = {0,0};
-
 //
 // Field in brtnode_fetch_extra that tells the 
 // partial fetch callback what piece of the node

newbrt/brt.c

@@ -579,18 +579,6 @@ toku_get_and_clear_basement_stats(BRTNODE leafnode) {
     return deltas;
 }
 
-static void
-update_header_stats(STAT64INFO headerstats, STAT64INFO delta) {
-    (void) __sync_fetch_and_add(&(headerstats->numrows),  delta->numrows);
-    (void) __sync_fetch_and_add(&(headerstats->numbytes), delta->numbytes);
-}
-
-static void
-decrease_header_stats(STAT64INFO headerstats, STAT64INFO delta) {
-    (void) __sync_fetch_and_sub(&(headerstats->numrows),  delta->numrows);
-    (void) __sync_fetch_and_sub(&(headerstats->numbytes), delta->numbytes);
-}
-
 // This is the ONLY place where a node is marked as dirty, other than toku_initialize_empty_brtnode().
 void
 toku_mark_node_dirty(BRTNODE node) {
@@ -635,9 +623,9 @@ static void brtnode_update_disk_stats(
     STAT64INFO_S deltas = ZEROSTATS;
     // capture deltas before rebalancing basements for serialization
     deltas = toku_get_and_clear_basement_stats(brtnode);
-    update_header_stats(&(h->on_disk_stats), &deltas);
+    toku_brt_header_update_stats(&h->on_disk_stats, deltas);
     if (for_checkpoint) {
-        update_header_stats(&(h->checkpoint_staging_stats), &deltas);
+        toku_brt_header_update_stats(&h->checkpoint_staging_stats, deltas);
     }
 }
 
@@ -760,7 +748,7 @@ void toku_brtnode_flush_callback (
                 for (int i = 0; i < brtnode->n_children; i++) {
                     if (BP_STATE(brtnode,i) == PT_AVAIL) {
                         BASEMENTNODE bn = BLB(brtnode, i);
-                        decrease_header_stats(&h->in_memory_stats, &bn->stat64_delta);
+                        toku_brt_header_decrease_stats(&h->in_memory_stats, bn->stat64_delta);
                     }
                 }
             }
@@ -890,7 +878,7 @@ void toku_evict_bn_from_memory(BRTNODE node, int childnum, struct brt_header* h)
     // free the basement node
     assert(!node->dirty);
     BASEMENTNODE bn = BLB(node, childnum);
-    decrease_header_stats(&h->in_memory_stats, &bn->stat64_delta);
+    toku_brt_header_decrease_stats(&h->in_memory_stats, bn->stat64_delta);
     struct mempool * mp = &bn->buffer_mempool;
     toku_mempool_destroy(mp);
     destroy_basement_node(bn);
@@ -2229,7 +2217,7 @@ brt_leaf_gc_all_les(BRTNODE node,
         delta.numrows = 0;
         delta.numbytes = 0;
         basement_node_gc_all_les(bn, snapshot_xids, live_list_reverse, live_root_txns, &delta);
-        update_header_stats(&(h->in_memory_stats), &(delta));
+        toku_brt_header_update_stats(&h->in_memory_stats, delta);
     }
 }
 
@@ -2260,7 +2248,7 @@ toku_bnc_flush_to_child(
                 );
         }));
     if (stats_delta.numbytes || stats_delta.numrows) {
-        update_header_stats(&h->in_memory_stats, &stats_delta);
+        toku_brt_header_update_stats(&h->in_memory_stats, stats_delta);
     }
     // Run garbage collection, if we are a leaf entry.
     TOKULOGGER logger = toku_cachefile_logger(h->cf);
@@ -2457,7 +2445,7 @@ static void push_something_at_root (struct brt_header *h, BRTNODE *nodep, BRT_MS
         &stats_delta
         );
     if (stats_delta.numbytes || stats_delta.numrows) {
-        update_header_stats(&h->in_memory_stats, &stats_delta);
+        toku_brt_header_update_stats(&h->in_memory_stats, stats_delta);
     }
     //
     // assumption is that toku_brt_node_put_cmd will
@@ -4133,7 +4121,7 @@ bnc_apply_messages_to_basement_node(
     // update stats
     //
     if (stats_delta.numbytes || stats_delta.numrows) {
-        update_header_stats(&t->h->in_memory_stats, &stats_delta);
+        toku_brt_header_update_stats(&t->h->in_memory_stats, stats_delta);
     }
     // We can't delete things out of the fresh tree inside the above
     // procedures because we're still looking at the fresh tree.  Instead

newbrt/brt_header.c

@@ -969,4 +969,14 @@ toku_brtheader_update_cmp_descriptor(struct brt_header* h) {
         );
 }
 
+void
+toku_brt_header_update_stats(STAT64INFO headerstats, STAT64INFO_S delta) {
+    (void) __sync_fetch_and_add(&(headerstats->numrows),  delta.numrows);
+    (void) __sync_fetch_and_add(&(headerstats->numbytes), delta.numbytes);
+}
 
+void
+toku_brt_header_decrease_stats(STAT64INFO headerstats, STAT64INFO_S delta) {
+    (void) __sync_fetch_and_sub(&(headerstats->numrows),  delta.numrows);
+    (void) __sync_fetch_and_sub(&(headerstats->numbytes), delta.numbytes);
+}

newbrt/brt_header.h

@@ -67,5 +67,8 @@ int toku_update_descriptor(struct brt_header * h, DESCRIPTOR d, int fd);
 // Note: See the locking discussion in brt.c for toku_brt_change_descriptor and toku_update_descriptor.
 void toku_brtheader_update_cmp_descriptor(struct brt_header* h);
 
+void toku_brt_header_update_stats(STAT64INFO headerstats, STAT64INFO_S delta);
+void toku_brt_header_decrease_stats(STAT64INFO headerstats, STAT64INFO_S delta);
+
 
 #endif

newbrt/brtloader.c

@@ -2143,7 +2143,7 @@ static struct leaf_buf *start_leaf (struct dbout *out, const DESCRIPTOR UU(desc)
 
 static void finish_leafnode (struct dbout *out, struct leaf_buf *lbuf, int progress_allocation, BRTLOADER bl, uint32_t target_basementnodesize);
 static int write_nonleaves (BRTLOADER bl, FIDX pivots_fidx, struct dbout *out, struct subtrees_info *sts, const DESCRIPTOR descriptor, uint32_t target_nodesize, uint32_t target_basementnodesize);
-static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int keylen, unsigned char *val, int vallen, int this_leafentry_size);
+static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int keylen, unsigned char *val, int vallen, int this_leafentry_size, STAT64INFO stats_to_update);
 static int write_translation_table (struct dbout *out, long long *off_of_translation_p);
 static int write_header (struct dbout *out, long long translation_location_on_disk, long long translation_size_on_disk);
 
@@ -2269,6 +2269,7 @@ static int toku_loader_write_brt_from_q (BRTLOADER bl,
 
     DBT maxkey = make_dbt(0, 0); // keep track of the max key of the current node
 
+    STAT64INFO_S deltas = ZEROSTATS;
     while (result == 0) {
 	void *item;
 	{
@@ -2327,7 +2328,7 @@ static int toku_loader_write_brt_from_q (BRTLOADER bl,
 		lbuf = start_leaf(&out, descriptor, lblock, le_xid, target_nodesize);
 	    }
 
-	    add_pair_to_leafnode(lbuf, (unsigned char *) key.data, key.size, (unsigned char *) val.data, val.size, this_leafentry_size);
+	    add_pair_to_leafnode(lbuf, (unsigned char *) key.data, key.size, (unsigned char *) val.data, val.size, this_leafentry_size, &deltas);
 	    n_rows_remaining--;
 
             update_maxkey(&maxkey, &key); // set the new maxkey to the current key
@@ -2343,6 +2344,10 @@ static int toku_loader_write_brt_from_q (BRTLOADER bl,
             result = brt_loader_get_error(&bl->error_callback); // check if an error was posted and terminate this quickly
     }
 
+    if (deltas.numrows || deltas.numbytes) {
+        toku_brt_header_update_stats(&h.in_memory_stats, deltas);
+    }
+
     cleanup_maxkey(&maxkey);
 
     if (lbuf) {
@@ -2679,7 +2684,7 @@ int toku_brt_loader_get_error(BRTLOADER bl, int *error) {
     return 0;
 }
 
-static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int keylen, unsigned char *val, int vallen, int this_leafentry_size) {
+static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int keylen, unsigned char *val, int vallen, int this_leafentry_size, STAT64INFO stats_to_update) {
     lbuf->nkeys++; // assume NODUP
     lbuf->ndata++;
     lbuf->dsize += keylen + vallen;
@@ -2694,7 +2699,7 @@ static void add_pair_to_leafnode (struct leaf_buf *lbuf, unsigned char *key, int
     DBT theval = { .data = val, .size = vallen };
     BRT_MSG_S cmd = { BRT_INSERT, ZERO_MSN, lbuf->xids, .u.id = { &thekey, &theval } };
     uint64_t workdone=0;
-    toku_brt_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, &workdone, NULL);
+    toku_brt_bn_apply_cmd_once(BLB(leafnode,0), &cmd, idx, NULL, &workdone, stats_to_update);
 }
 
 static int write_literal(struct dbout *out, void*data,  size_t len) {

newbrt/brttypes.h

@@ -106,6 +106,13 @@ typedef struct __toku_msn { u_int64_t msn; } MSN;
 #define MIN_MSN  ((MSN){(u_int64_t)1 << 62})  // first 2^62 values reserved for messages created before Dr. No (for upgrade)
 #define MAX_MSN  ((MSN){UINT64_MAX})
 
+typedef struct {
+    int64_t numrows;           // delta versions in basements could be negative
+    int64_t numbytes;
+} STAT64INFO_S, *STAT64INFO;
+
+static const STAT64INFO_S ZEROSTATS = {0,0};
+
 /* At the brt layer, a FILENUM uniquely identifies an open file.
  * At the ydb layer, a DICTIONARY_ID uniquely identifies an open dictionary.
  * With the introduction of the loader (ticket 2216), it is possible for the file that holds