diff --git a/CMakeLists.txt b/CMakeLists.txt index 254fc37dd..558c32ab8 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -197,6 +197,7 @@ set(memgraph_src_files ${src_dir}/database/graph_db.cpp ${src_dir}/database/graph_db_config.cpp ${src_dir}/database/graph_db_accessor.cpp + ${src_dir}/durability/paths.cpp ${src_dir}/durability/recovery.cpp ${src_dir}/durability/snapshooter.cpp ${src_dir}/durability/wal.cpp diff --git a/src/durability/paths.cpp b/src/durability/paths.cpp new file mode 100644 index 000000000..d297d4611 --- /dev/null +++ b/src/durability/paths.cpp @@ -0,0 +1,85 @@ +#include +#include +#include + +#include "glog/logging.h" + +#include "transactions/type.hpp" +#include "utils/datetime/timestamp.hpp" +#include "utils/string.hpp" + +namespace durability { +namespace fs = std::experimental::filesystem; +/// Returns true if the given directory path exists or is succesfully created. +bool EnsureDir(const fs::path &dir) { + if (fs::exists(dir)) return true; + std::error_code error_code; // Just for exception suppression. + return fs::create_directories(dir, error_code); +} + +/// Ensures the given durability directory exists and is ready for use. Creates +/// the directory if it doesn't exist. +void CheckDurabilityDir(const std::string &durability_dir) { + namespace fs = std::experimental::filesystem; + if (fs::exists(durability_dir)) { + CHECK(fs::is_directory(durability_dir)) << "The durability directory path '" + << durability_dir + << "' is not a directory!"; + } else { + bool success = EnsureDir(durability_dir); + CHECK(success) << "Failed to create durability directory '" + << durability_dir << "'."; + } +} + +/// Returns the transaction id contained in the file name. If the filename is +/// not a parseable WAL file name, nullopt is returned. If the filename +/// represents the "current" WAL file, then the maximum possible transaction ID +/// is returned because that's appropriate for the recovery logic (the current +/// WAL does not yet have a maximum transaction ID and can't be discarded by +/// the recovery regardless of the snapshot from which the transaction starts). +std::experimental::optional TransactionIdFromWalFilename( + const std::string &name) { + auto nullopt = std::experimental::nullopt; + // Get the max_transaction_id from the file name that has format + // "XXXXX__max_transaction_" + auto file_name_split = utils::RSplit(name, "__", 1); + if (file_name_split.size() != 2) { + LOG(WARNING) << "Unable to parse WAL file name: " << name; + return nullopt; + } + if (file_name_split[1] == "current") + return std::numeric_limits::max(); + file_name_split = utils::RSplit(file_name_split[1], "_", 1); + if (file_name_split.size() != 2) { + LOG(WARNING) << "Unable to parse WAL file name: " << name; + return nullopt; + } + auto &tx_id_str = file_name_split[1]; + try { + return std::stoll(tx_id_str); + } catch (std::invalid_argument &) { + LOG(WARNING) << "Unable to parse WAL file name tx ID: " << tx_id_str; + return nullopt; + } catch (std::out_of_range &) { + LOG(WARNING) << "WAL file name tx ID too large: " << tx_id_str; + return nullopt; + } +} + +/// Generates a file path for a write-ahead log file. If given a transaction ID +/// the file name will contain it. Otherwise the file path is for the "current" +/// WAL file for which the max tx id is still unknown. +fs::path WalFilenameForTransactionId( + const std::experimental::filesystem::path &wal_dir, + std::experimental::optional tx_id = + std::experimental::nullopt) { + auto file_name = Timestamp::now().to_iso8601(); + if (tx_id) { + file_name += "__max_transaction_" + std::to_string(*tx_id); + } else { + file_name += "__current"; + } + return wal_dir / file_name; +} +} // namespace durability diff --git a/src/durability/paths.hpp b/src/durability/paths.hpp index f1cd8fa29..07796ed99 100644 --- a/src/durability/paths.hpp +++ b/src/durability/paths.hpp @@ -2,32 +2,19 @@ #include #include -#include - -#include "glog/logging.h" #include "transactions/type.hpp" -#include "utils/datetime/timestamp.hpp" -#include "utils/string.hpp" namespace durability { const std::string kSnapshotDir = "snapshots"; const std::string kWalDir = "wal"; +/// Esures that the given dir either exists or is succsefully created. +bool EnsureDir(const std::experimental::filesystem::path &dir); + /// Ensures the given durability directory exists and is ready for use. Creates /// the directory if it doesn't exist. -inline void CheckDurabilityDir(const std::string &durability_dir) { - namespace fs = std::experimental::filesystem; - if (fs::exists(durability_dir)) { - CHECK(fs::is_directory(durability_dir)) << "The durability directory path '" - << durability_dir - << "' is not a directory!"; - } else { - bool success = fs::create_directory(durability_dir); - CHECK(success) << "Failed to create durability directory '" - << durability_dir << "'."; - } -} +void CheckDurabilityDir(const std::string &durability_dir); /// Returns the transaction id contained in the file name. If the filename is /// not a parseable WAL file name, nullopt is returned. If the filename @@ -35,48 +22,14 @@ inline void CheckDurabilityDir(const std::string &durability_dir) { /// is returned because that's appropriate for the recovery logic (the current /// WAL does not yet have a maximum transaction ID and can't be discarded by /// the recovery regardless of the snapshot from which the transaction starts). -inline std::experimental::optional -TransactionIdFromWalFilename(const std::string &name) { - auto nullopt = std::experimental::nullopt; - // Get the max_transaction_id from the file name that has format - // "XXXXX__max_transaction_" - auto file_name_split = utils::RSplit(name, "__", 1); - if (file_name_split.size() != 2) { - LOG(WARNING) << "Unable to parse WAL file name: " << name; - return nullopt; - } - if (file_name_split[1] == "current") - return std::numeric_limits::max(); - file_name_split = utils::RSplit(file_name_split[1], "_", 1); - if (file_name_split.size() != 2) { - LOG(WARNING) << "Unable to parse WAL file name: " << name; - return nullopt; - } - auto &tx_id_str = file_name_split[1]; - try { - return std::stoll(tx_id_str); - } catch (std::invalid_argument &) { - LOG(WARNING) << "Unable to parse WAL file name tx ID: " << tx_id_str; - return nullopt; - } catch (std::out_of_range &) { - LOG(WARNING) << "WAL file name tx ID too large: " << tx_id_str; - return nullopt; - } -} +std::experimental::optional TransactionIdFromWalFilename( + const std::string &name); /// Generates a file path for a write-ahead log file. If given a transaction ID /// the file name will contain it. Otherwise the file path is for the "current" /// WAL file for which the max tx id is still unknown. -inline auto WalFilenameForTransactionId( +std::experimental::filesystem::path WalFilenameForTransactionId( const std::experimental::filesystem::path &wal_dir, std::experimental::optional tx_id = - std::experimental::nullopt) { - auto file_name = Timestamp::now().to_iso8601(); - if (tx_id) { - file_name += "__max_transaction_" + std::to_string(*tx_id); - } else { - file_name += "__current"; - } - return wal_dir / file_name; -} -} + std::experimental::nullopt); +} // namespace durability diff --git a/src/durability/snapshooter.cpp b/src/durability/snapshooter.cpp index 989dfcabc..9c7479c4f 100644 --- a/src/durability/snapshooter.cpp +++ b/src/durability/snapshooter.cpp @@ -112,15 +112,7 @@ fs::path MakeSnapshotPath(const fs::path &durability_dir) { bool MakeSnapshot(GraphDbAccessor &db_accessor_, const fs::path &durability_dir, const int snapshot_max_retained) { - auto ensure_dir = [](const auto &dir) { - if (!fs::exists(dir) && !fs::create_directories(dir)) { - LOG(ERROR) << "Error while creating directory " << dir; - return false; - } - return true; - }; - if (!ensure_dir(durability_dir)) return false; - if (!ensure_dir(durability_dir / kSnapshotDir)) return false; + if (!EnsureDir(durability_dir / kSnapshotDir)) return false; const auto snapshot_file = MakeSnapshotPath(durability_dir); if (fs::exists(snapshot_file)) return false; if (Encode(snapshot_file, db_accessor_)) { diff --git a/src/durability/wal.cpp b/src/durability/wal.cpp index cdb09b551..b391cdf49 100644 --- a/src/durability/wal.cpp +++ b/src/durability/wal.cpp @@ -261,8 +261,7 @@ WriteAheadLog::WalFile::~WalFile() { } void WriteAheadLog::WalFile::Init() { - if (!std::experimental::filesystem::exists(wal_dir_) && - !std::experimental::filesystem::create_directories(wal_dir_)) { + if (!EnsureDir(wal_dir_)) { LOG(ERROR) << "Can't write to WAL directory: " << wal_dir_; current_wal_file_ = std::experimental::filesystem::path(); } else { diff --git a/src/utils/string.hpp b/src/utils/string.hpp index 7abd73c96..5ba59aafc 100644 --- a/src/utils/string.hpp +++ b/src/utils/string.hpp @@ -1,7 +1,9 @@ #pragma once +#include #include #include +#include #include #include #include @@ -201,4 +203,16 @@ inline bool EndsWith(const std::string &s, const std::string &suffix) { inline bool StartsWith(const std::string &s, const std::string &prefix) { return s.size() >= prefix.size() && s.compare(0, prefix.size(), prefix) == 0; } + +/** Creates a random alphanumeric string of the given length. */ +inline std::string RandomString(size_t length) { + static const char charset[] = + "0123456789" + "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + "abcdefghijklmnopqrstuvwxyz"; + std::string str(length, 0); + for (size_t i = 0; i < length; ++i) + str[i] = charset[rand() % strlen(charset)]; + return str; } +} // namespace utils diff --git a/tests/unit/durability.cpp b/tests/unit/durability.cpp index dc254c197..fa23c3a92 100644 --- a/tests/unit/durability.cpp +++ b/tests/unit/durability.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -222,15 +223,6 @@ void CompareDbs(GraphDb &a, GraphDb &b) { } } -const fs::path kDurabilityDir = - fs::temp_directory_path() / "MG_test_unit_durability"; -const fs::path kSnapshotDir = kDurabilityDir / durability::kSnapshotDir; -const fs::path kWalDir = kDurabilityDir / durability::kWalDir; - -void CleanDurability() { - if (fs::exists(kDurabilityDir)) fs::remove_all(kDurabilityDir); -} - std::vector DirFiles(fs::path dir) { std::vector files; if (fs::exists(dir)) @@ -238,23 +230,6 @@ std::vector DirFiles(fs::path dir) { return files; } -auto DbConfig() { - GraphDb::Config config; - config.durability_enabled = false; - config.durability_directory = kDurabilityDir; - config.snapshot_on_exit = false; - config.db_recover_on_startup = false; - - return config; -} - -void MakeSnapshot(GraphDb &db, int snapshot_max_retained = -1) { - GraphDbAccessor dba(db); - ASSERT_TRUE( - durability::MakeSnapshot(dba, kDurabilityDir, snapshot_max_retained)); - dba.Commit(); -} - fs::path GetLastFile(fs::path dir) { std::vector files = DirFiles(dir); CHECK(static_cast(files.size()) > 0) << "No files in folder."; @@ -294,7 +269,37 @@ void MakeDb(GraphDb &db, int scale, std::vector indices = {}) { class Durability : public ::testing::Test { protected: + fs::path tmp_dir_ = fs::temp_directory_path() / "MG_test_unit_durability"; + fs::path durability_dir_; + fs::path snapshot_dir_; + fs::path wal_dir_; + + void CleanDurability() { + if (fs::exists(tmp_dir_)) fs::remove_all(tmp_dir_); + } + + auto DbConfig() { + GraphDb::Config config; + config.durability_enabled = false; + config.durability_directory = durability_dir_; + config.snapshot_on_exit = false; + config.db_recover_on_startup = false; + + return config; + } + + void MakeSnapshot(GraphDb &db, int snapshot_max_retained = -1) { + GraphDbAccessor dba(db); + ASSERT_TRUE( + durability::MakeSnapshot(dba, durability_dir_, snapshot_max_retained)); + dba.Commit(); + } + void SetUp() override { + durability_dir_ = tmp_dir_ / utils::RandomString(24); + snapshot_dir_ = durability_dir_ / durability::kSnapshotDir; + wal_dir_ = durability_dir_ / durability::kWalDir; + srand(time(NULL)); FLAGS_wal_rotate_ops_count = 1000; CleanDurability(); } @@ -325,8 +330,8 @@ TEST_F(Durability, WalEncoding) { std::this_thread::sleep_for(std::chrono::milliseconds(50)); HashedFileReader reader; - ASSERT_EQ(DirFiles(kWalDir).size(), 1); - ASSERT_TRUE(reader.Open(GetLastFile(kWalDir))); + ASSERT_EQ(DirFiles(wal_dir_).size(), 1); + ASSERT_TRUE(reader.Open(GetLastFile(wal_dir_))); communication::bolt::Decoder decoder{reader}; std::vector ops; while (true) { @@ -405,7 +410,7 @@ TEST_F(Durability, SnapshotEncoding) { MakeSnapshot(db); } - auto snapshot = GetLastFile(kSnapshotDir); + auto snapshot = GetLastFile(snapshot_dir_); HashedFileReader buffer; communication::bolt::Decoder decoder(buffer); @@ -506,8 +511,8 @@ TEST_F(Durability, WalRecovery) { // Sleep to ensure the WAL gets flushed. std::this_thread::sleep_for(std::chrono::milliseconds(50)); - ASSERT_EQ(DirFiles(kSnapshotDir).size(), 0); - EXPECT_GT(DirFiles(kWalDir).size(), 1); + ASSERT_EQ(DirFiles(snapshot_dir_).size(), 0); + EXPECT_GT(DirFiles(wal_dir_).size(), 1); { auto recovered_config = DbConfig(); @@ -530,8 +535,8 @@ TEST_F(Durability, SnapshotAndWalRecovery) { // Sleep to ensure the WAL gets flushed. std::this_thread::sleep_for(std::chrono::milliseconds(50)); - ASSERT_EQ(DirFiles(kSnapshotDir).size(), 1); - EXPECT_GT(DirFiles(kWalDir).size(), 1); + ASSERT_EQ(DirFiles(snapshot_dir_).size(), 1); + EXPECT_GT(DirFiles(wal_dir_).size(), 1); { auto recovered_config = DbConfig(); @@ -581,8 +586,8 @@ TEST_F(Durability, SnapshotAndWalRecoveryAfterComplexTxSituation) { // Sleep to ensure the WAL gets flushed. std::this_thread::sleep_for(std::chrono::milliseconds(50)); - ASSERT_EQ(DirFiles(kSnapshotDir).size(), 1); - EXPECT_GT(DirFiles(kWalDir).size(), 1); + ASSERT_EQ(DirFiles(snapshot_dir_).size(), 1); + EXPECT_GT(DirFiles(wal_dir_).size(), 1); { auto recovered_config = DbConfig(); recovered_config.db_recover_on_startup = true; @@ -601,14 +606,14 @@ TEST_F(Durability, NoWalDuringRecovery) { // Sleep to ensure the WAL gets flushed. std::this_thread::sleep_for(std::chrono::milliseconds(50)); - auto wal_files_before = DirFiles(kWalDir); + auto wal_files_before = DirFiles(wal_dir_); ASSERT_GT(wal_files_before.size(), 3); { auto recovered_config = DbConfig(); recovered_config.db_recover_on_startup = true; GraphDb recovered{recovered_config}; CompareDbs(db, recovered); - auto wal_files_after = DirFiles(kWalDir); + auto wal_files_after = DirFiles(wal_dir_); EXPECT_EQ(wal_files_after.size(), wal_files_before.size()); } } @@ -624,14 +629,14 @@ TEST_F(Durability, SnapshotRetention) { std::unordered_set snapshots; for (int i = 0; i < count; ++i) { MakeSnapshot(db, retain); - auto latest = GetLastFile(kSnapshotDir); - snapshots.emplace(GetLastFile(kSnapshotDir)); + auto latest = GetLastFile(snapshot_dir_); + snapshots.emplace(GetLastFile(snapshot_dir_)); // Ensures that the latest snapshot was not in the snapshots collection // before. Thus ensures that it wasn't pruned. EXPECT_EQ(snapshots.size(), i + 1); } - EXPECT_EQ(DirFiles(kSnapshotDir).size(), + EXPECT_EQ(DirFiles(snapshot_dir_).size(), std::min(count, retain < 0 ? count : retain)); }; } @@ -644,14 +649,14 @@ TEST_F(Durability, WalRetention) { MakeDb(db, 100); MakeSnapshot(db); MakeDb(db, 100); - EXPECT_EQ(DirFiles(kSnapshotDir).size(), 1); + EXPECT_EQ(DirFiles(snapshot_dir_).size(), 1); std::this_thread::sleep_for(std::chrono::milliseconds(50)); // 1 current WAL file, plus retained ones - EXPECT_GT(DirFiles(kWalDir).size(), 1); + EXPECT_GT(DirFiles(wal_dir_).size(), 1); MakeSnapshot(db); // only 1 current WAL file - EXPECT_EQ(DirFiles(kSnapshotDir).size(), 2); - EXPECT_EQ(DirFiles(kWalDir).size(), 1); + EXPECT_EQ(DirFiles(snapshot_dir_).size(), 2); + EXPECT_EQ(DirFiles(wal_dir_).size(), 1); } TEST_F(Durability, SnapshotOnExit) { @@ -660,5 +665,5 @@ TEST_F(Durability, SnapshotOnExit) { config.snapshot_on_exit = true; GraphDb graph_db{config}; } - EXPECT_EQ(DirFiles(kSnapshotDir).size(), 1); + EXPECT_EQ(DirFiles(snapshot_dir_).size(), 1); } diff --git a/tests/unit/query_common.hpp b/tests/unit/query_common.hpp index df21afb4b..6589983ed 100644 --- a/tests/unit/query_common.hpp +++ b/tests/unit/query_common.hpp @@ -22,7 +22,6 @@ #pragma once -#include #include #include #include @@ -146,13 +145,6 @@ auto GetEdge(AstTreeStorage &storage, const std::string &name, EdgeAtom::Type::SINGLE, dir, edge_types); } -/// Generates a randomly chosen (uniformly) string from a population of 10 ** 20 -std::string random_string() { - std::string str = "rand_str_"; - for (int i = 0; i < 20; i++) str += std::to_string(rand() % 10); - return str; -} - /// /// Create a variable length expansion EdgeAtom with given name, direction and /// edge_type. @@ -167,10 +159,10 @@ auto GetEdgeVariable(AstTreeStorage &storage, const std::string &name, auto r_val = storage.Create(storage.Create(name), EdgeAtom::Type::DEPTH_FIRST, dir, edge_types); - r_val->inner_edge_ = - inner_edge ? inner_edge : storage.Create(random_string()); - r_val->inner_node_ = - inner_node ? inner_node : storage.Create(random_string()); + r_val->inner_edge_ = inner_edge ? inner_edge : storage.Create( + utils::RandomString(20)); + r_val->inner_node_ = inner_node ? inner_node : storage.Create( + utils::RandomString(20)); return r_val; } @@ -192,7 +184,8 @@ auto GetNode(AstTreeStorage &storage, const std::string &name, /// auto GetPattern(AstTreeStorage &storage, std::vector atoms) { auto pattern = storage.Create(); - pattern->identifier_ = storage.Create(random_string(), false); + pattern->identifier_ = + storage.Create(utils::RandomString(20), false); pattern->atoms_.insert(pattern->atoms_.begin(), atoms.begin(), atoms.end()); return pattern; } diff --git a/tests/unit/utils_string.cpp b/tests/unit/utils_string.cpp index 249274276..73245784d 100644 --- a/tests/unit/utils_string.cpp +++ b/tests/unit/utils_string.cpp @@ -111,3 +111,15 @@ TEST(String, EndsWith) { EXPECT_FALSE(EndsWith("memgraph", "GRAPH")); EXPECT_FALSE(EndsWith("memgraph", "the memgraph")); } + +TEST(String, RandomString) { + EXPECT_EQ(RandomString(0).size(), 0); + EXPECT_EQ(RandomString(1).size(), 1); + EXPECT_EQ(RandomString(42).size(), 42); + + std::set string_set; + for (int i = 0 ; i < 20 ; ++i) + string_set.emplace(RandomString(256)); + + EXPECT_EQ(string_set.size(), 20); +} diff --git a/tools/src/CMakeLists.txt b/tools/src/CMakeLists.txt index 1fd9293fe..1237d9e68 100644 --- a/tools/src/CMakeLists.txt +++ b/tools/src/CMakeLists.txt @@ -5,6 +5,7 @@ add_executable(mg_import_csv ${memgraph_src_dir}/communication/bolt/v1/decoder/decoded_value.cpp ${memgraph_src_dir}/data_structures/concurrent/skiplist_gc.cpp ${memgraph_src_dir}/database/graph_db_accessor.cpp + ${memgraph_src_dir}/durability/paths.cpp ${memgraph_src_dir}/durability/snapshooter.cpp ${memgraph_src_dir}/durability/wal.cpp ${memgraph_src_dir}/query/typed_value.cpp diff --git a/tools/tests/CMakeLists.txt b/tools/tests/CMakeLists.txt index 9853a055c..8f2916af2 100644 --- a/tools/tests/CMakeLists.txt +++ b/tools/tests/CMakeLists.txt @@ -7,6 +7,7 @@ add_executable(mg_recovery_check ${memgraph_src_dir}/database/graph_db.cpp ${memgraph_src_dir}/database/graph_db_config.cpp ${memgraph_src_dir}/database/graph_db_accessor.cpp + ${memgraph_src_dir}/durability/paths.cpp ${memgraph_src_dir}/durability/recovery.cpp ${memgraph_src_dir}/durability/snapshooter.cpp ${memgraph_src_dir}/durability/wal.cpp