DirectIO timer added

Author: denisbsu

Cargo.lock

@@ -21,6 +21,7 @@ futures-util = "0.3.31"
 uuid = { version = "1.18.1", features = ["v4"] }
 flate2 = "1.1.4"
 tar = "0.4.44"
+libc = "0.2.177"
 
 [dev-dependencies]
 criterion = "0.3"

Cargo.toml

@@ -1,5 +1,8 @@
-use std::time::Duration;
+use std::{alloc::{self, Layout}, cmp::min, fs::{File, OpenOptions}, io::{Read, Write}, os::unix::fs::OpenOptionsExt, time::{Duration, Instant}};
 
+use anyhow::Error;
+use futures::StreamExt;
+use indicatif::{ProgressBar, ProgressStyle};
 use sqd_query::ParquetChunk;
 
 pub type Dataset = String;
@@ -44,3 +47,84 @@ pub async fn execute_query(
     rx.await
         .unwrap_or_else(|_| Err(anyhow::anyhow!("Query processor didn't produce a result")))
 }
+
+pub fn time_direct_io(file_path: String) -> Result<Duration, Error> {
+    if std::fs::exists(&file_path)? {
+        std::fs::remove_file(&file_path)?;
+    };
+
+    let mut options = OpenOptions::new();
+    options.read(true);
+    options.write(true);
+    options.create(true);
+    options.custom_flags(libc::O_DIRECT | libc::O_SYNC);
+    let block_size = 4096;
+    let iterations = 10;
+    assert_eq!(block_size % 512, 0);
+    let elapsed;
+    let layout = Layout::from_size_align(block_size, 512).unwrap();
+    {
+        let mut ff = options.open(&file_path).unwrap();
+        let s = Instant::now();
+        let mut acc = 0;
+        unsafe {
+            let ptr = alloc::alloc(layout);
+            for i in 0..block_size {
+                *ptr.wrapping_add(i) = 1;
+            }
+
+            let wrap = core::slice::from_raw_parts_mut(ptr, block_size);
+            for _ in 0..iterations {
+                let _ = ff.write(wrap);
+            }
+
+            for _ in 0..iterations {
+                let _ = ff.read(wrap);
+                for i in 0..block_size {
+                    acc += wrap[i];
+                }
+            }
+            libc::free(ptr as *mut libc::c_void);
+        }
+        assert_eq!(acc, 0);
+        elapsed = s.elapsed();    
+    }
+
+    std::fs::remove_file(&file_path)?;
+
+    Ok(elapsed)
+}
+
+pub async fn download_file(url: &str, path: &str) -> Result<f32, String> {
+    let client = reqwest::Client::new();
+    let res = client
+        .get(url)
+        .send()
+        .await
+        .or(Err(format!("Failed to GET from '{}'", &url)))?;
+    let total_size = res
+        .content_length()
+        .ok_or(format!("Failed to get content length from '{}'", &url))?;
+
+    let pb = ProgressBar::new(total_size);
+    pb.set_style(ProgressStyle::with_template("{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})").unwrap()
+        .progress_chars("#>-"));
+    pb.set_message(format!("Downloading {url}").to_owned());
+
+    let mut file = File::create(path).or(Err(format!("Failed to create file '{path}'")))?;
+    let mut downloaded: u64 = 0;
+    let mut stream = res.bytes_stream();
+
+    while let Some(item) = stream.next().await {
+        let chunk = item.or(Err("Error while downloading file".to_owned()))?;
+        file.write_all(&chunk)
+            .or(Err("Error while writing to file".to_owned()))?;
+        let new = min(downloaded + (chunk.len() as u64), total_size);
+        downloaded = new;
+        pb.set_position(new);
+    }
+
+    pb.finish_with_message(format!("Downloaded {url} to {path}").to_owned());
+    let speed = total_size as f64 / pb.elapsed().as_secs_f64() / 1024.0 / 1024.0;
+    Ok(speed as f32)
+}

src/lib.rs

@@ -1,16 +1,13 @@
-use std::cmp::min;
 use std::{
     env,
     fs::{self, File},
-    io::Write,
     time::{Duration, Instant},
 };
 
-use benchmark::execute_query;
+use benchmark::{download_file, execute_query, time_direct_io};
 use clap::Parser;
 use flate2::read::GzDecoder;
 use futures::future::join_all;
-use futures_util::StreamExt;
 use indicatif::{ProgressBar, ProgressStyle};
 use serde::{Deserialize, Serialize};
 use tar::Archive;
@@ -91,38 +88,6 @@ pub struct Cli {
     pub force_update: bool,
 }
 
-pub async fn download_file(url: &str, path: &str) -> Result<(), String> {
-    let client = reqwest::Client::new();
-    let res = client
-        .get(url)
-        .send()
-        .await
-        .or(Err(format!("Failed to GET from '{}'", &url)))?;
-    let total_size = res
-        .content_length()
-        .ok_or(format!("Failed to get content length from '{}'", &url))?;
-
-    let pb = ProgressBar::new(total_size);
-    pb.set_style(ProgressStyle::with_template("{msg}\n{spinner:.green} [{elapsed_precise}] [{wide_bar:.cyan/blue}] {bytes}/{total_bytes} ({bytes_per_sec}, {eta})").unwrap()
-        .progress_chars("#>-"));
-    pb.set_message(format!("Downloading {url}").to_owned());
-
-    let mut file = File::create(path).or(Err(format!("Failed to create file '{path}'")))?;
-    let mut downloaded: u64 = 0;
-    let mut stream = res.bytes_stream();
-
-    while let Some(item) = stream.next().await {
-        let chunk = item.or(Err("Error while downloading file".to_owned()))?;
-        file.write_all(&chunk)
-            .or(Err("Error while writing to file".to_owned()))?;
-        let new = min(downloaded + (chunk.len() as u64), total_size);
-        downloaded = new;
-        pb.set_position(new);
-    }
-
-    pb.finish_with_message(format!("Downloaded {url} to {path}").to_owned());
-    Ok(())
-}
 
 #[tokio::main]
 async fn main() {
@@ -150,7 +115,10 @@ async fn main() {
         let id = Uuid::new_v4();
         dir.push(id.to_string());
 
-        let _ = download_file(&url, dir.to_str().unwrap()).await;
+        let big_download = download_file(&url, dir.to_str().unwrap()).await;
+        let download_speed = big_download.unwrap();
+        println!("Download speed: {:?}", download_speed);
+
         let tar_gz = File::open(dir).unwrap();
         let tar = GzDecoder::new(tar_gz);
         let mut archive = Archive::new(tar);
@@ -161,6 +129,19 @@ async fn main() {
         bench_set = serde_yaml::from_reader(reader).unwrap();
     }
 
+    {
+        // let file_path = "/home/denis/worker_bench/data/testfile.bin";
+        let file_path = "./data/testfile.bin";
+        let mut io_time = Duration::new(0, 0);
+        let iterations = 10;
+        for _ in 0..iterations {
+            io_time += time_direct_io(file_path.to_string()).unwrap();
+        }
+        io_time = io_time / iterations;
+        println!("Direct IO time: {:?}", io_time);
+    }
+    // exit(0);
+    
     let width = bench_set.len();
     let max_concurrent = args.max_concurrency;
     let total_samples = args.total_samples;