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

@@ -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)

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

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) {

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;

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<std::string> labels_;
   ConcurrentSet<std::string> edge_types_;
   ConcurrentSet<std::string> properties_;
-
+ 
   // indexes
   KeyIndex<GraphDbTypes::Label, Vertex> labels_index_;
   KeyIndex<GraphDbTypes::EdgeType, Edge> edge_types_index_;
   LabelPropertyIndex label_property_index_;
 
+  // snapshooter
+  Snapshooter snapshooter_;
+
   // Schedulers
   Scheduler<std::mutex> gc_scheduler_;
+  Scheduler<std::mutex> snapshot_creator_;
 };

src/durability/file_writer_buffer.hpp

@@ -0,0 +1,98 @@
+#pragma once
+
+#include <fstream>
+#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<const char *>(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<const char *>(&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;
+};

src/durability/snapshooter.cpp

@@ -0,0 +1,82 @@
+#include "durability/snapshooter.hpp"
+#include <algorithm>
+#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<FileWriterBuffer> 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<fs::path> Snapshooter::GetSnapshotFiles(
+    const fs::path &snapshot_folder) {
+  std::vector<fs::path> 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<fs::path> files = GetSnapshotFiles(snapshot_folder);
+  if (static_cast<int>(files.size()) <= max_retained_snapshots) return;
+  sort(files.begin(), files.end());
+  for (int i = 0; i < static_cast<int>(files.size()) - max_retained_snapshots;
+       ++i) {
+    if (!fs::remove(files[i])) {
+      logger.error("Error while removing file \"{}\"", files[i]);
+    }
+  }
+}

src/durability/snapshooter.hpp

@@ -0,0 +1,57 @@
+#pragma once
+
+#include "logging/loggable.hpp"
+
+#include <cstring>
+#include <experimental/filesystem>
+#include <vector>
+
+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<fs::path> 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);
+};

src/utils/datetime/timestamp.hpp

@@ -14,7 +14,8 @@ class Timestamp : public TotalOrdering<Timestamp> {
  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<Timestamp> {
                        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;

tests/unit/snapshot.cpp

@@ -0,0 +1,101 @@
+#include <experimental/filesystem>
+#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<fs::path> GetFilesFromDir(
+    const std::string &snapshots_default_db_dir) {
+  std::vector<fs::path> 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<fs::path> 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<fs::path> 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<fs::path> files_begin =
+          GetFilesFromDir(SNAPSHOTS_TEST_DEFAULT_DB_DIR);
+      EXPECT_EQ(files_begin.size(), 1);
+      first_snapshot = files_begin[0];
+    }
+  }
+
+  std::vector<fs::path> 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<fs::path> 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();
+}