immutable_csr.h

Go to the documentation of this file.

 
#ifndef STORAGES_RT_MUTABLE_GRAPH_CSR_IMMUTABLE_CSR_H_
#define STORAGES_RT_MUTABLE_GRAPH_CSR_IMMUTABLE_CSR_H_
 
namespace gs {
 
template <typename EDATA_T>
class ImmutableCsrConstEdgeIter : public CsrConstEdgeIterBase {
  using const_nbr_ptr_t = typename ImmutableNbrSlice<EDATA_T>::const_nbr_ptr_t;
 
 public:
  explicit ImmutableCsrConstEdgeIter(const ImmutableNbrSlice<EDATA_T>& slice)
      : cur_(slice.begin()), end_(slice.end()) {}
  ~ImmutableCsrConstEdgeIter() = default;
 
  vid_t get_neighbor() const override { return (*cur_).get_neighbor(); }
  Any get_data() const override {
    return AnyConverter<EDATA_T>::to_any((*cur_).get_data());
  }
  timestamp_t get_timestamp() const override { return 0; }
 
  void next() override { ++cur_; }
  CsrConstEdgeIterBase& operator+=(size_t offset) override {
    cur_ += offset;
    if (!(cur_ < end_)) {
      cur_ = end_;
    }
    return *this;
  }
  bool is_valid() const override { return cur_ != end_; }
  size_t size() const override { return end_ - cur_; }
 
 private:
  const_nbr_ptr_t cur_;
  const_nbr_ptr_t end_;
};
 
template <typename EDATA_T>
class ImmutableCsr : public TypedImmutableCsrBase<EDATA_T> {
 public:
  using nbr_t = ImmutableNbr<EDATA_T>;
  using slice_t = ImmutableNbrSlice<EDATA_T>;
 
  size_t batch_init(const std::string& name, const std::string& work_dir,
                    const std::vector<int>& degree,
                    double reserve_ratio) override {
    size_t vnum = degree.size();
    adj_lists_.open(work_dir + "/" + name + ".adj", true);
    adj_lists_.resize(vnum);
 
    size_t edge_num = 0;
    for (auto d : degree) {
      edge_num += d;
    }
 
    nbr_list_.open(work_dir + "/" + name + ".nbr", true);
    nbr_list_.resize(edge_num);
 
    degree_list_.open(work_dir + "/" + name + ".deg", true);
    degree_list_.resize(vnum);
 
    nbr_t* ptr = nbr_list_.data();
    for (vid_t i = 0; i < vnum; ++i) {
      int deg = degree[i];
      if (deg != 0) {
        adj_lists_[i] = ptr;
      } else {
        adj_lists_[i] = NULL;
      }
      ptr += deg;
 
      degree_list_[i] = 0;
    }
 
    unsorted_since_ = 0;
    return edge_num;
  }
 
  size_t batch_init_in_memory(const std::vector<int>& degree,
                              double reserve_ratio) override {
    size_t vnum = degree.size();
    adj_lists_.open("", false);
    adj_lists_.resize(vnum);
 
    size_t edge_num = 0;
    for (auto d : degree) {
      edge_num += d;
    }
 
    nbr_list_.open("", false);
    nbr_list_.resize(edge_num);
 
    degree_list_.open("", false);
    degree_list_.resize(vnum);
 
    nbr_t* ptr = nbr_list_.data();
    for (vid_t i = 0; i < vnum; ++i) {
      int deg = degree[i];
      if (deg != 0) {
        adj_lists_[i] = ptr;
      } else {
        adj_lists_[i] = NULL;
      }
      ptr += deg;
 
      degree_list_[i] = 0;
    }
 
    unsorted_since_ = 0;
    return edge_num;
  }
 
  void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
                      timestamp_t ts) override {
    auto& nbr = adj_lists_[src][degree_list_[src]++];
    nbr.neighbor = dst;
    nbr.data = data;
  }
 
