steel_core/player/
chunk_sender.rs

1//! This module is responsible for sending chunks to the client.
2//!
3//! Chunk sending runs on its own independent tick loop, separate from the game
4//! tick. The three-phase design (prepare → encode → commit) minimizes lock hold
5//! time on the per-player `ChunkSender` mutex so that game-tick operations like
6//! `mark_chunk_pending_to_send` and `drop_chunk` are never blocked for long.
7use rayon::{ThreadPool, prelude::*};
8use rustc_hash::{FxHashMap, FxHashSet};
9use smallvec::SmallVec;
10use std::{
11    mem,
12    sync::{Arc, Weak},
13};
14
15use steel_protocol::packet_traits::{ClientPacket, CompressionInfo, EncodedPacket};
16use steel_protocol::packets::game::{
17    CChunkBatchFinished, CChunkBatchStart, CForgetLevelChunk, CLevelChunkWithLight,
18};
19use steel_protocol::utils::ConnectionProtocol;
20use steel_utils::locks::SyncMutex;
21use steel_utils::{ChunkPos, PackedChunkPos};
22
23use crate::{
24    chunk::{
25        chunk_holder::{ChunkHolder, TickingReadinessSnapshot},
26        status::ChunkStatus,
27    },
28    player::PlayerConnection,
29    player::connection::NetworkConnection,
30    world::World,
31};
32
33/// Minimum chunks per tick (vanilla: 0.01)
34const MIN_CHUNKS_PER_TICK: f32 = 0.1f32;
35/// Maximum chunks per tick (vanilla: 64.0, we use 500.0 for faster loading)
36const MAX_CHUNKS_PER_TICK: f32 = 500.0;
37/// Starting chunks per tick (vanilla: 9.0)
38const START_CHUNKS_PER_TICK: f32 = 9.0;
39/// Maximum unacknowledged batches after first ack (vanilla: 10)
40const MAX_UNACKNOWLEDGED_BATCHES: u16 = 10;
41
42/// Connection-wide pacing, shared across world changes and player replacements.
43///
44/// Unlike vanilla's single-threaded sender, Steel unlocks the sender while encoding.
45/// ACKs are queued until the next prepare so they cannot reset an in-flight batch's quota.
46#[derive(Debug)]
47struct ChunkBatchPacing {
48    /// Committed batches whose ACK feedback has not yet been applied.
49    unacknowledged_batches: u16,
50    /// Client-reported rate, including fractional chunks of credit per sending tick.
51    desired_chunks_per_tick: f32,
52    /// Accumulated credit; preparation floors this to select whole chunks only.
53    batch_quota: f32,
54    /// Starts at one batch; the first ACK opens the full send window.
55    max_unacknowledged_batches: u16,
56    /// Accepted ACK rates in arrival order, bounded by the outstanding batch count.
57    accepted_feedback: SmallVec<[f32; MAX_UNACKNOWLEDGED_BATCHES as usize]>,
58}
59
60impl ChunkBatchPacing {
61    fn begin_prepare(&mut self) -> Option<usize> {
62        self.drain_feedback();
63
64        if self.unacknowledged_batches >= self.max_unacknowledged_batches {
65            return None;
66        }
67
68        let max_batch_size = self.desired_chunks_per_tick.max(1.0);
69        self.batch_quota = (self.batch_quota + self.desired_chunks_per_tick).min(max_batch_size);
70        Some(self.batch_quota.floor() as usize)
71    }
72
73    fn commit_batch(&mut self, batch_size: usize) {
74        debug_assert!(batch_size > 0);
75        debug_assert!(batch_size as f32 <= self.batch_quota);
76        self.unacknowledged_batches += 1;
77        self.batch_quota -= batch_size as f32;
78    }
79
80    fn record_feedback(&mut self, desired_chunks_per_tick: f32) -> bool {
81        let outstanding_batch_count = usize::from(self.unacknowledged_batches);
82        if self.accepted_feedback.len() >= outstanding_batch_count
83            || self.accepted_feedback.len() >= usize::from(MAX_UNACKNOWLEDGED_BATCHES)
84        {
85            return false;
86        }
87
88        self.accepted_feedback.push(desired_chunks_per_tick);
89        true
90    }
91
92    fn drain_feedback(&mut self) {
93        // Acceptance reserves at most one ACK per outstanding batch. Commits can
94        // only increase that count, so it reaches zero only on the final queued ACK.
95        debug_assert!(self.accepted_feedback.len() <= usize::from(self.unacknowledged_batches));
96        for desired_chunks_per_tick in mem::take(&mut self.accepted_feedback) {
97            self.unacknowledged_batches = self.unacknowledged_batches.saturating_sub(1);
98            self.desired_chunks_per_tick = if desired_chunks_per_tick.is_nan() {
99                MIN_CHUNKS_PER_TICK
100            } else {
101                desired_chunks_per_tick.clamp(MIN_CHUNKS_PER_TICK, MAX_CHUNKS_PER_TICK)
102            };
103
104            if self.unacknowledged_batches == 0 {
105                self.batch_quota = 1.0;
106            }
107
108            self.max_unacknowledged_batches = MAX_UNACKNOWLEDGED_BATCHES;
109        }
110    }
111}
112
113impl Default for ChunkBatchPacing {
114    fn default() -> Self {
115        Self {
116            unacknowledged_batches: 0,
117            desired_chunks_per_tick: START_CHUNKS_PER_TICK,
118            batch_quota: 0.0,
119            max_unacknowledged_batches: 1,
120            accepted_feedback: SmallVec::new(),
121        }
122    }
123}
124
125/// One chunk selected during the prepare phase.
126pub struct PreparedChunk {
127    /// Chunk position.
128    pub pos: ChunkPos,
129    /// Chunk holder to encode.
130    pub holder: Arc<ChunkHolder>,
131    /// Exact readiness generation observed while selecting the holder.
132    readiness: TickingReadinessSnapshot,
133}
134
135/// Data collected during the prepare phase, used to encode and then commit.
136pub struct PreparedBatch {
137    /// Chunk holders to encode.
138    pub chunks: Vec<PreparedChunk>,
139    /// Whether the world dimension has a vanilla sky-light layer.
140    pub has_skylight: bool,
141    /// Snapshot of the player's generation counter at prepare time.
142    pub epoch_snapshot: u32,
143}
144
145/// Encoded chunk packet plus the holder and readiness generation it was built from.
146#[derive(Clone)]
147pub struct EncodedChunk {
148    pos: ChunkPos,
149    packet: EncodedPacket,
150    content_revision: u64,
151    holder: Weak<ChunkHolder>,
152    readiness: TickingReadinessSnapshot,
153}
154
155impl EncodedChunk {
156    fn is_current_for(&self, prepared: &PreparedChunk) -> bool {
157        let Some(encoded_holder) = self.holder.upgrade() else {
158            return false;
159        };
160
161        self.pos == prepared.pos
162            && Arc::ptr_eq(&encoded_holder, &prepared.holder)
163            && self.readiness == prepared.readiness
164            && prepared.readiness.is_block_ticking()
165            && prepared.holder.ticking_readiness_snapshot() == prepared.readiness
166            && prepared.holder.packet_content_revision() == self.content_revision
167    }
168}
169
170/// This struct is responsible for sending chunks to the client.
171#[derive(Debug, Default)]
172pub struct ChunkSender {
173    /// A list of chunks that are waiting to be sent to the client.
174    pub pending_chunks: FxHashSet<ChunkPos>,
175    /// Chunks whose initial chunk packet has been queued for this client.
176    sent_chunks: FxHashSet<ChunkPos>,
177    pacing: ChunkBatchPacing,
178}
179
180impl ChunkSender {
181    /// Clears chunk membership from the previous world while preserving connection pacing.
182    pub(crate) fn clear_world_chunks(&mut self) {
183        self.pending_chunks.clear();
184        self.sent_chunks.clear();
185    }
186
187    /// Marks a chunk as pending to be sent to the client.
188    pub fn mark_chunk_pending_to_send(&mut self, pos: ChunkPos) {
189        self.sent_chunks.remove(&pos);
190        self.pending_chunks.insert(pos);
191    }
192
193    /// Drops a chunk from the client's view.
194    pub fn drop_chunk(&mut self, connection: &PlayerConnection, pos: ChunkPos) {
195        self.pending_chunks.remove(&pos);
196        if self.sent_chunks.remove(&pos) && !connection.closed() {
197            Self::send_packet(
198                connection,
199                CForgetLevelChunk {
200                    pos: PackedChunkPos::from(pos),
201                },
202            );
203        }
204    }
205
206    fn send_packet<P: ClientPacket>(connection: &PlayerConnection, packet: P) {
207        let encoded =
208            EncodedPacket::from_bare(packet, connection.compression(), ConnectionProtocol::Play)
209                .expect("Failed to encode packet");
210        connection.send_encoded(encoded);
211    }
212
213    /// Phase 1: Lock briefly to drain pending chunks and snapshot state.
214    ///
215    /// Returns `None` if there is nothing to send this tick.
216    /// The caller must complete or discard the returned batch before preparing
217    /// another one for this sender; the server's sending pass enforces this.
218    pub fn prepare_batch(
219        &mut self,
220        world: &Arc<World>,
221        player_chunk_pos: ChunkPos,
222        chunk_send_epoch: &SyncMutex<u32>,
223    ) -> Option<PreparedBatch> {
224        let max_batch_size = self.pacing.begin_prepare()?;
225        if max_batch_size == 0 || self.pending_chunks.is_empty() {
226            return None;
227        }
228
229        let holders = self.collect_candidates(world, player_chunk_pos, max_batch_size);
230        if holders.is_empty() {
231            return None;
232        }
233
234        let epoch_snapshot = *chunk_send_epoch.lock();
235
236        Some(PreparedBatch {
237            chunks: holders,
238            has_skylight: world.dimension_type.has_skylight,
239            epoch_snapshot,
240        })
241    }
242
243    /// Phase 2: Encode chunks without holding any lock. Called between prepare and commit.
244    ///
245    /// Uses the dedicated encoding pool to encode chunks in parallel. A per-tick
246    /// local cache prevents multiple players sharing the same chunks from
247    /// re-encoding them within the same sending tick.
248    ///
249    /// # Panics
250    /// Panics if a chunk packet fails to encode.
251    pub fn encode_batch(
252        batch: &PreparedBatch,
253        cache: &mut FxHashMap<ChunkPos, EncodedChunk>,
254        compression: Option<CompressionInfo>,
255        encoding_pool: &ThreadPool,
256    ) -> Vec<EncodedChunk> {
257        let cached_chunks = &*cache;
258        let encoded_chunks = encoding_pool.install(|| {
259            batch
260                .chunks
261                .par_iter()
262                .map(|prepared| {
263                    let holder = &prepared.holder;
264                    let pos = prepared.pos;
265
266                    if let Some(cached) = cached_chunks.get(&pos)
267                        && cached.is_current_for(prepared)
268                    {
269                        return Some(cached.clone());
270                    }
271
272                    if !prepared.readiness.is_block_ticking()
273                        || holder.ticking_readiness_snapshot() != prepared.readiness
274                    {
275                        return None;
276                    }
277                    let revision_before = holder.packet_content_revision();
278                    let chunk = holder.try_full_chunk()?;
279
280                    let packet = EncodedPacket::from_bare(
281                        CLevelChunkWithLight {
282                            x: pos.0.x,
283                            z: pos.0.y,
284                            chunk_data: chunk.extract_chunk_data(),
285                            light_data: chunk.extract_light_data(batch.has_skylight),
286                        },
287                        compression,
288                        ConnectionProtocol::Play,
289                    )
290                    .expect("Failed to encode chunk packet");
291                    let revision_after = holder.packet_content_revision();
292                    if revision_before != revision_after
293                        || holder.ticking_readiness_snapshot() != prepared.readiness
294                    {
295                        return None;
296                    }
297
298                    Some(EncodedChunk {
299                        pos,
300                        packet,
301                        content_revision: revision_after,
302                        holder: Arc::downgrade(holder),
303                        readiness: prepared.readiness,
304                    })
305                })
306                .collect::<Vec<_>>()
307        });
308        let encoded_chunks = encoded_chunks.into_iter().flatten().collect::<Vec<_>>();
309
310        for encoded in &encoded_chunks {
311            cache.insert(encoded.pos, encoded.clone());
312        }
313
314        encoded_chunks
315    }
316
317    /// Phase 3: Lock briefly to verify generation counter and send the batch.
318    ///
319    /// If the player teleported between prepare and commit (generation counter
320    /// changed), the batch is discarded.
321    pub fn commit_batch(
322        &mut self,
323        batch: &PreparedBatch,
324        encoded_chunks: Vec<EncodedChunk>,
325        connection: &PlayerConnection,
326        chunk_send_epoch: &SyncMutex<u32>,
327    ) -> Vec<ChunkPos> {
328        let epoch = chunk_send_epoch.lock();
329        if *epoch != batch.epoch_snapshot {
330            return Vec::new();
331        }
332        drop(epoch);
333
334        let valid_chunks =
335            Self::resolve_valid_chunks(&batch.chunks, encoded_chunks, &self.pending_chunks);
336
337        if valid_chunks.is_empty() {
338            return Vec::new();
339        }
340
341        self.pacing.commit_batch(valid_chunks.len());
342
343        Self::send_packet(connection, CChunkBatchStart {});
344
345        let batch_size = valid_chunks.len();
346        for encoded in &valid_chunks {
347            connection.send_encoded(encoded.packet.clone());
348        }
349
350        Self::send_packet(
351            connection,
352            CChunkBatchFinished {
353                batch_size: batch_size as i32,
354            },
355        );
356
357        let mut sent_chunks = Vec::with_capacity(valid_chunks.len());
358        for encoded in valid_chunks {
359            self.pending_chunks.remove(&encoded.pos);
360            self.sent_chunks.insert(encoded.pos);
361            sent_chunks.push(encoded.pos);
362        }
363        sent_chunks
364    }
365
366    /// Keeps the encoded chunks that are still pending and still match their prepared source.
367    ///
368    /// `encoded_chunks` is a subsequence of `prepared` in the same relative order:
369    /// [`Self::encode_batch`] maps over `prepared` with an order-preserving parallel
370    /// iterator and only drops entries. A single forward cursor over `prepared` therefore
371    /// resolves every encoded chunk in one pass, instead of restarting the scan per chunk.
372    fn resolve_valid_chunks(
373        prepared: &[PreparedChunk],
374        encoded_chunks: Vec<EncodedChunk>,
375        pending: &FxHashSet<ChunkPos>,
376    ) -> Vec<EncodedChunk> {
377        let mut prepared_chunks = prepared.iter();
378        let mut valid_chunks = Vec::with_capacity(encoded_chunks.len());
379
380        for encoded in encoded_chunks {
381            if !pending.contains(&encoded.pos) {
382                continue;
383            }
384            let Some(prepared) = prepared_chunks.find(|prepared| prepared.pos == encoded.pos)
385            else {
386                continue;
387            };
388            if encoded.is_current_for(prepared) {
389                valid_chunks.push(encoded);
390            }
391        }
392
393        valid_chunks
394    }
395
396    fn collect_candidates(
397        &mut self,
398        world: &Arc<World>,
399        player_chunk_pos: ChunkPos,
400        max_batch_size: usize,
401    ) -> Vec<PreparedChunk> {
402        let mut candidates: Vec<ChunkPos> = self.pending_chunks.iter().copied().collect();
403
404        candidates.sort_by_key(|pos| Self::chunk_distance_squared(*pos, player_chunk_pos));
405
406        let mut chunks_to_send = Vec::new();
407
408        for pos in candidates {
409            if chunks_to_send.len() >= max_batch_size {
410                break;
411            }
412
413            if let Some(holder) = world
414                .chunk_map
415                .chunks
416                .read_sync(&pos, |_, chunk| chunk.clone())
417                && holder.published_status() == Some(ChunkStatus::Full)
418            {
419                let readiness = holder.ticking_readiness_snapshot();
420                if readiness.is_block_ticking() {
421                    chunks_to_send.push(PreparedChunk {
422                        pos,
423                        holder,
424                        readiness,
425                    });
426                }
427            }
428        }
429        chunks_to_send
430    }
431
432    fn chunk_distance_squared(pos: ChunkPos, player_chunk_pos: ChunkPos) -> u64 {
433        let dx = u64::from(pos.0.x.abs_diff(player_chunk_pos.0.x));
434        let dz = u64::from(pos.0.y.abs_diff(player_chunk_pos.0.y));
435        dx.saturating_mul(dx).saturating_add(dz.saturating_mul(dz))
436    }
437
438    /// Queues accepted client rate feedback for the next prepare boundary.
439    pub fn on_chunk_batch_received_by_client(&mut self, desired_chunks_per_tick: f32) -> bool {
440        self.pacing.record_feedback(desired_chunks_per_tick)
441    }
442
443    /// Returns whether the client has been queued the initial chunk packet.
444    #[must_use]
445    pub fn is_chunk_sent(&self, pos: ChunkPos) -> bool {
446        self.sent_chunks.contains(&pos)
447    }
448
449    /// Returns a snapshot of all sent chunks for this player.
450    #[must_use]
451    pub fn sent_chunks_snapshot(&self) -> FxHashSet<ChunkPos> {
452        self.sent_chunks.clone()
453    }
454
455    #[cfg(test)]
456    pub(crate) fn mark_chunk_sent_for_test(&mut self, pos: ChunkPos) {
457        self.pending_chunks.remove(&pos);
458        self.sent_chunks.insert(pos);
459    }
460
461    #[cfg(test)]
462    pub(crate) const fn unacknowledged_batch_count_for_test(&self) -> u16 {
463        self.pacing.unacknowledged_batches
464    }
465}
466
467#[cfg(any(test, feature = "benchmark-support"))]
468/// Fixtures and direct entry points shared by unit tests and Criterion benchmarks.
469///
470/// The commit phase needs a [`PlayerConnection`] and cannot be driven from a
471/// benchmark, so this exposes the pure batch-resolution step that runs inside it
472/// along with the pieces needed to build a realistic batch.
473pub mod benchmark_support {
474    use std::sync::{Arc, Weak};
475
476    use rustc_hash::FxHashSet;
477    use steel_utils::ChunkPos;
478
479    use super::{ChunkSender, EncodedChunk, PreparedChunk};
480    use crate::chunk::{
481        Chunk,
482        chunk_holder::{ChunkHolder, TickingReadiness},
483        chunk_ticket_manager::ChunkTicketLevel,
484        heightmap::ChunkHeightmaps,
485        light::ChunkLightData,
486        section::{ChunkSection, Sections},
487        status::ChunkStatus,
488    };
489    use crate::world::tick_scheduler::{BlockTickList, FluidTickList};
490    use steel_worldgen::structure::{StructureReferenceMap, StructureStartMap};
491    /// Builds a block-ticking prepared chunk backed by a real empty full chunk.
492    ///
493    /// The holder is the same shape the prepare phase produces, so an encoded
494    /// chunk built from it passes every [`EncodedChunk::is_current_for`] check.
495    #[must_use]
496    pub fn prepared_full_chunk(pos: ChunkPos) -> PreparedChunk {
497        let chunk = Chunk::from_full_disk(
498            Sections::from_owned(vec![ChunkSection::new_empty()].into_boxed_slice()),
499            pos,
500            0,
501            16,
502            Weak::new(),
503            BlockTickList::new(),
504            FluidTickList::new(),
505            ChunkHeightmaps::new(0, 16),
506            Vec::new(),
507            StructureStartMap::default(),
508            StructureReferenceMap::default(),
509            ChunkLightData::for_valid_world_height(0, 16),
510        );
511        let holder = Arc::new(ChunkHolder::new(
512            pos,
513            ChunkTicketLevel::FULL_CHUNK,
514            Some(ChunkTicketLevel::FULL_CHUNK),
515            0,
516            16,
517        ));
518        holder.insert_chunk(chunk, ChunkStatus::Full);
519        holder.transition_ticking_readiness(TickingReadiness::BlockTicking);
520        let readiness = holder.ticking_readiness_snapshot();
521        PreparedChunk {
522            pos,
523            holder,
524            readiness,
525        }
526    }
527
528    /// Position the encoded chunk was built for.
529    #[must_use]
530    pub const fn encoded_pos(encoded: &EncodedChunk) -> ChunkPos {
531        encoded.pos
532    }
533
534    /// Whether an encoded chunk still matches the prepared chunk it came from.
535    ///
536    /// Exposed so a benchmark can drive an alternative resolution strategy through
537    /// the exact validity check the commit phase uses.
538    #[must_use]
539    pub fn encoded_is_current_for(encoded: &EncodedChunk, prepared: &PreparedChunk) -> bool {
540        encoded.is_current_for(prepared)
541    }
542
543    /// Runs the batch-resolution step of [`ChunkSender::commit_batch`].
544    #[must_use]
545    #[expect(
546        clippy::implicit_hasher,
547        reason = "mirrors the FxHashSet the commit phase actually passes"
548    )]
549    pub fn resolve_valid_chunks(
550        prepared: &[PreparedChunk],
551        encoded_chunks: Vec<EncodedChunk>,
552        pending: &FxHashSet<ChunkPos>,
553    ) -> Vec<EncodedChunk> {
554        ChunkSender::resolve_valid_chunks(prepared, encoded_chunks, pending)
555    }
556}
557
558#[cfg(test)]
559mod tests;
steel_core/player/chunk_sender.rs

steel_core/player/
chunk_sender.rs
