From 274c7e195dadd0198674fff85e55559a039696f8 Mon Sep 17 00:00:00 2001 From: Alexander Medvedev Date: Wed, 21 Jan 2026 00:22:13 +0100 Subject: [PATCH] chore: don't spam so many packets also use CancellationToken instead of Notify in some places --- pumpkin/src/entity/living.rs | 4 ++-- pumpkin/src/entity/mob/mod.rs | 30 +++++++++++++++--------------- pumpkin/src/lib.rs | 11 ++++++----- pumpkin/src/net/java/mod.rs | 22 +++++++++------------- pumpkin/src/net/lan_broadcast.rs | 2 +- pumpkin/src/net/query.rs | 2 +- pumpkin/src/net/rcon/mod.rs | 2 +- 7 files changed, 35 insertions(+), 38 deletions(-) diff --git a/pumpkin/src/entity/living.rs b/pumpkin/src/entity/living.rs index f213970a9..6ac6fbfed 100644 --- a/pumpkin/src/entity/living.rs +++ b/pumpkin/src/entity/living.rs @@ -1161,8 +1161,8 @@ impl EntityBase for LivingEntity { self.tick_movement(server, caller.clone()).await; // TODO if caller.get_player().is_none() { - self.entity.send_pos_rot().await; - self.entity.send_velocity().await; + // self.entity.send_pos_rot().await; + // self.entity.send_velocity().await; } self.tick_effects().await; // Current active item diff --git a/pumpkin/src/entity/mob/mod.rs b/pumpkin/src/entity/mob/mod.rs index 1b459ffe3..f78022cdd 100644 --- a/pumpkin/src/entity/mob/mod.rs +++ b/pumpkin/src/entity/mob/mod.rs @@ -75,22 +75,22 @@ impl MobEntity { } async fn set_mob_flag(&self, flag: u8, value: bool) { - let index = flag; - let mut b = self.mob_flags.load(Ordering::Relaxed); - if value { - b |= index; - } else { - b &= !index; + let old_b = self.mob_flags.load(Ordering::Relaxed); + + let new_b = if value { old_b | flag } else { old_b & !flag }; + + if new_b != old_b { + self.mob_flags.store(new_b, Ordering::Relaxed); + + self.living_entity + .entity + .send_meta_data(&[Metadata::new( + TrackedData::DATA_MOB_FLAGS, + MetaDataType::Byte, + new_b, + )]) + .await; } - self.mob_flags.store(b, Ordering::Relaxed); - self.living_entity - .entity - .send_meta_data(&[Metadata::new( - TrackedData::DATA_MOB_FLAGS, - MetaDataType::Byte, - b, - )]) - .await; } } diff --git a/pumpkin/src/lib.rs b/pumpkin/src/lib.rs index d5387c0d0..162301d9c 100644 --- a/pumpkin/src/lib.rs +++ b/pumpkin/src/lib.rs @@ -28,8 +28,9 @@ use std::time::Duration; use std::{net::SocketAddr, sync::LazyLock}; use tokio::net::{TcpListener, UdpSocket}; use tokio::select; -use tokio::sync::{Mutex, Notify, RwLock}; +use tokio::sync::{Mutex, RwLock}; use tokio::time::sleep; +use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; pub mod block; @@ -159,11 +160,11 @@ pub fn init_logger(advanced_config: &AdvancedConfiguration) { } pub static SHOULD_STOP: AtomicBool = AtomicBool::new(false); -pub static STOP_INTERRUPT: LazyLock = LazyLock::new(Notify::new); +pub static STOP_INTERRUPT: LazyLock = LazyLock::new(CancellationToken::new); pub fn stop_server() { SHOULD_STOP.store(true, Ordering::Relaxed); - STOP_INTERRUPT.notify_waiters(); + STOP_INTERRUPT.cancel(); } fn resolve_some T>( @@ -492,7 +493,7 @@ impl PumpkinServer { }, // Branch for the global stop signal - () = STOP_INTERRUPT.notified() => { + () = STOP_INTERRUPT.cancelled() => { return false; } } @@ -583,7 +584,7 @@ fn setup_console(mut rl: Editor, server: A server.clone().spawn_task(async move { while !SHOULD_STOP.load(Ordering::Relaxed) { let t1 = rx.recv(); - let t2 = STOP_INTERRUPT.notified(); + let t2 = STOP_INTERRUPT.cancelled(); let result = select! { line = t1 => line, diff --git a/pumpkin/src/net/java/mod.rs b/pumpkin/src/net/java/mod.rs index 52e9b40a7..aa691c312 100644 --- a/pumpkin/src/net/java/mod.rs +++ b/pumpkin/src/net/java/mod.rs @@ -37,7 +37,6 @@ use pumpkin_protocol::{ ser::{NetworkWriteExt, ReadingError, WritingError}, }; use pumpkin_util::text::TextComponent; -use tokio::sync::Notify; use tokio::{ io::{BufReader, BufWriter}, net::{ @@ -50,6 +49,7 @@ use tokio::{ sync::mpsc::{Receiver, Sender}, task::JoinHandle, }; +use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; pub mod config; @@ -82,7 +82,7 @@ pub struct JavaClient { /// A collection of tasks associated with this client. The tasks await completion when removing the client. tasks: TaskTracker, /// An notifier that is triggered when this client is closed. - close_interrupt: Arc, + close_token: CancellationToken, /// A queue of serialized packets to send to the network outgoing_packet_queue_send: Sender, /// A queue of serialized packets to send to the network @@ -106,7 +106,7 @@ impl JavaClient { address: Mutex::new(address), connection_state: AtomicCell::new(ConnectionState::HandShake), closed: Arc::new(AtomicBool::new(false)), - close_interrupt: Arc::new(Notify::new()), + close_token: CancellationToken::new(), tasks: TaskTracker::new(), outgoing_packet_queue_send: send, outgoing_packet_queue_recv: Some(recv), @@ -221,7 +221,7 @@ impl JavaClient { } pub async fn await_close_interrupt(&self) { - self.close_interrupt.notified().await; + self.close_token.cancelled().await; } pub async fn get_packet(&self) -> Option { @@ -406,19 +406,15 @@ impl JavaClient { .outgoing_packet_queue_recv .take() .expect("This was set in the new fn"); - let close_interrupt = self.close_interrupt.clone(); + let close_interrupt = self.close_token.clone(); let closed = self.closed.clone(); let writer = self.network_writer.clone(); let id = self.id; self.spawn_task(async move { while !closed.load(Ordering::Relaxed) { let recv_result = tokio::select! { - () = close_interrupt.notified() => { - None - }, - recv_result = packet_receiver.recv() => { - recv_result - } + () = close_interrupt.cancelled() => None, + res = packet_receiver.recv() => res, }; let Some(packet_data) = recv_result else { @@ -431,7 +427,7 @@ impl JavaClient { log::warn!("Failed to send packet to client {id}: {err}",); // We now need to close the connection to the client since the stream is in an // unknown state - close_interrupt.notify_waiters(); + close_interrupt.cancel(); closed.store(true, Ordering::Relaxed); break; } @@ -450,7 +446,7 @@ impl JavaClient { /// /// This function does not attempt to send any disconnect packets to the client. pub fn close(&self) { - self.close_interrupt.notify_waiters(); + self.close_token.cancel(); self.closed.store(true, Ordering::Relaxed); } diff --git a/pumpkin/src/net/lan_broadcast.rs b/pumpkin/src/net/lan_broadcast.rs index ac3f4e67d..222c3fee6 100644 --- a/pumpkin/src/net/lan_broadcast.rs +++ b/pumpkin/src/net/lan_broadcast.rs @@ -56,7 +56,7 @@ impl LANBroadcast { while !SHOULD_STOP.load(Ordering::Relaxed) { let t1 = interval.tick(); - let t2 = STOP_INTERRUPT.notified(); + let t2 = STOP_INTERRUPT.cancelled(); let should_continue = select! { _ = t1 => true, diff --git a/pumpkin/src/net/query.rs b/pumpkin/src/net/query.rs index 421ccf6c5..9b6102e85 100644 --- a/pumpkin/src/net/query.rs +++ b/pumpkin/src/net/query.rs @@ -51,7 +51,7 @@ pub async fn start_query_handler(server: Arc, query_addr: SocketAddr) { let recv_result = tokio::select! { result = socket.recv_from(&mut buf) => Some(result), - () = STOP_INTERRUPT.notified() => None, + () = STOP_INTERRUPT.cancelled() => None, }; let Some(Ok((_, addr))) = recv_result else { diff --git a/pumpkin/src/net/rcon/mod.rs b/pumpkin/src/net/rcon/mod.rs index 79a75ea25..4adfb9268 100644 --- a/pumpkin/src/net/rcon/mod.rs +++ b/pumpkin/src/net/rcon/mod.rs @@ -24,7 +24,7 @@ impl RCONServer { while !SHOULD_STOP.load(Ordering::Relaxed) { let await_new_client = || async { let t1 = listener.accept(); - let t2 = STOP_INTERRUPT.notified(); + let t2 = STOP_INTERRUPT.cancelled(); select! { client = t1 => Some(client),