add app kshell (#66)

* add app kshell

* providing kshell algorithm in python interface

* fix clang-format error

* python format

* update k_shell.py

* trigger GitHub actions

analytical_engine/apps/kshell/kshell.h

@@ -0,0 +1,138 @@
+/** Copyright 2020 Alibaba Group Holding Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * 	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef ANALYTICAL_ENGINE_APPS_KSHELL_KSHELL_H_
+#define ANALYTICAL_ENGINE_APPS_KSHELL_KSHELL_H_
+
+#include <functional>
+#include <memory>
+#include <vector>
+
+#include "kshell/kshell_context.h"
+
+namespace gs {
+/**
+ * @brief Get a subgraph induced by nodes with core number k.
+ * That is, nodes in the k-core that are not in the (k+1)-core.
+ * @tparam FRAG_T
+ */
+template <typename FRAG_T>
+class KShell : public grape::ParallelAppBase<FRAG_T, KShellContext<FRAG_T>>,
+               public grape::ParallelEngine,
+               public grape::Communicator {
+ public:
+  INSTALL_PARALLEL_WORKER(KShell<FRAG_T>, KShellContext<FRAG_T>, FRAG_T)
+  static constexpr grape::MessageStrategy message_strategy =
+      grape::MessageStrategy::kSyncOnOuterVertex;
+  static constexpr grape::LoadStrategy load_strategy =
+      grape::LoadStrategy::kBothOutIn;
+  using vertex_t = typename fragment_t::vertex_t;
+  using vid_t = typename FRAG_T::vid_t;
+
+  void UpdateDegree(const fragment_t& frag,
+                    const grape::DenseVertexSet<vid_t>& frontier,
+                    typename FRAG_T::template vertex_array_t<
+                        std::shared_ptr<std::atomic_int>>& degrees) {
+    ForEach(frontier, [&degrees, &frag](int tid, vertex_t u) {
+      for (auto& e : frag.GetOutgoingAdjList(u)) {
+        auto v = e.get_neighbor();
+        degrees[v]->operator--();
+      }
+
+      degrees[u]->store(0);
+    });
+  }
+
+  void PEval(const fragment_t& frag, context_t& ctx,
+             message_manager_t& messages) {
+    messages.InitChannels(thread_num());
+    // we put all computing logic in IncEval
+    messages.ForceContinue();
+  }
+
+  void IncEval(const fragment_t& frag, context_t& ctx,
+               message_manager_t& messages) {
+    int thrd_num = thread_num();
+    auto outer_vertices = frag.OuterVertices();
+    auto& remaining_vertices = ctx.remaining_vertices;
+    auto& next_remaining_vertices = ctx.next_remaining_vertices;
+    auto& to_remove_vertices_k = ctx.to_remove_vertices_k;
+    auto& to_remove_vertices_inc = ctx.to_remove_vertices_inc;
+    auto& curr_k = ctx.curr_k;
+    auto& degrees = ctx.degrees;
+
+    messages.ParallelProcess<fragment_t, int32_t>(
+        thrd_num, frag,
+        [&degrees](int tid, vertex_t v, int32_t msg) { *degrees[v] += msg; });
+
+    // remove vertices which degree less or equal than curr_k
+    ForEach(remaining_vertices, [&to_remove_vertices_k, &to_remove_vertices_inc,
+                                 &degrees, curr_k](int tid, vertex_t v) {
+      if (degrees[v]->load() <= curr_k) {
+        to_remove_vertices_k.Insert(v);
+        to_remove_vertices_inc.Insert(v);
+      }
+    });
+
+    // keep vertices which degree greater than currk
+    ForEach(remaining_vertices,
+            [&next_remaining_vertices, &degrees, curr_k](int tid, vertex_t v) {
+              if (degrees[v]->load() > curr_k) {
+                next_remaining_vertices.Insert(v);
+              }
+            });
+
+    UpdateDegree(frag, to_remove_vertices_inc, degrees);
+
+    ForEach(outer_vertices, [&frag, &degrees, &messages](int tid, vertex_t v) {
+      int degree = degrees[v]->load();
+      if (degree != 0) {
+        messages.Channels()[tid].SyncStateOnOuterVertex<fragment_t, int32_t>(
+            frag, v, degree);
+        degrees[v]->store(0);
+      }
+    });
+
+    bool curr_k_changed = false;
+    size_t global_removed_inc_count = 0;
+    Sum(to_remove_vertices_inc.Count(), global_removed_inc_count);
+    if (global_removed_inc_count == 0) {
+      curr_k++;
+      curr_k_changed = true;
+    }
+
+    to_remove_vertices_inc.Clear();
+    remaining_vertices.Clear();
+    remaining_vertices.Swap(next_remaining_vertices);
+
+    if (curr_k > ctx.k) {
+      auto inner_vertices = frag.InnerVertices();
+
+      for (auto v : inner_vertices) {
+        ctx.data()[v] = to_remove_vertices_k.Exist(v) ? 1 : 0;
+      }
+      return;
+    }
+
+    if (curr_k_changed) {
+      to_remove_vertices_k.Clear();
+    }
+
+    messages.ForceContinue();
+  }
+};
+};  // namespace gs
+
+#endif  // ANALYTICAL_ENGINE_APPS_KSHELL_KSHELL_H_

