feat: Integrated Java KV interface

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/

bencher/src/demo_javakv_test.rs

@@ -0,0 +1,178 @@
+use super::Cli;
+use base64::encode;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
+use reqwest::Response;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+use std::fmt::{format, Debug};
+use std::fs::{self, File};
+use std::io::BufReader;
+use std::process::{self, Command as Process};
+use std::time::{SystemTime, UNIX_EPOCH};
+use std::io::Write;
+
+use super::PlatformOpsBind;
+use crate::parse_app::App;
+use crate::{Metric, PlatformOps, SpecTarget, BUCKET};
+
+#[derive(Default)]
+pub struct JavaKvTest(Option<PlatformOpsBind>);
+
+impl SpecTarget for JavaKvTest {
+ fn set_platform(&mut self, platform: PlatformOpsBind) {
+ self.0 = Some(platform);
+ }
+
+ fn get_platform(&mut self) -> &mut PlatformOpsBind {
+ self.0.as_mut().unwrap()
+ }
+
+ async fn prepare_once(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ self.get_platform().upload_fn("javakv_test", "").await;
+ }
+
+ async fn call_once(&mut self, cli: Cli) -> Metric {
+
+ let arg = Args {
+ kv_test_arg: "javakv_test_arg".to_string(),
+ };
+
+ let start_call_ms = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+ let output = self
+ .get_platform()
+ .call_fn("javakv_test", "javakv", &serde_json::to_value(arg).unwrap())
+ .await;
+
+ print!("output {}\n",output);
+ let res: serde_json::Value = serde_json::from_str(&output).unwrap();
+ let req_arrive_time = res.get("req_arrive_time").unwrap().as_u64().unwrap();
+ let bf_exec_time = res.get("bf_exec_time").unwrap().as_u64().unwrap();
+ let recover_begin_time = res.get("recover_begin_time").unwrap().as_u64().unwrap();
+ let fn_start_ms = res.get("fn_start_time").unwrap().as_u64().unwrap();
+ let fn_end_ms = res.get("fn_end_time").unwrap().as_u64().unwrap();
+ let receive_resp_time = SystemTime::now()
+ .duration_since(UNIX_EPOCH)
+ .expect("Time went backwards")
+ .as_millis() as u64;
+
+ println!(
+ "\ntotal request latency: {}",
+ receive_resp_time - start_call_ms
+ );
+
+ println!("- req trans time: {}", req_arrive_time - start_call_ms);
+ println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ if fn_end_ms > receive_resp_time {
+ println!(
+ "- system time is not synced, lag with {} ms",
+ fn_end_ms - receive_resp_time
+ );
+ } else {
+ println!("- receive resp time: {}", receive_resp_time - fn_end_ms);
+ }
+
+ Metric {
+ start_call_time: start_call_ms,
+ req_arrive_time,
+ bf_exec_time,
+ recover_begin_time,
+ fn_start_time: fn_start_ms,
+ fn_end_time: fn_end_ms,
+ receive_resp_time,
+ }
+ }
+
+ async fn prepare_first_call(&mut self, seed: String, cli: Cli) {
+ self.get_platform().remove_all_fn().await;
+ }
+
+ async fn call_first_call(&mut self, cli: Cli) {
+ let mut metrics = vec![];
+ for _ in 0..20 {
+ self.get_platform().upload_fn("javakv_test", "").await;
+ metrics.push(self.call_once(cli.clone()).await);
+ }
+
+ println!(
+ "\ntotal request latency: {}",
+ metrics.iter().map(|v| v.get_total_req()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+
+ println!(
+ "- req trans time: {}",
+ metrics.iter().map(|v| v.get_req_trans_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- app verify time: {}",
+ metrics.iter().map(|v| v.get_app_verify_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time: {}",
+ metrics.iter().map(|v| v.get_cold_start_time()).sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- cold start time2: {}",
+ metrics
+ .iter()
+ .map(|v| v.get_cold_start_time2())
+ .sum::<u64>() as f32
+ / metrics.len() as f32
+ );
+
+ println!(
+ "- exec time: {}",
+ metrics.iter().map(|v| v.get_exec_time()).sum::<u64>() as f32 / metrics.len() as f32
+ );
+ // println!("- app verify time: {}", bf_exec_time - req_arrive_time);
+ // println!("- cold start time: {}", recover_begin_time - bf_exec_time);
+ // println!("- cold start time2: {}", fn_start_ms - recover_begin_time);
+ // println!("- exec time:{}", fn_end_ms - fn_start_ms);
+ }
+
+ fn app(&self) -> App {
+ App::JavaKvTest
+ }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Args {
+ kv_test_arg: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Resp {
+ javakv_test: String,
+}
+
+// 将字符串转换为u64哈希值
+fn hash_str(s: &str) -> u64 {
+ use std::hash::{Hash, Hasher};
+ let mut hasher = std::collections::hash_map::DefaultHasher::new();
+ s.hash(&mut hasher);
+ hasher.finish()
+}
+
+fn generate_random_text<R: Rng>(rng: &mut R, length: usize) -> Vec<u8> {
+ let mut text = Vec::with_capacity(length);
+ let chars: &[u8] = b"abcdefghijklmnopqrstuvwxyz ";
+
+ for _ in 0..length {
+ text.push(*chars.choose(rng).unwrap());
+ }
+
+ text
+}

bencher/src/main.rs

@@ -1,4 +1,5 @@
 mod demo_img_resize;
+mod demo_javakv_test;
 mod demo_parallel;
 mod demo_sequential;
 mod demo_word_count;
@@ -106,6 +107,7 @@ enum SpecTargetBind {
  WordCount(demo_word_count::WordCount),
  Parallel(demo_parallel::Parallel),
  Sequential(demo_sequential::Sequential),
+ JavaKvTest(demo_javakv_test::JavaKvTest),
 }
 
 /// unit: ms
@@ -223,6 +225,8 @@ async fn main() -> Result<(), GooseError> {
  SpecTargetBind::from(demo_parallel::Parallel::default())
  } else if cli.sequential > 0 {
  SpecTargetBind::from(demo_sequential::Sequential::default())
+ } else if cli.javakv_test > 0 {
+ SpecTargetBind::from(demo_javakv_test::JavaKvTest::default())
  } else {
  unreachable!()
  };

bencher/src/parse.rs

@@ -19,6 +19,9 @@ pub struct Cli {
  #[arg(long, action = clap::ArgAction::Count)]
  pub sequential: u8,
 
+ #[arg(long, action = clap::ArgAction::Count)]
+ pub javakv_test: u8,
+
  #[arg(long, action = clap::ArgAction::Count)]
  pub with_ow: u8,
 

bencher/src/parse_app.rs

@@ -5,6 +5,7 @@ pub enum App {
  WordCount,
  Parallel,
  Sequential,
+ JavaKvTest,
 }
 
 impl From<&Cli> for App {
@@ -30,6 +31,7 @@ impl ToString for App {
  App::WordCount => "word_count".to_owned(),
  App::Parallel => "parallel".to_owned(),
  App::Sequential => "sequential".to_owned(),
+ App::JavaKvTest => "java_kv_test".to_owned(),
  }
  }
 }

demos/_java_serverless_lib/core/pom.xml

@@ -15,7 +15,10 @@
  </parent>
 
 	<properties>
- <java.version>17</java.version>
+ <!-- <java.version>1.8</java.version> -->
+ <maven.compiler.source>1.8</maven.compiler.source>
+ <maven.compiler.target>1.8</maven.compiler.target>
+ <spring.boot.version>2.3.12.RELEASE</spring.boot.version>
  <protobuf.version>3.15.8</protobuf.version>
 	</properties>
 

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BeanConfig.java

@@ -24,4 +24,9 @@ public RpcHandleOwner rpcHandleOwner() {
  public CracManager cracManager() {
  return new CracManager();
  }
+
+ @Bean
+ public KvBatch kvBatch(){
+ return new KvBatch("javakv-test", "javakv-test-func", 9999);
+ }
 }

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/BootArgCheck.java

@@ -5,7 +5,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ConfigurableApplicationContext;;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.DefaultApplicationArguments;
 import org.springframework.core.env.Environment;

demos/_java_serverless_lib/core/src/main/java/io/serverless_lib/KvBatch.java

@@ -0,0 +1,124 @@
+package io.serverless_lib;
+
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+
+import process_rpc_proto.ProcessRpcProto.KvRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequests;
+import process_rpc_proto.ProcessRpcProto.KvResponses;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvPutRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvGetRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvDeleteRequest;
+import process_rpc_proto.ProcessRpcProto.KvRequest.KvLockRequest;
+import process_rpc_proto.ProcessRpcProto.KvPair;
+import process_rpc_proto.ProcessRpcProto.KeyRange;
+
+
+public class KvBatch {
+ private KvRequests batchArgs; // 用于发送批处理请求的KvRequests
+ private KvResponses result; // 用于接收响应的KvResponses
+ private int taskId; // 任务 ID
+
+ @Autowired 
+ UdsBackend uds;
+
+ // 构造函数，初始化 KvRequests 和 KvResponses
+ public KvBatch(String app, String func, int taskId) {
+ this.batchArgs = KvRequests.newBuilder().setApp(app).setFunc(func).setPrevKvOpeid(8888).build();
+ this.result = KvResponses.newBuilder().build();
+ this.taskId = taskId;
+ // TODO 注入this
+ // uds.kvBatch = this;
+ }
+
+ // 设置批处理请求
+ public void setBatchArgs(String app, String func, List<KvRequest> requests, long prevKvOpeId) {
+ KvRequests.Builder builder = KvRequests.newBuilder();
+ builder.setApp(app).setFunc(func).addAllRequests(requests).setPrevKvOpeid(prevKvOpeId);
+ this.batchArgs = builder.build();
+ }
+
+ // 添加单个 PUT 请求
+ public void addPutRequest(byte[] key, byte[] value) {
+ KvPair kvPair = KvPair.newBuilder().setKey(com.google.protobuf.ByteString.copyFrom(key))
+ .setValue(com.google.protobuf.ByteString.copyFrom(value)).build();
+ KvPutRequest putRequest = KvPutRequest.newBuilder().setKv(kvPair).build();
+ KvRequest request = KvRequest.newBuilder().setSet(putRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 GET 请求
+ public void addGetRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvGetRequest getRequest = KvGetRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setGet(getRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 DELETE 请求
+ public void addDeleteRequest(byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvDeleteRequest deleteRequest = KvDeleteRequest.newBuilder().setRange(range).build();
+ KvRequest request = KvRequest.newBuilder().setDelete(deleteRequest).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 添加单个 LOCK 请求
+ public void addLockRequest(boolean readOrWrite, List<Integer> releaseIds, byte[] startKey, byte[] endKey) {
+ KeyRange range = KeyRange.newBuilder().setStart(com.google.protobuf.ByteString.copyFrom(startKey))
+ .setEnd(com.google.protobuf.ByteString.copyFrom(endKey)).build();
+ KvLockRequest.Builder lockBuilder = KvLockRequest.newBuilder().setReadOrWrite(readOrWrite).setRange(range);
+ lockBuilder.addAllReleaseId(releaseIds);
+ KvRequest request = KvRequest.newBuilder().setLock(lockBuilder.build()).build();
+
+ this.batchArgs = this.batchArgs.toBuilder().addRequests(request).build();
+ }
+
+ // 发送请求并接收响应
+ public void sendBatch() throws Exception {
+ // 打包成 UdsPack 发送数据
+ UdsPack pack = new UdsPack(this.batchArgs, this.taskId);
+
+ System.out.println("KvBatch sendBatch");
+ // 发送请求
+ uds.send(pack);
+
+ // TODO 从 udsbackend 接收结果数据
+ // this.result = ;
+ }
+
+ // 获取请求内容
+ public KvRequests getBatchArgs() {
+ return this.batchArgs;
+ }
+
+ public Integer getTackId() {
+ return this.taskId;
+ }
+
+ // 获取响应内容
+ public KvResponses getResult() {
+ return this.result;
+ }
+
+ public void setResult(KvResponses result){
+ this.result = result;
+ }
+}
+
+
+
+/* 
+用法示例：
+KvBatch kvBatch = new KvBatch();
+kvBatch.addPutRequest("key1".getBytes(), "value1".getBytes());
+kvBatch.addGetRequest("key1".getBytes(), "key2".getBytes());
+KvResponses responses = kvBatch.sendBatch(); 
+*/