steel_core/chunk/
chunk_scheduler.rs

1//! Coordinates authoritative ticket storage and chunk-level propagation.
2
3use std::{
4    mem,
5    sync::atomic::{AtomicU64, Ordering},
6    time::Duration,
7};
8
9use rustc_hash::{FxHashMap, FxHashSet};
10use steel_registry::steel_ticket_types::CHUNK_REQUEST;
11use steel_utils::{ChunkPos, locks::SyncMutex};
12use uuid::Uuid;
13
14use crate::chunk::gameplay_chunk_lookup_cache::GameplayChunkLookupCacheStats;
15use crate::chunk::{
16    chunk_ticket::ChunkTicket,
17    chunk_ticket_manager::{ChunkTicketLevel, LoadLevelChange, LoadTicketManager},
18    chunk_ticket_storage::{
19        ChunkTicketStorage, PersistentChunkTickets, SourceLevelUpdate, SourceProjectionChanges,
20        TimedTicketExpiration,
21    },
22    player_ticket_tracker::PlayerTicketTracker,
23    simulation_ticket_manager::{SimulationLevelChange, SimulationTicketManager},
24};
25
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub(crate) enum PlayerTicketOperation {
28    Add { pos: ChunkPos, player_id: Uuid },
29    Remove { pos: ChunkPos, player_id: Uuid },
30}
31
32/// Timing information for one chunk-source update and its lifecycle commit.
33#[derive(Debug, Default)]
34pub struct ChunkMapSchedulingTimings {
35    /// Time spent applying queued source projections and propagating their levels.
36    pub ticket_updates: Duration,
37    /// Time spent finalizing block-entity unloads before lifecycle updates.
38    pub block_entity_unloads: Duration,
39    /// Time spent revoking ticking readiness before holder lifecycle changes.
40    pub readiness_demotions: Duration,
41    /// Time spent committing holder lifecycle changes at the game-tick boundary.
42    pub lifecycle_commit: Duration,
43    /// Time spent reconciling Full neighborhoods and applying ticking readiness.
44    pub readiness_reconcile: Duration,
45    /// Subset of `readiness_reconcile` spent running generation post-processing.
46    pub post_process_generation: Duration,
47    /// Number of chunks whose generation post-processing completed.
48    pub post_process_chunk_count: usize,
49    /// Number of packed generation post-processing positions attempted.
50    pub post_process_position_count: usize,
51    /// Number of readiness candidates considered during reconciliation.
52    pub readiness_candidate_count: usize,
53    /// Time spent rebuilding the published ticking-chunk snapshot.
54    pub ticking_snapshot_rebuild: Duration,
55    /// Number of block-ticking chunks in a snapshot rebuilt during this phase.
56    pub rebuilt_ticking_chunk_count: usize,
57    /// Scoped holder-cache activity during readiness reconciliation.
58    pub lookup_cache: GameplayChunkLookupCacheStats,
59    /// Time spent creating or updating chunk-generation tasks.
60    pub schedule_generation: Duration,
61    /// Number of holders scheduled for generation.
62    pub scheduled_count: usize,
63    /// Time spent refilling generation worker slots.
64    pub run_generation: Duration,
65    /// Time spent processing physical chunk unloads.
66    pub process_unloads: Duration,
67}
68
69/// Barrier assigned to one ordered, non-empty ticket-operation submission.
70#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
71pub(crate) struct ChunkTicketReceipt(u64);
72
73impl ChunkTicketReceipt {
74    const INITIAL: Self = Self(0);
75
76    fn next(self) -> Self {
77        assert_ne!(self.0, u64::MAX, "chunk ticket receipt exhausted");
78        Self(self.0 + 1)
79    }
80}
81
82/// Coherent loading and simulation changes through one ingress barrier.
83#[must_use = "chunk scheduling changes must be applied before publishing the receipt"]
84#[derive(Debug, Default)]
85pub(crate) struct ChunkSchedulingUpdateBatch {
86    pub(crate) through_receipt: ChunkTicketReceipt,
87    pub(crate) load_changes: Vec<LoadLevelChange>,
88    pub(crate) simulation_changes: Vec<SimulationLevelChange>,
89}
90
91#[derive(Debug)]
92struct SourceProjectionBatch {
93    through_receipt: ChunkTicketReceipt,
94    load_updates: Vec<SourceLevelUpdate>,
95    simulation_updates: Vec<SourceLevelUpdate>,
96}
97
98#[derive(Debug)]
99struct TicketOperationIngress {
100    storage: ChunkTicketStorage,
101    player_tickets: PlayerTicketTracker,
102    chunk_request_leases: FxHashMap<(ChunkPos, ChunkTicketLevel), usize>,
103    latest_receipt: ChunkTicketReceipt,
104    dirty_load_positions: FxHashSet<ChunkPos>,
105    dirty_simulation_positions: FxHashSet<ChunkPos>,
106}
107
108impl TicketOperationIngress {
109    fn new(storage: ChunkTicketStorage, view_distance: u8, simulation_distance: u8) -> Self {
110        Self {
111            storage,
112            player_tickets: PlayerTicketTracker::new(view_distance, simulation_distance),
113            chunk_request_leases: FxHashMap::default(),
114            latest_receipt: ChunkTicketReceipt::INITIAL,
115            dirty_load_positions: FxHashSet::default(),
116            dirty_simulation_positions: FxHashSet::default(),
117        }
118    }
119
120    fn push_player(&mut self, operation: PlayerTicketOperation) -> ChunkTicketReceipt {
121        let changes = self.apply_player_operation(operation);
122        self.record_changes(changes);
123        self.allocate_receipt()
124    }
125
126    fn push_player_batch(
127        &mut self,
128        operations: impl IntoIterator<Item = PlayerTicketOperation>,
129    ) -> Option<ChunkTicketReceipt> {
130        let mut operations = operations.into_iter().peekable();
131        operations.peek()?;
132        for operation in operations {
133            let changes = self.apply_player_operation(operation);
134            self.record_changes(changes);
135        }
136        Some(self.allocate_receipt())
137    }
138
139    fn apply_player_operation(
140        &mut self,
141        operation: PlayerTicketOperation,
142    ) -> SourceProjectionChanges {
143        match operation {
144            PlayerTicketOperation::Add { pos, player_id } => {
145                self.player_tickets
146                    .add_player(&mut self.storage, pos, player_id)
147            }
148            PlayerTicketOperation::Remove { pos, player_id } => {
149                self.player_tickets
150                    .remove_player(&mut self.storage, pos, player_id)
151            }
152        }
153    }
154
155    fn acquire_chunk_request_leases(
156        &mut self,
157        positions: impl IntoIterator<Item = ChunkPos>,
158        ticket_level: ChunkTicketLevel,
159    ) -> Option<ChunkTicketReceipt> {
160        let mut positions = positions.into_iter().peekable();
161        positions.peek()?;
162
163        for pos in positions {
164            let lease_count = self
165                .chunk_request_leases
166                .entry((pos, ticket_level))
167                .or_default();
168            assert_ne!(
169                *lease_count,
170                usize::MAX,
171                "chunk request lease count exhausted"
172            );
173            let needs_ticket = *lease_count == 0;
174            *lease_count += 1;
175
176            if needs_ticket {
177                let ticket = ChunkTicket::new(&CHUNK_REQUEST, ticket_level);
178                let changes = self.storage.add_ticket(pos, ticket);
179                self.record_changes(changes);
180            }
181        }
182
183        Some(self.allocate_receipt())
184    }
185
186    fn release_chunk_request_leases(
187        &mut self,
188        positions: impl IntoIterator<Item = ChunkPos>,
189        ticket_level: ChunkTicketLevel,
190    ) -> Option<ChunkTicketReceipt> {
191        let mut positions = positions.into_iter().peekable();
192        positions.peek()?;
193
194        for pos in positions {
195            let key = (pos, ticket_level);
196            let Some(lease_count) = self.chunk_request_leases.get_mut(&key) else {
197                panic!("released an unowned chunk request lease at {pos:?}");
198            };
199            let remove_ticket = *lease_count == 1;
200            if remove_ticket {
201                self.chunk_request_leases.remove(&key);
202                let ticket = ChunkTicket::new(&CHUNK_REQUEST, ticket_level);
203                let changes = self.storage.remove_ticket(pos, ticket);
204                self.record_changes(changes);
205            } else {
206                *lease_count -= 1;
207            }
208        }
209
210        Some(self.allocate_receipt())
211    }
212
213    fn record_changes(&mut self, changes: SourceProjectionChanges) {
214        let SourceProjectionChanges {
215            load_positions,
216            simulation_positions,
217        } = changes;
218
219        self.dirty_load_positions.extend(load_positions);
220        self.dirty_simulation_positions.extend(simulation_positions);
221    }
222
223    fn allocate_receipt(&mut self) -> ChunkTicketReceipt {
224        self.latest_receipt = self.latest_receipt.next();
225        self.latest_receipt
226    }
227
228    fn take_source_projections(
229        &mut self,
230        view_distance: u8,
231        simulation_distance: u8,
232    ) -> SourceProjectionBatch {
233        let view_distance_changes = self
234            .player_tickets
235            .set_view_distance(&mut self.storage, view_distance);
236        self.record_changes(view_distance_changes);
237        let simulation_distance_changes = self
238            .player_tickets
239            .set_simulation_distance(&mut self.storage, simulation_distance);
240        self.record_changes(simulation_distance_changes);
241
242        let load_positions = Self::take_sorted_positions(&mut self.dirty_load_positions);
243        let simulation_positions =
244            Self::take_sorted_positions(&mut self.dirty_simulation_positions);
245        let load_updates = load_positions
246            .into_iter()
247            .map(|pos| self.storage.load_source_update(pos))
248            .collect();
249        let simulation_updates = simulation_positions
250            .into_iter()
251            .map(|pos| self.storage.simulation_source_update(pos))
252            .collect();
253
254        SourceProjectionBatch {
255            through_receipt: self.latest_receipt,
256            load_updates,
257            simulation_updates,
258        }
259    }
260
261    fn take_sorted_positions(positions: &mut FxHashSet<ChunkPos>) -> Vec<ChunkPos> {
262        let mut positions: Vec<_> = mem::take(positions).into_iter().collect();
263        positions.sort_unstable_by_key(|pos| (pos.0.x, pos.0.y));
264        positions
265    }
266}
267
268#[derive(Debug)]
269struct ChunkSchedulingTrackers {
270    load: LoadTicketManager,
271    simulation: SimulationTicketManager,
272}
273
274/// Owns authoritative ticket sources and their two propagation trackers.
275///
276/// Gameplay submissions only lock ingress. Update propagation locks trackers
277/// first, snapshots both source projections under ingress, then releases
278/// ingress before either tracker propagates.
279pub(crate) struct ChunkSchedulingCoordinator {
280    ticket_ingress: SyncMutex<TicketOperationIngress>,
281    trackers: SyncMutex<ChunkSchedulingTrackers>,
282    committed_receipt: AtomicU64,
283}
284
285impl ChunkSchedulingCoordinator {
286    pub fn new(
287        ticket_storage: ChunkTicketStorage,
288        view_distance: u8,
289        simulation_distance: u8,
290    ) -> Self {
291        let initial_load_sources = ticket_storage.initial_load_sources();
292        let initial_simulation_sources = ticket_storage.initial_simulation_sources();
293        let mut load = LoadTicketManager::new();
294        load.apply_source_updates(initial_load_sources);
295        let mut simulation = SimulationTicketManager::new();
296        simulation.apply_source_updates(initial_simulation_sources);
297
298        Self {
299            ticket_ingress: SyncMutex::new(TicketOperationIngress::new(
300                ticket_storage,
301                view_distance,
302                simulation_distance,
303            )),
304            trackers: SyncMutex::new(ChunkSchedulingTrackers { load, simulation }),
305            committed_receipt: AtomicU64::new(ChunkTicketReceipt::INITIAL.0),
306        }
307    }
308
309    pub(crate) fn queue_player_ticket_operation(
310        &self,
311        operation: PlayerTicketOperation,
312    ) -> ChunkTicketReceipt {
313        self.ticket_ingress.lock().push_player(operation)
314    }
315
316    pub(crate) fn queue_player_ticket_operations(
317        &self,
318        operations: impl IntoIterator<Item = PlayerTicketOperation>,
319    ) -> Option<ChunkTicketReceipt> {
320        self.ticket_ingress.lock().push_player_batch(operations)
321    }
322
323    pub(crate) fn acquire_chunk_request_leases(
324        &self,
325        positions: impl IntoIterator<Item = ChunkPos>,
326        ticket_level: ChunkTicketLevel,
327    ) -> Option<ChunkTicketReceipt> {
328        self.ticket_ingress
329            .lock()
330            .acquire_chunk_request_leases(positions, ticket_level)
331    }
332
333    pub(crate) fn release_chunk_request_leases(
334        &self,
335        positions: impl IntoIterator<Item = ChunkPos>,
336        ticket_level: ChunkTicketLevel,
337    ) -> Option<ChunkTicketReceipt> {
338        self.ticket_ingress
339            .lock()
340            .release_chunk_request_leases(positions, ticket_level)
341    }
342
343    pub fn run_all_updates(
344        &self,
345        view_distance: u8,
346        simulation_distance: u8,
347    ) -> ChunkSchedulingUpdateBatch {
348        let mut trackers = self.trackers.lock();
349        let projections = {
350            let mut ingress = self.ticket_ingress.lock();
351            ingress.take_source_projections(view_distance, simulation_distance)
352        };
353
354        trackers
355            .simulation
356            .apply_source_updates(projections.simulation_updates);
357        trackers.simulation.run_all_updates();
358        let simulation_changes = trackers.simulation.take_changes();
359
360        trackers.load.apply_source_updates(projections.load_updates);
361        trackers.load.run_all_updates();
362        let load_changes = trackers.load.take_changes();
363
364        ChunkSchedulingUpdateBatch {
365            through_receipt: projections.through_receipt,
366            load_changes,
367            simulation_changes,
368        }
369    }
370
371    pub(crate) fn add_or_refresh_portal_ticket(&self, pos: ChunkPos) {
372        let mut ingress = self.ticket_ingress.lock();
373        let changes = ingress.storage.add_or_refresh_portal_ticket(pos);
374        ingress.record_changes(changes);
375    }
376
377    pub(crate) fn add_or_refresh_ender_pearl_ticket(&self, pos: ChunkPos) {
378        let mut ingress = self.ticket_ingress.lock();
379        let changes = ingress.storage.add_or_refresh_ender_pearl_ticket(pos);
380        ingress.record_changes(changes);
381    }
382
383    #[must_use]
384    pub(crate) fn timed_ticket_expirations(&self) -> Vec<TimedTicketExpiration> {
385        self.ticket_ingress
386            .lock()
387            .storage
388            .timed_ticket_expirations()
389    }
390
391    pub(crate) fn tick_timed_tickets(&self, expirations: &[TimedTicketExpiration]) {
392        let mut ingress = self.ticket_ingress.lock();
393        let changes = ingress.storage.tick_timed_tickets(expirations);
394        ingress.record_changes(changes);
395    }
396
397    #[must_use]
398    pub(crate) fn persistent_chunk_tickets(&self) -> PersistentChunkTickets {
399        self.ticket_ingress.lock().storage.to_persistent()
400    }
401
402    #[must_use]
403    pub fn simulation_level(&self, pos: ChunkPos) -> Option<ChunkTicketLevel> {
404        self.trackers.lock().simulation.get_level(pos)
405    }
406
407    pub fn recycle_update_batch(&self, batch: ChunkSchedulingUpdateBatch) {
408        let mut trackers = self.trackers.lock();
409        trackers.load.recycle_changes(batch.load_changes);
410        trackers
411            .simulation
412            .recycle_changes(batch.simulation_changes);
413    }
414
415    pub fn publish_committed(&self, receipt: ChunkTicketReceipt) {
416        let _ = self
417            .committed_receipt
418            .fetch_max(receipt.0, Ordering::Release);
419    }
420
421    #[must_use]
422    pub fn is_receipt_committed(&self, receipt: ChunkTicketReceipt) -> bool {
423        self.committed_receipt.load(Ordering::Acquire) >= receipt.0
424    }
425}
426
427#[cfg(test)]
428mod tests {
429    use std::iter;
430
431    use super::*;
432    use crate::chunk::chunk_ticket_manager::ChunkTicketLevel;
433    use uuid::Uuid;
434
435    const TEST_VIEW_DISTANCE_CHUNKS: u8 = 4;
436    const TEST_SIMULATION_DISTANCE_CHUNKS: u8 = 4;
437
438    fn coordinator(simulation_distance: u8) -> ChunkSchedulingCoordinator {
439        ChunkSchedulingCoordinator::new(
440            ChunkTicketStorage::new(),
441            TEST_VIEW_DISTANCE_CHUNKS,
442            simulation_distance,
443        )
444    }
445
446    fn has_load_change(
447        changes: &[LoadLevelChange],
448        pos: ChunkPos,
449        new_level: Option<ChunkTicketLevel>,
450    ) -> bool {
451        changes
452            .iter()
453            .any(|change| change.pos == pos && change.new_level == new_level)
454    }
455
456    fn has_simulation_change(
457        changes: &[SimulationLevelChange],
458        pos: ChunkPos,
459        new_level: Option<ChunkTicketLevel>,
460    ) -> bool {
461        changes
462            .iter()
463            .any(|change| change.pos == pos && change.new_level == new_level)
464    }
465
466    #[test]
467    fn chunk_request_leases_share_one_ticket_and_keep_ordered_receipts() {
468        let pos = ChunkPos::new(6, -9);
469        let level = ChunkTicketLevel::FULL_CHUNK;
470        let mut ingress = TicketOperationIngress::new(
471            ChunkTicketStorage::new(),
472            TEST_VIEW_DISTANCE_CHUNKS,
473            TEST_SIMULATION_DISTANCE_CHUNKS,
474        );
475
476        let first_acquire = ingress
477            .acquire_chunk_request_leases([pos], level)
478            .expect("non-empty acquisition should produce a receipt");
479        let second_acquire = ingress
480            .acquire_chunk_request_leases([pos], level)
481            .expect("non-empty acquisition should produce a receipt");
482        let additions = ingress
483            .take_source_projections(TEST_VIEW_DISTANCE_CHUNKS, TEST_SIMULATION_DISTANCE_CHUNKS);
484
485        assert!(second_acquire > first_acquire);
486        assert_eq!(additions.through_receipt, second_acquire);
487        assert_eq!(
488            additions.load_updates,
489            [SourceLevelUpdate {
490                pos,
491                level: Some(level),
492            }]
493        );
494
495        let first_release = ingress
496            .release_chunk_request_leases([pos], level)
497            .expect("non-empty release should produce a receipt");
498        let retained = ingress
499            .take_source_projections(TEST_VIEW_DISTANCE_CHUNKS, TEST_SIMULATION_DISTANCE_CHUNKS);
500        assert!(first_release > second_acquire);
501        assert_eq!(retained.through_receipt, first_release);
502        assert_eq!(retained.load_updates, []);
503
504        let final_release = ingress
505            .release_chunk_request_leases([pos], level)
506            .expect("non-empty release should produce a receipt");
507        let removal = ingress
508            .take_source_projections(TEST_VIEW_DISTANCE_CHUNKS, TEST_SIMULATION_DISTANCE_CHUNKS);
509        assert!(final_release > first_release);
510        assert_eq!(removal.through_receipt, final_release);
511        assert_eq!(
512            removal.load_updates,
513            [SourceLevelUpdate { pos, level: None }]
514        );
515    }
516
517    #[test]
518    fn empty_batches_have_no_receipt_and_true_no_ops_are_barriers() {
519        let coordinator = coordinator(TEST_SIMULATION_DISTANCE_CHUNKS);
520        assert_eq!(
521            coordinator.queue_player_ticket_operations(iter::empty()),
522            None
523        );
524
525        let pos = ChunkPos::new(0, 0);
526        let receipt = coordinator.queue_player_ticket_operation(PlayerTicketOperation::Remove {
527            pos,
528            player_id: Uuid::from_u128(1),
529        });
530        let batch =
531            coordinator.run_all_updates(TEST_VIEW_DISTANCE_CHUNKS, TEST_SIMULATION_DISTANCE_CHUNKS);
532
533        assert_eq!(receipt, ChunkTicketReceipt(1));
534        assert_eq!(batch.through_receipt, receipt);
535        assert_eq!(batch.load_changes, []);
536        assert_eq!(batch.simulation_changes, []);
537        assert!(!coordinator.is_receipt_committed(receipt));
538
539        coordinator.publish_committed(batch.through_receipt);
540        assert!(coordinator.is_receipt_committed(receipt));
541    }
542
543    #[test]
544    fn unified_batch_and_receipt_commit_atomically() {
545        let coordinator = coordinator(TEST_SIMULATION_DISTANCE_CHUNKS);
546        let pos = ChunkPos::new(0, 0);
547        let player_id = Uuid::from_u128(2);
548        let receipt = coordinator
549            .queue_player_ticket_operation(PlayerTicketOperation::Add { pos, player_id });
550
551        let batch =
552            coordinator.run_all_updates(TEST_VIEW_DISTANCE_CHUNKS, TEST_SIMULATION_DISTANCE_CHUNKS);
553
554        assert_eq!(batch.through_receipt, receipt);
555        assert!(has_load_change(
556            &batch.load_changes,
557            pos,
558            Some(ChunkTicketLevel::ENTITY_TICKING_CHUNK)
559        ));
560        assert!(has_simulation_change(
561            &batch.simulation_changes,
562            pos,
563            Some(ChunkTicketLevel::for_entity_ticking_radius(
564                TEST_SIMULATION_DISTANCE_CHUNKS
565            ))
566        ));
567        assert_eq!(
568            coordinator.simulation_level(pos),
569            Some(ChunkTicketLevel::for_entity_ticking_radius(
570                TEST_SIMULATION_DISTANCE_CHUNKS
571            ))
572        );
573        assert!(!coordinator.is_receipt_committed(receipt));
574
575        coordinator.publish_committed(batch.through_receipt);
576        assert!(coordinator.is_receipt_committed(receipt));
577        coordinator.recycle_update_batch(batch);
578
579        let removal = coordinator
580            .queue_player_ticket_operation(PlayerTicketOperation::Remove { pos, player_id });
581        let batch =
582            coordinator.run_all_updates(TEST_VIEW_DISTANCE_CHUNKS, TEST_SIMULATION_DISTANCE_CHUNKS);
583        assert_eq!(removal, ChunkTicketReceipt(2));
584        assert_eq!(batch.through_receipt, removal);
585        assert!(has_load_change(&batch.load_changes, pos, None));
586        assert!(has_simulation_change(&batch.simulation_changes, pos, None));
587        assert!(!coordinator.is_receipt_committed(removal));
588
589        coordinator.publish_committed(batch.through_receipt);
590        assert!(coordinator.is_receipt_committed(removal));
591    }
592
593    #[test]
594    fn simulation_distance_reprojects_players_without_allocating_a_receipt() {
595        let initial_distance = 2;
596        let updated_distance = 3;
597        let coordinator = coordinator(initial_distance);
598        let pos = ChunkPos::new(5, -3);
599        let receipt = coordinator.queue_player_ticket_operation(PlayerTicketOperation::Add {
600            pos,
601            player_id: Uuid::from_u128(5),
602        });
603
604        let initial = coordinator.run_all_updates(TEST_VIEW_DISTANCE_CHUNKS, initial_distance);
605        assert_eq!(initial.through_receipt, receipt);
606        assert_eq!(
607            coordinator.simulation_level(pos),
608            Some(ChunkTicketLevel::for_entity_ticking_radius(
609                initial_distance
610            ))
611        );
612
613        let updated = coordinator.run_all_updates(TEST_VIEW_DISTANCE_CHUNKS, updated_distance);
614
615        assert_eq!(updated.through_receipt, receipt);
616        assert_eq!(updated.load_changes, []);
617        assert!(has_simulation_change(
618            &updated.simulation_changes,
619            pos,
620            Some(ChunkTicketLevel::for_entity_ticking_radius(
621                updated_distance
622            ))
623        ));
624        assert_eq!(
625            coordinator.simulation_level(pos),
626            Some(ChunkTicketLevel::for_entity_ticking_radius(
627                updated_distance
628            ))
629        );
630    }
631}
steel_core/chunk/chunk_scheduler.rs

steel_core/chunk/
chunk_scheduler.rs
