Changeset 65c9208

Timestamp:

May 10, 2022, 12:28:54 PM (3 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, ast-experimental, master, pthread-emulation, qualifiedEnum

Children:

e07187d

Parents:

3613e25

Message:

Changed transfer benchmark to be more consistent with other rmit benchmarks

Location:

benchmark/readyQ

Files:

• transfer.cfa (modified) (6 diffs)
• transfer.cpp (modified) (8 diffs)
• transfer.go (modified) (11 diffs)
• transfer.rs (modified) (10 diffs)

benchmark/readyQ/transfer.cfa

@@ -14 +14 @@
 
 bool exhaust = false;
+volatile bool estop = false;
+
 
 thread$ * the_main;
@@ -35 +37 @@
 static void waitgroup() {
         Time start = timeHiRes();
-        for(i; nthreads) {
+        OUTER: for(i; nthreads) {
                 PRINT( sout | "Waiting for :" | i | "(" | threads[i]->idx | ")"; )
                 while( threads[i]->idx != lead_idx ) {
@@ -42 +44 @@
                                 print_stats_now( bench_cluster, CFA_STATS_READY_Q | CFA_STATS_IO );
                                 serr | "Programs has been blocked for more than 5 secs";
-                                exit(1);
+                                estop = true;
+                                unpark( the_main );
+                                break OUTER;
                         }
                 }
@@ -59 +63 @@
 static void lead(MyThread & this) {
         this.idx = ++lead_idx;
-        if(lead_idx > stop_count) {
+        if(lead_idx > stop_count || estop) {
                 PRINT( sout | "Leader" | this.id | "done"; )
                 unpark( the_main );
@@ -100 +104 @@
                         wait( this );
                 }
-                if(lead_idx > stop_count) break;
+                if(lead_idx > stop_count || estop) break;
         }
 }
@@ -172 +176 @@
         sout | "Number of processors    : " | nprocs;
         sout | "Number of threads       : " | nthreads;
-        sout | "Total Operations(ops)   : " | stop_count;
+        sout | "Total Operations(ops)   : " | lead_idx - 1;
         sout | "Threads parking on wait : " | (exhaust ? "yes" : "no");
         sout | "Rechecking              : " | rechecks;
+        sout | "ns per transfer         : " | (end - start)`dms / lead_idx;
 
 

benchmark/readyQ/transfer.cpp

@@ -12 +12 @@
 
 bool exhaust = false;
+volatile bool estop = false;
 
 bench_sem the_main;
@@ -42 +43 @@
                                 if( to_miliseconds(timeHiRes() - start) > 5'000 ) {
                                         std::cerr << "Programs has been blocked for more than 5 secs" << std::endl;
-                                        std::exit(1);
+                                        estop = true;
+                                        the_main.post();
+                                        goto END;
                                 }
                         }
                 }
+                END:;
                 PRINT( std::cout | "Waiting done"; )
         }
@@ -59 +63 @@
         void lead() {
                 this->idx = ++lead_idx;
-                if(lead_idx > stop_count) {
+                if(lead_idx > stop_count || estop) {
                         PRINT( std::cout << "Leader " << this->id << " done" << std::endl; )
                         the_main.post();
@@ -88 +92 @@
         }
 
-        static void main(void * arg) {
-                MyThread & self = *reinterpret_cast<MyThread*>(arg);
+        static void main(MyThread * arg) {
+                MyThread & self = *arg;
                 self.park();
 
@@ -101 +105 @@
                                 self.wait();
                         }
-                        if(lead_idx > stop_count) break;
+                        if(lead_idx > stop_count || estop) break;
                 }
         }
@@ -144 +148 @@
                         for(size_t i = 0; i < nthreads; i++) {
                                 threads[i] = new MyThread( i );
-                                handles[i] = new Fibre( MyThread::main, threads[i] );
+                                handles[i] = new Fibre();
+                                handles[i]->run( MyThread::main, threads[i] );
                         }
 
@@ -164 +169 @@
                                 PRINT( std::cout << i << " joined" << std::endl; )
                                 rechecks += thrd.rechecks;
-                                // delete( handles[i] );
                                 delete( threads[i] );
                         }
@@ -176 +180 @@
         std::cout << "Number of processors    : " << nprocs << std::endl;
         std::cout << "Number of threads       : " << nthreads << std::endl;
-        std::cout << "Total Operations(ops)   : " << stop_count << std::endl;
+        std::cout << "Total Operations(ops)   : " << (lead_idx - 1) << std::endl;
         std::cout << "Threads parking on wait : " << (exhaust ? "yes" : "no") << std::endl;
         std::cout << "Rechecking              : " << rechecks << std::endl;
+        std::cout << "ns per transfer         : " << std::fixed << (((double)(end - start)) / (lead_idx)) << std::endl;
 
 

benchmark/readyQ/transfer.go

@@ -6 +6 @@
         "math/rand"
         "os"
+        "regexp"
         "runtime"
         "sync/atomic"
@@ -16 +17 @@
         id uint64
         idx uint64
+        estop uint64
         seed uint64
 }
@@ -34 +36 @@
 
 func NewLeader(size uint64) (*LeaderInfo) {
-        this := &LeaderInfo{0, 0, uint64(os.Getpid())}
+        this := &LeaderInfo{0, 0, 0, uint64(os.Getpid())}
 
         r := rand.Intn(10)
@@ -51 +53 @@
 }
 
-func waitgroup(idx uint64, threads [] MyThread) {
+func waitgroup(leader * LeaderInfo, idx uint64, threads [] MyThread, main_sem chan struct {}) {
         start := time.Now()
+        Outer:
         for i := 0; i < len(threads); i++ {
                 // fmt.Fprintf(os.Stderr, "Waiting for :%d (%d)\n", threads[i].id, atomic.LoadUint64(&threads[i].idx) );
@@ -61 +64 @@
                         if delta.Seconds() > 5 {
                                 fmt.Fprintf(os.Stderr, "Programs has been blocked for more than 5 secs")
-                                os.Exit(1)
+                                atomic.StoreUint64(&leader.estop, 1);
+                                main_sem <- (struct {}{})
+                                break Outer
                         }
                 }
@@ -74 +79 @@
                 if i != me {
                         // debug!( "Leader waking {}", i);
+                        defer func() {
+                                if err := recover(); err != nil {
+                                        fmt.Fprintf(os.Stderr, "Panic occurred: %s\n", err)
+                                }
+                        }()
                         threads[i].sem <- (struct {}{})
                 }
@@ -84 +94 @@
         atomic.StoreUint64(&leader.idx, nidx);
 
-        if nidx > stop_count {
+        if nidx > stop_count || atomic.LoadUint64(&leader.estop) != 0 {
                 // debug!( "Leader {} done", this.id);
                 main_sem <- (struct {}{})
@@ -92 +102 @@
         // debug!( "====================\nLeader no {} : {}", nidx, this.id);
 
-        waitgroup(nidx, threads);
+        waitgroup(leader, nidx, threads, main_sem);
 
         leader.next( uint64(len(threads)) );
@@ -146 +156 @@
                         waitleader( exhaust, leader, &threads[me], &r )
                 }
-                if atomic.LoadUint64(&leader.idx) > stop_count { break; }
+                if atomic.LoadUint64(&leader.idx) > stop_count || atomic.LoadUint64(&leader.estop) != 0 { break; }
         }
 
@@ -155 +165 @@
 func main() {
         // Benchmark specific command line arguments
-        exhaustOpt := flag.Bool("e", false, "Whether or not threads that have seen the new epoch should park instead of yielding.")
+        exhaustOpt := flag.String("e", "no", "Whether or not threads that have seen the new epoch should park instead of yielding.")
 
         // General benchmark initialization and deinitialization
-        defer bench_init()()
-
-        exhaust := *exhaustOpt;
+        bench_init()
+
+        exhaustVal := *exhaustOpt;
+
+        var exhaust bool
+        re_yes := regexp.MustCompile("[Yy]|[Yy][Ee][Ss]")
+        re_no  := regexp.MustCompile("[Nn]|[Nn][Oo]")
+        if re_yes.Match([]byte(exhaustVal)) {
+                exhaust = true
+        } else if re_no.Match([]byte(exhaustVal)) {
+                exhaust = false
+        } else {
+                fmt.Fprintf(os.Stderr, "Unrecognized exhaust(-e) option '%s'\n", exhaustVal)
+                os.Exit(1)
+        }
+
         if clock_mode {
-                fmt.Fprintf(os.Stderr, "Programs does not support fixed duration mode")
+                fmt.Fprintf(os.Stderr, "Programs does not support fixed duration mode\n")
                 os.Exit(1)
         }
@@ -215 +238 @@
                 ws = "no"
         }
-        p.Printf("Duration (ms)           : %f\n", delta.Milliseconds() )
+        p.Printf("Duration (ms)           : %d\n", delta.Milliseconds() )
         p.Printf("Number of processors    : %d\n", nprocs )
         p.Printf("Number of threads       : %d\n", nthreads )
-        p.Printf("Total Operations(ops)   : %15d\n", stop_count )
+        p.Printf("Total Operations(ops)   : %15d\n", (leader.idx - 1) )
         p.Printf("Threads parking on wait : %s\n", ws)
         p.Printf("Rechecking              : %d\n", rechecks )
-}
+        p.Printf("ns per transfer         : %f\n", float64(delta.Nanoseconds()) / float64(leader.idx) )
+}

benchmark/readyQ/transfer.rs

@@ -6 +6 @@
 use std::hint;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::time::{Instant,Duration};
 
@@ -44 +44 @@
                         match val {
                                 "yes" => true,
+                                "Y" => true,
+                                "y" => true,
                                 "no"  => false,
+                                "N"  => false,
+                                "n"  => false,
                                 "maybe" | "I don't know" | "Can you repeat the question?" => {
                                         eprintln!("Lines for 'Malcolm in the Middle' are not acceptable values of parameter 'exhaust'");
@@ -64 +68 @@
         id: AtomicUsize,
         idx: AtomicUsize,
+        estop: AtomicBool,
         seed: u128,
 }
@@ -72 +77 @@
                         id: AtomicUsize::new(nthreads),
                         idx: AtomicUsize::new(0),
+                        estop: AtomicBool::new(false),
                         seed: process::id() as u128
                 };
@@ -100 +106 @@
 }
 
-fn waitgroup(idx: usize, threads: &Vec<Arc<MyThread>>) {
+fn waitgroup(leader: &LeaderInfo, idx: usize, threads: &Vec<Arc<MyThread>>, main_sem: &sync::Semaphore) {
         let start = Instant::now();
-        for t in threads {
+        'outer: for t in threads {
                 debug!( "Waiting for :{} ({})", t.id, t.idx.load(Ordering::Relaxed) );
                 while t.idx.load(Ordering::Relaxed) != idx {
@@ -108 +114 @@
                         if start.elapsed() > Duration::from_secs(5) {
                                 eprintln!("Programs has been blocked for more than 5 secs");
-                                std::process::exit(1);
+                                leader.estop.store(true, Ordering::Relaxed);
+                                main_sem.add_permits(1);
+                                break 'outer;
                         }
                 }
@@ -131 +139 @@
         leader.idx.store(nidx, Ordering::Relaxed);
 
-        if nidx as u64 > exp.stop_count {
+        if nidx as u64 > exp.stop_count || leader.estop.load(Ordering::Relaxed) {
                 debug!( "Leader {} done", this.id);
                 main_sem.add_permits(1);
@@ -139 +147 @@
         debug!( "====================\nLeader no {} : {}", nidx, this.id);
 
-        waitgroup(nidx, threads);
+        waitgroup(leader, nidx, threads, main_sem);
 
         leader.next( threads.len() );
@@ -192 +200 @@
                         wait( exhaust, &leader, &threads[me], &mut rechecks ).await;
                 }
-                if leader.idx.load(Ordering::Relaxed) as u64 > exp.stop_count { break; }
+                if leader.idx.load(Ordering::Relaxed) as u64 > exp.stop_count || leader.estop.load(Ordering::Relaxed) { break; }
         }
 
@@ -273 +281 @@
         println!("Number of processors    : {}", (nprocs).to_formatted_string(&Locale::en));
         println!("Number of threads       : {}", (nthreads).to_formatted_string(&Locale::en));
-        println!("Total Operations(ops)   : {:>15}", (exp.stop_count).to_formatted_string(&Locale::en));
+        println!("Total Operations(ops)   : {:>15}", (leader.idx.load(Ordering::Relaxed) - 1).to_formatted_string(&Locale::en));
         println!("Threads parking on wait : {}", if exhaust { "yes" } else { "no" });
         println!("Rechecking              : {}", rechecks );
-}
+        println!("ns per transfer         : {}", ((duration.as_nanos() as f64) / leader.idx.load(Ordering::Relaxed) as f64));
+
+}