FT-309 Fix up tracing code, add a lock (to extend fprintf atomicity to

multiple calls), support create_from_blockpairs in the allocator and
trace tool

ft/serialize/block_allocator.cc

@@ -147,24 +147,21 @@ void block_allocator::_create_internal(uint64_t reserve_at_beginning, uint64_t a
     _n_bytes_in_use = reserve_at_beginning;
     _strategy = BA_STRATEGY_FIRST_FIT;
 
+    memset(&_trace_lock, 0, sizeof(toku_mutex_t));
+    toku_mutex_init(&_trace_lock, nullptr);
+
     VALIDATE();
 }
 
 void block_allocator::create(uint64_t reserve_at_beginning, uint64_t alignment) {
     _create_internal(reserve_at_beginning, alignment);
-    if (ba_trace_file != nullptr) {
-        fprintf(ba_trace_file, "ba_trace_create %p\n", this);
-        fflush(ba_trace_file);
-    }
+    _trace_create();
 }
 
 void block_allocator::destroy() {
     toku_free(_blocks_array);
-
-    if (ba_trace_file != nullptr) {
-        fprintf(ba_trace_file, "ba_trace_destroy %p\n", this);
-        fflush(ba_trace_file);
-    }
+    _trace_destroy();
+    toku_mutex_destroy(&_trace_lock);
 }
 
 void block_allocator::set_strategy(enum allocation_strategy strategy) {
@@ -205,6 +202,8 @@ void block_allocator::create_from_blockpairs(uint64_t reserve_at_beginning, uint
     }
 
     VALIDATE();
+
+    _trace_create_from_blockpairs();
 }
 
 // Effect: align a value by rounding up.
@@ -282,13 +281,7 @@ void block_allocator::alloc_block(uint64_t size, uint64_t heat, uint64_t *offset
     _n_blocks++;
     VALIDATE();
 
-    if (ba_trace_file != nullptr) {
-        fprintf(ba_trace_file, "ba_trace_alloc %p %lu %lu %lu\n",
-                this, static_cast<unsigned long>(size),
-                static_cast<unsigned long>(heat),
-                static_cast<unsigned long>(*offset));
-        fflush(ba_trace_file);
-    }
+    _trace_alloc(size, heat, *offset);
 }
 
 // Find the index in the blocks array that has a particular offset.  Requires that the block exist.
@@ -330,12 +323,8 @@ void block_allocator::free_block(uint64_t offset) {
             (_n_blocks - bn - 1) * sizeof(struct blockpair));
     _n_blocks--;
     VALIDATE();
-
-    if (ba_trace_file != nullptr) {
-        fprintf(ba_trace_file, "ba_trace_free %p %lu\n",
-                this, static_cast<unsigned long>(offset));
-        fflush(ba_trace_file);
-    }
+    
+    _trace_free(offset);
 }
 
 uint64_t block_allocator::block_size(uint64_t offset) {
@@ -462,3 +451,65 @@ void block_allocator::validate() const {
     }
     assert(n_bytes_in_use == _n_bytes_in_use);
 }
