This is an automated email from the ASF dual-hosted git repository.
RexXiong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 61499a0c9 [CELEBORN-2319] Standalone LifecycleManager && rust sdk
61499a0c9 is described below
commit 61499a0c9ac40786c0e846009f0ea77e1b1d42ed
Author: 夷羿 <[email protected]>
AuthorDate: Thu Jun 11 21:55:41 2026 +0800
[CELEBORN-2319] Standalone LifecycleManager && rust sdk
### What changes were proposed in this pull request?
This PR introduces two major features to support **non-JVM (C++/Rust)
clients** using Apache Celeborn for shuffle:
**1. Standalone LifecycleManager Daemon (Scala/JVM)**
- Added `LifecycleManagerDaemon` — a standalone JVM process that hosts a
`LifecycleManager` independently from any compute engine (Spark/Flink) Driver.
It installs a shutdown hook (with a watchdog that force-halts if graceful stop
exceeds the timeout) and blocks until SIGINT/SIGTERM.
- Added `LifecycleManagerDaemonArguments` for CLI argument parsing
(`--app-id`, `--master-endpoints`, `--port`/`-p`, `--host`,
`--properties-file`, `-h`/`--help`). Parsing is a pure function that throws
`ArgumentParseException` (carrying an exit code) so every branch is
unit-testable; `parseOrExit` wraps it for the process entry point.
- Added `sbin/start-lifecycle-manager.sh` launch script with classpath
assembly, environment loading, required-argument validation, automatic
free-port selection, and **RPC-port polling** to confirm the daemon is actually
bound before reporting success.
- Added a new **`lifecycle-manager` Maven/sbt module** that depends on
`celeborn-service`, `celeborn-client` and `celeborn-common`. Registered the
module in both the root `pom.xml` and `project/CelebornBuild.scala`
(`projectDefinitions`), and wired it into `build/make-distribution.sh` for both
the Maven and sbt build paths.
- **Security note**: the standalone LM runs without authentication; it logs
a warning on startup and the code documents that operators must bind it to a
trusted network only.
**2. Rust SDK via C++ FFI (`rust/` directory)**
- `celeborn-client-sys`: Low-level FFI crate bridging Rust ↔ C++ via a
**plain C ABI** (no `cxx`). The C++ side exposes `celeborn_ffi_*` functions
(`create_client`, `setup_lifecycle_manager`, `shutdown`, `push_data`,
`mapper_end`, partition reader open/read/close, etc.) returning status codes
plus heap-allocated error strings.
- `build.rs`: links the single aggregated shared library
`libceleborn_client.{so,dylib}` (which whole-archives all internal static libs
and hides non-`celeborn` symbols), so downstream Rust never sees folly /
protobuf / glog / abseil.
- `celeborn-client`: Safe, ergonomic Rust wrapper providing `ShuffleClient`
with:
- Input validation extracted into a pure `validate_connect_args` (app_id
non-empty, port > 0, codec ∈ {NONE, LZ4, ZSTD}) — testable without a live
cluster.
- Documented `Send + Sync` rationale (the C++ `ShuffleClientImpl`
synchronizes internally), enabling `Arc<ShuffleClient>` sharing for concurrent
`&self` push/read.
- `Drop`-safe shutdown that nulls the handle to avoid a double
`celeborn_ffi_shutdown`. The native handle is **intentionally leaked** after
shutdown to dodge a folly `EventBase` teardown race; this implies a
**per-process-client** usage model, which is documented prominently on the type.
- Two example programs (`data_sum_writer.rs`, `data_sum_reader.rs`)
mirroring the existing C++ `DataSumWithWriterClient` /
`DataSumWithReaderClient` test programs. The writer seeds its RNG **per mapper
thread** so each thread emits a distinct byte stream, genuinely exercising the
concurrent push path.
**3. C++ build portability (`cpp/CMakeLists.txt`)**
- Discover the Homebrew prefix via `brew --prefix` instead of hard-coding
`/opt/homebrew` (so Intel macOS under `/usr/local` works too).
- Discover Abseil via `find_package(absl CONFIG)`, falling back to a
toolchain-derived GNU multiarch dir (covers `aarch64-linux-gnu`, which the
README advertises); a missing Abseil is now a `FATAL_ERROR` rather than a
silent warning.
- Guard the x86-only `-msse4.2` flag by architecture so aarch64 builds
compile.
### Why are the changes needed?
Currently, `LifecycleManager` can only run **embedded inside a JVM-based
compute engine Driver** (e.g., Spark Driver). This makes it impossible for
non-JVM applications (Daft engine, etc.) to use Celeborn as their shuffle
service, because:
1. The C++ client requires a running `LifecycleManager` to coordinate
shuffle metadata (register shuffles, allocate slots, manage partition
locations) with Celeborn Masters and Workers.
2. Without a standalone `LifecycleManager`, non-JVM applications have no
way to bootstrap this coordination layer.
By decoupling the `LifecycleManager` into a **standalone daemon process**,
any client — regardless of language runtime — can connect to it via RPC. The
Rust SDK then leverages this architecture to provide first-class Rust support
by bridging to the existing, battle-tested C++ client implementation via FFI.
### Does this PR resolve a correctness bug?
No
### Does this PR introduce _any_ user-facing change?
Yes.
- **New component**: Users can now start a standalone `LifecycleManager`
daemon via `sbin/start-lifecycle-manager.sh --app-id <id> --master-endpoints
<eps> [--port <port>] [--host <host>]`.
- **New SDK**: Rust applications can now use the `celeborn-client` crate to
perform shuffle read/write operations against a Celeborn cluster.
- **Limitation**: The standalone `LifecycleManager` does **not** support
auth (`celeborn.auth.enabled` must be `false`), as the C++/Rust clients lack
SASL support. Deploy it on a trusted network only.
### How was this patch tested?
- **Unit tests (JVM)**: `LifecycleManagerDaemonArgumentsSuite` covers the
happy paths plus help / unknown-arg / missing-arg / invalid-port branches and
`applyArgsToConf` (13 tests). Run via Maven (`mvn -pl lifecycle-manager test`)
and compiled/style-checked under sbt as a first-class module.
- **Unit tests (Rust)**: `validate_connect_args` is covered by `cargo test
-p celeborn-client` (codec/app_id/port validation) without requiring a cluster.
- **Integration**: The Rust SDK was validated using the `data_sum_writer` /
`data_sum_reader` examples (Rust ports of `DataSumWithWriterClient.cpp` /
`DataSumWithReaderClient.cpp`), which write random numeric data across
partitions and verify correctness by comparing partition sums between writer
and reader. The `LifecycleManagerDaemon` was tested by starting it against a
local Celeborn cluster (Master + Workers) and verifying the Rust examples
connect, push, and read through the daemon.
Closes #3677 from gavin9402/standalone_and_rust.
Lead-authored-by: 夷羿 <[email protected]>
Co-authored-by: Zhou <[email protected]>
Signed-off-by: Shuang <[email protected]>
---
.rat-excludes | 5 +
build/make-distribution.sh | 27 +-
cpp/CMakeLists.txt | 182 +++++++-
cpp/celeborn/CMakeLists.txt | 1 +
cpp/celeborn/{ => ffi}/CMakeLists.txt | 29 +-
cpp/celeborn/ffi/CelebornFfi.cc | 347 +++++++++++++++
cpp/celeborn/ffi/CelebornFfi.h | 141 ++++++
cpp/dummy.cc | 27 ++
cpp/exports.map | 26 ++
cpp/{celeborn/CMakeLists.txt => exports.txt} | 16 +-
lifecycle-manager/pom.xml | 68 +++
.../lifecyclemanager/LifecycleManagerDaemon.scala | 146 +++++++
.../LifecycleManagerDaemonArguments.scala | 145 +++++++
.../LifecycleManagerDaemonArgumentsSuite.scala | 185 ++++++++
pom.xml | 1 +
project/CelebornBuild.scala | 14 +
rust/.gitignore | 7 +
cpp/celeborn/CMakeLists.txt => rust/Cargo.toml | 16 +-
.../celeborn-client-sys/Cargo.toml | 17 +-
rust/celeborn-client-sys/build.rs | 186 ++++++++
rust/celeborn-client-sys/src/lib.rs | 158 +++++++
.rat-excludes => rust/celeborn-client/Cargo.toml | 47 +-
rust/celeborn-client/build.rs | 28 ++
rust/celeborn-client/src/lib.rs | 478 +++++++++++++++++++++
rust/examples/data_sum_reader.rs | 137 ++++++
rust/examples/data_sum_writer.rs | 154 +++++++
rust/resource/lib/README.md | 51 +++
sbin/start-lifecycle-manager.sh | 217 ++++++++++
sbin/stop-lifecycle-manager.sh | 165 +++++++
29 files changed, 2947 insertions(+), 74 deletions(-)
diff --git a/.rat-excludes b/.rat-excludes
index 22c97119e..0cf626add 100644
--- a/.rat-excludes
+++ b/.rat-excludes
@@ -38,3 +38,8 @@ build/sbt-config/**
**/benchmarks/**
**/node_modules/**
cpp/cmake/FindSodium.cmake
+cpp/build/**
+cpp/build_static/**
+cpp/thirdparty/**
+rust/Cargo.lock
+rust/.gitignore
diff --git a/build/make-distribution.sh b/build/make-distribution.sh
index e160b758d..67e97017e 100755
--- a/build/make-distribution.sh
+++ b/build/make-distribution.sh
@@ -146,7 +146,7 @@ function build_service {
# Store the command as an array because $MVN variable might have spaces in
it.
# Normal quoting tricks don't work.
# See: http://mywiki.wooledge.org/BashFAQ/050
- BUILD_COMMAND=("$MVN" clean package $MVN_DIST_OPT -pl master,worker,cli -am
$@)
+ BUILD_COMMAND=("$MVN" clean package $MVN_DIST_OPT -pl
master,worker,cli,lifecycle-manager -am $@)
# Actually build the jar
echo -e "\nBuilding with..."
@@ -158,6 +158,7 @@ function build_service {
mkdir -p "$DIST_DIR/master-jars"
mkdir -p "$DIST_DIR/worker-jars"
mkdir -p "$DIST_DIR/cli-jars"
+ mkdir -p "$DIST_DIR/lifecycle-manager-jars"
## Copy master jars
cp "$PROJECT_DIR"/master/target/celeborn-master_$SCALA_VERSION-$VERSION.jar
"$DIST_DIR/master-jars/"
@@ -177,6 +178,21 @@ function build_service {
for jar in $(ls "$PROJECT_DIR/cli/target/scala-$SCALA_VERSION/jars"); do
(cd $DIST_DIR/cli-jars; ln -snf "../jars/$jar" .)
done
+ ## Copy lifecycle-manager jars
+ # lifecycle-manager depends on celeborn-client which is not a dependency of
master/worker,
+ # so we copy its project-internal dependency jars that are missing from
jars/.
+ for module_jar in \
+
"$PROJECT_DIR/lifecycle-manager/target/celeborn-lifecycle-manager_$SCALA_VERSION-$VERSION.jar"
\
+ "$PROJECT_DIR/client/target/celeborn-client_$SCALA_VERSION-$VERSION.jar";
do
+ jarname=$(basename "$module_jar")
+ if [ ! -f "$DIST_DIR/jars/$jarname" ]; then
+ cp "$module_jar" "$DIST_DIR/jars/"
+ fi
+ done
+ cp
"$PROJECT_DIR"/lifecycle-manager/target/celeborn-lifecycle-manager_$SCALA_VERSION-$VERSION.jar
"$DIST_DIR/lifecycle-manager-jars/"
+ for jar in $(ls "$DIST_DIR/jars"); do
+ (cd $DIST_DIR/lifecycle-manager-jars; ln -snf "../jars/$jar" .)
+ done
}
function build_spark_client {
@@ -304,12 +320,13 @@ function sbt_build_service {
"${BUILD_COMMAND[@]}"
- $SBT
"celeborn-master/copyJars;celeborn-worker/copyJars;celeborn-cli/copyJars"
+ $SBT
"celeborn-master/copyJars;celeborn-worker/copyJars;celeborn-cli/copyJars;celeborn-lifecycle-manager/copyJars"
mkdir -p "$DIST_DIR/jars"
mkdir -p "$DIST_DIR/master-jars"
mkdir -p "$DIST_DIR/worker-jars"
mkdir -p "$DIST_DIR/cli-jars"
+ mkdir -p "$DIST_DIR/lifecycle-manager-jars"
## Copy master jars
cp
"$PROJECT_DIR"/master/target/scala-$SCALA_VERSION/celeborn-master_$SCALA_VERSION-$VERSION.jar
"$DIST_DIR/master-jars/"
@@ -329,6 +346,12 @@ function sbt_build_service {
for jar in $(ls "$PROJECT_DIR/cli/target/scala-$SCALA_VERSION/jars"); do
(cd $DIST_DIR/cli-jars; ln -snf "../jars/$jar" .)
done
+ ## Copy lifecycle-manager jars
+ cp
"$PROJECT_DIR"/lifecycle-manager/target/scala-$SCALA_VERSION/celeborn-lifecycle-manager_$SCALA_VERSION-$VERSION.jar
"$DIST_DIR/lifecycle-manager-jars/"
+ cp "$PROJECT_DIR"/lifecycle-manager/target/scala-$SCALA_VERSION/jars/*.jar
"$DIST_DIR/jars/"
+ for jar in $(ls
"$PROJECT_DIR/lifecycle-manager/target/scala-$SCALA_VERSION/jars"); do
+ (cd $DIST_DIR/lifecycle-manager-jars; ln -snf "../jars/$jar" .)
+ done
}
function sbt_build_client {
diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
index 8d54b62e3..151a70f65 100644
--- a/cpp/CMakeLists.txt
+++ b/cpp/CMakeLists.txt
@@ -23,6 +23,13 @@ enable_testing()
set(CMAKE_CXX_STANDARD 17)
set(CMAKE_CXX_STANDARD_REQUIRED True)
+
+# All internal static libs are whole-archived into the shared library
+# `celeborn_client`, which on Linux/ELF requires every contained object
+# to be position-independent. Without this, `ld` complains:
+# relocation R_X86_64_TPOFF32 against hidden symbol ... can not be used
+# when making a shared object
+set(CMAKE_POSITION_INDEPENDENT_CODE ON)
message("Appending CMAKE_CXX_FLAGS with ${SCRIPT_CXX_FLAGS}")
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${SCRIPT_CXX_FLAGS}")
if ("${TREAT_WARNINGS_AS_ERRORS}")
@@ -30,7 +37,33 @@ if ("${TREAT_WARNINGS_AS_ERRORS}")
endif ()
# Avoid folly::f14::detail::F14LinkCheck problem on x86-64 platform.
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -msse4.2")
+# -msse4.2 is an x86-only flag; emitting it on aarch64 (Apple Silicon or
+# aarch64 Linux) makes the compiler error out, so guard it by architecture.
+if (CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64|AMD64|amd64")
+ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -msse4.2")
+endif ()
+
+# Resolve the Homebrew prefix on macOS instead of hard-coding the
+# Apple-Silicon path. Intel Macs install under /usr/local while Apple
+# Silicon uses /opt/homebrew; `brew --prefix` reports whichever applies.
+if (APPLE)
+ if (DEFINED ENV{HOMEBREW_PREFIX})
+ set(HOMEBREW_PREFIX "$ENV{HOMEBREW_PREFIX}")
+ else ()
+ find_program(BREW_EXECUTABLE brew)
+ if (BREW_EXECUTABLE)
+ execute_process(
+ COMMAND ${BREW_EXECUTABLE} --prefix
+ OUTPUT_VARIABLE HOMEBREW_PREFIX
+ OUTPUT_STRIP_TRAILING_WHITESPACE)
+ elseif (EXISTS /opt/homebrew)
+ set(HOMEBREW_PREFIX "/opt/homebrew")
+ else ()
+ set(HOMEBREW_PREFIX "/usr/local")
+ endif ()
+ endif ()
+ message(STATUS "Using HOMEBREW_PREFIX=${HOMEBREW_PREFIX}")
+endif ()
# Set CMAKE_BUILD_TYPE to 'Release' if it is not specified.
if (NOT CMAKE_BUILD_TYPE)
@@ -102,7 +135,12 @@ if (APPLE)
# Homebrew and only shared library is installed.
find_package(gflags REQUIRED COMPONENTS shared)
else ()
- find_package(gflags REQUIRED COMPONENTS static)
+ # We build celeborn_client as a SHARED library (whole-archives every
+ # internal .a). Ubuntu's libgflags.a from apt is not compiled with
+ # -fPIC, so it cannot be linked into a .so. The shared gflags
+ # (libgflags.so) is PIC and works for both standalone executables and
+ # the aggregated shared library.
+ find_package(gflags REQUIRED COMPONENTS shared)
endif ()
find_package(glog REQUIRED)
@@ -177,9 +215,140 @@ endif()
add_subdirectory(celeborn)
# ---------------------------------------------------------------------------
-# Install rules — headers + static libraries
+# Aggregate SHARED library — bundles all internal static libs (including
+# the C ABI shim `celeborn_ffi`) into a single libceleborn_client.{so,dylib}.
+# Downstream language bindings (Rust / ctypes / etc.) link only this file
+# and never see folly / protobuf / glog / abseil.
+# ---------------------------------------------------------------------------
+add_library(celeborn_client SHARED dummy.cc)
+
+# Internal targets propagate plain library names like "xxhash" / "fmt" via
+# their link interface. Those need an actual search path on macOS where the
+# Homebrew prefix is not on ld's default path.
+if(APPLE)
+ target_link_directories(celeborn_client PRIVATE
+ ${HOMEBREW_PREFIX}/lib
+ ${HOMEBREW_PREFIX}/opt/openssl@3/lib)
+endif()
+
+set(_CELEBORN_INTERNAL_LIBS
+ celeborn_ffi client protocol network proto memory conf utils)
+
+if(APPLE)
+ # ld64: -force_load pulls every object out of each .a so no symbol gets
+ # stripped by the linker before it ships in the .dylib. Wrapping the
+ # target file path in a generator expression does not register a
+ # build-order dependency on those targets, so add it explicitly.
+ add_dependencies(celeborn_client ${_CELEBORN_INTERNAL_LIBS})
+ foreach(_lib IN LISTS _CELEBORN_INTERNAL_LIBS)
+ target_link_libraries(celeborn_client PRIVATE
+ "-Wl,-force_load" "$<TARGET_FILE:${_lib}>")
+ endforeach()
+else()
+ target_link_libraries(celeborn_client PRIVATE
+ -Wl,--whole-archive
+ ${_CELEBORN_INTERNAL_LIBS}
+ -Wl,--no-whole-archive)
+endif()
+
+# Third-party deps remain NEEDED entries inside the produced .so/.dylib and
+# resolve at runtime from the system loader path.
+target_link_libraries(celeborn_client PRIVATE
+ ${WANGLE} ${FIZZ}
+ ${LIBSODIUM_LIBRARY}
+ ${FOLLY_WITH_DEPENDENCIES}
+ ${LZ4_WITH_DEPENDENCIES}
+ ${GLOG} ${GFLAGS_LIBRARIES}
+ ${RE2}
+ protobuf::libprotobuf
+ OpenSSL::SSL OpenSSL::Crypto)
+
+# protobuf v22+ (a.k.a. 4.22.0+) switched to an abseil-based implementation and
+# pulls in dozens of absl_* sub-libraries at link time. Older versions
+# (3.x / v21 and below) are self-contained and do not need abseil at all.
+#
+# Protobuf_VERSION is reported as the *library* version (e.g. "3.21.7"), while
+# the "v22" release corresponds to library version 4.22.0. We therefore check
+# whether the major version is >= 4 (i.e. protobuf v22+).
+if(Protobuf_VERSION VERSION_GREATER_EQUAL "4.0.0")
+ message(STATUS "Protobuf ${Protobuf_VERSION} requires abseil — searching …")
+ find_package(absl CONFIG QUIET)
+ if(absl_FOUND)
+ if(TARGET absl::absl)
+ target_link_libraries(celeborn_client PRIVATE absl::absl)
+ else()
+ target_link_libraries(celeborn_client PRIVATE
+ absl::base absl::strings absl::status absl::statusor
+ absl::synchronization absl::time absl::log absl::log_internal_check_op)
+ endif()
+ else()
+ # No CMake config — fall back to globbing the platform library directory.
+ if(APPLE)
+ set(_ABSEIL_LIB_DIR "${HOMEBREW_PREFIX}/lib")
+ set(_ABSEIL_LIB_EXT "dylib")
+ else()
+ set(_ABSEIL_LIB_EXT "so")
+ execute_process(
+ COMMAND ${CMAKE_CXX_COMPILER} -dumpmachine
+ OUTPUT_VARIABLE _MULTIARCH_TRIPLET
+ OUTPUT_STRIP_TRAILING_WHITESPACE
+ ERROR_QUIET)
+ set(_ABSEIL_SEARCH_DIRS
+ "/usr/lib/${_MULTIARCH_TRIPLET}"
+ "/usr/local/lib"
+ "/usr/lib64"
+ "/usr/lib")
+ set(_ABSEIL_LIB_DIR "")
+ foreach(_dir IN LISTS _ABSEIL_SEARCH_DIRS)
+ file(GLOB _probe "${_dir}/libabsl_*.${_ABSEIL_LIB_EXT}")
+ if(_probe)
+ set(_ABSEIL_LIB_DIR "${_dir}")
+ break()
+ endif()
+ endforeach()
+ endif()
+
+ file(GLOB _ABSEIL_LIBS "${_ABSEIL_LIB_DIR}/libabsl_*.${_ABSEIL_LIB_EXT}")
+ if(_ABSEIL_LIBS)
+ message(STATUS "Found abseil libraries in ${_ABSEIL_LIB_DIR}")
+ target_link_libraries(celeborn_client PRIVATE ${_ABSEIL_LIBS})
+ else()
+ message(FATAL_ERROR
+ "Could not locate libabsl_*.${_ABSEIL_LIB_EXT}. Protobuf "
+ "${Protobuf_VERSION} depends on abseil; install abseil-cpp "
+ "(providing absl CMake config or libabsl_*.${_ABSEIL_LIB_EXT}) "
+ "or point CMAKE_PREFIX_PATH at it.")
+ endif()
+ endif()
+else()
+ message(STATUS "Protobuf ${Protobuf_VERSION} does not require abseil —
skipping absl search")
+endif()
+
+# Hide every non-celeborn symbol so the dylib does not collide with the
+# host process's own protobuf / abseil / folly runtimes.
+set_target_properties(celeborn_client PROPERTIES
+ CXX_VISIBILITY_PRESET hidden
+ VISIBILITY_INLINES_HIDDEN ON)
+
+if(APPLE)
+ target_link_options(celeborn_client PRIVATE
+ "-Wl,-exported_symbols_list,${CMAKE_CURRENT_SOURCE_DIR}/exports.txt")
+ set_target_properties(celeborn_client PROPERTIES
+ INSTALL_RPATH
"@loader_path;${HOMEBREW_PREFIX}/lib;${HOMEBREW_PREFIX}/opt/openssl@3/lib"
+ BUILD_WITH_INSTALL_RPATH TRUE)
+else()
+ target_link_options(celeborn_client PRIVATE
+ "-Wl,--version-script=${CMAKE_CURRENT_SOURCE_DIR}/exports.map"
+ "-Wl,--exclude-libs,ALL")
+ set_target_properties(celeborn_client PROPERTIES
+ INSTALL_RPATH "$ORIGIN"
+ BUILD_WITH_INSTALL_RPATH TRUE)
+endif()
+
+# ---------------------------------------------------------------------------
+# Install rules — headers + the aggregated shared library.
+# Static libraries are still installed to keep the legacy build path working.
# ---------------------------------------------------------------------------
-# Install all public headers preserving directory structure.
install(
DIRECTORY celeborn/
DESTINATION include/celeborn
@@ -187,15 +356,14 @@ install(
PATTERN "tests" EXCLUDE
)
-# Install the generated proto header.
install(
FILES ${ProtoGenHeader}
DESTINATION include/celeborn/proto
)
-# Install static libraries.
install(
- TARGETS client conf memory network proto protocol utils
+ TARGETS celeborn_client celeborn_ffi client conf memory network proto
protocol utils
ARCHIVE DESTINATION lib
LIBRARY DESTINATION lib
+ RUNTIME DESTINATION bin
)
diff --git a/cpp/celeborn/CMakeLists.txt b/cpp/celeborn/CMakeLists.txt
index 1588488a7..fc6d9b2be 100644
--- a/cpp/celeborn/CMakeLists.txt
+++ b/cpp/celeborn/CMakeLists.txt
@@ -19,6 +19,7 @@ add_subdirectory(conf)
add_subdirectory(protocol)
add_subdirectory(network)
add_subdirectory(client)
+add_subdirectory(ffi)
if(CELEBORN_BUILD_TESTS)
add_subdirectory(tests)
diff --git a/cpp/celeborn/CMakeLists.txt b/cpp/celeborn/ffi/CMakeLists.txt
similarity index 58%
copy from cpp/celeborn/CMakeLists.txt
copy to cpp/celeborn/ffi/CMakeLists.txt
index 1588488a7..48e5a6356 100644
--- a/cpp/celeborn/CMakeLists.txt
+++ b/cpp/celeborn/ffi/CMakeLists.txt
@@ -12,14 +12,23 @@
# 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.
-add_subdirectory(proto)
-add_subdirectory(memory)
-add_subdirectory(utils)
-add_subdirectory(conf)
-add_subdirectory(protocol)
-add_subdirectory(network)
-add_subdirectory(client)
-if(CELEBORN_BUILD_TESTS)
- add_subdirectory(tests)
-endif()
+# C ABI shim that wraps celeborn::client::ShuffleClient* in pure-C entry
+# points. Built as a STATIC archive that is whole-archived into the
+# aggregated libceleborn_client.{so,dylib} (see top-level CMakeLists.txt).
+add_library(
+ celeborn_ffi
+ STATIC
+ CelebornFfi.cc)
+
+target_include_directories(celeborn_ffi PUBLIC
+ ${CMAKE_SOURCE_DIR}
+ ${CMAKE_BINARY_DIR})
+
+target_link_libraries(
+ celeborn_ffi
+ client
+ conf
+ ${FOLLY_WITH_DEPENDENCIES}
+ ${GLOG}
+ ${GFLAGS_LIBRARIES})
diff --git a/cpp/celeborn/ffi/CelebornFfi.cc b/cpp/celeborn/ffi/CelebornFfi.cc
new file mode 100644
index 000000000..10a00081a
--- /dev/null
+++ b/cpp/celeborn/ffi/CelebornFfi.cc
@@ -0,0 +1,347 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+#include "celeborn/ffi/CelebornFfi.h"
+
+#include <cstdlib>
+#include <cstring>
+#include <exception>
+#include <memory>
+#include <new>
+#include <string>
+#include <vector>
+
+#include "celeborn/client/ShuffleClient.h"
+#include "celeborn/client/reader/CelebornInputStream.h"
+#include "celeborn/conf/CelebornConf.h"
+
+namespace {
+
+struct ClientImpl {
+ std::shared_ptr<celeborn::conf::CelebornConf> conf;
+ std::shared_ptr<celeborn::client::ShuffleClientEndpoint> endpoint;
+ std::shared_ptr<celeborn::client::ShuffleClientImpl> client;
+ std::string app_id;
+ std::string lifecycle_manager_host;
+};
+
+// One open partition stream. The stream references the owning
+// ShuffleClient, so callers must close every reader before shutting down
+// the client.
+struct PartitionReaderImpl {
+ std::unique_ptr<celeborn::client::CelebornInputStream> stream;
+};
+
+inline ClientImpl* as_impl(celeborn_ffi_handle* h) {
+ return reinterpret_cast<ClientImpl*>(h);
+}
+
+inline PartitionReaderImpl* as_reader_impl(celeborn_ffi_partition_reader* r) {
+ return reinterpret_cast<PartitionReaderImpl*>(r);
+}
+
+char* dup_message(const std::string& msg) {
+ char* out = static_cast<char*>(std::malloc(msg.size() + 1));
+ if (!out) {
+ return nullptr;
+ }
+ std::memcpy(out, msg.data(), msg.size());
+ out[msg.size()] = '\0';
+ return out;
+}
+
+void set_error(char** err_out, const std::string& msg) {
+ if (err_out) {
+ *err_out = dup_message("celeborn-ffi: " + msg);
+ }
+}
+
+template <typename Fn>
+celeborn_ffi_status guarded(char** err_out, Fn&& fn) {
+ try {
+ fn();
+ return CELEBORN_FFI_OK;
+ } catch (const std::exception& e) {
+ set_error(err_out, e.what());
+ return CELEBORN_FFI_ERROR;
+ } catch (...) {
+ set_error(err_out, "unknown C++ exception");
+ return CELEBORN_FFI_ERROR;
+ }
+}
+
+// Reject null pointers at the FFI boundary so that misuse from non-Rust
+// callers surfaces as CELEBORN_FFI_ERROR with a descriptive message rather
+// than a SIGSEGV inside the C++ implementation.
+#define CELEBORN_FFI_REQUIRE_NON_NULL(ptr, err_out) \
+ do { \
+ if ((ptr) == nullptr) { \
+ set_error((err_out), "null pointer for argument '" #ptr "'"); \
+ return CELEBORN_FFI_ERROR; \
+ } \
+ } while (0)
+
+} // namespace
+
+extern "C" {
+
+void celeborn_ffi_free_error(char* err) {
+ std::free(err);
+}
+
+void celeborn_ffi_free_buffer(uint8_t* data) {
+ delete[] data;
+}
+
+celeborn_ffi_handle* celeborn_ffi_create_client(
+ const char* app_id,
+ size_t app_id_len,
+ int32_t push_buffer_max_size,
+ const char* codec,
+ size_t codec_len,
+ char** err_out) {
+ if (app_id == nullptr && app_id_len > 0) {
+ set_error(err_out, "null pointer for argument 'app_id'");
+ return nullptr;
+ }
+ if (codec == nullptr && codec_len > 0) {
+ set_error(err_out, "null pointer for argument 'codec'");
+ return nullptr;
+ }
+ try {
+ auto impl = std::make_unique<ClientImpl>();
+ if (app_id_len > 0) {
+ impl->app_id.assign(app_id, app_id_len);
+ }
+ impl->conf = std::make_shared<celeborn::conf::CelebornConf>();
+
+ if (push_buffer_max_size > 0) {
+ impl->conf->registerProperty(
+ celeborn::conf::CelebornConf::kClientPushBufferMaxSize,
+ std::to_string(push_buffer_max_size) + "b");
+ }
+ if (codec_len > 0) {
+ impl->conf->registerProperty(
+ celeborn::conf::CelebornConf::kShuffleCompressionCodec,
+ std::string(codec, codec_len));
+ }
+
+ impl->endpoint =
+ std::make_shared<celeborn::client::ShuffleClientEndpoint>(impl->conf);
+ impl->client = celeborn::client::ShuffleClientImpl::create(
+ impl->app_id, impl->conf, *(impl->endpoint));
+ return reinterpret_cast<celeborn_ffi_handle*>(impl.release());
+ } catch (const std::exception& e) {
+ set_error(err_out, e.what());
+ return nullptr;
+ } catch (...) {
+ set_error(err_out, "unknown C++ exception");
+ return nullptr;
+ }
+}
+
+celeborn_ffi_status celeborn_ffi_setup_lifecycle_manager(
+ celeborn_ffi_handle* handle,
+ const char* host,
+ size_t host_len,
+ int32_t port,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(handle, err_out);
+ if (host == nullptr && host_len > 0) {
+ set_error(err_out, "null pointer for argument 'host'");
+ return CELEBORN_FFI_ERROR;
+ }
+ return guarded(err_out, [&] {
+ auto* impl = as_impl(handle);
+ if (host_len > 0) {
+ impl->lifecycle_manager_host.assign(host, host_len);
+ } else {
+ impl->lifecycle_manager_host.clear();
+ }
+ impl->client->setupLifecycleManagerRef(impl->lifecycle_manager_host, port);
+ });
+}
+
+celeborn_ffi_status celeborn_ffi_shutdown(
+ celeborn_ffi_handle* handle,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(handle, err_out);
+ return guarded(err_out, [&] { as_impl(handle)->client->shutdown(); });
+}
+
+celeborn_ffi_status celeborn_ffi_push_data(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t map_id,
+ int32_t attempt_id,
+ int32_t partition_id,
+ const uint8_t* data,
+ size_t data_len,
+ int32_t num_mappers,
+ int32_t num_partitions,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(handle, err_out);
+ if (data == nullptr && data_len > 0) {
+ set_error(err_out, "null pointer for argument 'data'");
+ return CELEBORN_FFI_ERROR;
+ }
+ return guarded(err_out, [&] {
+ as_impl(handle)->client->pushData(
+ shuffle_id,
+ map_id,
+ attempt_id,
+ partition_id,
+ data,
+ 0,
+ static_cast<int>(data_len),
+ num_mappers,
+ num_partitions);
+ });
+}
+
+celeborn_ffi_status celeborn_ffi_mapper_end(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t map_id,
+ int32_t attempt_id,
+ int32_t num_mappers,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(handle, err_out);
+ return guarded(err_out, [&] {
+ as_impl(handle)->client->mapperEnd(
+ shuffle_id, map_id, attempt_id, num_mappers);
+ });
+}
+
+celeborn_ffi_status celeborn_ffi_update_reducer_file_group(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(handle, err_out);
+ return guarded(err_out, [&] {
+ as_impl(handle)->client->updateReducerFileGroup(shuffle_id);
+ });
+}
+
+celeborn_ffi_status celeborn_ffi_read_partition_full(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t partition_id,
+ int32_t attempt_number,
+ int32_t start_map_index,
+ int32_t end_map_index,
+ uint8_t** data_out,
+ size_t* len_out,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(handle, err_out);
+ CELEBORN_FFI_REQUIRE_NON_NULL(data_out, err_out);
+ CELEBORN_FFI_REQUIRE_NON_NULL(len_out, err_out);
+ return guarded(err_out, [&] {
+ auto stream = as_impl(handle)->client->readPartition(
+ shuffle_id,
+ partition_id,
+ attempt_number,
+ start_map_index,
+ end_map_index);
+
+ constexpr size_t kReadBufSize = 64 * 1024;
+ std::vector<uint8_t> accumulated;
+ accumulated.reserve(kReadBufSize);
+ std::vector<uint8_t> buf(kReadBufSize);
+
+ while (true) {
+ int n = stream->read(buf.data(), 0, buf.size());
+ if (n == -1) {
+ break;
+ }
+ if (n <= 0) {
+ throw std::runtime_error(
+ "CelebornInputStream::read returned unexpected non-positive " +
+ std::to_string(n));
+ }
+ accumulated.insert(accumulated.end(), buf.data(), buf.data() + n);
+ }
+
+ auto* out = new uint8_t[accumulated.size()];
+ std::memcpy(out, accumulated.data(), accumulated.size());
+ *data_out = out;
+ *len_out = accumulated.size();
+ });
+}
+
+celeborn_ffi_status celeborn_ffi_open_partition_reader(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t partition_id,
+ int32_t attempt_number,
+ int32_t start_map_index,
+ int32_t end_map_index,
+ celeborn_ffi_partition_reader** reader_out,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(handle, err_out);
+ CELEBORN_FFI_REQUIRE_NON_NULL(reader_out, err_out);
+ return guarded(err_out, [&] {
+ auto reader = std::make_unique<PartitionReaderImpl>();
+ reader->stream = as_impl(handle)->client->readPartition(
+ shuffle_id,
+ partition_id,
+ attempt_number,
+ start_map_index,
+ end_map_index);
+ *reader_out =
+ reinterpret_cast<celeborn_ffi_partition_reader*>(reader.release());
+ });
+}
+
+celeborn_ffi_status celeborn_ffi_read_partition_chunk(
+ celeborn_ffi_partition_reader* reader,
+ uint8_t* buf,
+ size_t buf_len,
+ size_t* bytes_read,
+ char** err_out) {
+ CELEBORN_FFI_REQUIRE_NON_NULL(reader, err_out);
+ CELEBORN_FFI_REQUIRE_NON_NULL(bytes_read, err_out);
+ if (buf == nullptr && buf_len > 0) {
+ set_error(err_out, "null pointer for argument 'buf'");
+ return CELEBORN_FFI_ERROR;
+ }
+ return guarded(err_out, [&] {
+ if (buf_len == 0) {
+ *bytes_read = 0;
+ return;
+ }
+ int n = as_reader_impl(reader)->stream->read(buf, 0, buf_len);
+ if (n == -1) {
+ // EOF — surface as 0 bytes read to match std::io::Read semantics.
+ *bytes_read = 0;
+ return;
+ }
+ if (n < 0) {
+ throw std::runtime_error(
+ "CelebornInputStream::read returned unexpected negative " +
+ std::to_string(n));
+ }
+ *bytes_read = static_cast<size_t>(n);
+ });
+}
+
+void celeborn_ffi_close_partition_reader(
+ celeborn_ffi_partition_reader* reader) {
+ delete as_reader_impl(reader);
+}
+
+} // extern "C"
diff --git a/cpp/celeborn/ffi/CelebornFfi.h b/cpp/celeborn/ffi/CelebornFfi.h
new file mode 100644
index 000000000..e686113ba
--- /dev/null
+++ b/cpp/celeborn/ffi/CelebornFfi.h
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+// Pure C ABI shim around the celeborn C++ shuffle client. Lives inside
+// libceleborn_client.{so,dylib} so downstream language bindings (Rust /
+// Python via ctypes / etc.) only need to link the single shared library
+// and never pull folly / protobuf / glog / abseil headers themselves.
+#ifndef CELEBORN_FFI_WRAPPER_H
+#define CELEBORN_FFI_WRAPPER_H
+
+#include <stddef.h>
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// Opaque handle. Allocated by celeborn_ffi_create_client; intentionally
+// never freed (folly's IOThreadPoolExecutor::join() races with
+// TransportClient teardown — calling the C++ destructor produces SIGSEGV).
+typedef struct celeborn_ffi_handle celeborn_ffi_handle;
+
+// Opaque handle for a single open partition reader. Allocated by
+// celeborn_ffi_open_partition_reader; released by
+// celeborn_ffi_close_partition_reader. Holds a CelebornInputStream that
+// references the owning client, so it must be closed before its
+// celeborn_ffi_handle is shut down.
+typedef struct celeborn_ffi_partition_reader celeborn_ffi_partition_reader;
+
+// Status codes. Every fallible function returns CELEBORN_FFI_OK on success
+// and writes a heap-allocated, NUL-terminated message into *err_out on
+// failure (caller must release with celeborn_ffi_free_error).
+typedef int32_t celeborn_ffi_status;
+#define CELEBORN_FFI_OK 0
+#define CELEBORN_FFI_ERROR 1
+
+void celeborn_ffi_free_error(char* err);
+
+// Releases a buffer returned via celeborn_ffi_read_partition_full.
+void celeborn_ffi_free_buffer(uint8_t* data);
+
+// Returns NULL on failure; in that case *err_out is set.
+celeborn_ffi_handle* celeborn_ffi_create_client(
+ const char* app_id,
+ size_t app_id_len,
+ int32_t push_buffer_max_size,
+ const char* codec,
+ size_t codec_len,
+ char** err_out);
+
+celeborn_ffi_status celeborn_ffi_setup_lifecycle_manager(
+ celeborn_ffi_handle* handle,
+ const char* host,
+ size_t host_len,
+ int32_t port,
+ char** err_out);
+
+celeborn_ffi_status celeborn_ffi_shutdown(
+ celeborn_ffi_handle* handle,
+ char** err_out);
+
+celeborn_ffi_status celeborn_ffi_push_data(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t map_id,
+ int32_t attempt_id,
+ int32_t partition_id,
+ const uint8_t* data,
+ size_t data_len,
+ int32_t num_mappers,
+ int32_t num_partitions,
+ char** err_out);
+
+celeborn_ffi_status celeborn_ffi_mapper_end(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t map_id,
+ int32_t attempt_id,
+ int32_t num_mappers,
+ char** err_out);
+
+celeborn_ffi_status celeborn_ffi_update_reducer_file_group(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ char** err_out);
+
+// On success, *data_out is a heap buffer (free with celeborn_ffi_free_buffer)
+// and *len_out is its byte length.
+celeborn_ffi_status celeborn_ffi_read_partition_full(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t partition_id,
+ int32_t attempt_number,
+ int32_t start_map_index,
+ int32_t end_map_index,
+ uint8_t** data_out,
+ size_t* len_out,
+ char** err_out);
+
+// On success, *reader_out is a heap-allocated reader that must eventually
+// be released with celeborn_ffi_close_partition_reader.
+celeborn_ffi_status celeborn_ffi_open_partition_reader(
+ celeborn_ffi_handle* handle,
+ int32_t shuffle_id,
+ int32_t partition_id,
+ int32_t attempt_number,
+ int32_t start_map_index,
+ int32_t end_map_index,
+ celeborn_ffi_partition_reader** reader_out,
+ char** err_out);
+
+// Reads up to buf_len bytes into buf. *bytes_read is set to the number of
+// bytes actually read; 0 indicates EOF (matching Rust std::io::Read).
+celeborn_ffi_status celeborn_ffi_read_partition_chunk(
+ celeborn_ffi_partition_reader* reader,
+ uint8_t* buf,
+ size_t buf_len,
+ size_t* bytes_read,
+ char** err_out);
+
+void celeborn_ffi_close_partition_reader(celeborn_ffi_partition_reader*
reader);
+
+#ifdef __cplusplus
+} // extern "C"
+#endif
+
+#endif // CELEBORN_FFI_WRAPPER_H
diff --git a/cpp/dummy.cc b/cpp/dummy.cc
new file mode 100644
index 000000000..ec4d1b542
--- /dev/null
+++ b/cpp/dummy.cc
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+// Placeholder TU for the aggregate `celeborn_client` SHARED target.
+// All real code is brought in via -Wl,-force_load / --whole-archive of the
+// internal static libraries (celeborn_ffi / client / protocol / network /
+// proto / memory / conf / utils). CMake still requires at least one source
+// file to create the shared library target.
+namespace celeborn_ffi {
+extern "C" const char* celeborn_client_build_tag() {
+ return "celeborn-client-shared";
+}
+} // namespace celeborn_ffi
diff --git a/cpp/exports.map b/cpp/exports.map
new file mode 100644
index 000000000..0db2907af
--- /dev/null
+++ b/cpp/exports.map
@@ -0,0 +1,26 @@
+/* Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+{
+ global:
+ extern "C++" {
+ celeborn::*;
+ };
+ celeborn_*;
+ celeborn_ffi_*;
+ local:
+ *;
+};
diff --git a/cpp/celeborn/CMakeLists.txt b/cpp/exports.txt
similarity index 69%
copy from cpp/celeborn/CMakeLists.txt
copy to cpp/exports.txt
index 1588488a7..0e3b1a549 100644
--- a/cpp/celeborn/CMakeLists.txt
+++ b/cpp/exports.txt
@@ -12,14 +12,10 @@
# 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.
-add_subdirectory(proto)
-add_subdirectory(memory)
-add_subdirectory(utils)
-add_subdirectory(conf)
-add_subdirectory(protocol)
-add_subdirectory(network)
-add_subdirectory(client)
-if(CELEBORN_BUILD_TESTS)
- add_subdirectory(tests)
-endif()
+# macOS ld64 -exported_symbols_list patterns. Mangled C++ names of
+# celeborn::* symbols contain the substring "celeborn", and the C ABI
+# shim exports celeborn_ffi_* symbols as plain C (which on Mach-O become
+# _celeborn_ffi_*). The single leading-glob pattern below matches both
+# (celeborn_ffi_* is a subset of *celeborn*).
+*celeborn*
diff --git a/lifecycle-manager/pom.xml b/lifecycle-manager/pom.xml
new file mode 100644
index 000000000..68177c3e4
--- /dev/null
+++ b/lifecycle-manager/pom.xml
@@ -0,0 +1,68 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You 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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>celeborn-lifecycle-manager_${scala.binary.version}</artifactId>
+ <packaging>jar</packaging>
+ <name>Celeborn Lifecycle Manager</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-service_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-client_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-common_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_${scala.binary.version}</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/lifecycle-manager/src/main/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemon.scala
b/lifecycle-manager/src/main/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemon.scala
new file mode 100644
index 000000000..70abca62d
--- /dev/null
+++
b/lifecycle-manager/src/main/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemon.scala
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package org.apache.celeborn.server.lifecyclemanager
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.atomic.AtomicReference
+
+import org.apache.celeborn.client.LifecycleManager
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.util.{SignalUtils, Utils}
+
+object LifecycleManagerDaemon extends Logging {
+
+ private[lifecyclemanager] val shutdownLatch: CountDownLatch = new
CountDownLatch(1)
+
+ private[lifecyclemanager] val currentInstance:
AtomicReference[LifecycleManager] =
+ new AtomicReference[LifecycleManager]()
+
+ private[lifecyclemanager] var exitFn: Int => Unit =
+ (code: Int) => System.exit(code)
+
+ def main(args: Array[String]): Unit = {
+ SignalUtils.registerLogger(log)
+
+ val parsedArgs = LifecycleManagerDaemonArguments.parseOrExit(args)
+ val conf = new CelebornConf()
+
+ // Load properties file before applying CLI args
+ Utils.loadDefaultCelebornProperties(conf, parsedArgs.propertiesFile.orNull)
+
+ applyArgsToConf(parsedArgs, conf)
+
+ // Auth check: standalone LM does not support auth (cpp/Rust client lacks
SASL).
+ //
+ // DEPLOYMENT WARNING: with auth disabled, this daemon exposes shuffle
+ // registration and slot allocation to anything that can reach its RPC
port.
+ // It performs no authentication of incoming clients. Operators MUST bind
it
+ // to a trusted network only (e.g. set --host to a private interface and
+ // restrict the RPC port via firewall / security groups / network policy);
+ // never expose the port on an untrusted or public network.
+ if (conf.authEnabledOnClient) {
+ logError(
+ "Standalone LifecycleManager does not support auth " +
+ "(cpp/Rust client lacks SASL); set celeborn.auth.enabled=false")
+ exitFn(1)
+ return
+ }
+
+ logWarning(
+ "Standalone LifecycleManager runs WITHOUT authentication. Ensure its RPC
" +
+ "port is reachable only from trusted networks (bind to a private " +
+ "interface and restrict access via firewall / network policy).")
+
+ // Propagate --host to Utils so LifecycleManager binds to the requested
hostname
+ parsedArgs.host.foreach { host =>
+ logInfo(s"Setting custom hostname from --host: $host")
+ Utils.setCustomHostname(host)
+ }
+
+ logInfo(s"Parsed args: appId=${parsedArgs.appId}, port=${parsedArgs.port},
" +
+ s"masterEndpoints=${parsedArgs.masterEndpoints}")
+
+ try {
+ val lm = new LifecycleManager(parsedArgs.appId, conf)
+ currentInstance.set(lm)
+
+ installShutdownHook(conf)
+
+ // scalastyle:off println
+ println(s"LifecycleManager bound at ${lm.getHost}:${lm.getPort}")
+ // scalastyle:on println
+
+ logInfo("shutdown hook installed; press Ctrl-C to stop.")
+
+ // Block until the shutdown hook fires (Ctrl-C / SIGTERM) and counts the
+ // latch down. The hook is what drives the actual JVM exit, so there is
no
+ // need for an explicit exitFn(0) here — calling System.exit again from
the
+ // main thread while a shutdown hook is already running is redundant.
+ shutdownLatch.await()
+ } catch {
+ case e: Exception =>
+ logError("Failed to start LifecycleManager", e)
+ exitFn(1)
+ }
+ }
+
+ private[lifecyclemanager] def applyArgsToConf(
+ args: LifecycleManagerDaemonArguments,
+ conf: CelebornConf): Unit = {
+ conf.set(CelebornConf.MASTER_ENDPOINTS.key, args.masterEndpoints)
+ conf.set(CelebornConf.CLIENT_SHUFFLE_MANAGER_PORT.key, args.port.toString)
+ }
+
+ private def installShutdownHook(conf: CelebornConf): Unit = {
+ val shutdownTimeoutMs = conf.appHeartbeatTimeoutMs / 2
+
+ // Watchdog: force halt if shutdown takes too long
+ val watchdog = new Thread("celeborn-lm-shutdown-watchdog") {
+ override def run(): Unit = {
+ try {
+ Thread.sleep(shutdownTimeoutMs)
+ } catch {
+ // Nothing interrupts this watchdog today, but if some future caller
+ // ever does, treat the interruption itself as a sign that shutdown
is
+ // wedged and force a halt rather than silently exiting.
+ case _: InterruptedException =>
+ Thread.currentThread().interrupt()
+ }
+ logError(s"Shutdown exceeded ${shutdownTimeoutMs}ms, forcing halt")
+ Runtime.getRuntime.halt(2)
+ }
+ }
+ watchdog.setDaemon(true)
+
+ Runtime.getRuntime.addShutdownHook(new Thread("celeborn-lm-shutdown") {
+ override def run(): Unit = {
+ watchdog.start()
+ val lm = currentInstance.get()
+ if (lm != null) {
+ try {
+ lm.stop()
+ } catch {
+ case t: Throwable => logError("lm.stop() failed", t)
+ }
+ }
+ shutdownLatch.countDown()
+ }
+ })
+ }
+}
diff --git
a/lifecycle-manager/src/main/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemonArguments.scala
b/lifecycle-manager/src/main/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemonArguments.scala
new file mode 100644
index 000000000..3b3286ef1
--- /dev/null
+++
b/lifecycle-manager/src/main/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemonArguments.scala
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package org.apache.celeborn.server.lifecyclemanager
+
+import scala.annotation.tailrec
+
+import org.apache.celeborn.common.util.IntParam
+
+private[lifecyclemanager] case class LifecycleManagerDaemonArguments(
+ appId: String,
+ masterEndpoints: String,
+ port: Int,
+ host: Option[String],
+ propertiesFile: Option[String])
+
+/**
+ * Signals that argument parsing requested termination. Carries the intended
+ * process exit code so callers (e.g. `main`) can exit, while unit tests can
+ * assert on the code and message without the JVM actually shutting down.
+ *
+ * `exitCode == 0` denotes a successful, user-requested `--help`.
+ */
+private[lifecyclemanager] class ArgumentParseException(
+ val exitCode: Int,
+ message: String) extends RuntimeException(message)
+
+private[lifecyclemanager] object LifecycleManagerDaemonArguments {
+
+ private val MIN_USER_PORT = 1024
+
+ /**
+ * Pure parser: validates `args` and either returns the parsed arguments or
+ * throws [[ArgumentParseException]]. It performs no I/O and never calls
+ * `sys.exit`, so every branch (help / unknown / missing-arg / bad-port) is
+ * unit-testable. Use [[parseOrExit]] for the process entry point.
+ */
+ def parse(args: Array[String]): LifecycleManagerDaemonArguments = {
+ var appId: Option[String] = None
+ var masterEndpoints: Option[String] = None
+ var port: Option[Int] = None
+ var host: Option[String] = None
+ var propertiesFile: Option[String] = None
+
+ @tailrec
+ def doParse(remaining: List[String]): Unit = remaining match {
+ case "--app-id" :: value :: tail =>
+ appId = Some(value)
+ doParse(tail)
+
+ case "--master-endpoints" :: value :: tail =>
+ masterEndpoints = Some(value)
+ doParse(tail)
+
+ case ("--port" | "-p") :: IntParam(value) :: tail =>
+ port = Some(value)
+ doParse(tail)
+
+ // Only the long form is accepted for host: `-h` conventionally means
+ // help, so reserving it for --host would be surprising.
+ case "--host" :: value :: tail =>
+ host = Some(value)
+ doParse(tail)
+
+ case "--properties-file" :: value :: tail =>
+ propertiesFile = Some(value)
+ doParse(tail)
+
+ case ("--help" | "-h") :: _ =>
+ throw new ArgumentParseException(0, usage)
+
+ case Nil => // done
+
+ case unknown :: _ =>
+ throw new ArgumentParseException(1, s"Unknown argument:
$unknown\n$usage")
+ }
+
+ doParse(args.toList)
+
+ if (appId.isEmpty) {
+ throw new ArgumentParseException(1, s"Error: --app-id is
required.\n$usage")
+ }
+ if (masterEndpoints.isEmpty) {
+ throw new ArgumentParseException(1, s"Error: --master-endpoints is
required.\n$usage")
+ }
+ if (port.isEmpty) {
+ throw new ArgumentParseException(1, s"Error: --port is
required.\n$usage")
+ }
+ if (port.get < MIN_USER_PORT) {
+ throw new ArgumentParseException(
+ 1,
+ s"Error: --port must be >= $MIN_USER_PORT, got ${port.get}.\n$usage")
+ }
+
+ LifecycleManagerDaemonArguments(
+ appId = appId.get,
+ masterEndpoints = masterEndpoints.get,
+ port = port.get,
+ host = host,
+ propertiesFile = propertiesFile)
+ }
+
+ /**
+ * Process entry point wrapper around [[parse]]: prints the message carried
by
+ * an [[ArgumentParseException]] and exits with its code.
+ */
+ def parseOrExit(args: Array[String]): LifecycleManagerDaemonArguments = {
+ try {
+ parse(args)
+ } catch {
+ case e: ArgumentParseException =>
+ // scalastyle:off println
+ System.err.println(e.getMessage)
+ // scalastyle:on println
+ sys.exit(e.exitCode)
+ }
+ }
+
+ val usage: String =
+ """Usage: LifecycleManagerDaemon [options]
+ |
+ |Options:
+ | --app-id ID Application unique identifier (required)
+ | --master-endpoints ENDPOINTS Comma-separated master host:port list
(required)
+ | -p PORT, --port PORT Port for LifecycleManager to listen on
(required, >= 1024)
+ | --host HOST Hostname to bind (optional, default:
auto-detect)
+ | --properties-file FILE Path to a custom Celeborn properties
file,
+ | default is conf/celeborn-defaults.conf
+ | -h, --help Print this help message
+ |""".stripMargin
+}
diff --git
a/lifecycle-manager/src/test/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemonArgumentsSuite.scala
b/lifecycle-manager/src/test/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemonArgumentsSuite.scala
new file mode 100644
index 000000000..bd2cd0b2b
--- /dev/null
+++
b/lifecycle-manager/src/test/scala/org/apache/celeborn/server/lifecyclemanager/LifecycleManagerDaemonArgumentsSuite.scala
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package org.apache.celeborn.server.lifecyclemanager
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
+
+class LifecycleManagerDaemonArgumentsSuite extends AnyFunSuite with Logging {
+
+ test("parse all required arguments") {
+ val args = Array(
+ "--app-id",
+ "test-app-1",
+ "--master-endpoints",
+ "host1:9097,host2:9097",
+ "--port",
+ "39099")
+ val parsed = LifecycleManagerDaemonArguments.parse(args)
+ assert(parsed.appId === "test-app-1")
+ assert(parsed.masterEndpoints === "host1:9097,host2:9097")
+ assert(parsed.port === 39099)
+ assert(parsed.host.isEmpty)
+ assert(parsed.propertiesFile.isEmpty)
+ }
+
+ test("parse all arguments including optional ones") {
+ val args = Array(
+ "--app-id",
+ "my-app",
+ "--master-endpoints",
+ "localhost:9097",
+ "--port",
+ "40000",
+ "--host",
+ "my-host",
+ "--properties-file",
+ "/tmp/celeborn.conf")
+ val parsed = LifecycleManagerDaemonArguments.parse(args)
+ assert(parsed.appId === "my-app")
+ assert(parsed.masterEndpoints === "localhost:9097")
+ assert(parsed.port === 40000)
+ assert(parsed.host === Some("my-host"))
+ assert(parsed.propertiesFile === Some("/tmp/celeborn.conf"))
+ }
+
+ test("parse with short port flag -p") {
+ val args = Array(
+ "--app-id",
+ "short-app",
+ "--master-endpoints",
+ "host:9097",
+ "-p",
+ "2048")
+ val parsed = LifecycleManagerDaemonArguments.parse(args)
+ assert(parsed.appId === "short-app")
+ assert(parsed.port === 2048)
+ assert(parsed.host.isEmpty)
+ }
+
+ test("-h is treated as help, not host") {
+ val args = Array("-h")
+ val ex = intercept[ArgumentParseException] {
+ LifecycleManagerDaemonArguments.parse(args)
+ }
+ assert(ex.exitCode === 0)
+ assert(ex.getMessage.contains("Usage"))
+ }
+
+ test("--help requests help with exit code 0") {
+ val ex = intercept[ArgumentParseException] {
+ LifecycleManagerDaemonArguments.parse(Array("--help"))
+ }
+ assert(ex.exitCode === 0)
+ assert(ex.getMessage.contains("Usage"))
+ }
+
+ test("unknown argument fails with exit code 1") {
+ val args = Array(
+ "--app-id",
+ "app",
+ "--master-endpoints",
+ "host:9097",
+ "--port",
+ "39099",
+ "--bogus")
+ val ex = intercept[ArgumentParseException] {
+ LifecycleManagerDaemonArguments.parse(args)
+ }
+ assert(ex.exitCode === 1)
+ assert(ex.getMessage.contains("Unknown argument: --bogus"))
+ }
+
+ test("missing --app-id fails with exit code 1") {
+ val args = Array("--master-endpoints", "host:9097", "--port", "39099")
+ val ex = intercept[ArgumentParseException] {
+ LifecycleManagerDaemonArguments.parse(args)
+ }
+ assert(ex.exitCode === 1)
+ assert(ex.getMessage.contains("--app-id is required"))
+ }
+
+ test("missing --master-endpoints fails with exit code 1") {
+ val args = Array("--app-id", "app", "--port", "39099")
+ val ex = intercept[ArgumentParseException] {
+ LifecycleManagerDaemonArguments.parse(args)
+ }
+ assert(ex.exitCode === 1)
+ assert(ex.getMessage.contains("--master-endpoints is required"))
+ }
+
+ test("missing --port fails with exit code 1") {
+ val args = Array("--app-id", "app", "--master-endpoints", "host:9097")
+ val ex = intercept[ArgumentParseException] {
+ LifecycleManagerDaemonArguments.parse(args)
+ }
+ assert(ex.exitCode === 1)
+ assert(ex.getMessage.contains("--port is required"))
+ }
+
+ test("port below 1024 fails with exit code 1") {
+ val args = Array(
+ "--app-id",
+ "app",
+ "--master-endpoints",
+ "host:9097",
+ "--port",
+ "1023")
+ val ex = intercept[ArgumentParseException] {
+ LifecycleManagerDaemonArguments.parse(args)
+ }
+ assert(ex.exitCode === 1)
+ assert(ex.getMessage.contains("must be >= 1024"))
+ }
+
+ test("applyArgsToConf sets master endpoints and shuffle manager port") {
+ val parsed = LifecycleManagerDaemonArguments(
+ appId = "app",
+ masterEndpoints = "host1:9097,host2:9097",
+ port = 39099,
+ host = None,
+ propertiesFile = None)
+ val conf = new CelebornConf()
+ LifecycleManagerDaemon.applyArgsToConf(parsed, conf)
+ assert(conf.get(CelebornConf.MASTER_ENDPOINTS.key) ===
"host1:9097,host2:9097")
+ assert(conf.get(CelebornConf.CLIENT_SHUFFLE_MANAGER_PORT.key) === "39099")
+ }
+
+ test("parse minimum valid port 1024") {
+ val args = Array(
+ "--app-id",
+ "app",
+ "--master-endpoints",
+ "host:9097",
+ "--port",
+ "1024")
+ val parsed = LifecycleManagerDaemonArguments.parse(args)
+ assert(parsed.port === 1024)
+ }
+
+ test("usage string contains all options") {
+ val usageText = LifecycleManagerDaemonArguments.usage
+ assert(usageText.contains("--app-id"))
+ assert(usageText.contains("--master-endpoints"))
+ assert(usageText.contains("--port"))
+ assert(usageText.contains("--host"))
+ assert(usageText.contains("--properties-file"))
+ }
+}
diff --git a/pom.xml b/pom.xml
index 04622055e..042cfa8e8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -38,6 +38,7 @@
<module>service</module>
<module>master</module>
<module>worker</module>
+ <module>lifecycle-manager</module>
<module>cli</module>
</modules>
diff --git a/project/CelebornBuild.scala b/project/CelebornBuild.scala
index e369ef80e..571134e30 100644
--- a/project/CelebornBuild.scala
+++ b/project/CelebornBuild.scala
@@ -471,6 +471,7 @@ object CelebornBuild extends sbt.internal.BuildDef {
CelebornService.service,
CelebornWorker.worker,
CelebornMaster.master,
+ CelebornLifecycleManager.lifecycleManager,
CelebornCli.cli
) ++ maybeSparkClientModules ++
maybeFlinkClientModules ++
@@ -598,6 +599,19 @@ object Utils {
}
}
+object CelebornLifecycleManager {
+ lazy val lifecycleManager = Project("celeborn-lifecycle-manager",
file("lifecycle-manager"))
+ .dependsOn(CelebornService.service % "test->test;compile->compile")
+ .dependsOn(CelebornClient.client % "test->test;compile->compile")
+ .dependsOn(CelebornCommon.common % "test->test;compile->compile")
+ .settings (
+ commonSettings,
+ libraryDependencies ++= Seq(
+ Dependencies.scalatestMockito % "test"
+ ) ++ commonUnitTestDependencies
+ )
+}
+
object CelebornCli {
lazy val cli = Project("celeborn-cli", file("cli"))
.dependsOn(CelebornCommon.common % "test->test;compile->compile")
diff --git a/rust/.gitignore b/rust/.gitignore
new file mode 100644
index 000000000..4a10facca
--- /dev/null
+++ b/rust/.gitignore
@@ -0,0 +1,7 @@
+target/
+Cargo.lock
+
+# Prebuilt celeborn-client shared libraries — built by cpp/, not tracked.
+resource/lib/**/*.so
+resource/lib/**/*.dylib
+resource/lib/**/*.dll
diff --git a/cpp/celeborn/CMakeLists.txt b/rust/Cargo.toml
similarity index 76%
copy from cpp/celeborn/CMakeLists.txt
copy to rust/Cargo.toml
index 1588488a7..8a966b4df 100644
--- a/cpp/celeborn/CMakeLists.txt
+++ b/rust/Cargo.toml
@@ -12,14 +12,10 @@
# 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.
-add_subdirectory(proto)
-add_subdirectory(memory)
-add_subdirectory(utils)
-add_subdirectory(conf)
-add_subdirectory(protocol)
-add_subdirectory(network)
-add_subdirectory(client)
-if(CELEBORN_BUILD_TESTS)
- add_subdirectory(tests)
-endif()
+[workspace]
+members = [
+ "celeborn-client-sys",
+ "celeborn-client",
+]
+resolver = "2"
diff --git a/cpp/celeborn/CMakeLists.txt b/rust/celeborn-client-sys/Cargo.toml
similarity index 76%
copy from cpp/celeborn/CMakeLists.txt
copy to rust/celeborn-client-sys/Cargo.toml
index 1588488a7..8e7a14993 100644
--- a/cpp/celeborn/CMakeLists.txt
+++ b/rust/celeborn-client-sys/Cargo.toml
@@ -12,14 +12,11 @@
# 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.
-add_subdirectory(proto)
-add_subdirectory(memory)
-add_subdirectory(utils)
-add_subdirectory(conf)
-add_subdirectory(protocol)
-add_subdirectory(network)
-add_subdirectory(client)
-if(CELEBORN_BUILD_TESTS)
- add_subdirectory(tests)
-endif()
+[package]
+name = "celeborn-client-sys"
+version = "0.1.0"
+edition = "2021"
+description = "Raw FFI bindings to libceleborn_client (single-dylib C ABI)"
+publish = false
+links = "celeborn_client"
diff --git a/rust/celeborn-client-sys/build.rs
b/rust/celeborn-client-sys/build.rs
new file mode 100644
index 000000000..fecb321f3
--- /dev/null
+++ b/rust/celeborn-client-sys/build.rs
@@ -0,0 +1,186 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+use std::path::{Path, PathBuf};
+use std::process::Command;
+
+/// Resolve the directory containing `libceleborn_client.{so,dylib}`.
+///
+/// Lookup order (first match wins):
+/// 1. `CELEBORN_CPP_PREFIX` env var → use `<prefix>/lib`. Intended for CI /
+/// custom installs.
+/// 2. In-repo prebuilt artifact at
+/// `rust/resource/lib/<target-triple>/libceleborn_client.{so,dylib}`.
+/// Drop the matching dylib in and `cargo build` picks it up with no
+/// extra environment variable.
+/// 3. Fall back to driving `cmake` against the in-repo `cpp/` source tree
+/// and installing into `$OUT_DIR/celeborn-cpp-install/lib`.
+fn resolve_lib_dir() -> PathBuf {
+ let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap();
+ let lib_filename = match target_os.as_str() {
+ "macos" => "libceleborn_client.dylib",
+ "linux" => "libceleborn_client.so",
+ other => panic!("unsupported target_os: {other} (only linux/macos
supported)"),
+ };
+
+ // 1. Explicit override.
+ if let Ok(prefix) = std::env::var("CELEBORN_CPP_PREFIX") {
+ let lib_dir = PathBuf::from(&prefix).join("lib");
+ if !lib_dir.join(lib_filename).exists() {
+ panic!(
+ "CELEBORN_CPP_PREFIX={prefix} but {} does not exist.",
+ lib_dir.join(lib_filename).display()
+ );
+ }
+ eprintln!("cargo:warning=Using prebuilt Celeborn dylib from {prefix}");
+ return lib_dir;
+ }
+
+ // 2. In-repo prebuilt: rust/resource/lib/<target>/libceleborn_client.<ext>
+ let manifest_dir =
PathBuf::from(std::env::var("CARGO_MANIFEST_DIR").unwrap());
+ let target = std::env::var("TARGET").unwrap();
+ let resource_lib_dir = manifest_dir
+ .join("../resource/lib")
+ .join(&target);
+ if resource_lib_dir.join(lib_filename).exists() {
+ eprintln!(
+ "cargo:warning=Using in-repo prebuilt Celeborn dylib at {}",
+ resource_lib_dir.display()
+ );
+ return resource_lib_dir
+ .canonicalize()
+ .expect("failed to canonicalize resource lib dir");
+ }
+
+ // 3. Fall back to cmake from source.
+ let cpp_source_dir = manifest_dir
+ .join("../../cpp")
+ .canonicalize()
+ .unwrap_or_else(|_| {
+ panic!(
+ "No prebuilt {lib_filename} found at {}, CELEBORN_CPP_PREFIX
is unset, \
+ and the in-repo cpp/ directory is not reachable from {}.",
+ resource_lib_dir.display(),
+ manifest_dir.display(),
+ )
+ });
+
+ eprintln!(
+ "cargo:warning=No prebuilt dylib at {}; building Celeborn C++ from
source at {}",
+ resource_lib_dir.display(),
+ cpp_source_dir.display()
+ );
+
+ cmake_build_cpp(&cpp_source_dir).join("lib")
+}
+
+fn cmake_build_cpp(source_dir: &Path) -> PathBuf {
+ let out_dir = PathBuf::from(std::env::var("OUT_DIR").unwrap());
+ let build_dir = out_dir.join("celeborn-cpp-build");
+ let install_dir = out_dir.join("celeborn-cpp-install");
+
+ std::fs::create_dir_all(&build_dir).expect("failed to create cmake build
directory");
+ std::fs::create_dir_all(&install_dir).expect("failed to create cmake
install directory");
+
+ let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap();
+
+ let mut configure_cmd = Command::new("cmake");
+ configure_cmd
+ .current_dir(&build_dir)
+ .arg(source_dir)
+ .arg(format!("-DCMAKE_INSTALL_PREFIX={}", install_dir.display()))
+ .arg("-DCMAKE_BUILD_TYPE=Release")
+ .arg("-DCELEBORN_BUILD_TESTS=OFF");
+
+ if target_os == "macos" {
+ let homebrew_prefix = std::env::var("HOMEBREW_PREFIX")
+ .unwrap_or_else(|_| "/opt/homebrew".to_string());
+ configure_cmd.arg(format!("-DCMAKE_PREFIX_PATH={homebrew_prefix}"));
+ configure_cmd.env(
+ "OPENSSL_ROOT_DIR",
+ format!("{homebrew_prefix}/opt/openssl@3"),
+ );
+ }
+
+ let configure_status = configure_cmd
+ .status()
+ .expect("failed to execute `cmake` – is cmake installed?");
+ if !configure_status.success() {
+ panic!("cmake configure step failed (exit code: {configure_status})");
+ }
+
+ let num_jobs = std::env::var("NUM_JOBS").unwrap_or_else(|_|
num_cpus().to_string());
+ let build_status = Command::new("cmake")
+ .current_dir(&build_dir)
+ .args(["--build", "."])
+ .args(["--config", "Release"])
+ .args(["--parallel", &num_jobs])
+ .status()
+ .expect("failed to execute cmake --build");
+ if !build_status.success() {
+ panic!("cmake build step failed (exit code: {build_status})");
+ }
+
+ let install_status = Command::new("cmake")
+ .current_dir(&build_dir)
+ .args(["--install", "."])
+ .status()
+ .expect("failed to execute cmake --install");
+ if !install_status.success() {
+ panic!("cmake install step failed (exit code: {install_status})");
+ }
+
+ install_dir
+}
+
+fn num_cpus() -> usize {
+ std::thread::available_parallelism()
+ .map(|n| n.get())
+ .unwrap_or(4)
+}
+
+fn main() {
+ let lib_dir = resolve_lib_dir();
+ let lib_dir_str = lib_dir.display().to_string();
+
+ // Single aggregated dylib: libceleborn_client.{so,dylib} bundles every
+ // internal static lib (including the celeborn_ffi C ABI shim) and pulls
+ // third-party deps via NEEDED entries that resolve at runtime. The Rust
+ // crate touches no C++ headers and links nothing else.
+ println!("cargo:rustc-link-search=native={lib_dir_str}");
+ println!("cargo:rustc-link-lib=dylib=celeborn_client");
+
+ // Re-export lib_dir so the downstream `celeborn-client` crate can pick
+ // it up via DEP_CELEBORN_CLIENT_LIB_DIR and embed an rpath in its
+ // examples / tests / binaries. (`cargo:rustc-link-arg` does not
+ // propagate from a sys crate to dependent crates' artifacts.)
+ println!("cargo:metadata=lib_dir={lib_dir_str}");
+
+ println!("cargo:rerun-if-env-changed=CELEBORN_CPP_PREFIX");
+ let manifest_dir = std::env::var("CARGO_MANIFEST_DIR").unwrap();
+ println!("cargo:rerun-if-changed={manifest_dir}/../resource/lib");
+
+ // When we fall back to building cpp/ from source (option 3 in
+ // resolve_lib_dir), edits under cpp/ must trigger a rebuild of the
+ // dylib. Declare the dependency unconditionally so that toggling
+ // between prebuilt and from-source modes does not require touching
+ // the build script.
+ let cpp_dir = PathBuf::from(&manifest_dir).join("../../cpp");
+ if cpp_dir.exists() {
+ let cpp_dir_str = cpp_dir.display();
+ println!("cargo:rerun-if-changed={cpp_dir_str}/CMakeLists.txt");
+ println!("cargo:rerun-if-changed={cpp_dir_str}/celeborn");
+ }
+}
diff --git a/rust/celeborn-client-sys/src/lib.rs
b/rust/celeborn-client-sys/src/lib.rs
new file mode 100644
index 000000000..811e67fa7
--- /dev/null
+++ b/rust/celeborn-client-sys/src/lib.rs
@@ -0,0 +1,158 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+//! Raw C ABI bindings for `libceleborn_client.{so,dylib}`.
+//!
+//! All FFI is plain `extern "C"` — no `cxx`, no C++ headers, no template
+//! instantiations leak across the language boundary. The dylib is the sole
+//! external link dependency; folly / protobuf / glog / abseil all stay
+//! hidden inside it.
+
+#![allow(non_camel_case_types, non_upper_case_globals)]
+
+use std::os::raw::c_char;
+
+/// Opaque handle. Returned by [`celeborn_ffi_create_client`].
+///
+/// The handle is *intentionally never freed* — folly's
+/// `IOThreadPoolExecutor::join()` races with `TransportClient` teardown,
+/// so calling the C++ destructor produces SIGSEGV. Process-lifetime use.
+#[repr(C)]
+pub struct celeborn_ffi_handle {
+ _opaque: [u8; 0],
+ _marker: core::marker::PhantomData<(*mut u8, core::marker::PhantomPinned)>,
+}
+
+/// Opaque handle to a single open partition reader. Returned by
+/// [`celeborn_ffi_open_partition_reader`]; released with
+/// [`celeborn_ffi_close_partition_reader`]. Holds a raw pointer into the
+/// owning client, so it must not outlive its [`celeborn_ffi_handle`].
+#[repr(C)]
+pub struct celeborn_ffi_partition_reader {
+ _opaque: [u8; 0],
+ _marker: core::marker::PhantomData<(*mut u8, core::marker::PhantomPinned)>,
+}
+
+pub type celeborn_ffi_status = i32;
+pub const CELEBORN_FFI_OK: celeborn_ffi_status = 0;
+pub const CELEBORN_FFI_ERROR: celeborn_ffi_status = 1;
+
+extern "C" {
+ pub fn celeborn_ffi_free_error(err: *mut c_char);
+ pub fn celeborn_ffi_free_buffer(data: *mut u8);
+
+ pub fn celeborn_ffi_create_client(
+ app_id: *const c_char,
+ app_id_len: usize,
+ push_buffer_max_size: i32,
+ codec: *const c_char,
+ codec_len: usize,
+ err_out: *mut *mut c_char,
+ ) -> *mut celeborn_ffi_handle;
+
+ pub fn celeborn_ffi_setup_lifecycle_manager(
+ handle: *mut celeborn_ffi_handle,
+ host: *const c_char,
+ host_len: usize,
+ port: i32,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ pub fn celeborn_ffi_shutdown(
+ handle: *mut celeborn_ffi_handle,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ pub fn celeborn_ffi_push_data(
+ handle: *mut celeborn_ffi_handle,
+ shuffle_id: i32,
+ map_id: i32,
+ attempt_id: i32,
+ partition_id: i32,
+ data: *const u8,
+ data_len: usize,
+ num_mappers: i32,
+ num_partitions: i32,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ pub fn celeborn_ffi_mapper_end(
+ handle: *mut celeborn_ffi_handle,
+ shuffle_id: i32,
+ map_id: i32,
+ attempt_id: i32,
+ num_mappers: i32,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ pub fn celeborn_ffi_update_reducer_file_group(
+ handle: *mut celeborn_ffi_handle,
+ shuffle_id: i32,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ pub fn celeborn_ffi_read_partition_full(
+ handle: *mut celeborn_ffi_handle,
+ shuffle_id: i32,
+ partition_id: i32,
+ attempt_number: i32,
+ start_map_index: i32,
+ end_map_index: i32,
+ data_out: *mut *mut u8,
+ len_out: *mut usize,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ pub fn celeborn_ffi_open_partition_reader(
+ handle: *mut celeborn_ffi_handle,
+ shuffle_id: i32,
+ partition_id: i32,
+ attempt_number: i32,
+ start_map_index: i32,
+ end_map_index: i32,
+ reader_out: *mut *mut celeborn_ffi_partition_reader,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ /// Reads up to `buf_len` bytes into `buf`. Writes the number of bytes
+ /// actually read into `*bytes_read`; `0` indicates EOF (`std::io::Read`
+ /// semantics).
+ pub fn celeborn_ffi_read_partition_chunk(
+ reader: *mut celeborn_ffi_partition_reader,
+ buf: *mut u8,
+ buf_len: usize,
+ bytes_read: *mut usize,
+ err_out: *mut *mut c_char,
+ ) -> celeborn_ffi_status;
+
+ pub fn celeborn_ffi_close_partition_reader(reader: *mut
celeborn_ffi_partition_reader);
+}
+
+/// Take ownership of a C error string and convert it to an owned `String`.
+///
+/// Releases the heap allocation via [`celeborn_ffi_free_error`].
+///
+/// # Safety
+/// `err` must be either null or a valid pointer returned by a
+/// `celeborn_ffi_*` function via its `err_out` parameter. After calling
+/// this, the pointer is dangling.
+pub unsafe fn take_error(err: *mut c_char) -> Option<String> {
+ if err.is_null() {
+ return None;
+ }
+ let msg = std::ffi::CStr::from_ptr(err).to_string_lossy().into_owned();
+ celeborn_ffi_free_error(err);
+ Some(msg)
+}
diff --git a/.rat-excludes b/rust/celeborn-client/Cargo.toml
similarity index 62%
copy from .rat-excludes
copy to rust/celeborn-client/Cargo.toml
index 22c97119e..1427c650d 100644
--- a/.rat-excludes
+++ b/rust/celeborn-client/Cargo.toml
@@ -1,4 +1,3 @@
-#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
@@ -13,28 +12,26 @@
# 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.
-#
-**/.*/**
-**/*.json
-**/*.prefs
-**/*.log
-**/*.md
-**/*.iml
-**/*.svg
-**/target/**
-**/out/**
-**/spark-warehouse/**
-**/metastore_db/**
-**/licenses/LICENSE*
-**/licenses-binary/LICENSE*
-**/dependency-reduced-pom.xml
-**/scalastyle-output.xml
-NOTICE*
-assets/**
-build/apache-maven-*/**
-build/scala-*/**
-build/sbt-config/**
-**/benchmarks/**
-**/node_modules/**
-cpp/cmake/FindSodium.cmake
+[package]
+name = "celeborn-client"
+version = "0.1.0"
+edition = "2021"
+description = "Rust-friendly wrapper around Apache Celeborn C++ client FFI"
+publish = false
+
+[dependencies]
+celeborn-client-sys = { path = "../celeborn-client-sys" }
+thiserror = "1.0"
+log = "0.4"
+
+[dev-dependencies]
+env_logger = "0.11"
+
+[[example]]
+name = "data_sum_writer"
+path = "../examples/data_sum_writer.rs"
+
+[[example]]
+name = "data_sum_reader"
+path = "../examples/data_sum_reader.rs"
diff --git a/rust/celeborn-client/build.rs b/rust/celeborn-client/build.rs
new file mode 100644
index 000000000..4e41b4ede
--- /dev/null
+++ b/rust/celeborn-client/build.rs
@@ -0,0 +1,28 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+// Re-emit the dylib directory exported by `celeborn-client-sys` (via its
+// `links = "celeborn_client"` metadata) as an rpath on every artifact this
+// crate produces (examples, integration tests, downstream binaries).
+//
+// `cargo:rustc-link-arg=...` from a sys crate's build script does *not*
+// propagate to dependent crates' link lines, so the rpath has to be emitted
+// from the crate that owns those artifacts.
+fn main() {
+ if let Ok(lib_dir) = std::env::var("DEP_CELEBORN_CLIENT_LIB_DIR") {
+ println!("cargo:rustc-link-arg=-Wl,-rpath,{lib_dir}");
+ }
+ println!("cargo:rerun-if-env-changed=DEP_CELEBORN_CLIENT_LIB_DIR");
+}
diff --git a/rust/celeborn-client/src/lib.rs b/rust/celeborn-client/src/lib.rs
new file mode 100644
index 000000000..6b73cee79
--- /dev/null
+++ b/rust/celeborn-client/src/lib.rs
@@ -0,0 +1,478 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+//! Rust-friendly wrapper around `celeborn-client-sys` (raw C ABI bindings
+//! to `libceleborn_client.{so,dylib}`).
+
+use std::marker::PhantomData;
+use std::os::raw::c_char;
+use std::ptr;
+
+use celeborn_client_sys as sys;
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+ #[error("celeborn ffi error: {0}")]
+ Ffi(String),
+ #[error("invalid argument: {0}")]
+ InvalidArg(&'static str),
+ #[error("celeborn ffi returned a null handle")]
+ NullHandle,
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+/// Build an [`Error::Ffi`] from a C error pointer (consumes the heap
+/// allocation). Returns a generic message if the pointer is null.
+unsafe fn ffi_error(err: *mut c_char) -> Error {
+ Error::Ffi(
+ sys::take_error(err)
+ .unwrap_or_else(|| "celeborn ffi returned no error
message".to_string()),
+ )
+}
+
+/// Configuration for connecting to a Celeborn LifecycleManager.
+pub struct Config {
+ pub app_id: String,
+ /// Max push buffer size in bytes. 0 means use cpp default (64kB).
+ pub push_buffer_max_size: i32,
+ /// Compression codec: "NONE", "LZ4", or "ZSTD".
+ pub shuffle_compression_codec: String,
+}
+
+impl Config {
+ pub fn new(app_id: String) -> Self {
+ Self {
+ app_id,
+ push_buffer_max_size: 0,
+ shuffle_compression_codec: "NONE".to_string(),
+ }
+ }
+}
+
+/// A Rust-friendly Celeborn shuffle client backed by the C++ implementation.
+///
+/// The handle is intentionally leaked on `Drop` (after a best-effort
+/// shutdown) to work around a folly `EventBase` use-after-free that
+/// triggers when `TransportClient` is destroyed concurrently with
+/// `IOThreadPoolExecutor::join()`. Use `shutdown()` for explicit teardown.
+///
+/// # Per-process client assumption (IMPORTANT)
+///
+/// Because the underlying C++ `ClientImpl` is **never freed** — neither
+/// `shutdown()` nor `Drop` deletes it, and `connect()` leaks it entirely if
+/// `setup_lifecycle_manager` fails — each call to [`ShuffleClient::connect`]
+/// permanently leaks the client's config, endpoint, native client and its
+/// thread pool. This is acceptable for one-shot processes (the typical mapper
+/// / writer use case) but means a long-lived process that creates **many**
+/// clients will steadily leak memory and threads.
+///
+/// Treat this type as a **per-process singleton**: create one client for the
+/// lifetime of the process and reuse it (share via `Arc<ShuffleClient>` across
+/// threads) rather than repeatedly constructing and dropping clients.
+pub struct ShuffleClient {
+ handle: *mut sys::celeborn_ffi_handle,
+}
+
+// SAFETY: The underlying C++ `ShuffleClientImpl` owns its own thread pool and
+// synchronizes all shared state internally:
+// - All shared maps (partition locations, push states, reducer file groups,
+// mapper-end sets) are `folly`-backed concurrent maps guarded by
+// `Synchronized`.
+// - `registerShuffle` is serialized per-shuffleId with a dedicated mutex
plus
+// double-checked lookup.
+// - A fresh compressor instance is created per `pushData` call, so no
+// non-thread-safe compressor is shared across concurrent operations.
+// `pushData` and `readPartition` are therefore safe to invoke concurrently
from
+// multiple threads on the same client, which is why every method below takes
+// `&self`. The handle is an opaque raw pointer with no Rust-level aliasing
+// concerns, so the client is both `Send` and `Sync` and can be shared via
+// `Arc<ShuffleClient>` for parallel push/read.
+//
+// Note: the `PartitionReader` returned by `open_partition` wraps a single
+// `CelebornInputStream` that holds mutable read state and is NOT thread-safe;
+// it keeps `&mut self` on its `Read` impl so a single reader cannot be read
+// concurrently from multiple threads. Opening multiple independent readers in
+// parallel is fine.
+unsafe impl Send for ShuffleClient {}
+unsafe impl Sync for ShuffleClient {}
+
+/// Pure validation of [`ShuffleClient::connect`] inputs, separated from any
+/// FFI so it can be unit-tested without a live cluster.
+fn validate_connect_args(config: &Config, lm_port: i32) -> Result<()> {
+ if config.app_id.is_empty() {
+ return Err(Error::InvalidArg("app_id is empty"));
+ }
+ if lm_port <= 0 {
+ return Err(Error::InvalidArg("lm_port must be > 0"));
+ }
+ let valid_codecs = ["NONE", "LZ4", "ZSTD"];
+ if !valid_codecs.contains(&config.shuffle_compression_codec.as_str()) {
+ return Err(Error::InvalidArg(
+ "shuffle_compression_codec must be NONE, LZ4, or ZSTD",
+ ));
+ }
+ Ok(())
+}
+
+impl ShuffleClient {
+ /// Connect to a running LifecycleManager at `lm_host:lm_port`.
+ pub fn connect(config: Config, lm_host: &str, lm_port: i32) ->
Result<Self> {
+ validate_connect_args(&config, lm_port)?;
+
+ let mut err: *mut c_char = ptr::null_mut();
+ let handle = unsafe {
+ sys::celeborn_ffi_create_client(
+ config.app_id.as_ptr() as *const c_char,
+ config.app_id.len(),
+ config.push_buffer_max_size,
+ config.shuffle_compression_codec.as_ptr() as *const c_char,
+ config.shuffle_compression_codec.len(),
+ &mut err,
+ )
+ };
+ if handle.is_null() {
+ return Err(unsafe { ffi_error(err) });
+ }
+
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe {
+ sys::celeborn_ffi_setup_lifecycle_manager(
+ handle,
+ lm_host.as_ptr() as *const c_char,
+ lm_host.len(),
+ lm_port,
+ &mut err,
+ )
+ };
+ if status != sys::CELEBORN_FFI_OK {
+ // Intentional leak of `handle`: do not call any destructor that
+ // would tear down the folly EventBase state.
+ return Err(unsafe { ffi_error(err) });
+ }
+
+ Ok(Self { handle })
+ }
+
+ /// Push data for a specific partition.
+ ///
+ /// Takes `&self` so the client can be shared via `Arc<ShuffleClient>` and
+ /// pushed to concurrently from multiple threads; the underlying C++ client
+ /// synchronizes internally.
+ pub fn push_data(
+ &self,
+ shuffle_id: i32,
+ map_id: i32,
+ attempt_id: i32,
+ partition_id: i32,
+ data: &[u8],
+ num_mappers: i32,
+ num_partitions: i32,
+ ) -> Result<()> {
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe {
+ sys::celeborn_ffi_push_data(
+ self.handle,
+ shuffle_id,
+ map_id,
+ attempt_id,
+ partition_id,
+ data.as_ptr(),
+ data.len(),
+ num_mappers,
+ num_partitions,
+ &mut err,
+ )
+ };
+ if status == sys::CELEBORN_FFI_OK {
+ Ok(())
+ } else {
+ Err(unsafe { ffi_error(err) })
+ }
+ }
+
+ /// Signal that a mapper has finished writing all its partitions.
+ pub fn mapper_end(
+ &self,
+ shuffle_id: i32,
+ map_id: i32,
+ attempt_id: i32,
+ num_mappers: i32,
+ ) -> Result<()> {
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe {
+ sys::celeborn_ffi_mapper_end(
+ self.handle,
+ shuffle_id,
+ map_id,
+ attempt_id,
+ num_mappers,
+ &mut err,
+ )
+ };
+ if status == sys::CELEBORN_FFI_OK {
+ Ok(())
+ } else {
+ Err(unsafe { ffi_error(err) })
+ }
+ }
+
+ /// Update reducer file group metadata for a given shuffle.
+ pub fn update_reducer_file_group(&self, shuffle_id: i32) -> Result<()> {
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe {
+ sys::celeborn_ffi_update_reducer_file_group(self.handle,
shuffle_id, &mut err)
+ };
+ if status == sys::CELEBORN_FFI_OK {
+ Ok(())
+ } else {
+ Err(unsafe { ffi_error(err) })
+ }
+ }
+
+ /// Read all data for a partition with full control over parameters.
+ ///
+ /// Takes `&self` so multiple partitions can be read in parallel from
+ /// different threads sharing one `Arc<ShuffleClient>`.
+ pub fn read_partition(
+ &self,
+ shuffle_id: i32,
+ partition_id: i32,
+ attempt_number: i32,
+ start_map_index: i32,
+ end_map_index: i32,
+ ) -> Result<Vec<u8>> {
+ let mut data_out: *mut u8 = ptr::null_mut();
+ let mut len_out: usize = 0;
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe {
+ sys::celeborn_ffi_read_partition_full(
+ self.handle,
+ shuffle_id,
+ partition_id,
+ attempt_number,
+ start_map_index,
+ end_map_index,
+ &mut data_out,
+ &mut len_out,
+ &mut err,
+ )
+ };
+ if status != sys::CELEBORN_FFI_OK {
+ return Err(unsafe { ffi_error(err) });
+ }
+ // Copy the C-allocated buffer into a Rust-owned Vec, then release
+ // the C buffer with the matching deallocator.
+ let out = unsafe { std::slice::from_raw_parts(data_out,
len_out).to_vec() };
+ unsafe { sys::celeborn_ffi_free_buffer(data_out) };
+ Ok(out)
+ }
+
+ /// Convenience: read all map outputs for a partition.
+ #[inline]
+ pub fn read_partition_all(
+ &self,
+ shuffle_id: i32,
+ partition_id: i32,
+ num_mappers: i32,
+ ) -> Result<Vec<u8>> {
+ self.read_partition(shuffle_id, partition_id, 0, 0, num_mappers)
+ }
+
+ /// Open a streaming reader for a partition. The returned
[`PartitionReader`]
+ /// implements [`std::io::Read`], so the caller can wrap it in a
+ /// [`std::io::BufReader`] and process bytes without materializing the
+ /// whole partition in memory.
+ ///
+ /// The reader borrows `&self`, so multiple partitions can be opened and
+ /// read in parallel from threads sharing one `Arc<ShuffleClient>`. Each
+ /// individual [`PartitionReader`] is single-threaded (its `Read` impl
takes
+ /// `&mut self`), so a given reader must not be read from multiple threads
+ /// at once.
+ pub fn open_partition(
+ &self,
+ shuffle_id: i32,
+ partition_id: i32,
+ attempt_number: i32,
+ start_map_index: i32,
+ end_map_index: i32,
+ ) -> Result<PartitionReader<'_>> {
+ let mut reader_out: *mut sys::celeborn_ffi_partition_reader =
ptr::null_mut();
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe {
+ sys::celeborn_ffi_open_partition_reader(
+ self.handle,
+ shuffle_id,
+ partition_id,
+ attempt_number,
+ start_map_index,
+ end_map_index,
+ &mut reader_out,
+ &mut err,
+ )
+ };
+ if status != sys::CELEBORN_FFI_OK {
+ return Err(unsafe { ffi_error(err) });
+ }
+ if reader_out.is_null() {
+ return Err(Error::NullHandle);
+ }
+ Ok(PartitionReader {
+ inner: reader_out,
+ _client: PhantomData,
+ })
+ }
+
+ /// Convenience: stream all map outputs for a partition.
+ #[inline]
+ pub fn open_partition_all(
+ &self,
+ shuffle_id: i32,
+ partition_id: i32,
+ num_mappers: i32,
+ ) -> Result<PartitionReader<'_>> {
+ self.open_partition(shuffle_id, partition_id, 0, 0, num_mappers)
+ }
+
+ /// Explicitly shut down the client. Preferred over relying on Drop.
+ ///
+ /// After calling `celeborn_ffi_shutdown`, the underlying C++ handle is
+ /// intentionally leaked (see the type-level docs).
+ pub fn shutdown(mut self) -> Result<()> {
+ let mut err: *mut c_char = ptr::null_mut();
+ let status =
+ unsafe { sys::celeborn_ffi_shutdown(self.handle, &mut err) };
+ // Null the handle so Drop does not call shutdown a second time.
+ self.handle = ptr::null_mut();
+ if status == sys::CELEBORN_FFI_OK {
+ Ok(())
+ } else {
+ Err(unsafe { ffi_error(err) })
+ }
+ }
+}
+
+impl Drop for ShuffleClient {
+ fn drop(&mut self) {
+ if self.handle.is_null() {
+ return;
+ }
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe { sys::celeborn_ffi_shutdown(self.handle, &mut
err) };
+ if status != sys::CELEBORN_FFI_OK {
+ let msg = unsafe { sys::take_error(err) }
+ .unwrap_or_else(|| "no error message".to_string());
+ log::error!(
+ "celeborn_ffi_shutdown failed during Drop: {msg}; \
+ caller should explicitly call ShuffleClient::shutdown()
before drop"
+ );
+ }
+ // Intentional leak — no celeborn_ffi_destroy call. See type docs.
+ self.handle = ptr::null_mut();
+ }
+}
+
+/// Streaming reader for a single partition, returned by
+/// [`ShuffleClient::open_partition`]. Implements [`std::io::Read`]; wrap in
+/// a [`std::io::BufReader`] to avoid one FFI call per byte.
+pub struct PartitionReader<'client> {
+ inner: *mut sys::celeborn_ffi_partition_reader,
+ _client: PhantomData<&'client ShuffleClient>,
+}
+
+// SAFETY: the underlying C++ `CelebornInputStream` is owned exclusively by
this
+// reader and only touched through `&mut self` (see the `Read` impl), so it can
+// be moved to another thread. It is deliberately NOT `Sync`: a single reader
+// holds mutable read state and must not be read from multiple threads at once.
+unsafe impl<'client> Send for PartitionReader<'client> {}
+
+impl<'client> std::io::Read for PartitionReader<'client> {
+ fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+ let mut bytes_read: usize = 0;
+ let mut err: *mut c_char = ptr::null_mut();
+ let status = unsafe {
+ sys::celeborn_ffi_read_partition_chunk(
+ self.inner,
+ buf.as_mut_ptr(),
+ buf.len(),
+ &mut bytes_read,
+ &mut err,
+ )
+ };
+ if status != sys::CELEBORN_FFI_OK {
+ let msg = unsafe { sys::take_error(err) }
+ .unwrap_or_else(|| "no error message".to_string());
+ return Err(std::io::Error::other(msg));
+ }
+ Ok(bytes_read)
+ }
+}
+
+impl<'client> Drop for PartitionReader<'client> {
+ fn drop(&mut self) {
+ if !self.inner.is_null() {
+ unsafe { sys::celeborn_ffi_close_partition_reader(self.inner) };
+ self.inner = ptr::null_mut();
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn config_with(app_id: &str, codec: &str) -> Config {
+ let mut config = Config::new(app_id.to_string());
+ config.shuffle_compression_codec = codec.to_string();
+ config
+ }
+
+ #[test]
+ fn validate_accepts_supported_codecs() {
+ for codec in ["NONE", "LZ4", "ZSTD"] {
+ let config = config_with("app", codec);
+ assert!(validate_connect_args(&config, 39099).is_ok());
+ }
+ }
+
+ #[test]
+ fn validate_rejects_empty_app_id() {
+ let config = config_with("", "NONE");
+ let err = validate_connect_args(&config, 39099).unwrap_err();
+ assert!(matches!(err, Error::InvalidArg("app_id is empty")));
+ }
+
+ #[test]
+ fn validate_rejects_non_positive_port() {
+ let config = config_with("app", "NONE");
+ assert!(matches!(
+ validate_connect_args(&config, 0).unwrap_err(),
+ Error::InvalidArg("lm_port must be > 0")
+ ));
+ assert!(matches!(
+ validate_connect_args(&config, -1).unwrap_err(),
+ Error::InvalidArg("lm_port must be > 0")
+ ));
+ }
+
+ #[test]
+ fn validate_rejects_unknown_codec() {
+ let config = config_with("app", "GZIP");
+ assert!(matches!(
+ validate_connect_args(&config, 39099).unwrap_err(),
+ Error::InvalidArg(_)
+ ));
+ }
+}
diff --git a/rust/examples/data_sum_reader.rs b/rust/examples/data_sum_reader.rs
new file mode 100644
index 000000000..18c34a26f
--- /dev/null
+++ b/rust/examples/data_sum_reader.rs
@@ -0,0 +1,137 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+//! Rust equivalent of cpp/celeborn/tests/DataSumWithReaderClient.cpp
+//! Usage: data_sum_reader <lm_host> <lm_port> <app_id> <shuffle_id>
<attempt_id>
+//! <num_mappers> <num_partitions> <result_file> <compress_codec>
+//!
+//! Concurrency: a single `Arc<ShuffleClient>` is shared across
`num_partitions`
+//! threads. Each thread opens and drains its own partition reader in parallel
—
+//! exercising the `&self` `open_partition` path. Each `PartitionReader` itself
+//! stays single-threaded (its `Read` impl takes `&mut self`).
+//!
+//! Set RUST_LOG=info for diagnostic output.
+
+use celeborn_client::{Config, ShuffleClient};
+use std::env;
+use std::fs::File;
+use std::io::{BufReader, Read, Write};
+use std::sync::Arc;
+use std::thread;
+
+fn main() {
+ env_logger::init();
+
+ let args: Vec<String> = env::args().collect();
+ if args.len() != 10 {
+ eprintln!(
+ "Usage: {} <lm_host> <lm_port> <app_id> <shuffle_id> <attempt_id> \
+ <num_mappers> <num_partitions> <result_file> <compress_codec>",
+ args[0]
+ );
+ std::process::exit(1);
+ }
+
+ let lm_host = &args[1];
+ let lm_port: i32 = args[2].parse().expect("lm_port must be an integer");
+ let app_id = args[3].clone();
+ let shuffle_id: i32 = args[4].parse().expect("shuffle_id must be an
integer");
+ let attempt_id: i32 = args[5].parse().expect("attempt_id must be an
integer");
+ let num_mappers: i32 = args[6].parse().expect("num_mappers must be an
integer");
+ let num_partitions: i32 = args[7].parse().expect("num_partitions must be
an integer");
+ let result_file = &args[8];
+ let compress_codec = args[9].clone();
+
+ println!(
+ "lm_host={lm_host}, lm_port={lm_port}, app_id={app_id}, \
+ shuffle_id={shuffle_id}, attempt_id={attempt_id}, \
+ num_mappers={num_mappers}, num_partitions={num_partitions}, \
+ result_file={result_file}, compress_codec={compress_codec}"
+ );
+
+ let mut config = Config::new(app_id);
+ config.shuffle_compression_codec = compress_codec;
+
+ let client =
+ Arc::new(ShuffleClient::connect(config, lm_host,
lm_port).expect("Failed to connect to LM"));
+
+ client
+ .update_reducer_file_group(shuffle_id)
+ .expect("update_reducer_file_group failed");
+
+ // Spawn one thread per partition. All threads share a single
+ // `Arc<ShuffleClient>` and open their readers concurrently via the `&self`
+ // API. Each thread returns its (partition_id, sum, data_count).
+ let handles: Vec<_> = (0..num_partitions)
+ .map(|partition_id| {
+ let client = Arc::clone(&client);
+ thread::spawn(move || {
+ let reader = client
+ .open_partition(shuffle_id, partition_id, attempt_id, 0,
num_mappers)
+ .expect("open_partition failed");
+ let mut buf_reader = BufReader::with_capacity(64 * 1024,
reader);
+
+ let mut sum: i64 = 0;
+ let mut current_number: i64 = 0;
+ let mut data_count: usize = 0;
+ let mut byte = [0u8; 1];
+
+ loop {
+ let n = buf_reader.read(&mut byte).expect("read failed");
+ if n == 0 {
+ break;
+ }
+ let c = byte[0] as char;
+ match c {
+ '-' => {
+ sum += current_number;
+ current_number = 0;
+ data_count += 1;
+ }
+ '+' => {}
+ '0'..='9' => {
+ current_number = current_number * 10 + (c as i64 -
'0' as i64);
+ }
+ _ => {
+ panic!("Unexpected character in partition data:
'{c}'");
+ }
+ }
+ }
+ // Add the last number (data after last '-')
+ sum += current_number;
+
+ (partition_id, sum, data_count)
+ })
+ })
+ .collect();
+
+ let mut result = vec![0i64; num_partitions as usize];
+ for handle in handles {
+ let (partition_id, sum, data_count) = handle.join().expect("reader
thread panicked");
+ result[partition_id as usize] = sum;
+ println!("partition {partition_id} sum result = {sum}, dataCnt =
{data_count}");
+ }
+
+ let mut file = File::create(result_file).expect("Failed to create result
file");
+ for sum in &result {
+ writeln!(file, "{sum}").expect("Failed to write result");
+ }
+
+ // All reader threads have joined, so this is the only remaining Arc
handle.
+ let client = Arc::try_unwrap(client)
+ .unwrap_or_else(|_| panic!("ShuffleClient still has other Arc
references"));
+ client.shutdown().expect("shutdown failed");
+ println!("Reader completed successfully.");
+}
diff --git a/rust/examples/data_sum_writer.rs b/rust/examples/data_sum_writer.rs
new file mode 100644
index 000000000..5d7780bd5
--- /dev/null
+++ b/rust/examples/data_sum_writer.rs
@@ -0,0 +1,154 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You 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.
+
+//! Rust equivalent of cpp/celeborn/tests/DataSumWithWriterClient.cpp
+//! Usage: data_sum_writer <lm_host> <lm_port> <app_id> <shuffle_id>
<attempt_id>
+//! <num_mappers> <num_partitions> <result_file> <compress_codec>
+//!
+//! Concurrency: a single `Arc<ShuffleClient>` is shared across `num_mappers`
+//! threads. Each thread owns one map task, pushes all of its partitions, and
+//! issues `mapper_end`, all in parallel — exercising the `&self` push path
that
+//! makes the client safe to share for concurrent writes.
+//!
+//! Requires env_logger initialized for diagnostic output from celeborn-client
Drop path.
+//! Set RUST_LOG=info (or debug) for verbose output.
+
+use celeborn_client::{Config, ShuffleClient};
+use std::env;
+use std::fs::File;
+use std::io::Write;
+use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::Arc;
+use std::thread;
+
+fn main() {
+ env_logger::init();
+
+ let args: Vec<String> = env::args().collect();
+ if args.len() != 10 {
+ eprintln!(
+ "Usage: {} <lm_host> <lm_port> <app_id> <shuffle_id> <attempt_id> \
+ <num_mappers> <num_partitions> <result_file> <compress_codec>",
+ args[0]
+ );
+ std::process::exit(1);
+ }
+
+ let lm_host = &args[1];
+ let lm_port: i32 = args[2].parse().expect("lm_port must be an integer");
+ let app_id = args[3].clone();
+ let shuffle_id: i32 = args[4].parse().expect("shuffle_id must be an
integer");
+ let attempt_id: i32 = args[5].parse().expect("attempt_id must be an
integer");
+ let num_mappers: i32 = args[6].parse().expect("num_mappers must be an
integer");
+ let num_partitions: i32 = args[7].parse().expect("num_partitions must be
an integer");
+ let result_file = &args[8];
+ let compress_codec = args[9].clone();
+
+ println!(
+ "lm_host={lm_host}, lm_port={lm_port}, app_id={app_id}, \
+ shuffle_id={shuffle_id}, attempt_id={attempt_id}, \
+ num_mappers={num_mappers}, num_partitions={num_partitions}, \
+ result_file={result_file}, compress_codec={compress_codec}"
+ );
+
+ let mut config = Config::new(app_id);
+ config.shuffle_compression_codec = compress_codec;
+
+ let client =
+ Arc::new(ShuffleClient::connect(config, lm_host,
lm_port).expect("Failed to connect to LM"));
+
+ let max_data: i64 = 1_000_000;
+ let num_data: usize = 1000;
+
+ // Shared per-partition sums; multiple mapper threads accumulate
concurrently.
+ let result: Arc<Vec<AtomicI64>> =
+ Arc::new((0..num_partitions).map(|_| AtomicI64::new(0)).collect());
+
+ // Spawn one thread per map task. All threads share a single
+ // `Arc<ShuffleClient>` and push concurrently via the `&self` API.
+ let handles: Vec<_> = (0..num_mappers)
+ .map(|map_id| {
+ let client = Arc::clone(&client);
+ let result = Arc::clone(&result);
+ thread::spawn(move || {
+ // Seed the RNG per map task so every mapper thread produces a
+ // DISTINCT byte stream. A shared/fixed seed would make all
+ // threads push identical data, which masks data races in the
+ // concurrent `&self` push path this example is meant to
stress.
+ let mut rng_state: u64 = 0x9E37_79B9_7F4A_7C15 ^ (map_id as
u64).wrapping_add(1);
+ for partition_id in 0..num_partitions {
+ let mut partition_data = String::new();
+ let mut partition_sum: i64 = 0;
+ for _ in 0..num_data {
+ let data = rand_simple(&mut rng_state, max_data);
+ partition_sum += data;
+ partition_data.push('-');
+ partition_data.push_str(&data.to_string());
+ }
+ result[partition_id as usize].fetch_add(partition_sum,
Ordering::Relaxed);
+
+ client
+ .push_data(
+ shuffle_id,
+ map_id,
+ attempt_id,
+ partition_id,
+ partition_data.as_bytes(),
+ num_mappers,
+ num_partitions,
+ )
+ .expect("push_data failed");
+ }
+ client
+ .mapper_end(shuffle_id, map_id, attempt_id, num_mappers)
+ .expect("mapper_end failed");
+ })
+ })
+ .collect();
+
+ for handle in handles {
+ handle.join().expect("mapper thread panicked");
+ }
+
+ let result: Vec<i64> = result.iter().map(|v|
v.load(Ordering::Relaxed)).collect();
+ for (partition_id, sum) in result.iter().enumerate() {
+ println!("partition {partition_id} sum result = {sum}");
+ }
+
+ let mut file = File::create(result_file).expect("Failed to create result
file");
+ for sum in &result {
+ writeln!(file, "{sum}").expect("Failed to write result");
+ }
+
+ // All mapper threads have joined, so this is the only remaining Arc
handle.
+ let client = Arc::try_unwrap(client)
+ .unwrap_or_else(|_| panic!("ShuffleClient still has other Arc
references"));
+ client.shutdown().expect("shutdown failed");
+ println!("Writer completed successfully.");
+}
+
+/// Simple xorshift pseudo-random number generator. The caller owns the
+/// `state`, so each mapper thread can seed it independently and generate a
+/// distinct byte stream — necessary for the concurrency stress this example
+/// performs (a shared/fixed seed would make all threads emit identical data
+/// and hide races in the `&self` push path).
+fn rand_simple(state: &mut u64, max_val: i64) -> i64 {
+ let mut x = *state;
+ x ^= x << 13;
+ x ^= x >> 7;
+ x ^= x << 17;
+ *state = x;
+ ((x >> 1) as i64) % max_val
+}
diff --git a/rust/resource/lib/README.md b/rust/resource/lib/README.md
new file mode 100644
index 000000000..503c5d6bf
--- /dev/null
+++ b/rust/resource/lib/README.md
@@ -0,0 +1,51 @@
+# Prebuilt Celeborn native artifacts
+
+`celeborn-client-sys/build.rs` looks up the aggregated dylib here before
+falling back to a from-source `cmake` build, so dropping the right file
+in is all it takes for `cargo build` to work without any environment
+variable.
+
+## Layout
+
+```
+resource/lib/
+ <target-triple>/
+ libceleborn_client.dylib # macOS
+ libceleborn_client.so # Linux
+```
+
+`<target-triple>` is the Cargo target identifier — the same string Cargo
+prints from `rustc -vV` (`host:` line) or accepts in `cargo build --target`:
+
+| Platform | Triple | Filename |
+|------------------|------------------------------|---------------------------|
+| macOS Apple Silicon | `aarch64-apple-darwin` | `libceleborn_client.dylib`|
+| macOS Intel | `x86_64-apple-darwin` | `libceleborn_client.dylib`|
+| Linux x86_64 | `x86_64-unknown-linux-gnu` | `libceleborn_client.so` |
+| Linux aarch64 | `aarch64-unknown-linux-gnu` | `libceleborn_client.so` |
+
+## How to refresh the artifact
+
+Build `cpp/` once, then copy the install output:
+
+```bash
+cd cpp && mkdir -p build && cd build
+OPENSSL_ROOT_DIR=/opt/homebrew/opt/openssl@3 cmake .. \
+ -DCMAKE_BUILD_TYPE=Release -DCELEBORN_BUILD_TESTS=OFF \
+ -DCMAKE_PREFIX_PATH=/opt/homebrew \
+ -DCMAKE_INSTALL_PREFIX=$PWD/_install
+cmake --build . --parallel && cmake --install .
+
+TRIPLE=$(rustc -vV | sed -n 's|host: ||p')
+mkdir -p ../../rust/resource/lib/$TRIPLE
+cp _install/lib/libceleborn_client.* ../../rust/resource/lib/$TRIPLE/
+```
+
+After that, `cargo build` from `rust/` finds the dylib without
+`CELEBORN_CPP_PREFIX`.
+
+## Lookup precedence (in `celeborn-client-sys/build.rs`)
+
+1. `CELEBORN_CPP_PREFIX` env var → `<prefix>/lib/libceleborn_client.*`
+2. `rust/resource/lib/<target-triple>/libceleborn_client.*` (this directory)
+3. From-source `cmake` build of `cpp/` into `$OUT_DIR`
diff --git a/sbin/start-lifecycle-manager.sh b/sbin/start-lifecycle-manager.sh
new file mode 100755
index 000000000..0560ccff3
--- /dev/null
+++ b/sbin/start-lifecycle-manager.sh
@@ -0,0 +1,217 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+# Starts a standalone Celeborn LifecycleManager daemon in background.
+# The RPC port is randomly selected from 30000~50000 if --port is not
specified.
+# After successful start, the port is exported as CELEBORN_LM_PORT.
+
+set -euo pipefail
+
+if [ -z "${CELEBORN_HOME:-}" ]; then
+ export CELEBORN_HOME="$(cd "$(dirname "$0")/.."; pwd)"
+fi
+
+usage() {
+ cat <<EOF
+Usage: $0 --app-id <id> --master-endpoints <ep1,ep2,...> [--port <port>]
[--host <host>] [--properties-file <file>]
+
+ --app-id REQUIRED unique application id
+ --master-endpoints REQUIRED comma-separated host:port of Celeborn Masters
+ --port OPTIONAL fixed RPC port to bind (default: random
available port in 30000~50000)
+ --host OPTIONAL bind host (default: hostname)
+ --properties-file OPTIONAL path to celeborn-defaults.conf
+EOF
+ exit 1
+}
+
+# Probe whether `port` is currently bound on ANY local interface.
+# Returns 0 when the port appears free, non-zero when something is listening.
+# Prefer lsof / ss / netstat (cover every interface, IPv4 and IPv6) and only
+# fall back to /dev/tcp loopback probing — the latter misses listeners bound
+# to a specific non-loopback interface, which would let us hand out a port
+# the daemon cannot actually bind.
+port_is_free() {
+ local port="$1"
+ if command -v lsof >/dev/null 2>&1; then
+ ! lsof -nP -iTCP:"${port}" -sTCP:LISTEN >/dev/null 2>&1
+ return
+ fi
+ if command -v ss >/dev/null 2>&1; then
+ ! ss -ltn "sport = :${port}" 2>/dev/null | awk 'NR>1 {found=1} END {exit
!found}'
+ return
+ fi
+ if command -v netstat >/dev/null 2>&1; then
+ ! netstat -an 2>/dev/null | awk -v p=":${port}$" '$1 ~ /^tcp/ && $0 ~
/LISTEN/ && $4 ~ p {found=1} END {exit !found}'
+ return
+ fi
+ # Last-resort loopback probe (incomplete — see note above).
+ ! (echo >/dev/tcp/127.0.0.1/"${port}") 2>/dev/null
+}
+
+# Find a random available port in range [30000, 50000]
+find_available_port() {
+ local port
+ local max_attempts=100
+ local attempt=0
+ while [ "$attempt" -lt "$max_attempts" ]; do
+ port=$(( RANDOM % 20001 + 30000 ))
+ if port_is_free "$port"; then
+ echo "$port"
+ return 0
+ fi
+ attempt=$(( attempt + 1 ))
+ done
+ echo "Error: failed to find an available port in 30000~50000 after
$max_attempts attempts." >&2
+ return 1
+}
+
+# Pre-check: required args must be present
+have_app_id=0; have_master=0; have_port=0
+for ((i=1; i<=$#; i++)); do
+ case "${!i}" in
+ --app-id) have_app_id=1 ;;
+ --master-endpoints) have_master=1 ;;
+ --port) have_port=1 ;;
+ esac
+done
+
+if [ "$have_app_id" -eq 0 ] || [ "$have_master" -eq 0 ]; then
+ echo "Error: --app-id and --master-endpoints are required." >&2
+ usage
+fi
+
+# If --port is not specified, find a random available port and append it to
args
+if [ "$have_port" -eq 0 ]; then
+ CELEBORN_LM_PORT=$(find_available_port)
+ echo "Auto-selected available port: ${CELEBORN_LM_PORT}"
+ set -- "$@" --port "${CELEBORN_LM_PORT}"
+else
+ # Extract the port value from args
+ for ((i=1; i<=$#; i++)); do
+ if [ "${!i}" = "--port" ]; then
+ j=$(( i + 1 ))
+ CELEBORN_LM_PORT="${!j}"
+ break
+ fi
+ done
+fi
+
+export CELEBORN_LM_PORT
+
+# Load environment (JAVA_HOME, CELEBORN_CONF_DIR, etc.)
+# Temporarily disable nounset because load-celeborn-env.sh checks unset vars
+if [ -f "${CELEBORN_HOME}/sbin/load-celeborn-env.sh" ]; then
+ set +u
+ # shellcheck source=/dev/null
+ . "${CELEBORN_HOME}/sbin/load-celeborn-env.sh"
+ set -u
+fi
+
+CELEBORN_CONF_DIR="${CELEBORN_CONF_DIR:-${CELEBORN_HOME}/conf}"
+
+# Determine Java
+if [ -n "${JAVA_HOME:-}" ]; then
+ JAVA="${JAVA_HOME}/bin/java"
+else
+ JAVA="java"
+fi
+
+# Build classpath: conf + service jars + client jars + common jars
+CLASSPATH="${CELEBORN_CONF_DIR}"
+for dir in \
+ "${CELEBORN_HOME}/lifecycle-manager/target/"*.jar \
+ "${CELEBORN_HOME}/service/target/"*.jar \
+ "${CELEBORN_HOME}/client/target/"*.jar \
+ "${CELEBORN_HOME}/common/target/"*.jar \
+ "${CELEBORN_HOME}/spi/target/"*.jar \
+ "${CELEBORN_HOME}/jars/"*.jar \
+ "${CELEBORN_HOME}/lifecycle-manager-jars/"*.jar \
+ "${CELEBORN_HOME}/service-jars/"*.jar \
+ "${CELEBORN_HOME}/client-jars/"*.jar; do
+ if [ -e "$dir" ]; then
+ CLASSPATH="${CLASSPATH}:${dir}"
+ fi
+done
+
+# JVM options
+CELEBORN_JAVA_OPTS="${CELEBORN_JAVA_OPTS:-}"
+if [ -f "${CELEBORN_CONF_DIR}/log4j2.xml" ]; then
+ CELEBORN_JAVA_OPTS="${CELEBORN_JAVA_OPTS}
-Dlog4j2.configurationFile=file:${CELEBORN_CONF_DIR}/log4j2.xml"
+fi
+
+MAIN_CLASS="org.apache.celeborn.server.lifecyclemanager.LifecycleManagerDaemon"
+
+# --- Background execution ---
+LOG_DIR="${CELEBORN_HOME}/logs"
+mkdir -p "${LOG_DIR}"
+LOG_FILE="${LOG_DIR}/lifecyclemanager-${CELEBORN_LM_PORT}.out"
+PID_FILE="${LOG_DIR}/lifecyclemanager-${CELEBORN_LM_PORT}.pid"
+
+nohup "${JAVA}" -cp "${CLASSPATH}" ${CELEBORN_JAVA_OPTS} "${MAIN_CLASS}" "$@" \
+ > "${LOG_FILE}" 2>&1 &
+
+CELEBORN_LM_PID=$!
+echo "${CELEBORN_LM_PID}" > "${PID_FILE}"
+
+# Wait until the daemon either binds its RPC port or dies. A fixed `sleep`
+# would report success while the JVM is still starting and the port is not yet
+# bound; poll the port (reusing port_is_free) so we only declare success once
+# the RPC endpoint is actually accepting connections.
+startup_ok=0
+max_wait_secs=30
+waited=0
+while [ "$waited" -lt "$max_wait_secs" ]; do
+ if ! kill -0 "${CELEBORN_LM_PID}" 2>/dev/null; then
+ # Process exited during startup.
+ break
+ fi
+ if ! port_is_free "${CELEBORN_LM_PORT}"; then
+ # Something (our daemon) is now listening on the port.
+ startup_ok=1
+ break
+ fi
+ sleep 1
+ waited=$(( waited + 1 ))
+done
+
+if [ "$startup_ok" -eq 1 ]; then
+ export CELEBORN_LM_PORT
+ export CELEBORN_LM_PID
+
+ # Write port to a env file so other scripts can source it
+ ENV_FILE="${LOG_DIR}/lifecyclemanager-${CELEBORN_LM_PID}.env"
+ cat > "${ENV_FILE}" <<ENVEOF
+export CELEBORN_LM_PORT=${CELEBORN_LM_PORT}
+export CELEBORN_LM_PID=${CELEBORN_LM_PID}
+ENVEOF
+
+ echo "============================================="
+ echo " LifecycleManager started successfully!"
+ echo " PID : ${CELEBORN_LM_PID}"
+ echo " PORT : ${CELEBORN_LM_PORT}"
+ echo " LOG : ${LOG_FILE}"
+ echo " ENV : ${ENV_FILE}"
+ echo ""
+ echo " To load the port in another shell:"
+ echo " source ${ENV_FILE}"
+ echo "============================================="
+else
+ echo "Error: LifecycleManager failed to start. Check log: ${LOG_FILE}" >&2
+ cat "${LOG_FILE}" >&2
+ rm -f "${PID_FILE}"
+ exit 1
+fi
diff --git a/sbin/stop-lifecycle-manager.sh b/sbin/stop-lifecycle-manager.sh
new file mode 100755
index 000000000..e2a061cc9
--- /dev/null
+++ b/sbin/stop-lifecycle-manager.sh
@@ -0,0 +1,165 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+# Stops a Celeborn LifecycleManager daemon by PID file or port.
+# Usage:
+# stop-lifecycle-manager.sh --port <port> Stop the instance bound to
<port>
+# stop-lifecycle-manager.sh --all Stop all LifecycleManager
instances
+
+set -euo pipefail
+
+if [ -z "${CELEBORN_HOME:-}" ]; then
+ export CELEBORN_HOME="$(cd "$(dirname "$0")/.."; pwd)"
+fi
+
+LOG_DIR="${CELEBORN_HOME}/logs"
+GRACEFUL_TIMEOUT=10
+
+usage() {
+ cat <<EOF
+Usage: $0 --port <port> | --all
+
+ --port <port> Stop the LifecycleManager instance running on the specified
port
+ --all Stop all LifecycleManager instances managed by this script
+EOF
+ exit 1
+}
+
+# Stop a single instance by its PID file.
+# Returns:
+# 0 successfully stopped (or nothing to do — empty/stale PID file cleaned
up)
+# 1 PID file missing entirely (caller decides whether to warn)
+# 2 the process is alive but we failed to kill it
+stop_by_pid_file() {
+ local pid_file="$1"
+ local pid
+
+ if [ ! -f "${pid_file}" ]; then
+ echo "PID file not found: ${pid_file}" >&2
+ return 1
+ fi
+
+ pid=$(cat "${pid_file}")
+ if [ -z "${pid}" ]; then
+ echo "PID file is empty: ${pid_file}, removing." >&2
+ rm -f "${pid_file}"
+ return 0
+ fi
+
+ if kill -0 "${pid}" 2>/dev/null; then
+ echo "Stopping LifecycleManager (PID: ${pid}) with SIGTERM ..."
+ if ! kill -TERM "${pid}" 2>/dev/null; then
+ echo "Failed to send SIGTERM to PID ${pid}." >&2
+ return 2
+ fi
+
+ # Wait for graceful shutdown
+ local waited=0
+ while [ "${waited}" -lt "${GRACEFUL_TIMEOUT}" ]; do
+ if ! kill -0 "${pid}" 2>/dev/null; then
+ echo "LifecycleManager (PID: ${pid}) stopped."
+ rm -f "${pid_file}"
+ # Clean up the env file
+ rm -f "${LOG_DIR}/lifecyclemanager-${pid}.env"
+ return 0
+ fi
+ sleep 1
+ waited=$(( waited + 1 ))
+ done
+
+ # Force kill if still alive
+ echo "LifecycleManager (PID: ${pid}) did not stop after
${GRACEFUL_TIMEOUT}s, sending SIGKILL ..."
+ kill -9 "${pid}" 2>/dev/null || true
+ sleep 1
+ if kill -0 "${pid}" 2>/dev/null; then
+ echo "LifecycleManager (PID: ${pid}) is still alive after SIGKILL." >&2
+ return 2
+ fi
+ rm -f "${pid_file}"
+ rm -f "${LOG_DIR}/lifecyclemanager-${pid}.env"
+ echo "LifecycleManager (PID: ${pid}) killed."
+ return 0
+ else
+ echo "LifecycleManager (PID: ${pid}) is not running. Cleaning up stale PID
file."
+ rm -f "${pid_file}"
+ rm -f "${LOG_DIR}/lifecyclemanager-${pid}.env"
+ return 0
+ fi
+}
+
+# Parse arguments
+if [ $# -eq 0 ]; then
+ usage
+fi
+
+MODE=""
+TARGET_PORT=""
+
+while [ $# -gt 0 ]; do
+ case "$1" in
+ --port)
+ MODE="port"
+ TARGET_PORT="${2:-}"
+ if [ -z "${TARGET_PORT}" ]; then
+ echo "Error: --port requires a port number." >&2
+ usage
+ fi
+ shift 2
+ ;;
+ --all)
+ MODE="all"
+ shift
+ ;;
+ *)
+ echo "Unknown option: $1" >&2
+ usage
+ ;;
+ esac
+done
+
+if [ -z "${MODE}" ]; then
+ usage
+fi
+
+case "${MODE}" in
+ port)
+ PID_FILE="${LOG_DIR}/lifecyclemanager-${TARGET_PORT}.pid"
+ stop_by_pid_file "${PID_FILE}"
+ ;;
+ all)
+ found=0
+ failed=0
+ for pid_file in "${LOG_DIR}"/lifecyclemanager-*.pid; do
+ [ -f "${pid_file}" ] || continue
+ found=1
+ # Capture the return code without letting `set -e` abort the loop on
+ # the first non-zero exit (e.g. a stale PID file we cleaned up, or a
+ # single instance we failed to kill — we still want to try the rest).
+ rc=0
+ stop_by_pid_file "${pid_file}" || rc=$?
+ if [ "${rc}" -ne 0 ]; then
+ echo "Warning: stop_by_pid_file exited ${rc} for ${pid_file}" >&2
+ failed=1
+ fi
+ done
+ if [ "${found}" -eq 0 ]; then
+ echo "No running LifecycleManager instances found."
+ fi
+ exit "${failed}"
+ ;;
+esac
