solve deadlock (#1091)

pumpkin-world/src/level.rs

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