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 std::sync::{Arc, Weak};
10
11use steel_protocol::packet_traits::{ClientPacket, CompressionInfo, EncodedPacket};
12use steel_protocol::packets::game::{
13    CChunkBatchFinished, CChunkBatchStart, CForgetLevelChunk, CLevelChunkWithLight,
14};
15use steel_protocol::utils::ConnectionProtocol;
16use steel_utils::locks::SyncMutex;
17use steel_utils::{ChunkPos, PackedChunkPos};
18
19use crate::{
20    chunk::{
21        chunk_holder::{ChunkHolder, TickingReadinessSnapshot},
22        status::ChunkStatus,
23    },
24    player::PlayerConnection,
25    player::connection::NetworkConnection,
26    world::World,
27};
28
29/// Minimum chunks per tick (vanilla: 0.01)
30const MIN_CHUNKS_PER_TICK: f32 = 0.1f32;
31/// Maximum chunks per tick (vanilla: 64.0, we use 500.0 for faster loading)
32const MAX_CHUNKS_PER_TICK: f32 = 500.0;
33/// Starting chunks per tick (vanilla: 9.0)
34const START_CHUNKS_PER_TICK: f32 = 9.0;
35/// Maximum unacknowledged batches after first ack (vanilla: 10)
36const MAX_UNACKNOWLEDGED_BATCHES: u16 = 10;
37
38/// One chunk selected during the prepare phase.
39pub struct PreparedChunk {
40    /// Chunk position.
41    pub pos: ChunkPos,
42    /// Chunk holder to encode.
43    pub holder: Arc<ChunkHolder>,
44    /// Exact readiness generation observed while selecting the holder.
45    readiness: TickingReadinessSnapshot,
46}
47
48/// Data collected during the prepare phase, used to encode and then commit.
49pub struct PreparedBatch {
50    /// Chunk holders to encode.
51    pub chunks: Vec<PreparedChunk>,
52    /// Whether the world dimension has a vanilla sky-light layer.
53    pub has_skylight: bool,
54    /// Snapshot of the player's generation counter at prepare time.
55    pub epoch_snapshot: u32,
56}
57
58/// Encoded chunk packet plus the holder and readiness generation it was built from.
59#[derive(Clone)]
60pub struct EncodedChunk {
61    pos: ChunkPos,
62    packet: EncodedPacket,
63    content_revision: u64,
64    holder: Weak<ChunkHolder>,
65    readiness: TickingReadinessSnapshot,
66}
67
68impl EncodedChunk {
69    fn is_current_for(&self, prepared: &PreparedChunk) -> bool {
70        let Some(encoded_holder) = self.holder.upgrade() else {
71            return false;
72        };
73
74        self.pos == prepared.pos
75            && Arc::ptr_eq(&encoded_holder, &prepared.holder)
76            && self.readiness == prepared.readiness
77            && prepared.readiness.is_block_ticking()
78            && prepared.holder.ticking_readiness_snapshot() == prepared.readiness
79            && prepared.holder.packet_content_revision() == self.content_revision
80    }
81}
82
83/// This struct is responsible for sending chunks to the client.
84#[derive(Debug)]
85pub struct ChunkSender {
86    /// A list of chunks that are waiting to be sent to the client.
87    pub pending_chunks: FxHashSet<ChunkPos>,
88    /// Chunks whose initial chunk packet has been queued for this client.
89    sent_chunks: FxHashSet<ChunkPos>,
90    /// The number of batches that have been sent to the client but have not been acknowledged yet.
91    pub unacknowledged_batches: u16,
92    /// The number of chunks that should be sent to the client per tick.
93    /// This is dynamically adjusted based on client feedback.
94    pub desired_chunks_per_tick: f32,
95    /// The number of chunks that can be sent to the client in the current batch.
96    pub batch_quota: f32,
97    /// The maximum number of unacknowledged batches allowed.
98    /// Starts at 1 and increases to `MAX_UNACKNOWLEDGED_BATCHES` after first ack.
99    pub max_unacknowledged_batches: u16,
100}
101
102impl ChunkSender {
103    /// Marks a chunk as pending to be sent to the client.
104    pub fn mark_chunk_pending_to_send(&mut self, pos: ChunkPos) {
105        self.sent_chunks.remove(&pos);
106        self.pending_chunks.insert(pos);
107    }
108
109    /// Drops a chunk from the client's view.
110    pub fn drop_chunk(&mut self, connection: &PlayerConnection, pos: ChunkPos) {
111        self.pending_chunks.remove(&pos);
112        if self.sent_chunks.remove(&pos) && !connection.closed() {
113            Self::send_packet(
114                connection,
115                CForgetLevelChunk {
116                    pos: PackedChunkPos::from(pos),
117                },
118            );
119        }
120    }
121
122    /// Encodes and sends a packet through the connection.
123    fn send_packet<P: ClientPacket>(connection: &PlayerConnection, packet: P) {
124        let encoded =
125            EncodedPacket::from_bare(packet, connection.compression(), ConnectionProtocol::Play)
126                .expect("Failed to encode packet");
127        connection.send_encoded(encoded);
128    }
129
130    /// Phase 1: Lock briefly to drain pending chunks and snapshot state.
131    ///
132    /// Returns `None` if there is nothing to send this tick.
133    pub fn prepare_batch(
134        &mut self,
135        world: &Arc<World>,
136        player_chunk_pos: ChunkPos,
137        chunk_send_epoch: &SyncMutex<u32>,
138    ) -> Option<PreparedBatch> {
139        if self.unacknowledged_batches >= self.max_unacknowledged_batches {
140            return None;
141        }
142
143        let max_batch_size = self.desired_chunks_per_tick.max(1.0);
144        self.batch_quota = (self.batch_quota + self.desired_chunks_per_tick).min(max_batch_size);
145
146        if self.batch_quota < 1.0 || self.pending_chunks.is_empty() {
147            return None;
148        }
149
150        let holders = self.collect_candidates(world, player_chunk_pos);
151        if holders.is_empty() {
152            return None;
153        }
154
155        let epoch_snapshot = *chunk_send_epoch.lock();
156
157        Some(PreparedBatch {
158            chunks: holders,
159            has_skylight: world.dimension_type.has_skylight,
160            epoch_snapshot,
161        })
162    }
163
164    /// Phase 2: Encode chunks without holding any lock. Called between prepare and commit.
165    ///
166    /// Uses the dedicated encoding pool to encode chunks in parallel. A per-tick
167    /// local cache prevents multiple players sharing the same chunks from
168    /// re-encoding them within the same sending tick.
169    ///
170    /// # Panics
171    /// Panics if a chunk packet fails to encode.
172    pub fn encode_batch(
173        batch: &PreparedBatch,
174        cache: &mut FxHashMap<ChunkPos, EncodedChunk>,
175        compression: Option<CompressionInfo>,
176        encoding_pool: &ThreadPool,
177    ) -> Vec<EncodedChunk> {
178        let cached_chunks = &*cache;
179        let encoded_chunks = encoding_pool.install(|| {
180            batch
181                .chunks
182                .par_iter()
183                .map(|prepared| {
184                    let holder = &prepared.holder;
185                    let pos = prepared.pos;
186
187                    if let Some(cached) = cached_chunks.get(&pos)
188                        && cached.is_current_for(prepared)
189                    {
190                        return Some(cached.clone());
191                    }
192
193                    if !prepared.readiness.is_block_ticking()
194                        || holder.ticking_readiness_snapshot() != prepared.readiness
195                    {
196                        return None;
197                    }
198                    let revision_before = holder.packet_content_revision();
199                    let chunk = holder.try_full_chunk()?;
200
201                    let packet = EncodedPacket::from_bare(
202                        CLevelChunkWithLight {
203                            x: pos.0.x,
204                            z: pos.0.y,
205                            chunk_data: chunk.extract_chunk_data(),
206                            light_data: chunk.extract_light_data(batch.has_skylight),
207                        },
208                        compression,
209                        ConnectionProtocol::Play,
210                    )
211                    .expect("Failed to encode chunk packet");
212                    let revision_after = holder.packet_content_revision();
213                    if revision_before != revision_after
214                        || holder.ticking_readiness_snapshot() != prepared.readiness
215                    {
216                        return None;
217                    }
218
219                    Some(EncodedChunk {
220                        pos,
221                        packet,
222                        content_revision: revision_after,
223                        holder: Arc::downgrade(holder),
224                        readiness: prepared.readiness,
225                    })
226                })
227                .collect::<Vec<_>>()
228        });
229        let encoded_chunks = encoded_chunks.into_iter().flatten().collect::<Vec<_>>();
230
231        for encoded in &encoded_chunks {
232            cache.insert(encoded.pos, encoded.clone());
233        }
234
235        encoded_chunks
236    }
237
238    /// Phase 3: Lock briefly to verify generation counter and send the batch.
239    ///
240    /// If the player teleported between prepare and commit (generation counter
241    /// changed), the batch is discarded.
242    pub fn commit_batch(
243        &mut self,
244        batch: &PreparedBatch,
245        encoded_chunks: Vec<EncodedChunk>,
246        connection: &PlayerConnection,
247        chunk_send_epoch: &SyncMutex<u32>,
248    ) -> Vec<ChunkPos> {
249        let epoch = chunk_send_epoch.lock();
250        if *epoch != batch.epoch_snapshot {
251            return Vec::new();
252        }
253        drop(epoch);
254
255        let mut valid_chunks = Vec::with_capacity(encoded_chunks.len());
256        for encoded in encoded_chunks {
257            if !self.pending_chunks.contains(&encoded.pos) {
258                continue;
259            }
260            let Some(prepared) = batch.chunks.iter().find(|chunk| chunk.pos == encoded.pos) else {
261                continue;
262            };
263            if !encoded.is_current_for(prepared) {
264                continue;
265            }
266            valid_chunks.push(encoded);
267        }
268
269        if valid_chunks.is_empty() {
270            return Vec::new();
271        }
272
273        self.unacknowledged_batches += 1;
274        self.batch_quota -= valid_chunks.len() as f32;
275
276        Self::send_packet(connection, CChunkBatchStart {});
277
278        let batch_size = valid_chunks.len();
279        for encoded in &valid_chunks {
280            connection.send_encoded(encoded.packet.clone());
281        }
282
283        Self::send_packet(
284            connection,
285            CChunkBatchFinished {
286                batch_size: batch_size as i32,
287            },
288        );
289
290        let mut sent_chunks = Vec::with_capacity(valid_chunks.len());
291        for encoded in valid_chunks {
292            self.pending_chunks.remove(&encoded.pos);
293            self.sent_chunks.insert(encoded.pos);
294            sent_chunks.push(encoded.pos);
295        }
296        sent_chunks
297    }
298
299    fn collect_candidates(
300        &mut self,
301        world: &Arc<World>,
302        player_chunk_pos: ChunkPos,
303    ) -> Vec<PreparedChunk> {
304        let max_batch_size = self.batch_quota.floor() as usize;
305        let mut candidates: Vec<ChunkPos> = self.pending_chunks.iter().copied().collect();
306
307        // Sort by distance to player
308        candidates.sort_by_key(|pos| Self::chunk_distance_squared(*pos, player_chunk_pos));
309
310        let mut chunks_to_send = Vec::new();
311
312        for pos in candidates {
313            if chunks_to_send.len() >= max_batch_size {
314                break;
315            }
316
317            if let Some(holder) = world
318                .chunk_map
319                .chunks
320                .read_sync(&pos, |_, chunk| chunk.clone())
321                && holder.published_status() == Some(ChunkStatus::Full)
322            {
323                let readiness = holder.ticking_readiness_snapshot();
324                if readiness.is_block_ticking() {
325                    chunks_to_send.push(PreparedChunk {
326                        pos,
327                        holder,
328                        readiness,
329                    });
330                }
331            }
332        }
333        chunks_to_send
334    }
335
336    fn chunk_distance_squared(pos: ChunkPos, player_chunk_pos: ChunkPos) -> u64 {
337        let dx = u64::from(pos.0.x.abs_diff(player_chunk_pos.0.x));
338        let dz = u64::from(pos.0.y.abs_diff(player_chunk_pos.0.y));
339        dx.saturating_mul(dx).saturating_add(dz.saturating_mul(dz))
340    }
341
342    /// Handles the acknowledgement of a chunk batch from the client.
343    ///
344    /// The client sends back its desired chunks per tick based on how fast it can
345    /// process chunks. We clamp this value and use it to adjust our sending rate.
346    pub const fn on_chunk_batch_received_by_client(
347        &mut self,
348        desired_chunks_per_tick: f32,
349    ) -> bool {
350        if self.unacknowledged_batches == 0 {
351            return false;
352        }
353
354        self.unacknowledged_batches = self.unacknowledged_batches.saturating_sub(1);
355
356        // Handle NaN and clamp to valid range (vanilla uses 0.01-64, we use 0.01-500)
357        self.desired_chunks_per_tick = if desired_chunks_per_tick.is_nan() {
358            MIN_CHUNKS_PER_TICK
359        } else {
360            desired_chunks_per_tick.clamp(MIN_CHUNKS_PER_TICK, MAX_CHUNKS_PER_TICK)
361        };
362
363        // Reset batch quota when all batches are acknowledged
364        if self.unacknowledged_batches == 0 {
365            self.batch_quota = 1.0;
366        }
367
368        // After receiving the first acknowledgement, allow more unacknowledged batches
369        // for better pipelining (vanilla behavior)
370        self.max_unacknowledged_batches = MAX_UNACKNOWLEDGED_BATCHES;
371        true
372    }
373
374    /// Returns whether the client has been queued the initial chunk packet.
375    #[must_use]
376    pub fn is_chunk_sent(&self, pos: ChunkPos) -> bool {
377        self.sent_chunks.contains(&pos)
378    }
379
380    /// Returns a snapshot of all sent chunks for this player.
381    #[must_use]
382    pub fn sent_chunks_snapshot(&self) -> FxHashSet<ChunkPos> {
383        self.sent_chunks.clone()
384    }
385
386    #[cfg(test)]
387    pub(crate) fn mark_chunk_sent_for_test(&mut self, pos: ChunkPos) {
388        self.pending_chunks.remove(&pos);
389        self.sent_chunks.insert(pos);
390    }
391}
392
393impl Default for ChunkSender {
394    fn default() -> Self {
395        Self {
396            pending_chunks: FxHashSet::default(),
397            sent_chunks: FxHashSet::default(),
398            unacknowledged_batches: 0,
399            desired_chunks_per_tick: START_CHUNKS_PER_TICK,
400            batch_quota: 0.0,
401            max_unacknowledged_batches: 1,
402        }
403    }
404}
405
406#[cfg(test)]
407mod tests {
408    use super::*;
409    use crate::behavior::init_behaviors;
410    use crate::chunk::{
411        Chunk,
412        chunk_holder::TickingReadiness,
413        chunk_ticket_manager::ChunkTicketLevel,
414        heightmap::ChunkHeightmaps,
415        light::ChunkLightData,
416        section::{ChunkSection, Sections},
417    };
418    use crate::world::tick_scheduler::{BlockTickList, FluidTickList};
419    use std::sync::Weak;
420    use steel_registry::init_vanilla_registry;
421    use steel_worldgen::structure::{StructureReferenceMap, StructureStartMap};
422
423    fn prepared_full_chunk(pos: ChunkPos) -> PreparedChunk {
424        let chunk = Chunk::from_full_disk(
425            Sections::from_owned(vec![ChunkSection::new_empty()].into_boxed_slice()),
426            pos,
427            0,
428            16,
429            Weak::new(),
430            BlockTickList::new(),
431            FluidTickList::new(),
432            ChunkHeightmaps::new(0, 16),
433            Vec::new(),
434            StructureStartMap::default(),
435            StructureReferenceMap::default(),
436            ChunkLightData::for_valid_world_height(0, 16),
437        );
438        let holder = Arc::new(ChunkHolder::new(
439            pos,
440            ChunkTicketLevel::FULL_CHUNK,
441            Some(ChunkTicketLevel::FULL_CHUNK),
442            0,
443            16,
444        ));
445        holder.insert_chunk(chunk, ChunkStatus::Full);
446        holder.transition_ticking_readiness(TickingReadiness::BlockTicking);
447        let readiness = holder.ticking_readiness_snapshot();
448        PreparedChunk {
449            pos,
450            holder,
451            readiness,
452        }
453    }
454
455    #[test]
456    fn parallel_chunk_encoding_preserves_batch_order_and_cache_entries() {
457        init_vanilla_registry();
458        init_behaviors();
459        let positions = [
460            ChunkPos::new(3, -2),
461            ChunkPos::new(-1, 4),
462            ChunkPos::new(8, 5),
463            ChunkPos::new(0, 0),
464        ];
465        let batch = PreparedBatch {
466            chunks: positions.into_iter().map(prepared_full_chunk).collect(),
467            has_skylight: true,
468            epoch_snapshot: 0,
469        };
470        let encoding_pool = rayon::ThreadPoolBuilder::new()
471            .num_threads(2)
472            .build()
473            .expect("test chunk encoding pool should initialize");
474        let mut cache = FxHashMap::default();
475
476        let encoded = ChunkSender::encode_batch(&batch, &mut cache, None, &encoding_pool);
477
478        assert_eq!(
479            encoded.iter().map(|chunk| chunk.pos).collect::<Vec<_>>(),
480            positions
481        );
482        assert_eq!(cache.len(), positions.len());
483        for chunk in &encoded {
484            let cached = cache
485                .get(&chunk.pos)
486                .expect("every encoded chunk should be cached");
487            assert!(Arc::ptr_eq(
488                &cached.packet.encoded_data,
489                &chunk.packet.encoded_data
490            ));
491        }
492
493        let encoded_again = ChunkSender::encode_batch(&batch, &mut cache, None, &encoding_pool);
494        for (first, second) in encoded.iter().zip(&encoded_again) {
495            assert_eq!(first.pos, second.pos);
496            assert!(Arc::ptr_eq(
497                &first.packet.encoded_data,
498                &second.packet.encoded_data
499            ));
500        }
501    }
502
503    #[test]
504    fn readiness_demotion_invalidates_prepared_chunk_encoding() {
505        init_vanilla_registry();
506        init_behaviors();
507        let prepared = prepared_full_chunk(ChunkPos::new(4, -7));
508        prepared
509            .holder
510            .transition_ticking_readiness(TickingReadiness::Unready);
511        let batch = PreparedBatch {
512            chunks: vec![prepared],
513            has_skylight: true,
514            epoch_snapshot: 0,
515        };
516        let encoding_pool = rayon::ThreadPoolBuilder::new()
517            .num_threads(1)
518            .build()
519            .expect("test chunk encoding pool should initialize");
520        let mut cache = FxHashMap::default();
521
522        assert!(ChunkSender::encode_batch(&batch, &mut cache, None, &encoding_pool).is_empty());
523        assert!(cache.is_empty());
524    }
525
526    #[test]
527    fn encoding_cache_requires_holder_identity_and_exact_readiness_generation() {
528        init_vanilla_registry();
529        init_behaviors();
530        let pos = ChunkPos::new(-5, 9);
531        let first_batch = PreparedBatch {
532            chunks: vec![prepared_full_chunk(pos)],
533            has_skylight: true,
534            epoch_snapshot: 0,
535        };
536        let encoding_pool = rayon::ThreadPoolBuilder::new()
537            .num_threads(1)
538            .build()
539            .expect("test chunk encoding pool should initialize");
540        let mut cache = FxHashMap::default();
541
542        let first = ChunkSender::encode_batch(&first_batch, &mut cache, None, &encoding_pool);
543        assert_eq!(first.len(), 1);
544
545        let replacement_batch = PreparedBatch {
546            chunks: vec![prepared_full_chunk(pos)],
547            has_skylight: true,
548            epoch_snapshot: 0,
549        };
550        let replacement =
551            ChunkSender::encode_batch(&replacement_batch, &mut cache, None, &encoding_pool);
552        assert_eq!(replacement.len(), 1);
553        assert!(!Arc::ptr_eq(
554            &first[0].packet.encoded_data,
555            &replacement[0].packet.encoded_data
556        ));
557
558        let holder = Arc::clone(&replacement_batch.chunks[0].holder);
559        holder.transition_ticking_readiness(TickingReadiness::Unready);
560        holder.transition_ticking_readiness(TickingReadiness::BlockTicking);
561        let rebound_batch = PreparedBatch {
562            chunks: vec![PreparedChunk {
563                pos,
564                readiness: holder.ticking_readiness_snapshot(),
565                holder,
566            }],
567            has_skylight: true,
568            epoch_snapshot: 0,
569        };
570        let rebound = ChunkSender::encode_batch(&rebound_batch, &mut cache, None, &encoding_pool);
571        assert_eq!(rebound.len(), 1);
572        assert!(!Arc::ptr_eq(
573            &replacement[0].packet.encoded_data,
574            &rebound[0].packet.encoded_data
575        ));
576    }
577
578    #[test]
579    fn chunk_batch_ack_without_outstanding_batch_does_not_update_pacing() {
580        let mut sender = ChunkSender::default();
581
582        assert!(!sender.on_chunk_batch_received_by_client(64.0));
583        assert_eq!(sender.unacknowledged_batches, 0);
584        assert_eq!(
585            sender.desired_chunks_per_tick.to_bits(),
586            START_CHUNKS_PER_TICK.to_bits()
587        );
588        assert_eq!(sender.batch_quota.to_bits(), 0.0_f32.to_bits());
589        assert_eq!(sender.max_unacknowledged_batches, 1);
590    }
591
592    #[test]
593    fn chunk_batch_ack_updates_pacing_for_outstanding_batch() {
594        let mut sender = ChunkSender {
595            unacknowledged_batches: 1,
596            ..ChunkSender::default()
597        };
598
599        assert!(sender.on_chunk_batch_received_by_client(f32::NAN));
600        assert_eq!(sender.unacknowledged_batches, 0);
601        assert_eq!(
602            sender.desired_chunks_per_tick.to_bits(),
603            MIN_CHUNKS_PER_TICK.to_bits()
604        );
605        assert_eq!(sender.batch_quota.to_bits(), 1.0_f32.to_bits());
606        assert_eq!(
607            sender.max_unacknowledged_batches,
608            MAX_UNACKNOWLEDGED_BATCHES
609        );
610    }
611
612    #[test]
613    fn marking_chunk_pending_clears_sent_state() {
614        let mut sender = ChunkSender::default();
615        let pos = ChunkPos::new(2, -3);
616        sender.sent_chunks.insert(pos);
617
618        sender.mark_chunk_pending_to_send(pos);
619
620        assert!(sender.pending_chunks.contains(&pos));
621        assert!(!sender.is_chunk_sent(pos));
622    }
623
624    #[test]
625    fn chunk_distance_squared_handles_far_chunk_coordinates() {
626        let distance = ChunkSender::chunk_distance_squared(
627            ChunkPos::new(1_250_000, -1_250_000),
628            ChunkPos::new(0, 0),
629        );
630
631        assert_eq!(distance, 3_125_000_000_000);
632    }
633
634    #[test]
635    fn chunk_distance_squared_handles_valid_world_extremes() {
636        let max = ChunkPos::MAX_COORDINATE_VALUE;
637        let delta = u64::from(max.abs_diff(-max));
638        let expected = delta * delta * 2;
639
640        let distance =
641            ChunkSender::chunk_distance_squared(ChunkPos::new(max, max), ChunkPos::new(-max, -max));
642
643        assert_eq!(distance, expected);
644    }
645
646    #[test]
647    fn chunk_distance_squared_saturates_for_invalid_i32_extremes() {
648        let distance = ChunkSender::chunk_distance_squared(
649            ChunkPos::new(i32::MIN, i32::MIN),
650            ChunkPos::new(i32::MAX, i32::MAX),
651        );
652
653        assert_eq!(distance, u64::MAX);
654    }
655}
steel_core/player/chunk_sender.rs

steel_core/player/
chunk_sender.rs
