[720b1a9] | 1 | use std::ptr; |
---|
| 2 | use std::sync::Arc; |
---|
[751e2eb] | 3 | use std::sync::atomic::{AtomicU64, Ordering}; |
---|
| 4 | use std::time::Instant; |
---|
[720b1a9] | 5 | use std::thread::{self, ThreadId}; |
---|
| 6 | |
---|
| 7 | use tokio::runtime::Builder; |
---|
| 8 | use tokio::sync; |
---|
| 9 | |
---|
[751e2eb] | 10 | use clap::{App, Arg}; |
---|
[720b1a9] | 11 | use num_format::{Locale, ToFormattedString}; |
---|
| 12 | use rand::Rng; |
---|
| 13 | |
---|
[751e2eb] | 14 | #[path = "../bench.rs"] |
---|
| 15 | mod bench; |
---|
[720b1a9] | 16 | |
---|
| 17 | // ================================================== |
---|
| 18 | struct MyData { |
---|
| 19 | data: Vec<u64>, |
---|
| 20 | ttid: ThreadId, |
---|
| 21 | _id: usize, |
---|
| 22 | } |
---|
| 23 | |
---|
| 24 | impl 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 | |
---|
| 47 | struct MyDataPtr { |
---|
| 48 | ptr: *mut MyData, |
---|
| 49 | } |
---|
| 50 | |
---|
| 51 | unsafe impl std::marker::Send for MyDataPtr{} |
---|
| 52 | |
---|
| 53 | // ================================================== |
---|
| 54 | struct MyCtx { |
---|
| 55 | s: sync::Semaphore, |
---|
| 56 | d: MyDataPtr, |
---|
| 57 | ttid: ThreadId, |
---|
| 58 | _id: usize, |
---|
| 59 | } |
---|
| 60 | |
---|
| 61 | impl 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 |
---|
| 82 | struct MySpot { |
---|
| 83 | ptr: AtomicU64, |
---|
| 84 | _id: usize, |
---|
| 85 | } |
---|
| 86 | |
---|
| 87 | impl 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 |
---|
| 157 | struct Result { |
---|
| 158 | count: u64, |
---|
| 159 | gmigs: u64, |
---|
| 160 | dmigs: u64, |
---|
| 161 | } |
---|
| 162 | |
---|
| 163 | impl 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 |
---|
| 177 | fn __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 |
---|
| 188 | fn work(data: &mut MyData, cnt: u64, state : &mut u64) { |
---|
| 189 | for _ in 0..cnt { |
---|
| 190 | data.access(__xorshift64(state)) |
---|
| 191 | } |
---|
| 192 | } |
---|
| 193 | |
---|
[751e2eb] | 194 | async fn local(start: Arc<sync::Barrier>, idata: MyDataPtr, spots : Arc<Vec<MySpot>>, cnt: u64, share: bool, id: usize, exp: Arc<bench::BenchData>) -> Result{ |
---|
[720b1a9] | 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 |
---|
[751e2eb] | 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 |
---|
[720b1a9] | 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 | |
---|
[751e2eb] | 238 | exp.threads_left.fetch_sub(1, Ordering::SeqCst); |
---|
[720b1a9] | 239 | r |
---|
| 240 | } |
---|
| 241 | |
---|
| 242 | |
---|
| 243 | // ================================================== |
---|
| 244 | fn main() { |
---|
| 245 | let options = App::new("Locality Tokio") |
---|
[751e2eb] | 246 | .args(&bench::args()) |
---|
[720b1a9] | 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 | |
---|
[751e2eb] | 266 | let exp = Arc::new(bench::BenchData::new(options, nprocs)); |
---|
[720b1a9] | 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, |
---|
[751e2eb] | 293 | exp.clone(), |
---|
[720b1a9] | 294 | )) |
---|
| 295 | }).collect(); |
---|
| 296 | |
---|
| 297 | |
---|
| 298 | println!("Starting"); |
---|
| 299 | |
---|
| 300 | let start = Instant::now(); |
---|
| 301 | barr.wait().await; |
---|
| 302 | |
---|
[751e2eb] | 303 | elapsed = exp.wait(&start).await; |
---|
[720b1a9] | 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 | } |
---|