source: benchmark/readyQ/locality.rs @ 751e2eb

arm-ehjacob/cs343-translationnew-ast-unique-expr
Last change on this file since 751e2eb was 751e2eb, checked in by Thierry Delisle <tdelisle@…>, 13 months ago

Added bench.rs for common benchmark rust code

  • Property mode set to 100644
File size: 9.2 KB
Line 
1use std::ptr;
2use std::sync::Arc;
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::time::Instant;
5use std::thread::{self, ThreadId};
6
7use tokio::runtime::Builder;
8use tokio::sync;
9
10use clap::{App, Arg};
11use num_format::{Locale, ToFormattedString};
12use rand::Rng;
13
14#[path = "../bench.rs"]
15mod bench;
16
17// ==================================================
18struct MyData {
19        data: Vec<u64>,
20        ttid: ThreadId,
21        _id: usize,
22}
23
24impl MyData {
25        fn new(id: usize, size: usize) -> MyData {
26                MyData {
27                        data: vec![0; size],
28                        ttid: thread::current().id(),
29                        _id: id,
30                }
31        }
32
33        fn moved(&mut self, ttid: ThreadId) -> u64 {
34                if self.ttid == ttid {
35                        return 0;
36                }
37                self.ttid = ttid;
38                return 1;
39        }
40
41        fn access(&mut self, idx: usize) {
42                let l = self.data.len();
43                self.data[idx % l] += 1;
44        }
45}
46
47struct MyDataPtr {
48        ptr: *mut MyData,
49}
50
51unsafe impl std::marker::Send for MyDataPtr{}
52
53// ==================================================
54struct MyCtx {
55        s: sync::Semaphore,
56        d: MyDataPtr,
57        ttid: ThreadId,
58        _id: usize,
59}
60
61impl MyCtx {
62        fn new(d: *mut MyData, id: usize) -> MyCtx {
63                MyCtx {
64                        s: sync::Semaphore::new(0),
65                        d: MyDataPtr{ ptr: d },
66                        ttid: thread::current().id(),
67                        _id: id
68                }
69        }
70
71        fn moved(&mut self, ttid: ThreadId) -> u64 {
72                if self.ttid == ttid {
73                        return 0;
74                }
75                self.ttid = ttid;
76                return 1;
77        }
78}
79// ==================================================
80// Atomic object where a single thread can wait
81// May exchanges data
82struct MySpot {
83        ptr: AtomicU64,
84        _id: usize,
85}
86
87impl MySpot {
88        fn new(id: usize) -> MySpot {
89                let r = MySpot{
90                        ptr: AtomicU64::new(0),
91                        _id: id,
92                };
93                r
94        }
95
96        fn one() -> u64 {
97                1
98        }
99
100        // Main handshake of the code
101        // Single seat, first thread arriving waits
102        // Next threads unblocks current one and blocks in its place
103        // if share == true, exchange data in the process
104        async fn put( &self, ctx: &mut MyCtx, data: MyDataPtr, share: bool) -> (*mut MyData, bool) {
105                {
106                        // Attempt to CAS our context into the seat
107                        let raw = {
108                                loop {
109                                        let expected = self.ptr.load(Ordering::Relaxed) as u64;
110                                        if expected == MySpot::one() { // Seat is closed, return
111                                                let r: *const MyData = ptr::null();
112                                                return (r as *mut MyData, true);
113                                        }
114                                        let got = self.ptr.compare_and_swap(expected, ctx as *mut MyCtx as u64, Ordering::SeqCst);
115                                        if got == expected {
116                                                break expected;// We got the seat
117                                        }
118                                }
119                        };
120
121                        // If we aren't the fist in, wake someone
122                        if raw != 0 {
123                                let val: &mut MyCtx = unsafe{ &mut *(raw as *mut MyCtx) };
124                                // If we are sharing, give them our data
125                                if share {
126                                        val.d.ptr = data.ptr;
127                                }
128
129                                // Wake them up
130                                val.s.add_permits(1);
131                        }
132                }
133
134                // Block once on the seat
135                ctx.s.acquire().await.forget();
136
137                // Someone woke us up, get the new data
138                let ret = ctx.d.ptr;
139                return (ret, false);
140        }
141
142        // Shutdown the spot
143        // Wake current thread and mark seat as closed
144        fn release(&self) {
145                let val = self.ptr.swap(MySpot::one(), Ordering::SeqCst);
146                if val == 0 {
147                        return
148                }
149
150                // Someone was there, release them
151                unsafe{ &mut *(val as *mut MyCtx) }.s.add_permits(1)
152        }
153}
154
155// ==================================================
156// Struct for result, Go doesn't support passing tuple in channels
157struct Result {
158        count: u64,
159        gmigs: u64,
160        dmigs: u64,
161}
162
163impl Result {
164        fn new() -> Result {
165                Result{ count: 0, gmigs: 0, dmigs: 0}
166        }
167
168        fn add(&mut self, o: Result) {
169                self.count += o.count;
170                self.gmigs += o.gmigs;
171                self.dmigs += o.dmigs;
172        }
173}
174
175// ==================================================
176// Random number generator, Go's native one is to slow and global
177fn __xorshift64( state: &mut u64 ) -> usize {
178        let mut x = *state;
179        x ^= x << 13;
180        x ^= x >> 7;
181        x ^= x << 17;
182        *state = x;
183        x as usize
184}
185
186// ==================================================
187// Do some work by accessing 'cnt' cells in the array
188fn work(data: &mut MyData, cnt: u64, state : &mut u64) {
189        for _ in 0..cnt {
190                data.access(__xorshift64(state))
191        }
192}
193
194async fn local(start: Arc<sync::Barrier>, idata: MyDataPtr, spots : Arc<Vec<MySpot>>, cnt: u64, share: bool, id: usize, exp: Arc<bench::BenchData>) -> Result{
195        let mut state = rand::thread_rng().gen::<u64>();
196        let mut data = idata;
197        let mut ctx = MyCtx::new(data.ptr, id);
198        let _size = unsafe{ &mut *data.ptr }.data.len();
199
200        // Prepare results
201        let mut r = Result::new();
202
203        // Wait for start
204        start.wait().await;
205
206        // Main loop
207        loop {
208                // Touch our current data, write to invalidate remote cache lines
209                work(unsafe{ &mut *data.ptr }, cnt, &mut state);
210
211                // Wait on a random spot
212                let i = (__xorshift64(&mut state) as usize) % spots.len();
213                let closed = {
214                        let (d, c) = spots[i].put(&mut ctx, data, share).await;
215                        data = MyDataPtr{ ptr: d };
216                        c
217                };
218
219                // Check if the experiment is over
220                if closed { break }                                                   // yes, spot was closed
221                if  exp.clock_mode && exp.stop.load(Ordering::Relaxed) { break }  // yes, time's up
222                if !exp.clock_mode && r.count >= exp.stop_count { break }         // yes, iterations reached
223
224                assert_ne!(data.ptr as *const MyData, ptr::null());
225
226                let d = unsafe{ &mut *data.ptr };
227
228                // Check everything is consistent
229                debug_assert_eq!(d.data.len(), _size);
230
231                // write down progress and check migrations
232                let ttid = thread::current().id();
233                r.count += 1;
234                r.gmigs += ctx .moved(ttid);
235                r.dmigs += d.moved(ttid);
236        }
237
238        exp.threads_left.fetch_sub(1, Ordering::SeqCst);
239        r
240}
241
242
243// ==================================================
244fn main() {
245        let options = App::new("Locality Tokio")
246                .args(&bench::args())
247                .arg(Arg::with_name("size")      .short("w").long("worksize")  .takes_value(true).default_value("2").help("Size of the array for each threads, in words (64bit)"))
248                .arg(Arg::with_name("work")      .short("c").long("workcnt")   .takes_value(true).default_value("2").help("Number of words to touch when working (random pick, cells can be picked more than once)"))
249                .arg(Arg::with_name("share")     .short("s").long("share")     .takes_value(true).default_value("true").help("Pass the work data to the next thread when blocking"))
250                .get_matches();
251
252        let nthreads   = options.value_of("nthreads").unwrap().parse::<usize>().unwrap();
253        let nprocs     = options.value_of("nprocs").unwrap().parse::<usize>().unwrap();
254        let wsize      = options.value_of("size").unwrap().parse::<usize>().unwrap();
255        let wcnt       = options.value_of("work").unwrap().parse::<u64>().unwrap();
256        let share      = options.value_of("share").unwrap().parse::<bool>().unwrap();
257
258        // Check params
259        if ! (nthreads > nprocs) {
260                panic!("Must have more threads than procs");
261        }
262
263        let s = (1000000 as u64).to_formatted_string(&Locale::en);
264        assert_eq!(&s, "1,000,000");
265
266        let exp = Arc::new(bench::BenchData::new(options, nprocs));
267        let mut results = Result::new();
268
269        let mut elapsed : std::time::Duration = std::time::Duration::from_secs(0);
270
271        let mut data_arrays : Vec<MyData> = (0..nthreads).map(|i| MyData::new(i, wsize)).rev().collect();
272        let spots : Arc<Vec<MySpot>> = Arc::new((0..nthreads - nprocs).map(|i| MySpot::new(i)).rev().collect());
273        let barr = Arc::new(sync::Barrier::new(nthreads + 1));
274
275        let runtime = Builder::new_multi_thread()
276                .worker_threads(nprocs)
277                .enable_all()
278                .build()
279                .unwrap();
280
281        runtime.block_on(async
282                {
283                        let thrds: Vec<_> = (0..nthreads).map(|i| {
284                                debug_assert!(i < data_arrays.len());
285
286                                runtime.spawn(local(
287                                        barr.clone(),
288                                        MyDataPtr{ ptr: &mut data_arrays[i] },
289                                        spots.clone(),
290                                        wcnt,
291                                        share,
292                                        i,
293                                        exp.clone(),
294                                ))
295                        }).collect();
296
297
298                        println!("Starting");
299
300                        let start = Instant::now();
301                        barr.wait().await;
302
303                        elapsed = exp.wait(&start).await;
304
305                        println!("\nDone");
306
307                        // release all the blocked threads
308                        for s in &* spots {
309                                s.release();
310                        }
311
312                        println!("Threads released");
313
314                        // Join and accumulate results
315                        for t in thrds {
316                                results.add( t.await.unwrap() );
317                        }
318
319                        println!("Threads joined");
320                }
321        );
322
323        println!("Duration (ms)          : {}", (elapsed.as_millis()).to_formatted_string(&Locale::en));
324        println!("Number of processors   : {}", (nprocs).to_formatted_string(&Locale::en));
325        println!("Number of threads      : {}", (nthreads).to_formatted_string(&Locale::en));
326        println!("Work size (64bit words): {}", (wsize).to_formatted_string(&Locale::en));
327        println!("Total Operations(ops)  : {:>15}", (results.count).to_formatted_string(&Locale::en));
328        println!("Total G Migrations     : {:>15}", (results.gmigs).to_formatted_string(&Locale::en));
329        println!("Total D Migrations     : {:>15}", (results.dmigs).to_formatted_string(&Locale::en));
330        println!("Ops per second         : {:>15}", (((results.count as f64) / elapsed.as_secs() as f64) as u64).to_formatted_string(&Locale::en));
331        println!("ns per ops             : {:>15}", ((elapsed.as_nanos() as f64 / results.count as f64) as u64).to_formatted_string(&Locale::en));
332        println!("Ops per threads        : {:>15}", (results.count / nthreads as u64).to_formatted_string(&Locale::en));
333        println!("Ops per procs          : {:>15}", (results.count / nprocs as u64).to_formatted_string(&Locale::en));
334        println!("Ops/sec/procs          : {:>15}", ((((results.count as f64) / nprocs as f64) / elapsed.as_secs() as f64) as u64).to_formatted_string(&Locale::en));
335        println!("ns per ops/procs       : {:>15}", ((elapsed.as_nanos() as f64 / (results.count as f64 / nprocs as f64)) as u64).to_formatted_string(&Locale::en));
336}
Note: See TracBrowser for help on using the repository browser.