You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2020/06/17 14:58:35 UTC
[impala] 03/03: IMPALA-9435: Usability enhancements for data cache
access trace
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit d38e4d10de98d46c4b5b6a4f5ef0a9b9bbb61dae
Author: Joe McDonnell <jo...@cloudera.com>
AuthorDate: Mon Mar 9 11:52:54 2020 -0700
IMPALA-9435: Usability enhancements for data cache access trace
The data cache access trace was added in IMPALA-8542 as a way
to capture a workload's cache accesses to allow later analysis.
This modifies the data cache access trace to improve usability:
1. The access trace now uses a SimpleLogger to limit the total
number of trace entries per file and total number of trace
files. This caps the disk usage for the access trace. The
behavior is controlled by the data_cache_trace_dir,
max_data_cache_trace_file_size, and max_data_cache_trace_files
startup parameters.
2. This introduces the data_cache_trace_percentage, which allows
tracing only a subset of the entries produced. It traces
accesses for a consistent subset of the cache (i.e. accesses
for a filename/mtime/offset are either always traced or
never traced). This allows for better analysis than a random
sample. Tracing a subset of accesses can reduce any performance
overhead from tracing. It also provides a way to trace a longer
time period in the same number of entries.
This also implements the ability to replay traces against a
specific cache configuration. The replayer can produce JSON output
with cache hit/miss information for the original trace and the
replay. This provides a building block for building analysis
comparing different cache sizes or cache eviction policies.
Testing:
- New backend tests in data-cache-test, data-cache-trace-test
- Manually testing the data-cache-trace-replayer
Change-Id: I0f84204d8e5145f5fa8d4851d9c19ac317db168e
Reviewed-on: http://gerrit.cloudera.org:8080/15914
Reviewed-by: Joe McDonnell <jo...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/runtime/io/CMakeLists.txt | 7 +
be/src/runtime/io/data-cache-test.cc | 156 +++++++---
be/src/runtime/io/data-cache-trace-replayer.cc | 203 +++++++++++++
be/src/runtime/io/data-cache-trace-test.cc | 338 ++++++++++++++++++++++
be/src/runtime/io/data-cache-trace.cc | 374 ++++++++++++++++++++++++
be/src/runtime/io/data-cache-trace.h | 247 ++++++++++++++++
be/src/runtime/io/data-cache.cc | 378 ++++++++++---------------
be/src/runtime/io/data-cache.h | 50 +++-
bin/start-impala-cluster.py | 26 ++
9 files changed, 1507 insertions(+), 272 deletions(-)
diff --git a/be/src/runtime/io/CMakeLists.txt b/be/src/runtime/io/CMakeLists.txt
index 239d901..7c2637b 100644
--- a/be/src/runtime/io/CMakeLists.txt
+++ b/be/src/runtime/io/CMakeLists.txt
@@ -33,10 +33,12 @@ add_library(Io
hdfs-file-reader.cc
local-file-reader.cc
hdfs-monitored-ops.cc
+ data-cache-trace.cc
)
add_dependencies(Io gen-deps)
add_library(IoTests STATIC
+ data-cache-trace-test.cc
disk-io-mgr-test.cc
)
add_dependencies(IoTests gen-deps)
@@ -45,6 +47,11 @@ add_dependencies(IoTests gen-deps)
add_executable(disk-io-mgr-stress-test disk-io-mgr-stress-test.cc)
target_link_libraries(disk-io-mgr-stress-test ${IMPALA_TEST_LINK_LIBS})
+# This is a utility to replay access traces
+add_executable(data-cache-trace-replayer data-cache-trace-replayer.cc)
+target_link_libraries(data-cache-trace-replayer ${IMPALA_TEST_LINK_LIBS})
+
ADD_UNIFIED_BE_LSAN_TEST(disk-io-mgr-test DiskIoMgrTest.*)
+ADD_UNIFIED_BE_LSAN_TEST(data-cache-trace-test DataCacheTraceTest.*)
# Exception to unified be: Custom main function (platform tests)
ADD_BE_LSAN_TEST(data-cache-test)
diff --git a/be/src/runtime/io/data-cache-test.cc b/be/src/runtime/io/data-cache-test.cc
index 6ea4ca5..9369323 100644
--- a/be/src/runtime/io/data-cache-test.cc
+++ b/be/src/runtime/io/data-cache-test.cc
@@ -17,6 +17,7 @@
#include <algorithm>
#include <boost/bind.hpp>
+#include <boost/filesystem.hpp>
#include <fstream>
#include <gflags/gflags.h>
#include <rapidjson/document.h>
@@ -25,6 +26,7 @@
#include "gutil/strings/join.h"
#include "gutil/strings/util.h"
#include "runtime/io/data-cache.h"
+#include "runtime/io/data-cache-trace.h"
#include "runtime/io/request-ranges.h"
#include "runtime/test-env.h"
#include "service/fe-support.h"
@@ -32,6 +34,7 @@
#include "testutil/scoped-flag-setter.h"
#include "util/counting-barrier.h"
#include "util/filesystem-util.h"
+#include "util/simple-logger.h"
#include "util/thread.h"
#include "common/names.h"
@@ -60,10 +63,15 @@ DECLARE_int64(data_cache_file_max_size_bytes);
DECLARE_int32(data_cache_max_opened_files);
DECLARE_int32(data_cache_write_concurrency);
DECLARE_string(data_cache_eviction_policy);
+DECLARE_string(data_cache_trace_dir);
+DECLARE_int32(max_data_cache_trace_file_size);
+DECLARE_int32(data_cache_trace_percentage);
namespace impala {
namespace io {
+using boost::filesystem::path;
+
class DataCacheBaseTest : public testing::Test {
public:
const uint8_t* test_buffer() {
@@ -74,6 +82,10 @@ class DataCacheBaseTest : public testing::Test {
return data_cache_dirs_;
}
+ const string& data_cache_trace_dir() {
+ return data_cache_trace_dir_;
+ }
+
//
// Use multiple threads to insert and read back a set of ranges from test_buffer().
// Depending on the setting, the working set may or may not fit in the cache.
@@ -138,6 +150,8 @@ class DataCacheBaseTest : public testing::Test {
ASSERT_OK(FileSystemUtil::RemoveAndCreateDirectory(path));
data_cache_dirs_.push_back(path);
}
+ data_cache_trace_dir_ = ("/tmp" / boost::filesystem::unique_path()).string();
+ ASSERT_OK(FileSystemUtil::RemoveAndCreateDirectory(data_cache_trace_dir_));
// Force a single shard to avoid imbalance between shards in the Kudu LRU cache which
// may lead to unintended eviction. This allows the capacity of the cache is utilized
@@ -158,13 +172,10 @@ class DataCacheBaseTest : public testing::Test {
for (const string& dir_path : data_cache_dirs_) {
vector<string> entries;
ASSERT_OK(FileSystemUtil::Directory::GetEntryNames(dir_path, &entries));
- if (entries.size() == 1) {
- EXPECT_EQ(DataCache::Partition::TRACE_FILE_NAME, entries[0]);
- } else {
- ASSERT_EQ(0, entries.size());
- }
+ ASSERT_EQ(0, entries.size());
}
ASSERT_OK(FileSystemUtil::RemovePaths(data_cache_dirs_));
+ boost::filesystem::remove_all(path(data_cache_trace_dir_));
flag_saver_.reset();
test_env_.reset();
}
@@ -173,6 +184,7 @@ class DataCacheBaseTest : public testing::Test {
std::unique_ptr<TestEnv> test_env_;
char test_buffer_[TEST_BUFFER_SIZE];
vector<string> data_cache_dirs_;
+ string data_cache_trace_dir_;
// Saved configuration flags for restoring the values at the end of the test.
std::unique_ptr<google::FlagSaver> flag_saver_;
@@ -545,55 +557,125 @@ TEST_P(DataCacheTest, LargeFootprint) {
}
}
-TEST_P(DataCacheTest, TestAccessTrace) {
+TEST_P(DataCacheTest, AccessTraceAnonymization) {
FLAGS_data_cache_enable_tracing = true;
+ FLAGS_data_cache_trace_dir = data_cache_trace_dir();
+ // Use a small trace file size so that the trace is split over multiple files
+ FLAGS_max_data_cache_trace_file_size = 10000;
+ const int64_t cache_size = DEFAULT_CACHE_SIZE;
+ int64_t max_start_offset = 1024;
+ uint64_t expected_total_events = NUM_THREADS * max_start_offset * 2;
for (bool anon : { false, true }) {
SCOPED_TRACE(anon);
FLAGS_data_cache_anonymize_trace = anon;
+ // Remove files between iterations to avoid crosstalk between the anonymized
+ // iteration and non-anonymized iteration
+ ASSERT_OK(FileSystemUtil::RemoveAndCreateDirectory(FLAGS_data_cache_trace_dir));
{
- int64_t cache_size = DEFAULT_CACHE_SIZE;
DataCache cache(Substitute(
"$0:$1", data_cache_dirs()[0], std::to_string(cache_size)));
ASSERT_OK(cache.Init());
- int64_t max_start_offset = 1024;
bool use_per_thread_filename = true;
bool expect_misses = true;
MultiThreadedReadWrite(&cache, max_start_offset, use_per_thread_filename,
expect_misses);
}
- // Read the trace file and ensure that all of the lines are valid JSON with the
- // expected fields.
- std::ifstream trace(Substitute("$0/$1", data_cache_dirs()[0],
- DataCache::Partition::TRACE_FILE_NAME));
- int line_num = 1;
- while (trace.good()) {
- SCOPED_TRACE(line_num++);
- string line;
- std::getline(trace, line);
- if (line.empty()) {
- ASSERT_TRUE(trace.eof());
- break;
- }
- SCOPED_TRACE(line);
- rapidjson::Document d;
- d.Parse<0>(line.c_str());
- ASSERT_TRUE(d.IsObject());
- ASSERT_TRUE(d["ts"].IsDouble());
- ASSERT_TRUE(d["s"].IsString());
- ASSERT_TRUE(d["m"].IsInt64());
- ASSERT_TRUE(d["f"].IsString());
- if (anon) {
- // We expect anonymized filenames to be 22-character fingerprints:
- // - 128 bit fingerprint = 16 bytes
- // - 16 * 4/3 (base64 encoding) = 21.3333
- // - Round up to 22.
- EXPECT_EQ(22, d["f"].GetStringLength());
- } else {
- EXPECT_TRUE(MatchPattern(d["f"].GetString(), "thread-*file"));
+ // Part 1: Read the trace files and ensure that the JSON entries are valid
+ // TraceEvent entries
+ vector<string> trace_files;
+ string trace_dir = Substitute("$0/partition-0/", FLAGS_data_cache_trace_dir);
+ ASSERT_OK(SimpleLogger::GetLogFiles(trace_dir, trace::TRACE_FILE_PREFIX,
+ &trace_files));
+
+ for (string filename : trace_files) {
+ trace::TraceFileIterator trace_file_iter(filename);
+ ASSERT_OK(trace_file_iter.Init());
+ while (true) {
+ trace::TraceEvent trace_event;
+ bool done = false;
+ ASSERT_OK(trace_file_iter.GetNextEvent(&trace_event, &done));
+ if (done) break;
+ if (anon) {
+ // We expect anonymized filenames to be 22-character fingerprints:
+ // - 128 bit fingerprint = 16 bytes
+ // - 16 * 4/3 (base64 encoding) = 21.3333
+ // - Round up to 22.
+ EXPECT_EQ(22, trace_event.filename.size()) << trace_event.filename;
+ } else {
+ EXPECT_TRUE(MatchPattern(trace_event.filename, "thread-*file"));
+ }
}
}
+
+ // Part 2: Create a replayer and verify that it can successfully replay from that
+ // directory.
+ trace::TraceReplayer replayer(Substitute("/this_directory_doesnt_matter:$0",
+ std::to_string(cache_size)));
+ EXPECT_OK(replayer.Init());
+ EXPECT_OK(replayer.ReplayDirectory(trace_dir));
+ trace::CacheHitStatistics replay_stats = replayer.GetReplayStatistics();
+ int64_t total_events = replay_stats.hits + replay_stats.partial_hits +
+ replay_stats.misses + replay_stats.stores + replay_stats.failed_stores;
+ EXPECT_EQ(total_events, expected_total_events);
+ }
+}
+
+TEST_P(DataCacheTest, AccessTraceSubsetPercentage) {
+ FLAGS_data_cache_enable_tracing = true;
+ FLAGS_data_cache_trace_dir = data_cache_trace_dir();
+ // Use a small trace file size so that the trace is split over multiple files
+ FLAGS_max_data_cache_trace_file_size = 10000;
+ const int64_t cache_size = DEFAULT_CACHE_SIZE;
+ int64_t max_start_offset = 1024;
+ // Each thread does a store and a lookup for each offset
+ uint64_t expected_total_events = NUM_THREADS * max_start_offset * 2;
+ for (int trace_percentage : {100, 50, 20, 10, 5, 1}) {
+ SCOPED_TRACE(trace_percentage);
+ FLAGS_data_cache_trace_percentage = trace_percentage;
+ // Remove files between iterations to avoid crosstalk between the anonymized
+ // iteration and non-anonymized iteration
+ ASSERT_OK(FileSystemUtil::RemoveAndCreateDirectory(FLAGS_data_cache_trace_dir));
+ {
+ DataCache cache(Substitute(
+ "$0:$1", data_cache_dirs()[0], std::to_string(cache_size)));
+ ASSERT_OK(cache.Init());
+
+ bool use_per_thread_filename = true;
+ bool expect_misses = true;
+ MultiThreadedReadWrite(&cache, max_start_offset, use_per_thread_filename,
+ expect_misses);
+ }
+
+ // Replay the trace and record the counts
+ string trace_dir = Substitute("$0/partition-0/", FLAGS_data_cache_trace_dir);
+ trace::TraceReplayer replayer(Substitute("/this_directory_doesnt_matter:$0",
+ std::to_string(cache_size)));
+ EXPECT_OK(replayer.Init());
+ EXPECT_OK(replayer.ReplayDirectory(trace_dir));
+
+ // The trace percentage is enforced when generating the trace, so verify that
+ // the number of events in the original trace is within +/-1% of the expected value.
+ // This is deterministic, because the actual cache entries accessed are always
+ // the same (even if the order is not).
+ trace::CacheHitStatistics original_trace_stats =
+ replayer.GetOriginalTraceStatistics();
+ int64_t num_original_trace_events = original_trace_stats.hits +
+ original_trace_stats.partial_hits + original_trace_stats.misses +
+ original_trace_stats.stores + original_trace_stats.failed_stores;
+ EXPECT_LE(100 * num_original_trace_events,
+ (expected_total_events * (trace_percentage + 1)));
+ EXPECT_GE(100 * num_original_trace_events,
+ (expected_total_events * (trace_percentage - 1)));
+ LOG(INFO) << "Trace percentage " << trace_percentage << " num events: "
+ << num_original_trace_events;
+
+ // The replay should have the same number of events.
+ trace::CacheHitStatistics replay_stats = replayer.GetReplayStatistics();
+ int64_t num_replay_trace_events = replay_stats.hits + replay_stats.partial_hits +
+ replay_stats.misses + replay_stats.stores + replay_stats.failed_stores;
+ EXPECT_EQ(num_original_trace_events, num_replay_trace_events);
}
}
diff --git a/be/src/runtime/io/data-cache-trace-replayer.cc b/be/src/runtime/io/data-cache-trace-replayer.cc
new file mode 100644
index 0000000..e2ea8cb
--- /dev/null
+++ b/be/src/runtime/io/data-cache-trace-replayer.cc
@@ -0,0 +1,203 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <boost/filesystem.hpp>
+#include <fstream>
+#include <gflags/gflags.h>
+#include <iostream>
+#include <memory>
+#include <rapidjson/document.h>
+#include <rapidjson/ostreamwrapper.h>
+#include <rapidjson/writer.h>
+#include <string>
+
+#include "common/init.h"
+#include "common/status.h"
+#include "gutil/strings/substitute.h"
+#include "runtime/io/data-cache-trace.h"
+
+#include "common/names.h"
+
+// This provides a command-line utility for replaying data cache trace files.
+// It is simply a wrapper around the DataCacheTrace::TraceReplayer to take
+// command line arguments. Beyond the explicitly provided commandline flags
+// below, it supports the other command line flags that influence cache behavior.
+// For example, the data_cache_eviction_policy can be specified, along with policy
+// specific flags such as lirs_unprotected_percentage or similar flags.
+//
+// Usage example:
+// Suppose there is a trace of a workload using a single-directory 100GB data cache with
+// LRU eviction policy. (i.e. data_cache_eviction_policy=LRU and
+// data_cache="/cache_path:100GB"). The replayer can run the same workload against
+// different cache configurations to see how things change.
+//
+// To run against a larger cache still using LRU:
+// data-cache-trace-replayer --trace_directory /path/to/trace/directory
+// --data_cache="/cache_path:150GB" --data_cache_eviction_policy=LRU
+//
+// To run against the same cache size, but with LIRS cache eviction policy:
+// data-cache-trace-replayer --trace_directory /path/to/trace/directory
+// --data_cache="/cache_path:100GB" --data_cache_eviction_policy=LIRS
+//
+// The replayer produces two different types of cache hit statistics. The first is
+// the cache hit statistics from the original trace (i.e. the original 100GB cache
+// using LRU). This is a fixed property of a given set of trace files, and it will
+// always be the same regardless of the replay configuration. The second is the cache
+// hit statistics from the replay using the replay configuration (e.g. a replay using
+// a smaller cache may show fewer hits, etc).
+//
+// These commands put output in the glog INFO file. For JSON output, see the
+// output_file option.
+
+// One of either trace_file or trace_directory must be specified
+DEFINE_string(trace_file, "", "Single trace file to replay");
+DEFINE_string(trace_directory, "", "Directory of trace files to replay. Files will be "
+ "replayed sequentially in sorted order.");
+
+// The replay may use any configuration supported by the DataCache.
+DEFINE_string(data_cache_configuration, "", "Cache configuration string in the same "
+ "format as the 'data_cache' Impala startup parameter. Specifically, it takes "
+ "a list of directories, separated by ',', followed by a ':' and a capacity quota "
+ "per directory. For example '/data/0,/data/1:1TB' means the cache may use up to 2TB, "
+ "with 1TB max in each /data/0 and /data/1.");
+
+// If specified, the cache hit statistics are written to a JSON file with the provided
+// filename. If not specified, output goes to the INFO log.
+DEFINE_string(output_file, "", "File to write with JSON output containing hits/misses");
+
+using namespace impala;
+using namespace impala::io;
+using namespace impala::io::trace;
+using namespace rapidjson;
+using strings::Substitute;
+
+// Convert the CacheHitStatistics 'stats' to a JSON struct using the allocator from
+// 'document'.
+Value CacheHitStatisticsToJson(Document* document, const CacheHitStatistics& stats) {
+ Value json_value(kObjectType);
+ json_value.AddMember("hits", Value(stats.hits), document->GetAllocator());
+ json_value.AddMember("partial_hits", Value(stats.partial_hits),
+ document->GetAllocator());
+ json_value.AddMember("hit_bytes", Value(stats.hit_bytes), document->GetAllocator());
+ json_value.AddMember("misses", Value(stats.misses), document->GetAllocator());
+ json_value.AddMember("miss_bytes", Value(stats.miss_bytes), document->GetAllocator());
+ json_value.AddMember("stores", Value(stats.stores), document->GetAllocator());
+ json_value.AddMember("failed_stores", Value(stats.failed_stores),
+ document->GetAllocator());
+ return json_value;
+}
+
+// Output CacheHitStatistics to the INFO glog.
+void DumpStatisticsToLog(const CacheHitStatistics& stats) {
+ LOG(INFO) << "Hits: " << std::to_string(stats.hits)
+ << " hit bytes: " << std::to_string(stats.hit_bytes)
+ << " partial hits: " << std::to_string(stats.partial_hits);
+ LOG(INFO) << "Misses: " << std::to_string(stats.misses)
+ << " miss bytes: " << std::to_string(stats.miss_bytes);
+ LOG(INFO) << "Stores: " << std::to_string(stats.stores)
+ << " failed stores: " << std::to_string(stats.failed_stores);
+}
+
+// Write a JSON structure with both the original trace cache hit statistics and
+// the replay cache hit statistics.
+void DumpStatisticsToJSON(const CacheHitStatistics& trace_stats,
+ const CacheHitStatistics& replay_stats, std::string filename) {
+ Document document;
+ document.SetObject();
+
+ // Add trace stats
+ Value trace_stats_json = CacheHitStatisticsToJson(&document, trace_stats);
+ document.AddMember("original_trace_stats", trace_stats_json, document.GetAllocator());
+ Value replay_stats_json = CacheHitStatisticsToJson(&document, replay_stats);
+ document.AddMember("replay_stats", replay_stats_json, document.GetAllocator());
+
+ ofstream ofs(filename);
+ OStreamWrapper osw(ofs);
+ Writer<OStreamWrapper> writer(osw);
+ document.Accept(writer);
+}
+
+Status ValidateFlags() {
+ // data_cache_configuration is required
+ if (FLAGS_data_cache_configuration.size() == 0) {
+ return Status("data_cache_configuration must be specified.");
+ }
+ // trace_file and trace_directory are mutually exclusive
+ if (FLAGS_trace_file.size() > 0 && FLAGS_trace_directory.size() > 0) {
+ return Status(Substitute("Cannot specify both trace_file and trace_directory"
+ "trace_file: $0 trace_directory: $1", FLAGS_trace_file, FLAGS_trace_directory));
+ }
+ // Requires one of trace_file or trace_directory
+ if (FLAGS_trace_file.size() == 0 && FLAGS_trace_directory.size() == 0) {
+ return Status("Either trace_file or trace_directory must be specified");
+ }
+
+ if (FLAGS_trace_file.size() > 0) {
+ // Verify the file exists
+ boost::filesystem::path trace_file(FLAGS_trace_file);
+ if (!boost::filesystem::exists(trace_file)) {
+ return Status(Substitute("Invalid trace_file $0. File does not exist.",
+ FLAGS_trace_file));
+ }
+ } else if (FLAGS_trace_directory.size() > 0) {
+ // Verify this is a directory that exists
+ boost::filesystem::path dir(FLAGS_trace_directory);
+ if (!boost::filesystem::exists(dir)) {
+ return Status(Substitute("Invalid trace_directory $0. Directory doesn't exist",
+ FLAGS_trace_directory));
+ }
+ if (!boost::filesystem::is_directory(dir)) {
+ return Status(Substitute("Invalid trace_directory $0. Not a directory.",
+ FLAGS_trace_directory));
+ }
+ }
+ return Status::OK();
+}
+
+int main(int argc, char **argv) {
+ InitCommonRuntime(argc, argv, false);
+
+ Status status = ValidateFlags();
+ if (!status.ok()) CLEAN_EXIT_WITH_ERROR(status.GetDetail());
+
+ LOG(INFO) << "Initialize cache with configuration: " << FLAGS_data_cache_configuration;
+ TraceReplayer replayer(FLAGS_data_cache_configuration);
+ status = replayer.Init();
+ if (!status.ok()) CLEAN_EXIT_WITH_ERROR(status.GetDetail());
+
+ if (FLAGS_trace_file.size() != 0) {
+ LOG(INFO) << "Replaying file: " << FLAGS_trace_file;
+ status = replayer.ReplayFile(FLAGS_trace_file);
+ } else if (FLAGS_trace_directory.size() != 0) {
+ LOG(INFO) << "Replaying directory: " << FLAGS_trace_directory;
+ status = replayer.ReplayDirectory(FLAGS_trace_directory);
+ }
+ if (!status.ok()) CLEAN_EXIT_WITH_ERROR(status.GetDetail());
+
+ CacheHitStatistics original_trace_stats = replayer.GetOriginalTraceStatistics();
+ CacheHitStatistics replay_stats = replayer.GetReplayStatistics();
+
+ if (FLAGS_output_file.size() != 0) {
+ DumpStatisticsToJSON(original_trace_stats, replay_stats, FLAGS_output_file);
+ } else {
+ LOG(INFO) << "Cache hit statistics from the original trace:";
+ DumpStatisticsToLog(original_trace_stats);
+ LOG(INFO) << "Cache hit statistics from the replay:";
+ DumpStatisticsToLog(replay_stats);
+ }
+ return 0;
+}
diff --git a/be/src/runtime/io/data-cache-trace-test.cc b/be/src/runtime/io/data-cache-trace-test.cc
new file mode 100644
index 0000000..31ced1d
--- /dev/null
+++ b/be/src/runtime/io/data-cache-trace-test.cc
@@ -0,0 +1,338 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "runtime/io/data-cache-trace.h"
+
+#include <boost/filesystem.hpp>
+#include <vector>
+
+#include "common/status.h"
+#include "kudu/util/slice.h"
+#include "testutil/gtest-util.h"
+
+#include "common/names.h"
+
+namespace impala {
+namespace io {
+namespace trace {
+
+using boost::filesystem::path;
+using kudu::Slice;
+using strings::Substitute;
+
+// Basic defaults for number of log files and entries per log file. These do not
+// impact the correctness of tests.
+static const int MAX_LOG_FILES = 10;
+static const size_t MAX_ENTRIES_PER_FILE = 100;
+
+class DataCacheTraceTest : public testing::Test {
+protected:
+ DataCacheTraceTest() : tmp_dir_("/tmp" / boost::filesystem::unique_path()) {}
+
+ path tmp_dir() const { return tmp_dir_; }
+
+ void SetUp() override {
+ boost::filesystem::create_directories(tmp_dir());
+ }
+
+ void TearDown() override {
+ boost::filesystem::remove_all(tmp_dir());
+ }
+
+ // Helper function to create a tracer without thinking about the number of files
+ // or entries per file.
+ unique_ptr<Tracer> CreateSimpleTracer(path directory, bool anonymize_trace = false) {
+ return std::make_unique<Tracer>(directory.string(), MAX_ENTRIES_PER_FILE,
+ MAX_LOG_FILES, anonymize_trace);
+ }
+
+ // Since most tests don't need to modify most fields, this provides a basic TraceEvent
+ // to use as a template.
+ TraceEvent GetTemplateTraceEvent() {
+ TraceEvent event;
+ event.type = EventType::HIT;
+ event.timestamp = 1;
+ event.filename = "fname.txt";
+ event.mtime = 1;
+ event.offset = 0;
+ // Entry length and offset length should be filled in by caller
+ event.entry_length = -1;
+ event.lookup_length = -1;
+
+ return event;
+ }
+
+ // Helper function to trace from a TraceEvent, allowing simpler test code
+ void TraceFromTraceEvent(Tracer* tracer, const TraceEvent& event) {
+ tracer->Trace(event.type, event.timestamp, Slice(event.filename), event.mtime,
+ event.offset, event.lookup_length, event.entry_length);
+ }
+
+ // This constructs valid trace events of all types. It uses unique numbers for each
+ // numeric field unless there is a required value (i.e. -1 for some event types).
+ std::vector<TraceEvent> GetSampleTraceEvents() {
+ std::vector<TraceEvent> events;
+ events.emplace_back(EventType::HIT, 1, "hit_filename.txt", 2, 3, 4, 5);
+ events.emplace_back(EventType::MISS, 6, "miss_filename.txt", 7, 8,
+ /* entry_length */-1, 9);
+ events.emplace_back(EventType::STORE, 10, "store_filename.txt", 11, 12, 13,
+ /* lookup_length */-1);
+ events.emplace_back(EventType::STORE_FAILED, 14, "store_failed.txt", 15, 16, 17,
+ /* lookup_length */ -1);
+ events.emplace_back(EventType::STORE_FAILED_BUSY, 18, "store_failed_busy.txt",
+ 19, 20, 21, /* lookup_length */ -1);
+ return events;
+ }
+
+private:
+ // Temporary directory for the test run
+ path tmp_dir_;
+};
+
+TEST_F(DataCacheTraceTest, FromJSONFailures) {
+ TraceEvent event;
+ Status status;
+
+ // Empty JSON (missing required fields)
+ status = JsonToTraceEvent("{}", &event);
+ EXPECT_FALSE(status.ok());
+
+ // Invalid JSON
+ status = JsonToTraceEvent("{ x = 1234", &event);
+ EXPECT_FALSE(status.ok());
+
+ // Gibberish
+ status = JsonToTraceEvent("\\896437 a189047623fgdjkh", &event);
+ EXPECT_FALSE(status.ok());
+}
+
+TEST_F(DataCacheTraceTest, ToFromJSON) {
+ // Create an array of TraceEvents, go to JSON and back, verify the same
+ vector<TraceEvent> sample_events = GetSampleTraceEvents();
+ for (const TraceEvent& event : sample_events) {
+ string json = TraceEventToJson(event);
+ TraceEvent fromjson_event;
+ EXPECT_OK(JsonToTraceEvent(json, &fromjson_event));
+ EXPECT_TRUE(event == fromjson_event);
+ }
+}
+
+TEST_F(DataCacheTraceTest, TraceReplayBasic1) {
+ // Trace repeated hits to the same file, then replay them and verify appropriate
+ // cache hits.
+ path basictest_path = tmp_dir() / "basic1";
+ unique_ptr<Tracer> tracer = CreateSimpleTracer(basictest_path);
+ ASSERT_OK(tracer->Init());
+ TraceEvent event = GetTemplateTraceEvent();
+
+ // Five hits to the same location (provided by the template)
+ event.type = EventType::HIT;
+ event.lookup_length = 1024;
+ event.entry_length = 1024;
+ for (int i = 0; i < 5; ++i) {
+ TraceFromTraceEvent(tracer.get(), event);
+ }
+ tracer->Flush();
+
+ TraceReplayer replayer("/tmp:50MB");
+ EXPECT_OK(replayer.Init());
+ EXPECT_OK(replayer.ReplayDirectory(basictest_path.string()));
+
+ // Every event was a HIT in the original trace
+ CacheHitStatistics original_trace_stats = replayer.GetOriginalTraceStatistics();
+ EXPECT_EQ(original_trace_stats.hits, 5);
+ EXPECT_EQ(original_trace_stats.hit_bytes, 5120);
+ EXPECT_EQ(original_trace_stats.partial_hits, 0);
+ EXPECT_EQ(original_trace_stats.misses, 0);
+ EXPECT_EQ(original_trace_stats.miss_bytes, 0);
+ EXPECT_EQ(original_trace_stats.stores, 0);
+ EXPECT_EQ(original_trace_stats.failed_stores, 0);
+
+ // The replay stats are different from the original trace stats. The first access to
+ // the location is a HIT in the original trace, but the replay doesn't have an entry
+ // for the location yet. So, the replay has a MISS and then a STORE for the first
+ // access to the location. The rest of the entries are the same.
+ CacheHitStatistics replay_stats = replayer.GetReplayStatistics();
+ EXPECT_EQ(replay_stats.hits, 4);
+ EXPECT_EQ(replay_stats.hit_bytes, 4096);
+ EXPECT_EQ(replay_stats.partial_hits, 0);
+ EXPECT_EQ(replay_stats.misses, 1);
+ EXPECT_EQ(replay_stats.miss_bytes, 1024);
+ EXPECT_EQ(replay_stats.stores, 1);
+ EXPECT_EQ(replay_stats.failed_stores, 0);
+}
+
+TEST_F(DataCacheTraceTest, TraceReplayBasic2) {
+ // Trace and replay trace events that reference unique files (and thus won't be hits)
+ path basictest_path = tmp_dir() / "basic2";
+ unique_ptr<Tracer> tracer = CreateSimpleTracer(basictest_path);
+ ASSERT_OK(tracer->Init());
+ TraceEvent event = GetTemplateTraceEvent();
+
+ // Five misses to different files
+ event.type = EventType::MISS;
+ event.lookup_length = 1024;
+ event.entry_length = -1;
+ for (int i = 0; i < 5; ++i) {
+ event.filename = Substitute("file$0.txt", i);
+ TraceFromTraceEvent(tracer.get(), event);
+ }
+ tracer->Flush();
+
+ TraceReplayer replayer("/tmp:50MB");
+ EXPECT_OK(replayer.Init());
+ EXPECT_OK(replayer.ReplayDirectory(basictest_path.string()));
+
+ CacheHitStatistics original_trace_stats = replayer.GetOriginalTraceStatistics();
+ EXPECT_EQ(original_trace_stats.hits, 0);
+ EXPECT_EQ(original_trace_stats.hit_bytes, 0);
+ EXPECT_EQ(original_trace_stats.partial_hits, 0);
+ EXPECT_EQ(original_trace_stats.misses, 5);
+ EXPECT_EQ(original_trace_stats.miss_bytes, 5120);
+ EXPECT_EQ(original_trace_stats.stores, 0);
+ EXPECT_EQ(original_trace_stats.failed_stores, 0);
+
+ // When the replayer sees a cache miss, it stores the entry to the cache, so the
+ // replay trace stats will have an additional 5 STOREs compared to the original
+ // trace stats.
+ CacheHitStatistics replay_stats = replayer.GetReplayStatistics();
+ EXPECT_EQ(replay_stats.hits, 0);
+ EXPECT_EQ(replay_stats.hit_bytes, 0);
+ EXPECT_EQ(replay_stats.partial_hits, 0);
+ EXPECT_EQ(replay_stats.misses, 5);
+ EXPECT_EQ(replay_stats.miss_bytes, 5120);
+ EXPECT_EQ(replay_stats.stores, 5);
+ EXPECT_EQ(replay_stats.failed_stores, 0);
+}
+
+TEST_F(DataCacheTraceTest, TraceReplayIgnoredEvents) {
+ // The trace replayer only replays HIT/MISS events. Verify that other events are
+ // ignored.
+ path ignoredevents_path = tmp_dir() / "ignoredevents";
+ unique_ptr<Tracer> tracer = CreateSimpleTracer(ignoredevents_path);
+ ASSERT_OK(tracer->Init());
+ TraceEvent event = GetTemplateTraceEvent();
+
+ // Normal store
+ event.type = EventType::STORE;
+ event.lookup_length = -1;
+ event.entry_length = 1024;
+ TraceFromTraceEvent(tracer.get(), event);
+
+ // Failed store (non-busy)
+ event.type = EventType::STORE_FAILED;
+ event.lookup_length = -1;
+ event.entry_length = 1024;
+ TraceFromTraceEvent(tracer.get(), event);
+
+ // Failed stored (busy)
+ event.type = EventType::STORE_FAILED_BUSY;
+ event.lookup_length = -1;
+ event.entry_length = 1024;
+ TraceFromTraceEvent(tracer.get(), event);
+ tracer->Flush();
+
+ TraceReplayer replayer("/tmp:50MB");
+ EXPECT_OK(replayer.Init());
+ EXPECT_OK(replayer.ReplayDirectory(ignoredevents_path.string()));
+ CacheHitStatistics original_trace_stats = replayer.GetOriginalTraceStatistics();
+ EXPECT_EQ(original_trace_stats.hits, 0);
+ EXPECT_EQ(original_trace_stats.hit_bytes, 0);
+ EXPECT_EQ(original_trace_stats.partial_hits, 0);
+ EXPECT_EQ(original_trace_stats.misses, 0);
+ EXPECT_EQ(original_trace_stats.miss_bytes, 0);
+ EXPECT_EQ(original_trace_stats.stores, 1);
+ EXPECT_EQ(original_trace_stats.failed_stores, 2);
+
+ // The replayer only cares about HIT and MISS events, so the replay trace stats
+ // show nothing.
+ CacheHitStatistics replay_stats = replayer.GetReplayStatistics();
+ EXPECT_EQ(replay_stats.hits, 0);
+ EXPECT_EQ(replay_stats.hit_bytes, 0);
+ EXPECT_EQ(replay_stats.partial_hits, 0);
+ EXPECT_EQ(replay_stats.misses, 0);
+ EXPECT_EQ(replay_stats.miss_bytes, 0);
+ EXPECT_EQ(replay_stats.stores, 0);
+ EXPECT_EQ(replay_stats.failed_stores, 0);
+}
+
+TEST_F(DataCacheTraceTest, TraceReplayPartialHits) {
+ // The trace replayer needs to replicate the data cache's behaviors for partial
+ // hits.
+ path partialhits_path = tmp_dir() / "partialhits";
+ unique_ptr<Tracer> tracer = CreateSimpleTracer(partialhits_path);
+ ASSERT_OK(tracer->Init());
+ TraceEvent event = GetTemplateTraceEvent();
+
+ // Initial miss loads small entry (512 bytes)
+ event.type = EventType::MISS;
+ event.lookup_length = 512;
+ event.entry_length = -1;
+ TraceFromTraceEvent(tracer.get(), event);
+
+ // Store this small entry (skipped by replayer, but mimics what is happening)
+ event.type = EventType::STORE;
+ event.lookup_length = -1;
+ event.entry_length = 512;
+ TraceFromTraceEvent(tracer.get(), event);
+
+ // Trying to read a larger element at the same location is a partial hit
+ event.type = EventType::HIT;
+ event.lookup_length = 1024;
+ event.entry_length = 512;
+ TraceFromTraceEvent(tracer.get(), event);
+
+ // Store the larger element (skipped by replayer, but mimics what is happening)
+ event.type = EventType::STORE;
+ event.lookup_length = -1;
+ event.entry_length = 1024;
+ TraceFromTraceEvent(tracer.get(), event);
+
+ // Trying to read the shorter length again is a hit
+ event.type = EventType::HIT;
+ event.lookup_length = 512;
+ event.entry_length = 1024;
+ TraceFromTraceEvent(tracer.get(), event);
+
+ tracer->Flush();
+
+ TraceReplayer replayer("/tmp:50MB");
+ EXPECT_OK(replayer.Init());
+ EXPECT_OK(replayer.ReplayDirectory(partialhits_path.string()));
+ CacheHitStatistics original_trace_stats = replayer.GetOriginalTraceStatistics();
+ EXPECT_EQ(original_trace_stats.hits, 1);
+ EXPECT_EQ(original_trace_stats.hit_bytes, 1024);
+ EXPECT_EQ(original_trace_stats.partial_hits, 1);
+ EXPECT_EQ(original_trace_stats.misses, 1);
+ EXPECT_EQ(original_trace_stats.miss_bytes, 1024);
+ EXPECT_EQ(original_trace_stats.stores, 2);
+ EXPECT_EQ(original_trace_stats.failed_stores, 0);
+
+ // The events above were created in a way that the original trace is identical to
+ // what the replay will do. So, the stats are identical.
+ CacheHitStatistics replay_stats = replayer.GetReplayStatistics();
+ EXPECT_EQ(replay_stats.hits, 1);
+ EXPECT_EQ(replay_stats.hit_bytes, 1024);
+ EXPECT_EQ(replay_stats.partial_hits, 1);
+ EXPECT_EQ(replay_stats.misses, 1);
+ EXPECT_EQ(replay_stats.miss_bytes, 1024);
+ EXPECT_EQ(replay_stats.stores, 2);
+ EXPECT_EQ(replay_stats.failed_stores, 0);
+}
+} // namespace trace
+} // namespace io
+} // namespace impala
diff --git a/be/src/runtime/io/data-cache-trace.cc b/be/src/runtime/io/data-cache-trace.cc
new file mode 100644
index 0000000..68bf31d
--- /dev/null
+++ b/be/src/runtime/io/data-cache-trace.cc
@@ -0,0 +1,374 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "runtime/io/data-cache-trace.h"
+
+#include <boost/filesystem.hpp>
+#include <fstream>
+#include <glog/logging.h>
+#include <rapidjson/document.h>
+#include <rapidjson/error/en.h>
+#include <rapidjson/ostreamwrapper.h>
+#include <rapidjson/writer.h>
+#include <string>
+
+#include "gutil/hash/city.h"
+#include "gutil/strings/escaping.h"
+#include "kudu/util/async_logger.h"
+#include "kudu/util/jsonwriter.h"
+#include "kudu/util/path_util.h"
+#include "runtime/io/data-cache.h"
+#include "util/filesystem-util.h"
+#include "util/simple-logger.h"
+
+#include "common/names.h"
+
+using kudu::Slice;
+using kudu::JoinPathSegments;
+using strings::Substitute;
+
+namespace impala {
+namespace io {
+namespace trace {
+
+const string TRACE_FILE_PREFIX = "impala_cache_trace-";
+
+EventType StringToEventType(std::string s) {
+ if (s == "H") return EventType::HIT;
+ if (s == "M") return EventType::MISS;
+ if (s == "S") return EventType::STORE;
+ if (s == "B") return EventType::STORE_FAILED_BUSY;
+ if (s == "F") return EventType::STORE_FAILED;
+ CHECK(false) << "Unknown EventType string: " << s;
+ return EventType::HIT;
+}
+
+std::string EventTypeToString(EventType s) {
+ switch (s) {
+ case EventType::HIT:
+ return "H";
+ case EventType::MISS:
+ return "M";
+ case EventType::STORE:
+ return "S";
+ case EventType::STORE_FAILED_BUSY:
+ return "B";
+ case EventType::STORE_FAILED:
+ return "F";
+ default:
+ CHECK(false) << "Unknown EventType";
+ return "Invalid";
+ }
+}
+
+Status JsonToTraceEvent(std::string json, TraceEvent* event) {
+ rapidjson::Document d;
+ d.Parse<0>(json.c_str());
+ // Check for any parse failure
+ if (d.HasParseError()) {
+ return Status(Substitute("Failed to parse TraceEvent JSON. Error: $0 JSON: $1",
+ rapidjson::GetParseError_En(d.GetParseError()), json));
+ }
+ // Verify that this is an object and the required fields are present
+ if (!d.IsObject() || !d.HasMember("ts") || !d.HasMember("s") || !d.HasMember("f") ||
+ !d.HasMember("m") || !d.HasMember("o")) {
+ return Status(Substitute("Invalid TraceEvent JSON: $0", json));
+ }
+ event->timestamp = d["ts"].GetDouble();
+ event->type = StringToEventType(d["s"].GetString());
+ event->filename = d["f"].GetString();
+ event->mtime = d["m"].GetInt64();
+ event->offset = d["o"].GetInt64();
+ event->entry_length = d.HasMember("eLen") ? d["eLen"].GetInt64() : -1;
+ event->lookup_length = d.HasMember("lLen") ? d["lLen"].GetInt64() : -1;
+ return Status::OK();
+}
+
+std::string TraceEventToJson(const TraceEvent& event) {
+ ostringstream buf;
+ kudu::JsonWriter jw(&buf, kudu::JsonWriter::COMPACT);
+
+ jw.StartObject();
+ jw.String("ts");
+ jw.Double(event.timestamp);
+ jw.String("s");
+ jw.String(EventTypeToString(event.type));
+ jw.String("f");
+ jw.String(event.filename);
+ jw.String("m");
+ jw.Int64(event.mtime);
+ jw.String("o");
+ jw.Int64(event.offset);
+
+ if (event.lookup_length != -1) {
+ jw.String("lLen");
+ jw.Int64(event.lookup_length);
+ }
+ if (event.entry_length != -1) {
+ jw.String("eLen");
+ jw.Int64(event.entry_length);
+ }
+ jw.EndObject();
+
+ return buf.str();
+}
+
+Status TraceFileIterator::Init() {
+ boost::filesystem::path file_path(filename_);
+ if (!boost::filesystem::exists(file_path)) {
+ return Status(Substitute("Trace file does not exist: $0", filename_));
+ }
+ stream_.open(filename_);
+ initialized_ = true;
+ return Status::OK();
+}
+
+Status TraceFileIterator::GetNextEvent(TraceEvent* event, bool* done) {
+ string line;
+ DCHECK(initialized_);
+ DCHECK(event != nullptr);
+ DCHECK(done != nullptr);
+ if (getline(stream_, line)) {
+ RETURN_IF_ERROR(JsonToTraceEvent(line, event));
+ *done = false;
+ } else {
+ *done = true;
+ }
+ return Status::OK();
+}
+
+// Simple implementation of a glog Logger that writes to a file, used for
+// cache access tracing.
+//
+// This doesn't fully implement the Logger interface -- only the bare minimum
+// to be usable with kudu::AsyncLogger.
+class SimpleLoggerWrapper : public google::base::Logger {
+ public:
+ explicit SimpleLoggerWrapper(string log_dir, string log_file_name_prefix,
+ uint64_t max_entries_per_file, int32_t max_log_files)
+ : logger_(log_dir, log_file_name_prefix, max_entries_per_file, max_log_files) {}
+
+ virtual ~SimpleLoggerWrapper() {
+ Flush();
+ }
+
+ Status Open() {
+ return logger_.Init();
+ }
+
+ void Write(bool force_flush,
+ time_t timestamp,
+ const char* message,
+ int message_len) override {
+ string message_str(message, message_len);
+ Status status = logger_.AppendEntry(message_str);
+ if (!status.ok()) {
+ LOG_EVERY_N(WARNING, 1000) << "Could not write to data cache access trace ("
+ << google::COUNTER << " attempts failed): " << status.GetDetail();
+ }
+ if (force_flush) Flush();
+ }
+
+ // Flush any buffered messages.
+ // NOTE: declared 'final' to allow safe calls from the destructor.
+ void Flush() override final {
+ Status status = logger_.Flush();
+ if (!status.ok()) {
+ LOG_EVERY_N(WARNING, 1000) << "Could not write to data cache access trace ("
+ << google::COUNTER << " attempts failed): " << status.GetDetail();
+ }
+ }
+
+ uint32 LogSize() override {
+ LOG(FATAL) << "Unimplemented";
+ return 0;
+ }
+
+ private:
+ SimpleLogger logger_;
+};
+
+Tracer::Tracer(string log_dir, uint64_t max_entries_per_file, int32_t max_log_files,
+ bool anonymize_trace)
+ : underlying_logger_(new SimpleLoggerWrapper(log_dir, TRACE_FILE_PREFIX,
+ max_entries_per_file, max_log_files)),
+ anonymize_trace_(anonymize_trace) {}
+
+Tracer::~Tracer() {
+ if (logger_) logger_->Stop();
+}
+
+Status Tracer::Init() {
+ RETURN_IF_ERROR(underlying_logger_->Open());
+ logger_.reset(new kudu::AsyncLogger(underlying_logger_.get(), 8 * 1024 * 1024));
+ logger_->Start();
+ return Status::OK();
+}
+
+void Tracer::Flush() {
+ logger_->Flush();
+}
+
+void Tracer::Trace(EventType type, double timestamp, Slice filename, int64_t mtime,
+ int64_t offset, int64_t lookup_len, int64_t entry_len) {
+
+ // Sanity checks for lookup_len / entry_len
+ if (type == EventType::HIT) {
+ DCHECK(lookup_len > 0 && entry_len > 0);
+ } else if (type == EventType::MISS) {
+ DCHECK(lookup_len > 0 && entry_len == -1);
+ } else if (type == EventType::STORE || type == EventType::STORE_FAILED ||
+ type == EventType::STORE_FAILED_BUSY) {
+ DCHECK(lookup_len == -1 && entry_len > 0);
+ } else {
+ DCHECK(false) << "Unrecognized EventType";
+ }
+
+ string filename_str =
+ anonymize_trace_ ? AnonymizeFilename(filename) : filename.ToString();
+ TraceEvent event(type, timestamp, filename_str, mtime, offset, entry_len,
+ lookup_len);
+ string json = TraceEventToJson(event);
+ logger_->Write(/*force_flush="*/false, /*timestamp=*/0, json.data(), json.size());
+}
+
+string Tracer::AnonymizeFilename(Slice filename) {
+ uint128 hash = util_hash::CityHash128(reinterpret_cast<const char*>(filename.data()),
+ filename.size());
+ // A 128-bit (16-byte) hash results in a 22-byte base64-encoded string. We opt to
+ // generate the string without the typical two characters of padding.
+ const int ESCAPED_LEN = 22;
+ DCHECK_EQ(ESCAPED_LEN, CalculateBase64EscapedLen(sizeof(hash), /* padding */ false));
+ string b64_out;
+ Base64Escape(reinterpret_cast<const unsigned char*>(&hash), sizeof(hash),
+ &b64_out, /* padding */ false);
+ DCHECK_EQ(b64_out.size(), ESCAPED_LEN);
+ return b64_out;
+}
+
+TraceReplayer::TraceReplayer(string trace_configuration)
+ : trace_configuration_(trace_configuration) {}
+
+TraceReplayer::~TraceReplayer() {}
+
+Status TraceReplayer::Init() {
+ data_cache_.reset(new DataCache(trace_configuration_, /* trace_replay */ true));
+ RETURN_IF_ERROR(data_cache_->Init());
+ initialized_ = true;
+ return Status::OK();
+}
+
+Status TraceReplayer::ReplayFile(string filename) {
+ DCHECK(initialized_);
+ TraceFileIterator file_iter(filename);
+ RETURN_IF_ERROR(file_iter.Init());
+ while (true) {
+ bool done = false;
+ TraceEvent trace_event;
+ RETURN_IF_ERROR(file_iter.GetNextEvent(&trace_event, &done));
+ if (done) break;
+ ReplayEntry(trace_event);
+ }
+ return Status::OK();
+}
+
+Status TraceReplayer::ReplayDirectory(string directory) {
+ DCHECK(initialized_);
+ vector<string> trace_files;
+ RETURN_IF_ERROR(SimpleLogger::GetLogFiles(directory, TRACE_FILE_PREFIX, &trace_files));
+ for (const string& trace_file : trace_files) {
+ RETURN_IF_ERROR(ReplayFile(trace_file));
+ }
+ return Status::OK();
+}
+
+void TraceReplayer::UpdateTraceStats(const TraceEvent& entry) {
+ switch (entry.type) {
+ case EventType::STORE:
+ ++original_trace_stats_.stores;
+ break;
+ case EventType::STORE_FAILED:
+ case EventType::STORE_FAILED_BUSY:
+ ++original_trace_stats_.failed_stores;
+ break;
+ case EventType::HIT:
+ if (entry.entry_length >= entry.lookup_length) {
+ ++original_trace_stats_.hits;
+ original_trace_stats_.hit_bytes += entry.lookup_length;
+ } else {
+ ++original_trace_stats_.partial_hits;
+ original_trace_stats_.hit_bytes += entry.entry_length;
+ original_trace_stats_.miss_bytes += entry.lookup_length - entry.entry_length;
+ }
+ break;
+ case EventType::MISS:
+ ++original_trace_stats_.misses;
+ original_trace_stats_.miss_bytes += entry.lookup_length;
+ break;
+ default:
+ CHECK(false) << "Invalid TraceEvent";
+ }
+}
+
+void TraceReplayer::ReplayEntry(const TraceEvent& entry) {
+ // First, update hit/miss counts as reported by the trace itself.
+ UpdateTraceStats(entry);
+
+ // Second, do the actual replay against the current cache (which may have different
+ // settings from the original cache). This mirrors the behavior of DiskIoMgr.
+ // Replay only needs hits and misses.
+ if (entry.type != EventType::HIT && entry.type != EventType::MISS) {
+ return;
+ }
+ DCHECK_GT(entry.lookup_length, 0);
+ // Try to read the whole chunk from the cache. If it does a partial read,
+ // the rest is a miss, but it will try to store the complete read into the cache.
+ int64_t bytes_read = data_cache_->Lookup(entry.filename, entry.mtime, entry.offset,
+ entry.lookup_length, /* buffer */ nullptr);
+ DCHECK_LE(bytes_read, entry.lookup_length);
+ if (bytes_read == 0) {
+ // Complete miss, and we try to store the whole chunk into the cache
+ ++replay_stats_.misses;
+ replay_stats_.miss_bytes += entry.lookup_length;
+ bool success = data_cache_->Store(entry.filename, entry.mtime, entry.offset,
+ /* buffer */ nullptr, entry.lookup_length);
+ if (success) {
+ ++replay_stats_.stores;
+ } else {
+ ++replay_stats_.failed_stores;
+ }
+ } else if (bytes_read == entry.lookup_length) {
+ // Total hit, nothing to store to the cache
+ ++replay_stats_.hits;
+ replay_stats_.hit_bytes += bytes_read;
+ } else {
+ // Partial hit, store complete entry to the cache
+ ++replay_stats_.partial_hits;
+ replay_stats_.hit_bytes += bytes_read;
+ replay_stats_.miss_bytes += entry.lookup_length - bytes_read;
+ bool success = data_cache_->Store(entry.filename, entry.mtime, entry.offset,
+ /* buffer */ nullptr, entry.lookup_length);
+ if (success) {
+ ++replay_stats_.stores;
+ } else {
+ ++replay_stats_.failed_stores;
+ }
+ }
+}
+} // namespace trace
+} // namespace io
+} // namespace impala
diff --git a/be/src/runtime/io/data-cache-trace.h b/be/src/runtime/io/data-cache-trace.h
new file mode 100644
index 0000000..7cb226b
--- /dev/null
+++ b/be/src/runtime/io/data-cache-trace.h
@@ -0,0 +1,247 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <fstream>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "kudu/util/slice.h"
+
+namespace kudu {
+ class AsyncLogger;
+}
+
+namespace impala {
+namespace io {
+
+class DataCache;
+
+// When working on data cache tuning or optimization, the actual cache eviction behavior
+// is only a function of the metadata operations (which file locations are accessed in
+// what order) and do not require actual data. Since the data for caching workloads is
+// often large and sensitive, the data cache trace provides a way to improve cache
+// policies without access to the original data/workloads. It also reduces the cost of
+// testing the cache, since the metadata operations are significantly faster than the
+// data operations.
+//
+// The DataCache is instrumented to trace interesting events (e.g. hits misses, stores,
+// etc). This file implements the Tracer that writes the events out to disk and a
+// corresponding TraceReplayer which can read the events and replay them against a
+// cache with a particular configuration.
+namespace trace {
+
+class SimpleLoggerWrapper;
+
+enum class EventType {
+ HIT,
+ MISS,
+ STORE,
+ STORE_FAILED_BUSY,
+ STORE_FAILED
+};
+
+// Functions to convert EventType to a string representation and back
+EventType StringToEventType(std::string s);
+std::string EventTypeToString(EventType s);
+
+struct TraceEvent {
+ TraceEvent(EventType type, double timestamp, std::string filename, int64_t mtime,
+ int64_t offset, int64_t entry_length, int64_t lookup_length)
+ : type(type), timestamp(timestamp), filename(filename), mtime(mtime),
+ offset(offset), entry_length(entry_length), lookup_length(lookup_length) {}
+
+ TraceEvent() = default;
+
+ // Type of cache event: HIT, MISS, etc.
+ EventType type;
+
+ // Floating point timestamp of the event in seconds (see WallTime_Now())
+ double timestamp;
+
+ // These fields are the standard pieces of a cache key: filename, mtime, offset
+ // The one distinction is that the filename can be anonymized by being hashed.
+ // See Tracer::AnonymizeFilename()
+ std::string filename;
+ int64_t mtime;
+ int64_t offset;
+
+ // The 'entry_length' is the size of the cache entry. For a HIT, this may be
+ // larger or smaller than the lookup length. For a MISS, this is always -1.
+ // For STORE*, this is the size of the cache entry being added.
+ int64_t entry_length;
+ // The lookup length is the size requested by the cache user. For a HIT/MISS,
+ // this is the total size that the user requested. For STORE*, this is always -1.
+ int64_t lookup_length;
+
+ // Implement the == operator to make test logic cleaner
+ bool operator==(const TraceEvent& rhs) const {
+ return type == rhs.type && timestamp == rhs.timestamp &&
+ filename == rhs.filename && mtime == rhs.mtime && offset == rhs.offset &&
+ entry_length == rhs.entry_length && lookup_length == rhs.lookup_length;
+ }
+};
+
+// Parse the json input and fill in a TraceEvent representing the data
+Status JsonToTraceEvent(std::string json, TraceEvent* event);
+
+// Construct a JSON entry from the provided TraceEvent
+std::string TraceEventToJson(const TraceEvent& event);
+
+/// The file name prefix pattern used for the access trace.
+extern const string TRACE_FILE_PREFIX;
+
+class Tracer {
+ public:
+ /// The Tracer records data cache trace events to a representation on disk.
+ /// 'log_dir' is the directory that the Tracer uses for output. It is using
+ /// the SimpleLogger under the covers, so it obeys the 'max_entries_per_file'
+ /// and 'max_trace_files' to limit the total number of entries on disk.
+ /// If 'anonymize_trace' is true, the filenames in the trace are anonymized
+ /// by hashing them (see AnonymizeFilename()).
+ explicit Tracer(std::string log_dir, uint64_t max_entries_per_file,
+ int32_t max_trace_files, bool anonymize_trace);
+
+ ~Tracer();
+
+ // Initialize the tracer
+ Status Init();
+
+ // Force the logger to flush its buffers
+ void Flush();
+
+ // Record an entry into the log
+ void Trace(EventType type, double timestamp, kudu::Slice filename,
+ int64_t mtime, int64_t offset, int64_t lookup_len, int64_t entry_len);
+
+ private:
+ // The underlying logger that we wrap with the AsyncLogger wrapper
+ // 'logger_'. NOTE: AsyncLogger consumes a raw pointer which must
+ // outlive the AsyncLogger instance, so it's important that these
+ // are declared in this order (logger_ must destruct before
+ // underlying_logger_).
+ std::unique_ptr<SimpleLoggerWrapper> underlying_logger_;
+ // The async wrapper around underlying_logger_ (see above).
+ std::unique_ptr<kudu::AsyncLogger> logger_;
+
+ // Whether to anonymize the trace by hashing the filenames
+ bool anonymize_trace_;
+
+ // This anonymizes the filename by hashing it and encoding that in a Base64 string.
+ // This returns a 22 character string.
+ std::string AnonymizeFilename(kudu::Slice filename);
+};
+
+// Information about the cache outcomes for a particular workload
+struct CacheHitStatistics {
+ uint64_t hits = 0;
+ uint64_t partial_hits = 0;
+ uint64_t hit_bytes = 0;
+ uint64_t misses = 0;
+ uint64_t miss_bytes = 0;
+ uint64_t stores = 0;
+ uint64_t failed_stores = 0;
+};
+
+// The access trace has a TraceEvent JSON entry per line in the file. This is a helper
+// class to iterate over the TraceEvents from a single file.
+class TraceFileIterator {
+ public:
+ TraceFileIterator(std::string filename)
+ : filename_(filename) {}
+
+ // Initialize the iterator
+ Status Init();
+
+ // Get the next TraceEvent. If not at the end of the file, 'event' is filled with
+ // the next TraceEvent. If at the end of the file, 'done' is set to true and
+ // 'event' is untouched. Returns status in the event of an error.
+ Status GetNextEvent(TraceEvent* event, bool* done);
+
+ private:
+ // File to iterate over
+ std::string filename_;
+ // Set to true in Init()
+ bool initialized_ = false;
+ // Input stream reading the file
+ std::ifstream stream_;
+};
+
+class TraceReplayer {
+ public:
+ // The TraceReplayer creates a DataCache with the provided 'trace_configuration'
+ // in the special trace replay mode. This allows replaying the trace without any
+ // underlying data or filesystem operations. The 'trace_configuration' matches the
+ // configuration format from the DataCache constructor. See data-cache.h.
+ TraceReplayer(std::string trace_configuration);
+
+ ~TraceReplayer();
+
+ // Initialize the replayer
+ Status Init();
+
+ // Replay a single trace file
+ Status ReplayFile(std::string filename);
+
+ // Replay a directory of trace files as generated by the Tracer above.
+ Status ReplayDirectory(std::string directory);
+
+ // Get the hit statistics that are recorded in the original trace.
+ CacheHitStatistics GetOriginalTraceStatistics() const {
+ DCHECK(initialized_);
+ return original_trace_stats_;
+ }
+
+ // Get the hit statistics for the replay of the trace.
+ CacheHitStatistics GetReplayStatistics() const {
+ DCHECK(initialized_);
+ return replay_stats_;
+ }
+
+ private:
+
+ // Update the original trace statistics given this TraceEvent
+ void UpdateTraceStats(const TraceEvent& entry);
+
+ // Replay an individual TraceEvent against the current cache.
+ void ReplayEntry(const TraceEvent& entry);
+
+ // Configuration of the cache used for the replay
+ std::string trace_configuration_;
+
+ // Set to true in Init().
+ bool initialized_ = false;
+
+ // DataCache used for the trace replay
+ std::unique_ptr<DataCache> data_cache_;
+
+ // The hit statistics that are recorded in the original trace. This is
+ // purely a summation of the contents in the trace (i.e. hits, misses, etc).
+ CacheHitStatistics original_trace_stats_;
+
+ // The hit statistics for the replay of the trace. The outcomes of the
+ // trace replay are independent from the original trace and are dependent
+ // on the cache configuration. Even with equivalent cache configurations, the
+ // outcome can differ from the original trace due to concurrency.
+ CacheHitStatistics replay_stats_;
+};
+} // namespace trace
+} // namespace io
+} // namespace impala
diff --git a/be/src/runtime/io/data-cache.cc b/be/src/runtime/io/data-cache.cc
index febb6d9..9d86784 100644
--- a/be/src/runtime/io/data-cache.cc
+++ b/be/src/runtime/io/data-cache.cc
@@ -26,17 +26,15 @@
#include <glog/logging.h>
+#include "common/compiler-util.h"
#include "exec/kudu-util.h"
-#include "kudu/util/async_logger.h"
#include "kudu/util/env.h"
-#include "kudu/util/jsonwriter.h"
#include "kudu/util/locks.h"
#include "kudu/util/path_util.h"
-#include "gutil/hash/city.h"
#include "gutil/port.h"
-#include "gutil/strings/escaping.h"
#include "gutil/strings/split.h"
#include "gutil/walltime.h"
+#include "runtime/io/data-cache-trace.h"
#include "util/bit-util.h"
#include "util/cache/cache.h"
#include "util/error-util.h"
@@ -91,6 +89,19 @@ DEFINE_bool(data_cache_enable_tracing, false,
DEFINE_bool(data_cache_anonymize_trace, false,
"(Advanced) Use hashes of filenames rather than file paths in the data "
"cache access trace.");
+DEFINE_int32(data_cache_trace_percentage, 100, "The percentage of cache lookups that "
+ "should be emitted to the trace file.");
+DECLARE_string(log_dir);
+DEFINE_string(data_cache_trace_dir, "", "The base directory for data cache tracing. "
+ "The data cache trace files for each cache directory are placed in separate "
+ "subdirectories underneath this base directory. If blank, defaults to "
+ "<log_file_dir>/data_cache_trace/");
+DEFINE_int32(max_data_cache_trace_file_size, 100000, "The maximum size (in "
+ "log entries) of the data cache trace file before a new one is created.");
+DEFINE_int32(max_data_cache_trace_files, 10, "Maximum number of data cache trace "
+ "files to retain for each cache directory specified by the data_cache startup "
+ "parameter. The most recent trace files are retained. If set to 0, all trace files "
+ "are retained.");
DEFINE_string(data_cache_eviction_policy, "LRU",
"(Advanced) The cache eviction policy to use for the data cache. "
@@ -101,7 +112,6 @@ namespace io {
static const int64_t PAGE_SIZE = 1L << 12;
const char* DataCache::Partition::CACHE_FILE_PREFIX = "impala-cache-file-";
-const char* DataCache::Partition::TRACE_FILE_NAME = "impala-cache-trace.txt";
const int MAX_FILE_DELETER_QUEUE_SIZE = 500;
static const char* PARTITION_PATH_METRIC_KEY_TEMPLATE =
"impala-server.io-mgr.remote-data-cache-partition-$0.path";
@@ -113,100 +123,6 @@ static const char* PARTITION_EVICTION_LATENCY_METRIC_KEY_TEMPLATE =
"impala-server.io-mgr.remote-data-cache-partition-$0.eviction-latency";
-namespace {
-
-class FileLogger;
-
-// Simple implementation of a glog Logger that writes to a file, used for
-// cache access tracing.
-//
-// This doesn't fully implement the Logger interface -- only the bare minimum
-// to be usable with kudu::AsyncLogger.
-class FileLogger : public google::base::Logger {
- public:
- explicit FileLogger(string path) : path_(std::move(path)) {}
-
- virtual ~FileLogger() {
- if (file_) Flush();
- }
-
- Status Open() {
- KUDU_RETURN_IF_ERROR(Env::Default()->NewWritableFile({}, path_, &file_),
- "Failed to create trace log file");
- return Status::OK();
- }
-
- void Write(bool force_flush,
- time_t timestamp,
- const char* message,
- int message_len) override {
- buf_.append(message, message_len);
- if (force_flush || buf_.size() > kBufSize) {
- Flush();
- }
- }
-
- // Flush any buffered messages.
- // NOTE: declared 'final' to allow safe calls from the destructor.
- void Flush() override final {
- if (buf_.empty()) return;
-
- KUDU_WARN_NOT_OK(file_->Append(buf_), "Could not append to trace log");
- buf_.clear();
- }
-
- uint32 LogSize() override {
- LOG(FATAL) << "Unimplemented";
- return 0;
- }
-
- private:
- const string path_;
- string buf_;
- unique_ptr<WritableFile> file_;
-
- static constexpr int kBufSize = 64*1024;
-};
-
-} // anonymous namespace
-
-class DataCache::Partition::Tracer {
- public:
- explicit Tracer(string path) : underlying_logger_(new FileLogger(std::move(path))) {}
-
- ~Tracer() {
- if (logger_) logger_->Stop();
- }
-
- Status Start() {
- RETURN_IF_ERROR(underlying_logger_->Open());
- logger_.reset(new kudu::AsyncLogger(underlying_logger_.get(), 8 * 1024 * 1024));
- logger_->Start();
- return Status::OK();
- }
-
- enum CacheStatus {
- HIT,
- MISS,
- STORE,
- STORE_FAILED_BUSY
- };
-
- void Trace(CacheStatus status, const DataCache::CacheKey& key,
- int64_t lookup_len, int64_t entry_len);
-
- private:
- // The underlying logger that we wrap with the AsyncLogger wrapper
- // 'logger_'. NOTE: AsyncLogger consumes a raw pointer which must
- // outlive the AsyncLogger instance, so it's important that these
- // are declared in this order (logger_ must destruct before
- // underlying_logger_).
- unique_ptr<FileLogger> underlying_logger_;
- // The async wrapper around underlying_logger_ (see above).
- unique_ptr<kudu::AsyncLogger> logger_;
-};
-
-
/// This class is an implementation of backing files in a cache partition.
///
/// A partition uses the interface Create() to create a backing file. A reader can read
@@ -462,11 +378,13 @@ static Cache::EvictionPolicy GetCacheEvictionPolicy(const std::string& policy_st
}
DataCache::Partition::Partition(
- int32_t index, const string& path, int64_t capacity, int max_opened_files)
+ int32_t index, const string& path, int64_t capacity, int max_opened_files,
+ bool trace_replay)
: index_(index),
path_(path),
capacity_(max<int64_t>(capacity, PAGE_SIZE)),
max_opened_files_(max_opened_files),
+ trace_replay_(trace_replay),
meta_cache_(NewCache(GetCacheEvictionPolicy(FLAGS_data_cache_eviction_policy),
capacity_, path_)) {}
@@ -475,6 +393,7 @@ DataCache::Partition::~Partition() {
}
Status DataCache::Partition::CreateCacheFile() {
+ DCHECK(!trace_replay_);
lock_.DCheckLocked();
const string& path =
JoinPathSegments(path_, CACHE_FILE_PREFIX + PrintId(GenerateUUID()));
@@ -486,6 +405,7 @@ Status DataCache::Partition::CreateCacheFile() {
}
Status DataCache::Partition::DeleteExistingFiles() const {
+ DCHECK(!trace_replay_);
vector<string> entries;
RETURN_IF_ERROR(FileSystemUtil::Directory::GetEntryNames(path_, &entries, 0,
FileSystemUtil::Directory::EntryType::DIR_ENTRY_REG));
@@ -505,6 +425,10 @@ Status DataCache::Partition::Init() {
RETURN_IF_ERROR(meta_cache_->Init());
+ // Trace replay does not require further initialization, as it is only doing
+ // metadata operations and does not do filesystem operations.
+ if (trace_replay_) return Status::OK();
+
// Verify the validity of the path specified.
if (!FileSystemUtil::IsCanonicalPath(path_)) {
return Status(Substitute("$0 is not a canonical path", path_));
@@ -529,8 +453,26 @@ Status DataCache::Partition::Init() {
RETURN_IF_ERROR(FileSystemUtil::CheckHolePunch(path_));
if (FLAGS_data_cache_enable_tracing) {
- tracer_.reset(new Tracer(path_ + "/" + TRACE_FILE_NAME));
- RETURN_IF_ERROR(tracer_->Start());
+ if (FLAGS_data_cache_trace_percentage > 100 ||
+ FLAGS_data_cache_trace_percentage < 0) {
+ return Status(Substitute("Misconfigured data_cache_trace_percentage: $0."
+ "Must be between 0 and 100.", FLAGS_data_cache_trace_percentage));
+ }
+
+ // If unspecified, use a directory under the "log_dir".
+ if (FLAGS_data_cache_trace_dir.empty()) {
+ stringstream default_trace_dir;
+ default_trace_dir << FLAGS_log_dir << "/data_cache_traces/";
+ FLAGS_data_cache_trace_dir = default_trace_dir.str();
+ }
+ // To avoid mixing trace files from different partitions, give each partition
+ // its own subdirectory.
+ stringstream trace_dir;
+ trace_dir << FLAGS_data_cache_trace_dir << Substitute("/partition-$0/", index_);
+ tracer_.reset(new trace::Tracer(trace_dir.str(),
+ FLAGS_max_data_cache_trace_file_size, FLAGS_max_data_cache_trace_files,
+ FLAGS_data_cache_anonymize_trace));
+ RETURN_IF_ERROR(tracer_->Init());
}
// Create metrics for this partition
@@ -588,6 +530,7 @@ void DataCache::Partition::InitMetrics() {
}
Status DataCache::Partition::CloseFilesAndVerifySizes() {
+ if (trace_replay_) return Status::OK();
int64_t total_size = 0;
for (auto& file : cache_files_) {
uint64_t sz_on_disk;
@@ -623,42 +566,42 @@ void DataCache::Partition::ReleaseResources() {
int64_t DataCache::Partition::Lookup(const CacheKey& cache_key, int64_t bytes_to_read,
uint8_t* buffer) {
DCHECK(!closed_);
+ DCHECK(trace_replay_ ? buffer == nullptr : buffer != nullptr);
Slice key = cache_key.ToSlice();
Cache::UniqueHandle handle(meta_cache_->Lookup(key));
if (handle.get() == nullptr) {
- if (tracer_ != nullptr) {
- tracer_->Trace(Tracer::MISS, cache_key, bytes_to_read, /*entry_len=*/-1);
- }
+ Trace(trace::EventType::MISS, cache_key, bytes_to_read, /*entry_len=*/-1);
return 0;
}
// Read from the backing file.
CacheEntry entry(meta_cache_->Value(handle));
- if (tracer_ != nullptr) {
- tracer_->Trace(Tracer::HIT, cache_key, bytes_to_read, entry.len());
- }
+ Trace(trace::EventType::HIT, cache_key, bytes_to_read, entry.len());
- CacheFile* cache_file = entry.file();
bytes_to_read = min(entry.len(), bytes_to_read);
- VLOG(3) << Substitute("Reading file $0 offset $1 len $2 checksum $3 bytes_to_read $4",
- cache_file->path(), entry.offset(), entry.len(), entry.checksum(), bytes_to_read);
- bool read_success;
- {
- ScopedHistogramTimer read_timer(read_latency_);
- read_success = cache_file->Read(entry.offset(), buffer, bytes_to_read);
- }
- if (UNLIKELY(!read_success)) {
- meta_cache_->Erase(key);
- return 0;
- }
+ // Skip the actual reads if doing trace replay
+ if (LIKELY(!trace_replay_)) {
+ CacheFile* cache_file = entry.file();
+ VLOG(3) << Substitute("Reading file $0 offset $1 len $2 checksum $3 bytes_to_read $4",
+ cache_file->path(), entry.offset(), entry.len(), entry.checksum(), bytes_to_read);
+ bool read_success;
+ {
+ ScopedHistogramTimer read_timer(read_latency_);
+ read_success = cache_file->Read(entry.offset(), buffer, bytes_to_read);
+ }
+ if (UNLIKELY(!read_success)) {
+ meta_cache_->Erase(key);
+ return 0;
+ }
- // Verify checksum if enabled. Delete entry on checksum mismatch.
- if (FLAGS_data_cache_checksum && bytes_to_read == entry.len() &&
- !VerifyChecksum("read", entry, buffer, bytes_to_read)) {
- meta_cache_->Erase(key);
- return 0;
+ // Verify checksum if enabled. Delete entry on checksum mismatch.
+ if (FLAGS_data_cache_checksum && bytes_to_read == entry.len() &&
+ !VerifyChecksum("read", entry, buffer, bytes_to_read)) {
+ meta_cache_->Erase(key);
+ return 0;
+ }
}
return bytes_to_read;
}
@@ -668,13 +611,16 @@ bool DataCache::Partition::HandleExistingEntry(const Slice& key,
// Unpack the cache entry.
CacheEntry entry(meta_cache_->Value(handle));
- // Try verifying the checksum of the new buffer matches that of the existing entry.
- // On checksum mismatch, delete the existing entry and don't install the new entry
- // as it's unclear which one is right.
- if (FLAGS_data_cache_checksum && buffer_len >= entry.len()) {
- if (!VerifyChecksum("write", entry, buffer, buffer_len)) {
- meta_cache_->Erase(key);
- return true;
+ // Trace replays have no data and cannot do checksums.
+ if (LIKELY(!trace_replay_)) {
+ // Try verifying the checksum of the new buffer matches that of the existing entry.
+ // On checksum mismatch, delete the existing entry and don't install the new entry
+ // as it's unclear which one is right.
+ if (FLAGS_data_cache_checksum && buffer_len >= entry.len()) {
+ if (!VerifyChecksum("write", entry, buffer, buffer_len)) {
+ meta_cache_->Erase(key);
+ return true;
+ }
}
}
// If the new entry is not any longer than the existing entry, no work to do.
@@ -683,6 +629,10 @@ bool DataCache::Partition::HandleExistingEntry(const Slice& key,
bool DataCache::Partition::InsertIntoCache(const Slice& key, CacheFile* cache_file,
int64_t insertion_offset, const uint8_t* buffer, int64_t buffer_len) {
+ if (UNLIKELY(trace_replay_)) {
+ DCHECK(buffer == nullptr);
+ DCHECK(cache_file == nullptr);
+ }
DCHECK_EQ(insertion_offset % PAGE_SIZE, 0);
const int64_t charge_len = BitUtil::RoundUp(buffer_len, PAGE_SIZE);
@@ -691,19 +641,23 @@ bool DataCache::Partition::InsertIntoCache(const Slice& key, CacheFile* cache_fi
meta_cache_->Allocate(key, sizeof(CacheEntry), charge_len));
if (UNLIKELY(pending_handle.get() == nullptr)) return false;
- // Compute checksum if necessary.
- int64_t checksum = FLAGS_data_cache_checksum ? Checksum(buffer, buffer_len) : 0;
-
- // Write to backing file.
- VLOG(3) << Substitute("Storing file $0 offset $1 len $2 checksum $3 ",
- cache_file->path(), insertion_offset, buffer_len, checksum);
- bool write_success;
- {
- ScopedHistogramTimer write_timer(write_latency_);
- write_success = cache_file->Write(insertion_offset, buffer, buffer_len);
- }
- if (UNLIKELY(!write_success)) {
- return false;
+ int64_t checksum = 0;
+ // Trace replays have no data and cannot do checksums
+ if (LIKELY(!trace_replay_)) {
+ // Compute checksum if necessary.
+ checksum = FLAGS_data_cache_checksum ? Checksum(buffer, buffer_len) : 0;
+
+ // Write to backing file.
+ VLOG(3) << Substitute("Storing file $0 offset $1 len $2 checksum $3 ",
+ cache_file->path(), insertion_offset, buffer_len, checksum);
+ bool write_success;
+ {
+ ScopedHistogramTimer write_timer(write_latency_);
+ write_success = cache_file->Write(insertion_offset, buffer, buffer_len);
+ }
+ if (UNLIKELY(!write_success)) {
+ return false;
+ }
}
// Insert the new entry into the cache.
@@ -712,12 +666,18 @@ bool DataCache::Partition::InsertIntoCache(const Slice& key, CacheFile* cache_fi
Cache::UniqueHandle handle(meta_cache_->Insert(std::move(pending_handle), this));
// Check for failure of Insert(), which means the entry was evicted during Insert()
if (UNLIKELY(handle.get() == nullptr)){
- ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_INSTANT_EVICTIONS->Increment(1);
+ // Trace replays do not keep metrics
+ if (LIKELY(!trace_replay_)) {
+ ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_INSTANT_EVICTIONS->Increment(1);
+ }
return false;
}
- ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_TOTAL_BYTES->Increment(charge_len);
- ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_NUM_ENTRIES->Increment(1);
- ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_NUM_WRITES->Increment(1);
+ // Trace replays do not keep metrics
+ if (LIKELY(!trace_replay_)) {
+ ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_TOTAL_BYTES->Increment(charge_len);
+ ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_NUM_ENTRIES->Increment(1);
+ ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_NUM_WRITES->Increment(1);
+ }
return true;
}
@@ -739,7 +699,7 @@ bool DataCache::Partition::Store(const CacheKey& cache_key, const uint8_t* buffe
CacheFile* cache_file;
int64_t insertion_offset;
- {
+ if (LIKELY(!trace_replay_)) {
std::unique_lock<SpinLock> partition_lock(lock_);
// Limit the write concurrency to avoid blocking the caller (which could be calling
@@ -754,10 +714,8 @@ bool DataCache::Partition::Store(const CacheKey& cache_key, const uint8_t* buffe
pending_insert_set_.find(key.ToString()) != pending_insert_set_.end()) {
ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_DROPPED_BYTES->Increment(buffer_len);
ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_DROPPED_ENTRIES->Increment(1);
- if (tracer_ != nullptr) {
- tracer_->Trace(Tracer::STORE_FAILED_BUSY, cache_key, /*lookup_len=*/-1,
- buffer_len);
- }
+ Trace(trace::EventType::STORE_FAILED_BUSY, cache_key, /*lookup_len=*/-1,
+ buffer_len);
return false;
}
@@ -778,20 +736,29 @@ bool DataCache::Partition::Store(const CacheKey& cache_key, const uint8_t* buffe
// Do this last. At this point, we are committed to inserting 'key' into the cache.
pending_insert_set_.emplace(key.ToString());
- }
-
- if (tracer_ != nullptr) {
- tracer_->Trace(Tracer::STORE, cache_key, /* lookup_len=*/-1, buffer_len);
+ } else {
+ DCHECK(buffer == nullptr);
+ cache_file = nullptr;
+ insertion_offset = 0;
}
// Set up a scoped exit to always remove entry from the pending insertion set.
+ // This is not needed for trace replay.
auto remove_from_pending_set = MakeScopeExitTrigger([this, &key]() {
+ if (UNLIKELY(trace_replay_)) return;
std::unique_lock<SpinLock> partition_lock(lock_);
pending_insert_set_.erase(key.ToString());
});
// Try inserting into the cache.
- return InsertIntoCache(key, cache_file, insertion_offset, buffer, buffer_len);
+ bool insert_success = InsertIntoCache(key, cache_file, insertion_offset, buffer,
+ buffer_len);
+ if (insert_success) {
+ Trace(trace::EventType::STORE, cache_key, /* lookup_len=*/-1, buffer_len);
+ } else {
+ Trace(trace::EventType::STORE_FAILED, cache_key, /*lookup_len=*/ -1, buffer_len);
+ }
+ return insert_success;
}
void DataCache::Partition::DeleteOldFiles() {
@@ -805,6 +772,7 @@ void DataCache::Partition::DeleteOldFiles() {
void DataCache::Partition::EvictedEntry(Slice key, Slice value) {
if (closed_) return;
+ if (UNLIKELY(trace_replay_)) return;
ScopedHistogramTimer eviction_timer(eviction_latency_);
// Unpack the cache entry.
CacheEntry entry(value);
@@ -879,21 +847,23 @@ Status DataCache::Init() {
<< PrettyPrinter::PrintBytes(capacity);
std::unique_ptr<Partition> partition =
make_unique<Partition>(partition_idx, dir_path, capacity,
- max_opened_files_per_partition);
+ max_opened_files_per_partition, trace_replay_);
RETURN_IF_ERROR(partition->Init());
partitions_.emplace_back(move(partition));
++partition_idx;
}
CHECK_GT(partitions_.size(), 0);
- // Starts a thread pool which deletes old files from partitions. DataCache::Store()
- // will enqueue a request (i.e. a partition index) when it notices the number of files
- // in a partition exceeds the per-partition limit. The files in a partition will be
- // closed in the order they are created until it's within the per-partition limit.
- file_deleter_pool_.reset(new ThreadPool<int>("impala-server",
- "data-cache-file-deleter", 1, MAX_FILE_DELETER_QUEUE_SIZE,
- bind<void>(&DataCache::DeleteOldFiles, this, _1, _2)));
- RETURN_IF_ERROR(file_deleter_pool_->Init());
+ if (LIKELY(!trace_replay_)) {
+ // Starts a thread pool which deletes old files from partitions. DataCache::Store()
+ // will enqueue a request (i.e. a partition index) when it notices the number of files
+ // in a partition exceeds the per-partition limit. The files in a partition will be
+ // closed in the order they are created until it's within the per-partition limit.
+ file_deleter_pool_.reset(new ThreadPool<int>("impala-server",
+ "data-cache-file-deleter", 1, MAX_FILE_DELETER_QUEUE_SIZE,
+ bind<void>(&DataCache::DeleteOldFiles, this, _1, _2)));
+ RETURN_IF_ERROR(file_deleter_pool_->Init());
+ }
return Status::OK();
}
@@ -964,63 +934,29 @@ void DataCache::DeleteOldFiles(uint32_t thread_id, int partition_idx) {
partitions_[partition_idx]->DeleteOldFiles();
}
-void DataCache::Partition::Tracer::Trace(
- CacheStatus status, const DataCache::CacheKey& key,
+void DataCache::Partition::Trace(
+ const trace::EventType& type, const DataCache::CacheKey& key,
int64_t lookup_len, int64_t entry_len) {
+ if (tracer_ == nullptr) return;
- ostringstream buf;
- kudu::JsonWriter jw(&buf, kudu::JsonWriter::COMPACT);
-
- jw.StartObject();
- jw.String("ts");
- jw.Double(WallTime_Now());
- jw.String("s");
- switch (status) {
- case HIT: jw.String("H"); break;
- case MISS: jw.String("M"); break;
- case STORE: jw.String("S"); break;
- case STORE_FAILED_BUSY: jw.String("F"); break;
- }
-
- jw.String("f");
- if (FLAGS_data_cache_anonymize_trace) {
- uint128 hash = util_hash::CityHash128(
- reinterpret_cast<const char*>(key.filename().data()),
- key.filename().size());
- // A 128-bit (16-byte) hash results in a 24-byte base64-encoded string, including
- // two characters of padding.
- const int BUF_LEN = 24;
- DCHECK_EQ(BUF_LEN, CalculateBase64EscapedLen(sizeof(hash)));
- char b64_buf[BUF_LEN];
- int out_len = Base64Escape(reinterpret_cast<const unsigned char*>(&hash),
- sizeof(hash), b64_buf, BUF_LEN);
- DCHECK_EQ(out_len, BUF_LEN);
- // Chop off the two padding bytes.
- DCHECK(b64_buf[23] == '=' && b64_buf[22] == '=');
- out_len = 22;
- jw.String(b64_buf, out_len);
- } else {
- jw.String(reinterpret_cast<const char*>(key.filename().data()),
- key.filename().size());
- }
- jw.String("m");
- jw.Int64(key.mtime());
- jw.String("o");
- jw.Int64(key.offset());
-
- if (lookup_len != -1) {
- jw.String("lLen");
- jw.Int64(lookup_len);
- }
- if (entry_len != -1) {
- jw.String("eLen");
- jw.Int64(entry_len);
+ // When tracing a percentage of the requests, we want to trace all the accesses for a
+ // consistent subset of the entries rather than a subset of accesses for all entries.
+ // This gives the access trace more useful data.
+ //
+ // This uses the hash to determine a consistent subset to trace. Note that this is
+ // tracing at the partition level. If there are multiple partitions, the entry has
+ // been mapped to a specific partition by taking a modulus of the hash value. This
+ // can impact which bits are still useful. For example, if there are two partitions,
+ // this would have only even hash values or only odd hash values. To minimize the
+ // impact of this, we use 101 rather than 100, because 101 is prime.
+ uint64_t unsigned_key_hash = static_cast<uint64_t>(key.Hash());
+ if (FLAGS_data_cache_trace_percentage < 100 &&
+ unsigned_key_hash % 101 >= FLAGS_data_cache_trace_percentage) {
+ return;
}
- jw.EndObject();
- buf << "\n";
- string s = buf.str();
- logger_->Write(/*force_flush=*/false, /*timestamp=*/0, s.data(), s.size());
+ tracer_->Trace(type, WallTime_Now(), key.filename(), key.mtime(), key.offset(),
+ lookup_len, entry_len);
}
} // namespace io
diff --git a/be/src/runtime/io/data-cache.h b/be/src/runtime/io/data-cache.h
index 98c27d4..d892bce 100644
--- a/be/src/runtime/io/data-cache.h
+++ b/be/src/runtime/io/data-cache.h
@@ -107,6 +107,11 @@
namespace impala {
namespace io {
+namespace trace {
+ class Tracer;
+ enum class EventType;
+}
+
class DataCache {
public:
@@ -114,8 +119,12 @@ class DataCache {
/// in which <dir1>, <dirN> are part of a list of directories for storing cached data
/// and each directory corresponds to a cache partition. <quota> is the storage quota
/// for each directory. Impala daemons running on the same host will not share any
- /// caching directories.
- explicit DataCache(const std::string config) : config_(std::move(config)) { }
+ /// caching directories. If 'trace_replay' is set to true, the cache operates in a
+ /// an optimized mode that skips all file operations and only does the metadata
+ /// operations. This is used to replay the access trace and compare different cache
+ /// configurations. See data-cache-trace.h
+ explicit DataCache(const std::string config, bool trace_replay = false)
+ : config_(config), trace_replay_(trace_replay) { }
~DataCache() { ReleaseResources(); }
@@ -135,7 +144,8 @@ class DataCache {
/// 'mtime' : the modification time of the requested file
/// 'offset' : starting offset of the requested region in the file
/// 'bytes_to_read' : number of bytes to be read from the cache
- /// 'buffer' : output buffer to be written into on cache hit
+ /// 'buffer' : output buffer to be written into on cache hit or nullptr for
+ /// trace replay
///
/// Returns the number of bytes read from the cache on cache hit; Returns 0 otherwise.
///
@@ -150,7 +160,8 @@ class DataCache {
/// 'filename' : name of the file being inserted
/// 'mtime' : the modification time of the file being inserted.
/// 'offset' : the starting offset of the region in the file being inserted
- /// 'buffer' : buffer holding the data to be inserted
+ /// 'buffer' : buffer holding the data to be inserted or nullptr for trace
+ /// replay
/// 'buffer_len' : size of 'buffer'
///
/// The cache key is hashed and the resulting hash determines the partition to use.
@@ -194,8 +205,10 @@ class DataCache {
public:
/// Creates a partition at the given directory 'path' with quota 'capacity' in bytes.
/// 'max_opened_files' is the maximum number of opened files allowed per partition.
+ /// If 'trace_replay' is true, this only performs metadata operations for the
+ /// access trace functionality.
Partition(int32_t index, const std::string& path, int64_t capacity,
- int max_opened_files);
+ int max_opened_files, bool trace_replay);
~Partition();
@@ -214,14 +227,15 @@ class DataCache {
void ReleaseResources();
/// Looks up in the meta-data cache with key 'cache_key'. If found, try copying
- /// 'bytes_to_read' bytes from the backing file into 'buffer'. Returns number
+ /// 'bytes_to_read' bytes from the backing file into 'buffer'. If trace_replay
+ /// is enabled, the buffer is null and no bytes are copied. Returns number
/// of bytes read from the cache. Returns 0 if there is a cache miss.
int64_t Lookup(const CacheKey& cache_key, int64_t bytes_to_read, uint8_t* buffer);
/// Inserts a entry with key 'cache_key' and data in 'buffer' into the cache.
- /// 'buffer_len' is the length of buffer. 'start_reclaim' is set to true if
- /// the number of backing files exceeds the per partition limit. Returns true if
- /// the entry is inserted. Returns false otherwise.
+ /// 'buffer' is nullptr for trace replay. 'buffer_len' is the length of buffer.
+ /// 'start_reclaim' is set to true if the number of backing files exceeds the per
+ /// partition limit. Returns true if the entry is inserted. Returns false otherwise.
bool Store(const CacheKey& cache_key, const uint8_t* buffer, int64_t buffer_len,
bool* start_reclaim);
@@ -251,7 +265,6 @@ class DataCache {
friend class DataCacheBaseTest;
friend class DataCacheTest;
FRIEND_TEST(DataCacheTest, TestAccessTrace);
- class Tracer;
/// Index of this partition. This is used for naming metrics or other items that
/// need separate values for each partition. It does not impact cache behavior.
@@ -266,6 +279,11 @@ class DataCache {
/// Maximum number of opened files allowed in a partition.
const int max_opened_files_;
+ /// Whether this is only a trace replay. For trace replay, this is only
+ /// performing the metadata operations to determine the hit/miss rate.
+ /// There is no need to perform any filesystem operations.
+ bool trace_replay_;
+
/// True if this partition has been closed. Expected to be set after all IO
/// threads have been joined.
bool closed_ = false;
@@ -273,9 +291,6 @@ class DataCache {
/// The prefix of the names of the cache backing files.
static const char* CACHE_FILE_PREFIX;
- /// The file name used for the access trace.
- static const char* TRACE_FILE_NAME;
-
/// Protects the following fields.
SpinLock lock_;
@@ -304,7 +319,7 @@ class DataCache {
/// content. Please see comments at CachedEntry for details.
std::unique_ptr<Cache> meta_cache_;
- std::unique_ptr<Tracer> tracer_;
+ std::unique_ptr<trace::Tracer> tracer_;
/// Metrics to track performance of the underlying filesystem for the data cache
/// These are all latency histograms for the operations on the data cache files for
@@ -360,11 +375,18 @@ class DataCache {
/// Returns false if the checksum of 'buffer' doesn't match 'entry->checksum'.
static bool VerifyChecksum(const std::string& ops_name, const CacheEntry& entry,
const uint8_t* buffer, int64_t buffer_len);
+
+ void Trace(const trace::EventType& status, const DataCache::CacheKey& key,
+ int64_t lookup_len, int64_t entry_len);
};
/// The configuration string for the data cache.
const std::string config_;
+ /// Set to true if this is only doing trace replay. Trace replay does only metadata
+ /// operations, and no filesystem operations are required.
+ bool trace_replay_;
+
/// The set of all cache partitions.
std::vector<std::unique_ptr<Partition>> partitions_;
diff --git a/bin/start-impala-cluster.py b/bin/start-impala-cluster.py
index cccf9f1..c708ce4 100755
--- a/bin/start-impala-cluster.py
+++ b/bin/start-impala-cluster.py
@@ -124,6 +124,12 @@ parser.add_option("--data_cache_dir", dest="data_cache_dir", default=None,
parser.add_option("--data_cache_size", dest="data_cache_size", default=0,
help="This specifies the maximum storage usage of the IO data cache "
"each Impala daemon can use.")
+parser.add_option("--data_cache_eviction_policy", dest="data_cache_eviction_policy",
+ default="LRU", help="This specifies the cache eviction policy to use "
+ "for the data cache")
+parser.add_option("--data_cache_enable_tracing", dest="data_cache_enable_tracing",
+ action="store_true", default=False,
+ help="If the data cache is enabled, this enables tracing accesses.")
# For testing: list of comma-separated delays, in milliseconds, that delay impalad catalog
# replica initialization. The ith delay is applied to the ith impalad.
@@ -374,6 +380,26 @@ def build_impalad_arg_lists(cluster_size, num_coordinators, use_exclusive_coordi
args = "-data_cache={dir}:{quota} {args}".format(
dir=data_cache_path_arg, quota=options.data_cache_size, args=args)
+ # Add the eviction policy
+ args = "-data_cache_eviction_policy={policy} {args}".format(
+ policy=options.data_cache_eviction_policy, args=args)
+
+ # Add access tracing arguments if requested
+ if options.data_cache_enable_tracing:
+ tracing_args = ""
+ if options.docker_network is None:
+ # To avoid collisions in log files, use different data_cache_trace_dir values
+ # for different Impalads. The default directory is fine for the docker-based
+ # tests.
+ data_cache_trace_dir = "{log_dir}/data_cache_traces_{impalad_num}".format(
+ log_dir=options.log_dir, impalad_num=i)
+ tracing_args = "-data_cache_trace_dir={trace_dir} {tracing_args}".format(
+ trace_dir=data_cache_trace_dir, tracing_args=tracing_args)
+
+ tracing_args = "-data_cache_enable_tracing=true {tracing_args}".format(
+ tracing_args=tracing_args)
+ args = "{tracing_args} {args}".format(tracing_args=tracing_args, args=args)
+
# Appended at the end so they can override previous args.
if i < len(per_impalad_args):
args = "{args} {per_impalad_args}".format(