shuffle: implement basic shuffle server functions (#16)

Author: bigbigxu

Cargo.toml

@@ -19,7 +19,8 @@ members = [
     "curvine-fuse",
     "curvine-web",
     "curvine-ufs",
-    "curvine-cli"
+    "curvine-cli",
+    "curvine-shuffle"
 ]
 
 [workspace.package]
@@ -37,6 +38,7 @@ curvine-ufs = { path = "curvine-ufs" }
 curvine-tests = { path = "curvine-tests" }
 curvine-fuse = { path = "curvine-fuse" }
 curvine-web = { path = "curvine-web" }
+curvine-shuffle = { path = "curvine-shuffle" }
 
 # The most important library start
 
@@ -134,4 +136,5 @@ num_cpus = "1.16.0"
 slog = "2.7.0"
 slog-stdlog = "4.1.0"
 num-bigint = "0.4.6"
-bigdecimal = "0.4.8"
+bigdecimal = "0.4.8"
+bitvec = "1.0.1"

curvine-client/src/unified/unified_filesystem.rs

@@ -21,7 +21,7 @@ use curvine_common::fs::{FileSystem, Path};
 use curvine_common::proto::MountOptions;
 use curvine_common::state::{FileStatus, MasterInfo, MountInfo};
 use curvine_common::FsResult;
-use log::{info, warn};
+use log::{debug, info, warn};
 use orpc::common::{FastHashMap, LocalTime};
 use orpc::runtime::Runtime;
 use orpc::sync::AtomicCounter;

curvine-shuffle/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "curvine-shuffle"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+orpc = { workspace = true }
+tokio = { workspace = true }
+curvine-common = { workspace = true }
+curvine-server = { workspace = true }
+curvine-client = { workspace = true }
+curvine-web = { workspace = true }
+serde = { workspace = true }
+dashmap = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+bytes = { workspace = true }
+num_enum = { workspace = true }
+prost = { workspace = true }
+log = { workspace = true }
+bitvec = { workspace = true }
+
+[build-dependencies]
+prost-build = { workspace = true }
+
+[features]
+default = ["server", "client"]
+server = []
+client = []

curvine-shuffle/build.rs

@@ -0,0 +1,31 @@
+// Copyright 2025 OPPO.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::{env, fs};
+
+fn main() {
+    let src = vec!["shuffle.proto"];
+
+    let base = env::var("OUT_DIR").unwrap_or_else(|_| ".".to_string());
+    let output = format!("{}/protos", base);
+    fs::create_dir_all(&output).unwrap();
+
+    let mut build = prost_build::Config::new();
+    build.type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]");
+
+    build
+        .out_dir(&output)
+        .compile_protos(&src, &["proto/"])
+        .unwrap();
+}

curvine-shuffle/examples/shuffle_cluster.rs

@@ -0,0 +1,50 @@
+// Copyright 2025 OPPO.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use curvine_common::conf::ClusterConf;
+use curvine_shuffle::common::ShuffleConf;
+use curvine_shuffle::master::ShuffleMaster;
+use curvine_shuffle::worker::ShuffleWorker;
+use orpc::common::Logger;
+use orpc::CommonResult;
+use std::thread;
+use std::time::Duration;
+
+fn main() -> CommonResult<()> {
+    Logger::default();
+    let mut conf = ClusterConf::from("etc/curvine-cluster.toml")?;
+    conf.format_master = true;
+    conf.format_worker = true;
+    conf.journal.snapshot_interval = "10s".to_owned();
+    conf.master.min_block_size = 1024 * 1024;
+
+    let shuffle_conf = ShuffleConf {
+        cluster_conf: conf.clone(),
+        ..Default::default()
+    };
+
+    let master = ShuffleMaster::with_conf(shuffle_conf.clone())?;
+
+    thread::spawn(move || {
+        master.block_on_start();
+    });
+
+    let worker = ShuffleWorker::with_conf(shuffle_conf)?;
+    thread::spawn(move || {
+        worker.block_on_start();
+    });
+
+    thread::sleep(Duration::from_secs(u64::MAX));
+    Ok(())
+}

curvine-shuffle/proto/shuffle.proto

