Initial commit

This is the first working version with proper thread synchronization.
2017-09-07 18:30:04 -05:00 · 2017-09-07 18:30:04 -05:00 · 5b022693b0
commit 5b022693b0
4 changed files with 256 additions and 0 deletions
--- a/.gitignore
+++ b/.gitignore
@ -0,0 +1,2 @@
+/target/
+**/*.rs.bk
--- a/Cargo.lock
+++ b/Cargo.lock
@ -0,0 +1,121 @@
+[root]
+name = "iostress"
+version = "0.1.0"
+dependencies = [
+ "clap 2.26.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "ansi_term"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "atty"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.29 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "bitflags"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "clap"
+version = "2.26.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "ansi_term 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "atty 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "bitflags 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "strsim 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "term_size 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "textwrap 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "unicode-segmentation 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "unicode-width 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "vec_map 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "kernel32-sys"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi-build 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "libc"
+version = "0.2.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "strsim"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "term_size"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.29 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "textwrap"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "term_size 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "unicode-width 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "unicode-segmentation"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "unicode-width"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "vec_map"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "winapi"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "winapi-build"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[metadata]
+"checksum ansi_term 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)" = "23ac7c30002a5accbf7e8987d0632fa6de155b7c3d39d0067317a391e00a2ef6"
+"checksum atty 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "d912da0db7fa85514874458ca3651fe2cddace8d0b0505571dbdcd41ab490159"
+"checksum bitflags 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)" = "4efd02e230a02e18f92fc2735f44597385ed02ad8f831e7c1c1156ee5e1ab3a5"
+"checksum clap 2.26.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2267a8fdd4dce6956ba6649e130f62fb279026e5e84b92aa939ac8f85ce3f9f0"
+"checksum kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7507624b29483431c0ba2d82aece8ca6cdba9382bff4ddd0f7490560c056098d"
+"checksum libc 0.2.29 (registry+https://github.com/rust-lang/crates.io-index)" = "8a014d9226c2cc402676fbe9ea2e15dd5222cd1dd57f576b5b283178c944a264"
+"checksum strsim 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)" = "b4d15c810519a91cf877e7e36e63fe068815c678181439f2f29e2562147c3694"
+"checksum term_size 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "e2b6b55df3198cc93372e85dd2ed817f0e38ce8cc0f22eb32391bfad9c4bf209"
+"checksum textwrap 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "f728584ea33b0ad19318e20557cb0a39097751dbb07171419673502f848c7af6"
+"checksum unicode-segmentation 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "a8083c594e02b8ae1654ae26f0ade5158b119bd88ad0e8227a5d8fcd72407946"
+"checksum unicode-width 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "bf3a113775714a22dcb774d8ea3655c53a32debae63a063acc00a91cc586245f"
+"checksum vec_map 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "887b5b631c2ad01628bbbaa7dd4c869f80d3186688f8d0b6f58774fbe324988c"
+"checksum winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)" = "167dc9d6949a9b857f3451275e911c3f44255842c1f7a76f33c55103a909087a"
+"checksum winapi-build 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "2d315eee3b34aca4797b2da6b13ed88266e6d612562a0c46390af8299fc699bc"
--- a/Cargo.toml
+++ b/Cargo.toml
@ -0,0 +1,7 @@
+[package]
+name = "iostress"
+version = "0.1.0"
+authors = ["Dustin C. Hatch <dustin.hatch@firemon.com>"]
+
+[dependencies]
+clap = "2.26"
--- a/src/main.rs
+++ b/src/main.rs
@ -0,0 +1,126 @@
+extern crate clap;
+
+use std::process;
+use std::sync;
+use std::thread;
+
+
+struct Args {
+    num_threads: usize,
+}
+
+
+struct IOStressContext {
+    data: sync::RwLock<Vec<u8>>,
+    barrier: sync::Barrier,
+    lock: sync::Mutex<usize>,
+    cond: sync::Condvar,
+}
+
+
+impl IOStressContext {
+    fn new(t: usize) -> IOStressContext {
+        IOStressContext {
+            data: sync::RwLock::new(Vec::with_capacity(4096)),
+            barrier: sync::Barrier::new(t + 1),
+            lock: sync::Mutex::new(0),
+            cond: sync::Condvar::new(),
+        }
+    }
+}
+
+
+fn parse_args() -> Args {
+    let app = clap::App::new("iostress")
+        .version("0.1")
+        .author("Dustin C. Hatch <dustin.hatch@firemon.com>")
+        .about("I/O stress test tool")
+        .arg(clap::Arg::with_name("threads")
+            .short("t")
+            .long("threads")
+            .value_name("COUNT")
+            .default_value("1")
+            .help("Number of simultaneous tests")
+            .takes_value(true));
+    let matches = app.get_matches();
+
+    Args {
+        num_threads: match matches.value_of("threads").unwrap().parse() {
+            Ok(n) => n,
+            Err(_) => {
+                eprintln!("Invalid thread count");
+                process::exit(2);
+            },
+        },
+    }
+}
+
+
+fn worker_thread(ctx: sync::Arc<IOStressContext>) {
+    let thread = thread::current();
+    let name = thread.name().unwrap();
+    let mut prev_ctr: usize = 0;
+    loop {
+        {
+            let mut ctr = ctx.lock.lock().unwrap();
+            while *ctr <= prev_ctr {
+                println!("{} waiting on condition", name);
+                ctr = ctx.cond.wait(ctr).unwrap();
+            }
+            println!("{} woke up, counter: {} -> {}", name, prev_ctr, *ctr);
+            prev_ctr = *ctr;
+        }
+        {
+            let data = ctx.data.read().unwrap();
+            // Here is where the actual processing would occur
+            println!("{} data: {:?}", name, *data);
+            if (*data).len() == 0 {
+                println!("{} done!", name);
+                break;
+            }
+        }
+        println!("{} waiting at barrier", name);
+        ctx.barrier.wait();
+    }
+    println!("{} waiting at final barrier", name);
+    ctx.barrier.wait();
+    println!("{} terminating", name);
+}
+
+
+fn main() {
+    let args = parse_args();
+
+    let ctx = sync::Arc::new(IOStressContext::new(args.num_threads));
+    let mut threads = Vec::with_capacity(args.num_threads);
+    for i in 0..args.num_threads {
+        let c = ctx.clone();
+        let t = thread::Builder::new().name(format!("thread{}", i + 1));
+        threads.push(t.spawn(move || worker_thread(c)).unwrap());
+    }
+
+    for i in 0..5 {
+        {
+            let mut data = ctx.data.write().unwrap();
+            if i == 4 {
+                (*data).clear();
+            } else {
+                // Here is where data would be fetched and sent to the workers
+                (*data).push(i + 1);
+            }
+        }
+        {
+            let mut ctr = ctx.lock.lock().unwrap();
+            *ctr += 1;
+            println!("Notifying threads");
+            ctx.cond.notify_all();
+        }
+        println!("Waiting for threads to finish iteration {}", i + 1);
+        ctx.barrier.wait();
+    }
+
+    for thread in threads {
+        println!("waiting for {}", thread.thread().name().unwrap());
+        thread.join().unwrap();
+    }
+}