Address MR 2207 review comments.

- Add metrics for which branch of the compression heuristic was taken.
- Reduce the threshold for the heuristic.
- Deduplicate code for dealing with lazy messages.
- Make jpeg dependency only scoped to the compression benchmark.
- Remove commented code.

Cargo.lock

@@ -5516,7 +5516,6 @@ dependencies = [
  "flate2",
  "hashbrown",
  "image",
- "inline_tweak",
  "num-traits",
  "serde",
  "specs",

common/net/src/msg/compression.rs

@@ -397,8 +397,6 @@ impl<const N: u32> VoxelImageEncoding for QuadPngEncoding<N> {
     #[inline(always)]
     fn put_solid(ws: &mut Self::Workspace, x: u32, y: u32, kind: BlockKind, rgb: Rgb<u8>) {
         ws.0.put_pixel(x, y, image::Luma([kind as u8]));
-        //ws.1.put_pixel(x, y, image::Luma([0]));
-        //ws.2.put_pixel(x, y, image::Luma([0]));
         ws.3.put_pixel(x / N, y / N, image::Rgb([rgb.r, rgb.g, rgb.b]));
     }
 
@@ -558,15 +556,21 @@ impl<const N: u32> VoxelImageDecoding for QuadPngEncoding<N> {
                                         - (N - 1) as f64 / 2.0;
                                     let d = (y.wrapping_add(dy as u32) % N) as f64
                                         - (N - 1) as f64 / 2.0;
-                                    let _euclid = Vec2::new(c, d).magnitude();
+                                    let euclid = Vec2::new(c, d).magnitude();
                                     let _manhattan = c.abs() + d.abs();
-                                    //println!("{:?}, {:?}, {:?}: {} {}", (x, y), (i, j), (c, d),
+                                    match 2 {
-                                    // euclid, manhattan);
+                                        0 => {
-                                    // rgb += lanczos(a * euclid, b) * pix;
+                                            rgb += lanczos(a * euclid, b) * pix;
-                                    //rgb += lanczos(a * c, b) * lanczos(a * d, b) * pix;
+                                        },
+                                        1 => {
+                                            rgb += lanczos(a * c, b) * lanczos(a * d, b) * pix;
+                                        },
+                                        _ => {
                                             rgb += lanczos(a * (i as f64 - (x / N) as f64), b)
                                                 * lanczos(a * (j as f64 - (y / N) as f64), b)
                                                 * pix;
+                                        },
+                                    }
                                 }
                             }
                         }

common/net/src/msg/server.rs

@@ -75,7 +75,7 @@ pub enum SerializedTerrainChunk {
 
 /// If someone has less than this number of bytes per second of bandwidth, spend
 /// more CPU generating a smaller encoding of terrain data.
-pub const TERRAIN_LOW_BANDWIDTH: f32 = 5_000_000.0;
+pub const TERRAIN_LOW_BANDWIDTH: f32 = 500_000.0;
 
 impl SerializedTerrainChunk {
     pub fn via_heuristic(chunk: &TerrainChunk, low_bandwidth: bool) -> Self {

server/src/lib.rs

@@ -6,8 +6,10 @@
     label_break_value,
     bool_to_option,
     drain_filter,
+    never_type,
     option_unwrap_none,
-    option_zip
+    option_zip,
+    unwrap_infallible
 )]
 #![cfg_attr(not(feature = "worldgen"), feature(const_panic))]
 

server/src/metrics.rs

@@ -34,6 +34,8 @@ pub struct NetworkRequestMetrics {
     pub chunks_request_dropped: IntCounter,
     pub chunks_served_from_memory: IntCounter,
     pub chunks_generation_triggered: IntCounter,
+    pub chunks_served_lo_bandwidth: IntCounter,
+    pub chunks_served_hi_bandwidth: IntCounter,
 }
 
 pub struct ChunkGenMetrics {
@@ -187,15 +189,29 @@ impl NetworkRequestMetrics {
             "chunks_generation_triggered",
             "number of all chunks that were requested and needs to be generated",
         ))?;
+        let chunks_served_lo_bandwidth = IntCounter::with_opts(Opts::new(
+            "chunks_served_lo_bandwidth",
+            "number of chunks that were sent with compression setting by the low bandwidth \
+             heuristic",
+        ))?;
+        let chunks_served_hi_bandwidth = IntCounter::with_opts(Opts::new(
+            "chunks_served_hi_bandwidth",
+            "number of chunks that were sent with compression setting by the high bandwidth \
+             heuristic",
+        ))?;
 
         registry.register(Box::new(chunks_request_dropped.clone()))?;
         registry.register(Box::new(chunks_served_from_memory.clone()))?;
         registry.register(Box::new(chunks_generation_triggered.clone()))?;
+        registry.register(Box::new(chunks_served_lo_bandwidth.clone()))?;
+        registry.register(Box::new(chunks_served_hi_bandwidth.clone()))?;
 
         Ok(Self {
             chunks_request_dropped,
             chunks_served_from_memory,
             chunks_generation_triggered,
+            chunks_served_lo_bandwidth,
+            chunks_served_hi_bandwidth,
         })
     }
 }

