parent
6e4a05d66d
commit
14db025fa6
@ -28,8 +28,6 @@ use std::{
|
|||||||
atomic::{AtomicBool, AtomicU64, Ordering},
|
atomic::{AtomicBool, AtomicU64, Ordering},
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
use tokio::sync::watch;
|
|
||||||
use tokio::sync::watch::{Receiver, Sender};
|
|
||||||
use tokio::{
|
use tokio::{
|
||||||
select,
|
select,
|
||||||
sync::{
|
sync::{
|
||||||
@ -75,7 +73,7 @@ pub struct Level {
|
|||||||
/// Semaphore to limit concurrent chunk generation tasks
|
/// Semaphore to limit concurrent chunk generation tasks
|
||||||
//chunk_generation_semaphore: Arc<Semaphore>,
|
//chunk_generation_semaphore: Arc<Semaphore>,
|
||||||
/// Map to deduplicate chunk generation and avoid DashMap write lock
|
/// Map to deduplicate chunk generation and avoid DashMap write lock
|
||||||
chunk_generation_locks: Arc<Mutex<HashMap<Vector2<i32>, Receiver<bool>>>>,
|
chunk_generation_locks: Arc<Mutex<HashMap<Vector2<i32>, Arc<Notify>>>>,
|
||||||
/// Tracks tasks associated with this world instance
|
/// Tracks tasks associated with this world instance
|
||||||
tasks: TaskTracker,
|
tasks: TaskTracker,
|
||||||
/// Notification that interrupts tasks for shutdown
|
/// Notification that interrupts tasks for shutdown
|
||||||
@ -787,29 +785,20 @@ impl Level {
|
|||||||
let block_registry = block_registry.clone();
|
let block_registry = block_registry.clone();
|
||||||
let self_clone = self_clone.clone();
|
let self_clone = self_clone.clone();
|
||||||
|
|
||||||
enum Notify {
|
|
||||||
Send(Sender<bool>),
|
|
||||||
Recv(Receiver<bool>),
|
|
||||||
}
|
|
||||||
|
|
||||||
let notify = {
|
let notify = {
|
||||||
let mut locks = self_clone.chunk_generation_locks.lock().await;
|
let mut locks = self_clone.chunk_generation_locks.lock().await;
|
||||||
if let Some(existing) = locks.get(&pos) {
|
if let Some(existing) = locks.get(&pos) {
|
||||||
Notify::Recv(existing.clone())
|
Some(existing.clone())
|
||||||
} else {
|
} else {
|
||||||
let (send, recv) = watch::channel(true);
|
let notify = Arc::new(Notify::new());
|
||||||
locks.insert(pos, recv);
|
locks.insert(pos, notify.clone());
|
||||||
Notify::Send(send)
|
None
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
if let Notify::Recv(mut notify) = notify {
|
if let Some(notify) = notify {
|
||||||
// Wait for the chunk to be generated by another task
|
// Wait for the chunk to be generated by another task
|
||||||
'notify: while *notify.borrow_and_update() {
|
notify.notified().await;
|
||||||
if notify.changed().await.is_err() {
|
|
||||||
break 'notify;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// After being notified, the chunk should be in loaded_chunks
|
// After being notified, the chunk should be in loaded_chunks
|
||||||
// However, it might have been unloaded between notification and access
|
// However, it might have been unloaded between notification and access
|
||||||
if let Some(chunk) = loaded_chunks.get(&pos) {
|
if let Some(chunk) = loaded_chunks.get(&pos) {
|
||||||
@ -823,7 +812,7 @@ impl Level {
|
|||||||
// The chunk generation will be retried if needed
|
// The chunk generation will be retried if needed
|
||||||
log::info!("Chunk at {pos:?} was unloaded after generation notification");
|
log::info!("Chunk at {pos:?} was unloaded after generation notification");
|
||||||
}
|
}
|
||||||
} else if let Notify::Send(notify) = notify {
|
} else {
|
||||||
//let _permit = chunk_generation_semaphore
|
//let _permit = chunk_generation_semaphore
|
||||||
// .acquire()
|
// .acquire()
|
||||||
// .await
|
// .await
|
||||||
@ -851,15 +840,19 @@ impl Level {
|
|||||||
loaded_chunks.insert(pos, arc_chunk.clone());
|
loaded_chunks.insert(pos, arc_chunk.clone());
|
||||||
|
|
||||||
// Store the notify for later removal
|
// Store the notify for later removal
|
||||||
arc_chunk
|
(arc_chunk, pos)
|
||||||
};
|
};
|
||||||
|
|
||||||
// Remove the notify and wake up any waiters
|
// Remove the notify and wake up any waiters
|
||||||
// Do this outside the rayon thread to avoid deadlock
|
// Do this outside the rayon thread to avoid deadlock
|
||||||
let arc_chunk = result;
|
let (arc_chunk, pos) = result;
|
||||||
{
|
{
|
||||||
|
let self_clone = self_clone.clone();
|
||||||
handle.spawn(async move {
|
handle.spawn(async move {
|
||||||
notify.send(true).unwrap();
|
let mut locks = self_clone.chunk_generation_locks.lock().await;
|
||||||
|
if let Some(notify) = locks.remove(&pos) {
|
||||||
|
notify.notify_waiters();
|
||||||
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user