Revise apps and add a bunch of optimized version of ldbc apps (#3034)

alibaba · Jul 20, 2023 · 95b7130 · 95b7130
1 parent 8e4131d
commit 95b7130
Show file tree

Hide file tree

Showing 13 changed files with 416 additions and 542 deletions.
diff --git a/analytical_engine/core/grape_instance.cc b/analytical_engine/core/grape_instance.cc
@@ -322,6 +322,8 @@ bl::result<std::string> GrapeInstance::query(const rpc::GSParams& params,
     context_type = ctx_wrapper->context_type();
     context_schema = ctx_wrapper->schema();
     BOOST_LEAF_CHECK(object_manager_.PutObject(ctx_wrapper));
+  } else {
+    LOG(ERROR) << "Error occur when querying";
   }
   return toJson({{"context_type", context_type},
                  {"context_key", context_key},

diff --git a/analytical_engine/test/app_tests.sh b/analytical_engine/test/app_tests.sh
@@ -89,20 +89,17 @@ function get_test_data() {
 # Arguments:
 #   None
 # !!!WARNING!!!:
-#   Kill all started vineyardd and etcd
+#   Kill all started vineyardd
 ########################################################
 function start_vineyard() {
   pushd "${ENGINE_HOME}/build"
   pkill vineyardd || true
-  pkill etcd || true
   echo "[INFO] vineyardd will using the socket_file on ${socket_file}"
 
   timestamp=$(date +%Y-%m-%d_%H-%M-%S)
   vineyardd \
-    --socket ${socket_file} \
-    --size 2000000000 \
-    --etcd_prefix "${timestamp}" \
-    --etcd_endpoint=http://127.0.0.1:3457 &
+    -socket ${socket_file} \
+    -meta local &
   set +m
   sleep 5
   info "vineyardd started."
@@ -359,8 +356,15 @@ function run_local_vertex_map() {
 #
 # sssp_average_length is a time-consuming app, so we skip it for graph p2p.
 
-declare -a apps=(
+declare -a ldbc_apps=(
+  # "wcc"
   "sssp"
+  "lcc"
+  "bfs"
+  "cdlp"
+  # "pr"  # need eps match
+)
+declare -a other_apps=(
   "sssp_has_path"
   # "sssp_path"
   "cdlp_auto"
@@ -371,7 +375,6 @@ declare -a apps=(
   # "pagerank_auto"
   "kcore"
   "hits"
-  # "bfs"
   "avg_clustering"
   "transitivity"
   "triangles"
@@ -396,7 +399,12 @@ pushd "${ENGINE_HOME}"/build
 
 get_test_data
 
-for app in "${apps[@]}"; do
+for app in "${ldbc_apps[@]}"; do
+  run ${np} ./run_app --vfile "${test_dir}"/p2p-31.v --efile "${test_dir}"/p2p-31.e --application "${app}" --out_prefix ./test_output --sssp_source=6 --sssp_target=10 --bfs_source=6
+  exact_verify "${test_dir}"/property/ldbc/p2p-31-"${app^^}"
+done
+
+for app in "${other_apps[@]}"; do
   run ${np} ./run_app --vfile "${test_dir}"/p2p-31.v --efile "${test_dir}"/p2p-31.e --application "${app}" --out_prefix ./test_output --sssp_source=6 --sssp_target=10 --bfs_source=6
   exact_verify "${test_dir}"/p2p-31-"${app}"
 done

diff --git a/analytical_engine/test/run_app.h b/analytical_engine/test/run_app.h
@@ -36,19 +36,22 @@ limitations under the License.
 #include "grape/fragment/loader.h"
 #include "grape/grape.h"
 
-#include "bfs/bfs.h"
 #include "bfs/bfs_auto.h"
+#include "bfs/bfs_opt.h"
 #include "cdlp/cdlp.h"
 #include "cdlp/cdlp_auto.h"
-#include "lcc/lcc.h"
+#include "cdlp/cdlp_opt.h"
 #include "lcc/lcc_auto.h"
-#include "pagerank/pagerank.h"
+#include "lcc/lcc_opt.h"
 #include "pagerank/pagerank_auto.h"
-#include "sssp/sssp.h"
+#include "pagerank/pagerank_directed.h"
+#include "pagerank/pagerank_opt.h"
 #include "sssp/sssp_auto.h"
+#include "sssp/sssp_opt.h"
 #include "voterank/voterank.h"
 #include "wcc/wcc.h"
 #include "wcc/wcc_auto.h"
+#include "wcc/wcc_opt.h"
 
 #include "apps/bfs/bfs_generic.h"
 #include "apps/centrality/degree/degree_centrality.h"
@@ -203,13 +206,14 @@ void Run() {
   }
   int fnum = comm_spec.fnum();
   std::string name = FLAGS_application;
+
   using VertexMapType = grape::GlobalVertexMap<OID_T, VID_T, PARTITIONER_T>;
   if (name == "sssp") {
     using GraphType =
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, double,
                                         grape::LoadStrategy::kOnlyOut,
                                         VertexMapType>;
-    using AppType = grape::SSSP<GraphType>;
+    using AppType = grape::SSSPOpt<GraphType>;
     CreateAndQuery<GraphType, AppType, OID_T>(comm_spec, efile, vfile,
                                               out_prefix, FLAGS_datasource,
                                               fnum, spec, FLAGS_sssp_source);
@@ -252,6 +256,8 @@ void Run() {
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
                                         grape::LoadStrategy::kBothOutIn,
                                         VertexMapType>;
+    // TODO(siyuan): uncomment once latest libgrape-lite is released.
+    // using AppType = grape::CDLPOpt<GraphType, int64_t>;
     using AppType = grape::CDLP<GraphType>;
     CreateAndQuery<GraphType, AppType, int>(comm_spec, efile, vfile, out_prefix,
                                             FLAGS_datasource, fnum, spec, 10);
@@ -277,6 +283,7 @@ void Run() {
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
                                         grape::LoadStrategy::kBothOutIn,
                                         VertexMapType>;
+    // using AppType = grape::WCCOpt<GraphType>;
     using AppType = grape::WCC<GraphType>;
     CreateAndQuery<GraphType, AppType>(comm_spec, efile, vfile, out_prefix,
                                        FLAGS_datasource, fnum, spec);
@@ -293,7 +300,7 @@ void Run() {
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
                                         grape::LoadStrategy::kBothOutIn,
                                         VertexMapType>;
-    using AppType = grape::LCC<GraphType>;
+    using AppType = grape::LCCOpt<GraphType>;
     CreateAndQuery<GraphType, AppType>(comm_spec, efile, vfile, out_prefix,
                                        FLAGS_datasource, fnum, spec);
   } else if (name == "bfs_auto") {
@@ -310,7 +317,7 @@ void Run() {
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
                                         grape::LoadStrategy::kBothOutIn,
                                         VertexMapType>;
-    using AppType = grape::BFS<GraphType>;
+    using AppType = grape::BFSOpt<GraphType>;
     CreateAndQuery<GraphType, AppType, OID_T>(comm_spec, efile, vfile,
                                               out_prefix, FLAGS_datasource,
                                               fnum, spec, FLAGS_bfs_source);
@@ -323,15 +330,23 @@ void Run() {
     CreateAndQuery<GraphType, AppType, double, int>(
         comm_spec, efile, vfile, out_prefix, FLAGS_datasource, fnum, spec, 0.85,
         10);
-  } else if (name == "pagerank") {
+  } else if (name == "pr" || name == "pagerank") {
     using GraphType =
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
                                         grape::LoadStrategy::kBothOutIn,
                                         VertexMapType>;
-    using AppType = grape::PageRank<GraphType>;
-    CreateAndQuery<GraphType, AppType, double, int>(
-        comm_spec, efile, vfile, out_prefix, FLAGS_datasource, fnum, spec, 0.85,
-        10);
+    if (FLAGS_directed) {
+      using AppType = grape::PageRankDirected<GraphType>;
+
+      CreateAndQuery<GraphType, AppType, double, int>(
+          comm_spec, efile, vfile, out_prefix, FLAGS_datasource, fnum, spec,
+          0.85, 10);
+    } else {
+      using AppType = grape::PageRankOpt<GraphType>;
+      CreateAndQuery<GraphType, AppType, double, int>(
+          comm_spec, efile, vfile, out_prefix, FLAGS_datasource, fnum, spec,
+          0.85, 10);
+    }
   } else if (name == "kcore") {
     using GraphType =
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
@@ -382,7 +397,7 @@ void Run() {
                                          FLAGS_eigenvector_centrality_tolerance,
                                          FLAGS_eigenvector_centrality_max_round);
     */
-  } else if (name == "bfs") {
+  } else if (name == "bfs_generic") {
     using GraphType =
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
                                         grape::LoadStrategy::kBothOutIn,
@@ -441,12 +456,12 @@ void Run() {
     CreateAndQuery<GraphType, AppType>(comm_spec, efile, vfile, out_prefix,
                                        FLAGS_datasource, fnum, spec,
                                        FLAGS_dfs_source, FLAGS_dfs_format);
-  } else if (name == "bfs_original") {
+  } else if (name == "bfs") {
     using GraphType =
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
                                         grape::LoadStrategy::kBothOutIn,
                                         VertexMapType>;
-    using AppType = grape::BFS<GraphType>;
+    using AppType = grape::BFSOpt<GraphType>;
     CreateAndQuery<GraphType, AppType>(comm_spec, efile, vfile, out_prefix,
                                        FLAGS_datasource, fnum, spec,
                                        FLAGS_bfs_source);

diff --git a/analytical_engine/test/run_vy_app.cc b/analytical_engine/test/run_vy_app.cc
@@ -30,14 +30,15 @@
 #include "apps/property/sssp_property.h"
 #include "apps/property/wcc_property.h"
 #include "apps/sampling_path/sampling_path.h"
-#include "bfs/bfs.h"
+#include "bfs/bfs_opt.h"
 #include "cdlp/cdlp.h"
-#include "lcc/lcc.h"
-#include "pagerank/pagerank.h"
+#include "cdlp/cdlp_opt.h"
+#include "lcc/lcc_opt.h"
 #include "pagerank/pagerank_auto.h"
 #include "pagerank/pagerank_local_parallel.h"
-#include "sssp/sssp.h"
-#include "wcc/wcc.h"
+#include "pagerank/pagerank_opt.h"
+#include "sssp/sssp_opt.h"
+#include "wcc/wcc_opt.h"
 
 #include "core/fragment/arrow_projected_fragment.h"
 #include "core/loader/arrow_fragment_loader.h"
@@ -55,8 +56,9 @@ using ProjectedFragmentType =
 using ProjectedFragmentType2 =
     gs::ArrowProjectedFragment<oid_t, vid_t, grape::EmptyType, int64_t>;
 
-void RunWCC(std::shared_ptr<FragmentType> fragment,
-            const grape::CommSpec& comm_spec, const std::string& out_prefix) {
+void RunWCCProperty(std::shared_ptr<FragmentType> fragment,
+                    const grape::CommSpec& comm_spec,
+                    const std::string& out_prefix) {
   using AppType = gs::WCCProperty<FragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
@@ -76,8 +78,9 @@ void RunWCC(std::shared_ptr<FragmentType> fragment,
   worker->Finalize();
 }
 
-void RunSSSP(std::shared_ptr<FragmentType> fragment,
-             const grape::CommSpec& comm_spec, const std::string& out_prefix) {
+void RunSSSPProperty(std::shared_ptr<FragmentType> fragment,
+                     const grape::CommSpec& comm_spec,
+                     const std::string& out_prefix) {
   using AppType = gs::SSSPProperty<FragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
@@ -155,9 +158,9 @@ void RunSamplingPath(std::shared_ptr<FragmentType> fragment,
   worker->Finalize();
 }
 
-void RunAutoWCC(std::shared_ptr<FragmentType> fragment,
-                const grape::CommSpec& comm_spec,
-                const std::string& out_prefix) {
+void RunAutoWCCProperty(std::shared_ptr<FragmentType> fragment,
+                        const grape::CommSpec& comm_spec,
+                        const std::string& out_prefix) {
   using AppType = gs::AutoWCCProperty<FragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
@@ -177,9 +180,9 @@ void RunAutoWCC(std::shared_ptr<FragmentType> fragment,
   worker->Finalize();
 }
 
-void RunAutoSSSP(std::shared_ptr<FragmentType> fragment,
-                 const grape::CommSpec& comm_spec,
-                 const std::string& out_prefix) {
+void RunAutoSSSPProperty(std::shared_ptr<FragmentType> fragment,
+                         const grape::CommSpec& comm_spec,
+                         const std::string& out_prefix) {
   using AppType = gs::AutoSSSPProperty<FragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
@@ -202,9 +205,7 @@ void RunAutoSSSP(std::shared_ptr<FragmentType> fragment,
 void RunProjectedWCC(std::shared_ptr<ProjectedFragmentType> fragment,
                      const grape::CommSpec& comm_spec,
                      const std::string& out_prefix) {
-  // using AppType = grape::WCCProjected<ProjectedFragmentType>;
-  // using AppType = grape::WCCAuto<ProjectedFragmentType>;
-  using AppType = grape::WCC<ProjectedFragmentType>;
+  using AppType = grape::WCCOpt<ProjectedFragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
   auto spec = grape::DefaultParallelEngineSpec();
@@ -228,7 +229,7 @@ void RunProjectedSSSP(std::shared_ptr<ProjectedFragmentType2> fragment,
                       const std::string& out_prefix) {
   // using AppType = grape::SSSPProjected<ProjectedFragmentType2>;
   // using AppType = grape::SSSPAuto<ProjectedFragmentType2>;
-  using AppType = grape::SSSP<ProjectedFragmentType2>;
+  using AppType = grape::SSSPOpt<ProjectedFragmentType2>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
   auto spec = grape::DefaultParallelEngineSpec();
@@ -250,7 +251,8 @@ void RunProjectedSSSP(std::shared_ptr<ProjectedFragmentType2> fragment,
 void RunProjectedCDLP(std::shared_ptr<ProjectedFragmentType> fragment,
                       const grape::CommSpec& comm_spec,
                       const std::string& out_prefix) {
-  // using AppType = grape::CDLPAuto<ProjectedFragmentType>;
+  // TODO(siyuan): uncomment once latest libgrape-lite is released.
+  // using AppType = grape::CDLPOpt<GraphType, int64_t>;
   using AppType = grape::CDLP<ProjectedFragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
@@ -274,7 +276,7 @@ void RunProjectedBFS(std::shared_ptr<ProjectedFragmentType> fragment,
                      const grape::CommSpec& comm_spec,
                      const std::string& out_prefix) {
   // using AppType = grape::BFSAuto<ProjectedFragmentType>;
-  using AppType = grape::BFS<ProjectedFragmentType>;
+  using AppType = grape::BFSOpt<ProjectedFragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
   auto spec = grape::DefaultParallelEngineSpec();
@@ -297,7 +299,7 @@ void RunProjectedLCC(std::shared_ptr<ProjectedFragmentType> fragment,
                      const grape::CommSpec& comm_spec,
                      const std::string& out_prefix) {
   // using AppType = grape::LCCAuto<ProjectedFragmentType>;
-  using AppType = grape::LCC<ProjectedFragmentType>;
+  using AppType = grape::LCCOpt<ProjectedFragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
   auto spec = grape::DefaultParallelEngineSpec();
@@ -320,7 +322,7 @@ void RunProjectedPR(std::shared_ptr<ProjectedFragmentType> fragment,
                     const grape::CommSpec& comm_spec,
                     const std::string& out_prefix) {
   // using AppType = grape::PageRankAuto<ProjectedFragmentType>;
-  using AppType = grape::PageRank<ProjectedFragmentType>;
+  using AppType = grape::PageRankOpt<ProjectedFragmentType>;
   // using AppType = grape::PageRankLocalParallel<ProjectedFragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
@@ -353,11 +355,11 @@ void Run(vineyard::Client& client, const grape::CommSpec& comm_spec,
                     path_pattern);
   } else {
     if (!run_projected) {
-      RunWCC(fragment, comm_spec, "./outputs_wcc/");
-      RunSSSP(fragment, comm_spec, "./outputs_sssp/");
+      RunWCCProperty(fragment, comm_spec, "./outputs_wcc/");
+      RunSSSPProperty(fragment, comm_spec, "./outputs_sssp/");
 
-      RunAutoWCC(fragment, comm_spec, "./outputs_auto_wcc/");
-      RunAutoSSSP(fragment, comm_spec, "./outputs_auto_sssp/");
+      RunAutoWCCProperty(fragment, comm_spec, "./outputs_auto_wcc/");
+      RunAutoSSSPProperty(fragment, comm_spec, "./outputs_auto_sssp/");
     } else {
       {
         // v_prop is grape::EmptyType, e_prop is grape::EmptyType