Merge pull request #102 from arkedge/exec_kble_dump

kble-dump

Cargo.toml

@@ -15,6 +15,7 @@ members = [
   "kble-c2a",
   "kble-eb90",
   "kble-tcp",
+  "kble-dump",
 ]
 
 [workspace.dependencies]

kble-dump/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "kble-dump"
+description = "Virtual Harness Toolkit"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+readme.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[build-dependencies]
+notalawyer-build.workspace = true
+
+[dependencies]
+anyhow.workspace = true
+futures.workspace = true
+tokio = { workspace = true, features = ["full"] }
+kble-socket = { workspace = true, features = ["stdio", "tungstenite"] }
+tokio-util.workspace = true
+bytes.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+clap.workspace = true
+notalawyer.workspace = true
+notalawyer-clap.workspace = true
+rmp-serde = "1.3.0"
+chrono = { version = "0.4.38", features = ["serde"] }
+miniz_oxide = "0.7.3"
+serde = { workspace = true, features = ["derive"] }

kble-dump/README.md

@@ -0,0 +1,26 @@
+## 概説
+kble-dump は、デバッグを目的として通信データをダンプし、またダンプしたデータを再生するためのプログラムである。
+
+kble-dump は、サブコマンドとして record と replay をもつ。
+
+kble-dump record は、kble から受信したそのままkbleに送信すると同時に、kble から受信したメッセージの内容をダンプ形式でファイルに保存する。
+
+kble-dump replay は、kble から受信した内容を全て無視し、引数として与えらえたダンプ形式のファイルに記録された内容を kble に 送信する。送信のタイミングは、ダンプ形式のファイルに記録されたタイムスタンプの相対的タイミングと可能な限り一致させる。
+
+## ダンプ形式
+出力ファイルは、レコードをMessagePack形式で並べたものである。
+
+レコードは、未圧縮レコード1つをdeflate圧縮したものである。
+
+未圧縮のレコードの構造は12バイトのタイムスタンプと可変長のデータ部からなる（下図参照）。
+タイムスタンプは、kble-dumpをデータが通過した時刻を UNIX EPOCH からの経過時間で表したものである。
+データ部の内容はkble-dumpを通過したメッセージのバイト列をそのまま記録したものである。
+
+```
++---------------------------------------------------+------+
+| timestamp (12 bytes)                              | data |
++-------------------+-------------------------------+      |
+| seconds (8 bytes) | subseconds in nanos (4 bytes) |      |
++-------------------+-------------------------------+------+
+```
+

kble-dump/build.rs

@@ -0,0 +1,3 @@
+fn main() {
+    notalawyer_build::build();
+}

kble-dump/src/main.rs

@@ -0,0 +1,133 @@
+use anyhow::Result;
+use clap::{Parser, Subcommand};
+use futures::{SinkExt, StreamExt};
+use notalawyer_clap::*;
+use tracing_subscriber::{prelude::*, EnvFilter};
+
+#[derive(Parser, Debug)]
+#[clap(author, version, about, long_about = None)]
+struct Args {
+    #[clap(subcommand)]
+    command: Commands,
+}
+
+use std::path::PathBuf;
+
+#[derive(Subcommand, Debug)]
+enum Commands {
+    Record { output_dir: PathBuf },
+    Replay { input_file: PathBuf },
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::fmt::layer()
+                .with_ansi(false)
+                .with_writer(std::io::stderr),
+        )
+        .with(EnvFilter::from_default_env())
+        .init();
+
+    let args = Args::parse_with_license_notice(include_notice!());
+    match args.command {
+        Commands::Record { output_dir } => run_record(&output_dir).await,
+        Commands::Replay { input_file } => run_replay(&input_file).await,
+    }
+}
+
+use std::borrow::Cow;
+struct DumpRecord<'a> {
+    timestamp: std::time::SystemTime,
+    data: Cow<'a, [u8]>,
+}
+
+impl<'a> DumpRecord<'a> {
+    fn now(data: &'a [u8]) -> Self {
+        Self {
+            timestamp: std::time::SystemTime::now(),
+            data: Cow::Borrowed(data),
+        }
+    }
+
+    fn write_to(&self, mut writer: impl std::io::Write) -> anyhow::Result<()> {
+        let since_epoch = self.timestamp.duration_since(std::time::UNIX_EPOCH)?;
+        writer.write_all(&since_epoch.as_secs().to_le_bytes())?;
+        writer.write_all(&since_epoch.subsec_nanos().to_le_bytes())?;
+        writer.write_all(&self.data[..])?;
+        Ok(())
+    }
+
+    fn read_from(mut reader: impl std::io::Read) -> anyhow::Result<Self> {
+        let mut secs = [0u8; 8];
+        reader.read_exact(&mut secs)?;
+        let secs = u64::from_le_bytes(secs);
+        let mut nanos = [0u8; 4];
+        reader.read_exact(&mut nanos)?;
+        let nanos = u32::from_le_bytes(nanos);
+        let timestamp = std::time::UNIX_EPOCH + std::time::Duration::new(secs, nanos);
+        let mut data = vec![];
+        reader.read_to_end(&mut data)?;
+        Ok(Self {
+            timestamp,
+            data: Cow::Owned(data),
+        })
+    }
+}
+
+async fn run_record(output_dir: &PathBuf) -> Result<()> {
+    tokio::fs::create_dir_all(output_dir).await?;
+    let time = chrono::Local::now().format("%Y%m%d_%H%M%S_%f");
+    let path = output_dir.join(format!("dump_{}.bin", time));
+    tracing::info!("Recording to {:?}", path);
+    let mut file = tokio::fs::File::create(&path).await?;
+
+    let (mut tx, mut rx) = kble_socket::from_stdio().await;
+    while let Some(data) = rx.next().await {
+        let data = data?;
+        let mut buf = vec![];
+        let record = DumpRecord::now(&data[..]);
+        record.write_to(&mut buf)?;
+
+        use miniz_oxide::deflate::compress_to_vec;
+        let compressed = compress_to_vec(&buf, 6);
+        let bin = rmp_serde::encode::to_vec(&compressed)?;
+        use tokio::io::AsyncWriteExt;
+        file.write_all(&bin).await?;
+        tx.send(data).await?;
+    }
+    Ok(())
+}
+
+async fn run_replay(input_file: &PathBuf) -> Result<()> {
+    let file = std::fs::File::open(input_file)?;
+    let mut file = std::io::BufReader::new(file);
+
+    let mut replay_time_offset = None;
+
+    let (mut tx, rx) = kble_socket::from_stdio().await;
+    tokio::spawn(/* just consume everything */ rx.count());
+    while let Ok(compressed) = rmp_serde::decode::from_read::<_, Vec<u8>>(&mut file) {
+        let decompressed =
+            miniz_oxide::inflate::decompress_to_vec(&compressed).map_err(|e| anyhow::anyhow!(e))?;
+        let record = DumpRecord::read_from(&mut decompressed.as_slice())?;
+        let replay_time_offset = match replay_time_offset {
+            Some(offset) => offset,
+            None => {
+                let offset_value = std::time::SystemTime::now().duration_since(record.timestamp)?;
+                replay_time_offset = Some(offset_value);
+                offset_value
+            }
+        };
+        let replay_time = record.timestamp + replay_time_offset;
+        let sleep_time = replay_time
+            .duration_since(std::time::SystemTime::now())
+            // Err if the time is in the past
+            .unwrap_or_else(|_| std::time::Duration::from_secs(0));
+        tokio::time::sleep(sleep_time).await;
+        tx.send(record.data.into_owned().into()).await?;
+    }
+
+    Ok(())
+}