analytical_engine/apps/kshell/kshell_context.h

@@ -0,0 +1,84 @@
+/** Copyright 2020 Alibaba Group Holding Limited.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+#ifndef ANALYTICAL_ENGINE_APPS_KSHELL_KSHELL_CONTEXT_H_
+#define ANALYTICAL_ENGINE_APPS_KSHELL_KSHELL_CONTEXT_H_
+
+#include <limits>
+#include <map>
+#include <memory>
+#include <queue>
+#include <unordered_set>
+#include <utility>
+
+#include "grape/grape.h"
+
+namespace gs {
+
+template <typename FRAG_T>
+class KShellContext
+    : public grape::VertexDataContext<FRAG_T, typename FRAG_T::oid_t> {
+ public:
+  using oid_t = typename FRAG_T::oid_t;
+  using vid_t = typename FRAG_T::vid_t;
+
+  explicit KShellContext(const FRAG_T& fragment)
+      : grape::VertexDataContext<FRAG_T, typename FRAG_T::oid_t>(fragment) {}
+
+  typename FRAG_T::template vertex_array_t<std::shared_ptr<std::atomic_int>>
+      degrees;
+  grape::DenseVertexSet<vid_t> to_remove_vertices_k, to_remove_vertices_inc,
+      remaining_vertices, next_remaining_vertices;
+  int k;
+  int curr_k;
+
+  void Init(grape::ParallelMessageManager& messages, int k) {
+    auto& frag = this->fragment();
+    auto vertices = frag.Vertices();
+    auto inner_vertices = frag.InnerVertices();
+
+    degrees.Init(vertices);
+    to_remove_vertices_k.Init(inner_vertices);
+    to_remove_vertices_inc.Init(inner_vertices);
+    remaining_vertices.Init(inner_vertices);
+    next_remaining_vertices.Init(inner_vertices);
+    this->k = k;
+    curr_k = 0;
+
+    for (auto& v : vertices) {
+      degrees[v] = std::make_shared<std::atomic_int>(0);
+      if (frag.IsInnerVertex(v)) {
+        remaining_vertices.Insert(v);
+        degrees[v]->store(frag.GetLocalOutDegree(v));
+      } else {
+        degrees[v]->store(0);
+      }
+    }
+  }
+
+  void Output(std::ostream& os) override {
+    auto& frag = this->fragment();
+    auto inner_vertices = frag.InnerVertices();
+
+    for (auto& v : inner_vertices) {
+      if (to_remove_vertices_k.Exist(v)) {
+        os << frag.GetId(v) << '\n';
+      }
+    }
+  }
+};
+}  // namespace gs
+
+#endif  // ANALYTICAL_ENGINE_APPS_KSHELL_KSHELL_CONTEXT_H_

analytical_engine/test/run_app.cc

@@ -44,6 +44,8 @@ DEFINE_bool(hits_normalized, true,
 
 DEFINE_int32(kcore_k, 3, "The order of the core");
 
+DEFINE_int32(kshell_k, 3, "The order of the shell");
+
 DEFINE_double(katz_centrality_alpha, 0.1, "Attenuation factor");
 DEFINE_double(katz_centrality_beta, 1.0,
               "Weight attributed to the immediate neighborhood.");

analytical_engine/test/run_app.h

@@ -60,6 +60,7 @@ limitations under the License.
 #include "apps/dfs/dfs.h"
 #include "apps/hits/hits.h"
 #include "apps/kcore/kcore.h"
+#include "apps/kshell/kshell.h"
 #include "apps/sssp/sssp_average_length.h"
 #include "apps/sssp/sssp_has_path.h"
 #include "apps/sssp/sssp_path.h"
@@ -86,6 +87,8 @@ DECLARE_bool(hits_normalized);
 
 DECLARE_int32(kcore_k);
 
+DECLARE_int32(kshell_k);
+
 DECLARE_double(katz_centrality_alpha);
 DECLARE_double(katz_centrality_beta);
 DECLARE_double(katz_centrality_tolerance);
@@ -324,6 +327,14 @@ void Run() {
     CreateAndQuery<GraphType, AppType>(comm_spec, efile, vfile, out_prefix,
                                        FLAGS_datasource, fnum, spec,
                                        FLAGS_kcore_k);
+  } else if (name == "kshell") {
+    using GraphType =
+        grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,
+                                        grape::LoadStrategy::kBothOutIn>;
+    using AppType = KShell<GraphType>;
+    CreateAndQuery<GraphType, AppType>(comm_spec, efile, vfile, out_prefix,
+                                       FLAGS_datasource, fnum, spec,
+                                       FLAGS_kshell_k);
   } else if (name == "hits") {
     using GraphType =
         grape::ImmutableEdgecutFragment<OID_T, VID_T, VDATA_T, EDATA_T,

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

@@ -104,6 +104,12 @@ app:
     src: apps/kcore/kcore.h
     compatible_graph:
       - gs::DynamicFragment
+  - algo: kshell
+    type: cpp_pie
+    class_name: gs::KShell
+    src: apps/kshell/kshell.h
+    compatible_graph:
+      - gs::DynamicFragment
   - algo: clustering
     type: cpp_pie
     class_name: gs::Clustering

python/graphscope/analytical/app/__init__.py

@@ -25,6 +25,7 @@
 from graphscope.analytical.app.eigenvector_centrality import eigenvector_centrality
 from graphscope.analytical.app.hits import hits
 from graphscope.analytical.app.k_core import k_core
+from graphscope.analytical.app.k_shell import k_shell
 from graphscope.analytical.app.katz_centrality import katz_centrality
 from graphscope.analytical.app.lpa import lpa
 from graphscope.analytical.app.pagerank import pagerank

python/graphscope/analytical/app/k_shell.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+#
+# Copyright 2020 Alibaba Group Holding Limited. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+
+from graphscope.framework.app import AppAssets
+from graphscope.framework.app import not_compatible_for
+
+__all__ = ["k_shell"]
+
+
+@not_compatible_for("arrow_property", "dynamic_property")
+def k_shell(graph, k: int):
+    """The k-shell is the subgraph induced by nodes with core number k.
+    That is, nodes in the k-core that are not in the (k+1)-core.
+
+    Args:
+        graph (:class:`Graph`): A projected simple graph.
+        k (int): The `k` for k-shell.
+
+    Returns:
+        :class:`VertexDataContext`: A context with each vertex assigned with a boolean:
+        1 if the vertex satisfies k-shell, otherwise 0.
+
+    Examples:
+
+    .. code:: python
+
+        import graphscope as gs
+        s = gs.session()
+        g = s.load_from('The parameters for loading a graph...')
+        pg = g.project_to_simple(v_label='vlabel', e_label='elabel')
+        r = gs.k_shell(pg)
+        s.close()
+
+    """
+    k = int(k)
+    return AppAssets(algo="kshell")(graph, k)

python/tests/conftest.py

@@ -552,6 +552,12 @@ def wcc_result():
     yield ret
 
 
+@pytest.fixture(scope="module")
+def kshell_result():
+    ret = np.loadtxt("{}/../p2p-31-kshell-3".format(property_dir), dtype=int)
+    yield ret
+
+
 @pytest.fixture(scope="module")
 def pagerank_result():
     ret = {}

python/tests/test_app.py

@@ -30,6 +30,7 @@
 from graphscope import eigenvector_centrality
 from graphscope import hits
 from graphscope import k_core
+from graphscope import k_shell
 from graphscope import katz_centrality
 from graphscope import lpa
 from graphscope import pagerank
@@ -189,6 +190,7 @@ def test_app_on_undirected_graph(
     wcc_result,
     cdlp_result,
     triangles_result,
+    kshell_result,
 ):
     # sssp
     ctx1 = sssp(p2p_project_undirected_graph, src=6)
@@ -320,6 +322,24 @@ def test_app_on_undirected_graph(
         sorted(ctx9.to_numpy("r", vertex_range={"begin": 1, "end": 4})) == [1, 2, 2]
     )
 
+    # kshell
+    ctx10 = k_shell(p2p_project_undirected_graph, k=3)
+    r10 = (
+        ctx10.to_dataframe({"node": "v.id", "r": "r"})
+        .sort_values(by=["node"])
+        .to_numpy(dtype=int)
+    )
+    assert np.all(r10 == kshell_result)
+    assert np.all(
+        ctx10.to_dataframe(
+            {"node": "v.id", "r": "r"}, vertex_range={"begin": 1, "end": 4}
+        )
+        .sort_values(by=["node"])
+        .to_numpy()
+        == [[1, 0], [2, 0], [3, 0]]
+    )
+    assert np.all(ctx10.to_numpy("r", vertex_range={"begin": 1, "end": 4}) == [0, 0, 0])
+
 
 def test_error_on_parameters_not_correct(arrow_project_graph):
     # Incorrect type of parameters