1use arc_swap::ArcSwap;
2use rayon::ThreadPool;
3use rustc_hash::{FxBuildHasher, FxHashMap, FxHashSet};
4use std::{
5 io, mem,
6 sync::{
7 Arc, Weak,
8 atomic::{AtomicBool, AtomicUsize, Ordering},
9 },
10 time::{Duration, Instant},
11};
12use steel_protocol::packet_traits::EncodedPacket;
13use steel_protocol::packets::game::{
14 BlockChange, CBlockUpdate, CLightUpdate, CSectionBlocksUpdate, CSetChunkCenter,
15};
16use steel_protocol::utils::ConnectionProtocol;
17use steel_registry::dimension_type::DimensionTypeRef;
18use steel_registry::{
19 blocks::{BlockRef, block_state_ext::BlockStateExt},
20 fluid::FluidRef,
21};
22use steel_utils::{BlockPos, BlockStateId, ChunkPos, PackedChunkPos, SectionPos, locks::SyncMutex};
23use tokio::runtime::Runtime;
24use tokio::sync::Notify;
25use tokio::time::sleep;
26use tokio_util::sync::CancellationToken;
27use tokio_util::task::TaskTracker;
28use tracing::instrument;
29
30use crate::behavior::{BLOCK_BEHAVIORS, FLUID_BEHAVIORS};
31use crate::block_entity::{BlockEntityLifecycleExt as _, ClearedBlockEntities, SharedBlockEntity};
32use crate::chunk::chunk_holder::{
33 ChunkHolder, ChunkSaveDependency, PostProcessGenerationError, TickingReadiness,
34};
35use crate::chunk::chunk_request::ChunkRequestLease;
36pub use crate::chunk::chunk_scheduler::ChunkMapSchedulingTimings;
37#[cfg(test)]
38use crate::chunk::chunk_scheduler::PlayerTicketOperation;
39use crate::chunk::chunk_scheduler::{ChunkSchedulingCoordinator, ChunkTicketReceipt};
40use crate::chunk::chunk_ticket_manager::{
41 ChunkTicketLevel, LoadLevelChange, generation_status, is_block_ticking, is_entity_ticking,
42 is_full,
43};
44use crate::chunk::chunk_ticket_storage::{
45 ChunkTicketStorage, ENDER_PEARL_TICKET_TIMEOUT_TICKS, PersistentChunkTickets,
46 TimedTicketExpiration,
47};
48use crate::chunk::full_chunk_readiness::{
49 FullNeighborhoodCounts, FullNeighborhoodError, FullNeighborhoodIndex, FullPublication,
50 FullPublicationQueue,
51};
52pub use crate::chunk::gameplay_chunk_lookup_cache::GameplayChunkLookupCacheStats;
53use crate::chunk::gameplay_chunk_lookup_cache::{
54 GameplayChunkLookupCacheScope, lookup_or_insert_with,
55};
56use crate::chunk::light::{
57 LIGHT_CACHE_RADIUS, LightCacheLayout, LightCacheSetupRadius, LightLayer,
58 LightSectionEmptinessChange, LightSectionRange, LightWorkWindowGate, LightWorkset,
59 build_chunk_light_update_packet_for_sections,
60 propagate_block_light_changes_with_empty_sections,
61 propagate_sky_light_changes_with_empty_sections,
62};
63use crate::chunk::player_chunk_view::PlayerChunkView;
64use crate::chunk::simulation_ticket_manager::SimulationLevelChange;
65use crate::chunk::{
66 Chunk,
67 chunk_generation_task::ChunkGenerationTask,
68 full_chunk::{BlockRandomPositionGenerator, FullChunkRef},
69 section::RandomTickSectionBits,
70 status::ChunkStatus,
71};
72use crate::chunk_saver::ChunkStorage;
73use crate::player::connection::NetworkConnection;
74use crate::world::World;
75use crate::world::tick_scheduler::{BlockTick, FluidTick, ScheduledTickRunBatch};
76use crate::worldgen::{ChunkGeneratorType, WorldGenContext};
77use crate::{entity::Entity, player::Player};
78
79mod generation_readiness;
80mod light_update_state;
81mod light_updates;
82mod persistence;
83mod player_tracking;
84mod scheduled_ticks;
85
86#[cfg(test)]
87use light_update_state::PendingLightUpdates;
88use light_update_state::{InFlightLightUpdates, LightUpdateState, PendingChunkLightUpdates};
89
90const GENERATION_THREAD_MULTIPLE: usize = 2;
91const MAX_SCHEDULED_TICKS_PER_TICK: usize = 65_536;
93
94pub const ENDER_PEARL_TICKET_TIMEOUT: i64 = ENDER_PEARL_TICKET_TIMEOUT_TICKS;
98
99#[derive(Debug, Default)]
101pub struct ChunkMapGameTickTimings {
102 pub scheduling: ChunkMapSchedulingTimings,
104 pub broadcast_changes: Duration,
106 pub collect_tickable: Duration,
108 pub tick_chunks: Duration,
110 pub tick_block_entities: Duration,
112 pub tickable_count: usize,
114 pub total_chunks: usize,
116 pub lookup_cache: GameplayChunkLookupCacheStats,
118}
119
120#[derive(Clone)]
121struct TickableChunk {
122 pos: ChunkPos,
123 holder: Arc<ChunkHolder>,
124 randomly_ticking_sections: Arc<RandomTickSectionBits>,
125}
126
127#[derive(Default)]
134struct TickingChunkSnapshot {
135 block: Box<[TickableChunk]>,
136 random_chunk_indices: Box<[usize]>,
137 entity_indices: Box<[usize]>,
138}
139
140struct FinalizedBlockEntityUnload {
141 holder: Arc<ChunkHolder>,
142 lifecycle_dispatchers: Vec<SharedBlockEntity>,
143 positions: Vec<BlockPos>,
144}
145
146struct BlockTickBatchGuard<'a> {
147 world: &'a World,
148 batch: Arc<ScheduledTickRunBatch<BlockRef>>,
149}
150
151impl<'a> BlockTickBatchGuard<'a> {
152 fn new(world: &'a World, ticks: Vec<BlockTick>) -> Self {
153 Self {
154 world,
155 batch: world.begin_scheduled_block_tick_batch(ticks),
156 }
157 }
158
159 fn ticks(&self) -> &[BlockTick] {
160 self.batch.ticks()
161 }
162
163 fn start(&self, index: usize) {
164 self.batch.start(index);
165 }
166}
167
168impl Drop for BlockTickBatchGuard<'_> {
169 fn drop(&mut self) {
170 self.world.end_scheduled_block_tick_batch(&self.batch);
171 }
172}
173
174struct FluidTickBatchGuard<'a> {
175 world: &'a World,
176 batch: Arc<ScheduledTickRunBatch<FluidRef>>,
177}
178
179impl<'a> FluidTickBatchGuard<'a> {
180 fn new(world: &'a World, ticks: Vec<FluidTick>) -> Self {
181 Self {
182 world,
183 batch: world.begin_scheduled_fluid_tick_batch(ticks),
184 }
185 }
186
187 fn ticks(&self) -> &[FluidTick] {
188 self.batch.ticks()
189 }
190
191 fn start(&self, index: usize) {
192 self.batch.start(index);
193 }
194}
195
196impl Drop for FluidTickBatchGuard<'_> {
197 fn drop(&mut self) {
198 self.world.end_scheduled_fluid_tick_batch(&self.batch);
199 }
200}
201
202struct TickingReadinessCandidate {
203 pos: ChunkPos,
204 holder: Arc<ChunkHolder>,
205 desired: TickingReadiness,
206 target: TickingReadiness,
207}
208
209#[derive(Default)]
210struct ReadinessReconcileResult {
211 snapshot_changed: bool,
212 post_process_generation: Duration,
213 post_process_chunk_count: usize,
214 post_process_position_count: usize,
215 candidate_count: usize,
216}
217
218pub struct ChunkMap {
220 pub(crate) chunks: scc::HashMap<ChunkPos, Arc<ChunkHolder>, FxBuildHasher>,
222 pub(crate) unloading_chunks: scc::HashMap<ChunkPos, Arc<ChunkHolder>, FxBuildHasher>,
224 pub pending_generation_tasks: SyncMutex<Vec<Arc<ChunkGenerationTask>>>,
226 pub task_tracker: TaskTracker,
228 scheduling: ChunkSchedulingCoordinator,
230 source_phase_guard: SyncMutex<()>,
232 full_publications: Arc<FullPublicationQueue>,
234 full_neighborhood: SyncMutex<FullNeighborhoodIndex>,
236 ticking_chunks: ArcSwap<TickingChunkSnapshot>,
238 finalized_block_entity_unloads: SyncMutex<Vec<FinalizedBlockEntityUnload>>,
240 pub world_gen_context: Arc<WorldGenContext>,
242 pub generation_pool: Arc<ThreadPool>,
244 chunk_encoding_pool: Arc<ThreadPool>,
246 pub chunk_runtime: Arc<Runtime>,
250 pub storage: Arc<ChunkStorage>,
252 pub chunks_to_broadcast: SyncMutex<Vec<Arc<ChunkHolder>>>,
254 light_updates: SyncMutex<LightUpdateState>,
256 light_updates_progress_notify: Notify,
258 light_work_window_gate: Arc<LightWorkWindowGate>,
260 running_generation_tasks: AtomicUsize,
262 generation_refill_notify: Notify,
264 generation_refill_cancel_token: CancellationToken,
266 generation_refill_stopped: AtomicBool,
268 generation_refill_started: AtomicBool,
270 pub cancel_token: CancellationToken,
273}
274
275#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
276struct GenerationTaskPriority {
277 simulation_bucket: u8,
278 simulation_level: ChunkTicketLevel,
279 load_level: ChunkTicketLevel,
280}
281
282impl GenerationTaskPriority {
283 const fn for_levels(
284 load_level: Option<ChunkTicketLevel>,
285 simulation_level: Option<ChunkTicketLevel>,
286 ) -> Self {
287 let simulation_bucket = if simulation_level.is_some() { 0 } else { 1 };
288 Self {
289 simulation_bucket,
290 simulation_level: match simulation_level {
291 Some(level) => level,
292 None => ChunkTicketLevel::MAX,
293 },
294 load_level: match load_level {
295 Some(level) => level,
296 None => ChunkTicketLevel::MAX,
297 },
298 }
299 }
300}
301
302struct RunningGenerationTaskPermit {
303 chunk_map: Arc<ChunkMap>,
304 task: Arc<ChunkGenerationTask>,
305}
306
307impl Drop for RunningGenerationTaskPermit {
308 fn drop(&mut self) {
309 self.task
310 .center_holder()
311 .clear_generation_task_if_current(&self.task);
312 self.chunk_map
313 .running_generation_tasks
314 .fetch_sub(1, Ordering::AcqRel);
315 self.chunk_map.notify_generation_refill();
316 }
317}
318
319impl ChunkMap {
320 #[must_use]
324 #[expect(
325 clippy::too_many_arguments,
326 reason = "chunk-map construction requires its world and scheduling configuration"
327 )]
328 pub fn new_with_storage(
329 chunk_runtime: Arc<Runtime>,
330 world: Weak<World>,
331 dimension_type: DimensionTypeRef,
332 sea_level: i32,
333 view_distance: u8,
334 simulation_distance: u8,
335 storage: Arc<ChunkStorage>,
336 generator: Arc<ChunkGeneratorType>,
337 generation_pool: Arc<ThreadPool>,
338 ) -> Self {
339 let chunk_encoding_pool = Arc::clone(&generation_pool);
340 Self::new_with_storage_and_ticket_storage(
341 chunk_runtime,
342 world,
343 dimension_type,
344 sea_level,
345 storage,
346 generator,
347 generation_pool,
348 chunk_encoding_pool,
349 view_distance,
350 simulation_distance,
351 ChunkTicketStorage::new(),
352 )
353 }
354
355 #[must_use]
356 #[expect(
357 clippy::too_many_arguments,
358 reason = "extends ChunkMap::new_with_storage with restored runtime ticket state"
359 )]
360 pub(crate) fn new_with_storage_and_ticket_storage(
361 chunk_runtime: Arc<Runtime>,
362 world: Weak<World>,
363 dimension_type: DimensionTypeRef,
364 sea_level: i32,
365 storage: Arc<ChunkStorage>,
366 generator: Arc<ChunkGeneratorType>,
367 generation_pool: Arc<ThreadPool>,
368 chunk_encoding_pool: Arc<ThreadPool>,
369 view_distance: u8,
370 simulation_distance: u8,
371 ticket_storage: ChunkTicketStorage,
372 ) -> Self {
373 let full_publications = Arc::new(FullPublicationQueue::default());
374
375 Self {
376 chunks: scc::HashMap::default(),
377 unloading_chunks: scc::HashMap::default(),
378 pending_generation_tasks: SyncMutex::new(Vec::new()),
379 task_tracker: TaskTracker::new(),
380 scheduling: ChunkSchedulingCoordinator::new(
381 ticket_storage,
382 view_distance,
383 simulation_distance,
384 ),
385 source_phase_guard: SyncMutex::new(()),
386 full_publications,
387 full_neighborhood: SyncMutex::new(FullNeighborhoodIndex::default()),
388 ticking_chunks: ArcSwap::from_pointee(TickingChunkSnapshot::default()),
389 finalized_block_entity_unloads: SyncMutex::new(Vec::new()),
390 world_gen_context: Arc::new(WorldGenContext::new(
391 generator,
392 world,
393 dimension_type.min_y,
394 dimension_type.height,
395 sea_level,
396 )),
397 generation_pool,
398 chunk_encoding_pool,
399 chunk_runtime,
400 storage,
401 chunks_to_broadcast: SyncMutex::new(Vec::new()),
402 light_updates: SyncMutex::new(LightUpdateState::default()),
403 light_updates_progress_notify: Notify::new(),
404 light_work_window_gate: Arc::new(LightWorkWindowGate::new()),
405 running_generation_tasks: AtomicUsize::new(0),
406 generation_refill_notify: Notify::new(),
407 generation_refill_cancel_token: CancellationToken::new(),
408 generation_refill_stopped: AtomicBool::new(false),
409 generation_refill_started: AtomicBool::new(false),
410 cancel_token: CancellationToken::new(),
411 }
412 }
413
414 pub(crate) fn light_work_window_gate(&self) -> Arc<LightWorkWindowGate> {
415 Arc::clone(&self.light_work_window_gate)
416 }
417
418 pub fn start_generation_refill_loop(self: &Arc<Self>) {
420 if self.generation_refill_started.swap(true, Ordering::AcqRel) {
421 return;
422 }
423
424 let chunk_map = Arc::clone(self);
425 self.task_tracker.spawn_on(
426 async move {
427 loop {
428 tokio::select! {
429 () = chunk_map.generation_refill_cancel_token.cancelled() => break,
430 () = chunk_map.generation_refill_notify.notified() => {
431 chunk_map.run_generation_tasks_b();
432 }
433 }
434 }
435 },
436 self.chunk_runtime.handle(),
437 );
438 }
439
440 pub fn stop_generation_refill_loop(&self) {
443 self.generation_refill_stopped
444 .store(true, Ordering::Release);
445 self.generation_refill_cancel_token.cancel();
446 self.generation_refill_notify.notify_waiters();
447
448 let pending = mem::take(&mut *self.pending_generation_tasks.lock());
449 for task in pending {
450 task.cancel();
451 task.center_holder().clear_generation_task_if_current(&task);
452 }
453 }
454
455 pub(crate) fn notify_generation_refill(&self) {
456 self.generation_refill_notify.notify_one();
457 }
458
459 fn run_or_notify_generation_refill(&self) {
460 if self.generation_refill_started.load(Ordering::Acquire) {
461 self.notify_generation_refill();
462 } else {
463 self.run_generation_tasks_b();
464 }
465 }
466
467 pub fn with_full_chunk<F, R>(&self, pos: ChunkPos, f: F) -> Option<R>
470 where
471 F: FnOnce(FullChunkRef<'_>) -> R,
472 {
473 let holder = self.lookup_active_holder(pos)?;
474 if holder.is_status_disallowed(ChunkStatus::Full) {
475 return None;
476 }
477 holder.try_full_chunk().map(f)
478 }
479
480 pub(crate) fn active_full_chunk_holder(&self, pos: ChunkPos) -> Option<Arc<ChunkHolder>> {
482 let holder = self.lookup_active_holder(pos)?;
483 if holder.is_status_disallowed(ChunkStatus::Full)
484 || holder.try_chunk(ChunkStatus::Full).is_none()
485 {
486 return None;
487 }
488 Some(holder)
489 }
490
491 #[doc(hidden)]
496 #[cfg(feature = "benchmark-support")]
497 pub fn insert_benchmark_chunk_holder(&self, pos: ChunkPos, holder: Arc<ChunkHolder>) {
498 assert!(holder.simulation_level().is_none());
499 assert!(self.ticking_chunks.load().block.is_empty());
500 let _ = self.chunks.insert_sync(pos, holder);
501 }
502
503 #[inline]
504 fn lookup_active_holder(&self, pos: ChunkPos) -> Option<Arc<ChunkHolder>> {
505 lookup_or_insert_with(self, pos, || {
506 self.chunks.read_sync(&pos, |_, holder| Arc::clone(holder))
507 })
508 }
509
510 #[must_use]
512 pub(crate) fn is_block_ticking_full_chunk_loaded(&self, pos: ChunkPos) -> bool {
513 self.lookup_active_holder(pos).is_some_and(|holder| {
514 is_block_ticking(holder.load_level())
515 && holder.ticking_readiness_snapshot().is_block_ticking()
516 })
517 }
518
519 #[must_use]
521 pub fn is_block_ticking_full_chunk_simulated(&self, pos: ChunkPos) -> bool {
522 self.lookup_active_holder(pos).is_some_and(|holder| {
523 is_block_ticking(holder.simulation_level())
524 && holder.ticking_readiness_snapshot().is_block_ticking()
525 })
526 }
527
528 pub(crate) fn with_chunk_at_status<F, R>(
531 &self,
532 pos: ChunkPos,
533 status: ChunkStatus,
534 f: F,
535 ) -> Option<R>
536 where
537 F: FnOnce(&Chunk) -> R,
538 {
539 let chunk_holder = self.lookup_active_holder(pos)?;
540 if chunk_holder.is_status_disallowed(status) {
543 return None;
544 }
545 let chunk = chunk_holder.try_chunk(status)?;
546 Some(f(chunk))
547 }
548
549 #[cfg(test)]
550 pub(crate) fn queue_test_player_ticket_add(
551 &self,
552 pos: ChunkPos,
553 player_id: uuid::Uuid,
554 ) -> ChunkTicketReceipt {
555 self.scheduling
556 .queue_player_ticket_operation(PlayerTicketOperation::Add { pos, player_id })
557 }
558
559 #[cfg(test)]
560 pub(crate) fn queue_test_player_ticket_remove(
561 &self,
562 pos: ChunkPos,
563 player_id: uuid::Uuid,
564 ) -> ChunkTicketReceipt {
565 self.scheduling
566 .queue_player_ticket_operation(PlayerTicketOperation::Remove { pos, player_id })
567 }
568
569 pub(crate) fn acquire_chunk_request_leases(
570 &self,
571 positions: &[ChunkPos],
572 ticket_level: ChunkTicketLevel,
573 ) -> Option<ChunkTicketReceipt> {
574 self.scheduling
575 .acquire_chunk_request_leases(positions.iter().copied(), ticket_level)
576 }
577
578 pub(crate) fn release_chunk_request_leases(
579 &self,
580 positions: &[ChunkPos],
581 ticket_level: ChunkTicketLevel,
582 ) -> Option<ChunkTicketReceipt> {
583 self.scheduling
584 .release_chunk_request_leases(positions.iter().copied(), ticket_level)
585 }
586
587 pub(crate) fn is_ticket_receipt_committed(&self, receipt: ChunkTicketReceipt) -> bool {
588 self.scheduling.is_receipt_committed(receipt)
589 }
590
591 pub(crate) async fn with_full_chunks_in_radius<F, R>(
594 self: &Arc<Self>,
595 center: ChunkPos,
596 radius: u8,
597 f: F,
598 ) -> Option<R>
599 where
600 F: FnOnce() -> R,
601 {
602 let ticket_level = ChunkTicketLevel::for_full_chunk_radius(radius);
603 let lease = ChunkRequestLease::new(Arc::clone(self), Box::new([center]), ticket_level);
604 let Some(ticket_receipt) = lease.submission_receipt else {
605 unreachable!("one chunk request lease must produce a receipt");
606 };
607 let radius = i32::from(radius);
608
609 loop {
610 self.advance_scheduling();
611 if self.is_ticket_receipt_committed(ticket_receipt)
612 && self.full_square_is_ready(center, radius)
613 {
614 break;
615 }
616
617 if self.cancel_token.is_cancelled() {
618 drop(lease);
619 self.advance_scheduling();
620 return None;
621 }
622
623 sleep(Duration::from_millis(10)).await;
624 }
625
626 let result = f();
627 drop(lease);
628 self.advance_scheduling();
629
630 Some(result)
631 }
632
633 pub(crate) fn place_portal_ticket(&self, ticket_position: BlockPos) {
635 let center = ChunkPos::from_block_pos(ticket_position);
636 self.scheduling.add_or_refresh_portal_ticket(center);
637 }
638
639 pub(crate) fn tick_timed_tickets(&self) {
641 let expirations = self.eligible_timed_ticket_expirations();
642 self.scheduling.tick_timed_tickets(&expirations);
643 }
644
645 fn eligible_timed_ticket_expirations(&self) -> Vec<TimedTicketExpiration> {
646 let mut expirations = self.scheduling.timed_ticket_expirations();
648 expirations.retain(|expiration| {
649 expiration.can_expire_if_unloaded() || self.can_timed_ticket_expire(expiration.pos())
650 });
651 expirations
652 }
653
654 pub(crate) fn persistent_chunk_tickets(&self) -> PersistentChunkTickets {
655 self.scheduling.persistent_chunk_tickets()
656 }
657
658 fn can_timed_ticket_expire(&self, pos: ChunkPos) -> bool {
659 self.chunks
660 .read_sync(&pos, |_, holder| holder.is_ready_for_saving())
661 .unwrap_or(true)
662 }
663
664 fn full_square_is_ready(&self, center: ChunkPos, radius: i32) -> bool {
665 for dz in -radius..=radius {
666 for dx in -radius..=radius {
667 let pos = ChunkPos::new(center.0.x + dx, center.0.y + dz);
668 let Some(holder) = self.chunks.read_sync(&pos, |_, holder| Arc::clone(holder))
669 else {
670 return false;
671 };
672 if holder.try_chunk(ChunkStatus::Full).is_none() {
673 return false;
674 }
675 }
676 }
677 true
678 }
679
680 #[expect(
682 clippy::too_many_lines,
683 reason = "block and light packet construction share the same holder drain"
684 )]
685 pub fn broadcast_changed_chunks(&self) {
686 self.propagate_queued_light_changes();
687
688 let holders = {
689 let mut guard = self.chunks_to_broadcast.lock();
690 if guard.is_empty() {
691 return;
692 }
693 mem::take(&mut *guard)
694 };
695
696 let mut world = None;
697
698 for holder in holders {
699 let chunk_pos = holder.get_pos();
700 let world = world.get_or_insert_with(|| self.world_gen_context.world());
702 let has_skylight = world.dimension_type.has_skylight;
703 let min_y = holder.min_y();
704 holder.clear_broadcast_queued();
705
706 let light_changes = holder.take_changed_light_sections();
707 let changes_by_section = holder.take_changed_blocks();
709 let has_publishable_light_changes =
710 !light_changes.block.is_empty() || (has_skylight && !light_changes.sky.is_empty());
711
712 if !has_publishable_light_changes && changes_by_section.is_empty() {
713 continue;
714 }
715
716 if has_publishable_light_changes
717 && let Some(chunk) = holder.try_chunk(ChunkStatus::Full)
718 {
719 let tracking_players = world.get_light_packet_tracking_players(chunk_pos);
720 if !tracking_players.is_empty() {
721 let light_data = {
722 let light = chunk.light();
723 let sky_sections = if has_skylight {
724 light_changes.sky.as_slice()
725 } else {
726 &[]
727 };
728 build_chunk_light_update_packet_for_sections(
729 chunk_pos,
730 &light,
731 has_skylight,
732 sky_sections,
733 &light_changes.block,
734 )
735 };
736 let light_packet = CLightUpdate {
737 x: chunk_pos.0.x,
738 z: chunk_pos.0.y,
739 light_data,
740 };
741
742 let Ok(encoded) = EncodedPacket::from_bare(
743 light_packet,
744 world.compression,
745 ConnectionProtocol::Play,
746 ) else {
747 log::warn!("Failed to encode light update packet");
748 continue;
749 };
750
751 for entity_id in &tracking_players {
752 if let Some(player) = world.players.get_by_entity_id(*entity_id) {
753 player.connection.send_encoded(encoded.clone());
754 }
755 }
756 }
757 }
758
759 if changes_by_section.is_empty() {
760 continue;
761 }
762
763 let tracking_players = world.get_packet_tracking_players(chunk_pos);
765 if tracking_players.is_empty() {
766 continue;
767 }
768
769 for (section_index, changed_positions) in changes_by_section {
771 let section_y = min_y / 16 + section_index as i32;
772 let section_pos = SectionPos::new(chunk_pos.0.x, section_y, chunk_pos.0.y);
773
774 if changed_positions.len() == 1 {
775 let Some(&packed) = changed_positions.iter().next() else {
777 continue;
778 };
779 let block_pos = section_pos.relative_to_block_pos(packed);
780 let block_state = world.get_block_state(block_pos);
781
782 tracing::trace!(
783 ?block_pos,
784 ?block_state,
785 player_count = tracking_players.len(),
786 "Broadcasting single block update"
787 );
788
789 let update_packet = CBlockUpdate {
790 pos: block_pos,
791 block_state,
792 };
793
794 let Ok(encoded) = EncodedPacket::from_bare(
795 update_packet,
796 world.compression,
797 ConnectionProtocol::Play,
798 ) else {
799 log::warn!("Failed to encode block update packet");
800 continue;
801 };
802
803 for entity_id in &tracking_players {
804 if let Some(player) = world.players.get_by_entity_id(*entity_id) {
805 player.connection.send_encoded(encoded.clone());
806 }
807 }
808 world.broadcast_block_entity_if_needed(block_pos);
809 } else {
810 let changes: Vec<BlockChange> = changed_positions
812 .iter()
813 .map(|&packed| {
814 let block_pos = section_pos.relative_to_block_pos(packed);
815 let block_state = world.get_block_state(block_pos);
816 BlockChange {
817 pos: packed,
818 block_state,
819 }
820 })
821 .collect();
822
823 tracing::trace!(
824 change_count = changes.len(),
825 ?section_pos,
826 player_count = tracking_players.len(),
827 "Broadcasting section block updates"
828 );
829
830 let packet = CSectionBlocksUpdate {
831 section_pos,
832 changes,
833 };
834
835 let Ok(encoded) = EncodedPacket::from_bare(
836 packet,
837 world.compression,
838 ConnectionProtocol::Play,
839 ) else {
840 log::warn!("Failed to encode section block update packet");
841 continue;
842 };
843
844 for entity_id in &tracking_players {
845 if let Some(player) = world.players.get_by_entity_id(*entity_id) {
846 player.connection.send_encoded(encoded.clone());
847 }
848 }
849 for &packed in &changed_positions {
850 let block_pos = section_pos.relative_to_block_pos(packed);
851 world.broadcast_block_entity_if_needed(block_pos);
852 }
853 }
854 }
855 }
856 }
857
858 #[instrument(level = "trace", skip(self, world), name = "chunk_map_game_tick")]
865 pub fn tick_game(
866 self: &Arc<Self>,
867 world: &Arc<World>,
868 tick_count: u64,
869 random_tick_speed: u32,
870 runs_normally: bool,
871 ) -> ChunkMapGameTickTimings {
872 let _source_phase_guard = self.source_phase_guard.lock();
873 let mut timings = ChunkMapGameTickTimings::default();
874
875 if tick_count.is_multiple_of(100) {
876 tracing::debug!(
877 chunks = self.chunks.len(),
878 unloading = self.unloading_chunks.len(),
879 "Chunk map status"
880 );
881 }
882
883 if runs_normally {
884 let lookup_cache_scope = GameplayChunkLookupCacheScope::enter(self);
885 let _span = tracing::trace_span!("collect_tickable").entered();
886 let start = Instant::now();
887 let tickable_chunks = self.ticking_chunks.load();
888 timings.collect_tickable = start.elapsed();
889 if !tickable_chunks.block.is_empty() {
890 let _span = tracing::trace_span!(
891 "scheduled_ticks",
892 block_ticking_count = tickable_chunks.block.len(),
893 total_chunks = self.chunks.len()
894 )
895 .entered();
896 let start = Instant::now();
897 let current_tick = world.game_time();
900 let ready_block_ticks =
901 Self::collect_scheduled_block_ticks(world, &tickable_chunks, current_tick);
902 Self::execute_scheduled_block_ticks(world, ready_block_ticks);
903
904 let ready_fluid_ticks =
905 Self::collect_scheduled_fluid_ticks(world, &tickable_chunks, current_tick);
906 Self::execute_scheduled_fluid_ticks(world, ready_fluid_ticks);
907 timings.tick_chunks += start.elapsed();
908 }
909 timings.lookup_cache.merge(lookup_cache_scope.finish());
910 }
911
912 if runs_normally {
915 self.tick_timed_tickets();
916 }
917 timings.scheduling = self.run_chunk_source_updates();
918
919 {
920 let lookup_cache_scope = GameplayChunkLookupCacheScope::enter(self);
921 let _span = tracing::trace_span!("collect_tickable").entered();
922 let start = Instant::now();
923 let tickable_chunks = self.ticking_chunks.load();
924 timings.collect_tickable += start.elapsed();
925 timings.total_chunks = self.chunks.len();
926 timings.tickable_count = tickable_chunks.block.len();
927
928 if runs_normally && random_tick_speed > 0 && !tickable_chunks.block.is_empty() {
929 let _span = tracing::trace_span!(
930 "random_chunk_ticks",
931 block_ticking_count = tickable_chunks.block.len(),
932 total_chunks = timings.total_chunks
933 )
934 .entered();
935 let start = Instant::now();
936 let mut random_positions = BlockRandomPositionGenerator::from_runtime_rng();
939 for &index in &tickable_chunks.random_chunk_indices {
940 let tickable_chunk = &tickable_chunks.block[index];
943 if tickable_chunk.randomly_ticking_sections.is_empty() {
944 continue;
945 }
946 if let Some(chunk) = tickable_chunk.holder.try_full_chunk() {
947 chunk.tick_random_blocks(world, random_tick_speed, &mut random_positions);
948 }
949 }
950 timings.tick_chunks += start.elapsed();
951 }
952 timings.lookup_cache.merge(lookup_cache_scope.finish());
953 }
954
955 {
956 let _span = tracing::trace_span!("broadcast_changes").entered();
957 let start = Instant::now();
958 self.broadcast_changed_chunks();
959 timings.broadcast_changes = start.elapsed();
960 }
961
962 {
963 let _span = tracing::trace_span!("process_unloads").entered();
964 let start = Instant::now();
965 self.process_unloads();
966 timings.scheduling.process_unloads = start.elapsed();
967 }
968
969 timings
970 }
971
972 #[instrument(level = "trace", skip(self), name = "advance_chunk_scheduling")]
979 pub(crate) fn advance_scheduling(self: &Arc<Self>) -> ChunkMapSchedulingTimings {
980 let _source_phase_guard = self.source_phase_guard.lock();
981 let mut timings = self.run_chunk_source_updates();
982 let start = Instant::now();
983 self.process_unloads();
984 timings.process_unloads = start.elapsed();
985 timings
986 }
987
988 #[cfg(feature = "test-framework")]
992 pub fn flint_advance_scheduling(self: &Arc<Self>) -> ChunkMapSchedulingTimings {
993 self.advance_scheduling()
994 }
995
996 fn apply_simulation_changes(&self, changes: &[SimulationLevelChange]) -> bool {
997 let world = self.world_gen_context.world();
998 let mut snapshot_changed = false;
999 for change in changes {
1000 let Some(holder) = self
1001 .chunks
1002 .read_sync(&change.pos, |_, holder| Arc::clone(holder))
1003 else {
1004 continue;
1005 };
1006 let previous_level = holder.simulation_level();
1007 if previous_level == change.new_level {
1008 continue;
1009 }
1010
1011 let readiness = holder.ticking_readiness_snapshot().readiness();
1012 let previous_visibility = holder.entity_visibility();
1013 holder.set_simulation_level(change.new_level);
1014 let previous_membership = Self::ticking_snapshot_membership(readiness, previous_level);
1015 let current_membership = Self::ticking_snapshot_membership(readiness, change.new_level);
1016 snapshot_changed |= previous_membership != current_membership;
1017 let new_visibility = holder.entity_visibility();
1018 if previous_visibility != new_visibility
1019 && holder.try_chunk(ChunkStatus::Empty).is_some()
1020 {
1021 world.update_entity_chunk_visibility(change.pos, new_visibility);
1022 }
1023 }
1024 snapshot_changed
1025 }
1026
1027 #[expect(
1028 clippy::too_many_lines,
1029 reason = "the ordered chunk-source commit is kept together so its Vanilla phase guarantees stay auditable"
1030 )]
1031 fn run_chunk_source_updates(self: &Arc<Self>) -> ChunkMapSchedulingTimings {
1032 debug_assert!(!GameplayChunkLookupCacheScope::is_active_for(self));
1035 let mut timings = ChunkMapSchedulingTimings::default();
1036 let mut batch = {
1037 let _span = tracing::trace_span!("ticket_updates").entered();
1038 let start = Instant::now();
1039 let world = self.world_gen_context.world();
1040 let batch = self
1041 .scheduling
1042 .run_all_updates(world.view_distance, world.simulation_distance);
1043 timings.ticket_updates = start.elapsed();
1044 batch
1045 };
1046
1047 {
1048 let _span = tracing::trace_span!("block_entity_unloads").entered();
1049 let start = Instant::now();
1050 self.finish_block_entity_unloads();
1053 timings.block_entity_unloads = start.elapsed();
1054 }
1055
1056 let simulation_snapshot_changed = self.apply_simulation_changes(&batch.simulation_changes);
1057
1058 let (changed_positions, mut rebuild_ticking_snapshot, rebuild_readiness) = {
1059 let _span = tracing::trace_span!("readiness_demotions").entered();
1060 let start = Instant::now();
1061 let changed_positions = batch
1062 .load_changes
1063 .iter()
1064 .map(|change| change.pos)
1065 .collect::<Vec<_>>();
1066 let mut rebuild_ticking_snapshot = simulation_snapshot_changed;
1067 let rebuild_readiness = match self
1068 .prepare_ticking_readiness_demotions(&batch.load_changes)
1069 {
1070 Ok(changed) => {
1071 rebuild_ticking_snapshot |= changed;
1072 false
1073 }
1074 Err(error) => {
1075 tracing::error!(
1076 ?error,
1077 "Full-neighborhood index invariant failed before lifecycle commit; rebuilding after the commit"
1078 );
1079 self.clear_all_ticking_readiness();
1080 *self.full_neighborhood.lock() = FullNeighborhoodIndex::default();
1081 true
1082 }
1083 };
1084 timings.readiness_demotions = start.elapsed();
1085 (
1086 changed_positions,
1087 rebuild_ticking_snapshot,
1088 rebuild_readiness,
1089 )
1090 };
1091
1092 let holders_to_schedule = {
1093 let _span = tracing::trace_span!("lifecycle_commit").entered();
1094 let start = Instant::now();
1095 let holders: Vec<(Arc<ChunkHolder>, ChunkTicketLevel)> = batch
1096 .load_changes
1097 .drain(..)
1098 .filter_map(|change| {
1099 self.update_chunk_level(change.pos, change.new_level)
1100 .zip(change.new_level)
1101 })
1102 .collect();
1103 timings.lifecycle_commit = start.elapsed();
1104 holders
1105 };
1106
1107 let lookup_cache_scope = GameplayChunkLookupCacheScope::enter(self);
1108 let readiness_result = {
1109 let _span = tracing::trace_span!("readiness_reconcile").entered();
1110 let start = Instant::now();
1111 let result = if rebuild_readiness {
1112 rebuild_ticking_snapshot = true;
1113 match self.rebuild_ticking_readiness() {
1114 Ok(result) => result,
1115 Err(error) => self.recover_ticking_readiness_index(error),
1116 }
1117 } else {
1118 match self.reconcile_ticking_readiness_measured(&changed_positions) {
1119 Ok(result) => {
1120 rebuild_ticking_snapshot |= result.snapshot_changed;
1121 result
1122 }
1123 Err(error) => {
1124 rebuild_ticking_snapshot = true;
1125 self.recover_ticking_readiness_index(error)
1126 }
1127 }
1128 };
1129 timings.readiness_reconcile = start.elapsed();
1130 result
1131 };
1132 timings.lookup_cache = lookup_cache_scope.finish();
1133 timings.post_process_generation = readiness_result.post_process_generation;
1134 timings.post_process_chunk_count = readiness_result.post_process_chunk_count;
1135 timings.post_process_position_count = readiness_result.post_process_position_count;
1136 timings.readiness_candidate_count = readiness_result.candidate_count;
1137
1138 if rebuild_ticking_snapshot {
1139 let _span = tracing::trace_span!("ticking_snapshot_rebuild").entered();
1140 let start = Instant::now();
1141 timings.rebuilt_ticking_chunk_count = self.rebuild_ticking_chunk_snapshot();
1142 timings.ticking_snapshot_rebuild = start.elapsed();
1143 }
1144
1145 {
1146 let _span = tracing::trace_span!("schedule_generation").entered();
1147 let start = Instant::now();
1148 timings.scheduled_count = holders_to_schedule
1149 .iter()
1150 .filter(|(holder, level)| {
1151 let Some(status) = generation_status(Some(*level)) else {
1152 return false;
1153 };
1154 holder.schedule_chunk_generation_task_b(status, self)
1155 })
1156 .count();
1157 timings.schedule_generation = start.elapsed();
1158 }
1159
1160 {
1161 let _span = tracing::trace_span!("run_generation").entered();
1162 let start = Instant::now();
1163 self.run_or_notify_generation_refill();
1164 timings.run_generation = start.elapsed();
1165 }
1166
1167 self.scheduling.publish_committed(batch.through_receipt);
1168 self.scheduling.recycle_update_batch(batch);
1169 timings
1170 }
1171
1172 pub fn tickable_full_chunk_positions(&self) -> Vec<ChunkPos> {
1174 let snapshot = self.ticking_chunks.load();
1175 snapshot
1176 .entity_indices
1177 .iter()
1178 .map(|&index| snapshot.block[index].pos)
1179 .collect()
1180 }
1181
1182 pub fn is_entity_ticking_full_chunk_loaded(&self, pos: ChunkPos) -> bool {
1184 self.chunks
1185 .read_sync(&pos, |_, holder| holder.entity_visibility().is_ticking())
1186 .unwrap_or(false)
1187 }
1188
1189 pub(crate) fn block_entity_tick_state_if_owned(
1194 &self,
1195 holder: &Arc<ChunkHolder>,
1196 pos: BlockPos,
1197 expected: &SharedBlockEntity,
1198 ) -> Option<BlockStateId> {
1199 let chunk_pos = ChunkPos::from_block_pos(pos);
1200 let active = self
1201 .chunks
1202 .read_sync(&chunk_pos, |_, current| Arc::ptr_eq(current, holder))
1203 .unwrap_or(false);
1204 if !active
1205 || !is_block_ticking(holder.simulation_level())
1206 || !holder.ticking_readiness_snapshot().is_block_ticking()
1207 {
1208 return None;
1209 }
1210
1211 holder
1212 .try_full_chunk()?
1213 .block_entity_tick_state_if_owned(pos, expected)
1214 }
1215
1216 pub(crate) fn reconcile_block_entity_ticker(&self, holder: &Arc<ChunkHolder>, pos: BlockPos) {
1219 let world = self.world_gen_context.world();
1220 let chunk_pos = ChunkPos::from_block_pos(pos);
1221 let active = self
1222 .chunks
1223 .read_sync(&chunk_pos, |_, current| Arc::ptr_eq(current, holder))
1224 .unwrap_or(false);
1225 if !active {
1226 world.block_entity_tickers().remove(holder, pos);
1227 return;
1228 }
1229
1230 let target = {
1231 let Some(chunk) = holder.try_full_chunk() else {
1232 world.block_entity_tickers().remove(holder, pos);
1233 return;
1234 };
1235 chunk.block_entity_tick_target(pos)
1236 };
1237 let Some((state, block_entity)) = target else {
1238 world.block_entity_tickers().remove(holder, pos);
1239 return;
1240 };
1241 if block_entity.is_removed() {
1242 world.block_entity_tickers().remove(holder, pos);
1243 return;
1244 }
1245
1246 let behavior = BLOCK_BEHAVIORS.get_behavior(state.get_block());
1247 let ticker = behavior.get_block_entity_ticker(&world, state, block_entity.get_type());
1248 let ticker = ticker.filter(|ticker| {
1249 let valid = ticker.accepts(block_entity.get_type());
1250 if !valid {
1251 tracing::error!(
1252 block = %state.get_block().key,
1253 block_entity_type = %block_entity.get_type().key,
1254 ?pos,
1255 "Block behavior returned a ticker for the wrong block-entity type"
1256 );
1257 }
1258 valid
1259 });
1260 world
1261 .block_entity_tickers()
1262 .reconcile(holder, block_entity, ticker);
1263 }
1264
1265 pub(crate) fn activate_block_entities<'a>(
1266 &self,
1267 holders: impl IntoIterator<Item = &'a Arc<ChunkHolder>>,
1268 ) {
1269 for holder in holders {
1270 if !holder.load_level().is_some_and(is_full)
1271 || !self
1272 .chunks
1273 .read_sync(&holder.get_pos(), |_, active| Arc::ptr_eq(active, holder))
1274 .unwrap_or(false)
1275 || !holder.is_full_status_initialized()
1276 || holder.published_status() != Some(ChunkStatus::Full)
1277 {
1278 continue;
1279 }
1280 let batch = {
1281 let Some(chunk) = holder.try_full_chunk() else {
1282 continue;
1283 };
1284 chunk.prepare_block_entity_activation(holder)
1285 };
1286 let Some(batch) = batch else {
1287 continue;
1288 };
1289 for block_entity in batch.lifecycle_dispatchers {
1290 block_entity.dispatch_lifecycle_events();
1291 }
1292 for pos in batch.positions {
1293 {
1294 let Some(chunk) = holder.try_full_chunk() else {
1295 break;
1296 };
1297 chunk.reconcile_block_entity_game_event_listener(pos);
1298 }
1299 self.reconcile_block_entity_ticker(holder, pos);
1300 }
1301 }
1302 }
1303
1304 fn finish_block_entity_unloads(&self) {
1305 let finalized = mem::take(&mut *self.finalized_block_entity_unloads.lock());
1306 if finalized.is_empty() {
1307 return;
1308 }
1309
1310 let world = self.world_gen_context.world();
1311 for mut unload in finalized {
1312 let mut lifecycle_dispatchers = unload
1313 .holder
1314 .try_full_chunk()
1315 .map_or_default(|chunk| chunk.deactivate_block_entities(&unload.holder));
1316 world
1317 .block_entity_tickers()
1318 .remove_positions(&unload.holder, &unload.positions);
1319 lifecycle_dispatchers.append(&mut unload.lifecycle_dispatchers);
1320 for block_entity in lifecycle_dispatchers {
1321 block_entity.dispatch_lifecycle_events();
1322 }
1323 }
1324 }
1325
1326 pub fn place_ender_pearl_ticket(&self, chunk: ChunkPos) {
1335 self.scheduling.add_or_refresh_ender_pearl_ticket(chunk);
1336 }
1337}
1338
1339#[cfg(test)]
1340mod tests;