  void batch_sort_by_edge_data(timestamp_t ts) override {
    size_t vnum = adj_lists_.size();
    for (size_t i = 0; i != vnum; ++i) {
      std::sort(adj_lists_[i], adj_lists_[i] + degree_list_[i],
                [](const nbr_t& lhs, const nbr_t& rhs) {
                  return lhs.data < rhs.data;
                });
    }
    unsorted_since_ = ts;
  }
 
  timestamp_t unsorted_since() const override { return unsorted_since_; }
 
  void open(const std::string& name, const std::string& snapshot_dir,
            const std::string& work_dir) override {
    // Changes made to the CSR will not be synchronized to the file
    // TODO(luoxiaojian): Implement the insert operation on ImmutableCsr.
    if (snapshot_dir != "") {
      degree_list_.open(snapshot_dir + "/" + name + ".deg", false);
      nbr_list_.open(snapshot_dir + "/" + name + ".nbr", false);
      load_meta(snapshot_dir + "/" + name);
    }
 
    adj_lists_.open(work_dir + "/" + name + ".adj", true);
    adj_lists_.resize(degree_list_.size());
 
    nbr_t* ptr = nbr_list_.data();
    for (size_t i = 0; i < degree_list_.size(); ++i) {
      int deg = degree_list_[i];
      adj_lists_[i] = ptr;
      ptr += deg;
    }
  }
 
  void open_in_memory(const std::string& prefix, size_t v_cap) override {
    degree_list_.open(prefix + ".deg", false);
    load_meta(prefix);
    nbr_list_.open(prefix + ".nbr", false);
    adj_lists_.reset();
    v_cap = std::max(v_cap, degree_list_.size());
    adj_lists_.resize(v_cap);
    size_t old_degree_size = degree_list_.size();
    degree_list_.resize(v_cap);
 
    nbr_t* ptr = nbr_list_.data();
    for (size_t i = 0; i < old_degree_size; ++i) {
      int deg = degree_list_[i];
      if (deg != 0) {
        adj_lists_[i] = ptr;
      } else {
        adj_lists_[i] = NULL;
      }
      ptr += deg;
    }
    for (size_t i = old_degree_size; i < degree_list_.size(); ++i) {
      degree_list_[i] = 0;
      adj_lists_[i] = NULL;
    }
  }
 
  void open_with_hugepages(const std::string& prefix, size_t v_cap) override {
    degree_list_.open_with_hugepages(prefix + ".deg", v_cap);
    load_meta(prefix);
    nbr_list_.open_with_hugepages(prefix + ".nbr");
    adj_lists_.reset();
    v_cap = std::max(v_cap, degree_list_.size());
    adj_lists_.resize(v_cap);
    size_t old_degree_size = degree_list_.size();
    degree_list_.resize(v_cap);
 
    nbr_t* ptr = nbr_list_.data();
    for (size_t i = 0; i < old_degree_size; ++i) {
      int deg = degree_list_[i];
      if (deg != 0) {
        adj_lists_[i] = ptr;
      } else {
        adj_lists_[i] = NULL;
      }
      ptr += deg;
    }
    for (size_t i = old_degree_size; i < degree_list_.size(); ++i) {
      degree_list_[i] = 0;
      adj_lists_[i] = NULL;
    }
  }
 
  void dump(const std::string& name,
            const std::string& new_snapshot_dir) override {
    dump_meta(new_snapshot_dir + "/" + name);
    size_t vnum = adj_lists_.size();
    {
      FILE* fout =
          fopen((new_snapshot_dir + "/" + name + ".deg").c_str(), "wb");
      fwrite(degree_list_.data(), sizeof(int), vnum, fout);
      fflush(fout);
      fclose(fout);
    }
    {
      FILE* fout =
          fopen((new_snapshot_dir + "/" + name + ".nbr").c_str(), "wb");
      for (size_t k = 0; k < vnum; ++k) {
        if (adj_lists_[k] != NULL && degree_list_[k] != 0) {
          fwrite(adj_lists_[k], sizeof(nbr_t), degree_list_[k], fout);
        }
      }
      fflush(fout);
      fclose(fout);
    }
  }
 
  void warmup(int thread_num) const override {}
 
