Add distributed EasyGraph client and Kubernetes Worker release (1.6.4)

.github/workflows/distributed_image_publish.yml

@@ -0,0 +1,52 @@
+name: Publish distributed EasyGraph Worker image
+
+on:
+  workflow_dispatch:
+    inputs:
+      version:
+        description: Version from setup.py, for example 1.6.4
+        required: true
+        type: string
+
+permissions:
+  contents: read
+  packages: write
+
+jobs:
+  publish:
+    if: github.repository == 'easy-graph/Easy-Graph'
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: recursive
+      - name: Match image tag to Python package version
+        env:
+          RELEASE_VERSION: ${{ inputs.version }}
+        run: |
+          python - <<'PY'
+          import os
+          import pathlib
+          import re
+          source = pathlib.Path('setup.py').read_text()
+          match = re.search(r'^\s*version="([^"]+)"', source, re.MULTILINE)
+          assert match and match.group(1) == os.environ['RELEASE_VERSION']
+          PY
+      - uses: docker/setup-buildx-action@v3
+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+      - uses: docker/build-push-action@v6
+        id: image
+        with:
+          context: .
+          file: deploy/k8s/Dockerfile
+          platforms: linux/amd64
+          push: true
+          tags: ghcr.io/easy-graph/easygraph-distributed:${{ inputs.version }}
+          provenance: true
+          sbom: true
+      - name: Record digest
+        run: echo "ghcr.io/easy-graph/easygraph-distributed@${{ steps.image.outputs.digest }}"

.github/workflows/wheel_publish.yml

@@ -77,7 +77,7 @@ jobs:
       needs: [build_wheels, build_sdist]
       if: >-
         always() &&
-        ((github.ref == 'refs/heads/pybind11' && needs.build_wheels.result == 'success' && needs.build_sdist.result == 'success') || inputs.artifact_run_id != '')
+        (((github.ref == 'refs/heads/pybind11' || startsWith(github.ref, 'refs/tags/v')) && needs.build_wheels.result == 'success' && needs.build_sdist.result == 'success') || inputs.artifact_run_id != '')
       runs-on: ubuntu-latest
       environment: pypi
       permissions:

CMakeLists.txt

@@ -6,6 +6,8 @@ option(EASYGRAPH_ENABLE_OPENMP "Enable OpenMP acceleration (auto-detect)" ON)
 
 option(EASYGRAPH_ENABLE_GPU "EASYGRAPH_ENABLE_GPU" OFF)
 
+option(EASYGRAPH_ENABLE_DISTRIBUTED "Build the optional MPI graph analytics backend" OFF)
+
 add_subdirectory(cpp_easygraph)
 
 if (EASYGRAPH_ENABLE_GPU)

README.md

@@ -112,6 +112,11 @@ The conda package is no longer updated or maintained.
 
 If you've previously installed EasyGraph with conda, please uninstall it with ``conda`` and reinstall with ``pip``.
 
