Changes in / [7b1f6d4:276a94d7]

Files:

• benchmark/Cargo.toml.in (deleted)
• benchmark/Makefile.am (modified) (1 diff)
• benchmark/bench.rs (deleted)
• benchmark/readyQ/bench.go (modified) (3 diffs)
• benchmark/readyQ/cycle.rs (modified) (5 diffs)
• benchmark/readyQ/locality.go (modified) (3 diffs)
• benchmark/readyQ/locality.rs (deleted)
• configure.ac (modified) (2 diffs)

benchmark/Makefile.am

@@ -522 +522 @@
 size-cfa$(EXEEXT):
         $(BENCH_V_CFA)$(CFACOMPILE) $(srcdir)/size/size.cfa
-
-## =========================================================================================================
-
-%-tokio$(EXEEXT): $(srcdir)/readyQ/%.rs $(srcdir)/bench.rs
-        cd $(builddir) && cargo build --release
-        cp $(builddir)/target/release/$(basename $@) $@

benchmark/readyQ/bench.go

@@ -5 +5 @@
         "flag"
         "fmt"
-        "log"
         "os"
         "runtime"
-        "runtime/pprof"
         "sync/atomic"
         "time"
@@ -45 +43 @@
 }
 
-func bench_init() func() {
+func bench_init() {
         nprocsOpt := flag.Int("p", 1, "The number of processors")
         nthreadsOpt := flag.Int("t", 1, "The number of threads")
         durationOpt := flag.Float64("d", 0, "Duration of the experiment in seconds")
         stopOpt := flag.Uint64("i", 0, "Duration of the experiment in iterations")
-        cpuprofile := flag.String("cpuprofile", "", "write cpu profile to file")
 
         flag.Parse()
@@ -75 +72 @@
 
         runtime.GOMAXPROCS(nprocs)
-
-        if (*cpuprofile) != "" {
-                f, err := os.Create(*cpuprofile)
-                if err != nil {
-                    log.Fatal(err)
-                }
-                pprof.StartCPUProfile(f)
-        }
-
-        return func() {
-                if (*cpuprofile) != "" {
-                        pprof.StopCPUProfile()
-                }
-        }
 }

benchmark/readyQ/cycle.rs

@@ +1 @@
+#[cfg(any(
+        feature = "sync time rt-threaded",
+  ))]
+
+extern crate tokio;
+
+use std::io::{self, Write};
 use std::sync::Arc;
-use std::sync::atomic::Ordering;
-use std::time::Instant;
+use std::sync::atomic::{AtomicU64, AtomicBool,Ordering};
+use std::time::{Instant,Duration};
 
 use tokio::runtime::Builder;
 use tokio::sync;
-
+use tokio::time;
+
+extern crate isatty;
+use isatty::stdout_isatty;
+
+extern crate num_format;
+use num_format::{Locale, ToFormattedString};
+
+extern crate clap;
 use clap::{Arg, App};
-use num_format::{Locale, ToFormattedString};
-
-#[path = "../bench.rs"]
-mod bench;
-
-// ==================================================
+
+use std::cell::UnsafeCell;
+use std::mem::MaybeUninit;
+use std::ops;
+
+pub struct InitializeCell<T> {
+    inner: UnsafeCell<MaybeUninit<T>>,
+}
+
+unsafe impl<T> Sync for InitializeCell<T> {}
+
+impl<T> InitializeCell<T> {
+    pub const unsafe fn new_uninitialized() -> InitializeCell<T> {
+          InitializeCell {
+                inner: UnsafeCell::new(MaybeUninit::uninit()),
+          }
+    }
+    pub const fn new(init: T) -> InitializeCell<T> {
+          InitializeCell {
+                inner: UnsafeCell::new(MaybeUninit::new(init)),
+          }
+    }
+    pub unsafe fn init(&self, init: T) {
+          (*self.inner.get()) = MaybeUninit::new(init);
+    }
+}
+
+impl<T> ops::Deref for InitializeCell<T> {
+    type Target = T;
+    fn deref(&self) -> &T {
+          unsafe {
+                &*(*self.inner.get()).as_ptr()
+          }
+    }
+}
+
+static CLOCK_MODE: InitializeCell<bool> = unsafe { InitializeCell::new_uninitialized() };
+static STOP_COUNT: InitializeCell<u64>  = unsafe { InitializeCell::new_uninitialized() };
+static DURATION: InitializeCell<f64>    = unsafe { InitializeCell::new_uninitialized() };
+static STOP         : AtomicBool = AtomicBool::new(false);
+static THREADS_LEFT : AtomicU64  = AtomicU64 ::new(10);
+
 struct Partner {
         sem: sync::Semaphore,
@@ -18 +69 @@
 }
 
