Bencher upgrade

bencher/.gitignore

@@ -0,0 +1 @@
+prepare_data

bencher/Cargo.toml

@@ -7,19 +7,20 @@ edition = "2021"
 
 [dependencies]
 goose = "^0.16.4"
-tokio = {version="1.1",features=["process"]}
+tokio = { version = "1.1", features = ["process"] }
 clap = { version = "4.5.7", features = ["derive"] }
 image = "0.24.1"
 rand = "0.8.5"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
+serde_yaml = "0.9"
 base64 = "0.13"
 rust-s3 = "0.34.0"
 lazy_static = "1.4.0"
-reqwest = { version="0.12.5", features = ["json"]}
+reqwest = { version = "0.12.5", features = ["json"] }
 async-trait = "0.1.74"
 enum_dispatch = "0.3.13"
-serde_yaml = "0.9"
 interprocess = "2.2.0"
 tracing = "0.1.40"
-tracing-subscriber = "0.3"
+tracing-subscriber = "0.3"
+futures = "0.3.30"

bencher/app_fn_entries.yaml

@@ -0,0 +1,134 @@
+minio:
+  endpoint: "http://192.168.31.54:9009"
+  access_key: "minioadmin"
+  secret_key: "minioadmin123"
+  compose_path: "../middlewares/minio/"
+
+models:
+  simple_demo:
+    # name: "Simple Demo"
+    # description: "A simple demo function"
+    # language: "java"
+    # runtime: "java11"
+    # handler: "test.functions.Simple"
+    # memory: 256
+    # timeout: 30
+    prepare_data: []
+    prepare_scripts: []
+    fns:
+      simple:
+        args: {}
+
+  img_resize:
+    fns:
+      resize:
+        args: 
+          use_minio: true
+          image_s3_path: "image_to_resize.png"
+          target_width: 50
+          target_height: 50
+    prepare_data: ["image_to_resize.png"]
+    prepare_scripts: 
+    - |
+      from PIL import Image, ImageDraw
+      import random
+
+      def generate_random_image(width, height, output_path):
+          # 创建一个空白图像
+          image = Image.new('RGB', (width, height), 'white')
+          draw = ImageDraw.Draw(image)
+
+          # 生成随机颜色的像素
+          for x in range(width):
+              for y in range(height):
+                  r = random.randint(0, 255)
+                  g = random.randint(0, 255)
+                  b = random.randint(0, 255)
+                  draw.point((x, y), fill=(r, g, b))
+
+          # 保存图像
+          image.save(output_path)
+          print(f"Image saved to {output_path}")
+
+      # 指定图片的宽度和高度
+      width = 800
+      height = 600
+      output_path = 'image_to_resize.png'
+
+      # 生成随机图片
+      generate_random_image(width, height, output_path)
+
+  word_count:
+    prepare_data: 
+    - large_text_file.txt
+    prepare_scripts: 
+    - |
+      import os
+
+      def generate_paragraph():
+          paragraph = """
+          In a far-off land, nestled among the mountains, lies a small village known for its unique culture and friendly inhabitants. Each spring, the villagers hold a grand festival to celebrate the arrival of the flowering season. Visitors from all corners gather to enjoy this beautiful moment. During the festival, people don traditional attire and dance joyfully, filling the village with laughter and cheer.
+          """
+          return paragraph
+
+      def generate_large_text_file(file_path, target_size_gb):
+          target_size_bytes = target_size_gb * 1024 * 1024 * 1024
+          paragraph = generate_paragraph()
+          paragraph_length = len(paragraph.encode('utf-8'))
+
+          with open(file_path, 'w') as file:
+              while os.path.getsize(file_path) < target_size_bytes:
+                  file.write(paragraph)
+
+          final_size = os.path.getsize(file_path) / (1024 * 1024 * 1024)
+          print(f"Generated file size: {final_size:.2f} GB")
+
+      # 生成一个 1 GB 大小的文件
+      file_path = 'large_text_file.txt'
+      target_size_gb = 1
+      generate_large_text_file(file_path, target_size_gb)
+    fns:
+      count:
+        args:
+          text_s3_path: "large_text_file.txt"
+
+# 函数配置项变体
+replicas: {}
+benchlist: {}
+#   img_resize2: # 函数 和 是否使用minio排列组合，每种两个
+#     source: img_resize
+
+#   img_resize3:
+#     source: img_resize
+#     args:
+#       use_minio: false
+#   img_resize4:
+#     source: img_resize
+#     args:
+#       use_minio: false
+
+
+
+# benchlist:
+#   img_resize:
+#   img_resize2:
+#   img_resize3:
+#   img_resize4:
+
+
+# parallel_composition:
+#   parallel:
+#     args:
+#       loopTime: 10000000
+#       parallelIndex: 100
+#   sequential:
+#     args:
+#       loopTime: 10000000
+
+
+
+#   fns:
+#     split:
+#       args:
+#         text_s3_path: "random_words.txt"    
+#     count:

