source: benchmark/readyQ/locality.rs @ dc9dd94

Last change on this file since dc9dd94 was aec2c022, checked in by Thierry Delisle <tdelisle@…>, 2 years ago

Clean-up the benchmarks a little

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