-async fn partner_main(idx: usize, others: Arc<Vec<Arc<Partner>>>, exp: Arc<bench::BenchData> ) -> u64 {
+async fn partner_main(result: sync::oneshot::Sender<u64>, idx: usize, others: Arc<Vec<Arc<Partner>>> ) {
         let this = &others[idx];
         let mut count:u64 = 0;
@@ -26 +77 @@
                 count += 1;
 
-                if  exp.clock_mode && exp.stop.load(Ordering::Relaxed) { break; }
-                if !exp.clock_mode && count >= exp.stop_count { break; }
-        }
-
-        exp.threads_left.fetch_sub(1, Ordering::SeqCst);
-        count
-}
-
-// ==================================================
+                if  *CLOCK_MODE && STOP.load(Ordering::Relaxed) { break; }
+                if !*CLOCK_MODE && count >= *STOP_COUNT { break; }
+        }
+
+        THREADS_LEFT.fetch_sub(1, Ordering::SeqCst);
+        result.send( count ).unwrap();
+}
+
+fn prep(nthreads: usize, tthreads: usize) -> Vec<Arc<Partner>> {
+        let mut thddata = Vec::with_capacity(tthreads);
+        for i in 0..tthreads {
+                let pi = (i + nthreads) % tthreads;
+                thddata.push(Arc::new(Partner{
+                        sem: sync::Semaphore::new(0),
+                        next: pi,
+                }));
+        }
+        return thddata;
+}
+
+async fn wait(start: &Instant, is_tty: bool) {
+        loop {
+                time::sleep(Duration::from_micros(100000)).await;
+                let delta = start.elapsed();
+                if is_tty {
+                        print!(" {:.1}\r", delta.as_secs_f32());
+                        io::stdout().flush().unwrap();
+                }
+                if *CLOCK_MODE && delta >= Duration::from_secs_f64(*DURATION)  {
+                        break;
+                }
+                else if !*CLOCK_MODE && THREADS_LEFT.load(Ordering::Relaxed) == 0 {
+                        break;
+                }
+        }
+}
+
 fn main() {
         let options = App::new("Cycle Tokio")
-                .args(&bench::args())
+                .arg(Arg::with_name("duration")  .short("d").long("duration")  .takes_value(true).default_value("5").help("Duration of the experiments in seconds"))
+                .arg(Arg::with_name("iterations").short("i").long("iterations").takes_value(true).conflicts_with("duration").help("Number of iterations of the experiments"))
+                .arg(Arg::with_name("nthreads")  .short("t").long("nthreads")  .takes_value(true).default_value("1").help("Number of threads to use"))
+                .arg(Arg::with_name("nprocs")    .short("p").long("nprocs")    .takes_value(true).default_value("1").help("Number of processors to use"))
                 .arg(Arg::with_name("ringsize")  .short("r").long("ringsize")  .takes_value(true).default_value("1").help("Number of threads in a cycle"))
                 .get_matches();
@@ -45 +127 @@
         let nprocs    = options.value_of("nprocs").unwrap().parse::<usize>().unwrap();
 
-        let tthreads = nthreads * ring_size;
-        let exp = Arc::new(bench::BenchData::new(options, tthreads));
+        if options.is_present("iterations") {
+                unsafe{
+                        CLOCK_MODE.init( false );
+                        STOP_COUNT.init( options.value_of("iterations").unwrap().parse::<u64>().unwrap() );
+                }
+        }
+        else {
+                unsafe{
+                        CLOCK_MODE.init(true);
+                        DURATION  .init(options.value_of("duration").unwrap().parse::<f64>().unwrap());
+                }
+        }
 
         let s = (1000000 as u64).to_formatted_string(&Locale::en);
         assert_eq!(&s, "1,000,000");
 
-        let thddata : Arc<Vec<Arc<Partner>>> = Arc::new(
-                (0..tthreads).map(|i| {
-                        let pi = (i + nthreads) % tthreads;
-                        Arc::new(Partner{
-                                sem: sync::Semaphore::new(0),
-                                next: pi,
-                        })
-                }).collect()
-        );
+
+        let tthreads = nthreads * ring_size;
+        THREADS_LEFT.store(tthreads as u64, Ordering::SeqCst);
+        let thddata = Arc::new(prep(nthreads, tthreads));
 
         let mut global_counter :u64 = 0;
@@ -70 +157 @@
 
         runtime.block_on(async {
-                let threads: Vec<_> = (0..tthreads).map(|i| {
-                        tokio::spawn(partner_main(i, thddata.clone(), exp.clone()))
-                }).collect();
-                println!("Starting");
-
-                let start = Instant::now();
-
-                for i in 0..nthreads {
-                        thddata[i].sem.add_permits(1);
-                }
-
-                duration = exp.wait(&start).await;
-
-                println!("\nDone");
-
-                for i in 0..tthreads {
-                        thddata[i].sem.add_permits(1);
-                }
-
-                for t in threads {
-                        global_counter += t.await.unwrap();
+                let mut result  : Vec<sync::oneshot::Receiver::<u64>> = Vec::with_capacity(tthreads);
+                {
+                        let mut threads = Vec::with_capacity(tthreads);
+                        for i in 0..tthreads {
+                                let (s, r) = sync::oneshot::channel::<u64>();
+                                result.push(r);
+                                threads.push(tokio::spawn(partner_main(s, i, thddata.clone())));
+                        }
+                        println!("Starting");
+
+                        let is_tty = stdout_isatty();
+                        let start = Instant::now();
+
+                        for i in 0..nthreads {
+                                thddata[i].sem.add_permits(1);
+                        }
+
+                        wait(&start, is_tty).await;
+
+                        STOP.store(true, Ordering::SeqCst);
+                        duration = start.elapsed();
+
+                        println!("\nDone");
+
+                        for i in 0..tthreads {
+                                thddata[i].sem.add_permits(1);
+                        }
+
+                        for _ in 0..tthreads {
+                                global_counter += result.pop().unwrap().await.unwrap();
+                        }
                 }
         });

benchmark/readyQ/locality.go

@@ -215 +215 @@
 func main() {
         // Benchmark specific command line arguments
-        work_sizeOpt := flag.Uint64("w", 2    , "Size of the array for each threads, in words (64bit)")
-        countOpt     := flag.Uint64("c", 2    , "Number of words to touch when working (random pick, cells can be picked more than once)")
+        work_sizeOpt := flag.Uint64("w", 2    , "Number of words (uint64) per threads")
+        countOpt     := flag.Uint64("c", 2    , "Number of words (uint64) to touch")
         shareOpt     := flag.Bool  ("s", false, "Pass the work data to the next thread when blocking")
 
@@ -266 +266 @@
 
         // Join and accumulate results
-        results := NewResult()
+        global_result := NewResult()
         for i := 0; i < nthreads; i++ {
                 r := <- result
-                results.count += r.count
-                results.gmigs += r.gmigs
-                results.dmigs += r.dmigs
+                global_result.count += r.count
+                global_result.gmigs += r.gmigs
+                global_result.dmigs += r.dmigs
         }
 
@@ -280 +280 @@
         p.Printf("Number of threads      : %d\n", nthreads);
         p.Printf("Work size (64bit words): %d\n", size);
-        p.Printf("Total Operations(ops)  : %15d\n", results.count)
-        p.Printf("Total G Migrations     : %15d\n", results.gmigs)
-        p.Printf("Total D Migrations     : %15d\n", results.dmigs)
-        p.Printf("Ops per second         : %18.2f\n", float64(results.count) / delta.Seconds())
-        p.Printf("ns per ops             : %18.2f\n", float64(delta.Nanoseconds()) / float64(results.count))
-        p.Printf("Ops per threads        : %15d\n", results.count / uint64(nthreads))
-        p.Printf("Ops per procs          : %15d\n", results.count / uint64(nprocs))
-        p.Printf("Ops/sec/procs          : %18.2f\n", (float64(results.count) / float64(nprocs)) / delta.Seconds())
-        p.Printf("ns per ops/procs       : %18.2f\n", float64(delta.Nanoseconds()) / (float64(results.count) / float64(nprocs)))
-}
+        p.Printf("Total Operations(ops)  : %15d\n", global_result.count)
+        p.Printf("Total G Migrations     : %15d\n", global_result.gmigs)
+        p.Printf("Total D Migrations     : %15d\n", global_result.dmigs)
+        p.Printf("Ops per second         : %18.2f\n", float64(global_result.count) / delta.Seconds())
+        p.Printf("ns per ops             : %18.2f\n", float64(delta.Nanoseconds()) / float64(global_result.count))
+        p.Printf("Ops per threads        : %15d\n", global_result.count / uint64(nthreads))
+        p.Printf("Ops per procs          : %15d\n", global_result.count / uint64(nprocs))
+        p.Printf("Ops/sec/procs          : %18.2f\n", (float64(global_result.count) / float64(nprocs)) / delta.Seconds())
+        p.Printf("ns per ops/procs       : %18.2f\n", float64(delta.Nanoseconds()) / (float64(global_result.count) / float64(nprocs)))
+}

configure.ac

@@ -295 +295 @@
 # Some of our makefile don't need to be distributed
 AM_CONDITIONAL([CFORALL_DISTRIBUTE], [test -e $TOP_SRCDIR/autogen.sh])
-AM_COND_IF([CFORALL_DISTRIBUTE], [
-        AC_CONFIG_FILES([
+AM_COND_IF([CFORALL_DISTRIBUTE],
+        [AC_CONFIG_FILES([
                 longrun_tests/Makefile
                 benchmark/Makefile
@@ -302 +302 @@
                 tools/Makefile
                 tools/prettyprinter/Makefile
-        ])
-
-        AC_OUTPUT(benchmark/Cargo.toml)
-])
+                ])])
 
 AC_CONFIG_LINKS([tests/test.py:tests/test.py])