steel_core/server/
world_tick_workers.rs

1use std::{io, sync::Arc, thread};
2
3use crossbeam::channel::{self, Sender};
4use thiserror::Error;
5use tokio::sync::oneshot;
6
7use crate::world::{World, WorldGameTickTimings};
8
9struct WorldTickRequest {
10    tick_count: u64,
11    runs_normally: bool,
12    response: oneshot::Sender<WorldGameTickTimings>,
13}
14
15struct WorldTickWorker {
16    world_key: Arc<str>,
17    requests: Option<Sender<WorldTickRequest>>,
18    thread: Option<thread::JoinHandle<()>>,
19}
20
21impl WorldTickWorker {
22    fn spawn(index: usize, world: Arc<World>) -> io::Result<Self> {
23        let world_key = Arc::<str>::from(world.key.to_string());
24        let (request_sender, request_receiver) = channel::bounded::<WorldTickRequest>(1);
25        let thread = thread::Builder::new()
26            .name(format!("world-tick-{index}"))
27            .spawn(move || {
28                while let Ok(request) = request_receiver.recv() {
29                    if request.runs_normally {
30                        world.chunk_map.tick_timed_tickets();
31                    }
32                    let timings = world.tick_game(request.tick_count, request.runs_normally);
33                    let _ = request.response.send(timings);
34                }
35            })?;
36
37        Ok(Self {
38            world_key,
39            requests: Some(request_sender),
40            thread: Some(thread),
41        })
42    }
43
44    fn start_tick(
45        &self,
46        tick_count: u64,
47        runs_normally: bool,
48    ) -> Result<oneshot::Receiver<WorldGameTickTimings>, WorldTickWorkerError> {
49        let (response, receiver) = oneshot::channel();
50        let Some(requests) = &self.requests else {
51            return Err(WorldTickWorkerError::Unavailable {
52                world: Arc::clone(&self.world_key),
53            });
54        };
55        requests
56            .send(WorldTickRequest {
57                tick_count,
58                runs_normally,
59                response,
60            })
61            .map_err(|_| WorldTickWorkerError::Unavailable {
62                world: Arc::clone(&self.world_key),
63            })?;
64        Ok(receiver)
65    }
66}
67
68impl Drop for WorldTickWorker {
69    fn drop(&mut self) {
70        drop(self.requests.take());
71        let Some(thread) = self.thread.take() else {
72            return;
73        };
74        if thread.join().is_err() {
75            log::error!(
76                "World tick worker for {} panicked during execution",
77                self.world_key
78            );
79        }
80    }
81}
82
83#[derive(Debug, Error)]
84pub(super) enum WorldTickWorkerError {
85    #[error("world tick worker for {world} is unavailable")]
86    Unavailable { world: Arc<str> },
87    #[error("world tick worker for {world} stopped without returning timings")]
88    MissingResponse { world: Arc<str> },
89}
90
91pub(super) struct WorldTickWorkers {
92    workers: Vec<WorldTickWorker>,
93}
94
95impl WorldTickWorkers {
96    pub(super) fn spawn<'a>(worlds: impl IntoIterator<Item = &'a Arc<World>>) -> io::Result<Self> {
97        let mut workers = Vec::new();
98        for (index, world) in worlds.into_iter().enumerate() {
99            workers.push(WorldTickWorker::spawn(index, Arc::clone(world))?);
100        }
101        Ok(Self { workers })
102    }
103
104    pub(super) async fn tick_all(
105        &self,
106        tick_count: u64,
107        runs_normally: bool,
108    ) -> Result<Vec<WorldGameTickTimings>, WorldTickWorkerError> {
109        let mut responses = Vec::with_capacity(self.workers.len());
110        for worker in &self.workers {
111            responses.push(worker.start_tick(tick_count, runs_normally)?);
112        }
113
114        let mut timings = Vec::with_capacity(responses.len());
115        for (worker, response) in self.workers.iter().zip(responses) {
116            timings.push(
117                response
118                    .await
119                    .map_err(|_| WorldTickWorkerError::MissingResponse {
120                        world: Arc::clone(&worker.world_key),
121                    })?,
122            );
123        }
124        Ok(timings)
125    }
126}
127
128#[cfg(test)]
129mod tests {
130    use futures::executor::block_on;
131
132    use super::WorldTickWorkers;
133    use crate::test_support::fresh_test_world;
134
135    #[test]
136    fn persistent_workers_tick_every_world_across_boundaries() {
137        let first = fresh_test_world("persistent_worker_first");
138        let second = fresh_test_world("persistent_worker_second");
139        let Ok(workers) = WorldTickWorkers::spawn([&first, &second]) else {
140            panic!("world tick workers should start");
141        };
142
143        let Ok(first_tick) = block_on(workers.tick_all(1, true)) else {
144            panic!("world tick workers should finish the first tick");
145        };
146        assert_eq!(first_tick.len(), 2);
147        assert_eq!(first.game_time(), 1);
148        assert_eq!(second.game_time(), 1);
149
150        let Ok(second_tick) = block_on(workers.tick_all(2, true)) else {
151            panic!("world tick workers should finish the second tick");
152        };
153        assert_eq!(second_tick.len(), 2);
154        assert_eq!(first.game_time(), 2);
155        assert_eq!(second.game_time(), 2);
156    }
157}
steel_core/server/world_tick_workers.rs

steel_core/server/
world_tick_workers.rs
