feat(server): add dirty-chunk write-back on unload
This commit is contained in:
parent
b89b73b277
commit
c54eeb2216
|
|
@ -1,6 +1,6 @@
|
||||||
// SPDX-License-Identifier: AGPL-3.0-only
|
// SPDX-License-Identifier: AGPL-3.0-only
|
||||||
|
|
||||||
//! A dedicated thread that owns every open region file and answers load requests over a channel.
|
//! A dedicated thread that owns every open region file and services load, write-back, remove, and flush requests over a channel.
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
use std::collections::hash_map::Entry;
|
use std::collections::hash_map::Entry;
|
||||||
|
|
@ -10,10 +10,11 @@ use std::thread::{self, JoinHandle};
|
||||||
use crossbeam_channel::{Receiver, Sender};
|
use crossbeam_channel::{Receiver, Sender};
|
||||||
use shared::save::SaveError;
|
use shared::save::SaveError;
|
||||||
use shared::world::{ChunkData, ChunkPos};
|
use shared::world::{ChunkData, ChunkPos};
|
||||||
|
use tracing::warn;
|
||||||
|
|
||||||
use super::region_file::{RegionFile, region_coords, region_path};
|
use super::region_file::{RegionFile, region_coords, region_path};
|
||||||
|
|
||||||
/// A request sent to the save actor. Each variant carries its own one-shot reply channel.
|
/// A request sent to the save actor. Read and flush requests carry a one-shot reply channel; write and remove requests are fire-and-forget, mutating only the in-memory region image until a flush.
|
||||||
pub enum SaveRequest {
|
pub enum SaveRequest {
|
||||||
/// Reads the stored chunk at a position, replying with the saved modification if one exists.
|
/// Reads the stored chunk at a position, replying with the saved modification if one exists.
|
||||||
Read {
|
Read {
|
||||||
|
|
@ -22,6 +23,23 @@ pub enum SaveRequest {
|
||||||
/// The one-shot channel the actor replies on: `Ok(Some(data))` for a saved modification, `Ok(None)` when the chunk was never modified, or `Err` on a save-layer failure.
|
/// The one-shot channel the actor replies on: `Ok(Some(data))` for a saved modification, `Ok(None)` when the chunk was never modified, or `Err` on a save-layer failure.
|
||||||
reply: Sender<Result<Option<ChunkData>, SaveError>>,
|
reply: Sender<Result<Option<ChunkData>, SaveError>>,
|
||||||
},
|
},
|
||||||
|
/// Writes a modified chunk's diff into its region, replacing any prior record. Mutates only the in-memory image; durability waits for a `Flush`.
|
||||||
|
Write {
|
||||||
|
/// The chunk position the diff is stored under.
|
||||||
|
pos: ChunkPos,
|
||||||
|
/// The baseline-relative diff to persist.
|
||||||
|
data: ChunkData,
|
||||||
|
},
|
||||||
|
/// Drops any stored record for a position, reclaiming its space into the region free list. Used when a clean chunk is unloaded.
|
||||||
|
Remove {
|
||||||
|
/// The chunk position whose record is dropped.
|
||||||
|
pos: ChunkPos,
|
||||||
|
},
|
||||||
|
/// Flushes every dirty region to disk, replying once all are written.
|
||||||
|
Flush {
|
||||||
|
/// The one-shot channel the actor replies on: `Ok(())` when every dirty region flushed, or the first `Err` encountered.
|
||||||
|
reply: Sender<Result<(), SaveError>>,
|
||||||
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A handle to the running save actor: the request sender plus the owning thread's join handle.
|
/// A handle to the running save actor: the request sender plus the owning thread's join handle.
|
||||||
|
|
@ -61,6 +79,63 @@ fn actor_loop(region_dir: &Path, request_rx: &Receiver<SaveRequest>) {
|
||||||
// A send error means the requesting worker has gone away; the reply is simply dropped.
|
// A send error means the requesting worker has gone away; the reply is simply dropped.
|
||||||
let _ = reply.send(result);
|
let _ = reply.send(result);
|
||||||
}
|
}
|
||||||
|
SaveRequest::Write { pos, data } => {
|
||||||
|
// `last_modified` is record metadata only, not a worldgen input, so a zero placeholder is acceptable until a real timestamp source is wired.
|
||||||
|
let last_modified = 0;
|
||||||
|
match region_mut(&mut regions, region_dir, pos) {
|
||||||
|
Ok(region) => {
|
||||||
|
// A failed encode must not go unnoticed; the write is otherwise silently lost.
|
||||||
|
if let Err(error) = region.write_chunk(pos, &data, last_modified) {
|
||||||
|
warn!(?error, ?pos, "chunk write-back failed; edit dropped");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(error) => warn!(?error, ?pos, "region open failed; write-back dropped"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
SaveRequest::Remove { pos } => match region_mut(&mut regions, region_dir, pos) {
|
||||||
|
Ok(region) => region.remove_chunk(pos),
|
||||||
|
Err(error) => warn!(?error, ?pos, "region open failed; record not removed"),
|
||||||
|
},
|
||||||
|
SaveRequest::Flush { reply } => {
|
||||||
|
let result = flush_dirty(&mut regions);
|
||||||
|
// A send error means the requester has gone away; the reply is simply dropped.
|
||||||
|
let _ = reply.send(result);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Flushes every dirty region to disk, returning the first error while still attempting the rest.
|
||||||
|
fn flush_dirty(regions: &mut HashMap<(i32, i32), RegionFile>) -> Result<(), SaveError> {
|
||||||
|
let mut result = Ok(());
|
||||||
|
for region in regions.values_mut() {
|
||||||
|
// Clean regions are skipped so a flush never rewrites an unchanged file.
|
||||||
|
if !region.is_dirty() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if let Err(error) = region.save() {
|
||||||
|
warn!(?error, "region flush failed");
|
||||||
|
// The first failure is reported; later regions are still flushed.
|
||||||
|
if result.is_ok() {
|
||||||
|
result = Err(error);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
result
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the region file covering `pos`, opening and caching it on first access.
|
||||||
|
fn region_mut<'a>(
|
||||||
|
regions: &'a mut HashMap<(i32, i32), RegionFile>,
|
||||||
|
region_dir: &Path,
|
||||||
|
pos: ChunkPos,
|
||||||
|
) -> Result<&'a mut RegionFile, SaveError> {
|
||||||
|
let key = region_coords(pos.x, pos.z);
|
||||||
|
// The region file is opened once on first touch; every later access hits the in-memory copy.
|
||||||
|
match regions.entry(key) {
|
||||||
|
Entry::Occupied(entry) => Ok(entry.into_mut()),
|
||||||
|
Entry::Vacant(entry) => {
|
||||||
|
Ok(entry.insert(RegionFile::open(region_path(region_dir, pos.x, pos.z))?))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -71,13 +146,5 @@ fn read_chunk(
|
||||||
region_dir: &Path,
|
region_dir: &Path,
|
||||||
pos: ChunkPos,
|
pos: ChunkPos,
|
||||||
) -> Result<Option<ChunkData>, SaveError> {
|
) -> Result<Option<ChunkData>, SaveError> {
|
||||||
let key = region_coords(pos.x, pos.z);
|
region_mut(regions, region_dir, pos)?.read_chunk(pos)
|
||||||
// The region file is opened once on first touch; every later read of a chunk in it hits the in-memory copy.
|
|
||||||
let region = match regions.entry(key) {
|
|
||||||
Entry::Occupied(entry) => entry.into_mut(),
|
|
||||||
Entry::Vacant(entry) => {
|
|
||||||
entry.insert(RegionFile::open(region_path(region_dir, pos.x, pos.z))?)
|
|
||||||
}
|
|
||||||
};
|
|
||||||
region.read_chunk(pos)
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -50,9 +50,16 @@ pub struct ServerWorld {
|
||||||
#[expect(dead_code)]
|
#[expect(dead_code)]
|
||||||
save_actor: SaveActor,
|
save_actor: SaveActor,
|
||||||
/// Handles to the generation worker threads, retained so they can be joined on shutdown.
|
/// Handles to the generation worker threads, retained so they can be joined on shutdown.
|
||||||
// Retained ahead of a dedicated shutdown path; not yet read because the server has no graceful-stop sequence.
|
// * NOTE: Retained ahead of a dedicated shutdown path
|
||||||
|
// TODO: Remove once the server has a graceful-stop sequence
|
||||||
#[expect(dead_code)]
|
#[expect(dead_code)]
|
||||||
workers: Vec<JoinHandle<()>>,
|
workers: Vec<JoinHandle<()>>,
|
||||||
|
/// The same read-only generator handle the workers share, held so eviction can regenerate a chunk's baseline to diff against.
|
||||||
|
generator: Arc<VoxelGenerator>,
|
||||||
|
/// A handle onto the shared baseline cache, used to resolve the baseline during the unload diff.
|
||||||
|
cache: ChunkCache,
|
||||||
|
/// The save actor's request sender, used to issue `Write` and `Remove` on unload.
|
||||||
|
save_tx: Sender<SaveRequest>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ServerWorld {
|
impl ServerWorld {
|
||||||
|
|
@ -101,6 +108,8 @@ impl ServerWorld {
|
||||||
drop(job_rx);
|
drop(job_rx);
|
||||||
drop(result_tx);
|
drop(result_tx);
|
||||||
|
|
||||||
|
let save_tx = save_actor.sender();
|
||||||
|
|
||||||
Self {
|
Self {
|
||||||
chunks: HashMap::new(),
|
chunks: HashMap::new(),
|
||||||
job_tx,
|
job_tx,
|
||||||
|
|
@ -108,6 +117,9 @@ impl ServerWorld {
|
||||||
in_flight: HashSet::new(),
|
in_flight: HashSet::new(),
|
||||||
save_actor,
|
save_actor,
|
||||||
workers,
|
workers,
|
||||||
|
generator,
|
||||||
|
cache,
|
||||||
|
save_tx,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -156,7 +168,20 @@ impl ServerWorld {
|
||||||
.copied()
|
.copied()
|
||||||
.collect();
|
.collect();
|
||||||
for pos in &stale {
|
for pos in &stale {
|
||||||
self.chunks.remove(pos);
|
// The chunk is taken by value so it can be diffed against its baseline before being dropped.
|
||||||
|
let Some(chunk) = self.chunks.remove(pos) else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
let baseline = self.cache.get_or_generate(*pos, &self.generator);
|
||||||
|
// * NOTE: The diff is stamped with worldgen version 0: a single version exists today. This must become the chunk's stored version once worldgen versioning lands.
|
||||||
|
let data = ChunkData::from_diff(*pos, 0, &baseline, &chunk);
|
||||||
|
// A clean chunk drops any prior record into the region free list; a dirty chunk writes its diff back. Both only mutate the actor's in-memory image until a flush. A send error means the actor is gone, which the tick thread cannot act on.
|
||||||
|
let request = if data.is_unmodified() {
|
||||||
|
SaveRequest::Remove { pos: *pos }
|
||||||
|
} else {
|
||||||
|
SaveRequest::Write { pos: *pos, data }
|
||||||
|
};
|
||||||
|
let _ = self.save_tx.send(request);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Dispatch: request a load for every wanted position that is neither resident nor already in flight.
|
// Dispatch: request a load for every wanted position that is neither resident nor already in flight.
|
||||||
|
|
@ -242,14 +267,14 @@ pub fn cylinder_chunks<S: std::hash::BuildHasher>(
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::{ServerWorld, StreamStats, cylinder_chunks};
|
use super::*;
|
||||||
use shared::generator::{VoxelGenerator, WorldGenConfig};
|
use shared::generator::{VoxelGenerator, WorldGenConfig};
|
||||||
use shared::save::SaveError;
|
use shared::save::SaveError;
|
||||||
use shared::world::{BlockId, ChunkData, ChunkPos};
|
use shared::world::{BlockId, ChunkData, ChunkPos};
|
||||||
use std::collections::HashSet;
|
use std::collections::HashSet;
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use crate::save::{RegionFile, region_path};
|
use crate::save::{RegionFile, SaveRequest, region_path};
|
||||||
|
|
||||||
/// Builds a generator with a small, cheap terrain configuration for streaming tests.
|
/// Builds a generator with a small, cheap terrain configuration for streaming tests.
|
||||||
fn test_generator() -> VoxelGenerator {
|
fn test_generator() -> VoxelGenerator {
|
||||||
|
|
@ -285,6 +310,20 @@ mod tests {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Issues a flush against the world's save actor and blocks until every dirty region is written. Because write-backs and this flush travel the same sender to the single actor thread, the reply confirms the preceding writes are durable.
|
||||||
|
fn flush(world: &ServerWorld) -> Result<(), SaveError> {
|
||||||
|
let (reply_tx, reply_rx) = crossbeam_channel::bounded(1);
|
||||||
|
// A send error means the actor has already stopped, leaving nothing to flush.
|
||||||
|
if world
|
||||||
|
.save_tx
|
||||||
|
.send(SaveRequest::Flush { reply: reply_tx })
|
||||||
|
.is_err()
|
||||||
|
{
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
reply_rx.recv().unwrap_or(Ok(()))
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn reconcile_converges_over_multiple_passes() -> Result<(), SaveError> {
|
fn reconcile_converges_over_multiple_passes() -> Result<(), SaveError> {
|
||||||
// A fresh empty directory means every load is a miss and resolves to the baseline.
|
// A fresh empty directory means every load is a miss and resolves to the baseline.
|
||||||
|
|
@ -354,6 +393,65 @@ mod tests {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn dirty_chunk_is_written_back_on_eviction() -> Result<(), SaveError> {
|
||||||
|
// A resident chunk edited away from its baseline must survive an evict -> flush -> reload round-trip.
|
||||||
|
let dir = tempfile::tempdir()?;
|
||||||
|
let pos = ChunkPos::new(0, 0, 0);
|
||||||
|
let edited_index = 100usize;
|
||||||
|
let edited_block = BlockId(999);
|
||||||
|
|
||||||
|
let mut world = test_world(dir.path().to_path_buf());
|
||||||
|
let mut desired = HashSet::new();
|
||||||
|
desired.insert(pos);
|
||||||
|
drain_to_idle(&mut world, &desired);
|
||||||
|
|
||||||
|
// Mutate the resident chunk so it diverges from the baseline the eviction diff regenerates.
|
||||||
|
assert!(
|
||||||
|
world
|
||||||
|
.chunks
|
||||||
|
.get_mut(&pos)
|
||||||
|
.map(|chunk| chunk.blocks[edited_index] = edited_block)
|
||||||
|
.is_some()
|
||||||
|
);
|
||||||
|
|
||||||
|
// Reconciling against an empty desired set evicts the chunk, sending its diff to the actor.
|
||||||
|
world.reconcile(&HashSet::new());
|
||||||
|
// The flush shares the eviction's sender, so its reply confirms the write-back is on disk.
|
||||||
|
flush(&world)?;
|
||||||
|
|
||||||
|
// A fresh world over the same directory must stream the chunk back with the edit intact.
|
||||||
|
let mut reloaded = test_world(dir.path().to_path_buf());
|
||||||
|
drain_to_idle(&mut reloaded, &desired);
|
||||||
|
assert!(
|
||||||
|
reloaded
|
||||||
|
.chunk(pos)
|
||||||
|
.is_some_and(|chunk| chunk.blocks[edited_index] == edited_block)
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn clean_chunk_is_not_written_back_on_eviction() -> Result<(), SaveError> {
|
||||||
|
// An unmodified chunk equals its baseline, so eviction must persist no record for it.
|
||||||
|
let dir = tempfile::tempdir()?;
|
||||||
|
let pos = ChunkPos::new(0, 0, 0);
|
||||||
|
|
||||||
|
let mut world = test_world(dir.path().to_path_buf());
|
||||||
|
let mut desired = HashSet::new();
|
||||||
|
desired.insert(pos);
|
||||||
|
drain_to_idle(&mut world, &desired);
|
||||||
|
|
||||||
|
// Evict without modifying the chunk, then flush.
|
||||||
|
world.reconcile(&HashSet::new());
|
||||||
|
flush(&world)?;
|
||||||
|
|
||||||
|
// No record may exist for a chunk that never diverged from its baseline.
|
||||||
|
let region = RegionFile::open(region_path(dir.path(), pos.x, pos.z))?;
|
||||||
|
assert!(region.read_chunk(pos)?.is_none());
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn cylinder_contains_expected_columns() {
|
fn cylinder_contains_expected_columns() {
|
||||||
let mut set = HashSet::new();
|
let mut set = HashSet::new();
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue