Block cache analyzer: Compute correlation of features and human reada…

…ble trace file. (facebook#5596)

Summary:
- Compute correlation between a few features and predictions, e.g., number of accesses since the last access vs number of accesses till the next access on a block.
- Output human readable trace file so python can consume it.
Pull Request resolved: facebook#5596

Test Plan: make clean && USE_CLANG=1 make check -j32

Differential Revision: D16373200

Pulled By: HaoyuHuang

fbshipit-source-id: c848d26bc2e9210461f317d7dbee42d55be5a0cc

tools/block_cache_trace_analyzer.h

@@ -16,8 +16,23 @@
 #include "utilities/simulator_cache/cache_simulator.h"
 
 namespace rocksdb {
+
+// Statistics of a key refereneced by a Get.
+struct GetKeyInfo {
+  uint64_t key_id = 0;
+  std::vector<uint64_t> access_sequence_number_timeline;
+  std::vector<uint64_t> access_timeline;
+
+  void AddAccess(const BlockCacheTraceRecord& access,
+                 uint64_t access_sequnce_number) {
+    access_sequence_number_timeline.push_back(access_sequnce_number);
+    access_timeline.push_back(access.access_timestamp);
+  }
+};
+
 // Statistics of a block.
 struct BlockAccessInfo {
+  uint64_t block_id = 0;
   uint64_t num_accesses = 0;
   uint64_t block_size = 0;
   uint64_t first_access_time = 0;
@@ -39,7 +54,16 @@ struct BlockAccessInfo {
   // Number of reuses grouped by reuse distance.
   std::map<uint64_t, uint64_t> reuse_distance_count;
 
-  void AddAccess(const BlockCacheTraceRecord& access) {
+  // The access sequence numbers of this block.
+  std::vector<uint64_t> access_sequence_number_timeline;
+  std::map<TableReaderCaller, std::vector<uint64_t>>
+      caller_access_sequence__number_timeline;
+  // The access timestamp in microseconds of this block.
+  std::vector<uint64_t> access_timeline;
+  std::map<TableReaderCaller, std::vector<uint64_t>> caller_access_timeline;
+
+  void AddAccess(const BlockCacheTraceRecord& access,
+                 uint64_t access_sequnce_number) {
     if (block_size != 0 && access.block_size != 0) {
       assert(block_size == access.block_size);
     }
@@ -57,6 +81,12 @@ struct BlockAccessInfo {
     const uint64_t timestamp_in_seconds =
         access.access_timestamp / kMicrosInSecond;
     caller_num_accesses_timeline[access.caller][timestamp_in_seconds] += 1;
+    // Populate the feature vectors.
+    access_sequence_number_timeline.push_back(access_sequnce_number);
+    caller_access_sequence__number_timeline[access.caller].push_back(
+        access_sequnce_number);
+    access_timeline.push_back(access.access_timestamp);
+    caller_access_timeline[access.caller].push_back(access.access_timestamp);
     if (BlockCacheTraceHelper::IsGetOrMultiGetOnDataBlock(access.block_type,
                                                           access.caller)) {
       num_keys = access.num_keys_in_block;
@@ -94,11 +124,23 @@ struct ColumnFamilyAccessInfoAggregate {
   std::map<uint64_t, SSTFileAccessInfoAggregate> fd_aggregates_map;
 };
 
+struct Features {
+  std::vector<uint64_t> elapsed_time_since_last_access;
+  std::vector<uint64_t> num_accesses_since_last_access;
+  std::vector<uint64_t> num_past_accesses;
+};
+
+struct Predictions {
+  std::vector<uint64_t> elapsed_time_till_next_access;
+  std::vector<uint64_t> num_accesses_till_next_access;
+};
+
 class BlockCacheTraceAnalyzer {
  public:
   BlockCacheTraceAnalyzer(
       const std::string& trace_file_path, const std::string& output_dir,
-      bool compute_reuse_distance,
+      const std::string& human_readable_trace_file_path,
+      bool compute_reuse_distance, bool mrc_only,
       std::unique_ptr<BlockCacheTraceSimulator>&& cache_simulator);
   ~BlockCacheTraceAnalyzer() = default;
   // No copy and move.
@@ -184,6 +226,24 @@ class BlockCacheTraceAnalyzer {
   // "cache_name,num_shard_bits,capacity,miss_ratio,total_accesses".
   void WriteMissRatioCurves() const;
 
+  // Write miss ratio timeline of simulated cache configurations into several
+  // csv files, one per cache capacity saved in 'output_dir'.
+  //
+  // The file format is
+  // "time,label_1_access_per_second,label_2_access_per_second,...,label_N_access_per_second"
+  // where N is the number of unique cache names
+  // (cache_name+num_shard_bits+ghost_capacity).
+  void WriteMissRatioTimeline(uint64_t time_unit) const;
+
+  // Write misses timeline of simulated cache configurations into several
+  // csv files, one per cache capacity saved in 'output_dir'.
+  //
+  // The file format is
+  // "time,label_1_access_per_second,label_2_access_per_second,...,label_N_access_per_second"
+  // where N is the number of unique cache names
+  // (cache_name+num_shard_bits+ghost_capacity).
+  void WriteMissTimeline(uint64_t time_unit) const;
+
   // Write the access timeline into a csv file saved in 'output_dir'.
   //
   // The file is named "label_access_timeline".The file format is
@@ -236,6 +296,11 @@ class BlockCacheTraceAnalyzer {
       const std::string& label_str,
       const std::vector<uint64_t>& percent_buckets) const;
 
+  void WriteCorrelationFeatures(const std::string& label_str,
+                                uint32_t max_number_of_values) const;
+
+  void WriteCorrelationFeaturesForGet(uint32_t max_number_of_values) const;
+
   const std::map<std::string, ColumnFamilyAccessInfoAggregate>&
   TEST_cf_aggregates_map() const {
     return cf_aggregates_map_;
@@ -251,7 +316,7 @@ class BlockCacheTraceAnalyzer {
 
   void ComputeReuseDistance(BlockAccessInfo* info) const;
 
-  void RecordAccess(const BlockCacheTraceRecord& access);
+  Status RecordAccess(const BlockCacheTraceRecord& access);
 
   void UpdateReuseIntervalStats(
       const std::string& label, const std::vector<uint64_t>& time_buckets,
@@ -278,17 +343,41 @@ class BlockCacheTraceAnalyzer {
                          const BlockAccessInfo& /*block_access_info*/)>
           block_callback) const;
 
+  void UpdateFeatureVectors(
+      const std::vector<uint64_t>& access_sequence_number_timeline,
+      const std::vector<uint64_t>& access_timeline, const std::string& label,
+      std::map<std::string, Features>* label_features,
+      std::map<std::string, Predictions>* label_predictions) const;
+
+  void WriteCorrelationFeaturesToFile(
+      const std::string& label,
+      const std::map<std::string, Features>& label_features,
+      const std::map<std::string, Predictions>& label_predictions,
+      uint32_t max_number_of_values) const;
+
+  Status WriteHumanReadableTraceRecord(const BlockCacheTraceRecord& access,
+                                       uint64_t block_id, uint64_t get_key_id);
+
   rocksdb::Env* env_;
   const std::string trace_file_path_;
   const std::string output_dir_;
+  std::string human_readable_trace_file_path_;
   const bool compute_reuse_distance_;
+  const bool mrc_only_;
 
   BlockCacheTraceHeader header_;
   std::unique_ptr<BlockCacheTraceSimulator> cache_simulator_;
   std::map<std::string, ColumnFamilyAccessInfoAggregate> cf_aggregates_map_;
   std::map<std::string, BlockAccessInfo*> block_info_map_;
+  std::unordered_map<std::string, GetKeyInfo> get_key_info_map_;
+  uint64_t access_sequence_number_ = 0;
   uint64_t trace_start_timestamp_in_seconds_ = 0;
   uint64_t trace_end_timestamp_in_seconds_ = 0;
+  MissRatioStats miss_ratio_stats_;
+  uint64_t unique_block_id_ = 1;
+  uint64_t unique_get_key_id_ = 1;
+  char trace_record_buffer_[1024 * 1024];
+  std::unique_ptr<rocksdb::WritableFile> human_readable_trace_file_writer_;
 };
 
 int block_cache_trace_analyzer_tool(int argc, char** argv);

tools/block_cache_trace_analyzer_test.cc

@@ -117,7 +117,8 @@ class BlockCacheTracerTest : public testing::Test {
       // Provide these fields for all block types.
       // The writer should only write these fields for data blocks and the
       // caller is either GET or MGET.
-      record.referenced_key = kRefKeyPrefix + std::to_string(key_id);
+      record.referenced_key =
+          kRefKeyPrefix + std::to_string(key_id) + std::string(8, 0);
       record.referenced_key_exist_in_block = Boolean::kTrue;
       record.num_keys_in_block = kNumKeysInBlock;
       ASSERT_OK(writer->WriteBlockAccess(
@@ -179,7 +180,8 @@ class BlockCacheTracerTest : public testing::Test {
         "-analyze_get_spatial_locality_labels=" +
             analyze_get_spatial_locality_labels_,
         "-analyze_get_spatial_locality_buckets=" +
-            analyze_get_spatial_locality_buckets_};
+            analyze_get_spatial_locality_buckets_,
+        "-analyze_correlation_coefficients_labels=all"};
     char arg_buffer[kArgBufferSize];
     char* argv[kMaxArgCount];
     int argc = 0;
@@ -236,9 +238,9 @@ TEST_F(BlockCacheTracerTest, BlockCacheAnalyzer) {
   RunBlockCacheTraceAnalyzer();
   {
     // Validate the cache miss ratios.
-    const std::vector<uint64_t> expected_capacities{1024, 1024 * 1024,
-                                                    1024 * 1024 * 1024};
-    const std::string mrc_path = test_path_ + "/mrc";
+    std::vector<uint64_t> expected_capacities{1024, 1024 * 1024,
+                                              1024 * 1024 * 1024};
+    const std::string mrc_path = test_path_ + "/49_50_mrc";
     std::ifstream infile(mrc_path);
     uint32_t config_index = 0;
     std::string line;
@@ -266,6 +268,68 @@ TEST_F(BlockCacheTracerTest, BlockCacheAnalyzer) {
     ASSERT_EQ(expected_capacities.size(), config_index);
     infile.close();
     ASSERT_OK(env_->DeleteFile(mrc_path));
+
+    const std::vector<std::string> time_units{"1", "60", "3600"};
+    expected_capacities.push_back(port::kMaxUint64);
+    for (auto const& expected_capacity : expected_capacities) {
+      for (auto const& time_unit : time_units) {
+        const std::string miss_ratio_timeline_path =
+            test_path_ + "/" + std::to_string(expected_capacity) + "_" +
+            time_unit + "_miss_ratio_timeline";
+        std::ifstream mrt_file(miss_ratio_timeline_path);
+        // Read header.
+        ASSERT_TRUE(getline(mrt_file, line));
+        ASSERT_TRUE(getline(mrt_file, line));
+        std::stringstream ss(line);
+        bool read_header = false;
+        while (ss.good()) {
+          std::string substr;
+          getline(ss, substr, ',');
+          if (!read_header) {
+            if (expected_capacity == port::kMaxUint64) {
+              ASSERT_EQ("trace", substr);
+            } else {
+              ASSERT_EQ("lru-1-0", substr);
+            }
+            read_header = true;
+            continue;
+          }
+          ASSERT_DOUBLE_EQ(100.0, ParseDouble(substr));
+        }
+        ASSERT_FALSE(getline(mrt_file, line));
+        mrt_file.close();
+        ASSERT_OK(env_->DeleteFile(miss_ratio_timeline_path));
+      }
+      for (auto const& time_unit : time_units) {
+        const std::string miss_timeline_path =
+            test_path_ + "/" + std::to_string(expected_capacity) + "_" +
+            time_unit + "_miss_timeline";
+        std::ifstream mt_file(miss_timeline_path);
+        // Read header.
+        ASSERT_TRUE(getline(mt_file, line));
+        ASSERT_TRUE(getline(mt_file, line));
+        std::stringstream ss(line);
+        uint32_t num_misses = 0;
+        while (ss.good()) {
+          std::string substr;
+          getline(ss, substr, ',');
+          if (num_misses == 0) {
+            if (expected_capacity == port::kMaxUint64) {
+              ASSERT_EQ("trace", substr);
+            } else {
+              ASSERT_EQ("lru-1-0", substr);
+            }
+            num_misses++;
+            continue;
+          }
+          num_misses += ParseInt(substr);
+        }
+        ASSERT_EQ(51, num_misses);
+        ASSERT_FALSE(getline(mt_file, line));
+        mt_file.close();
+        ASSERT_OK(env_->DeleteFile(miss_timeline_path));
+      }
+    }
   }
   {
     // Validate the timeline csv files.
@@ -543,7 +607,9 @@ TEST_F(BlockCacheTracerTest, MixedBlocks) {
     // Read blocks.
     BlockCacheTraceAnalyzer analyzer(trace_file_path_,
                                      /*output_miss_ratio_curve_path=*/"",
+                                     /*human_readable_trace_file_path=*/"",
                                      /*compute_reuse_distance=*/true,
+                                     /*mrc_only=*/false,
                                      /*simulator=*/nullptr);
     // The analyzer ends when it detects an incomplete access record.
     ASSERT_EQ(Status::Incomplete(""), analyzer.Analyze());

trace_replay/block_cache_tracer.cc

@@ -6,6 +6,7 @@
 #include "trace_replay/block_cache_tracer.h"
 
 #include "db/db_impl/db_impl.h"
+#include "db/dbformat.h"
 #include "rocksdb/slice.h"
 #include "util/coding.h"
 #include "util/hash.h"
@@ -54,6 +55,19 @@ bool BlockCacheTraceHelper::IsUserAccess(TableReaderCaller caller) {
          caller == TableReaderCaller::kUserVerifyChecksum;
 }
 
+std::string BlockCacheTraceHelper::ComputeRowKey(
+    const BlockCacheTraceRecord& access) {
+  if (!IsGetOrMultiGet(access.caller)) {
+    return "";
+  }
+  Slice key = ExtractUserKey(access.referenced_key);
+  uint64_t seq_no = access.get_from_user_specified_snapshot == Boolean::kFalse
+                        ? 0
+                        : 1 + GetInternalKeySeqno(access.referenced_key);
+  return std::to_string(access.sst_fd_number) + "_" + key.ToString() + "_" +
+         std::to_string(seq_no);
+}
+
 BlockCacheTraceWriter::BlockCacheTraceWriter(
     Env* env, const TraceOptions& trace_options,
     std::unique_ptr<TraceWriter>&& trace_writer)

trace_replay/block_cache_tracer.h

@@ -20,14 +20,17 @@ extern const uint64_t kMicrosInSecond;
 extern const uint64_t kSecondInMinute;
 extern const uint64_t kSecondInHour;
 
+struct BlockCacheTraceRecord;
 
 class BlockCacheTraceHelper {
  public:
   static bool IsGetOrMultiGetOnDataBlock(TraceType block_type,
                                          TableReaderCaller caller);
   static bool IsGetOrMultiGet(TableReaderCaller caller);
   static bool IsUserAccess(TableReaderCaller caller);
-
+  // Row key is a concatenation of the access's fd_number and the referenced
+  // user key.
+  static std::string ComputeRowKey(const BlockCacheTraceRecord& access);
   static const std::string kUnknownColumnFamilyName;
   static const uint64_t kReservedGetId;
 };