Merge branch 'main' into ir_case_when

Author: longbinlai

analytical_engine/core/fragment/arrow_projected_fragment.h

@@ -1117,13 +1117,13 @@ class ArrowProjectedFragment
                        grape::PrepareConf conf) {
     if (conf.message_strategy ==
         grape::MessageStrategy::kAlongEdgeToOuterVertex) {
-      initDestFidList(true, true, iodst_, iodoffset_);
+      initDestFidList(comm_spec, true, true, iodst_, iodoffset_);
     } else if (conf.message_strategy ==
                grape::MessageStrategy::kAlongIncomingEdgeToOuterVertex) {
-      initDestFidList(true, false, idst_, idoffset_);
+      initDestFidList(comm_spec, true, false, idst_, idoffset_);
     } else if (conf.message_strategy ==
                grape::MessageStrategy::kAlongOutgoingEdgeToOuterVertex) {
-      initDestFidList(false, true, odst_, odoffset_);
+      initDestFidList(comm_spec, false, true, odst_, odoffset_);
     }
 
     initOuterVertexRanges();
@@ -1140,16 +1140,19 @@ class ArrowProjectedFragment
       ie_spliters_ptr_.clear();
       oe_spliters_ptr_.clear();
       if (directed_) {
-        initEdgeSpliters(ie_, ie_offsets_begin_, ie_offsets_end_, ie_spliters_);
-        initEdgeSpliters(oe_, oe_offsets_begin_, oe_offsets_end_, oe_spliters_);
+        initEdgeSpliters(comm_spec, ie_, ie_offsets_begin_, ie_offsets_end_,
+                         ie_spliters_);
+        initEdgeSpliters(comm_spec, oe_, oe_offsets_begin_, oe_offsets_end_,
+                         oe_spliters_);
         for (auto& vec : ie_spliters_) {
           ie_spliters_ptr_.push_back(vec.data());
         }
         for (auto& vec : oe_spliters_) {
           oe_spliters_ptr_.push_back(vec.data());
         }
       } else {
-        initEdgeSpliters(oe_, oe_offsets_begin_, oe_offsets_end_, oe_spliters_);
+        initEdgeSpliters(comm_spec, oe_, oe_offsets_begin_, oe_offsets_end_,
+                         oe_spliters_);
         for (auto& vec : oe_spliters_) {
           ie_spliters_ptr_.push_back(vec.data());
           oe_spliters_ptr_.push_back(vec.data());
@@ -1690,7 +1693,7 @@ class ArrowProjectedFragment
             ends[i] = range.second;
           }
         },
-        std::thread::hardware_concurrency());
+        std::thread::hardware_concurrency(), 1024);
     return {};
   }
 
@@ -1721,16 +1724,81 @@ class ArrowProjectedFragment
             bends[i] = range.second.second;
           }
         },
-        std::thread::hardware_concurrency());
+        std::thread::hardware_concurrency(), 1024);
     return {};
   }
 
