source: benchmark/readyQ/locality.rs @ 3f8baf4

arm-ehenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 3f8baf4 was 3f8baf4, checked in by Thierry Delisle <tdelisle@…>, 2 years ago

Fixed the padding

  • Property mode set to 100644
File size: 9.4 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_and_swap(expected, ctx as *mut MyCtx as u64, Ordering::SeqCst);
127                                        if got == 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("size") .short("w").long("worksize").takes_value(true).default_value("2").help("Size of the array for each threads, in words (64bit)"))
260                .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)"))
261                .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"))
262                .get_matches();
263
264        let nthreads   = options.value_of("nthreads").unwrap().parse::<usize>().unwrap();
265        let nprocs     = options.value_of("nprocs").unwrap().parse::<usize>().unwrap();
266        let wsize      = options.value_of("size").unwrap().parse::<usize>().unwrap();
267        let wcnt       = options.value_of("work").unwrap().parse::<u64>().unwrap();
268        let share      = options.value_of("share").unwrap().parse::<bool>().unwrap();
269
270        // Check params
271        if ! (nthreads > nprocs) {
272                panic!("Must have more threads than procs");
273        }
274
275        let s = (1000000 as u64).to_formatted_string(&Locale::en);
276        assert_eq!(&s, "1,000,000");
277
278        let exp = Arc::new(bench::BenchData::new(options, nprocs));
279        let mut results = Result::new();
280
281        let mut elapsed : std::time::Duration = std::time::Duration::from_secs(0);
282
283        let mut data_arrays : Vec<MyData> = (0..nthreads).map(|i| MyData::new(i, wsize)).rev().collect();
284        let spots : Arc<Vec<MySpot>> = Arc::new((0..nthreads - nprocs).map(|i| MySpot::new(i)).rev().collect());
285        let barr = Arc::new(sync::Barrier::new(nthreads + 1));
286
287        let runtime = Builder::new_multi_thread()
288                .worker_threads(nprocs)
289                .enable_all()
290                .build()
291                .unwrap();
292
293        runtime.block_on(async
294                {
295                        let thrds: Vec<_> = (0..nthreads).map(|i| {
296                                debug_assert!(i < data_arrays.len());
297
298                                runtime.spawn(local(
299                                        barr.clone(),
300                                        MyDataPtr{ ptr: &mut data_arrays[i] },
301                                        spots.clone(),
302                                        wcnt,
303                                        share,
304                                        i,
305                                        exp.clone(),
306                                ))
307                        }).collect();
308
309
310                        println!("Starting");
311
312                        let start = Instant::now();
313                        barr.wait().await;
314
315                        elapsed = exp.wait(&start).await;
316
317                        println!("\nDone");
318
319                        // release all the blocked threads
320                        for s in &* spots {
321                                s.release();
322                        }
323
324                        println!("Threads released");
325
326                        // Join and accumulate results
327                        for t in thrds {
328                                results.add( t.await.unwrap() );
329                        }
330
331                        println!("Threads joined");
332                }
333        );
334
335        println!("Duration (ms)          : {}", (elapsed.as_millis()).to_formatted_string(&Locale::en));
336        println!("Number of processors   : {}", (nprocs).to_formatted_string(&Locale::en));
337        println!("Number of threads      : {}", (nthreads).to_formatted_string(&Locale::en));
338        println!("Work size (64bit words): {}", (wsize).to_formatted_string(&Locale::en));
339        println!("Total Operations(ops)  : {:>15}", (results.count).to_formatted_string(&Locale::en));
340        println!("Total G Migrations     : {:>15}", (results.gmigs).to_formatted_string(&Locale::en));
341        println!("Total D Migrations     : {:>15}", (results.dmigs).to_formatted_string(&Locale::en));
342        println!("Ops per second         : {:>15}", (((results.count as f64) / elapsed.as_secs() as f64) as u64).to_formatted_string(&Locale::en));
343        println!("ns per ops             : {:>15}", ((elapsed.as_nanos() as f64 / results.count as f64) as u64).to_formatted_string(&Locale::en));
344        println!("Ops per threads        : {:>15}", (results.count / nthreads as u64).to_formatted_string(&Locale::en));
345        println!("Ops per procs          : {:>15}", (results.count / nprocs as u64).to_formatted_string(&Locale::en));
346        println!("Ops/sec/procs          : {:>15}", ((((results.count as f64) / nprocs as f64) / elapsed.as_secs() as f64) as u64).to_formatted_string(&Locale::en));
347        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));
348}
Note: See TracBrowser for help on using the repository browser.