janis/executor
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

works but sucks ass

Browse source
This commit is contained in:
Janis 2025-02-20 18:30:39 +01:00
parent bfa4a34f54
commit 6ddbd11076
2 changed files with 75 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Cargo.toml
View file

@ -14,13 +14,14 @@ never-local = []
[profile.bench] [profile.bench]
debug = true debug = true
# opt-level = 0
[dependencies] [dependencies]
futures = "0.3" futures = "0.3"
rayon = "1.10" rayon = "1.10"
bevy_tasks = "0.15.1" bevy_tasks = "0.15.1"
parking_lot = {version = "0.12.3", features = ["deadlock_detection"]} parking_lot = {version = "0.12.3"}
thread_local = "1.1.8" thread_local = "1.1.8"
crossbeam = "0.8.4" crossbeam = "0.8.4"
st3 = "0.4" st3 = "0.4"

150
src/praetor/mod.rs
View file

@ -101,7 +101,7 @@ mod util {
let mask = Self::mask(); let mask = Self::mask();
loop { loop {
let ptr = self.0.load(failure); let ptr = self.0.load(failure);
let new = ptr.with_addr(ptr.addr() | (tag & mask)); let new = ptr.with_addr((ptr.addr() & !mask) | (tag & mask));
if self if self
.0 .0
.compare_exchange_weak(ptr, new, success, failure) .compare_exchange_weak(ptr, new, success, failure)
@ -433,7 +433,7 @@ mod job {
val_or_this: UnsafeCell<ValueOrThis<T>>, val_or_this: UnsafeCell<ValueOrThis<T>>,
/// (prev,next) before execute(), Box<...> after /// (prev,next) before execute(), Box<...> after
err_or_link: UnsafeCell<LinkOrError<Job>>, err_or_link: UnsafeCell<LinkOrError<Job>>,
phantom: PhantomPinned, _phantom: PhantomPinned,
} }
impl<T> Debug for Job<T> { impl<T> Debug for Job<T> {
@ -501,7 +501,7 @@ mod job {
next: None, next: None,
}, },
}), }),
phantom: PhantomPinned, _phantom: PhantomPinned,
} }
} }
pub fn empty() -> Job<T> { pub fn empty() -> Job<T> {
@ -519,7 +519,7 @@ mod job {
next: None, next: None,
}, },
}), }),
phantom: PhantomPinned, _phantom: PhantomPinned,
} }
} }
@ -562,13 +562,11 @@ mod job {
Ordering::Release, Ordering::Release,
Ordering::Relaxed, Ordering::Relaxed,
); );
eprintln!("wait({:?}): eepy", self as *const _);
std::thread::park(); std::thread::park();
spin.reset(); spin.reset();
eprintln!("wait({:?}): woken", self as *const _);
// after sleeping, state should be `Finished` // after sleeping, state should be `Finished`
continue;
} }
Err(state) => { Err(state) => {
assert_ne!(state, JobState::Pending as usize); assert_ne!(state, JobState::Pending as usize);
@ -589,11 +587,11 @@ mod job {
return result; return result;
} else { } else {
// spin until lock is released. // spin until lock is released.
eprintln!( // eprintln!(
"wait({:?}): spinning ({:?})", // "wait({:?}): spinning ({:?})",
self as *const _, // self as *const _,
JobState::from_u8(state as u8).unwrap() // JobState::from_u8(state as u8).unwrap()
); // );
spin.spin(); spin.spin();
} }
} }
@ -638,13 +636,11 @@ mod job {
let harness: unsafe fn(*const (), *const Self) = mem::transmute(ptr.as_ptr()); let harness: unsafe fn(*const (), *const Self) = mem::transmute(ptr.as_ptr());
let this = (*self.val_or_this.get()).this; let this = (*self.val_or_this.get()).this;
eprintln!("{harness:?}({this:?}, {:?})", self as *const Self);
harness(this.as_ptr().cast(), (self as *const Self).cast()); harness(this.as_ptr().cast(), (self as *const Self).cast());
} }
} }
fn complete(&self, result: std::thread::Result<T>) { fn complete(&self, result: std::thread::Result<T>) {
eprintln!("complete({:?}) {:#?}", self as *const _, self);
let mut spin = SpinWait::new(); let mut spin = SpinWait::new();
loop { loop {
match self.harness_and_state.compare_exchange_weak_tag( match self.harness_and_state.compare_exchange_weak_tag(
@ -659,11 +655,6 @@ mod job {
} }
Err(state) => { Err(state) => {
assert_ne!(state, JobState::Pending as usize); assert_ne!(state, JobState::Pending as usize);
// eprintln!(
// "complete(): spin waiting for lock to complete with {:?}: ({:?})",
// result.as_ref().map(|_| ()),
// JobState::from_u8(state as u8).unwrap()
// );
spin.spin(); spin.spin();
} }
} }
@ -691,7 +682,6 @@ mod job {
Ordering::Release, Ordering::Release,
Ordering::Relaxed, Ordering::Relaxed,
); );
eprintln!("complete({:?}): finished", self as *const _);
} }
} }
@ -773,7 +763,7 @@ mod job {
} }
use std::{ use std::{
cell::UnsafeCell, cell::{Cell, UnsafeCell},
collections::BTreeMap, collections::BTreeMap,
mem, mem,
pin::{pin, Pin}, pin::{pin, Pin},
@ -790,6 +780,7 @@ use parking_lot::{Condvar, Mutex};
use util::DropGuard; use util::DropGuard;
pub struct Scope { pub struct Scope {
join_count: Cell<usize>,
context: Arc<Context>, context: Arc<Context>,
index: usize, index: usize,
heartbeat: Arc<AtomicBool>, heartbeat: Arc<AtomicBool>,
@ -815,6 +806,7 @@ impl Scope {
context, context,
index, index,
heartbeat, heartbeat,
join_count: Cell::new(0),
queue: UnsafeCell::new(JobList::new()), queue: UnsafeCell::new(JobList::new()),
} }
} }
@ -896,6 +888,43 @@ impl Scope {
} }
pub fn join<A, B, RA, RB>(&self, a: A, b: B) -> (RA, RB) pub fn join<A, B, RA, RB>(&self, a: A, b: B) -> (RA, RB)
where
RA: Send,
RB: Send,
A: FnOnce() -> RA + Send,
B: FnOnce() -> RB + Send,
{
self.join_heartbeat_every::<_, _, _, _, 64>(a, b)
}
pub fn join_seq<A, B, RA, RB>(&self, a: A, b: B) -> (RA, RB)
where
RA: Send,
RB: Send,
A: FnOnce() -> RA + Send,
B: FnOnce() -> RB + Send,
{
(a(), b())
}
pub fn join_heartbeat_every<A, B, RA, RB, const TIMES: usize>(&self, a: A, b: B) -> (RA, RB)
where
RA: Send,
RB: Send,
A: FnOnce() -> RA + Send,
B: FnOnce() -> RB + Send,
{
let count = self.join_count.get();
self.join_count.set(count.wrapping_add(1) % TIMES);
if count == 1 {
self.join_heartbeat(a, b)
} else {
self.join_seq(a, b)
}
}
pub fn join_heartbeat<A, B, RA, RB>(&self, a: A, b: B) -> (RA, RB)
where where
RA: Send, RA: Send,
RB: Send, RB: Send,
@ -926,6 +955,7 @@ impl Scope {
} }
}; };
drop(b);
(ra, rb) (ra, rb)
} }
@ -938,21 +968,18 @@ impl Scope {
#[inline] #[inline]
fn execute(&self, job: &Job) { fn execute(&self, job: &Job) {
eprintln!("execute()");
self.tick(); self.tick();
job.execute(); job.execute();
} }
#[cold] #[cold]
fn heartbeat_cold(&self) { fn heartbeat_cold(&self) {
eprintln!("heartbeat_cold()");
let mut guard = self.context.shared.lock(); let mut guard = self.context.shared.lock();
if !guard.jobs.contains_key(&self.index) { if !guard.jobs.contains_key(&self.index) {
if let Some(job) = self.pop_back() { if let Some(job) = self.pop_back() {
unsafe { unsafe {
job.as_ref().set_pending(); job.as_ref().set_pending();
eprintln!("sharing {job:?} {:#?}", job.as_ref());
} }
guard.jobs.insert(self.index, job); guard.jobs.insert(self.index, job);
self.context.shared_job.notify_one(); self.context.shared_job.notify_one();
@ -964,37 +991,32 @@ impl Scope {
#[cold] #[cold]
pub fn wait_until<T>(&self, job: Pin<&Job<T>>) -> Option<std::thread::Result<T>> { pub fn wait_until<T>(&self, job: Pin<&Job<T>>) -> Option<std::thread::Result<T>> {
// let shared_job = self.context.shared.lock().jobs.remove(&self.index); let shared_job = self.context.shared.lock().jobs.remove(&self.index);
// if let Some(ptr) = shared_job { if let Some(ptr) = shared_job {
// if ptr.as_ptr() == job as *const _ as *mut _ { if ptr.as_ptr() == &*job as *const _ as *mut _ {
// eprintln!("reclaimed shared job"); return None;
// return None; } else {
// } else { unsafe {
// unsafe { self.execute(ptr.as_ref());
// self.execute(ptr.as_ref()); }
// } }
// } }
// }
while job.state() != JobState::Finished as u8 { while job.state() != JobState::Finished as u8 {
let Some(job) = let Some(job) = self
// self .context
// .pop_front() .shared
// .inspect(|job| unsafe { .lock()
// job.as_ref().set_pending(); .jobs
// }) .pop_first()
None .map(|(_, job)| job)
.or_else(|| { .or_else(|| {
self.context self.pop_front().inspect(|job| unsafe {
.shared job.as_ref().set_pending();
.lock() })
.jobs
.pop_first()
.map(|(_, job)| job)
}) })
else { else {
eprintln!("no more jobs, sleep instead");
break; break;
}; };
@ -1014,7 +1036,7 @@ where
A: FnOnce() -> RA + Send, A: FnOnce() -> RA + Send,
B: FnOnce() -> RB + Send, B: FnOnce() -> RB + Send,
{ {
Scope::with(|scope| scope.join(a, b)) Scope::with(|scope| scope.join_heartbeat(a, b))
} }
struct Heartbeat { struct Heartbeat {
@ -1090,6 +1112,7 @@ impl Context {
eprintln!("created threadpool {:?}", Arc::as_ptr(&this)); eprintln!("created threadpool {:?}", Arc::as_ptr(&this));
let num_threads = available_parallelism(); let num_threads = available_parallelism();
// let num_threads = 2;
let barrier = Arc::new(std::sync::Barrier::new(num_threads + 1)); let barrier = Arc::new(std::sync::Barrier::new(num_threads + 1));
for _ in 0..num_threads { for _ in 0..num_threads {
@ -1101,32 +1124,6 @@ impl Context {
let ctx = this.clone(); let ctx = this.clone();
std::thread::spawn(|| heartbeat_worker(ctx)); std::thread::spawn(|| heartbeat_worker(ctx));
// {
// // only for #[cfg]
// use parking_lot::deadlock;
// use std::thread;
// use std::time::Duration;
// // Create a background thread which checks for deadlocks every 10s
// thread::spawn(move || loop {
// thread::sleep(Duration::from_secs(1));
// let deadlocks = deadlock::check_deadlock();
// if deadlocks.is_empty() {
// println!("no deadlocks detected");
// continue;
// }
// println!("{} deadlocks detected", deadlocks.len());
// for (i, threads) in deadlocks.iter().enumerate() {
// println!("Deadlock #{}", i);
// for t in threads {
// println!("Thread Id {:#?}", t.thread_id());
// println!("{:#?}", t.backtrace());
// }
// }
// });
// } // only for #[cfg]
barrier.wait(); barrier.wait();
this this
@ -1160,7 +1157,6 @@ fn worker(ctx: Arc<Context>, barrier: Arc<std::sync::Barrier>) {
let mut job = ctx.shared.lock().jobs.pop_first(); let mut job = ctx.shared.lock().jobs.pop_first();
loop { loop {
if let Some((_, job)) = job { if let Some((_, job)) = job {
eprintln!("worker(): found job {job:?}");
unsafe { unsafe {
scope.execute(job.as_ref()); scope.execute(job.as_ref());
} }