-  void initDestFidList(bool in_edge, bool out_edge,
-                       std::vector<fid_t>& fid_list,
+  void initDestFidList(const grape::CommSpec& comm_spec, const bool in_edge,
+                       const bool out_edge, std::vector<fid_t>& fid_list,
                        std::vector<fid_t*>& fid_list_offset) {
     if (!fid_list_offset.empty()) {
       return;
     }
+    fid_list_offset.resize(ivnum_ + 1, NULL);
+
+    int concurrency =
+        (std::thread::hardware_concurrency() + comm_spec.local_num() - 1) /
+        comm_spec.local_num();
+
+    // don't use std::vector<bool> due to its specialization
+    std::vector<uint8_t> fid_list_bitmap(ivnum_ * fnum_, 0);
+    std::atomic_size_t fid_list_size(0);
+
+    vineyard::parallel_for(
+        static_cast<vid_t>(0), static_cast<vid_t>(ivnum_),
+        [this, in_edge, out_edge, &fid_list_bitmap,
+         &fid_list_size](const vid_t& offset) {
+          vertex_t v = *(inner_vertices_.begin() + offset);
+          if (in_edge) {
+            auto es = GetIncomingAdjList(v);
+            fid_t last_fid = -1;
+            for (auto& e : es) {
+              fid_t f = GetFragId(e.neighbor());
+              if (f != last_fid && f != fid_ &&
+                  !fid_list_bitmap[offset * fnum_ + f]) {
+                last_fid = f;
+                fid_list_bitmap[offset * fnum_ + f] = 1;
+                fid_list_size.fetch_add(1);
+              }
+            }
+          }
+          if (out_edge) {
+            auto es = GetOutgoingAdjList(v);
+            fid_t last_fid = -1;
+            for (auto& e : es) {
+              fid_t f = GetFragId(e.neighbor());
+              if (f != last_fid && f != fid_ &&
+                  !fid_list_bitmap[offset * fnum_ + f]) {
+                last_fid = f;
+                fid_list_bitmap[offset * fnum_ + f] = 1;
+                fid_list_size.fetch_add(1);
+              }
+            }
+          }
+        },
+        concurrency, 1024);
+
+    fid_list.reserve(fid_list_size.load());
+    fid_list_offset[0] = fid_list.data();
+
+    for (vid_t i = 0; i < ivnum_; ++i) {
+      size_t nonzero = 0;
+      for (fid_t fid = 0; fid < fnum_; ++fid) {
+        if (fid_list_bitmap[i * fnum_ + fid]) {
+          nonzero += 1;
+          fid_list.push_back(fid);
+        }
+      }
+      fid_list_offset[i + 1] = fid_list_offset[i] + nonzero;
+    }
+  }
+
+  void initDestFidListSeq(const bool in_edge, const bool out_edge,
+                          std::vector<fid_t>& fid_list,
+                          std::vector<fid_t*>& fid_list_offset) {
+    if (!fid_list_offset.empty()) {
+      return;
+    }
 
     fid_list_offset.resize(ivnum_ + 1, NULL);
 
@@ -1773,6 +1841,7 @@ class ArrowProjectedFragment
   }
 
   void initEdgeSpliters(
+      const grape::CommSpec& comm_spec,
       const std::shared_ptr<arrow::FixedSizeBinaryArray>& edge_list,
       const std::shared_ptr<arrow::Int64Array>& offsets_begin,
       const std::shared_ptr<arrow::Int64Array>& offsets_end,
@@ -1784,28 +1853,38 @@ class ArrowProjectedFragment
     for (auto& vec : spliters) {
       vec.resize(ivnum_);
     }
-    std::vector<int> frag_count;
-    for (vid_t i = 0; i < ivnum_; ++i) {
-      frag_count.clear();
-      frag_count.resize(fnum_, 0);
-      int64_t begin = offsets_begin->Value(i);
-      int64_t end = offsets_end->Value(i);
-      for (int64_t j = begin; j != end; ++j) {
-        const nbr_unit_t* nbr_ptr =
-            reinterpret_cast<const nbr_unit_t*>(edge_list->GetValue(j));
-        vertex_t u(nbr_ptr->vid);
-        fid_t u_fid = GetFragId(u);
-        ++frag_count[u_fid];
-      }
-      begin += frag_count[fid_];
-      frag_count[fid_] = 0;
-      spliters[0][i] = begin;
-      for (fid_t j = 0; j < fnum_; ++j) {
-        begin += frag_count[j];
-        spliters[j + 1][i] = begin;
-      }
-      CHECK_EQ(begin, end);
-    }
+
+    int concurrency =
+        (std::thread::hardware_concurrency() + comm_spec.local_num() - 1) /
+        comm_spec.local_num();
+
+    vineyard::parallel_for(
+        static_cast<vid_t>(0), ivnum_,
+        [this, &offsets_begin, &offsets_end, &edge_list,
+         &spliters](const vid_t i) {
+          std::vector<int> frag_count(fnum_, 0);
+          int64_t begin = offsets_begin->Value(i);
+          int64_t end = offsets_end->Value(i);
+          for (int64_t j = begin; j != end; ++j) {
+            const nbr_unit_t* nbr_ptr =
+                reinterpret_cast<const nbr_unit_t*>(edge_list->GetValue(j));
+            vertex_t u(nbr_ptr->vid);
+            fid_t u_fid = GetFragId(u);
+            ++frag_count[u_fid];
+          }
+          begin += frag_count[fid_];
+          frag_count[fid_] = 0;
+          spliters[0][i] = begin;
+          for (fid_t j = 0; j < fnum_; ++j) {
+            begin += frag_count[j];
+            spliters[j + 1][i] = begin;
+          }
+          if (begin != end) {
+            LOG(ERROR) << "Unexpected edge spliters for ith vertex " << i
+                       << ", begin: " << begin << " vs. end: " << end;
+          }
+        },
+        concurrency, 1024);
   }
 
   void initOuterVertexRanges() {

analytical_engine/core/fragment/dynamic_fragment.h

@@ -418,7 +418,7 @@ class DynamicFragment
     if (conf.need_split_edges_by_fragment) {
       LOG(ERROR) << "MutableEdgecutFragment cannot split edges by fragment";
     } else if (conf.need_split_edges) {
-      splitEdges();
+      splitEdges(comm_spec);
     }
   }
 
@@ -905,30 +905,37 @@ class DynamicFragment
     return id_parser_.max_local_id() - index - 1;
   }
 
