From e791e9b6608abc6f07b5521703c428eb4a7750fb Mon Sep 17 00:00:00 2001 From: "matej.gradicek" Date: Wed, 17 May 2017 08:08:57 +0000 Subject: [PATCH] Summary:Added files for snapshot durability Summary: File buffer added Implemented little more of snapshoter. Resolved conflicts More stuff implemented of snapshot durability. More things in snapshoter. Refactored, added comments Merge branch 'dev' into durability_snapshot Merge branch 'dev' into durability_snapshot Resolved bug in scheduler, snapshoter is running in grpah_db. Reviewers: mferencevic, buda, dgleich, mislav.bradac Reviewed By: mferencevic, buda, dgleich, mislav.bradac Subscribers: pullbot Differential Revision: https://phabricator.memgraph.io/D232 --- CMakeLists.txt | 5 +- config/memgraph.yaml | 3 +- src/config/config.hpp | 2 +- src/database/graph_db.cpp | 40 +++++++++- src/database/graph_db.hpp | 7 +- src/durability/file_writer_buffer.hpp | 98 +++++++++++++++++++++++++ src/durability/snapshooter.cpp | 82 +++++++++++++++++++++ src/durability/snapshooter.hpp | 57 +++++++++++++++ src/utils/datetime/timestamp.hpp | 10 ++- tests/unit/snapshot.cpp | 101 ++++++++++++++++++++++++++ 10 files changed, 396 insertions(+), 9 deletions(-) create mode 100644 src/durability/file_writer_buffer.hpp create mode 100644 src/durability/snapshooter.cpp create mode 100644 src/durability/snapshooter.hpp create mode 100644 tests/unit/snapshot.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 92eb6cc0d..f1fbef14f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -330,6 +330,7 @@ set(memgraph_src_files ${src_dir}/io/network/socket.cpp ${src_dir}/threading/thread.cpp ${src_dir}/mvcc/id.cpp + ${src_dir}/durability/snapshooter.cpp # ${src_dir}/snapshot/snapshot_engine.cpp # ${src_dir}/snapshot/snapshoter.cpp # ${src_dir}/snapshot/snapshot_encoder.cpp @@ -340,7 +341,7 @@ set(memgraph_src_files ${src_dir}/storage/record_accessor.cpp ${src_dir}/storage/vertex_accessor.cpp ${src_dir}/storage/edge_accessor.cpp -# ${src_dir}/storage/record_accessor.cpp +# ${src_dir}/storage/record_accessor.cpp ${src_dir}/transactions/snapshot.cpp ${src_dir}/transactions/transaction.cpp ${src_dir}/template_engine/engine.cpp @@ -366,6 +367,7 @@ set(memgraph_src_files # STATIC library used by memgraph executables add_library(memgraph_lib STATIC ${memgraph_src_files}) +target_link_libraries(memgraph_lib stdc++fs) add_dependencies(memgraph_lib generate_opencypher_parser generate_plan_compiler_flags) # executables that require memgraph_lib should link MEMGRAPH_ALL_LIBS to link all dependant libraries @@ -377,6 +379,7 @@ endif() # STATIC PIC library used by query engine add_library(memgraph_pic STATIC ${memgraph_src_files}) +target_link_libraries(memgraph_pic stdc++fs) add_dependencies(memgraph_pic generate_opencypher_parser generate_plan_compiler_flags) set_property(TARGET memgraph_pic PROPERTY POSITION_INDEPENDENT_CODE TRUE) diff --git a/config/memgraph.yaml b/config/memgraph.yaml index a815b8ff1..9aeeaf937 100644 --- a/config/memgraph.yaml +++ b/config/memgraph.yaml @@ -22,7 +22,8 @@ cleaning_cycle_sec: "30" snapshot_cycle_sec: "60" # max number of snapshots which will be kept on the disk at some point -max_retained_snapshots: "3" +# if set to -1 the max number of snapshots is unlimited +max_retained_snapshots: "-1" # by default query engine runs in interpret mode interpret: true diff --git a/src/config/config.hpp b/src/config/config.hpp index 788d21885..9178baa90 100644 --- a/src/config/config.hpp +++ b/src/config/config.hpp @@ -29,7 +29,7 @@ constexpr const char *MAX_RETAINED_SNAPSHOTS = "max_retained_snapshots"; constexpr const char *INTERPRET = "interpret"; // -- all possible Memgraph's keys -- -inline size_t to_int(std::string &s) { return stoull(s); } +inline long long to_int(const std::string &s) { return stoll(s); } // TODO: move this to register args because it doesn't make sense to convert // str to bool for every lookup inline bool to_bool(std::string &s) { diff --git a/src/database/graph_db.cpp b/src/database/graph_db.cpp index bafb3f9c5..d0fdb11b8 100644 --- a/src/database/graph_db.cpp +++ b/src/database/graph_db.cpp @@ -3,12 +3,15 @@ #include "config/config.hpp" #include "database/creation_exception.hpp" #include "database/graph_db.hpp" +#include "database/graph_db_accessor.hpp" #include "logging/logger.hpp" #include "storage/edge.hpp" #include "storage/garbage_collector.hpp" -//#include "snapshot/snapshoter.hpp" const int DEFAULT_CLEANING_CYCLE_SEC = 30; // 30 seconds +const std::string DEFAULT_SNAPSHOT_FOLDER = "snapshots"; +const int DEFAULT_MAX_RETAINED_SNAPSHOTS = -1; // unlimited number of snapshots +const int DEFAULT_SNAPSHOT_CYCLE_SEC = -1; // off GraphDb::GraphDb(const std::string &name, bool import_snapshot) : name_(name), @@ -44,14 +47,45 @@ GraphDb::GraphDb(const std::string &name, bool import_snapshot) this->vertex_version_list_deleter_.FreeExpiredObjects(id); }); } - // if (import_snapshot) - // snap_engine.import(); + + // Creating snapshoter + const std::string max_retained_snapshots_str = + CONFIG(config::MAX_RETAINED_SNAPSHOTS); + const std::string snapshot_cycle_sec_str = + CONFIG(config::MAX_RETAINED_SNAPSHOTS); + const std::string snapshot_folder_str = CONFIG(config::SNAPSHOTS_PATH); + + int max_retained_snapshots_ = DEFAULT_MAX_RETAINED_SNAPSHOTS; + if (!max_retained_snapshots_str.empty()) + max_retained_snapshots_ = CONFIG_INTEGER(config::MAX_RETAINED_SNAPSHOTS); + + int snapshot_cycle_sec_ = DEFAULT_SNAPSHOT_CYCLE_SEC; + if (!snapshot_cycle_sec_str.empty()) + snapshot_cycle_sec_ = CONFIG_INTEGER(config::SNAPSHOT_CYCLE_SEC); + + std::string snapshot_folder_ = DEFAULT_SNAPSHOT_FOLDER; + if (!snapshot_folder_str.empty()) snapshot_folder_ = snapshot_folder_str; + + if (snapshot_cycle_sec_ != -1) { + auto create_snapshot = [this, snapshot_folder_, + max_retained_snapshots_]() -> void { + GraphDbAccessor db_accessor(*this); + snapshooter_.MakeSnapshot(db_accessor, fs::path(snapshot_folder_) / name_, + max_retained_snapshots_); + }; + snapshot_creator_.Run(std::chrono::seconds(snapshot_cycle_sec_), + create_snapshot); + } } GraphDb::~GraphDb() { // Stop the gc scheduler to not run into race conditions for deletions. gc_scheduler_.Stop(); + // Stop the snapshot creator to avoid snapshooting while database is beeing + // deleted. + snapshot_creator_.Stop(); + // Delete vertices and edges which weren't collected before, also deletes // records inside version list for (auto &vertex : this->vertices_.access()) delete vertex; diff --git a/src/database/graph_db.hpp b/src/database/graph_db.hpp index 6322a968f..c6efe71e1 100644 --- a/src/database/graph_db.hpp +++ b/src/database/graph_db.hpp @@ -18,6 +18,7 @@ #include "storage/vertex.hpp" #include "transactions/engine.hpp" #include "utils/scheduler.hpp" +#include "durability/snapshooter.hpp" // TODO: Maybe split this in another layer between Db and Dbms. Where the new // layer would hold SnapshotEngine and his kind of concept objects. Some @@ -85,12 +86,16 @@ class GraphDb { ConcurrentSet labels_; ConcurrentSet edge_types_; ConcurrentSet properties_; - + // indexes KeyIndex labels_index_; KeyIndex edge_types_index_; LabelPropertyIndex label_property_index_; + // snapshooter + Snapshooter snapshooter_; + // Schedulers Scheduler gc_scheduler_; + Scheduler snapshot_creator_; }; diff --git a/src/durability/file_writer_buffer.hpp b/src/durability/file_writer_buffer.hpp new file mode 100644 index 000000000..017b686a3 --- /dev/null +++ b/src/durability/file_writer_buffer.hpp @@ -0,0 +1,98 @@ +#pragma once + +#include +#include "utils/bswap.hpp" + +/** + * Buffer that writes data to file and calculates hash of written data. + * Implements template param Buffer interface from BaseEncoder class. Hash is + * incremented when Write is called. If any ofstream operation fails, + * std::ifstream::failure is thrown. + */ +class FileWriterBuffer { + public: + /** + * Constructor, initialize ofstream to throw exception on fail. + */ + FileWriterBuffer() { + output_stream_.exceptions(std::ifstream::failbit | std::ifstream::badbit); + } + + /** + * Opens ofstream to file given in constructor. + * @param file: + * path to ofstream file + */ + void Open(const std::string &file) { + output_stream_.open(file, std::ios::out | std::ios::binary); + } + + /** + * Closes ofstream. + */ + void Close() { output_stream_.close(); } + + /** + * Writes data to stream and increases hash. + * @param data: + * pointer to data. + * @param n: + * data length. + */ + void Write(const uint8_t *data, size_t n) { + UpdateHash(data, n); + output_stream_.write(reinterpret_cast(data), n); + } + /** + * BaseEncoder needs this method, it is not needed in this buffer. + */ + void Chunk() {} + /** + * Flushes data to stream. + */ + void Flush() { output_stream_.flush(); } + + /** + * Writes summary to ofstream in big endian format. This method should be + * called when writing all other data in the file is done. Returns true if + * writing was successful. + */ + void WriteSummary(int64_t vertex_num, int64_t edge_num) { + debug_assert(vertex_num >= 0, "Number of vertices should't be negative"); + debug_assert(vertex_num >= 0, "Number of edges should't be negative"); + WriteLong(vertex_num); + WriteLong(edge_num); + WriteLong(hash_); + } + + private: + /** + * Hash function is H(n) = H(n-1) * prime + data where data is unsigned char. + * TODO implement different hash function + */ + void UpdateHash(const uint8_t *data, size_t n) { + for (int i = 0; i < n; ++i) hash_ = hash_ * kPrime + data[i] + 1; + } + + /** + * Method writes uint64_t to ofstream. + */ + void WriteLong(uint64_t val) { + uint64_t bval = bswap(val); + output_stream_.write(reinterpret_cast(&bval), + sizeof(bval)); + } + + /** + * Stream used to write data to file. + */ + std::ofstream output_stream_; + /** + * Represents hash of current data. + */ + uint64_t hash_ = 0; + /** + * Prime number used for hashing. + */ + const uint64_t kPrime = 3137; +}; diff --git a/src/durability/snapshooter.cpp b/src/durability/snapshooter.cpp new file mode 100644 index 000000000..5fc089162 --- /dev/null +++ b/src/durability/snapshooter.cpp @@ -0,0 +1,82 @@ +#include "durability/snapshooter.hpp" +#include +#include "communication/bolt/v1/encoder/base_encoder.hpp" +#include "config/config.hpp" +#include "database/graph_db_accessor.hpp" +#include "durability/file_writer_buffer.hpp" +#include "utils/datetime/timestamp.hpp" + +bool Snapshooter::MakeSnapshot(GraphDbAccessor &db_accessor_, + const fs::path &snapshot_folder, + const int max_retained_snapshots) { + if (!fs::exists(snapshot_folder) && + !fs::create_directories(snapshot_folder)) { + logger.error("Error while creating directory \"{}\"", snapshot_folder); + return false; + } + const auto snapshot_file = GetSnapshotFileName(snapshot_folder); + if (fs::exists(snapshot_file)) return false; + if (Encode(snapshot_file, db_accessor_)) { + MaintainMaxRetainedFiles(snapshot_folder, max_retained_snapshots); + return true; + } + return false; +} + +bool Snapshooter::Encode(const fs::path &snapshot_file, + GraphDbAccessor &db_accessor_) { + try { + FileWriterBuffer buffer; + // BaseEncoder encodes graph elements. Flag true is for storing vertex IDs. + communication::bolt::BaseEncoder encoder(buffer, true); + int64_t vertex_num = 0, edge_num = 0; + + buffer.Open(snapshot_file); + for (const auto &vertex : db_accessor_.vertices()) { + encoder.WriteVertex(vertex); + vertex_num++; + } + for (const auto &edge : db_accessor_.edges()) { + encoder.WriteEdge(edge); + edge_num++; + } + buffer.WriteSummary(vertex_num, edge_num); + buffer.Close(); + } catch (std::ifstream::failure e) { + if (fs::exists(snapshot_file) && !fs::remove(snapshot_file)) { + logger.error("Error while removing corrupted snapshot file \"{}\"", + snapshot_file); + } + return false; + } + return true; +} + +fs::path Snapshooter::GetSnapshotFileName(const fs::path &snapshot_folder) { + std::string date_str = + Timestamp(Timestamp::now()) + .to_string("{:04d}_{:02d}_{:02d}__{:02d}_{:02d}_{:02d}_{:05d}"); + return snapshot_folder / date_str; +} + +std::vector Snapshooter::GetSnapshotFiles( + const fs::path &snapshot_folder) { + std::vector files; + for (auto &file : fs::directory_iterator(snapshot_folder)) + files.push_back(file.path()); + return files; +} + +void Snapshooter::MaintainMaxRetainedFiles(const fs::path &snapshot_folder, + int max_retained_snapshots) { + if (max_retained_snapshots == -1) return; + std::vector files = GetSnapshotFiles(snapshot_folder); + if (static_cast(files.size()) <= max_retained_snapshots) return; + sort(files.begin(), files.end()); + for (int i = 0; i < static_cast(files.size()) - max_retained_snapshots; + ++i) { + if (!fs::remove(files[i])) { + logger.error("Error while removing file \"{}\"", files[i]); + } + } +} diff --git a/src/durability/snapshooter.hpp b/src/durability/snapshooter.hpp new file mode 100644 index 000000000..a99adcaf3 --- /dev/null +++ b/src/durability/snapshooter.hpp @@ -0,0 +1,57 @@ +#pragma once + +#include "logging/loggable.hpp" + +#include +#include +#include + +namespace fs = std::experimental::filesystem; + +class GraphDbAccessor; + +/** + * Class responsible for making snapshots. Snapshots are stored in folder + * memgraph/build/$snapshot_folder/$db_name using bolt protocol. + */ +class Snapshooter : public Loggable { + public: + Snapshooter() : Loggable("Snapshoter"){}; + /** + * Make snapshot and save it in snapshots folder. Returns true if successful. + * @param db_accessor: + * GraphDbAccessor used to access elements of GraphDb. + * @param snapshot_folder: + * folder where snapshots are stored. + * @param max_retained_snapshots: + * maximum number of snapshots stored in snapshot folder. + */ + bool MakeSnapshot(GraphDbAccessor &db_accessor, + const fs::path &snapshot_folder, + int max_retained_snapshots); + + private: + /** + * Method returns path to new snapshot file in format + * memgraph/build/$snapshot_folder/$db_name/$timestamp + */ + fs::path GetSnapshotFileName(const fs::path &snapshot_folder); + /** + * Method used to keep given number of snapshots in snapshot folder. Newest + * max_retained_files snapshots are kept, other snapshots are deleted. If + * max_retained_files is -1, all snapshots are kept. + */ + void MaintainMaxRetainedFiles(const fs::path &snapshot_folder, + const int max_retained_files); + /** + * Function returns list of snapshot files in snapshot folder. + */ + std::vector GetSnapshotFiles(const fs::path &snapshot_folder); + + /** + * Encodes graph and stores it in file given as parameter. Graph elements are + * accessed using parameter db_accessor. If function is successfully executed, + * true is returned. + */ + bool Encode(const fs::path &snapshot_file, GraphDbAccessor &db_accessor); +}; diff --git a/src/utils/datetime/timestamp.hpp b/src/utils/datetime/timestamp.hpp index 0461a3afc..34baf47ce 100644 --- a/src/utils/datetime/timestamp.hpp +++ b/src/utils/datetime/timestamp.hpp @@ -14,7 +14,8 @@ class Timestamp : public TotalOrdering { public: Timestamp() : Timestamp(0, 0) {} - Timestamp(std::time_t time, long nsec = 0) : unix_time(time), nsec(nsec) { + Timestamp(std::time_t time, long nsec = 0) + : unix_time(time), nsec(nsec) { auto result = gmtime_r(&time, &this->time); if (result == nullptr) @@ -50,11 +51,16 @@ class Timestamp : public TotalOrdering { subsec()); } + const std::string to_string(const std::string &format = fiso8601) const { + return fmt::format(format, year(), month(), day(), hour(), min(), sec(), + subsec()); + } + friend std::ostream& operator<<(std::ostream& stream, const Timestamp& ts) { return stream << ts.to_iso8601(); } - operator std::string() const { return to_iso8601(); } + operator std::string() const { return to_string(); } constexpr friend bool operator==(const Timestamp& a, const Timestamp& b) { return a.unix_time == b.unix_time && a.nsec == b.nsec; diff --git a/tests/unit/snapshot.cpp b/tests/unit/snapshot.cpp new file mode 100644 index 000000000..52cd09a0d --- /dev/null +++ b/tests/unit/snapshot.cpp @@ -0,0 +1,101 @@ +#include +#include "dbms/dbms.hpp" +#include "durability/snapshooter.hpp" +#include "gtest/gtest.h" + +namespace fs = std::experimental::filesystem; + +const std::string SNAPSHOTS_FOLDER_ALL_DB = "snapshots_test"; +const std::string SNAPSHOTS_TEST_DEFAULT_DB_DIR = "snapshots_test/default"; + +// Other functionality will be tested in recovery tests. + +std::vector GetFilesFromDir( + const std::string &snapshots_default_db_dir) { + std::vector files; + for (auto &file : fs::directory_iterator(snapshots_default_db_dir)) + files.push_back(file.path()); + return files; +} + +void CleanDbDir() { + if (!fs::exists(SNAPSHOTS_TEST_DEFAULT_DB_DIR)) return; + std::vector files = GetFilesFromDir(SNAPSHOTS_TEST_DEFAULT_DB_DIR); + for (auto file : files) { + fs::remove(file); + } +} + +class SnapshotTest : public ::testing::Test { + protected: + virtual void TearDown() { + CleanDbDir(); + CONFIG(config::SNAPSHOT_CYCLE_SEC) = snapshot_cycle_sec_setup_; + } + + virtual void SetUp() { + CleanDbDir(); + snapshot_cycle_sec_setup_ = CONFIG(config::SNAPSHOT_CYCLE_SEC); + CONFIG(config::SNAPSHOT_CYCLE_SEC) = "-1"; + } + std::string snapshot_cycle_sec_setup_; +}; + +TEST_F(SnapshotTest, CreateLessThanMaxRetainedSnapshotsTests) { + const int max_retained_snapshots = 10; + Dbms dbms; + + for (int i = 0; i < 3; ++i) { + auto dba = dbms.active(); + Snapshooter snapshooter; + snapshooter.MakeSnapshot(*dba.get(), SNAPSHOTS_TEST_DEFAULT_DB_DIR, + max_retained_snapshots); + } + + std::vector files = GetFilesFromDir(SNAPSHOTS_TEST_DEFAULT_DB_DIR); + EXPECT_EQ(files.size(), 3); +} + +TEST_F(SnapshotTest, CreateMoreThanMaxRetainedSnapshotsTests) { + const int max_retained_snapshots = 2; + Dbms dbms; + + fs::path first_snapshot; + for (int i = 0; i < 3; ++i) { + auto dba = dbms.active(); + Snapshooter snapshooter; + snapshooter.MakeSnapshot(*dba.get(), SNAPSHOTS_TEST_DEFAULT_DB_DIR, + max_retained_snapshots); + if (i == 0) { + std::vector files_begin = + GetFilesFromDir(SNAPSHOTS_TEST_DEFAULT_DB_DIR); + EXPECT_EQ(files_begin.size(), 1); + first_snapshot = files_begin[0]; + } + } + + std::vector files_end = + GetFilesFromDir(SNAPSHOTS_TEST_DEFAULT_DB_DIR); + EXPECT_EQ(files_end.size(), 2); + EXPECT_EQ(fs::exists(first_snapshot), false); +} + +TEST_F(SnapshotTest, CreateSnapshotWithUnlimitedMaxRetainedSnapshots) { + const int max_retained_snapshots = -1; + Dbms dbms; + + for (int i = 0; i < 10; ++i) { + auto dba = dbms.active(); + Snapshooter snapshooter; + snapshooter.MakeSnapshot(*dba.get(), SNAPSHOTS_TEST_DEFAULT_DB_DIR, + max_retained_snapshots); + } + + std::vector files = GetFilesFromDir(SNAPSHOTS_TEST_DEFAULT_DB_DIR); + EXPECT_EQ(files.size(), 10); +} + +int main(int argc, char **argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}