graph_db.h

Go to the documentation of this file.

 
#ifndef GRAPHSCOPE_DATABASE_GRAPH_DB_H_
#define GRAPHSCOPE_DATABASE_GRAPH_DB_H_
 
#include <dlfcn.h>
 
#include <map>
#include <mutex>
#include <shared_mutex>
#include <thread>
#include <vector>
 
#include "flex/engines/graph_db/app/app_base.h"
#include "flex/engines/graph_db/database/insert_transaction.h"
#include "flex/engines/graph_db/database/read_transaction.h"
#include "flex/engines/graph_db/database/single_edge_insert_transaction.h"
#include "flex/engines/graph_db/database/single_vertex_insert_transaction.h"
#include "flex/engines/graph_db/database/update_transaction.h"
#include "flex/engines/graph_db/database/version_manager.h"
#include "flex/storages/rt_mutable_graph/loader/loader_factory.h"
#include "flex/storages/rt_mutable_graph/loading_config.h"
#include "flex/storages/rt_mutable_graph/mutable_property_fragment.h"
 
namespace gs {
 
class GraphDB;
class GraphDBSession;
struct SessionLocalContext;
 
struct GraphDBConfig {
  GraphDBConfig(const Schema& schema_, const std::string& data_dir_,
                const std::string& compiler_path_ = "", int thread_num_ = 1)
      : schema(schema_),
        data_dir(data_dir_),
        compiler_path(compiler_path_),
        thread_num(thread_num_),
        warmup(false),
        enable_monitoring(false),
        enable_auto_compaction(false),
        memory_level(1) {}
 
  Schema schema;
  std::string data_dir;
  std::string compiler_path;
  int thread_num;
  bool warmup;
  bool enable_monitoring;
  bool enable_auto_compaction;
 
  /*
    0 - sync with disk;
    1 - mmap virtual memory;
    2 - preferring hugepages;
    3 - force hugepages;
  */
  int memory_level;
};
 
struct QueryCache {
  bool get(const std::string& key, std::string_view& value) {
    std::shared_lock<std::shared_mutex> lock(mutex);
    auto it = cache.find(key);
    if (it != cache.end()) {
      value = it->second;
      return true;
    }
    return false;
  }
 
  void put(const std::string& key, const std::string_view& value) {
    std::unique_lock<std::shared_mutex> lock(mutex);
    cache[key] = value;
  }
  std::shared_mutex mutex;
  std::unordered_map<std::string, std::string> cache;
};
 
class GraphDB {
 public:
  GraphDB();
  ~GraphDB();
 
  static GraphDB& get();
 
  Result<bool> Open(const Schema& schema, const std::string& data_dir,
                    int32_t thread_num = 1, bool warmup = false,
                    bool memory_only = true,
                    bool enable_auto_compaction = false);
 
  Result<bool> Open(const GraphDBConfig& config);
 
  void Close();
 
  ReadTransaction GetReadTransaction(int thread_id = 0);
 
  InsertTransaction GetInsertTransaction(int thread_id = 0);
 
  SingleVertexInsertTransaction GetSingleVertexInsertTransaction(
      int thread_id = 0);
 
  SingleEdgeInsertTransaction GetSingleEdgeInsertTransaction(int thread_id = 0);
 
  UpdateTransaction GetUpdateTransaction(int thread_id = 0);
 
  inline const MutablePropertyFragment& graph() const { return graph_; }
  inline MutablePropertyFragment& graph() { return graph_; }
 
  inline const Schema& schema() const { return graph_.schema(); }
 
  inline std::shared_ptr<ColumnBase> get_vertex_property_column(
      uint8_t label, const std::string& col_name) const {
    return graph_.get_vertex_table(label).get_column(col_name);
  }
 
  inline std::shared_ptr<RefColumnBase> get_vertex_id_column(
      uint8_t label) const {
    return graph_.get_vertex_id_column(label);
  }
 
  AppWrapper CreateApp(uint8_t app_type, int thread_id);
 
  void GetAppInfo(Encoder& result);
 
  GraphDBSession& GetSession(int thread_id);
  const GraphDBSession& GetSession(int thread_id) const;
 
  int SessionNum() const;
 
  void UpdateCompactionTimestamp(timestamp_t ts);
  timestamp_t GetLastCompactionTimestamp() const;
 
  QueryCache& getQueryCache() const;
 
  std::string work_dir() const { return work_dir_; }
 
  void OutputCypherProfiles(const std::string& prefix);
 
 private:
  bool registerApp(const std::string& path, uint8_t index = 0);
 
  void ingestWals(const std::vector<std::string>& wals,
                  const std::string& work_dir, int thread_num);
 
  void initApps(
      const std::unordered_map<std::string, std::pair<std::string, uint8_t>>&
          plugins);
 
  void openWalAndCreateContexts(const std::string& data_dir_path,
                                MemoryStrategy allocator_strategy);
 
  void showAppMetrics() const;
 
  size_t getExecutedQueryNum() const;
 
  friend class GraphDBSession;
 
  std::string work_dir_;
  SessionLocalContext* contexts_;
 
  int thread_num_;
 
  MutablePropertyFragment graph_;
  VersionManager version_manager_;
 
  std::array<std::string, 256> app_paths_;
  std::array<std::shared_ptr<AppFactoryBase>, 256> app_factories_;
 
  mutable QueryCache query_cache_;
 
  std::thread monitor_thread_;
  bool monitor_thread_running_;
 
  timestamp_t last_compaction_ts_;
  bool compact_thread_running_ = false;
  std::thread compact_thread_;
};
 
}  // namespace gs
 
#endif  // GRAPHSCOPE_DATABASE_GRAPH_DB_H_