bencher/scripts/run_with_all_log.sh

@@ -0,0 +1,4 @@
+# run with following command:
+# source scripts/run_with_all_log.sh
+
+export RUST_LOG=trace,debug

bencher/src/demo_word_count.rs → bencher/src/_deperacated/demo_word_count.rs

@@ -97,10 +97,24 @@ impl SpecTarget for WordCount {
             receive_resp_time - start_call_ms
         );
 
+        // | cold start time
+        //   |
+        //     | fn_start_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!(
+            "- cold start time: {}",
+            if bf_exec_time > recover_begin_time {
+                recover_begin_time - bf_exec_time
+            } else {
+                0
+            }
+        );
+        println!(
+            "- cold start time2: {}",
+            fn_start_ms - recover_begin_time.max(req_arrive_time)
+        );
         println!("- exec time:{}", fn_end_ms - fn_start_ms);
         if fn_end_ms > receive_resp_time {
             println!(

bencher/src/common_prepare.rs

@@ -0,0 +1,74 @@
+use std::{collections::HashMap, path::PathBuf, process::Stdio};
+
+use tokio::{fs, process::Command};
+
+use crate::config::Config;
+
+/// return each app datas
+/// app->[data1,data2]
+pub async fn prepare_data(
+    target_apps: Vec<String>,
+    config: &Config,
+) -> HashMap<String, Vec<PathBuf>> {
+    let mut prepare_data = HashMap::new();
+    let model_apps: Vec<String> = target_apps
+        .clone()
+        .into_iter()
+        .filter(|app| config.models.contains_key(app))
+        .collect();
+
+    for app in model_apps {
+        fs::create_dir_all(PathBuf::from("prepare_data").join(&app))
+            .await
+            .unwrap();
+        let app_entry = config.models.get(&app).unwrap();
+        for (i, script) in app_entry.prepare_scripts.iter().enumerate() {
+            let script_path = PathBuf::from("prepare_data")
+                .join(&app)
+                .join(format!("prepare_{}.py", i));
+            let script_dir = PathBuf::from("prepare_data").join(&app);
+            let abs_script_dir = script_dir.canonicalize().unwrap();
+            // let abs_script_path = script_path.canonicalize().unwrap();
+            fs::write(&script_path, script).await.unwrap();
+
+            tracing::debug!(
+                "prepare data for {} with script {}",
+                app,
+                script_path.to_str().unwrap()
+            );
+            let res = Command::new("python3")
+                .args(&[script_path.file_name().unwrap().to_str().unwrap(), &*app])
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped())
+                .current_dir(abs_script_dir)
+                .spawn()
+                .unwrap()
+                .wait()
+                .await
+                .unwrap();
+            if !res.success() {
+                panic!(
+                    "prepare data for {} with script {} failed",
+                    app,
+                    script_path.to_str().unwrap()
+                );
+            }
+        }
+
+        for data in app_entry.prepare_data.iter() {
+            let data_path = PathBuf::from("prepare_data").join(&app).join(data);
+            if !data_path.exists() {
+                panic!("prepare data failed {:?}", data);
+            }
+            prepare_data
+                .entry(app.to_owned())
+                .or_insert(vec![])
+                .push(data_path);
+        }
+        // for data in app_entry.prepare_data.iter() {
+        //     prepare_data.push(data.clone());
+        // }
+    }
+
+    prepare_data
+}

bencher/src/config.rs

@@ -0,0 +1,104 @@
+use serde_yaml::Value;
+use std::{collections::HashMap, fs::File, io::BufReader};
+
+#[derive(serde::Deserialize, Debug, Clone)]
+pub struct FnDetails {
+    pub args: Option<HashMap<String, Value>>,
+}
+
+#[derive(serde::Deserialize, Debug, Clone)]
+pub struct AppFnEntry {
+    pub prepare_data: Vec<String>,
+    pub prepare_scripts: Vec<String>,
+    pub fns: HashMap<String, FnDetails>,
+}
+
+#[derive(serde::Deserialize, Debug, Clone)]
+pub struct AppFnReplica {
+    pub source: String,
+    pub fns: HashMap<String, FnDetails>,
+}
+
+// app:
+//   prepare_data:
+//   - file1
+//   prepare_scripts:
+//   - script1
+//   fns:
+//     fn1:
+//       args:
+// pub type AppFnEntries = HashMap<String, AppFnEntry>;
+
+#[derive(serde::Deserialize, Debug, Clone)]
+pub struct MinioConfig {
+    pub endpoint: String,
+    pub access_key: String,
+    pub secret_key: String,
+    pub compose_path: String,
+}
+
+impl MinioConfig {
+    pub fn one_line(&self) -> String {
+        format!("{},{},{}", self.endpoint, self.access_key, self.secret_key)
+    }
+}
+
+#[derive(serde::Deserialize, Debug, Clone)]
+pub struct Config {
+    pub models: HashMap<String, AppFnEntry>,
+    pub replicas: HashMap<String, AppFnReplica>,
+    pub benchlist: HashMap<String, ()>,
+    pub minio: MinioConfig,
+}
+
+impl Config {
+    // todo: add replica support
+    pub fn get(&self, app: &str) -> Option<&AppFnEntry> {
+        self.models.get(app)
+    }
+
+    pub fn get_fn_details(&self, app: &str, func: &str) -> Option<FnDetails> {
+        let mut fndetail = if self.models.contains_key(app) {
+            self.models.get(app).unwrap().fns.get(func).cloned()
+        } else if let Some(replica) = self.replicas.get(app) {
+            // replica args will cover model args
+            let source = replica.source.clone();
+            let source_fn_details = self.get_fn_details(&source, func);
+            if let Some(mut source_fn_details) = source_fn_details {
+                // cover by replica args
+                for (key, value) in replica.fns.get(func).unwrap().args.as_ref().unwrap() {
+                    source_fn_details
+                        .args
+                        .as_mut()
+                        .unwrap()
+                        .insert(key.clone(), value.clone());
+                }
+                Some(source_fn_details)
+            } else {
+                None
+            }
+        } else {
+            None
+        };
+        fndetail.map(|mut fndetail| {
+            let args = fndetail.args.as_mut().unwrap();
+            if let Some(Value::Bool(true)) = args.get("use_minio") {
+                args.insert("use_minio".to_owned(), Value::from(self.minio.one_line()));
+            } else {
+                // remove use_minio
+                args.remove("use_minio");
+            }
+
+            fndetail
+        })
+    }
+}
+
+pub fn load_config() -> Config {
+    let f = File::open("app_fn_entries.yaml").expect("open app_fn_entries.yaml failed");
+    let freader = BufReader::new(f);
+    let app_fn_entries: Config = serde_yaml::from_reader(freader).unwrap_or_else(|e| {
+        panic!("parse 'app_fn_entries.yaml' failed with {:?}", e);
+    });
+    app_fn_entries
+}