fix more bugs

This commit is contained in:
Carl Lerche
2023-06-12 12:51:23 -07:00
parent 0cfaaea2b9
commit 904dabb23d
3 changed files with 33 additions and 2 deletions
@@ -194,6 +194,11 @@ impl Idle {
}
self.num_idle.store(0, Release);
// Wake up any other workers
while let Some(index) = synced.idle.sleepers.pop() {
shared.condvars[index].notify_one();
}
}
/// The worker releases the given core, making it available to other workers
@@ -1025,6 +1025,13 @@ impl Worker {
}
}
// If the runtime is shutdown, skip parking
self.update_global_flags(cx, &mut synced, &mut core);
if core.is_shutdown {
return Ok((None, core));
}
// Core being returned must not be in the searching state
debug_assert!(!core.is_searching);
@@ -1038,10 +1045,15 @@ impl Worker {
// Wait for driver events
driver.park(&self.handle.driver);
synced = cx.shared().synced.lock();
// Put the driver back
cx.shared().driver.set(driver);
synced = cx.shared().synced.lock();
if cx.shared().inject.is_closed(&mut synced.inject) {
self.shutdown_finalize(cx, synced);
return Err(());
}
// Try to acquire an available core to schedule I/O events
if let Some(core) = self.try_acquire_available_core(cx, &mut synced) {
@@ -1098,10 +1110,20 @@ impl Worker {
let mut synced = cx.shared().synced.lock();
synced.shutdown_cores.push(core);
self.shutdown_finalize(cx, synced);
}
fn shutdown_finalize(&self, cx: &Context, mut synced: MutexGuard<'_, Synced>) {
// Wait for all cores
if synced.shutdown_cores.len() != cx.shared().remotes.len() {
return;
}
// Wait for driver
if cx.shared().driver.is_none() {
return;
}
debug_assert!(cx.shared().owned.is_empty());
for mut core in synced.shutdown_cores.drain(..) {
+5 -1
View File
@@ -1,7 +1,7 @@
use crate::loom::sync::atomic::AtomicPtr;
use std::ptr;
use std::sync::atomic::Ordering::AcqRel;
use std::sync::atomic::Ordering::{AcqRel, Acquire};
pub(crate) struct AtomicCell<T> {
data: AtomicPtr<T>,
@@ -29,6 +29,10 @@ impl<T> AtomicCell<T> {
pub(crate) fn take(&self) -> Option<Box<T>> {
self.swap(None)
}
pub(crate) fn is_none(&self) -> bool {
self.data.load(Acquire).is_null()
}
}
fn to_raw<T>(data: Option<Box<T>>) -> *mut T {