(#530) Add test cases to exercise print methods for mini-allocator me…

…tapages

This commit adds new print function to drive the functions for printing
keyed / unkeyed metadata pages of the mini-allocator. Existing test cases
are enhanced as follows to eventually exercise these print methods.

- Add bunch of print functions in trunk.c to find out routing filter
  metadata pages and to invoke underlying print methods on such pages.

- Extend splinter_test:test_splinter_print_diags() to exercise the print
  function that will walk through routing filter metadata pages for one
  pivot key off of the trunk's root node and eventually callss
  mini_unkeyed_print().

- Add new test_splinter_verbose_print_diags test case to exercise the
  verbose print logging in trunk.c (for a small data set).

- Modularize code in btree_stress_test.c to carve out code that is used
  to build new unit-test case, test_btree_print_diags(). First, this
  invokes btree_print_tree(), to see outputs of packed BTree node.
  Then, invokes mini_keyed_print(), to print keyed mini-allocator's
  metadata pages.

src/btree.c

@@ -1135,15 +1135,6 @@ btree_addrs_share_extent(cache *cc, uint64 left_addr, uint64 right_addr)
       allocator_get_config(al), right_addr, left_addr);
 }
 
-static inline uint64
-btree_root_to_meta_addr(const btree_config *cfg,
-                        uint64              root_addr,
-                        uint64              meta_page_no)
-{
-   return root_addr + (meta_page_no + 1) * btree_page_size(cfg);
-}
-
-
 /*----------------------------------------------------------
  * Creating and destroying B-trees.
  *----------------------------------------------------------

src/btree_private.h

@@ -305,3 +305,11 @@ btree_get_child_addr(const btree_config *cfg,
 {
    return index_entry_child_addr(btree_get_index_entry(cfg, hdr, k));
 }
+
+static inline uint64
+btree_root_to_meta_addr(const btree_config *cfg,
+                        uint64              root_addr,
+                        uint64              meta_page_no)
+{
+   return root_addr + (meta_page_no + 1) * btree_page_size(cfg);
+}

src/mini_allocator.c

@@ -255,7 +255,6 @@ static uint64
 allocator_page_number(allocator *al, uint64 page_addr)
 {
    allocator_config *allocator_cfg = allocator_get_config(al);
-   debug_assert(allocator_valid_page_addr(al, page_addr));
    return ((page_addr / allocator_cfg->io_cfg->page_size));
 }
 
@@ -275,7 +274,6 @@ static uint64
 allocator_extent_number(allocator *al, uint64 page_addr)
 {
    allocator_config *allocator_cfg = allocator_get_config(al);
-   debug_assert(allocator_valid_page_addr(al, page_addr));
    return ((allocator_extent_base_addr(al, page_addr)
             / allocator_cfg->io_cfg->extent_size));
 }
@@ -381,7 +379,7 @@ static uint64
 mini_num_entries(page_handle *meta_page)
 {
    mini_meta_hdr *hdr = (mini_meta_hdr *)meta_page->data;
-   return hdr->num_entries;
+   return (uint64)hdr->num_entries;
 }
 
 /*

src/routing_filter.h

@@ -43,12 +43,12 @@ typedef struct routing_config {
 /*
  * -----------------------------------------------------------------------------
  * Routing Filter: Disk-resident structure, on pages of type PAGE_TYPE_TRUNK.
- * Stored in trunk nodes, and is a pointer to a routing filter.
+ * Stored in trunk nodes, and is a pointer to a routing filter's page.
  * -----------------------------------------------------------------------------
  */
 typedef struct ONDISK routing_filter {
-   uint64 addr;
-   uint64 meta_head;
+   uint64 addr;      // Address of page holding filter
+   uint64 meta_head; // Address of metadata page holding mini-allocator info
    uint32 num_fingerprints;
    uint32 num_unique;
    uint32 value_size;

src/trunk.c

@@ -54,8 +54,8 @@ static const int64 latency_histo_buckets[LATENCYHISTO_SIZE] = {
  * structures sized by these limits can fit within 4K byte pages.
  *
  * NOTE: The bundle and sub-bundle related limits below are used to size arrays
- * of structures in splinter_trunk_hdr{}; i.e. Splinter pages of type
- * PAGE_TYPE_TRUNK. So these constants do affect disk-resident structures.
+ * of structures in trunk_hdr{}; i.e. Splinter pages of type PAGE_TYPE_TRUNK.
+ * So these constants do affect disk-resident structures.
  */
 #define TRUNK_MAX_PIVOTS            (20)
 #define TRUNK_MAX_BUNDLES           (12)
@@ -105,6 +105,19 @@ static const int64 latency_histo_buckets[LATENCYHISTO_SIZE] = {
  * If verbose_logging_enabled is enabled in trunk_config, these functions print
  * to cfg->log_handle.
  */
+void
+trunk_enable_verbose_logging(trunk_handle *spl, platform_log_handle *log_handle)
+{
+   spl->cfg.verbose_logging_enabled = TRUE;
+   spl->cfg.log_handle              = log_handle;
+}
+
+void
+trunk_disable_verbose_logging(trunk_handle *spl)
+{
+   spl->cfg.verbose_logging_enabled = FALSE;
+   spl->cfg.log_handle              = NULL;
+}
 
 static inline bool
 trunk_verbose_logging_enabled(trunk_handle *spl)
@@ -143,15 +156,19 @@ trunk_close_log_stream_if_enabled(trunk_handle           *spl,
 #define trunk_log_stream_if_enabled(spl, _stream, message, ...)                \
    do {                                                                        \
       if (trunk_verbose_logging_enabled(spl)) {                                \
-         platform_log_stream(                                                  \
-            (_stream), "[%3lu] " message, platform_get_tid(), ##__VA_ARGS__);  \
+         platform_log_stream((_stream),                                        \
+                             "trunk_log():%d [%lu] " message,                  \
+                             __LINE__,                                         \
+                             platform_get_tid(),                               \
+                             ##__VA_ARGS__);                                   \
       }                                                                        \
    } while (0)
 
 #define trunk_default_log_if_enabled(spl, message, ...)                        \
    do {                                                                        \
       if (trunk_verbose_logging_enabled(spl)) {                                \
-         platform_default_log(message, __VA_ARGS__);                           \
+         platform_default_log(                                                 \
+            "trunk_log():%d " message, __LINE__, __VA_ARGS__);                 \
       }                                                                        \
    } while (0)
 
@@ -355,7 +372,7 @@ trunk_log_node_if_enabled(platform_stream_handle *stream,
  *       Array of bundles
  *          When a collection of branches are flushed into a node, they are
  *          organized into a bundle. This bundle will be compacted into a
- *          single branch by a call to trunk_compact_bundle. Bundles are
+ *          single branch by a call to trunk_compact_bundle(). Bundles are
  *          implemented as a collection of subbundles, each of which covers a
  *          range of branches.
  *       ----------
@@ -2227,7 +2244,7 @@ trunk_leaf_rebundle_all_branches(trunk_handle *spl,
       routing_filter   *filter = trunk_subbundle_filter(spl, node, sb, 0);
       trunk_pivot_data *pdata  = trunk_get_pivot_data(spl, node, 0);
       *filter                  = pdata->filter;
-      debug_assert(filter->addr != 0);
+      debug_assert((filter->addr != 0), "addr=%lu\n", filter->addr);
       ZERO_STRUCT(pdata->filter);
       debug_assert(trunk_subbundle_branch_count(spl, node, sb) != 0);
    }
@@ -8180,6 +8197,62 @@ trunk_print(platform_log_handle *log_handle, trunk_handle *spl)
    trunk_print_subtree(log_handle, spl, spl->root_addr);
 }
 
+/*
+ * Print meta-page's linked list for one routing filter at address 'meta_head'.
+ */
+void
+trunk_print_filter_metapage_list(platform_log_handle *log_handle,
+                                 trunk_handle        *spl,
+                                 uint64               meta_head)
+{
+   platform_log(log_handle,
+                "\nFilter Metadata page starting from meta_head=%lu\n{\n",
+                meta_head);
+   mini_unkeyed_print(spl->cc, meta_head, PAGE_TYPE_FILTER);
+   platform_log(log_handle, "\n}\n");
+}
+
+void
+trunk_print_one_pivots_filter_metapages(platform_log_handle *log_handle,
+                                        trunk_handle        *spl,
+                                        trunk_node          *node,
+                                        uint16               pivot_no)
+{
+   trunk_pivot_data *pdata = trunk_get_pivot_data(spl, node, pivot_no);
+
+   // Last pivot won't have any filter metadata pages for it.
+   if (pivot_no == (trunk_num_pivot_keys(spl, node) - 1)) {
+      return;
+   }
+   trunk_print_filter_metapage_list(log_handle, spl, pdata->filter.meta_head);
+}
+
+/* Print filter's metadata pages for given node at address 'node_addr' */
+void
+trunk_print_nodes_filter_metapages(platform_log_handle *log_handle,
+                                   trunk_handle        *spl,
+                                   uint64               node_addr)
+{
+   trunk_node node;
+   trunk_node_get(spl->cc, node_addr, &node);
+
+   for (uint16 pivot_no = 0; pivot_no < trunk_num_pivot_keys(spl, &node);
+        pivot_no++)
+   {
+      trunk_print_one_pivots_filter_metapages(log_handle, spl, &node, pivot_no);
+   }
+
+   trunk_node_unget(spl->cc, &node);
+}
+
+void
+trunk_print_root_nodes_filter_metapages(platform_log_handle *log_handle,
+                                        trunk_handle        *spl)
+{
+   trunk_print_nodes_filter_metapages(log_handle, spl, spl->root_addr);
+}
+
+
 /*
  * trunk_print_super_block()
  *

src/trunk.h

@@ -413,6 +413,17 @@ trunk_print_space_use(platform_log_handle *log_handle, trunk_handle *spl);
 bool
 trunk_verify_tree(trunk_handle *spl);
 
+void
+trunk_print_root_nodes_filter_metapages(platform_log_handle *log_handle,
+                                        trunk_handle        *spl);
+
+void
+trunk_enable_verbose_logging(trunk_handle        *spl,
+                             platform_log_handle *log_handle);
+
+void
+trunk_disable_verbose_logging(trunk_handle *spl);
+
 static inline uint64
 trunk_max_key_size(trunk_handle *spl)
 {