diff --git a/.github/workflows/gaia.yml b/.github/workflows/gaia.yml
index edf0bc57a33b..1c0d284492cf 100644
--- a/.github/workflows/gaia.yml
+++ b/.github/workflows/gaia.yml
@@ -6,21 +6,22 @@ on:
branches:
- main
paths:
- - 'research/gaia/**'
+ - 'research/**'
- '.github/workflows/gaia.yml'
- - '!research/gaia/**.md'
+ - '!research/**.md'
pull_request:
branches:
- main
paths:
- - 'research/gaia/**'
+ - 'research/**'
- '.github/workflows/gaia.yml'
- - '!research/gaia/**.md'
+ - '!research/**.md'
jobs:
gaia-test:
- runs-on: ubuntu-20.04
- if: false
+ runs-on: self-hosted
+ container:
+ image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.3.21
defaults:
run:
shell: bash --noprofile --norc -eo pipefail {0}
@@ -31,40 +32,50 @@ jobs:
- name: Install Dependencies
run: |
- # Due to an observation of changing hostname in github runners,
- # append 127.0.0.1 to etc/hosts to avoid DNS lookup.
- r=`cat /etc/hosts | grep $(hostname) || true`
- if [ -z "${r}" ];then export hn=$(hostname); sudo -E bash -c 'echo "127.0.0.1 ${hn}" >> /etc/hosts'; fi
- cat /etc/hosts
+ # install cargo cause github change $HOME to /github/home, so we can't use
+ # `source ~/.bashrc` or `source ${HOME}/.bashrc` to find cargo command.
+ sudo chown -R $(id -u):$(id -g) ${HOME}
+ sudo chmod -R a+wrx ${HOME} ${GITHUB_WORKSPACE}
+ wget --no-verbose https://golang.org/dl/go1.15.5.linux-amd64.tar.gz
+ sudo tar -C /usr/local -xzf go1.15.5.linux-amd64.tar.gz
+ curl -sf -L https://static.rust-lang.org/rustup.sh | sh -s -- -y --profile minimal --default-toolchain 1.54.0
+ echo "source ~/.cargo/env" >> ~/.bashrc
+ source ${HOME}/.bashrc
+ rustup component add rustfmt
+ # install rdkafka
+ sudo yum install -y librdkafka-devel
+ # install cppkafka
+ git clone -b 0.4.0 --single-branch --depth=1 \
+ https://github.com/mfontanini/cppkafka.git /tmp/cppkafka && \
+ cd /tmp/cppkafka && git submodule update --init && \
+ mkdir -p build && cd build && \
+ cmake .. && make -j && sudo make install && \
+ rm -fr /tmp/cppkafka
- # install dependencies
- ${GITHUB_WORKSPACE}/scripts/install_deps.sh --dev --verbose
-
- - name: Build GraphScope
+ - name: Build Ir on Experimental Store
run: |
- source ${HOME}/.graphscope_env
- make BUILD_TYPE=debug
+ source ${HOME}/.bashrc
+ cd ${GITHUB_WORKSPACE}/research/query_service/ir/compiler
+ make build
- - name: Build MaxGraph Store
+ - name: Build Ir on Groot Store
run: |
- source ${HOME}/.graphscope_env
+ source ${HOME}/.bashrc
cd ${GITHUB_WORKSPACE}/interactive_engine
mvn clean install -DskipTests -Pv2
- - name: Gaia test on MaxGraph Store
+ - name: Ir Unit Test
run: |
- source ${HOME}/.graphscope_env
- cd ${GITHUB_WORKSPACE}/interactive_engine/gaia-adaptor && ./gremlin_test.sh
+ source ${HOME}/.bashrc
+ cd ${GITHUB_WORKSPACE}/research/query_service/ir/compiler && make test
- - name: Run Backup Test on Maxgraph Store
+ - name: Ir Integration Test on Experimental Store
run: |
- source ${HOME}/.graphscope_env
- cd ${GITHUB_WORKSPACE}/interactive_engine/sdk && ./backup_test.sh
+ source ${HOME}/.bashrc
+ cd ${GITHUB_WORKSPACE}/research/query_service/ir/compiler && ./ir_exprimental_ci.sh
- - name: Gaia runtime test
+ - name: Ir Integration Test on Groot Store
run: |
- source ${HOME}/.graphscope_env
+ source ${HOME}/.bashrc
+ cd ${GITHUB_WORKSPACE}/interactive_engine/ir-adaptor && ./ir_groot_ci.sh
- cd ${GITHUB_WORKSPACE}/research/query_service/gremlin/gremlin_core/tests
- sh prepare_all_test_plan.sh
- cargo test
diff --git a/.github/workflows/pegasus.yml b/.github/workflows/pegasus.yml
new file mode 100644
index 000000000000..618fbd67c15a
--- /dev/null
+++ b/.github/workflows/pegasus.yml
@@ -0,0 +1,84 @@
+name: GraphScope Pegasus CI
+
+on:
+ push:
+ branches:
+ - gaia-x
+ paths:
+ - 'research/engine/pegasus/**'
+ - '.github/workflows/pegasus.yml'
+
+ pull_request:
+ branches:
+ - gaia-x
+ paths:
+ - 'research/engine/pegasus/**'
+ - '.github/workflows/pegasus.yml'
+
+env:
+ CARGO_TERM_COLOR: always
+
+jobs:
+ peagsus-build:
+ runs-on: ubuntu-20.04
+ defaults:
+ run:
+ shell: bash --noprofile --norc -eo pipefail {0}
+
+ steps:
+ - uses: actions/checkout@v2
+
+ - name: Build & Test
+ run: |
+ echo $(pwd)
+ cd research/engine/pegasus
+ cargo build --verbose
+ cargo test --verbose
+
+ - name: Run example
+ run: |
+ cd research/engine/pegasus
+
+ cargo build --release --examples
+
+ # Run word_count_toy
+ target/release/examples/word_count_toy
+
+ target/release/examples/word_count_toy -w 2
+
+ target/release/examples/word_count_toy -w 4
+
+ # Run logistic regression
+ target/release/examples/logistic_regression --data pegasus/examples/data/binary.csv
+
+ k8s-test:
+ runs-on: self-hosted
+ defaults:
+ run:
+ shell: bash --noprofile --norc -eo pipefail {0}
+
+ steps:
+ - uses: actions/checkout@v2
+
+ - name : Prepare Image
+ run: |
+ sudo docker pull registry.cn-hongkong.aliyuncs.com/graphscope/pegasus-base:latest
+ sudo docker build -t registry.cn-hongkong.aliyuncs.com/graphscope/pegasus:${{ github.sha }} \
+ --network=host \
+ -f research/engine/pegasus/test/k8s/pegasus.Dockerfile .
+ echo " - name: CONFIG_MAP_NAME" >> research/engine/pegasus/test/k8s/pegasus-set.yaml
+ echo " value: \"${{ github.sha }}\"" >> research/engine/pegasus/test/k8s/pegasus-set.yaml
+ echo " image: registry.cn-hongkong.aliyuncs.com/graphscope/pegasus:${{ github.sha }}" >> research/engine/pegasus/test/k8s/pegasus-set.yaml
+
+ - name: Start k8s cluster
+ run: |
+ kubectl get ns pegasus-ci || kubectl create ns pegasus-ci
+ kubectl --namespace pegasus-ci create -f research/engine/pegasus/test/k8s/role_binding.yaml
+ kubectl --namespace pegasus-ci create -f research/engine/pegasus/test/k8s/pegasus-set.yaml
+ bash research/engine/pegasus/test/k8s/read_result.sh ${{ github.sha }}
+
+ - name: Clean
+ if: always()
+ run: |
+ kubectl delete ns pegasus-ci || true
+ docker rmi -f registry.cn-hongkong.aliyuncs.com/graphscope/pegasus:${{ github.sha }} --wait=false || true
diff --git a/.gitignore b/.gitignore
index 0b3286e2a164..c1f063a9169d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -65,6 +65,7 @@ interactive_engine/executor/store/src/db/proto/*
interactive_engine/data_load_tools/dependency-reduced-pom.xml
interactive_engine/gaia-adaptor/dependency-reduced-pom.xml
interactive_engine/executor/Cargo.lock
+research/engine/pegasus/benchmark/src/graph/storage/clickhouse/pb_gen/*
# java sdk
java/**/dependency-reduced-pom.xml
diff --git a/Makefile b/Makefile
index 1472f881c533..ee1b4bc8b27a 100644
--- a/Makefile
+++ b/Makefile
@@ -115,6 +115,7 @@ gie:
mkdir -p $(WORKING_DIR)/.install_prefix && \
tar -xf $(WORKING_DIR)/interactive_engine/assembly/target/maxgraph-assembly-0.0.1-SNAPSHOT.tar.gz --strip-components 1 -C $(WORKING_DIR)/.install_prefix && \
cp $(WORKING_DIR)/interactive_engine/executor/target/$(BUILD_TYPE)/executor $(WORKING_DIR)/.install_prefix/bin/executor && \
+ cp $(WORKING_DIR)/interactive_engine/executor/target/$(BUILD_TYPE)/gaia_executor $(WORKING_DIR)/.install_prefix/bin/gaia_executor && \
sudo cp -r $(WORKING_DIR)/.install_prefix/* $(INSTALL_PREFIX) && \
rm -fr $(WORKING_DIR)/.install_prefix
diff --git a/interactive_engine/distribution/src/assembly/assembly.xml b/interactive_engine/distribution/src/assembly/assembly.xml
index 7332fb429241..02603ee524c5 100644
--- a/interactive_engine/distribution/src/assembly/assembly.xml
+++ b/interactive_engine/distribution/src/assembly/assembly.xml
@@ -13,6 +13,13 @@
native
+
+ ${project.parent.basedir}/../research/query_service/ir/target/release
+
+ libir_core*
+
+ native
+
${project.parent.basedir}/lgraph/build/install
diff --git a/interactive_engine/executor/Cargo.toml b/interactive_engine/executor/Cargo.toml
index 658e68a2f525..ee1a4d92d138 100644
--- a/interactive_engine/executor/Cargo.toml
+++ b/interactive_engine/executor/Cargo.toml
@@ -5,7 +5,8 @@ members = [
"store",
"server",
"Pegasus",
- "ffi"
+ "ffi",
+ "gaia_runtime"
]
[profile.release]
diff --git a/interactive_engine/executor/ffi/Cargo.toml b/interactive_engine/executor/ffi/Cargo.toml
index 9160420c9cf0..48dbf86b4cd0 100644
--- a/interactive_engine/executor/ffi/Cargo.toml
+++ b/interactive_engine/executor/ffi/Cargo.toml
@@ -13,6 +13,10 @@ maxgraph-store = { path = "../store" }
maxgraph-common = { path = "../../rust-common" }
maxgraph-server = { path = "../server" }
pegasus = { path = "../Pegasus", package = "pegasus" }
+gaia_pegasus = { path = "../../../research/engine/pegasus/pegasus", package = "pegasus" }
+pegasus_network = { path = "../../../research/engine/pegasus/network" }
+pegasus_server = { path = "../../../research/engine/pegasus/server-v0" }
+graph_proxy = { path = "../../../research/query_service/ir/graph_proxy" }
itertools = "0.7.8"
log = "0.3"
log4rs = "0.8.0"
diff --git a/interactive_engine/executor/ffi/src/executor/gaia/engine_ports_response.rs b/interactive_engine/executor/ffi/src/executor/gaia/engine_ports_response.rs
new file mode 100644
index 000000000000..39a26d978284
--- /dev/null
+++ b/interactive_engine/executor/ffi/src/executor/gaia/engine_ports_response.rs
@@ -0,0 +1,64 @@
+//
+//! Copyright 2022 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+//!
+
+use std::os::raw::c_char;
+use std::ffi::CString;
+
+#[repr(C)]
+#[allow(non_snake_case)]
+pub struct EnginePortsResponse {
+ success: bool,
+ errMsg: *const c_char,
+ enginePort: i32,
+ rpcPort: i32,
+}
+
+impl EnginePortsResponse {
+ pub fn new(engine_port: i32, rpc_port: i32) -> Box {
+ Box::new(EnginePortsResponse {
+ success: true,
+ errMsg: std::ptr::null(),
+ enginePort: engine_port,
+ rpcPort: rpc_port,
+ })
+ }
+
+ pub fn new_with_error(err_msg: &str) -> Box {
+ let msg = CString::new(err_msg).unwrap();
+ let response = EnginePortsResponse {
+ success: false,
+ errMsg: msg.as_ptr(),
+ enginePort: 0,
+ rpcPort: 0,
+ };
+ ::std::mem::forget(msg);
+ Box::new(response)
+ }
+}
+
+impl Drop for EnginePortsResponse {
+ fn drop(&mut self) {
+ unsafe {
+ if !self.errMsg.is_null() {
+ CString::from_raw(self.errMsg as *mut c_char);
+ }
+ }
+ }
+}
+
+#[no_mangle]
+#[allow(non_snake_case)]
+pub extern fn dropEnginePortsResponse(_: Box) {}
diff --git a/interactive_engine/executor/ffi/src/executor/gaia/gaia_library.rs b/interactive_engine/executor/ffi/src/executor/gaia/gaia_library.rs
new file mode 100644
index 000000000000..0567723b1f45
--- /dev/null
+++ b/interactive_engine/executor/ffi/src/executor/gaia/gaia_library.rs
@@ -0,0 +1,101 @@
+//
+//! Copyright 2022 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+//!
+
+use std::os::raw::{c_void, c_char};
+use maxgraph_store::db::common::bytes::util::parse_pb;
+use maxgraph_store::db::api::GraphConfigBuilder;
+use std::sync::Arc;
+use maxgraph_store::db::common::unsafe_util::to_mut;
+use maxgraph_store::db::graph::store::GraphStore;
+use std::ffi::CStr;
+use std::net::{SocketAddr, ToSocketAddrs};
+use crate::executor::gaia::gaia_server::GaiaServer;
+use crate::executor::gaia::engine_ports_response::EnginePortsResponse;
+use maxgraph_store::db::proto::model::ConfigPb;
+
+pub type EngineHandle = *const c_void;
+pub type GraphHandle = *const c_void;
+
+#[no_mangle]
+pub extern fn initialize(config_bytes: *const u8, len: usize) -> EngineHandle {
+ let config_buf = unsafe { ::std::slice::from_raw_parts(config_bytes, len) };
+ let config_pb = parse_pb::(config_buf).expect("parse config pb failed");
+ let mut config_builder = GraphConfigBuilder::new();
+ config_builder.set_storage_options(config_pb.get_configs().clone());
+ let config = Arc::new(config_builder.build());
+ let handle = Box::new(GaiaServer::new(config));
+ Box::into_raw(handle) as EngineHandle
+}
+
+#[no_mangle]
+pub extern fn addPartition(engine_handle: EngineHandle, partition_id: i32, graph_handle: GraphHandle) {
+ let engine_ptr = unsafe {
+ to_mut(&*(engine_handle as *const GaiaServer))
+ };
+ let graph_ptr = unsafe {
+ Arc::from_raw(&*(graph_handle as *const GraphStore))
+ };
+ engine_ptr.add_partition(partition_id as u32, graph_ptr);
+}
+
+#[no_mangle]
+pub extern fn updatePartitionRouting(engine_handle: EngineHandle, partition_id: i32, server_id: i32) {
+ let engine_ptr = unsafe {
+ to_mut(&*(engine_handle as *const GaiaServer))
+ };
+ engine_ptr.update_partition_routing(partition_id as u32, server_id as u32);
+}
+
+#[no_mangle]
+pub extern fn startEngine(engine_handle: EngineHandle) -> Box {
+ let engine_ptr = unsafe {
+ to_mut(&*(engine_handle as *const GaiaServer))
+ };
+ match engine_ptr.start() {
+ Ok((engine_port, server_port)) => {
+ EnginePortsResponse::new(engine_port as i32, server_port as i32)
+ },
+ Err(e) => {
+ let msg = format!("{:?}", e);
+ EnginePortsResponse::new_with_error(&msg)
+ },
+ }
+}
+
+#[no_mangle]
+pub extern fn stopEngine(engine_handle: EngineHandle) {
+ let engine_ptr = unsafe {
+ to_mut(&*(engine_handle as *const GaiaServer))
+ };
+ engine_ptr.stop();
+}
+
+#[no_mangle]
+pub extern fn updatePeerView(engine_handle: EngineHandle, peer_view_string_raw: *const c_char) {
+ let slice = unsafe { CStr::from_ptr(peer_view_string_raw) }.to_bytes();
+ let peer_view_string = std::str::from_utf8(slice).unwrap();
+ let peer_view = peer_view_string.split(",").map(|item| {
+ let mut fields = item.split("#");
+ let id = fields.next().unwrap().parse::().unwrap();
+ let addr_str = fields.next().unwrap();
+ let mut addr_iter = addr_str.to_socket_addrs().expect(format!("parse addr failed [{}]", addr_str).as_str());
+ (id, addr_iter.next().unwrap())
+ }).collect::>();
+ let engine_ptr = unsafe {
+ to_mut(&*(engine_handle as *const GaiaServer))
+ };
+ engine_ptr.update_peer_view(peer_view);
+}
diff --git a/interactive_engine/executor/ffi/src/executor/gaia/gaia_server.rs b/interactive_engine/executor/ffi/src/executor/gaia/gaia_server.rs
new file mode 100644
index 000000000000..f42c27b99684
--- /dev/null
+++ b/interactive_engine/executor/ffi/src/executor/gaia/gaia_server.rs
@@ -0,0 +1,141 @@
+//
+//! Copyright 2021 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+//!
+
+use maxgraph_store::db::api::{GraphConfig, GraphResult, GraphError};
+use std::sync::Arc;
+use maxgraph_store::db::graph::store::GraphStore;
+use std::net::SocketAddr;
+use maxgraph_runtime::store::groot::global_graph::GlobalGraph;
+use gaia_pegasus::Configuration as GaiaConfig;
+use maxgraph_store::db::api::GraphErrorCode::EngineError;
+use tokio::runtime::Runtime;
+use pegasus_server::service::Service;
+use pegasus_server::rpc::{start_rpc_server, RpcService};
+use pegasus_network::SimpleServerDetector;
+use pegasus_network::config::{NetworkConfig, ServerAddr};
+use graph_proxy::{InitializeJobCompiler, QueryMaxGraph};
+use maxgraph_store::api::PartitionId;
+
+pub struct GaiaServer {
+ config: Arc,
+ graph: Arc,
+ detector: Arc,
+ rpc_runtime: Runtime,
+}
+
+impl GaiaServer {
+ pub fn new(config: Arc) -> Self {
+ let partition_count = config.get_storage_option("partition.count")
+ .expect("required config partition.count is missing").parse().expect("parse partition.count failed");
+ GaiaServer {
+ config,
+ graph: Arc::new(GlobalGraph::empty(partition_count)),
+ detector: Arc::new(SimpleServerDetector::new()),
+ rpc_runtime: Runtime::new().unwrap(),
+ }
+ }
+
+ pub fn add_partition(&mut self, partition_id: PartitionId, graph_partition: Arc) {
+ Arc::get_mut(&mut self.graph).unwrap().add_partition(partition_id, graph_partition);
+ }
+
+ pub fn update_partition_routing(&mut self, partition_id: PartitionId, worker_id: u32) {
+ Arc::get_mut(&mut self.graph).unwrap().update_partition_routing(partition_id, worker_id);
+ }
+
+ pub fn start(&self) -> GraphResult<(u16, u16)> {
+ let report = match self.config.get_storage_option("gaia.report") {
+ None => false,
+ Some(report_string) => report_string.parse()
+ .map_err(|e| GraphError::new(EngineError, format!("{:?}", e)))?,
+ };
+ let rpc_port = match self.config.get_storage_option("gaia.rpc.port") {
+ None => { 0 },
+ Some(server_port_string) => {
+ server_port_string.parse().map_err(|e| GraphError::new(EngineError, format!("{:?}", e)))?
+ },
+ };
+ let addr = format!("{}:{}", "0.0.0.0", rpc_port).parse()
+ .map_err(|e| GraphError::new(EngineError, format!("{:?}", e)))?;
+ let gaia_config = make_gaia_config(self.config.clone());
+ let socket_addr = gaia_pegasus::startup_with(gaia_config, self.detector.clone())
+ .map_err(|e| GraphError::new(EngineError, format!("{:?}", e)))?
+ .ok_or(GraphError::new(EngineError, "gaia engine return None addr".to_string()))?;
+
+ let rpc_port = self.rpc_runtime.block_on(async{
+ let query_maxgraph = QueryMaxGraph::new(self.graph.clone(), self.graph.clone());
+ let job_compiler = query_maxgraph.initialize_job_compiler();
+ let service = Service::new(job_compiler);
+ let rpc_service = RpcService::new(service, report);
+ let local_addr = start_rpc_server(addr, rpc_service, false).await.unwrap();
+ local_addr.port()
+ });
+ Ok((socket_addr.port(), rpc_port))
+ }
+
+ pub fn update_peer_view(&self, peer_view: Vec<(u64, SocketAddr)>) {
+ self.detector.update_peer_view(peer_view.into_iter());
+ }
+
+ pub fn stop(&self) {
+ gaia_pegasus::shutdown_all();
+ }
+}
+
+fn make_gaia_config(graph_config: Arc) -> GaiaConfig {
+ let server_id = graph_config.get_storage_option("node.idx").expect("required config node.idx is missing")
+ .parse().expect("parse node.idx failed");
+ let ip = "0.0.0.0".to_string();
+ let port = match graph_config.get_storage_option("gaia.engine.port") {
+ None => { 0 },
+ Some(server_port_string) => {
+ server_port_string.parse().expect("parse gaia.engine.port failed")
+ },
+ };
+ let worker_num = match graph_config.get_storage_option("worker.num") {
+ None => 1,
+ Some(worker_num_string) => worker_num_string.parse().expect("parse worker.num failed"),
+ };
+ let nonblocking = graph_config.get_storage_option("gaia.nonblocking")
+ .map(|config_str| config_str.parse().expect("parse gaia.nonblocking failed"));
+ let read_timeout_ms = graph_config.get_storage_option("gaia.read.timeout.ms")
+ .map(|config_str| config_str.parse().expect("parse gaia.read.timeout.ms failed"));
+ let write_timeout_ms = graph_config.get_storage_option("gaia.write.timeout.ms")
+ .map(|config_str| config_str.parse().expect("parse gaia.write.timeout.ms failed"));
+ let read_slab_size = graph_config.get_storage_option("gaia.read.slab.size")
+ .map(|config_str| config_str.parse().expect("parse gaia.read.slab.size failed"));
+ let no_delay = graph_config.get_storage_option("gaia.no.delay")
+ .map(|config_str| config_str.parse().expect("parse gaia.no.delay failed"));
+ let send_buffer = graph_config.get_storage_option("gaia.send.buffer")
+ .map(|config_str| config_str.parse().expect("parse gaia.send.buffer failed"));
+ let heartbeat_sec = graph_config.get_storage_option("gaia.heartbeat.sec")
+ .map(|config_str| config_str.parse().expect("parse gaia.heartbeat.sec failed"));
+ let max_pool_size = graph_config.get_storage_option("gaia.max.pool.size")
+ .map(|config_str| config_str.parse().expect("parse gaia.max.pool.size failed"));
+ let mut network_config = NetworkConfig::new(server_id, ServerAddr::new(ip, port), worker_num);
+ network_config.nonblocking(nonblocking)
+ .read_timeout_ms(read_timeout_ms)
+ .write_timeout_ms(write_timeout_ms)
+ .read_slab_size(read_slab_size)
+ .no_delay(no_delay)
+ .send_buffer(send_buffer)
+ .heartbeat_sec(heartbeat_sec);
+ GaiaConfig {
+ network: Some(network_config),
+ max_pool_size,
+ }
+}
+
diff --git a/interactive_engine/executor/ffi/src/executor/gaia/mod.rs b/interactive_engine/executor/ffi/src/executor/gaia/mod.rs
new file mode 100644
index 000000000000..883ff6b122fb
--- /dev/null
+++ b/interactive_engine/executor/ffi/src/executor/gaia/mod.rs
@@ -0,0 +1,19 @@
+//
+//! Copyright 2022 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+//!
+
+mod gaia_library;
+mod gaia_server;
+mod engine_ports_response;
diff --git a/interactive_engine/executor/ffi/src/executor/mod.rs b/interactive_engine/executor/ffi/src/executor/mod.rs
index 031312f4c7d2..90578ad08738 100644
--- a/interactive_engine/executor/ffi/src/executor/mod.rs
+++ b/interactive_engine/executor/ffi/src/executor/mod.rs
@@ -1,2 +1,18 @@
-mod pegasus;
+//
+//! Copyright 2022 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+//!
+mod gaia;
+mod pegasus;
diff --git a/interactive_engine/executor/ffi/src/lib.rs b/interactive_engine/executor/ffi/src/lib.rs
index 60702aec6c22..98d6e7f2f91b 100644
--- a/interactive_engine/executor/ffi/src/lib.rs
+++ b/interactive_engine/executor/ffi/src/lib.rs
@@ -2,6 +2,7 @@ extern crate maxgraph_runtime;
extern crate maxgraph_store;
extern crate maxgraph_common;
extern crate maxgraph_server;
+extern crate gaia_pegasus;
extern crate itertools;
extern crate pegasus;
#[macro_use]
@@ -9,6 +10,7 @@ extern crate log;
extern crate log4rs;
extern crate grpcio;
extern crate lazy_static;
+extern crate pegasus_network;
mod executor;
mod store;
diff --git a/interactive_engine/executor/gaia_runtime/Cargo.toml b/interactive_engine/executor/gaia_runtime/Cargo.toml
new file mode 100644
index 000000000000..614ff57db0be
--- /dev/null
+++ b/interactive_engine/executor/gaia_runtime/Cargo.toml
@@ -0,0 +1,51 @@
+[package]
+name = "gaia_runtime"
+version = "0.1.0"
+authors = ["BingqingLyu "]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+abomonation = "0.7"
+abomonation_derive = "0.3"
+bincode = "1.0.1"
+byteorder = "1.3.1"
+crossbeam-channel = "0.4.4"
+crossbeam-queue = "0.1"
+futures-cpupool = "0.1.8"
+getopts = "0.2.14"
+grpcio = "=0.4.1"
+quote = "=1.0.1"
+itertools = "0.7.8"
+lazy_static = "1.2.0"
+log = "0.3"
+log4rs = "0.8.0"
+maxgraph-common = { path = "../../rust-common" }
+maxgraph-store = { path = "../store" }
+maxgraph-server = { path = "../server" }
+maxgraph-runtime = { path = "../runtime" }
+protobuf = { version = "~2.0", features = ["with-bytes"] }
+rand = "0.7.3"
+regex = "1"
+serde = "1.0.72"
+serde_derive = "1.0.72"
+serde_json = "1.0.3"
+structopt = "0.2"
+zookeeper = { git = "https://github.com/bonifaido/rust-zookeeper.git", rev = "fd222dd" }
+lru_time_cache = "0.8.0"
+pegasus = { path = "../Pegasus" }
+libc = "0.2"
+
+tokio = { version = "1.0", features = ["macros", "sync"] }
+futures = { version = "0.3.0", features = ["thread-pool"] }
+gaia_pegasus = { path = "../../../research/engine/pegasus/pegasus", package = "pegasus" }
+pegasus_network = { path = "../../../research/engine/pegasus/network" }
+pegasus_server = { path = "../../../research/engine/pegasus/server-v0" }
+graph_proxy = { path = "../../../research/query_service/ir/graph_proxy" }
+
+[dev-dependencies]
+env_logger = "0.6"
+
+[build-dependencies]
+cmake = "0.1"
diff --git a/interactive_engine/executor/gaia_runtime/src/bin/gaia_executor.rs b/interactive_engine/executor/gaia_runtime/src/bin/gaia_executor.rs
new file mode 100644
index 000000000000..34c8b2790f28
--- /dev/null
+++ b/interactive_engine/executor/gaia_runtime/src/bin/gaia_executor.rs
@@ -0,0 +1,298 @@
+//
+//! Copyright 2020 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+
+#![allow(bare_trait_objects)]
+
+extern crate futures;
+extern crate grpcio;
+#[macro_use]
+extern crate log;
+extern crate gaia_pegasus;
+extern crate graph_proxy;
+extern crate log4rs;
+extern crate maxgraph_common;
+extern crate maxgraph_runtime;
+extern crate maxgraph_server;
+extern crate maxgraph_store;
+extern crate pegasus;
+extern crate pegasus_server;
+extern crate protobuf;
+extern crate structopt;
+
+use gaia_runtime::server::init_with_rpc_service;
+use gaia_runtime::server::manager::GaiaServerManager;
+use graph_proxy::{InitializeJobCompiler, QueryVineyard};
+use grpcio::ChannelBuilder;
+use grpcio::EnvBuilder;
+use maxgraph_common::proto::data::*;
+use maxgraph_common::proto::hb::*;
+use maxgraph_common::proto::query_flow::*;
+use maxgraph_common::util;
+use maxgraph_common::util::get_local_ip;
+use maxgraph_common::util::log4rs::init_log4rs;
+use maxgraph_runtime::server::manager::*;
+use maxgraph_runtime::server::RuntimeInfo;
+use maxgraph_server::StoreContext;
+use maxgraph_store::api::graph_partition::GraphPartitionManager;
+use maxgraph_store::api::prelude::*;
+use maxgraph_store::config::{StoreConfig, VINEYARD_GRAPH};
+use pegasus_server::rpc::{start_rpc_server, RpcService};
+use pegasus_server::service::Service;
+use protobuf::Message;
+use std::collections::HashMap;
+use std::env;
+use std::sync::atomic::AtomicBool;
+use std::sync::mpsc::{channel, Sender};
+use std::sync::{Arc, Mutex, RwLock};
+use std::thread;
+use std::time::Duration;
+use tokio::runtime::Runtime;
+
+fn main() {
+ if let Some(_) = env::args().find(|arg| arg == "--show-build-info") {
+ util::get_build_info();
+ return;
+ }
+ init_log4rs();
+ let mut store_config = {
+ let args: Vec = std::env::args().collect();
+ if args.len() <= 6 && args[1] == "--config" {
+ let mut store_config = StoreConfig::init_from_file(&args[2], &args[4]);
+ if args.len() == 6 {
+ store_config.graph_name = (&args[5]).to_string();
+ }
+ store_config
+ } else {
+ StoreConfig::init()
+ }
+ };
+ let (alive_id, partitions) = get_init_info(&store_config);
+ info!("alive_id: {:?}, partitions: {:?}", alive_id, partitions);
+ store_config.update_alive_id(alive_id);
+ info!("{:?}", store_config);
+
+ let worker_num = store_config.timely_worker_per_process;
+ let store_config = Arc::new(store_config);
+ if store_config.graph_type.to_lowercase().eq(VINEYARD_GRAPH) {
+ info!(
+ "Start executor with vineyard graph object id {:?}",
+ store_config.vineyard_graph_id
+ );
+ use maxgraph_runtime::store::ffi::FFIGraphStore;
+ let ffi_store = FFIGraphStore::new(store_config.vineyard_graph_id, worker_num as i32);
+ let partition_manager = ffi_store.get_partition_manager();
+ run_main(store_config, Arc::new(ffi_store), Arc::new(partition_manager));
+ } else {
+ unimplemented!("only start vineyard graph from executor")
+ }
+}
+
+fn run_main(
+ store_config: Arc,
+ graph: Arc>,
+ partition_manager: Arc,
+) where
+ V: Vertex + 'static,
+ VI: Iterator- + Send + 'static,
+ E: Edge + 'static,
+ EI: Iterator
- + Send + 'static,
+{
+ let process_partition_list = partition_manager.get_process_partition_list();
+ info!("process_partition_list: {:?}", process_partition_list);
+ let runtime_info = Arc::new(Mutex::new(RuntimeInfo::new(
+ store_config.timely_worker_per_process,
+ process_partition_list,
+ )));
+ let runtime_info_clone = runtime_info.clone();
+ let (hb_resp_sender, hb_resp_receiver) = channel();
+ let signal = Arc::new(AtomicBool::new(false));
+ let gaia_server_manager =
+ GaiaServerManager::new(hb_resp_receiver, runtime_info, signal.clone());
+
+ let partition_worker_mapping = gaia_server_manager.get_partition_worker_mapping();
+ let worker_partition_list_mapping = gaia_server_manager.get_worker_partition_list_mapping();
+ let server_manager = Box::new(gaia_server_manager);
+ let _manager_guards = ServerManager::start_server(
+ server_manager,
+ store_config.clone(),
+ Box::new(recover_prepare),
+ )
+ .unwrap();
+
+ let gaia_service = GaiaService::new(
+ store_config.clone(),
+ graph.clone(),
+ partition_manager.clone(),
+ partition_worker_mapping,
+ worker_partition_list_mapping,
+ );
+ let (_, gaia_rpc_service_port) = gaia_service.start_rpc_service();
+ let store_context = StoreContext::new(graph, partition_manager);
+ start_hb_rpc_service(
+ runtime_info_clone,
+ store_config,
+ gaia_rpc_service_port,
+ hb_resp_sender,
+ store_context,
+ );
+ thread::sleep(Duration::from_secs(u64::max_value()));
+}
+
+fn recover_prepare(prepared: &[u8]) -> Result, String> {
+ ::protobuf::parse_from_bytes::(prepared)
+ .map_err(|err| err.to_string())
+ .and_then(move |desc| {
+ info!("parse {} bytes to {:?} ", prepared.len(), desc);
+ Ok(desc.write_to_bytes().expect("query flow to bytes"))
+ })
+}
+
+fn start_hb_rpc_service(
+ runtime_info: Arc>,
+ store_config: Arc,
+ gaia_service_port: u16,
+ hb_resp_sender: Sender>,
+ store_context: StoreContext,
+) where
+ VV: 'static + Vertex,
+ VVI: 'static + Iterator
- + Send,
+ EE: 'static + Edge,
+ EEI: 'static + Iterator
- + Send,
+{
+ // build hb information
+ let mut hb_providers = Vec::new();
+ let mut hb_resp_senders = Vec::new();
+ let hb_provider = move |ref mut server_hb_req: &mut ServerHBReq| {
+ server_hb_req.set_runtimeReq(build_runtime_req(runtime_info.clone()));
+ server_hb_req
+ .mut_endpoint()
+ .set_runtimCtrlAndAsyncPort(gaia_service_port as i32);
+ };
+
+ hb_providers.push(Box::new(hb_provider));
+ hb_resp_senders.push(hb_resp_sender);
+
+ let store_config_clone = store_config.clone();
+ init_with_rpc_service(
+ store_config_clone,
+ hb_providers,
+ hb_resp_senders,
+ store_context,
+ );
+}
+
+fn build_runtime_req(runtime_info: Arc>) -> RuntimeHBReq {
+ let hb_req = runtime_info.lock().expect("Lock runtime hb req failed");
+
+ let mut runtime_req = RuntimeHBReq::new();
+ runtime_req.set_serverStatus(hb_req.get_server_status());
+ runtime_req.set_runtimePort(hb_req.get_server_port() as i32);
+ runtime_req.set_worker_num_per_process(hb_req.get_worker_num_per_process());
+ runtime_req.set_process_partition_list(hb_req.get_process_partition_list().to_vec());
+ debug!("Build runtime request {:?} in heartbeat", &runtime_req);
+
+ runtime_req
+}
+
+/// return: (aliveId, partiiton assignments)
+fn get_init_info(config: &StoreConfig) -> (u64, Vec) {
+ use maxgraph_common::proto::data_grpc::*;
+ use maxgraph_common::util::ip;
+ use maxgraph_server::client::ZKClient;
+
+ let zk_url = format!("{}/{}", config.zk_url, config.graph_name);
+ let zk = ZKClient::new(&zk_url, config.zk_timeout_ms, config.get_zk_auth());
+ let addr = zk.get_coordinator_addr();
+
+ let channel =
+ ChannelBuilder::new(Arc::new(EnvBuilder::new().build())).connect(addr.to_string().as_str());
+
+ let client = ServerDataApiClient::new(channel);
+
+ let mut request = GetExecutorAliveIdRequest::new();
+ request.set_serverId(config.worker_id);
+ request.set_ip(ip::get_local_ip());
+ let response = client.get_executor_alive_id(&request).unwrap();
+ let alive_id = response.get_aliveId();
+ let mut request = GetPartitionAssignmentRequest::new();
+ request.set_serverId(config.worker_id);
+ let response = client.get_partition_assignment(&request).unwrap();
+ let partitions = response.get_partitionId().to_vec();
+ (alive_id, partitions)
+}
+
+pub struct GaiaService
+where
+ V: Vertex + 'static,
+ VI: Iterator
- + Send + 'static,
+ E: Edge + 'static,
+ EI: Iterator
- + Send + 'static,
+{
+ store_config: Arc,
+ graph: Arc>,
+ partition_manager: Arc,
+ // mapping of partition id -> worker id
+ partition_worker_mapping: Arc>>>,
+ // mapping of worker id -> partition list
+ worker_partition_list_mapping: Arc>>>>,
+ rpc_runtime: Runtime,
+}
+
+impl GaiaService
+where
+ V: Vertex + 'static,
+ VI: Iterator
- + Send + 'static,
+ E: Edge + 'static,
+ EI: Iterator
- + Send + 'static,
+{
+ pub fn new(
+ store_config: Arc,
+ graph: Arc>,
+ partition_manager: Arc,
+ partition_worker_mapping: Arc>>>,
+ worker_partition_list_mapping: Arc>>>>,
+ ) -> GaiaService {
+ GaiaService {
+ store_config,
+ graph,
+ partition_manager,
+ partition_worker_mapping,
+ worker_partition_list_mapping,
+ rpc_runtime: Runtime::new().unwrap(),
+ }
+ }
+
+ pub fn start_rpc_service(&self) -> (String, u16) {
+ let rpc_port = self.rpc_runtime.block_on(async {
+ let query_vineyard = QueryVineyard::new(
+ self.graph.clone(),
+ self.partition_manager.clone(),
+ self.partition_worker_mapping.clone(),
+ self.worker_partition_list_mapping.clone(),
+ self.store_config.worker_num as usize,
+ );
+ let job_compiler = query_vineyard.initialize_job_compiler();
+ let service = Service::new(job_compiler);
+ let addr = format!("{}:{}", "0.0.0.0", self.store_config.rpc_port);
+ // TODO: add report in store_config
+ let rpc_service = RpcService::new(service, true);
+ let local_addr = start_rpc_server(addr.parse().unwrap(), rpc_service, false).await.unwrap();
+ local_addr.port()
+ });
+ let ip = get_local_ip();
+ info!("start rpc server on {} {}", ip, rpc_port);
+ (ip, rpc_port)
+ }
+}
diff --git a/interactive_engine/executor/gaia_runtime/src/lib.rs b/interactive_engine/executor/gaia_runtime/src/lib.rs
new file mode 100644
index 000000000000..dc7d73df61db
--- /dev/null
+++ b/interactive_engine/executor/gaia_runtime/src/lib.rs
@@ -0,0 +1,19 @@
+//
+//! Copyright 2022 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+//!
+
+#[macro_use]
+extern crate log;
+pub mod server;
diff --git a/interactive_engine/executor/gaia_runtime/src/server/manager.rs b/interactive_engine/executor/gaia_runtime/src/server/manager.rs
new file mode 100644
index 000000000000..5635b6db1e33
--- /dev/null
+++ b/interactive_engine/executor/gaia_runtime/src/server/manager.rs
@@ -0,0 +1,236 @@
+//
+//! Copyright 2020 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+
+//! timely_server_manager responsible for controlling the timely_server, including starting, running and down
+//! it also collects the status information of timely_server and reports to runtime_manager (coordinator)
+extern crate protobuf;
+
+use gaia_pegasus::Configuration;
+use maxgraph_common::proto::hb::*;
+use maxgraph_common::util::log;
+use maxgraph_runtime::server::allocate::register_tcp_listener;
+use maxgraph_runtime::server::manager::{ManagerGuards, ServerManager, ServerManagerCommon};
+use maxgraph_runtime::server::network_manager::{NetworkManager, PegasusNetworkCenter};
+use maxgraph_runtime::server::RuntimeAddress;
+use maxgraph_runtime::server::RuntimeInfo;
+use maxgraph_runtime::store::task_partition_manager::TaskPartitionManager;
+use maxgraph_store::config::StoreConfig;
+use pegasus::{network_connection, ConfigArgs};
+use pegasus::Pegasus;
+use pegasus_network::config::{NetworkConfig, ServerAddr};
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc::Receiver;
+use std::sync::{Arc, Mutex, RwLock};
+use std::thread;
+use std::time;
+use std::vec::Vec;
+
+pub struct GaiaServerManager {
+ server_manager_common: ServerManagerCommon,
+ // we preserve pegasus_runtime for heartbeat
+ pegasus_runtime: Arc