BlueGit

golden-age paperdoll-renderer

Files

paperdoll-renderer / src / io_queue.rs
Code

SSH

HTTPS

Username and a token as the password.

io_queue.rs

Parallel production IO + PBO readback; preview latest-wins (Tim t1063u/t1064u)
timfa Grok Bot · pushed by timfa · 4 days ago
2297f64d History
io_queue.rs 7.9 KiB Download Raw
1//! Ordered, bounded encode+write queue for production maps (Tim t1063u).
2//!
3//! The render worker resolves a view, then hands already-encoded PNG bytes to a
4//! single IO thread that publishes them with [`crate::resume::RunSession`] in
5//! deterministic journal order. A bounded channel provides backpressure so only
6//! a few views sit in flight between GPU/CPU and disk.
7use crate::resume::RunSession;
8use std::{
9 collections::BTreeMap,
10 sync::{
11 atomic::{AtomicBool, Ordering},
12 mpsc, Arc,
13 },
14 thread::{self, JoinHandle},
15};
16
17/// Max view-bundles waiting on the IO thread (backpressure).
18pub const IO_QUEUE_VIEWS: usize = 4;
19
20/// One planned PNG: relative path + encoded file bytes.
21#[derive(Debug)]
22pub struct EncodedMap {
23 pub rel: String,
24 pub bytes: Vec<u8>,
25}
26
27/// All maps of one view, in [`crate::production::SAVED_ORDER`].
28#[derive(Debug)]
29pub struct ViewPublish {
30 pub maps: Vec<EncodedMap>,
31}
32
33enum Msg {
34 View(ViewPublish),
35 /// Stop after draining; reply with the session and digests.
36 Finish(mpsc::SyncSender<IoFinish>),
37}
38
39/// Result of shutting down the IO thread.
40pub struct IoFinish {
41 pub session: RunSession,
42 pub png_sha256: BTreeMap<String, String>,
43 pub error: Option<String>,
44}
45
46/// Handle for the ordered publisher.
47pub struct IoQueue {
48 tx: Option<mpsc::SyncSender<Msg>>,
49 /// Shared cancel; IO stops accepting new publishes when set (still finishes Finish).
50 cancel: Arc<AtomicBool>,
51 handle: Option<JoinHandle<()>>,
52 finished: bool,
53}
54
55impl IoQueue {
56 /// Move `session` onto a dedicated IO thread. `seed` are digests already
57 /// verified (resume skip); they are merged into the final map.
58 pub fn start(
59 session: RunSession,
60 seed: BTreeMap<String, String>,
61 cancel: Arc<AtomicBool>,
62 ) -> Self {
63 let (tx, rx) = mpsc::sync_channel::<Msg>(IO_QUEUE_VIEWS);
64 let handle = thread::spawn(move || {
65 let mut session = session;
66 let mut png_sha256 = seed;
67 let mut error: Option<String> = None;
68 for msg in rx {
69 match msg {
70 Msg::View(view) => {
71 // Once queued, always publish — cancel only blocks new
72 // submit()s so Resume keeps already-finished maps.
73 if error.is_some() {
74 continue;
75 }
76 match publish_view(&mut session, &mut png_sha256, view) {
77 Ok(()) => {}
78 Err(e) => error = Some(e),
79 }
80 }
81 Msg::Finish(reply) => {
82 let _ = reply.send(IoFinish {
83 session,
84 png_sha256,
85 error,
86 });
87 return;
88 }
89 }
90 }
91 });
92 Self {
93 tx: Some(tx),
94 cancel,
95 handle: Some(handle),
96 finished: false,
97 }
98 }
99
100 /// Enqueue one view. Blocks when the queue is full (backpressure).
101 /// Returns Err if the IO thread is gone or cancel was set before send.
102 pub fn submit(&self, view: ViewPublish) -> Result<(), String> {
103 if self.cancel.load(Ordering::Relaxed) {
104 return Err("geannuleerd".into());
105 }
106 self.tx
107 .as_ref()
108 .ok_or_else(|| "IO-queue gestopt".to_string())?
109 .send(Msg::View(view))
110 .map_err(|_| "IO-queue gestopt".to_string())
111 }
112
113 /// Drain the queue, stop the thread, and return session + digests.
114 pub fn finish(mut self) -> Result<(RunSession, BTreeMap<String, String>), String> {
115 let finished = self.shutdown()?;
116 self.finished = true;
117 if let Some(e) = finished.error {
118 return Err(e);
119 }
120 Ok((finished.session, finished.png_sha256))
121 }
122
123 fn shutdown(&mut self) -> Result<IoFinish, String> {
124 let tx = self
125 .tx
126 .take()
127 .ok_or_else(|| "IO-queue al gestopt".to_string())?;
128 let (reply_tx, reply_rx) = mpsc::sync_channel(1);
129 tx.send(Msg::Finish(reply_tx))
130 .map_err(|_| "IO-queue gestopt".to_string())?;
131 drop(tx);
132 let finished = reply_rx
133 .recv()
134 .map_err(|_| "IO-queue gestopt zonder antwoord".to_string())?;
135 if let Some(h) = self.handle.take() {
136 let _ = h.join();
137 }
138 Ok(finished)
139 }
140}
141
142impl Drop for IoQueue {
143 fn drop(&mut self) {
144 if self.finished || self.tx.is_none() {
145 return;
146 }
147 let _ = self.shutdown();
148 }
149}
150
151fn publish_view(
152 session: &mut RunSession,
153 png_sha256: &mut BTreeMap<String, String>,
154 view: ViewPublish,
155) -> Result<(), String> {
156 for map in view.maps {
157 let digest = session.publish(&map.rel, &map.bytes)?;
158 png_sha256.insert(map.rel, digest);
159 }
160 Ok(())
161}
162
163#[cfg(test)]
164mod tests {
165 use super::*;
166 use crate::resume::RunSession;
167 use std::path::PathBuf;
168
169 struct Temp(PathBuf);
170 impl Temp {
171 fn new() -> Self {
172 let p = std::env::temp_dir().join(format!(
173 "ppd-io-queue-{}-{}",
174 std::process::id(),
175 std::time::SystemTime::now()
176 .duration_since(std::time::UNIX_EPOCH)
177 .unwrap()
178 .as_nanos()
179 ));
180 std::fs::create_dir_all(&p).unwrap();
181 Self(p)
182 }
183 }
184 impl Drop for Temp {
185 fn drop(&mut self) {
186 let _ = std::fs::remove_dir_all(&self.0);
187 }
188 }
189
190 #[test]
191 fn publishes_in_submit_order_and_respects_backpressure_capacity() {
192 let tmp = Temp::new();
193 let session = RunSession::fresh(&tmp.0).unwrap();
194 let cancel = Arc::new(AtomicBool::new(false));
195 let q = IoQueue::start(session, BTreeMap::new(), cancel);
196 for i in 0..6u32 {
197 q.submit(ViewPublish {
198 maps: vec![EncodedMap {
199 rel: format!("raw files/m/f{i}.png"),
200 bytes: format!("png-{i}").into_bytes(),
201 }],
202 })
203 .unwrap();
204 }
205 let (session, digests) = q.finish().unwrap();
206 drop(session);
207 let mut keys: Vec<_> = digests.keys().cloned().collect();
208 keys.sort();
209 assert_eq!(
210 keys,
211 (0..6)
212 .map(|i| format!("raw files/m/f{i}.png"))
213 .collect::<Vec<_>>()
214 );
215 // Journal order matches submit order.
216 let journal = std::fs::read_to_string(tmp.0.join(crate::resume::JOURNAL)).unwrap();
217 let mut seen = Vec::new();
218 for line in journal.lines() {
219 if line.trim().is_empty() {
220 continue;
221 }
222 let v: serde_json::Value = serde_json::from_str(line).unwrap();
223 if let Some(p) = v.get("path").and_then(|x| x.as_str()) {
224 seen.push(p.to_string());
225 }
226 }
227 assert_eq!(
228 seen,
229 (0..6)
230 .map(|i| format!("raw files/m/f{i}.png"))
231 .collect::<Vec<_>>()
232 );
233 }
234
235 #[test]
236 fn cancel_before_submit_returns_error() {
237 let tmp = Temp::new();
238 let session = RunSession::fresh(&tmp.0).unwrap();
239 let cancel = Arc::new(AtomicBool::new(true));
240 let q = IoQueue::start(session, BTreeMap::new(), cancel);
241 let err = q
242 .submit(ViewPublish {
243 maps: vec![EncodedMap {
244 rel: "a.png".into(),
245 bytes: b"x".to_vec(),
246 }],
247 })
248 .unwrap_err();
249 assert!(err.contains("geannuleerd"));
250 // finish still returns (with cancel error if anything was queued — here none)
251 let _ = q.finish();
252 }
253}