fix(simulation): address #843 review — seeding, atomics, API consistency
Review fixes from Hoshe + Tyre: - EntityRng seeding: splitmix64(seed) ^ splitmix64(id) instead of splitmix64(seed + id) — eliminates collision class where adjacent seeds produce identical streams - AtomicBool ordering: Relaxed → SeqCst for shutdown flag (correct on weakly-ordered architectures) - Worker Drop: join handles instead of detaching threads - Normalize stub API: remove ChunkGenWorker convenience wrappers, use .pool consistently across all 3 workers - trigger_monologue: downgrade &mut to shared refs (no-op anchor was blocking parallel systems) - Remove dead SimRng inserts from migrated monologue tests - Document determinism gap on poll_worker_results - Document bevy_tasks/rayon dep rationale in Cargo.toml Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -54,7 +54,7 @@ pub struct BackgroundWorkerPool<Req, Resp> {
|
||||
tx: Sender<WorkRequest<Req>>,
|
||||
rx: Receiver<(Resp, DeliveryStrategy)>,
|
||||
shutdown: Arc<AtomicBool>,
|
||||
_workers: Vec<JoinHandle<()>>,
|
||||
workers: Vec<JoinHandle<()>>,
|
||||
}
|
||||
|
||||
impl<Req, Resp> BackgroundWorkerPool<Req, Resp>
|
||||
@@ -84,7 +84,7 @@ where
|
||||
let handle = thread::Builder::new()
|
||||
.name(format!("worker-{id}"))
|
||||
.spawn(move || {
|
||||
while !shutdown.load(Ordering::Relaxed) {
|
||||
while !shutdown.load(Ordering::SeqCst) {
|
||||
match rx.recv_timeout(std::time::Duration::from_millis(100)) {
|
||||
Ok(work) => {
|
||||
let strategy = work.strategy;
|
||||
@@ -107,7 +107,7 @@ where
|
||||
tx: req_tx,
|
||||
rx: resp_rx,
|
||||
shutdown,
|
||||
_workers: workers,
|
||||
workers,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -146,13 +146,16 @@ where
|
||||
|
||||
/// Signal all workers to shut down gracefully.
|
||||
pub fn shutdown(&self) {
|
||||
self.shutdown.store(true, Ordering::Relaxed);
|
||||
self.shutdown.store(true, Ordering::SeqCst);
|
||||
}
|
||||
}
|
||||
|
||||
impl<Req, Resp> Drop for BackgroundWorkerPool<Req, Resp> {
|
||||
fn drop(&mut self) {
|
||||
self.shutdown.store(true, Ordering::Relaxed);
|
||||
// Workers will exit on next recv_timeout cycle (100ms max)
|
||||
self.shutdown.store(true, Ordering::SeqCst);
|
||||
// Join all worker threads — wait for clean exit (100ms max per thread)
|
||||
for handle in self.workers.drain(..) {
|
||||
let _ = handle.join();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
|
||||
use bevy_ecs::prelude::*;
|
||||
|
||||
use super::pool::{BackgroundWorkerPool, DeliveryStrategy, WorkRequest};
|
||||
use super::pool::BackgroundWorkerPool;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// ChunkGen — terrain, buildings, props, navmesh for a world chunk
|
||||
@@ -53,17 +53,6 @@ impl ChunkGenWorker {
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn submit(&self, req: ChunkGenRequest, strategy: DeliveryStrategy) {
|
||||
self.pool.submit(WorkRequest {
|
||||
payload: req,
|
||||
strategy,
|
||||
});
|
||||
}
|
||||
|
||||
pub fn drain(&self) -> Vec<(ChunkGenResult, DeliveryStrategy)> {
|
||||
self.pool.drain()
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -12,16 +12,25 @@ use super::stubs::{ChunkGenWorker, NpcPrepWorker, OffscreenTickWorker};
|
||||
|
||||
/// PreInput: poll all worker pools for completed results.
|
||||
///
|
||||
/// Incorporates results into the world state. Currently logs completions
|
||||
/// at debug level — real incorporation logic (spawning chunks, activating
|
||||
/// NPCs) plugs in at Phase 5+.
|
||||
/// Currently logs completions at debug level — real incorporation logic
|
||||
/// (spawning chunks, activating NPCs) plugs in at Phase 5+.
|
||||
///
|
||||
/// **DETERMINISM GAP (D-010):** Results are incorporated at the tick they
|
||||
/// happen to be ready, not at a deterministic tick. Two runs with the same
|
||||
/// seed may incorporate a chunk at tick 500 vs tick 502 depending on CPU
|
||||
/// load — diverging world state from that point.
|
||||
///
|
||||
/// This is acceptable while all handlers are no-ops (no simulation state
|
||||
/// is modified). Before any handler produces real state changes, implement
|
||||
/// a `WorkerResultLog` resource that records `(tick, worker_type, result)`.
|
||||
/// Replay reads from the log instead of polling channels. See plan #843.
|
||||
pub fn poll_worker_results(
|
||||
chunk_worker: Option<Res<ChunkGenWorker>>,
|
||||
npc_worker: Option<Res<NpcPrepWorker>>,
|
||||
offscreen_worker: Option<Res<OffscreenTickWorker>>,
|
||||
) {
|
||||
if let Some(worker) = chunk_worker {
|
||||
for (result, _strategy) in worker.drain() {
|
||||
for (result, _strategy) in worker.pool.drain() {
|
||||
tracing::debug!(
|
||||
chunk_x = result.chunk_x,
|
||||
chunk_y = result.chunk_y,
|
||||
|
||||
Reference in New Issue
Block a user