alex/beta-minecraft-rust
Minecraft Beta 1.7.3 Rewrite
BRANCH main
SNAPSHOT Draw terrain GPU-driven from packed per-section buffers with GPU occlusion culling, add a render distance slider (2-128) with a quad/disk shape, cut chunk bookkeeping at high render distances, and fix a light underflow at y=0
DOWNLOAD
root crates server src save writer.rs
use std::io::{self, Write};
use std::path::PathBuf;
use std::sync::mpsc::{self, Sender, SyncSender};
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use std::thread::JoinHandle;
use std::time::Duration;
use common::nbt::Compound;
use common::world::{Chunk, ChunkPos};
use flate2::Compression;
use flate2::write::ZlibEncoder;
use rustc_hash::FxHashMap;
use super::{LoadedChunk, Regions, SaveError, SessionLock, chunk, io_error, lock};
const MAX_PENDING: usize = 1024;
struct PendingWrite {
data: Arc<Vec<u8>>,
queued: bool,
}
#[derive(Default)]
struct Queue {
jobs: Mutex<FxHashMap<ChunkPos, PendingWrite>>,
drained: Condvar,
}
impl Queue {
fn lock(&self) -> MutexGuard<'_, FxHashMap<ChunkPos, PendingWrite>> {
lock(&self.jobs)
}
}
enum Command {
Write(ChunkPos),
Flush(SyncSender<Result<(), SaveError>>),
}
pub struct ChunkStorage {
dir: PathBuf,
session: SessionLock,
pub(super) regions: Arc<Mutex<Regions>>,
pending: Arc<Queue>,
error: Arc<Mutex<Option<String>>>,
writer: Option<(Sender<Command>, JoinHandle<()>)>,
}
impl ChunkStorage {
pub(super) fn with_session(dir: PathBuf, session: SessionLock) -> Result<Self, SaveError> {
let regions = Arc::new(Mutex::new(Regions {
dir: dir.join("region"),
..Regions::default()
}));
let pending = Arc::<Queue>::default();
let error = Arc::new(Mutex::new(None));
let (sender, receiver) = mpsc::channel();
let thread = std::thread::Builder::new()
.name("chunk writer".into())
.spawn({
let regions = regions.clone();
let pending = pending.clone();
let failure = error.clone();
let session = session.clone();
move || {
let write = |pos: ChunkPos| {
write_latest(®ions, &pending, pos, &session).map_err(|error| {
format!("cannot save chunk {},{}: {error}", pos.x, pos.z)
})
};
for command in receiver {
match command {
Command::Write(pos) => {
if let Err(error) = write(pos) {
tracing::error!("{error}");
*lock(&failure) = Some(error);
}
}
Command::Flush(reply) => {
let positions: Vec<_> = pending.lock().keys().copied().collect();
let mut result = Ok(());
for pos in positions {
if let Err(error) = write(pos) {
result = Err(error);
}
}
*lock(&failure) = result.clone().err();
let _ = reply.send(result.map_err(SaveError::Invalid));
}
}
}
}
})
.map_err(io_error(&dir))?;
Ok(Self {
dir,
session,
regions,
pending,
error,
writer: Some((sender, thread)),
})
}
pub fn check(&self) -> Result<(), SaveError> {
self.session.verify()?;
if let Some(error) = &*lock(&self.error) {
return Err(SaveError::Invalid(error.clone()));
}
if self
.writer
.as_ref()
.is_none_or(|(_, thread)| thread.is_finished())
{
return Err(SaveError::Invalid("chunk writer stopped".into()));
}
Ok(())
}
pub fn load(&self, pos: ChunkPos) -> Result<Option<LoadedChunk>, SaveError> {
let queued = self.pending.lock().get(&pos).map(|job| job.data.clone());
let data = match queued {
Some(data) => data,
None => {
let mut regions = lock(&self.regions);
let Some(region) = regions.region(pos, false).map_err(io_error(&self.dir))? else {
return Ok(None);
};
let Some(data) = region.read(pos).map_err(io_error(&self.dir))? else {
return Ok(None);
};
Arc::new(data)
}
};
let nbt = Compound::read(&mut &data[..]).map_err(|source| SaveError::Nbt {
path: self.dir.clone(),
source,
})?;
chunk::read(&nbt, pos).map(Some).ok_or_else(|| {
SaveError::Invalid(format!(
"invalid chunk {},{} in {}",
pos.x,
pos.z,
self.dir.display()
))
})
}
fn sender(&self) -> Result<&Sender<Command>, SaveError> {
self.writer
.as_ref()
.map(|(sender, _)| sender)
.ok_or_else(|| SaveError::Invalid("chunk writer stopped".into()))
}
pub fn save(&self, chunk: &Chunk, entities: &[Compound], time: i64) -> Result<(), SaveError> {
self.session.verify()?;
let sender = self.sender()?;
let mut data = Vec::with_capacity(96 * 1024);
chunk::write(chunk, entities, time)
.write(&mut data)
.map_err(io_error(&self.dir))?;
let mut pending = self.pending.lock();
// Like the original's synchronous writes, a full backlog waits for the writer
while pending.len() >= MAX_PENDING && !pending.contains_key(&chunk.pos) {
self.check()?;
pending = self
.pending
.drained
.wait_timeout(pending, Duration::from_millis(50))
.unwrap_or_else(|e| e.into_inner())
.0;
}
let enqueue = !pending.get(&chunk.pos).is_some_and(|job| job.queued);
pending.insert(
chunk.pos,
PendingWrite {
data: Arc::new(data),
queued: true,
},
);
if enqueue && sender.send(Command::Write(chunk.pos)).is_err() {
pending.get_mut(&chunk.pos).expect("inserted write").queued = false;
return Err(SaveError::Invalid("chunk writer disconnected".into()));
}
Ok(())
}
pub fn flush(&self) -> Result<(), SaveError> {
let (reply, result) = mpsc::sync_channel(1);
self.sender()?
.send(Command::Flush(reply))
.map_err(|_| SaveError::Invalid("chunk writer disconnected".into()))?;
result
.recv()
.map_err(|_| SaveError::Invalid("chunk writer stopped before flushing".into()))?
}
pub(super) fn take_growth(&self) -> u64 {
std::mem::take(&mut lock(&self.regions).growth)
}
}
fn write_latest(
regions: &Mutex<Regions>,
queue: &Queue,
pos: ChunkPos,
session: &SessionLock,
) -> io::Result<()> {
loop {
let Some(data) = queue.lock().get(&pos).map(|job| job.data.clone()) else {
return Ok(());
};
let result = (|| {
let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());
encoder.write_all(&data)?;
let compressed = encoder.finish()?;
let mut regions = lock(regions);
session.check()?;
let region = regions
.region(pos, true)?
.ok_or_else(|| io::Error::other("region was not created"))?;
regions.growth += region.write(pos, &compressed)?;
Ok(())
})();
let mut pending = queue.lock();
if result.is_err() {
if let Some(job) = pending.get_mut(&pos) {
job.queued = false;
}
return result;
}
if pending
.get(&pos)
.is_some_and(|job| Arc::ptr_eq(&job.data, &data))
{
pending.remove(&pos);
queue.drained.notify_all();
return Ok(());
}
}
}
impl Drop for ChunkStorage {
fn drop(&mut self) {
if let Err(error) = self.flush() {
tracing::error!("save shutdown failed: {error}");
}
if let Some((sender, thread)) = self.writer.take() {
drop(sender);
if thread.join().is_err() {
tracing::error!("chunk writer panicked");
}
}
}
}