-  void splitEdges() {
+  void splitEdges(const grape::CommSpec& comm_spec) {
     auto& inner_vertices = InnerVertices();
     iespliter_.Init(inner_vertices);
     oespliter_.Init(inner_vertices);
-    int inner_neighbor_count = 0;
-    for (auto& v : inner_vertices) {
-      inner_neighbor_count = 0;
-      auto ie = GetIncomingAdjList(v);
-      for (auto& e : ie) {
-        if (IsInnerVertex(e.neighbor)) {
-          ++inner_neighbor_count;
-        }
-      }
-      iespliter_[v] = get_ie_begin(v) + inner_neighbor_count;
 
-      inner_neighbor_count = 0;
-      auto oe = GetOutgoingAdjList(v);
-      for (auto& e : oe) {
-        if (IsInnerVertex(e.neighbor)) {
-          ++inner_neighbor_count;
-        }
-      }
-      oespliter_[v] = get_oe_begin(v) + inner_neighbor_count;
-    }
+    int concurrency =
+        (std::thread::hardware_concurrency() + comm_spec.local_num() - 1) /
+        comm_spec.local_num();
+    vineyard::parallel_for(
+        static_cast<vid_t>(0), static_cast<vid_t>(inner_vertices.size()),
+        [this, &inner_vertices](const vid_t& offset) {
+          vertex_t v = *(inner_vertices.begin() + offset);
+          size_t inner_neighbor_count = 0;
+          auto ie = GetIncomingAdjList(v);
+          for (auto& e : ie) {
+            if (IsInnerVertex(e.neighbor)) {
+              ++inner_neighbor_count;
+            }
+          }
+          iespliter_[v] = get_ie_begin(v) + inner_neighbor_count;
+
+          inner_neighbor_count = 0;
+          auto oe = GetOutgoingAdjList(v);
+          for (auto& e : oe) {
+            if (IsInnerVertex(e.neighbor)) {
+              ++inner_neighbor_count;
+            }
+          }
+          oespliter_[v] = get_oe_begin(v) + inner_neighbor_count;
+        },
+        concurrency, 1024);
   }
 
   vid_t parseOrAddOuterVertexGid(vid_t gid) {

analytical_engine/core/fragment/fragment_reporter.h

@@ -18,8 +18,6 @@
 
 #ifdef NETWORKX
 
-#include <glog/logging.h>
-
 #include <cstddef>
 #include <cstdint>
 #include <memory>
@@ -29,6 +27,7 @@
 
 #include "boost/leaf/error.hpp"
 #include "boost/leaf/result.hpp"
+#include "glog/logging.h"
 #include "grape/communication/communicator.h"
 #include "grape/serialization/in_archive.h"
 #include "grape/worker/comm_spec.h"

analytical_engine/core/server/rpc_utils.h

@@ -131,6 +131,17 @@ class GSParams {
 
   const rpc::LargeAttrValue& GetLargeAttr() const { return large_attr_; }
 
+  const std::string DebugString() const {
+    std::ostringstream ss;
+    ss << "GSParams: {";
+    for (auto const& kv : params_) {
+      ss << rpc::ParamKey_Name(kv.first) << ": " << kv.second.DebugString()
+         << ", ";
+    }
+    ss << "}";
+    return ss.str();
+  }
+
  private:
   const std::map<int, rpc::AttrValue> params_;
   const rpc::LargeAttrValue& large_attr_;

analytical_engine/test/run_vy_app.cc

@@ -35,7 +35,7 @@
 #include "lcc/lcc.h"
 #include "pagerank/pagerank.h"
 #include "pagerank/pagerank_auto.h"
-#include "pagerank/pagerank_local.h"
+#include "pagerank/pagerank_local_parallel.h"
 #include "sssp/sssp.h"
 #include "wcc/wcc.h"
 
@@ -321,7 +321,7 @@ void RunProjectedPR(std::shared_ptr<ProjectedFragmentType> fragment,
                     const std::string& out_prefix) {
   // using AppType = grape::PageRankAuto<ProjectedFragmentType>;
   using AppType = grape::PageRank<ProjectedFragmentType>;
-  // using AppType = grape::PageRankLocal<ProjectedFragmentType>;
+  // using AppType = grape::PageRankLocalParallel<ProjectedFragmentType>;
   auto app = std::make_shared<AppType>();
   auto worker = AppType::CreateWorker(app, fragment);
   auto spec = grape::DefaultParallelEngineSpec();

coordinator/gscoordinator/builtin/app/.gs_conf.yaml

@@ -1,8 +1,24 @@
 app:
   - algo: pagerank
     type: cpp_pie
-    class_name: grape::PageRankLocal
-    src: pagerank/pagerank_local.h
+    class_name: grape::PageRankLocalParallel
+    src: pagerank/pagerank_local_parallel.h
+    compatible_graph:
+      - grape::ImmutableEdgecutFragment
+      - gs::ArrowProjectedFragment
+      - gs::DynamicProjectedFragment
+  - algo: pagerank_directed
+    type: cpp_pie
+    class_name: grape::PageRankDirected
+    src: pagerank/pagerank_directed.h
+    compatible_graph:
+      - grape::ImmutableEdgecutFragment
+      - gs::ArrowProjectedFragment
+      - gs::DynamicProjectedFragment
+  - algo: pagerank_opt
+    type: cpp_pie
+    class_name: grape::PageRankOpt
+    src: pagerank/pagerank_opt.h
     compatible_graph:
       - grape::ImmutableEdgecutFragment
       - gs::ArrowProjectedFragment
@@ -53,6 +69,12 @@ app:
       - grape::ImmutableEdgecutFragment
       - gs::ArrowProjectedFragment
       - gs::DynamicProjectedFragment
+  - algo: lcc
+    type: cpp_pie
+    class_name: grape::LCC
+    src: lcc/lcc.h
+    compatible_graph:
+      - gs::DynamicFragment
   - algo: sssp_path
     type: cpp_pie
     class_name: gs::SSSPPath
@@ -118,12 +140,6 @@ app:
     src: apps/kshell/kshell.h
     compatible_graph:
       - gs::DynamicFragment
-  - algo: lcc
-    type: cpp_pie
-    class_name: grape::LCC
-    src: lcc/lcc.h
-    compatible_graph:
-      - gs::DynamicFragment
   - algo: clustering
     type: cpp_pie
     class_name: gs::Clustering

interactive_engine/executor/assembly/v6d/src/bin/gaia_executor.rs

@@ -169,7 +169,9 @@ fn sync_global_process_partition_lists(
         if servers.len() > 1 {
             assert_eq!(partitions.len() % servers.len(), 0);
             let nchunk = partitions.len() / servers.len();
-            for (index, server) in servers.iter().enumerate() {
+            let mut sorted_servers = servers.clone();
+            sorted_servers.sort();
+            for (index, server) in sorted_servers.iter().enumerate() {
                 partition_lists.insert(*server, partitions[index * nchunk..(index + 1) * nchunk].to_vec());
             }
         }