1use crate::{
8 DecodeCapability, Decoder, ImageDescriptor, PixelsError, Producer, Region, Result, TileBuf,
9 TileMut, copy_region,
10};
11use std::sync::{Arc, Mutex};
12
13#[derive(Debug, Clone)]
18pub struct BufferSource {
19 buffer: Arc<TileBuf>,
20 descriptor: ImageDescriptor,
21}
22
23impl BufferSource {
24 pub fn new(descriptor: ImageDescriptor, buffer: Arc<TileBuf>) -> Result<Self> {
31 if buffer.region() != descriptor.region() {
32 return Err(PixelsError::invalid_argument(
33 "buffer",
34 format!(
35 "buffer covers {} but the image is {}",
36 buffer.region(),
37 descriptor.region()
38 ),
39 ));
40 }
41 if buffer.pixel() != descriptor.pixel {
42 return Err(PixelsError::invalid_argument(
43 "buffer",
44 format!(
45 "buffer is {} but the image is {}",
46 buffer.pixel(),
47 descriptor.pixel
48 ),
49 ));
50 }
51 Ok(Self { buffer, descriptor })
52 }
53
54 #[must_use]
56 pub fn buffer(&self) -> &Arc<TileBuf> {
57 &self.buffer
58 }
59}
60
61impl Producer for BufferSource {
62 fn name(&self) -> &'static str {
63 "buffer"
64 }
65
66 fn descriptor(&self) -> ImageDescriptor {
67 self.descriptor
68 }
69
70 fn capability(&self) -> DecodeCapability {
71 DecodeCapability::Regions
75 }
76
77 fn produce(&self, region: Region, output: &mut TileMut<'_>) -> Result<()> {
78 let tile = self.buffer.as_tile()?;
79 copy_region(&tile, output, region)
80 }
81}
82
83#[derive(Debug)]
85enum DecodeState {
86 Reading {
88 decoder: Box<dyn Decoder>,
90 cursor: u32,
92 },
93 Failed(String),
96}
97
98#[derive(Debug)]
125pub struct DecodedSource {
126 descriptor: Mutex<ImageDescriptor>,
130 capability: DecodeCapability,
131 state: Mutex<DecodeState>,
132 window: Mutex<Option<(Region, Arc<TileBuf>)>>,
134}
135
136impl DecodedSource {
137 #[must_use]
141 pub fn new(decoder: Box<dyn Decoder>) -> Self {
142 Self {
143 descriptor: Mutex::new(decoder.descriptor()),
144 capability: decoder.capability(),
145 state: Mutex::new(DecodeState::Reading { decoder, cursor: 0 }),
146 window: Mutex::new(None),
147 }
148 }
149
150 fn shape(&self) -> ImageDescriptor {
156 *self
157 .descriptor
158 .lock()
159 .unwrap_or_else(std::sync::PoisonError::into_inner)
160 }
161
162 fn decode_band(&self, band: Region) -> Result<Arc<TileBuf>> {
177 let mut state = self
178 .state
179 .lock()
180 .map_err(|_| PixelsError::graph("decoder state was poisoned by a panicking thread"))?;
181
182 if let DecodeState::Failed(detail) = &*state {
183 return Err(PixelsError::malformed("stream", detail.clone()));
184 }
185 let DecodeState::Reading { decoder, cursor } = &mut *state else {
186 return Err(PixelsError::graph("decoder state changed unexpectedly"));
187 };
188
189 let carry = if band.y < *cursor {
192 let held = self
193 .window
194 .lock()
195 .ok()
196 .and_then(|window| window.clone())
197 .filter(|(covered, _)| covered.y <= band.y);
198 match held {
199 Some((_, buffer)) => Some(buffer),
200 None => {
201 return Err(PixelsError::graph(format!(
202 "source cannot rewind: row {} was requested but the stream is at row \
203 {cursor} and the retained band no longer covers it; this pipeline \
204 needed a materialization point",
205 band.y
206 )));
207 }
208 }
209 } else {
210 None
211 };
212
213 let outcome = Self::fill_band(
214 decoder.as_mut(),
215 cursor,
216 band,
217 self.shape(),
218 carry.as_deref(),
219 );
220 match outcome {
221 Ok(buffer) => Ok(Arc::new(buffer)),
222 Err(error) => {
223 *state = DecodeState::Failed(error.to_string());
224 Err(error)
225 }
226 }
227 }
228
229 fn fill_band(
234 decoder: &mut dyn Decoder,
235 cursor: &mut u32,
236 band: Region,
237 descriptor: ImageDescriptor,
238 carry: Option<&TileBuf>,
239 ) -> Result<TileBuf> {
240 let mut scratch = vec![0_u8; descriptor.row_bytes()];
241 while *cursor < band.y {
242 decoder.read_row(&mut scratch)?;
243 *cursor += 1;
244 }
245 let mut buffer = TileBuf::zeroed(band, descriptor.pixel)?;
246 {
247 let mut tile = buffer.as_tile_mut()?;
248 let carried = carry.map(TileBuf::as_tile).transpose()?;
249 for y in band.y..band.y.saturating_add(band.height) {
250 let row = tile.row_mut(y).ok_or_else(|| {
251 PixelsError::malformed("stream", format!("row {y} is outside the band"))
252 })?;
253 if y < *cursor {
254 let source =
257 carried
258 .as_ref()
259 .and_then(|tile| tile.row(y))
260 .ok_or_else(|| {
261 PixelsError::graph(format!(
262 "row {y} is behind the stream and outside the retained band"
263 ))
264 })?;
265 let take = row.len().min(source.len());
266 row.get_mut(..take)
267 .zip(source.get(..take))
268 .map(|(target, source)| target.copy_from_slice(source))
269 .ok_or_else(|| {
270 PixelsError::graph(format!("row {y} could not be carried over"))
271 })?;
272 continue;
273 }
274 decoder.read_row(row)?;
275 *cursor += 1;
276 }
277 }
278 Ok(buffer)
279 }
280
281 fn produce_region(&self, region: Region, output: &mut TileMut<'_>) -> Result<()> {
292 let mut state = self
293 .state
294 .lock()
295 .map_err(|_| PixelsError::graph("decoder state was poisoned by a panicking thread"))?;
296
297 if let DecodeState::Failed(detail) = &*state {
298 return Err(PixelsError::malformed("stream", detail.clone()));
299 }
300 let DecodeState::Reading { decoder, .. } = &mut *state else {
301 return Err(PixelsError::graph("decoder state changed unexpectedly"));
302 };
303
304 match decoder.read_region(region, output) {
305 Ok(()) => Ok(()),
306 Err(error) => {
307 *state = DecodeState::Failed(error.to_string());
311 Err(error)
312 }
313 }
314 }
315
316 fn band_for(&self, region: Region) -> Result<Arc<TileBuf>> {
318 {
319 let window = self
320 .window
321 .lock()
322 .map_err(|_| PixelsError::graph("source window was poisoned"))?;
323 if let Some((covered, buffer)) = &*window {
324 if covered.contains(region) {
325 return Ok(Arc::clone(buffer));
326 }
327 }
328 }
329 let band = Region::new(0, region.y, self.shape().width, region.height);
332 let buffer = self.decode_band(band)?;
333 if let Ok(mut window) = self.window.lock() {
334 *window = Some((band, Arc::clone(&buffer)));
335 }
336 Ok(buffer)
337 }
338}
339
340impl Producer for DecodedSource {
341 fn name(&self) -> &'static str {
342 "decoded"
343 }
344
345 fn descriptor(&self) -> ImageDescriptor {
346 self.shape()
347 }
348
349 fn capability(&self) -> DecodeCapability {
350 self.capability
351 }
352
353 fn reduced_descriptor(&self, target: (u32, u32)) -> Option<ImageDescriptor> {
354 let state = self.state.lock().ok()?;
355 let DecodeState::Reading { decoder, cursor } = &*state else {
356 return None;
357 };
358 if *cursor > 0 {
361 return None;
362 }
363 decoder.reduced_descriptor(target)
364 }
365
366 fn reduce_to(&self, descriptor: ImageDescriptor) -> Result<()> {
367 let mut state = self
368 .state
369 .lock()
370 .map_err(|_| PixelsError::graph("decoder state was poisoned by a panicking thread"))?;
371 let DecodeState::Reading { decoder, cursor } = &mut *state else {
372 return Err(PixelsError::graph("decoder state changed unexpectedly"));
373 };
374 if *cursor > 0 {
375 return Err(PixelsError::invalid_argument(
376 "descriptor",
377 format!("{cursor} rows have already been produced; the resolution is settled"),
378 ));
379 }
380 decoder.reduce_to(descriptor)?;
381
382 let mut shape = self
383 .descriptor
384 .lock()
385 .map_err(|_| PixelsError::graph("descriptor was poisoned by a panicking thread"))?;
386 *shape = descriptor;
387 if let Ok(mut window) = self.window.lock() {
390 *window = None;
391 }
392 Ok(())
393 }
394
395 fn produce(&self, region: Region, output: &mut TileMut<'_>) -> Result<()> {
396 if self.capability == DecodeCapability::Regions {
399 return self.produce_region(region, output);
400 }
401 let buffer = self.band_for(region)?;
402 let tile = buffer.as_tile()?;
403 copy_region(&tile, output, region)
404 }
405}
406
407#[cfg(test)]
408#[allow(
409 clippy::unwrap_used,
410 clippy::indexing_slicing,
411 reason = "tests operate on known-good values and assert shapes directly"
412)]
413mod tests {
414 use super::*;
415 use crate::{ErrorCode, PixelFormat};
416
417 #[derive(Debug)]
420 struct StubDecoder {
421 descriptor: ImageDescriptor,
422 row: u32,
423 fail_at: Option<u32>,
424 rows_read: Arc<std::sync::atomic::AtomicU32>,
425 }
426
427 impl Decoder for StubDecoder {
428 fn descriptor(&self) -> ImageDescriptor {
429 self.descriptor
430 }
431 fn read_row(&mut self, out: &mut [u8]) -> Result<()> {
432 if Some(self.row) == self.fail_at {
433 return Err(PixelsError::malformed("stub", "corrupt row"));
434 }
435 self.rows_read
436 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
437 out.fill(self.row as u8);
438 self.row += 1;
439 Ok(())
440 }
441 }
442
443 fn sized_stub(width: u32, height: u32) -> (DecodedSource, Arc<std::sync::atomic::AtomicU32>) {
445 let rows_read = Arc::new(std::sync::atomic::AtomicU32::new(0));
446 let decoder = StubDecoder {
447 descriptor: ImageDescriptor::new(width, height, PixelFormat::Gray8).unwrap(),
448 row: 0,
449 fail_at: None,
450 rows_read: Arc::clone(&rows_read),
451 };
452 (DecodedSource::new(Box::new(decoder)), rows_read)
453 }
454
455 fn rows_of(source: &DecodedSource, region: Region) -> Vec<u8> {
457 let mut buffer = TileBuf::zeroed(region, PixelFormat::Gray8).unwrap();
458 {
459 let mut tile = buffer.as_tile_mut().unwrap();
460 source.produce(region, &mut tile).unwrap();
461 }
462 let tile = buffer.as_tile().unwrap();
463 (region.y..region.y + region.height)
464 .map(|y| tile.row(y).unwrap()[0])
465 .collect()
466 }
467
468 #[test]
473 fn overlapping_bands_carry_the_rows_already_consumed() {
474 let (source, rows_read) = sized_stub(2, 16);
475
476 assert_eq!(
477 rows_of(&source, Region::new(0, 0, 2, 6)),
478 [0, 1, 2, 3, 4, 5]
479 );
480 assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 6);
481
482 assert_eq!(
484 rows_of(&source, Region::new(0, 4, 2, 6)),
485 [4, 5, 6, 7, 8, 9],
486 "the carried rows are wrong"
487 );
488 assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 10);
491 }
492
493 #[test]
495 fn a_request_behind_the_retained_band_is_still_an_error() {
496 let (source, _) = sized_stub(2, 16);
497 assert_eq!(rows_of(&source, Region::new(0, 8, 2, 4)), [8, 9, 10, 11]);
498
499 let region = Region::new(0, 2, 2, 4);
500 let mut buffer = TileBuf::zeroed(region, PixelFormat::Gray8).unwrap();
501 let mut tile = buffer.as_tile_mut().unwrap();
502 let error = source.produce(region, &mut tile).unwrap_err();
503 assert_eq!(error.code(), ErrorCode::Graph, "{error}");
504 }
505
506 fn stub(fail_at: Option<u32>) -> (DecodedSource, Arc<std::sync::atomic::AtomicU32>) {
507 let rows_read = Arc::new(std::sync::atomic::AtomicU32::new(0));
508 let decoder = StubDecoder {
509 descriptor: ImageDescriptor::new(2, 3, PixelFormat::Gray8).unwrap(),
510 row: 0,
511 fail_at,
512 rows_read: Arc::clone(&rows_read),
513 };
514 (DecodedSource::new(Box::new(decoder)), rows_read)
515 }
516
517 fn tall_stub() -> (DecodedSource, Arc<std::sync::atomic::AtomicU32>) {
519 let rows_read = Arc::new(std::sync::atomic::AtomicU32::new(0));
520 let decoder = StubDecoder {
521 descriptor: ImageDescriptor::new(2, 64, PixelFormat::Gray8).unwrap(),
522 row: 0,
523 fail_at: None,
524 rows_read: Arc::clone(&rows_read),
525 };
526 (DecodedSource::new(Box::new(decoder)), rows_read)
527 }
528
529 #[test]
530 fn construction_reads_no_pixel_rows() {
531 let (source, rows_read) = stub(None);
532 assert_eq!(source.descriptor().width, 2);
533 assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 0);
534 }
535
536 #[test]
537 fn the_stream_is_decoded_once_and_reused() {
538 let (source, rows_read) = stub(None);
539 let mut out = TileBuf::zeroed(Region::from_size(2, 3), PixelFormat::Gray8).unwrap();
540 for _ in 0..3 {
541 let mut tile = out.as_tile_mut().unwrap();
542 source.produce(Region::from_size(2, 3), &mut tile).unwrap();
543 }
544 assert_eq!(
545 rows_read.load(std::sync::atomic::Ordering::Relaxed),
546 3,
547 "decoded exactly once"
548 );
549 assert_eq!(out.bytes(), &[0, 0, 1, 1, 2, 2]);
550 }
551
552 #[test]
553 fn a_partial_region_pull_returns_just_that_region() {
554 let (source, _) = stub(None);
555 let mut out = TileBuf::zeroed(Region::new(0, 1, 2, 1), PixelFormat::Gray8).unwrap();
556 let mut tile = out.as_tile_mut().unwrap();
557 source.produce(Region::new(0, 1, 2, 1), &mut tile).unwrap();
558 assert_eq!(out.bytes(), &[1, 1]);
559 }
560
561 #[test]
562 fn a_decode_failure_is_sticky_and_never_partial() {
563 let (source, _) = stub(Some(1));
564 let mut out = TileBuf::zeroed(Region::from_size(2, 3), PixelFormat::Gray8).unwrap();
565 for attempt in 0..2 {
566 let mut tile = out.as_tile_mut().unwrap();
567 let err = source
568 .produce(Region::from_size(2, 3), &mut tile)
569 .unwrap_err();
570 assert_eq!(err.code(), ErrorCode::Malformed, "attempt {attempt}");
571 }
572 }
573
574 #[test]
575 fn decoding_advances_only_as_far_as_demanded() {
576 let (source, rows_read) = tall_stub();
579 let mut out = TileBuf::zeroed(Region::new(0, 0, 2, 4), PixelFormat::Gray8).unwrap();
580 let mut tile = out.as_tile_mut().unwrap();
581 source.produce(Region::new(0, 0, 2, 4), &mut tile).unwrap();
582 assert_eq!(
583 rows_read.load(std::sync::atomic::Ordering::Relaxed),
584 4,
585 "decoded past the requested band"
586 );
587 }
588
589 #[test]
590 fn successive_bands_stream_forward() {
591 let (source, rows_read) = tall_stub();
592 for start in (0..64).step_by(4) {
593 let band = Region::new(0, start, 2, 4);
594 let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
595 let mut tile = out.as_tile_mut().unwrap();
596 source.produce(band, &mut tile).unwrap();
597 assert_eq!(out.bytes()[0], start as u8, "band at row {start} is wrong");
599 }
600 assert_eq!(
601 rows_read.load(std::sync::atomic::Ordering::Relaxed),
602 64,
603 "rows re-read"
604 );
605 }
606
607 #[test]
608 fn repeated_demand_for_a_band_is_served_from_the_window() {
609 let (source, rows_read) = tall_stub();
612 let band = Region::new(0, 0, 2, 4);
613 for _ in 0..5 {
614 let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
615 let mut tile = out.as_tile_mut().unwrap();
616 source.produce(band, &mut tile).unwrap();
617 }
618 assert_eq!(
619 rows_read.load(std::sync::atomic::Ordering::Relaxed),
620 4,
621 "band was re-decoded"
622 );
623 }
624
625 #[test]
626 fn a_sub_band_of_the_window_is_served_without_re_reading() {
627 let (source, rows_read) = tall_stub();
628 let band = Region::new(0, 0, 2, 8);
629 let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
630 source
631 .produce(band, &mut out.as_tile_mut().unwrap())
632 .unwrap();
633
634 let inner = Region::new(0, 2, 2, 2);
635 let mut small = TileBuf::zeroed(inner, PixelFormat::Gray8).unwrap();
636 source
637 .produce(inner, &mut small.as_tile_mut().unwrap())
638 .unwrap();
639 assert_eq!(small.bytes(), &[2, 2, 3, 3]);
640 assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 8);
641 }
642
643 #[test]
644 fn rewinding_is_a_reported_error_not_silent_corruption() {
645 let (source, _) = tall_stub();
649 let later = Region::new(0, 16, 2, 4);
650 let mut out = TileBuf::zeroed(later, PixelFormat::Gray8).unwrap();
651 source
652 .produce(later, &mut out.as_tile_mut().unwrap())
653 .unwrap();
654
655 let earlier = Region::new(0, 0, 2, 4);
656 let mut back = TileBuf::zeroed(earlier, PixelFormat::Gray8).unwrap();
657 let err = source
658 .produce(earlier, &mut back.as_tile_mut().unwrap())
659 .unwrap_err();
660 assert_eq!(err.code(), ErrorCode::Graph);
661 assert!(err.to_string().contains("rewind"), "{err}");
662 assert!(err.to_string().contains("materialization"), "{err}");
663 }
664
665 #[test]
666 fn skipped_rows_are_consumed_not_lost() {
667 let (source, rows_read) = tall_stub();
670 let band = Region::new(0, 32, 2, 4);
671 let mut out = TileBuf::zeroed(band, PixelFormat::Gray8).unwrap();
672 source
673 .produce(band, &mut out.as_tile_mut().unwrap())
674 .unwrap();
675 assert_eq!(rows_read.load(std::sync::atomic::Ordering::Relaxed), 36);
676 assert_eq!(out.bytes()[0], 32, "landed on the wrong row");
677 }
678
679 #[test]
680 fn producers_report_their_capability() {
681 let (source, _) = tall_stub();
683 assert_eq!(source.capability(), DecodeCapability::Sequential);
684
685 let descriptor = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
686 let buffer = Arc::new(TileBuf::for_image(&descriptor).unwrap());
687 let buffered = BufferSource::new(descriptor, buffer).unwrap();
688 assert_eq!(buffered.capability(), DecodeCapability::Regions);
689 }
690
691 #[test]
692 fn buffer_source_validates_against_its_descriptor() {
693 let desc = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
694 let good = Arc::new(TileBuf::zeroed(desc.region(), PixelFormat::Gray8).unwrap());
695 assert!(BufferSource::new(desc, good).is_ok());
696
697 let wrong_size =
698 Arc::new(TileBuf::zeroed(Region::from_size(3, 3), PixelFormat::Gray8).unwrap());
699 assert_eq!(
700 BufferSource::new(desc, wrong_size).unwrap_err().code(),
701 ErrorCode::InvalidArgument
702 );
703
704 let wrong_format = Arc::new(TileBuf::zeroed(desc.region(), PixelFormat::Rgb8).unwrap());
705 assert_eq!(
706 BufferSource::new(desc, wrong_format).unwrap_err().code(),
707 ErrorCode::InvalidArgument
708 );
709 }
710
711 #[test]
712 fn buffer_source_serves_regions() {
713 let desc = ImageDescriptor::new(2, 2, PixelFormat::Gray8).unwrap();
714 let buffer = Arc::new(
715 TileBuf::from_vec(desc.region(), PixelFormat::Gray8, vec![1, 2, 3, 4]).unwrap(),
716 );
717 let source = BufferSource::new(desc, buffer).unwrap();
718 let mut out = TileBuf::zeroed(Region::new(1, 0, 1, 2), PixelFormat::Gray8).unwrap();
719 let mut tile = out.as_tile_mut().unwrap();
720 source.produce(Region::new(1, 0, 1, 2), &mut tile).unwrap();
721 assert_eq!(out.bytes(), &[2, 4]);
722 }
723}