  void resize(vid_t vnum) override {
    if (vnum > adj_lists_.size()) {
      size_t old_size = adj_lists_.size();
      adj_lists_.resize(vnum);
      for (size_t k = old_size; k != vnum; ++k) {
        adj_lists_[k] = NULL;
        degree_list_[k] = 0;
      }
    } else {
      adj_lists_.resize(vnum);
      degree_list_.resize(vnum);
    }
  }
 
  size_t size() const override { return adj_lists_.size(); }
 
  size_t edge_num() const override {
    size_t ret = 0;
    for (size_t i = 0; i < adj_lists_.size(); ++i) {
      ret += degree_list_[i];
    }
    return ret;
  }
 
  std::shared_ptr<CsrConstEdgeIterBase> edge_iter(vid_t v) const override {
    return std::make_shared<ImmutableCsrConstEdgeIter<EDATA_T>>(get_edges(v));
  }
  CsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
    return new ImmutableCsrConstEdgeIter<EDATA_T>(get_edges(v));
  }
  std::shared_ptr<CsrEdgeIterBase> edge_iter_mut(vid_t v) override {
    return nullptr;
  }
 
  void put_edge(vid_t src, vid_t dst, const EDATA_T& data, timestamp_t ts,
                Allocator& alloc) override {
    LOG(FATAL) << "Put single edge is not supported";
  }
 
  slice_t get_edges(vid_t v) const override {
    slice_t ret;
    ret.set_begin(adj_lists_[v]);
    ret.set_size(degree_list_[v]);
    return ret;
  }
 
  void close() override {
    adj_lists_.reset();
    degree_list_.reset();
    nbr_list_.reset();
  }
 
 private:
  void load_meta(const std::string& prefix) {
    std::string meta_file_path = prefix + ".meta";
    if (std::filesystem::exists(meta_file_path)) {
      FILE* meta_file_fd = fopen(meta_file_path.c_str(), "r");
      CHECK_EQ(fread(&unsorted_since_, sizeof(timestamp_t), 1, meta_file_fd),
               1);
      fclose(meta_file_fd);
    } else {
      unsorted_since_ = 0;
    }
  }
 
  void dump_meta(const std::string& prefix) const {
    std::string meta_file_path = prefix + ".meta";
    FILE* meta_file_fd = fopen((prefix + ".meta").c_str(), "wb");
    CHECK_EQ(fwrite(&unsorted_since_, sizeof(timestamp_t), 1, meta_file_fd), 1);
    fflush(meta_file_fd);
    fclose(meta_file_fd);
  }
 
  mmap_array<nbr_t*> adj_lists_;
  mmap_array<int> degree_list_;
  mmap_array<nbr_t> nbr_list_;
  timestamp_t unsorted_since_;
};
 
template <typename EDATA_T>
class SingleImmutableCsr : public TypedImmutableCsrBase<EDATA_T> {
 public:
  using nbr_t = ImmutableNbr<EDATA_T>;
  using slice_t = ImmutableNbrSlice<EDATA_T>;
 
  SingleImmutableCsr() {}
  ~SingleImmutableCsr() {}
 
  size_t batch_init(const std::string& name, const std::string& work_dir,
                    const std::vector<int>& degree,
                    double reserve_ratio) override {
    size_t vnum = degree.size();
    nbr_list_.open(work_dir + "/" + name + ".snbr", true);
    nbr_list_.resize(vnum);
    for (size_t k = 0; k != vnum; ++k) {
      nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
    }
    return vnum;
  }
 
  size_t batch_init_in_memory(const std::vector<int>& degree,
                              double reserve_ratio) override {
    size_t vnum = degree.size();
    nbr_list_.open("", false);
    nbr_list_.resize(vnum);
    for (size_t k = 0; k != vnum; ++k) {
      nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
    }
    return vnum;
  }
 
  void batch_put_edge(vid_t src, vid_t dst, const EDATA_T& data,
                      timestamp_t ts) override {
    CHECK_EQ(nbr_list_[src].neighbor, std::numeric_limits<vid_t>::max());
    nbr_list_[src].neighbor = dst;
    nbr_list_[src].data = data;
  }
 