server/src/sys/msg/terrain.rs

@@ -88,7 +88,12 @@ impl<'a> System<'a> for Sys {
                                                     &chunk,
                                                     low_bandwidth,
                                                 )),
-                                            })?
+                                            })?;
+                                            if low_bandwidth {
+                                                network_metrics.chunks_served_lo_bandwidth.inc();
+                                            } else {
+                                                network_metrics.chunks_served_hi_bandwidth.inc();
+                                            }
                                         }
                                     },
                                     None => {

server/src/sys/terrain.rs

@@ -1,6 +1,6 @@
 use crate::{
-    chunk_generator::ChunkGenerator, client::Client, presence::Presence, rtsim::RtSim,
+    chunk_generator::ChunkGenerator, client::Client, metrics::NetworkRequestMetrics,
-    settings::Settings, SpawnPoint, Tick,
+    presence::Presence, rtsim::RtSim, settings::Settings, SpawnPoint, Tick,
 };
 use common::{
     comp::{
@@ -17,10 +17,60 @@ use common_ecs::{Job, Origin, Phase, System};
 use common_net::msg::{SerializedTerrainChunk, ServerGeneral, TERRAIN_LOW_BANDWIDTH};
 use common_state::TerrainChanges;
 use comp::Behavior;
-use specs::{Join, Read, ReadStorage, Write, WriteExpect};
+use specs::{Join, Read, ReadExpect, ReadStorage, Write, WriteExpect};
 use std::sync::Arc;
 use vek::*;
 
+pub struct LazyTerrainMessage {
+    lazy_msg_lo: Option<crate::client::PreparedMsg>,
+    lazy_msg_hi: Option<crate::client::PreparedMsg>,
+}
+
+impl LazyTerrainMessage {
+    #[allow(clippy::new_without_default)]
+    pub fn new() -> Self {
+        Self {
+            lazy_msg_lo: None,
+            lazy_msg_hi: None,
+        }
+    }
+
+    pub fn prepare_and_send<'a, A, F: FnOnce() -> Result<&'a common::terrain::TerrainChunk, A>>(
+        &mut self,
+        network_metrics: &NetworkRequestMetrics,
+        client: &Client,
+        chunk_key: &vek::Vec2<i32>,
+        generate_chunk: F,
+    ) -> Result<(), A> {
+        if let Some(participant) = &client.participant {
+            let low_bandwidth = participant.bandwidth() < TERRAIN_LOW_BANDWIDTH;
+            let lazy_msg = if low_bandwidth {
+                &mut self.lazy_msg_lo
+            } else {
+                &mut self.lazy_msg_hi
+            };
+            if lazy_msg.is_none() {
+                *lazy_msg = Some(client.prepare(ServerGeneral::TerrainChunkUpdate {
+                    key: *chunk_key,
+                    chunk: Ok(match generate_chunk() {
+                        Ok(chunk) => SerializedTerrainChunk::via_heuristic(&chunk, low_bandwidth),
+                        Err(e) => return Err(e),
+                    }),
+                }));
+            }
+            lazy_msg.as_ref().map(|ref msg| {
+                let _ = client.send_prepared(&msg);
+                if low_bandwidth {
+                    network_metrics.chunks_served_lo_bandwidth.inc();
+                } else {
+                    network_metrics.chunks_served_hi_bandwidth.inc();
+                }
+            });
+        }
+        Ok(())
+    }
+}
+
 /// This system will handle loading generated chunks and unloading
 /// unneeded chunks.
 ///     1. Inserts newly generated chunks into the TerrainGrid
@@ -36,6 +86,7 @@ impl<'a> System<'a> for Sys {
         Read<'a, Tick>,
         Read<'a, SpawnPoint>,
         Read<'a, Settings>,
+        ReadExpect<'a, NetworkRequestMetrics>,
         WriteExpect<'a, ChunkGenerator>,
         WriteExpect<'a, TerrainGrid>,
         Write<'a, TerrainChanges>,
@@ -56,6 +107,7 @@ impl<'a> System<'a> for Sys {
             tick,
             spawn_point,
             server_settings,
+            network_metrics,
             mut chunk_generator,
             mut terrain,
             mut terrain_changes,
@@ -222,8 +274,7 @@ impl<'a> System<'a> for Sys {
         // Send the chunk to all nearby players.
         use rayon::iter::{IntoParallelIterator, ParallelIterator};
         new_chunks.into_par_iter().for_each(|(key, chunk)| {
-            let mut lazy_msg_lo = None;
+            let mut lazy_msg = LazyTerrainMessage::new();
-            let mut lazy_msg_hi = None;
 
             (&presences, &positions, &clients)
                 .join()
@@ -237,26 +288,11 @@ impl<'a> System<'a> for Sys {
                         .magnitude_squared();
 
                     if adjusted_dist_sqr <= presence.view_distance.pow(2) {
-                        if let Some(participant) = &client.participant {
+                        lazy_msg
-                            let low_bandwidth = participant.bandwidth() < TERRAIN_LOW_BANDWIDTH;
+                            .prepare_and_send::<!, _>(&network_metrics, &client, &key, || {
-                            let lazy_msg = if low_bandwidth {
+                                Ok(&*chunk)
-                                &mut lazy_msg_lo
+                            })
-                            } else {
+                            .into_ok();
-                                &mut lazy_msg_hi
-                            };
-                            if lazy_msg.is_none() {
-                                *lazy_msg =
-                                    Some(client.prepare(ServerGeneral::TerrainChunkUpdate {
-                                        key,
-                                        chunk: Ok(SerializedTerrainChunk::via_heuristic(
-                                            &*chunk,
-                                            low_bandwidth,
-                                        )),
-                                    }));
-                            };
-
-                            lazy_msg.as_ref().map(|msg| client.send_prepared(msg));
-                        }
                     }
                 });
         });

server/src/sys/terrain_sync.rs

@@ -1,9 +1,8 @@
-use crate::{client::Client, presence::Presence};
+use super::terrain::LazyTerrainMessage;
+use crate::{client::Client, metrics::NetworkRequestMetrics, presence::Presence};
 use common::{comp::Pos, terrain::TerrainGrid};
 use common_ecs::{Job, Origin, Phase, System};
-use common_net::msg::{
+use common_net::msg::{CompressedData, ServerGeneral};
-    CompressedData, SerializedTerrainChunk, ServerGeneral, TERRAIN_LOW_BANDWIDTH,
-};
 use common_state::TerrainChanges;
 use specs::{Join, Read, ReadExpect, ReadStorage};
 
@@ -19,6 +18,7 @@ impl<'a> System<'a> for Sys {
         ReadStorage<'a, Pos>,
         ReadStorage<'a, Presence>,
         ReadStorage<'a, Client>,
+        ReadExpect<'a, NetworkRequestMetrics>,
     );
 
     const NAME: &'static str = "terrain_sync";
@@ -27,39 +27,20 @@ impl<'a> System<'a> for Sys {
 
     fn run(
         _job: &mut Job<Self>,
-        (terrain, terrain_changes, positions, presences, clients): Self::SystemData,
+        (terrain, terrain_changes, positions, presences, clients, network_metrics): Self::SystemData,
     ) {
         // Sync changed chunks
         'chunk: for chunk_key in &terrain_changes.modified_chunks {
-            let mut lazy_msg_hi = None;
+            let mut lazy_msg = LazyTerrainMessage::new();
-            let mut lazy_msg_lo = None;
-
             for (presence, pos, client) in (&presences, &positions, &clients).join() {
-                if let Some(participant) = &client.participant {
+                if super::terrain::chunk_in_vd(pos.0, *chunk_key, &terrain, presence.view_distance)
-                    let low_bandwidth = participant.bandwidth() < TERRAIN_LOW_BANDWIDTH;
+                {
-                    let lazy_msg = if low_bandwidth {
+                    if let Err(()) =
-                        &mut lazy_msg_lo
+                        lazy_msg.prepare_and_send(&network_metrics, &client, chunk_key, || {
-                    } else {
+                            terrain.get_key(*chunk_key).ok_or(())
-                        &mut lazy_msg_hi
+                        })
-                    };
+                    {
-                    if super::terrain::chunk_in_vd(
+                        break 'chunk;
-                        pos.0,
-                        *chunk_key,
-                        &terrain,
-                        presence.view_distance,
-                    ) {
-                        if lazy_msg.is_none() {
-                            *lazy_msg = Some(client.prepare(ServerGeneral::TerrainChunkUpdate {
-                                key: *chunk_key,
-                                chunk: Ok(match terrain.get_key(*chunk_key) {
-                                    Some(chunk) => {
-                                        SerializedTerrainChunk::via_heuristic(&chunk, low_bandwidth)
-                                    },
-                                    None => break 'chunk,
-                                }),
-                            }));
-                        }
-                        lazy_msg.as_ref().map(|ref msg| client.send_prepared(&msg));
                     }
                 }
             }

world/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2018"
 [features]
 tracy = ["common/tracy", "common-net/tracy"]
 simd = ["vek/platform_intrinsics"]
-bin_compression = ["lz-fear", "deflate", "flate2", "common-frontend"]
+bin_compression = ["lz-fear", "deflate", "flate2", "common-frontend", "image/jpeg"]
 
 default = ["simd"]
 
@@ -19,7 +19,7 @@ bincode = "1.3.1"
 bitvec = "0.22"
 enum-iterator = "0.6"
 fxhash = "0.2.1"
-image = { version = "0.23.12", default-features = false, features = ["png", "jpeg"] }
+image = { version = "0.23.12", default-features = false, features = ["png"] }
 itertools = "0.10"
 vek = { version = "0.14.1", features = ["serde"] }
 noise = { version = "0.7", default-features = false }

world/src/bin/chunk_compression_benchmarks.rs

@@ -361,7 +361,7 @@ fn main() {
             .map(|v| v + sitepos.as_())
             .enumerate()
         {
-            let chunk = world.generate_chunk(index.as_index_ref(), spiralpos, || false);
+            let chunk = world.generate_chunk(index.as_index_ref(), spiralpos, || false, None);
             if let Ok((chunk, _)) = chunk {
                 let uncompressed = bincode::serialize(&chunk).unwrap();
                 let n = uncompressed.len();
@@ -376,8 +376,6 @@ fn main() {
                 let lz4chonk_pre = Instant::now();
                 let lz4_chonk = lz4_with_dictionary(&bincode::serialize(&chunk).unwrap(), &[]);
                 let lz4chonk_post = Instant::now();
-                //let lz4_dict_chonk = SerializedTerrainChunk::from_chunk(&chunk,
-                // &*dictionary);
                 for _ in 0..ITERS {
                     let _deflate0_chonk =
                         do_deflate_flate2_zero(&bincode::serialize(&chunk).unwrap());
@@ -493,7 +491,6 @@ fn main() {
                         }
                     }
                     let lz4_dyna = lz4_with_dictionary(&*ser_dyna, &[]);
-                    //let lz4_dict_dyna = lz4_with_dictionary(&*ser_dyna, &dictionary2);
                     let deflate_dyna = do_deflate_flate2::<5>(&*ser_dyna);
                     let deflate_channeled_dyna = do_deflate_flate2::<5>(
                         &bincode::serialize(&channelize_dyna(&dyna)).unwrap(),
@@ -507,6 +504,10 @@ fn main() {
                             deflate_channeled_dyna.len() as f32 / n as f32,
                         ),
                     ]);
+                    if HISTOGRAMS {
+                        let lz4_dict_dyna = lz4_with_dictionary(&*ser_dyna, &dictionary2);
+                        sizes.push(("lz4_dict_dyna", lz4_dyna.len() as f32 / n as f32));
+                    }
                 }
 
                 if !SKIP_IMAGECHONK {