fix(interactive): close csr and free memory after dumped (#3936)

alibaba · Jun 18, 2024 · ce17b4c · ce17b4c
1 parent f1cf69d
commit ce17b4c
Show file tree

Hide file tree

Showing 5 changed files with 53 additions and 1 deletion.
diff --git a/flex/storages/rt_mutable_graph/csr/csr_base.h b/flex/storages/rt_mutable_graph/csr/csr_base.h
@@ -89,6 +89,8 @@ class CsrBase {
   virtual void resize(vid_t vnum) = 0;
   virtual size_t size() const = 0;
 
+  virtual void close() = 0;
+
   virtual std::shared_ptr<CsrConstEdgeIterBase> edge_iter(vid_t v) const = 0;
   virtual CsrConstEdgeIterBase* edge_iter_raw(vid_t v) const = 0;
   virtual std::shared_ptr<CsrEdgeIterBase> edge_iter_mut(vid_t v) = 0;

diff --git a/flex/storages/rt_mutable_graph/csr/immutable_csr.h b/flex/storages/rt_mutable_graph/csr/immutable_csr.h
@@ -247,6 +247,12 @@ class ImmutableCsr : public TypedImmutableCsrBase<EDATA_T> {
     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";
@@ -425,6 +431,8 @@ class SingleImmutableCsr : public TypedImmutableCsrBase<EDATA_T> {
     return ret;
   }
 
+  void close() override { nbr_list_.reset(); }
+
   const nbr_t& get_edge(vid_t i) const { return nbr_list_[i]; }
 
  private:
@@ -591,11 +599,13 @@ class SingleImmutableCsr<std::string_view>
     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_
+#endif  // STORAGES_RT_MUTABLE_GRAPH_CSR_IMMUTABLE_CSR_H_
diff --git a/flex/storages/rt_mutable_graph/csr/mutable_csr.h b/flex/storages/rt_mutable_graph/csr/mutable_csr.h
@@ -421,6 +421,15 @@ class MutableCsr : public TypedMutableCsrBase<EDATA_T> {
 
   mut_slice_t get_edges_mut(vid_t i) { return adj_lists_[i].get_edges_mut(); }
 
+  void close() override {
+    if (locks_ != nullptr) {
+      delete[] locks_;
+      locks_ = nullptr;
+    }
+    adj_lists_.reset();
+    nbr_list_.reset();
+  }
+
  private:
   void load_meta(const std::string& prefix) {
     std::string meta_file_path = prefix + ".meta";
@@ -651,6 +660,14 @@ class MutableCsr<std::string_view>
     return adj_lists_[i].get_edges_mut(column_);
   }
 
+  void close() override {
+    if (locks_ != nullptr) {
+      delete[] locks_;
+    }
+    adj_lists_.reset();
+    nbr_list_.reset();
+  }
+
  private:
   StringColumn& column_;
   grape::SpinLock* locks_;
@@ -827,6 +844,8 @@ class SingleMutableCsr : public TypedMutableCsrBase<EDATA_T> {
 
   const nbr_t& get_edge(vid_t i) const { return nbr_list_[i]; }
 
+  void close() override { nbr_list_.reset(); }
+
  private:
   mmap_array<nbr_t> nbr_list_;
 };
@@ -993,6 +1012,8 @@ class SingleMutableCsr<std::string_view>
     return nbr;
   }
 
+  void close() override { nbr_list_.reset(); }
+
  private:
   StringColumn& column_;
   mmap_array<nbr_t> nbr_list_;
@@ -1053,6 +1074,8 @@ class EmptyCsr : public TypedMutableCsrBase<EDATA_T> {
   }
 
   slice_t get_edges(vid_t v) const override { return slice_t::empty(); }
+
+  void close() override {}
 };
 
 template <>
@@ -1103,6 +1126,8 @@ class EmptyCsr<std::string_view>
 
   slice_t get_edges(vid_t v) const override { return slice_t::empty(column_); }
 
+  void close() override {}
+
   StringColumn& column_;
 };
 

diff --git a/flex/storages/rt_mutable_graph/dual_csr.h b/flex/storages/rt_mutable_graph/dual_csr.h
@@ -75,6 +75,7 @@ class DualCsrBase {
   virtual CsrBase* GetOutCsr() = 0;
   virtual const CsrBase* GetInCsr() const = 0;
   virtual const CsrBase* GetOutCsr() const = 0;
+  virtual void Close() = 0;
 };
 
 template <typename EDATA_T>
@@ -150,6 +151,7 @@ class DualCsr : public DualCsrBase {
             const std::string& new_snapshot_dir) override {
     in_csr_->dump(ie_name, new_snapshot_dir);
     out_csr_->dump(oe_name, new_snapshot_dir);
+    Close();
   }
 
   CsrBase* GetInCsr() override { return in_csr_; }
@@ -204,6 +206,11 @@ class DualCsr : public DualCsrBase {
     out_csr_->batch_put_edge(src, dst, data);
   }
 
+  void Close() override {
+    in_csr_->close();
+    out_csr_->close();
+  }
+
  private:
   TypedCsrBase<EDATA_T>* in_csr_;
   TypedCsrBase<EDATA_T>* out_csr_;
@@ -285,6 +292,7 @@ class DualCsr<std::string_view> : public DualCsrBase {
     out_csr_->dump(oe_name, new_snapshot_dir);
     column_.resize(column_idx_.load());
     column_.dump(new_snapshot_dir + "/" + edata_name);
+    Close();
   }
 
   CsrBase* GetInCsr() override { return in_csr_; }
@@ -355,6 +363,12 @@ class DualCsr<std::string_view> : public DualCsrBase {
     out_csr_->batch_put_edge_with_index(src, dst, row_id);
   }
 
+  void Close() override {
+    in_csr_->close();
+    out_csr_->close();
+    column_.close();
+  }
+
  private:
   TypedCsrBase<std::string_view>* in_csr_;
   TypedCsrBase<std::string_view>* out_csr_;

diff --git a/flex/utils/id_indexer.h b/flex/utils/id_indexer.h
@@ -422,6 +422,7 @@ class LFIndexer {
     keys_->dump(snapshot_dir + "/" + name + ".keys");
     indices_.dump(snapshot_dir + "/" + name + ".indices");
     dump_meta(snapshot_dir + "/" + name + ".meta");
+    close();
   }
 
   void close() {