Refractored read_chunk
Added new error types to indicate a chunk should be generated. Renamed to `fetch_chunks` to indicate that it no longer just reads chunks (might write to the worldfile in the future) Take a reference to a slice instead of an owned vector because the data is Copy anyways. Remove the ChunkCoordinates that are sent over the mpsc Sender. - The coordinates are stored in ChunkData anyways. - If there is an Error the coordinates should not matter because the error can't be recovered from. Moved the reading from file logic to `read_chunk`. - So that we don't always have to repeat `channel.blocking_send(...); return;` - To make it clearly distinct from the main program logic. Replaced some verbose match statements with `Result.map_err()`.
This commit is contained in:
DaniD3v
parent
cdb0e6b761
commit
17b285081a
@ -1,7 +1,7 @@
|
||||
use std::{
|
||||
fs::OpenOptions,
|
||||
io::{Read, Seek},
|
||||
path::PathBuf,
|
||||
path::{Path, PathBuf},
|
||||
};
|
||||
|
||||
use flate2::{bufread::ZlibDecoder, read::GzDecoder};
|
||||
@ -24,12 +24,10 @@ pub enum WorldError {
|
||||
// using ErrorKind instead of Error, beacuse the function read_chunks and read_region_chunks is designed to return an error on a per-chunk basis, while std::io::Error does not implement Copy or Clone
|
||||
#[error("Io error: {0}")]
|
||||
IoError(std::io::ErrorKind),
|
||||
#[error("Region not found")]
|
||||
RegionNotFound,
|
||||
#[error("Region is invalid")]
|
||||
RegionIsInvalid,
|
||||
#[error("Chunk not found")]
|
||||
ChunkNotFound,
|
||||
#[error("The chunk isn't generated yet: {0}")]
|
||||
ChunkNotGenerated(ChunkNotGeneratedError),
|
||||
#[error("Compression Error")]
|
||||
Compression(CompressionError),
|
||||
#[error("Error deserializing chunk: {0}")]
|
||||
@ -42,6 +40,18 @@ pub enum WorldError {
|
||||
BlockOutsideChunk,
|
||||
}
|
||||
|
||||
#[derive(Error, Debug)]
|
||||
pub enum ChunkNotGeneratedError {
|
||||
#[error("The region file does not exist.")]
|
||||
RegionFileMissing,
|
||||
|
||||
#[error("The chunks generation is incomplete.")]
|
||||
IncompleteGeneration,
|
||||
|
||||
#[error("Chunk not found.")]
|
||||
NotFound,
|
||||
}
|
||||
|
||||
#[derive(Error, Debug)]
|
||||
pub enum CompressionError {
|
||||
#[error("Compression scheme not recognised")]
|
||||
@ -98,132 +108,110 @@ impl Level {
|
||||
// .1
|
||||
// }
|
||||
|
||||
/// Read many chunks in a world
|
||||
/// Reads/Generates many chunks in a world
|
||||
/// MUST be called from a tokio runtime thread
|
||||
///
|
||||
/// Note: The order of the output chunks will almost never be in the same order as the order of input chunks
|
||||
pub async fn read_chunks(
|
||||
pub async fn fetch_chunks(
|
||||
&self,
|
||||
chunks: Vec<ChunkCoordinates>,
|
||||
channel: mpsc::Sender<(ChunkCoordinates, Result<ChunkData, WorldError>)>,
|
||||
chunks: &[ChunkCoordinates],
|
||||
channel: mpsc::Sender<Result<ChunkData, WorldError>>,
|
||||
) {
|
||||
chunks.into_par_iter().for_each(|chunk| {
|
||||
let region = (
|
||||
((chunk.x as f32) / 32.0).floor() as i32,
|
||||
((chunk.z as f32) / 32.0).floor() as i32,
|
||||
);
|
||||
chunks.into_par_iter().copied().for_each(|at| {
|
||||
let channel = channel.clone();
|
||||
|
||||
// return different error when file is not found (because that means that the chunks have just not been generated yet)
|
||||
let mut region_file = match OpenOptions::new().read(true).open(
|
||||
self.region_folder
|
||||
.join(format!("r.{}.{}.mca", region.0, region.1)),
|
||||
) {
|
||||
Ok(f) => f,
|
||||
Err(err) => match err.kind() {
|
||||
std::io::ErrorKind::NotFound => {
|
||||
let _ = channel.blocking_send((chunk, Err(WorldError::RegionNotFound)));
|
||||
return;
|
||||
}
|
||||
_ => {
|
||||
let _ =
|
||||
channel.blocking_send((chunk, Err(WorldError::IoError(err.kind()))));
|
||||
return;
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
let mut location_table: [u8; 4096] = [0; 4096];
|
||||
let mut timestamp_table: [u8; 4096] = [0; 4096];
|
||||
|
||||
// fill the location and timestamp tables
|
||||
{
|
||||
match region_file.read_exact(&mut location_table) {
|
||||
Ok(_) => {}
|
||||
Err(err) => {
|
||||
let _ =
|
||||
channel.blocking_send((chunk, Err(WorldError::IoError(err.kind()))));
|
||||
return;
|
||||
}
|
||||
}
|
||||
match region_file.read_exact(&mut timestamp_table) {
|
||||
Ok(_) => {}
|
||||
Err(err) => {
|
||||
let _ =
|
||||
channel.blocking_send((chunk, Err(WorldError::IoError(err.kind()))));
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let modulus = |a: i32, b: i32| ((a % b) + b) % b;
|
||||
let chunk_x = modulus(chunk.x, 32) as u32;
|
||||
let chunk_z = modulus(chunk.z, 32) as u32;
|
||||
let channel = channel.clone();
|
||||
let table_entry = (chunk_x + chunk_z * 32) * 4;
|
||||
|
||||
let mut offset = vec![0u8];
|
||||
offset
|
||||
.extend_from_slice(&location_table[table_entry as usize..table_entry as usize + 3]);
|
||||
let offset = u32::from_be_bytes(offset.try_into().unwrap()) as u64 * 4096;
|
||||
let size = location_table[table_entry as usize + 3] as usize * 4096;
|
||||
|
||||
if offset == 0 && size == 0 {
|
||||
let _ = channel.blocking_send((chunk, Err(WorldError::ChunkNotFound)));
|
||||
return;
|
||||
}
|
||||
// Read the file using the offset and size
|
||||
let mut file_buf = {
|
||||
let seek_result = region_file.seek(std::io::SeekFrom::Start(offset));
|
||||
if seek_result.is_err() {
|
||||
let _ = channel.blocking_send((chunk, Err(WorldError::RegionIsInvalid)));
|
||||
return;
|
||||
}
|
||||
let mut out = vec![0; size];
|
||||
let read_result = region_file.read_exact(&mut out);
|
||||
if read_result.is_err() {
|
||||
let _ = channel.blocking_send((chunk, Err(WorldError::RegionIsInvalid)));
|
||||
return;
|
||||
}
|
||||
out
|
||||
};
|
||||
|
||||
// TODO: check checksum to make sure chunk is not corrupted
|
||||
let header = file_buf.drain(0..5).collect_vec();
|
||||
|
||||
let compression = match Compression::from_byte(header[4]) {
|
||||
Some(c) => c,
|
||||
None => {
|
||||
let _ = channel.blocking_send((
|
||||
chunk,
|
||||
Err(WorldError::Compression(
|
||||
CompressionError::UnknownCompression,
|
||||
)),
|
||||
));
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let size = u32::from_be_bytes(header[..4].try_into().unwrap());
|
||||
|
||||
// size includes the compression scheme byte, so we need to subtract 1
|
||||
let chunk_data = file_buf.drain(0..size as usize - 1).collect_vec();
|
||||
let decompressed_chunk = match Self::decompress_data(compression, chunk_data) {
|
||||
Ok(data) => data,
|
||||
Err(e) => {
|
||||
channel
|
||||
.blocking_send((chunk, Err(WorldError::Compression(e))))
|
||||
.expect("Failed to send Compression error");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
channel
|
||||
.blocking_send((chunk, ChunkData::from_bytes(decompressed_chunk, chunk)))
|
||||
.expect("Error sending decompressed chunk");
|
||||
.blocking_send(match Self::read_chunk(&self.region_folder, at) {
|
||||
Err(WorldError::ChunkNotGenerated(_)) => {
|
||||
// This chunk was not generated yet.
|
||||
todo!("generate chunk here")
|
||||
}
|
||||
// TODO this doesn't warn the user about the error. fix.
|
||||
result => result,
|
||||
})
|
||||
.expect("Failed sending ChunkData.");
|
||||
})
|
||||
}
|
||||
|
||||
fn read_chunk(region_folder: &Path, at: ChunkCoordinates) -> Result<ChunkData, WorldError> {
|
||||
let region = (
|
||||
((at.x as f32) / 32.0).floor() as i32,
|
||||
((at.z as f32) / 32.0).floor() as i32,
|
||||
);
|
||||
|
||||
let mut region_file = OpenOptions::new()
|
||||
.read(true)
|
||||
.open(region_folder.join(format!("r.{}.{}.mca", region.0, region.1)))
|
||||
.map_err(|err| match err.kind() {
|
||||
std::io::ErrorKind::NotFound => {
|
||||
WorldError::ChunkNotGenerated(ChunkNotGeneratedError::RegionFileMissing)
|
||||
}
|
||||
kind => WorldError::IoError(kind),
|
||||
})?;
|
||||
|
||||
let mut location_table: [u8; 4096] = [0; 4096];
|
||||
let mut timestamp_table: [u8; 4096] = [0; 4096];
|
||||
|
||||
// fill the location and timestamp tables
|
||||
region_file
|
||||
.read_exact(&mut location_table)
|
||||
.map_err(|err| WorldError::IoError(err.kind()))?;
|
||||
region_file
|
||||
.read_exact(&mut timestamp_table)
|
||||
.map_err(|err| WorldError::IoError(err.kind()))?;
|
||||
|
||||
let modulus = |a: i32, b: i32| ((a % b) + b) % b;
|
||||
let chunk_x = modulus(at.x, 32) as u32;
|
||||
let chunk_z = modulus(at.z, 32) as u32;
|
||||
let table_entry = (chunk_x + chunk_z * 32) * 4;
|
||||
|
||||
let mut offset = vec![0u8];
|
||||
offset.extend_from_slice(&location_table[table_entry as usize..table_entry as usize + 3]);
|
||||
let offset = u32::from_be_bytes(offset.try_into().unwrap()) as u64 * 4096;
|
||||
let size = location_table[table_entry as usize + 3] as usize * 4096;
|
||||
|
||||
if offset == 0 && size == 0 {
|
||||
return Err(WorldError::ChunkNotGenerated(
|
||||
ChunkNotGeneratedError::NotFound,
|
||||
));
|
||||
}
|
||||
|
||||
// Read the file using the offset and size
|
||||
let mut file_buf = {
|
||||
let seek_result = region_file.seek(std::io::SeekFrom::Start(offset));
|
||||
if seek_result.is_err() {
|
||||
return Err(WorldError::RegionIsInvalid);
|
||||
}
|
||||
let mut out = vec![0; size];
|
||||
let read_result = region_file.read_exact(&mut out);
|
||||
if read_result.is_err() {
|
||||
return Err(WorldError::RegionIsInvalid);
|
||||
}
|
||||
out
|
||||
};
|
||||
|
||||
// TODO: check checksum to make sure chunk is not corrupted
|
||||
let header = file_buf.drain(0..5).collect_vec();
|
||||
|
||||
let compression = match Compression::from_byte(header[4]) {
|
||||
Some(c) => c,
|
||||
None => {
|
||||
return Err(WorldError::Compression(
|
||||
CompressionError::UnknownCompression,
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
let size = u32::from_be_bytes(header[..4].try_into().unwrap());
|
||||
|
||||
// size includes the compression scheme byte, so we need to subtract 1
|
||||
let chunk_data = file_buf.drain(0..size as usize - 1).collect_vec();
|
||||
let decompressed_chunk =
|
||||
Self::decompress_data(compression, chunk_data).map_err(WorldError::Compression)?;
|
||||
|
||||
ChunkData::from_bytes(decompressed_chunk, at)
|
||||
}
|
||||
|
||||
fn decompress_data(
|
||||
compression: Compression,
|
||||
compressed_data: Vec<u8>,
|
||||
|
||||
@ -353,13 +353,10 @@ impl Server {
|
||||
let inst = std::time::Instant::now();
|
||||
let (sender, mut chunk_receiver) = mpsc::channel(distance as usize);
|
||||
let world = self.world.clone();
|
||||
|
||||
let chunks: Vec<_> = RadialIterator::new(distance).collect();
|
||||
tokio::spawn(async move {
|
||||
world
|
||||
.lock()
|
||||
.await
|
||||
.level
|
||||
.read_chunks(RadialIterator::new(distance).collect(), sender)
|
||||
.await;
|
||||
world.lock().await.level.fetch_chunks(&chunks, sender).await;
|
||||
});
|
||||
|
||||
player.client.send_packet(&CCenterChunk {
|
||||
@ -367,14 +364,15 @@ impl Server {
|
||||
chunk_z: 0.into(),
|
||||
});
|
||||
|
||||
while let Some((_chunk_pos, chunk_data)) = chunk_receiver.recv().await {
|
||||
while let Some(chunk_data) = chunk_receiver.recv().await {
|
||||
// dbg!(chunk_pos);
|
||||
let chunk_data = match chunk_data {
|
||||
Ok(d) => d,
|
||||
Err(_) => continue,
|
||||
};
|
||||
#[cfg(debug_assertions)]
|
||||
if _chunk_pos == (pumpkin_world::coordinates::ChunkCoordinates { x: 0, z: 0 }) {
|
||||
if chunk_data.position == (pumpkin_world::coordinates::ChunkCoordinates { x: 0, z: 0 })
|
||||
{
|
||||
use pumpkin_protocol::bytebuf::ByteBuffer;
|
||||
let mut test = ByteBuffer::empty();
|
||||
CChunkData(&chunk_data).write(&mut test);
|
||||
|
||||
Loading…
Reference in New Issue
Block a user