steel_core/server/
run_loop.rs

1use super::world_tick_workers::{WorldTickWorkerError, WorldTickWorkers};
2use super::{
3    Arc, CCommandSuggestions, CHUNK_SENDING_TPS, COMMAND_DATA_AUTOSAVE_INTERVAL,
4    COMMAND_REQUESTS_PER_TICK, COMMAND_RESUMPTIONS_PER_TICK, CancellationToken, ChunkPos,
5    ChunkSender, CommandExecutionContext, CommandExecutionOwner, CommandRequest,
6    CommandResultCallback, CommandSender, CommandSource, Duration, EncodedChunk,
7    ExecutionCommandSource, ExecutionStop, GameTickTaskGuard, Instant, JoinSet, NetworkConnection,
8    PendingCommandExecutionQueue, Player, SEND_PLAYER_INFO_INTERVAL, SLOW_CHUNK_TICK_THRESHOLD,
9    Server, StringReader, SuggestionError, Suggestions, TAB_LIST_UPDATE_INTERVAL, TabListTickStats,
10    ThreadPool, World, command_suggestions_packet, configured_packet_workers, sleep,
11    spawn_blocking,
12};
13
14impl Server {
15    /// Runs gameplay packets, game ticks, and chunk sending. Game-tick boundaries
16    /// fork background chunk-scheduling epochs through each world's task tracker.
17    pub async fn run(self: Arc<Self>, cancel_token: CancellationToken) {
18        self.packet_processor.open_after_tick();
19        let packet_worker_count = configured_packet_workers(self.config.packet_workers);
20        let mut packet_handles = Vec::with_capacity(packet_worker_count);
21        for worker_id in 0..packet_worker_count {
22            let s = self.clone();
23            let t = cancel_token.clone();
24            packet_handles.push(tokio::spawn(async move {
25                if let Err(error) = spawn_blocking(move || s.packet_processor.run(&s)).await {
26                    log::error!("Gameplay packet worker {worker_id} failed: {error}");
27                    t.cancel();
28                }
29            }));
30        }
31        let packet_supervisor_cancel = cancel_token.clone();
32        let packet_workers = async move {
33            for handle in packet_handles {
34                if let Err(error) = handle.await {
35                    log::error!("Gameplay packet supervisor failed: {error}");
36                    packet_supervisor_cancel.cancel();
37                }
38            }
39        };
40        let game_handle = {
41            let s = self.clone();
42            let t = cancel_token.clone();
43            let task_guard = GameTickTaskGuard::new(self.clone(), cancel_token.clone());
44            tokio::spawn(async move {
45                let _task_guard = task_guard;
46                s.run_game_tick(t).await;
47            })
48        };
49        let chunk_send_handle = {
50            let s = self.clone();
51            let t = cancel_token.clone();
52            tokio::spawn(async move { s.run_chunk_sending_tick(t).await })
53        };
54        let ((), game_result, chunk_send_result) =
55            tokio::join!(packet_workers, game_handle, chunk_send_handle);
56        for (task, result) in [
57            ("Game tick", game_result),
58            ("Chunk sending tick", chunk_send_result),
59        ] {
60            if let Err(error) = result {
61                log::error!("{task} task failed: {error}");
62            }
63        }
64    }
65
66    /// The main game tick loop (20 TPS, governed by tick rate manager).
67    #[expect(
68        clippy::too_many_lines,
69        reason = "the ordered tick phases and their shutdown joins remain easier to audit together"
70    )]
71    async fn run_game_tick(self: Arc<Self>, cancel_token: CancellationToken) {
72        let world_tick_workers = match WorldTickWorkers::spawn(self.worlds.values()) {
73            Ok(workers) => workers,
74            Err(error) => {
75                log::error!("Failed to start world tick workers: {error}");
76                cancel_token.cancel();
77                return;
78            }
79        };
80        let mut next_tick_time = Instant::now();
81        let mut next_command_data_autosave = Instant::now() + COMMAND_DATA_AUTOSAVE_INTERVAL;
82        let mut player_info_ticks = 0_u64;
83        let mut pending_command_executions = PendingCommandExecutionQueue::<CommandSource>::new();
84        let mut player_disconnect_saves = JoinSet::new();
85        let mut command_data_autosaves = JoinSet::new();
86
87        loop {
88            if cancel_token.is_cancelled() {
89                break;
90            }
91
92            let (nanoseconds_per_tick, should_sprint_this_tick) = {
93                let mut tick_manager = self.tick_rate_manager.write();
94                let nanoseconds_per_tick = tick_manager.nanoseconds_per_tick;
95                let (should_sprint, sprint_report) = tick_manager.check_should_sprint_this_tick();
96                drop(tick_manager);
97
98                if let Some(report) = sprint_report {
99                    self.broadcast_sprint_report(&report);
100                    self.broadcast_ticking_state();
101                }
102
103                (nanoseconds_per_tick, should_sprint)
104            };
105
106            if should_sprint_this_tick {
107                next_tick_time = Instant::now();
108            } else {
109                let now = Instant::now();
110                if now < next_tick_time {
111                    tokio::select! {
112                        () = cancel_token.cancelled() => break,
113                        () = sleep(next_tick_time - now) => {}
114                    }
115                }
116                next_tick_time += Duration::from_nanos(nanoseconds_per_tick);
117            }
118
119            if cancel_token.is_cancelled() {
120                break;
121            }
122
123            let tick_start = Instant::now();
124            self.packet_processor.close_for_tick().await;
125            self.advance_chunk_scheduling();
126            self.start_player_disconnect_saves(&mut player_disconnect_saves);
127
128            let (tick_count, runs_normally) = {
129                let mut tick_manager = self.tick_rate_manager.write();
130                tick_manager.tick();
131                let runs_normally = tick_manager.runs_normally();
132                tick_manager.increment_tick_count();
133                (tick_manager.tick_count, runs_normally)
134            };
135
136            self.tick_pending_command_executions(&mut pending_command_executions);
137            self.tick_command_requests(&mut pending_command_executions);
138            if let Err(error) = self
139                .tick_worlds_game(&world_tick_workers, tick_count, runs_normally)
140                .await
141            {
142                log::error!("World game tick failed: {error}");
143                cancel_token.cancel();
144                break;
145            }
146            player_info_ticks += 1;
147            if player_info_ticks > SEND_PLAYER_INFO_INTERVAL {
148                let _span = tracing::trace_span!("broadcast_latency").entered();
149                self.broadcast_player_latency_updates();
150                player_info_ticks = 0;
151            }
152            self.tick_jobs(tick_count, runs_normally);
153            self.process_player_joins();
154
155            {
156                let server = self.clone();
157                let _ =
158                    spawn_blocking(move || server.process_world_changes(tick_count, runs_normally))
159                        .await;
160            }
161
162            self.process_domain_switches();
163
164            self.tick_command_data_autosave(
165                &mut next_command_data_autosave,
166                &mut command_data_autosaves,
167            );
168
169            let tab_list_tick_stats = self.record_tick_and_capture_tab_stats(
170                tick_count,
171                tick_start.elapsed().as_nanos() as u64,
172            );
173
174            if let Some(tick_stats) = tab_list_tick_stats {
175                self.broadcast_tab_list(tick_stats);
176            }
177
178            if should_sprint_this_tick {
179                let mut tick_manager = self.tick_rate_manager.write();
180                tick_manager.end_tick_work();
181            }
182
183            self.packet_processor.open_after_tick();
184            if should_sprint_this_tick || Instant::now() >= next_tick_time {
185                self.packet_processor.wait_for_overload_progress().await;
186            }
187        }
188
189        self.jobs.cancel_all();
190        pending_command_executions.cancel_all();
191        self.command_requests.clear();
192        self.packet_processor.stop();
193        self.start_player_disconnect_saves(&mut player_disconnect_saves);
194        self.pending_player_disconnects.clear();
195        while let Some(result) = player_disconnect_saves.join_next().await {
196            if let Err(error) = result {
197                log::error!("Player disconnect save task failed during shutdown: {error}");
198            }
199        }
200        while let Some(result) = command_data_autosaves.join_next().await {
201            if let Err(error) = result {
202                log::error!("Command data autosave task failed during shutdown: {error}");
203            }
204        }
205    }
206
207    fn record_tick_and_capture_tab_stats(
208        &self,
209        tick_count: u64,
210        tick_duration_nanos: u64,
211    ) -> Option<TabListTickStats> {
212        let mut tick_manager = self.tick_rate_manager.write();
213        tick_manager.record_tick_time(tick_duration_nanos);
214        tick_count
215            .is_multiple_of(TAB_LIST_UPDATE_INTERVAL)
216            .then(|| TabListTickStats::capture(&tick_manager))
217    }
218
219    async fn autosave_command_data(&self) {
220        tracing::debug!("Command data autosave started");
221        let results = self.save_command_data().await;
222        match results.scoreboards {
223            Ok(saved) => tracing::debug!(saved, "Domain scoreboard autosave completed"),
224            Err(error) => tracing::error!(%error, "Domain scoreboard autosave failed"),
225        }
226        match results.storage {
227            Ok(saved) => tracing::debug!(saved, "Domain command-storage autosave completed"),
228            Err(error) => tracing::error!(%error, "Domain command-storage autosave failed"),
229        }
230    }
231
232    fn tick_command_data_autosave(
233        self: &Arc<Self>,
234        next_autosave: &mut Instant,
235        saves: &mut JoinSet<()>,
236    ) {
237        while let Some(result) = saves.try_join_next() {
238            if let Err(error) = result {
239                log::error!("Command data autosave task failed: {error}");
240            }
241        }
242        if Instant::now() < *next_autosave {
243            return;
244        }
245        if saves.is_empty() {
246            let server = Arc::clone(self);
247            saves.spawn(async move {
248                server.autosave_command_data().await;
249            });
250        } else {
251            tracing::warn!("Skipping command data autosave while the previous save runs");
252        }
253        *next_autosave = Instant::now() + COMMAND_DATA_AUTOSAVE_INTERVAL;
254    }
255
256    fn tick_pending_command_executions(
257        &self,
258        pending: &mut PendingCommandExecutionQueue<CommandSource>,
259    ) {
260        let stats = pending.tick(COMMAND_RESUMPTIONS_PER_TICK, |owner| owner.is_current(self));
261        if stats.polled == COMMAND_RESUMPTIONS_PER_TICK && stats.pending > 0 {
262            tracing::debug!(
263                polled = stats.polled,
264                finished = stats.finished,
265                pending = stats.pending,
266                "Command resumption tick reached per-tick processing limit"
267            );
268        }
269    }
270
271    fn tick_command_requests(
272        self: &Arc<Self>,
273        pending: &mut PendingCommandExecutionQueue<CommandSource>,
274    ) {
275        let mut handled = 0;
276        for _ in 0..COMMAND_REQUESTS_PER_TICK {
277            let Some(request) = self
278                .command_requests
279                .pop_front_runnable(|owner| !pending.blocks(owner.key()))
280            else {
281                break;
282            };
283            handled += 1;
284
285            match request {
286                CommandRequest::Execute { owner, command } => {
287                    if !owner.is_current(self) {
288                        continue;
289                    }
290                    self.execute_command_request(pending, owner, &command);
291                }
292                CommandRequest::Suggestions {
293                    owner,
294                    transaction_id,
295                    input,
296                } => {
297                    if !owner.is_current(self) {
298                        continue;
299                    }
300                    let Some(player) = owner.sender().get_player() else {
301                        tracing::error!("command suggestion request has a non-player owner");
302                        continue;
303                    };
304                    self.send_command_suggestions(player, transaction_id, &input);
305                }
306            }
307        }
308
309        if handled == COMMAND_REQUESTS_PER_TICK {
310            tracing::debug!(handled, "Command request tick reached its processing limit");
311        }
312    }
313
314    fn execute_command_request(
315        self: &Arc<Self>,
316        pending: &mut PendingCommandExecutionQueue<CommandSource>,
317        owner: CommandExecutionOwner,
318        command: &str,
319    ) {
320        let source = CommandSource::new(owner.sender().clone(), Arc::clone(self));
321        let command = command.strip_prefix('/').unwrap_or(command);
322        let chain = {
323            let dispatcher = self.command_dispatcher.read();
324            let parse = dispatcher.parse(command, source.clone());
325            dispatcher.context_chain(parse)
326        };
327        let chain = match chain {
328            Ok(chain) => chain,
329            Err(error) => {
330                source.handle_error(&error, false);
331                return;
332            }
333        };
334
335        let mut execution = CommandExecutionContext::for_source(&source);
336        execution.queue_initial_command(chain, source, CommandResultCallback::empty());
337        if execution.run() == ExecutionStop::Suspended && !pending.push_suspended(owner, execution)
338        {
339            tracing::error!("suspended command execution could not be retained");
340        }
341    }
342
343    fn send_command_suggestions(
344        self: &Arc<Self>,
345        player: &Arc<Player>,
346        transaction_id: i32,
347        input: &str,
348    ) {
349        let suggestions =
350            self.build_command_suggestions(CommandSender::Player(Arc::clone(player)), input);
351        match suggestions {
352            Ok(suggestions) => {
353                player.send_packet(command_suggestions_packet(transaction_id, &suggestions));
354            }
355            Err(error) => {
356                tracing::warn!(%error, "failed to build command suggestions");
357                player.send_packet(CCommandSuggestions::new(transaction_id, 0, 0, Vec::new()));
358            }
359        }
360    }
361
362    pub(super) fn build_command_suggestions(
363        self: &Arc<Self>,
364        sender: CommandSender,
365        input: &str,
366    ) -> Result<Suggestions, SuggestionError> {
367        let source = CommandSource::new(sender, Arc::clone(self));
368        let mut reader = StringReader::new(input);
369        if reader.peek() == Some('/') {
370            reader.skip();
371        }
372        let dispatcher = self.command_dispatcher.read();
373        let parse = dispatcher.parse_reader(reader, source);
374        dispatcher.completion_suggestions(&parse)
375    }
376
377    /// Chunk sending tick loop — encodes and sends chunks to players independently.
378    async fn run_chunk_sending_tick(self: Arc<Self>, cancel_token: CancellationToken) {
379        let nanos_per_tick = 1_000_000_000 / CHUNK_SENDING_TPS;
380        let mut next_tick_time = Instant::now();
381
382        loop {
383            if cancel_token.is_cancelled() {
384                break;
385            }
386
387            let now = Instant::now();
388            if now < next_tick_time {
389                tokio::select! {
390                    () = cancel_token.cancelled() => break,
391                    () = sleep(next_tick_time - now) => {}
392                }
393            }
394            next_tick_time += Duration::from_nanos(nanos_per_tick);
395
396            if cancel_token.is_cancelled() {
397                break;
398            }
399
400            let server = self.clone();
401            let _ = spawn_blocking(move || {
402                server.tick_chunk_sending();
403            })
404            .await;
405        }
406    }
407
408    /// Executes one chunk sending tick across all worlds and players.
409    ///
410    /// A per-world per-tick encode cache is used so overlapping view areas
411    /// don't re-encode the same chunk within a single tick.
412    fn tick_chunk_sending(&self) {
413        let tick_start = Instant::now();
414        for world in self.worlds.values() {
415            let mut encode_cache = rustc_hash::FxHashMap::default();
416            world.players.iter_players(|_uuid, player| {
417                Self::send_chunks_for_player(
418                    player,
419                    world,
420                    &mut encode_cache,
421                    self.chunk_encoding_pool.as_ref(),
422                );
423                true
424            });
425        }
426
427        let elapsed = tick_start.elapsed();
428        if elapsed >= SLOW_CHUNK_TICK_THRESHOLD {
429            tracing::warn!(?elapsed, "Chunk sending tick slow");
430        }
431    }
432
433    /// Three-phase chunk send for a single player: prepare (lock briefly),
434    /// encode (no lock), commit (lock briefly + generation check).
435    fn send_chunks_for_player(
436        player: &Arc<Player>,
437        world: &Arc<World>,
438        encode_cache: &mut rustc_hash::FxHashMap<ChunkPos, EncodedChunk>,
439        encoding_pool: &ThreadPool,
440    ) {
441        let chunk_pos = *player.last_chunk_pos.lock();
442        let connection = &player.connection;
443
444        // Phase 1: prepare (brief lock)
445        let prepared = {
446            let mut sender = player.chunk_sender.lock();
447            sender.prepare_batch(world, chunk_pos, &player.chunk_send_epoch)
448        };
449
450        let Some(batch) = prepared else {
451            return;
452        };
453
454        // Phase 2: encode (no lock held — uses per-tick local cache)
455        let compression = connection.compression();
456        let encoded = ChunkSender::encode_batch(&batch, encode_cache, compression, encoding_pool);
457
458        // Phase 3: commit while holding the tracking view that world detachment invalidates.
459        // This makes membership validation, packet commit, and tracker refresh one side of
460        // the same synchronization boundary.
461        let tracking_view = player.last_tracking_view.lock();
462        let Some(view) = *tracking_view else {
463            return;
464        };
465        if !world.contains_player(player) {
466            return;
467        }
468        let sent_chunks = {
469            let mut sender = player.chunk_sender.lock();
470            sender.commit_batch(&batch, encoded, connection, &player.chunk_send_epoch)
471        };
472
473        if sent_chunks.is_empty() {
474            return;
475        }
476
477        let sent_chunks = player.chunk_sender.lock().sent_chunks_snapshot();
478        world
479            .entity_tracker()
480            .update_player(player, &view, |chunk| sent_chunks.contains(&chunk));
481    }
482
483    /// Commits ready chunk lifecycle epochs and forks the next background work.
484    fn advance_chunk_scheduling(&self) {
485        for (i, world) in self.worlds.values().enumerate() {
486            let timings = world.chunk_map.advance_scheduling();
487
488            let background_elapsed = timings.ticket_updates
489                + timings.schedule_generation
490                + timings.run_generation
491                + timings.process_unloads;
492            let boundary_elapsed = timings.block_entity_unloads
493                + timings.readiness_demotions
494                + timings.lifecycle_commit
495                + timings.readiness_reconcile
496                + timings.ticking_snapshot_rebuild;
497            let work_elapsed = background_elapsed + boundary_elapsed;
498
499            if work_elapsed >= SLOW_CHUNK_TICK_THRESHOLD {
500                tracing::warn!(
501                    world = i,
502                    work_elapsed = ?work_elapsed,
503                    background_elapsed = ?background_elapsed,
504                    boundary_elapsed = ?boundary_elapsed,
505                    ticket_updates = ?timings.ticket_updates,
506                    block_entity_unloads = ?timings.block_entity_unloads,
507                    readiness_demotions = ?timings.readiness_demotions,
508                    lifecycle_commit = ?timings.lifecycle_commit,
509                    readiness_reconcile = ?timings.readiness_reconcile,
510                    post_process_generation = ?timings.post_process_generation,
511                    post_process_chunk_count = timings.post_process_chunk_count,
512                    post_process_position_count = timings.post_process_position_count,
513                    readiness_candidate_count = timings.readiness_candidate_count,
514                    ticking_snapshot_rebuild = ?timings.ticking_snapshot_rebuild,
515                    rebuilt_ticking_chunk_count = timings.rebuilt_ticking_chunk_count,
516                    lookup_cache_holder_hits = timings.lookup_cache.holder_hits,
517                    lookup_cache_missing_hits = timings.lookup_cache.missing_hits,
518                    lookup_cache_scc_lookups = timings.lookup_cache.scc_lookups,
519                    lookup_cache_foreign_map_bypasses = timings.lookup_cache.foreign_map_bypasses,
520                    lookup_cache_evictions = timings.lookup_cache.evictions,
521                    schedule_generation = ?timings.schedule_generation,
522                    scheduled_count = timings.scheduled_count,
523                    run_generation = ?timings.run_generation,
524                    process_unloads = ?timings.process_unloads,
525                    "Chunk scheduling epoch slow"
526                );
527            }
528        }
529    }
530
531    #[tracing::instrument(level = "trace", skip(self, workers), name = "tick_worlds")]
532    async fn tick_worlds_game(
533        &self,
534        workers: &WorldTickWorkers,
535        tick_count: u64,
536        runs_normally: bool,
537    ) -> Result<(), WorldTickWorkerError> {
538        let all_timings = workers.tick_all(tick_count, runs_normally).await?;
539        for (i, timings) in all_timings.iter().enumerate() {
540            if timings.elapsed.as_millis() < 50 {
541                continue;
542            }
543            let cm = &timings.chunk_map;
544            tracing::warn!(
545                world = i,
546                elapsed = ?timings.elapsed,
547                tick_count,
548                entity_tick = ?timings.entity_tick,
549                broadcast_changes = ?cm.broadcast_changes,
550                collect_tickable = ?cm.collect_tickable,
551                tick_chunks = ?cm.tick_chunks,
552                tick_block_entities = ?cm.tick_block_entities,
553                tickable_count = cm.tickable_count,
554                total_chunks = cm.total_chunks,
555                lookup_cache_holder_hits = cm.lookup_cache.holder_hits,
556                lookup_cache_missing_hits = cm.lookup_cache.missing_hits,
557                lookup_cache_scc_lookups = cm.lookup_cache.scc_lookups,
558                lookup_cache_foreign_map_bypasses = cm.lookup_cache.foreign_map_bypasses,
559                lookup_cache_evictions = cm.lookup_cache.evictions,
560                "Game tick slow"
561            );
562        }
563        Ok(())
564    }
565
566    pub(super) fn tick_jobs(self: &Arc<Self>, tick_count: u64, runs_normally: bool) {
567        let stats = self
568            .jobs
569            .tick(Arc::downgrade(self), tick_count, runs_normally);
570        if stats.polled > 0 && stats.pending > 0 && tick_count.is_multiple_of(100) {
571            tracing::debug!(
572                polled = stats.polled,
573                finished = stats.finished,
574                pending = stats.pending,
575                "Server jobs pending"
576            );
577        }
578    }
579}
580
581#[cfg(test)]
582mod tests {
583    use std::sync::Arc;
584
585    use rustc_hash::FxHashMap;
586    use uuid::Uuid;
587
588    use super::Server;
589    use crate::{
590        player::ResetReason,
591        test_support::{TestPlayerBuilder, fresh_test_world, insert_ready_full_chunk},
592    };
593    use steel_utils::ChunkPos;
594
595    #[test]
596    fn chunk_send_commit_rechecks_live_world_membership() {
597        let world = fresh_test_world("chunk_send_membership_revalidation");
598        let center = ChunkPos::new(0, 0);
599        insert_ready_full_chunk(&world, center);
600        let player =
601            TestPlayerBuilder::new(Arc::clone(&world), Uuid::from_u128(1), "ChunkTester", 1)
602                .build();
603        assert!(world.add_player(Arc::clone(&player), ResetReason::InitialJoin));
604        assert!(world.players.remove_player_sync(&player).is_some());
605
606        let encoding_pool = rayon::ThreadPoolBuilder::new().num_threads(1).build();
607        let Ok(encoding_pool) = encoding_pool else {
608            panic!("test chunk encoding pool should initialize");
609        };
610        let mut encode_cache = FxHashMap::default();
611        Server::send_chunks_for_player(&player, &world, &mut encode_cache, &encoding_pool);
612
613        let sender = player.chunk_sender.lock();
614        assert!(sender.pending_chunks.contains(&center));
615        assert!(!sender.is_chunk_sent(center));
616        assert_eq!(sender.unacknowledged_batches, 0);
617        drop(sender);
618
619        assert!(world.players.insert(Arc::clone(&player)));
620        world.remove_player_for_world_change(&player);
621    }
622}
steel_core/server/run_loop.rs

steel_core/server/
run_loop.rs