  void batch_sort_by_edge_data(timestamp_t ts) override {}
 
  timestamp_t unsorted_since() const override {
    return std::numeric_limits<timestamp_t>::max();
  }
 
  void open(const std::string& name, const std::string& snapshot_dir,
            const std::string& work_dir) override {
    if (!std::filesystem::exists(work_dir + "/" + name + ".snbr")) {
      copy_file(snapshot_dir + "/" + name + ".snbr",
                work_dir + "/" + name + ".snbr");
    }
    nbr_list_.open(work_dir + "/" + name + ".snbr", true);
  }
 
  void open_in_memory(const std::string& prefix, size_t v_cap) override {
    nbr_list_.open(prefix + ".snbr", false);
    if (nbr_list_.size() < v_cap) {
      size_t old_size = nbr_list_.size();
      nbr_list_.reset();
      nbr_list_.resize(v_cap);
      FILE* fin = fopen((prefix + ".snbr").c_str(), "r");
      CHECK_EQ(fread(nbr_list_.data(), sizeof(nbr_t), old_size, fin), old_size);
      fclose(fin);
      for (size_t k = old_size; k != v_cap; ++k) {
        nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
      }
    }
  }
 
  void open_with_hugepages(const std::string& prefix, size_t v_cap) override {
    nbr_list_.open_with_hugepages(prefix + ".snbr", v_cap);
    size_t old_size = nbr_list_.size();
    if (old_size < v_cap) {
      nbr_list_.resize(v_cap);
      for (size_t k = old_size; k != v_cap; ++k) {
        nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
      }
    }
  }
 
  void dump(const std::string& name,
            const std::string& new_snapshot_dir) override {
    if (!nbr_list_.filename().empty() &&
        std::filesystem::exists(nbr_list_.filename())) {
      std::filesystem::create_hard_link(
          nbr_list_.filename(), new_snapshot_dir + "/" + name + ".snbr");
    } else {
      FILE* fp = fopen((new_snapshot_dir + "/" + name + ".snbr").c_str(), "wb");
      fwrite(nbr_list_.data(), sizeof(nbr_t), nbr_list_.size(), fp);
      fflush(fp);
      fclose(fp);
    }
  }
 
  void warmup(int thread_num) const override {
    size_t vnum = nbr_list_.size();
    std::vector<std::thread> threads;
    std::atomic<size_t> v_i(0);
    std::atomic<size_t> output(0);
    const size_t chunk = 4096;
    for (int i = 0; i < thread_num; ++i) {
      threads.emplace_back([&]() {
        size_t ret = 0;
        while (true) {
          size_t begin = std::min(v_i.fetch_add(chunk), vnum);
          size_t end = std::min(begin + chunk, vnum);
          if (begin == end) {
            break;
          }
          while (begin < end) {
            auto& nbr = nbr_list_[begin];
            ret += nbr.neighbor;
            ++begin;
          }
        }
        output.fetch_add(ret);
      });
    }
    for (auto& thrd : threads) {
      thrd.join();
    }
    (void) output.load();
  }
 
  void resize(vid_t vnum) override {
    if (vnum > nbr_list_.size()) {
      size_t old_size = nbr_list_.size();
      nbr_list_.resize(vnum);
      for (size_t k = old_size; k != vnum; ++k) {
        nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
      }
    } else {
      nbr_list_.resize(vnum);
    }
  }
 
  size_t size() const override { return nbr_list_.size(); }
 
  size_t edge_num() const override {
    size_t ret = 0;
    for (size_t i = 0; i < nbr_list_.size(); ++i) {
      if (nbr_list_[i].neighbor != std::numeric_limits<vid_t>::max()) {
        ++ret;
      }
    }
    return ret;
  }
 
  std::shared_ptr<CsrConstEdgeIterBase> edge_iter(vid_t v) const override {
    return std::make_shared<ImmutableCsrConstEdgeIter<EDATA_T>>(get_edges(v));
  }
 
  CsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
    return new ImmutableCsrConstEdgeIter<EDATA_T>(get_edges(v));
  }
 
  std::shared_ptr<CsrEdgeIterBase> edge_iter_mut(vid_t v) override {
    return nullptr;
  }
 
  void put_edge(vid_t src, vid_t dst, const EDATA_T& data, timestamp_t ts,
                Allocator&) override {
    CHECK_LT(src, nbr_list_.size());
    CHECK_EQ(nbr_list_[src].neighbor, std::numeric_limits<vid_t>::max());
    nbr_list_[src].neighbor = dst;
    nbr_list_[src].data = data;
  }
 
  slice_t get_edges(vid_t i) const override {
    slice_t ret;
    ret.set_size(
        nbr_list_[i].neighbor == std::numeric_limits<vid_t>::max() ? 0 : 1);
    if (ret.size() != 0) {
      ret.set_begin(&nbr_list_[i]);
    }
    return ret;
  }
 
  void close() override { nbr_list_.reset(); }
 
  const nbr_t& get_edge(vid_t i) const { return nbr_list_[i]; }
 
 private:
  mmap_array<nbr_t> nbr_list_;
};
 
template <>
class SingleImmutableCsr<std::string_view>
    : public TypedImmutableCsrBase<std::string_view> {
 public:
  using nbr_t = ImmutableNbr<size_t>;
  using slice_t = ImmutableNbrSlice<std::string_view>;
 
  SingleImmutableCsr(StringColumn& column) : column_(column) {}
  ~SingleImmutableCsr() {}
 
  size_t batch_init(const std::string& name, const std::string& work_dir,
                    const std::vector<int>& degree,
                    double reserve_ratio) override {
    size_t vnum = degree.size();
    nbr_list_.open(work_dir + "/" + name + ".snbr", true);
    nbr_list_.resize(vnum);
    for (size_t k = 0; k != vnum; ++k) {
      nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
    }
    return vnum;
  }
 
  size_t batch_init_in_memory(const std::vector<int>& degree,
                              double reserve_ratio) override {
    size_t vnum = degree.size();
    nbr_list_.open("", false);
    nbr_list_.resize(vnum);
    for (size_t k = 0; k != vnum; ++k) {
      nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
    }
    return vnum;
  }
 
  void batch_put_edge_with_index(vid_t src, vid_t dst, size_t data,
                                 timestamp_t ts) override {
    CHECK_EQ(nbr_list_[src].neighbor, std::numeric_limits<vid_t>::max());
    nbr_list_[src].neighbor = dst;
    nbr_list_[src].data = data;
  }
 
  void batch_sort_by_edge_data(timestamp_t ts) override {}
 
  timestamp_t unsorted_since() const override {
    return std::numeric_limits<timestamp_t>::max();
  }
 
  void open(const std::string& name, const std::string& snapshot_dir,
            const std::string& work_dir) override {
    if (!std::filesystem::exists(work_dir + "/" + name + ".snbr")) {
      copy_file(snapshot_dir + "/" + name + ".snbr",
                work_dir + "/" + name + ".snbr");
    }
    nbr_list_.open(work_dir + "/" + name + ".snbr", true);
  }
 
  void open_in_memory(const std::string& prefix, size_t v_cap) override {
    nbr_list_.open(prefix + ".snbr", false);
    if (nbr_list_.size() < v_cap) {
      size_t old_size = nbr_list_.size();
      nbr_list_.reset();
      nbr_list_.resize(v_cap);
      FILE* fin = fopen((prefix + ".snbr").c_str(), "r");
      CHECK_EQ(fread(nbr_list_.data(), sizeof(nbr_t), old_size, fin), old_size);
      fclose(fin);
      for (size_t k = old_size; k != v_cap; ++k) {
        nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
      }
    }
  }
 
  void open_with_hugepages(const std::string& prefix, size_t v_cap) override {
    nbr_list_.open_with_hugepages(prefix + ".snbr", v_cap);
    size_t old_size = nbr_list_.size();
    if (old_size < v_cap) {
      nbr_list_.resize(v_cap);
      for (size_t k = old_size; k != v_cap; ++k) {
        nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
      }
    }
  }
 
  void dump(const std::string& name,
            const std::string& new_snapshot_dir) override {
    if (!nbr_list_.filename().empty() &&
        std::filesystem::exists(nbr_list_.filename())) {
      std::filesystem::create_hard_link(
          nbr_list_.filename(), new_snapshot_dir + "/" + name + ".snbr");
    } else {
      FILE* fp = fopen((new_snapshot_dir + "/" + name + ".snbr").c_str(), "wb");
      fwrite(nbr_list_.data(), sizeof(nbr_t), nbr_list_.size(), fp);
      fflush(fp);
      fclose(fp);
    }
  }
 
  void warmup(int thread_num) const override {
    size_t vnum = nbr_list_.size();
    std::vector<std::thread> threads;
    std::atomic<size_t> v_i(0);
    std::atomic<size_t> output(0);
    const size_t chunk = 4096;
    for (int i = 0; i < thread_num; ++i) {
      threads.emplace_back([&]() {
        size_t ret = 0;
        while (true) {
          size_t begin = std::min(v_i.fetch_add(chunk), vnum);
          size_t end = std::min(begin + chunk, vnum);
          if (begin == end) {
            break;
          }
          while (begin < end) {
            auto& nbr = nbr_list_[begin];
            ret += nbr.neighbor;
            ++begin;
          }
        }
        output.fetch_add(ret);
      });
    }
    for (auto& thrd : threads) {
      thrd.join();
    }
    (void) output.load();
  }
 
  void resize(vid_t vnum) override {
    if (vnum > nbr_list_.size()) {
      size_t old_size = nbr_list_.size();
      nbr_list_.resize(vnum);
      for (size_t k = old_size; k != vnum; ++k) {
        nbr_list_[k].neighbor = std::numeric_limits<vid_t>::max();
      }
    } else {
      nbr_list_.resize(vnum);
    }
  }
 
  size_t size() const override { return nbr_list_.size(); }
 
  size_t edge_num() const override {
    size_t ret = 0;
    for (size_t i = 0; i < nbr_list_.size(); ++i) {
      if (nbr_list_[i].neighbor != std::numeric_limits<vid_t>::max()) {
        ++ret;
      }
    }
    return ret;
  }
 
  std::shared_ptr<CsrConstEdgeIterBase> edge_iter(vid_t v) const override {
    return std::make_shared<ImmutableCsrConstEdgeIter<std::string_view>>(
        get_edges(v));
  }
 
  CsrConstEdgeIterBase* edge_iter_raw(vid_t v) const override {
    return new ImmutableCsrConstEdgeIter<std::string_view>(get_edges(v));
  }
 
  std::shared_ptr<CsrEdgeIterBase> edge_iter_mut(vid_t v) override {
    return nullptr;
  }
 
  void put_edge_with_index(vid_t src, vid_t dst, size_t data, timestamp_t ts,
                           Allocator&) override {
    CHECK_LT(src, nbr_list_.size());
    CHECK_EQ(nbr_list_[src].neighbor, std::numeric_limits<vid_t>::max());
    nbr_list_[src].neighbor = dst;
    nbr_list_[src].data = data;
  }
 
  slice_t get_edges(vid_t i) const override {
    slice_t ret(column_);
    ret.set_size(
        nbr_list_[i].neighbor == std::numeric_limits<vid_t>::max() ? 0 : 1);
    if (ret.size() != 0) {
      ret.set_begin(&nbr_list_[i]);
    }
    return ret;
  }
 
  ImmutableNbr<std::string_view> get_edge(vid_t i) const {
    ImmutableNbr<std::string_view> nbr;
    nbr.neighbor = nbr_list_[i].neighbor;
    nbr.data = column_.get_view(nbr_list_[i].data);
    return nbr;
  }
  void close() override { nbr_list_.reset(); }
 
 private:
  StringColumn& column_;
  mmap_array<nbr_t> nbr_list_;
};
 
}  // namespace gs
 
#endif  // STORAGES_RT_MUTABLE_GRAPH_CSR_IMMUTABLE_CSR_H_