@@ -0,0 +1,74 @@
+syntax = "proto2";
+package proto;
+
+option java_package = "io.curvine.proto";
+option java_multiple_files = true;
+option java_outer_classname = "ShuffleProto";
+
+message StageKeyProto {
+    required string app_id = 1;
+    required string stage_id = 2;
+}
+
+message SplitKeyProto {
+    required int32 part_id = 1;
+    required int32 split_id = 2;
+}
+
+message InetAddrProto {
+    required string hostname = 1;
+    required int32 port = 2;
+}
+
+message SplitInfoProto {
+    required int32 part_id = 1;
+    required int32 split_id = 2;
+    required int64 write_len = 3;
+    required InetAddrProto worker_addr = 4;
+}
+
+
+message AllocWriterRequest    {
+    required StageKeyProto stage = 1;
+    required int32 part_id = 2;
+    required int64 split_size = 3;
+    optional SplitInfoProto full_split = 4;
+    repeated uint32 exclude_workers = 5;
+}
+
+message AllocWriterResponse {
+    required StageKeyProto stage = 1;
+    required int32 part_id = 2;
+    required SplitInfoProto cur_split = 3;
+}
+
+message WriteDataRequest {
+    required StageKeyProto stage = 1;
+    required int32 part_id = 2;
+    required int32 split_id = 3;
+    required int64 split_size = 4;
+}
+
+message WriteDataResponse {
+    optional SplitInfoProto full_split = 1;
+}
+
+message TaskCommitRequest {
+    required StageKeyProto stage = 1;
+    required int32 task_id = 2;
+    required int32 num_tasks = 3;
+}
+
+message TaskCommitResponse {
+}
+
+// shuffle stage submission. This submission has two meanings:
+// 1. Aggregate shuffle, is_merge = true, submission can only be made after all tasks are completed
+// 2. Normal shuffle, is_merge = false, after the task is completed, the writer can submit immediately.
+message StageCommitRequest {
+    required StageKeyProto stage = 1;
+}
+
+message StageCommitResponse {
+    repeated SplitKeyProto splits = 1;
+}

curvine-shuffle/src/bin/curvine-shuffle.rs

@@ -0,0 +1,19 @@
+// Copyright 2025 OPPO.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use orpc::CommonResult;
+
+fn main() -> CommonResult<()> {
+    Ok(())
+}

curvine-shuffle/src/client/master_client.rs

@@ -0,0 +1,75 @@
+// Copyright 2025 OPPO.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::common::ShuffleCode;
+use crate::proto::{
+    AllocWriterRequest, AllocWriterResponse, TaskCommitRequest, TaskCommitResponse,
+};
+use crate::protocol::{ProtoUtils, SplitInfo, StageKey};
+use curvine_common::error::FsError;
+use curvine_common::FsResult;
+use orpc::client::ClusterConnector;
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub struct MasterClient {
+    connector: Arc<ClusterConnector>,
+}
+
+impl MasterClient {
+    pub fn new(connector: Arc<ClusterConnector>) -> Self {
+        Self { connector }
+    }
+
+    pub async fn alloc_writer(
+        &self,
+        stage: &StageKey,
+        part_id: i32,
+        split_size: i64,
+        full_split: Option<SplitInfo>,
+    ) -> FsResult<SplitInfo> {
+        let header = AllocWriterRequest {
+            stage: ProtoUtils::stage_key_to_proto(stage.clone()),
+            part_id,
+            split_size,
+            full_split: full_split.map(ProtoUtils::split_info_to_proto),
+            exclude_workers: vec![],
+        };
+
+        let rep_header: AllocWriterResponse = self
+            .connector
+            .proto_rpc::<_, _, FsError>(ShuffleCode::AllocWriter, header)
+            .await?;
+        let split = ProtoUtils::split_info_from_proto(rep_header.cur_split);
+        Ok(split)
+    }
+
+    pub async fn task_commit(
+        &self,
+        stage: &StageKey,
+        task_id: i32,
+        num_tasks: i32,
+    ) -> FsResult<()> {
+        let header = TaskCommitRequest {
+            stage: ProtoUtils::stage_key_to_proto(stage.clone()),
+            task_id,
+            num_tasks,
+        };
+        let _: TaskCommitResponse = self
+            .connector
+            .proto_rpc::<_, _, FsError>(ShuffleCode::TaskCommit, header)
+            .await?;
+        Ok(())
+    }
+}

curvine-shuffle/src/client/mod.rs

@@ -0,0 +1,25 @@
+// Copyright 2025 OPPO.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod shuffle_writer;
+pub use self::shuffle_writer::ShuffleWriter;
+
+mod master_client;
+pub use self::master_client::MasterClient;
+
+mod worker_client;
+pub use self::worker_client::WorkerClient;
+
+mod shuffle_context;
+pub use self::shuffle_context::ShuffleContext;

curvine-shuffle/src/client/shuffle_context.rs

@@ -0,0 +1,52 @@
+// Copyright 2025 OPPO.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::client::WorkerClient;
+use crate::common::ShuffleConf;
+use curvine_common::FsResult;
+use orpc::client::ClusterConnector;
+use orpc::io::net::InetAddr;
+use orpc::runtime::Runtime;
+use std::sync::Arc;
+
+pub struct ShuffleContext {
+    connector: Arc<ClusterConnector>,
+    conf: ShuffleConf,
+}
+
+impl ShuffleContext {
+    pub fn new(conf: ShuffleConf, connector: Arc<ClusterConnector>) -> Self {
+        Self { connector, conf }
+    }
+
+    pub async fn worker_client(&self, addr: &InetAddr) -> FsResult<WorkerClient> {
+        WorkerClient::new(&self.connector, addr.clone()).await
+    }
+
+    pub fn conf(&self) -> &ShuffleConf {
+        &self.conf
+    }
+
+    pub fn connector(&self) -> Arc<ClusterConnector> {
+        self.connector.clone()
+    }
+
+    pub fn clone_runtime(&self) -> Arc<Runtime> {
+        self.connector.clone_runtime()
+    }
+
+    pub fn root_dir(&self) -> &str {
+        &self.conf.root_dir
+    }
+}