+
+// Tracing
+
+void block_allocator::_trace_create(void) {
+    if (ba_trace_file != nullptr) {
+        toku_mutex_lock(&_trace_lock);
+        fprintf(ba_trace_file, "ba_trace_create %p %lu %lu\n", this,
+                _reserve_at_beginning, _alignment);
+        toku_mutex_unlock(&_trace_lock);
+
+        fflush(ba_trace_file);
+    }
+}
+
+void block_allocator::_trace_create_from_blockpairs(void) {
+    if (ba_trace_file != nullptr) {
+        toku_mutex_lock(&_trace_lock);
+        fprintf(ba_trace_file, "ba_trace_create_from_blockpairs %p %lu %lu ", this,
+                _reserve_at_beginning, _alignment);
+        for (uint64_t i = 0; i < _n_blocks; i++) {
+            fprintf(ba_trace_file, "[%lu %lu] ", _blocks_array[i].offset, _blocks_array[i].size);
+        }
+        fprintf(ba_trace_file, "\n");
+        toku_mutex_unlock(&_trace_lock);
+
+        fflush(ba_trace_file);
+    }
+}
+
+void block_allocator::_trace_destroy(void) {
+    if (ba_trace_file != nullptr) {
+        toku_mutex_lock(&_trace_lock);
+        fprintf(ba_trace_file, "ba_trace_destroy %p\n", this);
+        toku_mutex_unlock(&_trace_lock);
+
+        fflush(ba_trace_file);
+    }
+}
+
+void block_allocator::_trace_alloc(uint64_t size, uint64_t heat, uint64_t offset) {
+    if (ba_trace_file != nullptr) {
+        toku_mutex_lock(&_trace_lock);
+        fprintf(ba_trace_file, "ba_trace_alloc %p %lu %lu %lu\n", this,
+                static_cast<unsigned long>(size),
+                static_cast<unsigned long>(heat),
+                static_cast<unsigned long>(offset));
+        toku_mutex_unlock(&_trace_lock);
+
+        fflush(ba_trace_file);
+    }
+}
+
+void block_allocator::_trace_free(uint64_t offset) {
+    if (ba_trace_file != nullptr) {
+        toku_mutex_lock(&_trace_lock);
+        fprintf(ba_trace_file, "ba_trace_free %p %lu\n", this,
+                static_cast<unsigned long>(offset));
+        toku_mutex_unlock(&_trace_lock);
+
+        fflush(ba_trace_file);
+    }
+}

ft/serialize/block_allocator.h

@@ -94,6 +94,7 @@ PATENT RIGHTS GRANT:
 
 #include <db.h>
 
+#include "portability/toku_pthread.h"
 #include "portability/toku_stdint.h"
 
 // Block allocator.
@@ -241,6 +242,14 @@ class block_allocator {
     int64_t find_block(uint64_t offset);
     struct blockpair *choose_block_to_alloc_after(size_t size, uint64_t heat);
 
+    // Tracing
+    toku_mutex_t _trace_lock;
+    void _trace_create(void);
+    void _trace_create_from_blockpairs(void);
+    void _trace_destroy(void);
+    void _trace_alloc(uint64_t size, uint64_t heat, uint64_t offset);
+    void _trace_free(uint64_t offset);
+
     // How much to reserve at the beginning
     uint64_t _reserve_at_beginning;
     // Block alignment

tools/ba_replay.cc

@@ -120,21 +120,16 @@ static void ba_replay_assert(bool pred, const char *msg, const char *line, int l
     }
 }
 
-// return line with whitespace skipped, and any newline replaced with a null byte
-static char *tidy_line(char *line) {
+static char *trim_whitespace(char *line) {
     // skip leading whitespace
     while (isspace(*line)) {
         line++;
     }
-    char *ptr = strchr(line, '\n');
-    if (ptr != nullptr) {
-        *ptr = '\0';
-    }
     return line;
 }
 
 static int64_t parse_number(char **ptr, int line_num, int base) {
-    *ptr = tidy_line(*ptr);
+    *ptr = trim_whitespace(*ptr);
 
     char *new_ptr;
     int64_t n = strtoll(*ptr, &new_ptr, base);
@@ -152,7 +147,7 @@ static uint64_t parse_uint64(char **ptr, int line_num) {
 }
 
 static string parse_token(char **ptr, int line_num) {
-    char *line = *ptr;
+    char *line = trim_whitespace(*ptr);
 
     // parse the first token, which represents the traced function
     char token[64];
@@ -162,6 +157,44 @@ static string parse_token(char **ptr, int line_num) {
     return string(token);
 }
 
+static block_allocator::blockpair parse_blockpair(char **ptr, int line_num) {
+    char *line = trim_whitespace(*ptr);
+    uint64_t offset, size;
+    int bytes_read;
+    int r = sscanf(line, "[%" PRIu64 " %" PRIu64 "]%n", &offset, &size, &bytes_read);
+    ba_replay_assert(r == 3, "malformed trace", line, line_num);
+    *ptr += bytes_read;
+    return block_allocator::blockpair(offset, size);
+}
+
+static char *strip_newline(char *line, bool *found) {
+    char *ptr = strchr(line, '\n');
+    if (ptr != nullptr) {
+        *found = true;
+        *ptr = '\0';
+    }
+    return line;
+}
+
+static char *read_trace_line(FILE *file) {
+    const int buf_size = 4096;
+    char buf[buf_size];
+    std::stringstream ss;
+    while (true) {
+        if (fgets(buf, buf_size, file) == nullptr) {
+            break;
+        }
+        bool has_newline = false;
+        ss << strip_newline(buf, &has_newline);
+        if (has_newline) {
+            // end of the line, we're done out
+            break;
+        }
+    }
+    std::string s = ss.str();
+    return toku_strdup(s.c_str());
+}
+
 static vector<string> canonicalize_trace_from(FILE *file) {
     // new trace, canonicalized from a raw trace
     vector<string> canonicalized_trace;
@@ -182,24 +215,22 @@ static vector<string> canonicalize_trace_from(FILE *file) {
     map<uint64_t, offset_seq_map> offset_to_seq_num_maps;
 
     int line_num = 0;
-    const int max_line = 512;
-    char line[max_line];
-    while (fgets(line, max_line, file) != nullptr) {
+    char *line;
+    while ((line = read_trace_line(file)) != nullptr) {
         line_num++;
-
-        // removes leading whitespace and trailing newline
-        char *ptr = tidy_line(line);
+        char *ptr = line;
 
         string fn = parse_token(&ptr, line_num);
         int64_t allocator_id = parse_number(&ptr, line_num, 16);
 
         std::stringstream ss;
-        if (fn == "ba_trace_create") {
-            // only allocators created in the raw traec will be part of the
-            // canonical trace, so save the next canonical allocator id here.
+        if (fn.find("ba_trace_create") != string::npos) {
+            // either a create or a create_from_blockpairs. either way,
+            // we only convert the allocator_id to an allocator_id_seq_num
+            // in the canonical trace and leave the rest of the line as-is.
             ba_replay_assert(allocator_ids.count(allocator_id) == 0, "corrupted trace: double create", line, line_num);
             allocator_ids[allocator_id] = allocator_id_seq_num;
-            ss << fn << ' ' << allocator_id_seq_num << ' ' << std::endl;
+            ss << fn << ' ' << allocator_id_seq_num << ' ' << trim_whitespace(ptr) << std::endl;
             allocator_id_seq_num++;
         } else if (allocator_ids.count(allocator_id) > 0) {
             // this allocator is part of the canonical trace
@@ -243,6 +274,8 @@ static vector<string> canonicalize_trace_from(FILE *file) {
             continue;
         }
         canonicalized_trace.push_back(ss.str());
+
+        toku_free(line);
     }
 
     return canonicalized_trace;
@@ -265,18 +298,31 @@ static void replay_canonicalized_trace(const vector<string> &canonicalized_trace
             printf("playing canonical trace line #%d: %s", line_num, line);
         }
 
-        char *ptr = tidy_line(line);
+        char *ptr = trim_whitespace(line);
 
         // canonical allocator id is in base 10, not 16
         string fn = parse_token(&ptr, line_num);
         int64_t allocator_id = parse_number(&ptr, line_num, 10);
 
-        if (fn == "ba_trace_create") {
+        if (fn.find("ba_trace_create") != string::npos) {
+            const uint64_t reserve_at_beginning = parse_uint64(&ptr, line_num);
+            const uint64_t alignment = parse_uint64(&ptr, line_num);
             ba_replay_assert(allocator_map->count(allocator_id) == 0,
                              "corrupted canonical trace: double create", line, line_num);
 
             block_allocator *ba = new block_allocator();
-            ba->create(8096, 4096); // header reserve, alignment - taken from block_table.cc
+            if (fn == "ba_trace_create") {
+                ba->create(reserve_at_beginning, alignment);
+            } else {
+                ba_replay_assert(fn == "ba_trace_create_from_blockpairs",
+                                 "corrupted canonical trace: bad create fn", line, line_num);
+                vector<block_allocator::blockpair> pairs;
+                while (*trim_whitespace(ptr) != '\0') {
+                    block_allocator::blockpair bp = parse_blockpair(&ptr, line_num);
+                    pairs.push_back(bp);
+                }
+                ba->create_from_blockpairs(reserve_at_beginning, alignment, &pairs[0], pairs.size());
+            }
             ba->set_strategy(strategy);
 
             // caller owns the allocator_map and its contents