+The same Python package includes the distributed Kubernetes client. After an
+administrator prepares MPI Operator and shared storage, users can launch a
+Session without building a Worker image locally. See the
+[distributed Kubernetes guide](https://github.com/easy-graph/Easy-Graph/blob/pybind11/docs/distributed_k8s.md).
+
 ## Build From Source
 If prebuilt EasyGraph wheels are not supported for your platform (OS / CPU arch, check [here](https://pypi.org/simple/python-easygraph/)), or you want to have GPU-based functions enabled, you can build it locally.
 

cpp_easygraph/CMakeLists.txt

@@ -9,7 +9,17 @@ file(GLOB SOURCES
     cpp_easygraph.cpp
 )
 
-add_subdirectory(pybind11)
+if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/pybind11/CMakeLists.txt")
+    add_subdirectory(pybind11)
+else()
+    find_package(pybind11 CONFIG REQUIRED)
+endif()
+
+include("${CMAKE_CURRENT_LIST_DIR}/cmake/EasyGraphOpenMP.cmake")
+
+if(EASYGRAPH_ENABLE_DISTRIBUTED)
+    add_subdirectory(distributed)
+endif()
 
 option(EASYGRAPH_ENABLE_OPENMP "Enable OpenMP acceleration (auto-detect)" ON)
 option(EASYGRAPH_ENABLE_GPU "EASYGRAPH_ENABLE_GPU" OFF)
@@ -38,46 +48,4 @@ else()
 
 endif()
 
-if (EASYGRAPH_ENABLE_OPENMP)
-    if (APPLE AND CMAKE_CXX_COMPILER_ID MATCHES "Clang")
-        find_path(EG_LIBOMP_INCLUDE_DIR
-            NAMES omp.h
-            PATHS
-                /opt/homebrew/opt/libomp/include
-                /usr/local/opt/libomp/include
-        )
-
-        find_library(EG_LIBOMP_LIBRARY
-            NAMES omp libomp
-            PATHS
-                /opt/homebrew/opt/libomp/lib
-                /usr/local/opt/libomp/lib
-        )
-
-        if (EG_LIBOMP_INCLUDE_DIR AND EG_LIBOMP_LIBRARY)
-            message(STATUS "libomp found: ${EG_LIBOMP_LIBRARY}")
-
-            set(OpenMP_C_FLAGS   "-Xpreprocessor -fopenmp" CACHE STRING "" FORCE)
-            set(OpenMP_CXX_FLAGS "-Xpreprocessor -fopenmp" CACHE STRING "" FORCE)
-
-            set(OpenMP_CXX_INCLUDE_DIR "${EG_LIBOMP_INCLUDE_DIR}" CACHE PATH "" FORCE)
-            set(OpenMP_omp_LIBRARY     "${EG_LIBOMP_LIBRARY}"     CACHE FILEPATH "" FORCE)
-        else()
-            message(STATUS
-                "libomp not found on macOS. OpenMP will be disabled.\n"
-                "To enable OpenMP, run: brew install libomp"
-            )
-        endif()
-    endif()
-
-    find_package(OpenMP QUIET)
-endif()
-
-if (OpenMP_CXX_FOUND)
-    message(STATUS "OpenMP found, enabling parallel acceleration.")
-    target_link_libraries(cpp_easygraph PRIVATE OpenMP::OpenMP_CXX)
-    target_compile_definitions(cpp_easygraph PRIVATE EASYGRAPH_USE_OPENMP=1)
-else()
-    message(STATUS "OpenMP not found, building in single-thread mode.")
-    target_compile_definitions(cpp_easygraph PRIVATE EASYGRAPH_USE_OPENMP=0)
-endif()
+target_link_libraries(cpp_easygraph PRIVATE easygraph_openmp)

cpp_easygraph/cmake/EasyGraphOpenMP.cmake

@@ -0,0 +1,29 @@
+include_guard(GLOBAL)
+
+option(EASYGRAPH_ENABLE_OPENMP "Enable OpenMP acceleration (auto-detect)" ON)
+
+if(EASYGRAPH_ENABLE_OPENMP)
+    if(APPLE AND CMAKE_CXX_COMPILER_ID MATCHES "Clang")
+        find_path(EG_LIBOMP_INCLUDE_DIR NAMES omp.h
+            PATHS /opt/homebrew/opt/libomp/include /usr/local/opt/libomp/include)
+        find_library(EG_LIBOMP_LIBRARY NAMES omp libomp
+            PATHS /opt/homebrew/opt/libomp/lib /usr/local/opt/libomp/lib)
+        if(EG_LIBOMP_INCLUDE_DIR AND EG_LIBOMP_LIBRARY)
+            set(OpenMP_C_FLAGS "-Xpreprocessor -fopenmp" CACHE STRING "" FORCE)
+            set(OpenMP_CXX_FLAGS "-Xpreprocessor -fopenmp" CACHE STRING "" FORCE)
+            set(OpenMP_CXX_INCLUDE_DIR "${EG_LIBOMP_INCLUDE_DIR}" CACHE PATH "" FORCE)
+            set(OpenMP_omp_LIBRARY "${EG_LIBOMP_LIBRARY}" CACHE FILEPATH "" FORCE)
+        endif()
+    endif()
+    find_package(OpenMP QUIET)
+endif()
+
+add_library(easygraph_openmp INTERFACE)
+if(EASYGRAPH_ENABLE_OPENMP AND OpenMP_CXX_FOUND)
+    target_link_libraries(easygraph_openmp INTERFACE OpenMP::OpenMP_CXX)
+    target_compile_definitions(easygraph_openmp INTERFACE EASYGRAPH_USE_OPENMP=1)
+    message(STATUS "EasyGraph OpenMP acceleration enabled")
+else()
+    target_compile_definitions(easygraph_openmp INTERFACE EASYGRAPH_USE_OPENMP=0)
+    message(STATUS "EasyGraph OpenMP unavailable or disabled; using one thread")
+endif()

cpp_easygraph/common/pagerank_parallel.h

@@ -0,0 +1,66 @@
+#pragma once
+
+#include <cstddef>
+#include <stdexcept>
+#ifdef _OPENMP
+#include <omp.h>
+#endif
+
+namespace easygraph_parallel {
+inline void configure(int threads) {
+    if (threads < 1) throw std::invalid_argument("OpenMP thread count must be positive");
+    #ifdef _OPENMP
+    omp_set_dynamic(0);
+    omp_set_num_threads(threads);
+    #else
+    if (threads != 1) throw std::runtime_error("OpenMP is unavailable in this build");
+    #endif
+}
+
+template <typename Function>
+void for_each(std::size_t count, Function function) {
+    #pragma omp parallel for schedule(dynamic, 128)
+    for (std::ptrdiff_t i = 0; i < static_cast<std::ptrdiff_t>(count); ++i) function(i);
+}
+
+template <typename Function>
+double sum(std::size_t count, Function function) {
+    double result = 0.0;
+    #pragma omp parallel for reduction(+:result)
+    for (std::ptrdiff_t i = 0; i < static_cast<std::ptrdiff_t>(count); ++i) result += function(i);
+    return result;
+}
+
+inline void add(double& target, double value) {
+    #pragma omp atomic
+    target += value;
+}
+
+inline bool enabled() {
+    #ifdef _OPENMP
+    return true;
+    #else
+    return false;
+    #endif
+}
+
+inline int max_threads() {
+    #ifdef _OPENMP
+    return omp_get_max_threads();
+    #else
+    return 1;
+    #endif
+}
+
+inline int team_size() {
+    int result = 1;
+    #ifdef _OPENMP
+    #pragma omp parallel
+    {
+        #pragma omp single
+        result = omp_get_num_threads();
+    }
+    #endif
+    return result;
+}
+}

cpp_easygraph/distributed/CMakeLists.txt

@@ -0,0 +1,21 @@
+cmake_minimum_required(VERSION 3.23)
+if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR)
+    project(easygraph_distributed LANGUAGES CXX)
+    set(PYBIND11_FINDPYTHON ON)
+    find_package(Python COMPONENTS Interpreter Development.Module REQUIRED)
+    find_package(pybind11 CONFIG REQUIRED)
+endif()
+
+include("${CMAKE_CURRENT_LIST_DIR}/../cmake/EasyGraphOpenMP.cmake")
+find_package(MPI REQUIRED COMPONENTS CXX)
+pybind11_add_module(cpp_easygraph_distributed
+    bindings.cpp
+    communicator.cpp
+    distributed_graph.cpp
+    partitioner.cpp
+    loader.cpp
+    worker.cpp
+    algorithms/pagerank.cpp
+)
+target_compile_features(cpp_easygraph_distributed PRIVATE cxx_std_11)
+target_link_libraries(cpp_easygraph_distributed PRIVATE MPI::MPI_CXX easygraph_openmp)

cpp_easygraph/distributed/algorithms/pagerank.cpp

@@ -0,0 +1,8 @@
+#include "pagerank.h"
+
+namespace egdist {
+PageRankResult pagerank(const DistributedGraph& graph, double alpha, int max_iter,
+                        double tol, bool weighted) {
+    return PageRankWorker(graph).run(alpha, max_iter, tol, weighted);
+}
+}

cpp_easygraph/distributed/algorithms/pagerank.h

@@ -0,0 +1,7 @@
+#pragma once
+#include "../worker.h"
+
+namespace egdist {
+PageRankResult pagerank(const DistributedGraph& graph, double alpha, int max_iter,
+                        double tol, bool weighted);
+}

cpp_easygraph/distributed/bindings.cpp

@@ -0,0 +1,46 @@
+#include <pybind11/pybind11.h>
+#include <pybind11/stl.h>
+
+#include "algorithms/pagerank.h"
+#include "communicator.h"
+#include "distributed_graph.h"
+#include "loader.h"
+#include "../common/pagerank_parallel.h"
+
+namespace py = pybind11;
+using namespace egdist;
+
+PYBIND11_MODULE(cpp_easygraph_distributed, m) {
+    m.doc() = "Optional MPI backend for EasyGraph distributed analytics";
+    m.def("finalize_mpi", &finalize_mpi);
+    m.def("configure_parallelism", &easygraph_parallel::configure, py::arg("threads"));
+    m.def("parallel_info", []() {
+        py::dict info;
+        info["openmp_enabled"] = easygraph_parallel::enabled();
+        info["max_threads"] = easygraph_parallel::max_threads();
+        info["team_size"] = easygraph_parallel::team_size();
+        return info;
+    });
+    py::class_<Communicator, std::shared_ptr<Communicator>>(m, "Communicator")
+        .def(py::init<>())
+        .def_property_readonly("rank", &Communicator::rank)
+        .def_property_readonly("size", &Communicator::size)
+        .def_property_readonly("closed", &Communicator::closed)
+        .def("close", &Communicator::close)
+        .def("barrier", &Communicator::barrier)
+        .def("check_error", &Communicator::check_error)
+        .def("agree", &Communicator::agree)
+        .def("broadcast_string", &Communicator::broadcast_string)
+        .def("gather_result", &Communicator::gather_result);
+    py::class_<DistributedGraph, std::shared_ptr<DistributedGraph>>(m, "DistributedGraph")
+        .def_readonly("vertices", &DistributedGraph::vertices)
+        .def_readonly("global_num_vertices", &DistributedGraph::global_num_vertices)
+        .def_readonly("global_num_edges", &DistributedGraph::global_num_edges)
+        .def_property_readonly("local_num_arcs", [](const DistributedGraph& g) { return g.targets.size(); });
+    m.def("read_edgelist", &read_edgelist, py::arg("path"), py::arg("comm"), py::arg("directed"));
+    m.def("pagerank", [](const DistributedGraph& graph, double alpha, int max_iter,
+                          double tol, bool weighted) {
+        auto result = pagerank(graph, alpha, max_iter, tol, weighted);
+        return py::make_tuple(result.ids, result.values, result.iterations);
+    }, py::arg("graph"), py::arg("alpha"), py::arg("max_iter"), py::arg("tol"), py::arg("weighted"));
+}