steel_core/chunk/
chunk_scheduler.rs

1//! Sequencing between gameplay ticket changes and background chunk scheduling.
2
3use std::{
4    mem,
5    sync::atomic::{AtomicU64, Ordering},
6    time::Duration,
7};
8
9use steel_utils::{ChunkPos, locks::SyncMutex};
10
11use crate::chunk::chunk_ticket_manager::{ChunkTicket, ChunkTicketManager, LevelChange};
12use crate::chunk::gameplay_chunk_lookup_cache::GameplayChunkLookupCacheStats;
13
14/// Timing information for one background epoch and its boundary commit.
15#[derive(Debug, Default)]
16pub(crate) struct ChunkMapSchedulingTimings {
17    /// Time spent applying queued ticket operations and propagating their levels.
18    pub(crate) ticket_updates: Duration,
19    /// Time spent finalizing block-entity unloads before the boundary commit.
20    pub(crate) block_entity_unloads: Duration,
21    /// Time spent revoking ticking readiness before holder lifecycle changes.
22    pub(crate) readiness_demotions: Duration,
23    /// Time spent committing holder lifecycle changes at the game-tick boundary.
24    pub(crate) lifecycle_commit: Duration,
25    /// Time spent reconciling Full neighborhoods and applying ticking readiness.
26    pub(crate) readiness_reconcile: Duration,
27    /// Subset of `readiness_reconcile` spent running generation post-processing.
28    pub(crate) post_process_generation: Duration,
29    /// Number of chunks whose generation post-processing completed.
30    pub(crate) post_process_chunk_count: usize,
31    /// Number of packed generation post-processing positions attempted.
32    pub(crate) post_process_position_count: usize,
33    /// Number of readiness candidates considered during reconciliation.
34    pub(crate) readiness_candidate_count: usize,
35    /// Time spent rebuilding the published ticking-chunk snapshot.
36    pub(crate) ticking_snapshot_rebuild: Duration,
37    /// Number of block-ticking chunks in a snapshot rebuilt during this epoch.
38    pub(crate) rebuilt_ticking_chunk_count: usize,
39    /// Scoped holder-cache activity during readiness reconciliation.
40    pub(crate) lookup_cache: GameplayChunkLookupCacheStats,
41    /// Time spent creating or updating chunk-generation tasks.
42    pub(crate) schedule_generation: Duration,
43    /// Number of holders scheduled for generation.
44    pub(crate) scheduled_count: usize,
45    /// Time spent refilling generation worker slots.
46    pub(crate) run_generation: Duration,
47    /// Time spent processing physical chunk unloads.
48    pub(crate) process_unloads: Duration,
49}
50
51/// Timing information produced by the background half of a scheduling epoch.
52///
53/// Boundary-only fields stay out of `PreparedChunkSchedulingEpoch` so the
54/// cross-thread scheduling state does not grow with game-thread observability.
55#[derive(Debug, Default)]
56pub(crate) struct ChunkMapPreparationTimings {
57    pub(crate) ticket_updates: Duration,
58    pub(crate) schedule_generation: Duration,
59    pub(crate) scheduled_count: usize,
60    pub(crate) run_generation: Duration,
61    pub(crate) process_unloads: Duration,
62}
63
64impl ChunkMapPreparationTimings {
65    pub(crate) fn into_scheduling_timings(self) -> ChunkMapSchedulingTimings {
66        ChunkMapSchedulingTimings {
67            ticket_updates: self.ticket_updates,
68            schedule_generation: self.schedule_generation,
69            scheduled_count: self.scheduled_count,
70            run_generation: self.run_generation,
71            process_unloads: self.process_unloads,
72            ..ChunkMapSchedulingTimings::default()
73        }
74    }
75}
76
77/// Revision assigned to an ordered batch of ticket operations.
78#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
79pub(crate) struct ChunkTicketRevision(u64);
80
81impl ChunkTicketRevision {
82    const INITIAL: Self = Self(0);
83
84    fn next(self) -> Self {
85        assert_ne!(self.0, u64::MAX, "chunk ticket revision exhausted");
86        Self(self.0 + 1)
87    }
88}
89
90/// One source-level ticket mutation submitted by gameplay.
91#[derive(Debug, Clone, Copy, PartialEq, Eq)]
92pub(crate) enum ChunkTicketOperation {
93    Add { pos: ChunkPos, ticket: ChunkTicket },
94    Remove { pos: ChunkPos, ticket: ChunkTicket },
95}
96
97impl ChunkTicketOperation {
98    fn apply(self, ticket_manager: &mut ChunkTicketManager) {
99        match self {
100            Self::Add { pos, ticket } => ticket_manager.add_ticket(pos, ticket),
101            Self::Remove { pos, ticket } => {
102                ticket_manager.remove_ticket(pos, ticket);
103            }
104        }
105    }
106}
107
108#[derive(Debug, Clone, Copy)]
109struct QueuedChunkTicketOperation {
110    revision: ChunkTicketRevision,
111    operation: ChunkTicketOperation,
112}
113
114#[derive(Debug)]
115struct PendingChunkTicketOperations {
116    next_revision: ChunkTicketRevision,
117    operations: Vec<QueuedChunkTicketOperation>,
118    recycled_operations: Vec<QueuedChunkTicketOperation>,
119}
120
121impl Default for PendingChunkTicketOperations {
122    fn default() -> Self {
123        Self {
124            next_revision: ChunkTicketRevision::INITIAL,
125            operations: Vec::new(),
126            recycled_operations: Vec::new(),
127        }
128    }
129}
130
131impl PendingChunkTicketOperations {
132    fn push(&mut self, operation: ChunkTicketOperation) -> ChunkTicketRevision {
133        let revision = self.next_revision.next();
134        self.next_revision = revision;
135        self.operations.push(QueuedChunkTicketOperation {
136            revision,
137            operation,
138        });
139        revision
140    }
141
142    fn push_batch(
143        &mut self,
144        operations: impl IntoIterator<Item = ChunkTicketOperation>,
145    ) -> Option<ChunkTicketRevision> {
146        let mut operations = operations.into_iter();
147        let first = operations.next()?;
148        let revision = self.next_revision.next();
149        self.next_revision = revision;
150        self.operations.push(QueuedChunkTicketOperation {
151            revision,
152            operation: first,
153        });
154        self.operations
155            .extend(operations.map(|operation| QueuedChunkTicketOperation {
156                revision,
157                operation,
158            }));
159        Some(revision)
160    }
161
162    fn take(&mut self) -> Vec<QueuedChunkTicketOperation> {
163        mem::replace(
164            &mut self.operations,
165            mem::take(&mut self.recycled_operations),
166        )
167    }
168
169    fn recycle(&mut self, mut operations: Vec<QueuedChunkTicketOperation>) {
170        operations.clear();
171        self.recycled_operations = operations;
172    }
173}
174
175pub(crate) struct PreparedChunkSchedulingEpoch {
176    pub ticket_manager: ChunkTicketManager,
177    pub applied_revision: ChunkTicketRevision,
178    pub changes: Vec<LevelChange>,
179    pub timings: ChunkMapPreparationTimings,
180}
181
182enum ChunkSchedulingState {
183    Idle {
184        ticket_manager: ChunkTicketManager,
185        applied_revision: ChunkTicketRevision,
186    },
187    Running,
188    Ready(PreparedChunkSchedulingEpoch),
189}
190
191pub(crate) enum ChunkSchedulingBoundaryStep {
192    Running,
193    Start {
194        ticket_manager: ChunkTicketManager,
195        applied_revision: ChunkTicketRevision,
196    },
197    Commit(PreparedChunkSchedulingEpoch),
198}
199
200/// Owns the short ticket-ingress lock and the non-blocking epoch handoff.
201/// The propagation manager moves between epochs instead of being cloned or
202/// locked by gameplay.
203pub(crate) struct ChunkSchedulingCoordinator {
204    pending_ticket_operations: SyncMutex<PendingChunkTicketOperations>,
205    state: SyncMutex<ChunkSchedulingState>,
206    committed_revision: AtomicU64,
207}
208
209impl ChunkSchedulingCoordinator {
210    pub fn new(ticket_manager: ChunkTicketManager) -> Self {
211        Self {
212            pending_ticket_operations: SyncMutex::new(PendingChunkTicketOperations::default()),
213            state: SyncMutex::new(ChunkSchedulingState::Idle {
214                ticket_manager,
215                applied_revision: ChunkTicketRevision::INITIAL,
216            }),
217            committed_revision: AtomicU64::new(ChunkTicketRevision::INITIAL.0),
218        }
219    }
220
221    pub fn queue_ticket_operation(&self, operation: ChunkTicketOperation) -> ChunkTicketRevision {
222        self.pending_ticket_operations.lock().push(operation)
223    }
224
225    pub fn queue_ticket_operations(
226        &self,
227        operations: impl IntoIterator<Item = ChunkTicketOperation>,
228    ) -> Option<ChunkTicketRevision> {
229        self.pending_ticket_operations.lock().push_batch(operations)
230    }
231
232    pub fn apply_pending_ticket_operations(
233        &self,
234        ticket_manager: &mut ChunkTicketManager,
235        applied_revision: ChunkTicketRevision,
236    ) -> ChunkTicketRevision {
237        let mut operations = self.pending_ticket_operations.lock().take();
238        let mut latest_revision = applied_revision;
239        for queued in operations.drain(..) {
240            queued.operation.apply(ticket_manager);
241            latest_revision = queued.revision;
242        }
243        self.pending_ticket_operations.lock().recycle(operations);
244        latest_revision
245    }
246
247    pub fn take_boundary_step(&self) -> ChunkSchedulingBoundaryStep {
248        let mut state = self.state.lock();
249        match mem::replace(&mut *state, ChunkSchedulingState::Running) {
250            ChunkSchedulingState::Idle {
251                ticket_manager,
252                applied_revision,
253            } => ChunkSchedulingBoundaryStep::Start {
254                ticket_manager,
255                applied_revision,
256            },
257            ChunkSchedulingState::Running => ChunkSchedulingBoundaryStep::Running,
258            ChunkSchedulingState::Ready(epoch) => ChunkSchedulingBoundaryStep::Commit(epoch),
259        }
260    }
261
262    pub fn finish_epoch(&self, epoch: PreparedChunkSchedulingEpoch) {
263        let mut state = self.state.lock();
264        assert!(
265            matches!(*state, ChunkSchedulingState::Running),
266            "chunk scheduling epoch finished while another epoch was not running"
267        );
268        *state = ChunkSchedulingState::Ready(epoch);
269    }
270
271    pub fn publish_committed_revision(&self, revision: ChunkTicketRevision) {
272        self.committed_revision.store(revision.0, Ordering::Release);
273    }
274
275    #[must_use]
276    pub fn is_revision_committed(&self, revision: ChunkTicketRevision) -> bool {
277        self.committed_revision.load(Ordering::Acquire) >= revision.0
278    }
279}
280
281#[cfg(test)]
282mod tests {
283    use super::*;
284
285    #[test]
286    fn ticket_batches_apply_in_submission_order() {
287        let coordinator = ChunkSchedulingCoordinator::new(ChunkTicketManager::new());
288        let pos = ChunkPos::new(3, -2);
289        let ticket = ChunkTicket::full_chunks(0);
290        let add_revision = coordinator
291            .queue_ticket_operations([ChunkTicketOperation::Add { pos, ticket }])
292            .expect("non-empty batch should receive a revision");
293        let remove_revision = coordinator
294            .queue_ticket_operations([ChunkTicketOperation::Remove { pos, ticket }])
295            .expect("non-empty batch should receive a revision");
296        let mut manager = ChunkTicketManager::new();
297
298        let applied_revision =
299            coordinator.apply_pending_ticket_operations(&mut manager, ChunkTicketRevision::INITIAL);
300
301        assert!(add_revision < remove_revision);
302        assert_eq!(applied_revision, remove_revision);
303        assert_eq!(manager.ticket_count(), 0);
304
305        let next_revision =
306            coordinator.queue_ticket_operation(ChunkTicketOperation::Add { pos, ticket });
307        let applied_revision =
308            coordinator.apply_pending_ticket_operations(&mut manager, applied_revision);
309
310        assert_eq!(applied_revision, next_revision);
311        assert_eq!(manager.ticket_count(), 1);
312    }
313
314    #[test]
315    fn prepared_revision_is_not_visible_before_boundary_publication() {
316        let coordinator = ChunkSchedulingCoordinator::new(ChunkTicketManager::new());
317        let revision = coordinator
318            .queue_ticket_operations([ChunkTicketOperation::Add {
319                pos: ChunkPos::new(0, 0),
320                ticket: ChunkTicket::full_chunks(0),
321            }])
322            .expect("non-empty batch should receive a revision");
323        let ChunkSchedulingBoundaryStep::Start {
324            mut ticket_manager,
325            applied_revision,
326        } = coordinator.take_boundary_step()
327        else {
328            panic!("idle coordinator should start its first epoch");
329        };
330        assert!(matches!(
331            coordinator.take_boundary_step(),
332            ChunkSchedulingBoundaryStep::Running
333        ));
334        let applied =
335            coordinator.apply_pending_ticket_operations(&mut ticket_manager, applied_revision);
336        ticket_manager.run_all_updates();
337        let changes = ticket_manager.take_changes();
338        coordinator.finish_epoch(PreparedChunkSchedulingEpoch {
339            ticket_manager,
340            applied_revision: applied,
341            changes,
342            timings: ChunkMapPreparationTimings::default(),
343        });
344
345        assert_eq!(applied, revision);
346        assert!(!coordinator.is_revision_committed(revision));
347
348        let ChunkSchedulingBoundaryStep::Commit(epoch) = coordinator.take_boundary_step() else {
349            panic!("finished epoch should be committed at the next boundary");
350        };
351        coordinator.publish_committed_revision(epoch.applied_revision);
352
353        assert!(coordinator.is_revision_committed(revision));
354    }
355}
steel_core/chunk/chunk_scheduler.rs

steel_core/chunk/
chunk_scheduler.rs
