1use crate::compression_config::EncodedGeometryType;
2#[cfg(feature = "point_cloud_decode")]
3use crate::decoder_buffer::DecoderBuffer;
4#[cfg(feature = "point_cloud_decode")]
5use crate::draco_types::DataType;
6#[cfg(feature = "point_cloud_decode")]
7use crate::geometry_attribute::{GeometryAttributeType, PointAttribute};
8#[cfg(feature = "point_cloud_decode")]
9use crate::kd_tree_attributes_decoder::KdTreeAttributesDecoder;
10#[cfg(feature = "point_cloud_decode")]
11use crate::parallel;
12#[cfg(feature = "point_cloud_decode")]
13use crate::point_cloud::PointCloud;
14#[cfg(feature = "point_cloud_decode")]
15use crate::prediction_scheme::EntryToPointIdMap;
16#[cfg(feature = "point_cloud_decode")]
17use crate::sequential_integer_attribute_decoder::{
18 PortableExtent, SequentialIntegerAttributeDecoder,
19};
20#[cfg(feature = "point_cloud_decode")]
21use crate::status::{DracoError, Status};
22#[cfg(feature = "point_cloud_decode")]
23use crate::symbol_encoding::{decode_raw_symbol_pair, decode_raw_symbol_pair_in};
24
25#[cfg(feature = "point_cloud_decode")]
26use crate::attribute_octahedron_transform::AttributeOctahedronTransform;
27#[cfg(feature = "point_cloud_decode")]
28use crate::attribute_quantization_transform::AttributeQuantizationTransform;
29#[cfg(feature = "point_cloud_decode")]
30use crate::attribute_transform::AttributeTransform;
31#[cfg(feature = "point_cloud_decode")]
32use crate::sequential_generic_attribute_decoder::SequentialGenericAttributeDecoder;
33#[cfg(feature = "point_cloud_decode")]
34use crate::sequential_normal_attribute_decoder::SequentialNormalAttributeDecoder;
35#[cfg(feature = "point_cloud_decode")]
36use crate::sequential_quantization_attribute_decoder::SequentialQuantizationAttributeDecoder;
37#[cfg(feature = "point_cloud_decode")]
38use crate::version::{version_at_least, VERSION_FLAGS_INTRODUCED};
39
40#[cfg(feature = "legacy_bitstream_decode")]
58pub(crate) fn carries_transform_byte(method_byte: u8) -> bool {
59 method_byte != 0xFF && method_byte != 0xFE
60}
61
62pub struct PointCloudDecoder {
71 geometry_type: EncodedGeometryType,
72 #[cfg(feature = "point_cloud_decode")]
73 method: u8,
74 #[cfg(feature = "point_cloud_decode")]
75 flags: u16,
76 version_major: u8,
80 version_minor: u8,
81 #[cfg(feature = "point_cloud_decode")]
82 threads: i32,
83}
84
85#[cfg(all(feature = "point_cloud_decode", not(fuzzing)))]
92const PARALLEL_MIN_STREAM_BYTES: usize = 1 << 20;
93#[cfg(all(feature = "point_cloud_decode", fuzzing))]
94const PARALLEL_MIN_STREAM_BYTES: usize = 64;
95
96#[cfg(feature = "point_cloud_decode")]
100fn coded_components(decoder_type: u8, attribute: &PointAttribute) -> usize {
101 match decoder_type {
102 3 => 2,
103 _ => attribute.num_components() as usize,
104 }
105}
106
107#[cfg(all(test, feature = "point_cloud_decode"))]
110thread_local! {
111 static ENGAGED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
112 static PAIRED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
114 static IN_ORDER: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
117 static PAIRS_OFF: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
120}
121
122#[cfg(feature = "point_cloud_decode")]
125struct PendingQuant {
126 att_id: i32,
127 portable: PointAttribute,
128 transform: AttributeQuantizationTransform,
129}
130
131#[cfg(feature = "point_cloud_decode")]
134struct PendingNormal {
135 att_id: i32,
136 portable: PointAttribute,
137 quantization_bits: u8,
138}
139
140#[cfg(feature = "point_cloud_decode")]
146fn dequantize_in_parallel(
147 pc: &mut PointCloud,
148 mut pending: Vec<PendingQuant>,
149 threads: usize,
150) -> Status {
151 pending.sort_by_key(|q| q.att_id);
152 let mut work: Vec<(PendingQuant, PointAttribute, Option<DracoError>)> =
153 Vec::with_capacity(pending.len());
154 for q in pending {
155 let dst = std::mem::take(pc.try_attribute_mut(q.att_id)?);
156 work.push((q, dst, None));
157 }
158 parallel::for_each_chunk_mut(&mut work, 1, threads, |_, chunk| {
159 let (q, dst, error) = &mut chunk[0];
160 if let Err(e) = q.transform.inverse_transform_attribute(&q.portable, dst) {
161 *error = Some(DracoError::general(format!(
162 "Failed to dequantize attribute: {e}"
163 )));
164 }
165 });
166 let mut first_error = None;
167 for (q, dst, error) in work {
168 *pc.try_attribute_mut(q.att_id)? = dst;
169 first_error = first_error.or(error);
170 }
171 first_error.map_or(Ok(()), Err)
172}
173
174#[cfg(feature = "point_cloud_decode")]
176fn undo_octahedra_in_parallel(
177 pc: &mut PointCloud,
178 mut pending: Vec<PendingNormal>,
179 bitstream_version: u16,
180 threads: usize,
181) -> Status {
182 pending.sort_by_key(|n| n.att_id);
183 let mut work: Vec<(PendingNormal, PointAttribute, Option<DracoError>)> =
184 Vec::with_capacity(pending.len());
185 for n in pending {
186 let dst = std::mem::take(pc.try_attribute_mut(n.att_id)?);
187 work.push((n, dst, None));
188 }
189 parallel::for_each_chunk_mut(&mut work, 1, threads, |_, chunk| {
190 let (n, dst, error) = &mut chunk[0];
191 let mut oct = AttributeOctahedronTransform::new(-1);
192 let done = oct
193 .set_parameters(n.quantization_bits as i32)
194 .and_then(|()| {
195 oct.inverse_transform_attribute_with_legacy_octahedron(
196 &n.portable,
197 dst,
198 bitstream_version < 0x0200,
199 )
200 });
201 if let Err(e) = done {
202 *error = Some(DracoError::general(format!(
203 "Failed to decode normals: {e}"
204 )));
205 }
206 });
207 let mut first_error = None;
208 for (n, dst, error) in work {
209 *pc.try_attribute_mut(n.att_id)? = dst;
210 first_error = first_error.or(error);
211 }
212 first_error.map_or(Ok(()), Err)
213}
214
215impl Default for PointCloudDecoder {
216 fn default() -> Self {
217 Self::new()
218 }
219}
220
221#[cfg(feature = "point_cloud_decode")]
222fn validate_num_attributes_in_decoder(
223 num_attributes_in_decoder: usize,
224 remaining_bytes: usize,
225) -> Result<(), DracoError> {
226 const MIN_ATTRIBUTE_BYTES: usize = 6;
230 if num_attributes_in_decoder == 0
231 || num_attributes_in_decoder > remaining_bytes / MIN_ATTRIBUTE_BYTES
232 {
233 return Err(DracoError::general(
234 "Invalid number of attributes".to_string(),
235 ));
236 }
237 Ok(())
238}
239
240#[cfg(feature = "point_cloud_decode")]
241fn validate_num_components(num_components: u8) -> Result<(), DracoError> {
242 if num_components == 0 {
243 return Err(DracoError::general(
244 "Invalid attribute component count".to_string(),
245 ));
246 }
247 Ok(())
248}
249
250impl PointCloudDecoder {
251 pub fn new() -> Self {
253 Self {
254 geometry_type: EncodedGeometryType::PointCloud,
255 #[cfg(feature = "point_cloud_decode")]
256 method: 0,
257 #[cfg(feature = "point_cloud_decode")]
258 flags: 0,
259 version_major: 0,
260 version_minor: 0,
261 #[cfg(feature = "point_cloud_decode")]
262 threads: 1,
263 }
264 }
265
266 #[cfg(feature = "point_cloud_decode")]
287 pub fn set_threads(&mut self, threads: i32) {
288 self.threads = threads;
289 }
290
291 pub(crate) fn bitstream_version(&self) -> u16 {
297 crate::version::bitstream_version(self.version_major, self.version_minor)
298 }
299
300 pub(crate) fn set_bitstream_version(&mut self, major: u8, minor: u8) {
307 self.version_major = major;
308 self.version_minor = minor;
309 }
310
311 #[cfg(feature = "point_cloud_decode")]
312 pub fn decode(&mut self, in_buffer: &mut DecoderBuffer, out_pc: &mut PointCloud) -> Status {
324 out_pc.clear();
325
326 self.decode_header(in_buffer)?;
328
329 if version_at_least(
330 self.version_major,
331 self.version_minor,
332 VERSION_FLAGS_INTRODUCED,
333 ) && (self.flags & crate::metadata::METADATA_FLAG_MASK) != 0
334 {
335 let metadata = crate::metadata::GeometryMetadata::decode(in_buffer)
336 .map_err(|_| DracoError::general("Failed to decode metadata".to_string()))?;
337 out_pc.set_metadata(Some(metadata));
338 }
339
340 self.decode_geometry_data(in_buffer, out_pc)
342 }
343
344 #[cfg(feature = "point_cloud_decode")]
347 pub fn decode_after_header(
348 &mut self,
349 version_major: u8,
350 version_minor: u8,
351 method: u8,
352 buffer: &mut DecoderBuffer,
353 out_pc: &mut PointCloud,
354 ) -> Status {
355 self.version_major = version_major;
356 self.version_minor = version_minor;
357 self.method = method;
358 self.flags = 0;
359 self.geometry_type = EncodedGeometryType::PointCloud;
360 self.decode_geometry_data(buffer, out_pc)
361 }
362
363 #[cfg(feature = "point_cloud_decode")]
364 fn decode_header(&mut self, buffer: &mut DecoderBuffer) -> Status {
365 let mut magic = [0u8; 5];
366 buffer.decode_bytes(&mut magic)?;
367 if &magic != b"DRACO" {
368 return Err(DracoError::general("Invalid magic".to_string()));
369 }
370
371 self.version_major = buffer.decode_u8()?;
372 self.version_minor = buffer.decode_u8()?;
373 buffer.set_version(self.version_major, self.version_minor);
374
375 let g_type = buffer.decode_u8()?;
376 self.geometry_type = match g_type {
377 0 => EncodedGeometryType::PointCloud,
378 1 => EncodedGeometryType::TriangularMesh,
379 _ => return Err(DracoError::general("Invalid geometry type".to_string())),
380 };
381 if self.geometry_type != EncodedGeometryType::PointCloud {
382 return Err(DracoError::general(
383 "PointCloudDecoder cannot decode mesh bitstreams".to_string(),
384 ));
385 }
386
387 self.method = buffer.decode_u8()?;
388
389 self.flags = buffer
391 .decode_u16()
392 .map_err(|_| DracoError::general("Failed to decode flags".to_string()))?;
393
394 Ok(())
395 }
396
397 #[cfg(feature = "point_cloud_decode")]
398 fn decode_geometry_data(&mut self, buffer: &mut DecoderBuffer, pc: &mut PointCloud) -> Status {
399 let bitstream_version: u16 =
400 crate::version::bitstream_version(self.version_major, self.version_minor);
401 let declared_points = buffer.decode_u32()? as i32;
419 if declared_points < 0 {
420 return Err(DracoError::general(format!(
421 "Point cloud declares {declared_points} points"
422 )));
423 }
424 let num_points: usize = declared_points as usize;
425 buffer.check_points(num_points)?;
426 pc.set_num_points(num_points);
427
428 let num_attributes_decoders = buffer.decode_u8()? as usize;
429
430 if self.method == 1 {
431 for _ in 0..num_attributes_decoders {
433 let mut att_decoder = KdTreeAttributesDecoder::new(0);
434 att_decoder
435 .decode_attributes_decoder_data(pc, buffer)
436 .map_err(|err| err.context("Failed to decode attribute metadata"))?;
437 att_decoder
438 .decode_attributes(pc, buffer)
439 .map_err(|err| err.context("Failed to decode attributes"))?;
440 }
441 } else {
442 struct AttributeSpec {
444 att_type: GeometryAttributeType,
445 data_type: DataType,
446 num_components: u8,
447 normalized: bool,
448 unique_id: u32,
449 }
450
451 for _ in 0..num_attributes_decoders {
452 let num_attributes_in_decoder: usize = if bitstream_version < 0x0200 {
453 buffer.decode_u32()? as usize
454 } else {
455 buffer.decode_varint()? as usize
456 };
457 if num_attributes_in_decoder == 0 {
458 return Err(DracoError::general(
459 "Invalid number of attributes".to_string(),
460 ));
461 }
462 validate_num_attributes_in_decoder(
463 num_attributes_in_decoder,
464 buffer.remaining_size(),
465 )?;
466
467 let mut attribute_specs: Vec<AttributeSpec> =
468 Vec::with_capacity(num_attributes_in_decoder);
469 let mut att_ids: Vec<i32> = Vec::with_capacity(num_attributes_in_decoder);
470 let mut decoder_types: Vec<u8> = Vec::with_capacity(num_attributes_in_decoder);
471 let mut pending_quant: Vec<PendingQuant> = Vec::new();
472 let mut pending_normals: Vec<PendingNormal> = Vec::new();
473
474 for _ in 0..num_attributes_in_decoder {
475 let att_type_val = buffer.decode_u8()?;
476 let att_type = GeometryAttributeType::try_from(att_type_val)?;
477
478 let data_type_val = buffer.decode_u8()?;
479 let data_type = DataType::try_from(data_type_val)?;
480
481 let num_components = buffer.decode_u8()?;
482 validate_num_components(num_components)?;
483 let normalized = buffer.decode_u8()? != 0;
484 let unique_id: u32 = if bitstream_version < 0x0103 {
485 buffer.decode_u16()? as u32
486 } else {
487 buffer.decode_varint()? as u32
488 };
489
490 attribute_specs.push(AttributeSpec {
491 att_type,
492 data_type,
493 num_components,
494 normalized,
495 unique_id,
496 });
497 }
498
499 for _ in 0..num_attributes_in_decoder {
500 decoder_types.push(buffer.decode_u8()?);
501 }
502
503 for (local_i, spec) in attribute_specs.iter().enumerate() {
504 if decoder_types[local_i] == 0 {
505 let entry_size =
506 spec.num_components as usize * spec.data_type.byte_length();
507 let bytes_needed = entry_size.checked_mul(num_points).ok_or_else(|| {
508 DracoError::general(
509 "Raw point cloud attribute byte count overflow".to_string(),
510 )
511 })?;
512 if buffer.remaining_size() < bytes_needed {
513 return Err(DracoError::general(
514 "Not enough data for raw point cloud attribute values".to_string(),
515 ));
516 }
517 }
518
519 buffer.admit_attribute(
520 num_points,
521 spec.num_components as usize,
522 spec.data_type.byte_length(),
523 )?;
524 let mut att = PointAttribute::new();
525 att.init_deferred(
536 spec.att_type,
537 spec.num_components,
538 spec.data_type,
539 spec.normalized,
540 num_points,
541 )?;
542 att.set_unique_id(spec.unique_id);
543 let att_id = pc.add_attribute_preserve_unique_id(att);
544 att_ids.push(att_id);
545 }
546
547 let point_ids = if decoder_types.iter().any(|&decoder_type| decoder_type != 0) {
553 Some(EntryToPointIdMap::identity(num_points))
554 } else {
555 None
556 };
557
558 let threads = parallel::resolve(self.threads);
567 #[cfg(test)]
568 let in_order = IN_ORDER.with(|in_order| in_order.get());
569 #[cfg(not(test))]
570 let in_order = false;
571 let in_parallel = bitstream_version >= 0x0200
572 && threads > 1
573 && !in_order
574 && num_points.saturating_mul(num_attributes_in_decoder)
575 >= crate::parallel::ATTRIBUTES_MIN_VALUES
576 && buffer.remaining_size() >= PARALLEL_MIN_STREAM_BYTES
577 && decoder_types
578 .iter()
579 .filter(|&&decoder_type| (1..=3).contains(&decoder_type))
580 .count()
581 >= 2;
582 let decoded_in_parallel = in_parallel
583 && self.decode_attributes_in_parallel(
584 pc,
585 buffer,
586 &att_ids,
587 &decoder_types,
588 point_ids,
589 num_points,
590 threads,
591 &mut pending_quant,
592 &mut pending_normals,
593 )?;
594 if !decoded_in_parallel {
595 let mut ahead: Option<Vec<u32>> = None;
598 for (local_i, &att_id) in att_ids.iter().enumerate() {
599 let decoder_type = decoder_types[local_i];
600 let mut predecoded = ahead.take();
601 if predecoded.is_none() && bitstream_version >= 0x0200 {
602 if let (Some(&next_id), Some(&next_type)) =
603 (att_ids.get(local_i + 1), decoder_types.get(local_i + 1))
604 {
605 if let Some((first, second)) = Self::pair_symbols(
606 pc,
607 buffer,
608 (att_id, decoder_type),
609 (next_id, next_type),
610 num_points,
611 ) {
612 predecoded = Some(first);
613 ahead = Some(second);
614 }
615 }
616 }
617 self.decode_attribute_in_place(
618 pc,
619 buffer,
620 att_id,
621 decoder_type,
622 point_ids,
623 num_points,
624 &mut pending_quant,
625 &mut pending_normals,
626 predecoded,
627 )?;
628 }
629 }
630
631 for (local_i, &att_id) in att_ids.iter().enumerate() {
632 match decoder_types[local_i] {
633 2 if bitstream_version >= 0x0200 => {
634 let idx = pending_quant
635 .iter()
636 .position(|p| p.att_id == att_id)
637 .ok_or_else(|| {
638 DracoError::general(
639 "Missing pending quantized attribute transform".to_string(),
640 )
641 })?;
642 let original = pc.try_attribute(att_id)?;
643 pending_quant[idx]
644 .transform
645 .decode_parameters(original, buffer)
646 .map_err(|e| {
647 DracoError::general(format!(
648 "Failed to decode quantization parameters: {e}"
649 ))
650 })?;
651 }
652 3 if bitstream_version >= 0x0200 => {
653 let idx = pending_normals
654 .iter()
655 .position(|p| p.att_id == att_id)
656 .ok_or_else(|| {
657 DracoError::general(
658 "Missing pending normal attribute transform".to_string(),
659 )
660 })?;
661 let quantization_bits = buffer.decode_u8()?;
662 if !AttributeOctahedronTransform::is_valid_quantization_bits(
663 quantization_bits as i32,
664 ) {
665 return Err(DracoError::general(
666 "Invalid normal quantization bits".to_string(),
667 ));
668 }
669 pending_normals[idx].quantization_bits = quantization_bits;
670 }
671 _ => {}
672 }
673 }
674
675 if in_parallel {
676 dequantize_in_parallel(pc, pending_quant, threads)?;
677 undo_octahedra_in_parallel(pc, pending_normals, bitstream_version, threads)?;
678 } else {
679 for q in pending_quant {
680 let dst = pc.try_attribute_mut(q.att_id)?;
681 q.transform
682 .inverse_transform_attribute(&q.portable, dst)
683 .map_err(|e| {
684 DracoError::general(format!("Failed to dequantize attribute: {e}"))
685 })?;
686 }
687 for n in pending_normals {
688 let mut oct = AttributeOctahedronTransform::new(-1);
689 oct.set_parameters(n.quantization_bits as i32)?;
690 let dst = pc.try_attribute_mut(n.att_id)?;
691 oct.inverse_transform_attribute_with_legacy_octahedron(
692 &n.portable,
693 dst,
694 bitstream_version < 0x0200,
695 )
696 .map_err(|e| {
697 DracoError::general(format!("Failed to decode normals: {e}"))
698 })?;
699 }
700 }
701 }
702 }
703
704 Ok(())
705 }
706
707 pub fn get_geometry_type(&self) -> EncodedGeometryType {
709 self.geometry_type
710 }
711}
712
713#[cfg(feature = "point_cloud_decode")]
714impl PointCloudDecoder {
715 fn pair_symbols(
732 pc: &PointCloud,
733 buffer: &mut DecoderBuffer,
734 first: (i32, u8),
735 second: (i32, u8),
736 num_points: usize,
737 ) -> Option<(Vec<u32>, Vec<u32>)> {
738 #[cfg(test)]
739 if PAIRS_OFF.with(|off| off.get()) {
740 return None;
741 }
742 let coded = |(att_id, decoder_type): (i32, u8)| -> Option<(usize, usize)> {
743 if !(1..=3).contains(&decoder_type) {
744 return None;
745 }
746 let components = coded_components(decoder_type, pc.try_attribute(att_id).ok()?);
747 Some((num_points.checked_mul(components)?, components))
748 };
749 let pairs = |at: &DecoderBuffer| -> bool {
754 let Ok(mut look) = at.fork_at(at.position()) else {
755 return false;
756 };
757 matches!(
758 (look.decode_u8(), look.decode_u8()),
759 (Ok(1), Ok(bits)) if (1..=11).contains(&bits)
760 )
761 };
762 let (first, second) = (coded(first)?, coded(second)?);
763 let start = buffer.position();
764 let budget = buffer.budget();
765 let mut attempt = || -> Option<(Vec<u32>, Vec<u32>)> {
766 if !SequentialIntegerAttributeDecoder::seek_symbols(buffer) || !pairs(buffer) {
767 return None;
768 }
769 let first_symbols = buffer.position();
770 buffer.set_position(start).ok()?;
771 if !SequentialIntegerAttributeDecoder::skip_values(first.0, first.1, buffer).ok()? {
772 return None;
773 }
774 let mut probe = buffer.fork_at(buffer.position()).ok()?;
775 if !SequentialIntegerAttributeDecoder::seek_symbols(&mut probe) || !pairs(&probe) {
776 return None;
777 }
778 let second_symbols = probe.position();
779 buffer.set_position(first_symbols).ok()?;
780 decode_raw_symbol_pair_in(buffer, first, (second_symbols, second.0, second.1))
781 };
782 let symbols = attempt();
783 let _ = buffer.set_position(start);
785 if symbols.is_none() {
786 buffer.restore_budget(budget);
787 }
788 #[cfg(test)]
789 if symbols.is_some() {
790 PAIRED.with(|paired| paired.set(paired.get() + 2));
791 }
792 symbols
793 }
794
795 #[allow(clippy::too_many_arguments)]
798 fn decode_attribute_in_place(
799 &self,
800 pc: &mut PointCloud,
801 buffer: &mut DecoderBuffer,
802 att_id: i32,
803 decoder_type: u8,
804 point_ids: Option<EntryToPointIdMap<'_>>,
805 num_points: usize,
806 pending_quant: &mut Vec<PendingQuant>,
807 pending_normals: &mut Vec<PendingNormal>,
808 symbols: Option<Vec<u32>>,
809 ) -> Status {
810 let bitstream_version = self.bitstream_version();
811 match decoder_type {
812 1 => {
813 let point_ids = point_ids.ok_or_else(|| {
814 DracoError::general(
815 "Point ids missing for integer attribute decoder".to_string(),
816 )
817 })?;
818 let mut att_decoder = SequentialIntegerAttributeDecoder::new();
819 att_decoder.init(self, att_id);
820 if let Some(symbols) = symbols {
821 att_decoder.set_predecoded_symbols(symbols);
822 }
823 att_decoder
824 .decode_values(pc, point_ids, buffer, None, None, None, None, None, None)?;
825 }
826 2 => {
827 let mut att_decoder = SequentialQuantizationAttributeDecoder::new();
828 att_decoder.init(self, pc, att_id)?;
829 if let Some(symbols) = symbols {
830 att_decoder.set_predecoded_symbols(symbols);
831 }
832 let portable = att_decoder.decode_values(
833 pc,
834 point_ids.ok_or_else(|| {
835 DracoError::general(
836 "Point ids missing for quantized attribute decoder".to_string(),
837 )
838 })?,
839 buffer,
840 bitstream_version,
841 PortableExtent::Declared(num_points),
842 None,
843 None,
844 None,
845 None,
846 )?;
847 pending_quant.push(PendingQuant {
848 att_id,
849 portable,
850 transform: att_decoder.into_transform(),
851 });
852 }
853 3 => {
854 let mut att_decoder = SequentialNormalAttributeDecoder::new();
855 att_decoder.init(self, pc, att_id)?;
856 if let Some(symbols) = symbols {
857 att_decoder.set_predecoded_symbols(symbols);
858 }
859 let portable = att_decoder.decode_values(
860 pc,
861 point_ids.ok_or_else(|| {
862 DracoError::general(
863 "Point ids missing for normal attribute decoder".to_string(),
864 )
865 })?,
866 buffer,
867 bitstream_version,
868 PortableExtent::Declared(num_points),
869 None,
870 None,
871 None,
872 None,
873 )?;
874 pending_normals.push(PendingNormal {
875 att_id,
876 portable,
877 quantization_bits: att_decoder.quantization_bits(),
878 });
879 }
880 0 => {
881 let mut att_decoder = SequentialGenericAttributeDecoder::new();
886 att_decoder.init(self, att_id);
887 att_decoder.decode_values(pc, EntryToPointIdMap::identity(num_points), buffer)?;
888 }
889 _ => {
890 return Err(DracoError::general(format!(
891 "Unsupported sequential decoder type: {}",
892 decoder_type
893 )));
894 }
895 }
896 Ok(())
897 }
898
899 #[allow(clippy::too_many_arguments)]
922 fn decode_attributes_in_parallel(
923 &self,
924 pc: &mut PointCloud,
925 buffer: &mut DecoderBuffer,
926 att_ids: &[i32],
927 decoder_types: &[u8],
928 point_ids: Option<EntryToPointIdMap<'_>>,
929 num_points: usize,
930 threads: usize,
931 pending_quant: &mut Vec<PendingQuant>,
932 pending_normals: &mut Vec<PendingNormal>,
933 ) -> Result<bool, DracoError> {
934 let start = buffer.position();
935 let stream = buffer.remaining_data();
936 let template = buffer.child_template();
937
938 let (walked, results) = {
939 let queue = std::sync::Mutex::new(None::<std::sync::mpsc::Receiver<Job>>);
940 let (sender, receiver) = std::sync::mpsc::channel::<Job>();
941 *queue.lock().unwrap() = Some(receiver);
942
943 let work = || {
944 let mut done = Vec::new();
945 loop {
946 let (first, second) = {
952 let guard = queue.lock().unwrap();
953 let receiver = guard.as_ref().expect("the queue is open");
954 let Ok(first) = receiver.recv() else { break };
955 (first, receiver.try_recv().ok())
956 };
957 let index = first.index;
958 match second {
959 Some(second) => {
960 let second_index = second.index;
961 let [a, b] =
962 Job::run_pair(first, second, self, &template, stream, num_points);
963 done.push((index, a));
964 done.push((second_index, b));
965 }
966 None => {
967 done.push((index, first.run(self, &template, stream, num_points, None)))
968 }
969 }
970 }
971 done
972 };
973
974 std::thread::scope(|scope| {
979 let workers: Vec<_> = (0..threads.min(att_ids.len()))
980 .map_while(|_| std::thread::Builder::new().spawn_scoped(scope, work).ok())
981 .collect();
982
983 let walked = self.walk_streams(
984 pc,
985 buffer,
986 att_ids,
987 decoder_types,
988 point_ids,
989 num_points,
990 pending_quant,
991 pending_normals,
992 start,
993 |job| sender.send(job).is_ok(),
994 );
995 drop(sender);
999 let mut finished = vec![work()];
1000 finished.extend(workers.into_iter().map(|worker| {
1001 worker
1002 .join()
1003 .unwrap_or_else(|panic| std::panic::resume_unwind(panic))
1004 }));
1005 let mut results: Vec<Option<Result<Decoded, DracoError>>> = Vec::new();
1006 for done in finished {
1007 for (index, result) in done {
1008 if results.len() <= index {
1009 results.resize_with(index + 1, || None);
1010 }
1011 results[index] = Some(result);
1012 }
1013 }
1014 (walked, results)
1015 })
1016 };
1017
1018 let give_up = |buffer: &mut DecoderBuffer,
1019 pending_quant: &mut Vec<PendingQuant>,
1020 pending_normals: &mut Vec<PendingNormal>|
1021 -> Result<bool, DracoError> {
1022 buffer.rejoin(&template, false);
1025 buffer.set_position(start)?;
1026 pending_quant.clear();
1027 pending_normals.clear();
1028 Ok(false)
1029 };
1030 let jobs = match walked {
1031 Ok(Some(count)) => count,
1032 _ => return give_up(buffer, pending_quant, pending_normals),
1033 };
1034 if jobs == 0 {
1035 buffer.rejoin(&template, true);
1036 return Ok(true);
1037 }
1038 let mut decoded_jobs = Vec::with_capacity(jobs);
1039 for result in results {
1040 match result {
1041 Some(Ok(decoded)) => decoded_jobs.push(decoded),
1042 _ => return give_up(buffer, pending_quant, pending_normals),
1043 }
1044 }
1045 if decoded_jobs.len() != jobs {
1048 return give_up(buffer, pending_quant, pending_normals);
1049 }
1050 buffer.rejoin(&template, true);
1051 #[cfg(test)]
1052 ENGAGED.with(|engaged| engaged.set(engaged.get() + jobs));
1053 for decoded in decoded_jobs {
1054 match decoded {
1055 Decoded::Values { att_id, attribute } => *pc.try_attribute_mut(att_id)? = attribute,
1056 Decoded::Quantized {
1057 att_id,
1058 portable,
1059 transform,
1060 } => pending_quant.push(PendingQuant {
1061 att_id,
1062 portable,
1063 transform,
1064 }),
1065 Decoded::Normal {
1066 att_id,
1067 portable,
1068 quantization_bits,
1069 } => pending_normals.push(PendingNormal {
1070 att_id,
1071 portable,
1072 quantization_bits,
1073 }),
1074 }
1075 }
1076 Ok(true)
1077 }
1078
1079 #[allow(clippy::too_many_arguments)]
1085 fn walk_streams(
1086 &self,
1087 pc: &mut PointCloud,
1088 buffer: &mut DecoderBuffer,
1089 att_ids: &[i32],
1090 decoder_types: &[u8],
1091 point_ids: Option<EntryToPointIdMap<'_>>,
1092 num_points: usize,
1093 pending_quant: &mut Vec<PendingQuant>,
1094 pending_normals: &mut Vec<PendingNormal>,
1095 start: usize,
1096 mut send: impl FnMut(Job) -> bool,
1097 ) -> Result<Option<usize>, DracoError> {
1098 let mut count = 0usize;
1099 for (local_i, &att_id) in att_ids.iter().enumerate() {
1100 let decoder_type = decoder_types[local_i];
1101 if (1..=3).contains(&decoder_type) {
1102 let before = buffer.position();
1103 let components = coded_components(decoder_type, pc.try_attribute(att_id)?);
1104 let skipped = match num_points.checked_mul(components) {
1105 Some(values) => {
1106 SequentialIntegerAttributeDecoder::skip_values(values, components, buffer)
1107 }
1108 None => Ok(false),
1109 };
1110 if matches!(skipped, Ok(true)) {
1111 let job = Job {
1112 index: count,
1113 att_id,
1114 decoder_type,
1115 start: before - start,
1116 end: buffer.position() - start,
1117 seed: pc.try_attribute(att_id)?.clone(),
1118 };
1119 if !send(job) {
1120 return Ok(None);
1121 }
1122 count += 1;
1123 continue;
1124 }
1125 buffer.set_position(before)?;
1126 }
1127 if self
1128 .decode_attribute_in_place(
1129 pc,
1130 buffer,
1131 att_id,
1132 decoder_type,
1133 point_ids,
1134 num_points,
1135 pending_quant,
1136 pending_normals,
1137 None,
1138 )
1139 .is_err()
1140 {
1141 return Ok(None);
1142 }
1143 }
1144 Ok(Some(count))
1145 }
1146}
1147
1148#[cfg(feature = "point_cloud_decode")]
1151struct Job {
1152 index: usize,
1153 att_id: i32,
1154 decoder_type: u8,
1155 start: usize,
1156 end: usize,
1157 seed: PointAttribute,
1158}
1159
1160#[cfg(feature = "point_cloud_decode")]
1162enum Decoded {
1163 Values {
1164 att_id: i32,
1165 attribute: PointAttribute,
1166 },
1167 Quantized {
1168 att_id: i32,
1169 portable: PointAttribute,
1170 transform: AttributeQuantizationTransform,
1171 },
1172 Normal {
1173 att_id: i32,
1174 portable: PointAttribute,
1175 quantization_bits: u8,
1176 },
1177}
1178
1179#[cfg(feature = "point_cloud_decode")]
1180impl Job {
1181 fn run(
1185 self,
1186 decoder: &PointCloudDecoder,
1187 template: &crate::decoder_buffer::ChildTemplate,
1188 stream: &[u8],
1189 num_points: usize,
1190 symbols: Option<Vec<u32>>,
1191 ) -> Result<Decoded, DracoError> {
1192 let bitstream_version = decoder.bitstream_version();
1193 let mut child = template.open(&stream[self.start..self.end]);
1194 let mut mini = PointCloud::new();
1195 mini.set_num_points(num_points);
1196 mini.add_attribute_preserve_unique_id(self.seed);
1197 let ids = EntryToPointIdMap::identity(num_points);
1198 let att_id = self.att_id;
1199 let decoded = match self.decoder_type {
1200 1 => {
1201 let mut att_decoder = SequentialIntegerAttributeDecoder::new();
1202 att_decoder.init(decoder, 0);
1203 if let Some(symbols) = symbols {
1204 att_decoder.set_predecoded_symbols(symbols);
1205 }
1206 att_decoder.decode_values(
1207 &mut mini, ids, &mut child, None, None, None, None, None, None,
1208 )?;
1209 Decoded::Values {
1210 att_id,
1211 attribute: std::mem::take(mini.try_attribute_mut(0)?),
1212 }
1213 }
1214 2 => {
1215 let mut att_decoder = SequentialQuantizationAttributeDecoder::new();
1216 att_decoder.init(decoder, &mini, 0)?;
1217 if let Some(symbols) = symbols {
1218 att_decoder.set_predecoded_symbols(symbols);
1219 }
1220 let portable = att_decoder.decode_values(
1221 &mut mini,
1222 ids,
1223 &mut child,
1224 bitstream_version,
1225 PortableExtent::Declared(num_points),
1226 None,
1227 None,
1228 None,
1229 None,
1230 )?;
1231 Decoded::Quantized {
1232 att_id,
1233 portable,
1234 transform: att_decoder.into_transform(),
1235 }
1236 }
1237 _ => {
1238 let mut att_decoder = SequentialNormalAttributeDecoder::new();
1239 att_decoder.init(decoder, &mini, 0)?;
1240 if let Some(symbols) = symbols {
1241 att_decoder.set_predecoded_symbols(symbols);
1242 }
1243 let portable = att_decoder.decode_values(
1244 &mut mini,
1245 ids,
1246 &mut child,
1247 bitstream_version,
1248 PortableExtent::Declared(num_points),
1249 None,
1250 None,
1251 None,
1252 None,
1253 )?;
1254 Decoded::Normal {
1255 att_id,
1256 portable,
1257 quantization_bits: att_decoder.quantization_bits(),
1258 }
1259 }
1260 };
1261 if child.remaining_size() != 0 {
1262 return Err(DracoError::general(
1263 "An attribute's stream is not the length it was stepped over as".to_string(),
1264 ));
1265 }
1266 Ok(decoded)
1267 }
1268
1269 fn run_pair(
1274 a: Job,
1275 b: Job,
1276 decoder: &PointCloudDecoder,
1277 template: &crate::decoder_buffer::ChildTemplate,
1278 stream: &[u8],
1279 num_points: usize,
1280 ) -> [Result<Decoded, DracoError>; 2] {
1281 let mut child_a = template.open(&stream[a.start..a.end]);
1282 let mut child_b = template.open(&stream[b.start..b.end]);
1283 let values = |job: &Job| {
1284 let components = coded_components(job.decoder_type, &job.seed);
1285 (num_points.checked_mul(components), components)
1286 };
1287 let ((values_a, components_a), (values_b, components_b)) = (values(&a), values(&b));
1288 let symbols = match (values_a, values_b) {
1289 (Some(values_a), Some(values_b))
1290 if SequentialIntegerAttributeDecoder::seek_symbols(&mut child_a)
1291 && SequentialIntegerAttributeDecoder::seek_symbols(&mut child_b) =>
1292 {
1293 decode_raw_symbol_pair(
1294 (&mut child_a, values_a, components_a),
1295 (&mut child_b, values_b, components_b),
1296 )
1297 }
1298 _ => None,
1299 };
1300 #[cfg(test)]
1301 if symbols.is_some() {
1302 PAIRED.with(|paired| paired.set(paired.get() + 2));
1303 }
1304 let (symbols_a, symbols_b) = symbols.map_or((None, None), |(a, b)| (Some(a), Some(b)));
1305 [
1306 a.run(decoder, template, stream, num_points, symbols_a),
1307 b.run(decoder, template, stream, num_points, symbols_b),
1308 ]
1309 }
1310}
1311
1312#[cfg(all(test, feature = "point_cloud_decode", feature = "encoder"))]
1313mod parallel_tests {
1314 use super::*;
1315 use crate::encoder_buffer::EncoderBuffer;
1316 use crate::encoder_options::EncoderOptions;
1317 use crate::point_cloud_encoder::PointCloudEncoder;
1318
1319 struct Xorshift(u64);
1320
1321 impl Xorshift {
1322 fn next(&mut self) -> u64 {
1323 self.0 ^= self.0 << 13;
1324 self.0 ^= self.0 >> 7;
1325 self.0 ^= self.0 << 17;
1326 self.0
1327 }
1328
1329 fn unit(&mut self) -> f32 {
1330 (self.next() >> 40) as f32 / (1u64 << 24) as f32
1331 }
1332 }
1333
1334 fn float_attribute(
1335 kind: GeometryAttributeType,
1336 components: u8,
1337 values: &[f32],
1338 ) -> PointAttribute {
1339 let points = values.len() / components as usize;
1340 let mut attribute = PointAttribute::new();
1341 attribute.init(kind, components, DataType::Float32, false, points);
1342 for (i, value) in values.iter().enumerate() {
1343 attribute.buffer_mut().write(i * 4, &value.to_le_bytes());
1344 }
1345 attribute
1346 }
1347
1348 fn byte_attribute(
1349 kind: GeometryAttributeType,
1350 components: u8,
1351 values: &[u8],
1352 ) -> PointAttribute {
1353 let points = values.len() / components as usize;
1354 let mut attribute = PointAttribute::new();
1355 attribute.init(kind, components, DataType::Uint8, false, points);
1356 for (i, value) in values.iter().enumerate() {
1357 attribute.buffer_mut().write(i, &[*value]);
1358 }
1359 attribute
1360 }
1361
1362 fn stream(points: usize, seed: u64) -> Vec<u8> {
1366 stream_with_threads(points, seed, 1)
1367 }
1368
1369 fn stream_with_threads(points: usize, seed: u64, threads: i32) -> Vec<u8> {
1370 let mut rng = Xorshift(seed);
1371 let mut cloud = PointCloud::new();
1372 cloud.set_num_points(points);
1373 let mut options = EncoderOptions::new();
1374 options.set_encoding_method(0);
1375 options.set_threads(threads);
1376 let mut next_id = 0;
1377 let mut add = |cloud: &mut PointCloud, attribute: PointAttribute, bits: i32| {
1378 cloud.add_attribute(attribute);
1379 if bits > 0 {
1380 options.set_attribute_int(next_id, "quantization_bits", bits);
1381 }
1382 next_id += 1;
1383 };
1384 let positions: Vec<f32> = (0..points * 3).map(|_| rng.unit() * 100.0).collect();
1385 add(
1386 &mut cloud,
1387 float_attribute(GeometryAttributeType::Position, 3, &positions),
1388 16,
1389 );
1390 let normals: Vec<f32> = (0..points * 3).map(|_| rng.unit() * 2.0 - 1.0).collect();
1391 add(
1392 &mut cloud,
1393 float_attribute(GeometryAttributeType::Normal, 3, &normals),
1394 8,
1395 );
1396 for _ in 0..40 {
1397 let values: Vec<f32> = (0..points).map(|_| rng.unit()).collect();
1398 add(
1399 &mut cloud,
1400 float_attribute(GeometryAttributeType::Generic, 1, &values),
1401 8,
1402 );
1403 }
1404 let colours: Vec<u8> = (0..points * 3).map(|_| (rng.next() >> 56) as u8).collect();
1405 add(
1406 &mut cloud,
1407 byte_attribute(GeometryAttributeType::Color, 3, &colours),
1408 0,
1409 );
1410 let tags: Vec<u8> = (0..points).map(|_| (rng.next() >> 56) as u8).collect();
1411 add(
1412 &mut cloud,
1413 byte_attribute(GeometryAttributeType::Generic, 1, &tags),
1414 0,
1415 );
1416 let raw: Vec<f32> = (0..points).map(|_| rng.unit()).collect();
1418 add(
1419 &mut cloud,
1420 float_attribute(GeometryAttributeType::Generic, 1, &raw),
1421 0,
1422 );
1423 for _ in 0..3 {
1428 let zeros = vec![0.0f32; points];
1429 add(
1430 &mut cloud,
1431 float_attribute(GeometryAttributeType::Generic, 1, &zeros),
1432 8,
1433 );
1434 }
1435 let same = vec![7u8; points];
1436 add(
1437 &mut cloud,
1438 byte_attribute(GeometryAttributeType::Generic, 1, &same),
1439 0,
1440 );
1441 let mut encoder = PointCloudEncoder::new();
1442 encoder.set_point_cloud(cloud);
1443 let mut buffer = EncoderBuffer::new();
1444 encoder.encode(&options, &mut buffer).expect("encodes");
1445 buffer.data().to_vec()
1446 }
1447
1448 fn decode(bytes: &[u8], threads: i32) -> Result<Vec<Vec<u8>>, String> {
1450 let mut decoder = PointCloudDecoder::new();
1451 decoder.set_threads(threads);
1452 let mut cloud = PointCloud::new();
1453 decoder
1454 .decode(&mut DecoderBuffer::new(bytes), &mut cloud)
1455 .map_err(|e| e.to_string())?;
1456 Ok((0..cloud.num_attributes())
1457 .map(|id| cloud.attribute(id).buffer().data().to_vec())
1458 .collect())
1459 }
1460
1461 fn decode_in_order(bytes: &[u8]) -> Result<Vec<Vec<u8>>, String> {
1464 IN_ORDER.with(|in_order| in_order.set(true));
1465 let decoded = decode(bytes, 1);
1466 IN_ORDER.with(|in_order| in_order.set(false));
1467 decoded
1468 }
1469
1470 fn engaged_by<T>(f: impl FnOnce() -> T) -> (T, usize) {
1471 let before = ENGAGED.with(|e| e.get());
1472 let out = f();
1473 (out, ENGAGED.with(|e| e.get()) - before)
1474 }
1475
1476 #[test]
1477 fn side_by_side_decodes_the_same_cloud_value_for_value() {
1478 let bytes = stream(40_000, 1);
1479 assert!(
1480 bytes.len() >= PARALLEL_MIN_STREAM_BYTES,
1481 "{} bytes",
1482 bytes.len()
1483 );
1484 let (serial, engaged) = engaged_by(|| decode_in_order(&bytes));
1485 assert_eq!(engaged, 0, "the reference took the side-by-side path");
1486 let serial = serial.expect("decodes");
1487 for threads in [1, 2, 5, 16] {
1488 let paired_before = PAIRED.with(|paired| paired.get());
1489 let (parallel, engaged) = engaged_by(|| decode(&bytes, threads));
1490 if threads == 1 {
1494 let paired = PAIRED.with(|paired| paired.get()) - paired_before;
1495 assert_eq!(engaged, 0, "one thread took the side-by-side path");
1496 assert!(paired >= 40, "{paired} paired in order");
1497 } else {
1498 assert!(
1499 engaged >= 44,
1500 "the parallel path ran for {engaged} streams at {threads} threads"
1501 );
1502 }
1503 assert_eq!(parallel.expect("decodes"), serial, "{threads} threads");
1504 }
1505 }
1506
1507 #[test]
1511 fn the_encoded_stream_is_the_same_bytes_at_any_number_of_threads() {
1512 let single = stream_with_threads(40_000, 5, 1);
1513 for threads in [2, 7, 16, 0] {
1514 assert_eq!(
1515 stream_with_threads(40_000, 5, threads),
1516 single,
1517 "{threads} threads"
1518 );
1519 }
1520 }
1521
1522 #[test]
1527 fn an_attribute_split_across_threads_writes_the_same_bytes() {
1528 let points = 400_000;
1529 let mut rng = Xorshift(11);
1530 let mut cloud = PointCloud::new();
1531 cloud.set_num_points(points);
1532 let positions: Vec<f32> = (0..points * 3)
1535 .map(|i| match i % 97 {
1536 0 => -0.0,
1537 1 => 0.0,
1538 _ => (rng.unit() * 64.0).floor() - 32.0,
1539 })
1540 .collect();
1541 cloud.add_attribute(float_attribute(
1542 GeometryAttributeType::Position,
1543 3,
1544 &positions,
1545 ));
1546 let colours: Vec<u8> = (0..points * 3).map(|_| (rng.next() >> 58) as u8).collect();
1547 cloud.add_attribute(byte_attribute(GeometryAttributeType::Color, 3, &colours));
1548 let encode = |threads: i32, search: bool| {
1549 let mut options = EncoderOptions::new();
1550 options.set_encoding_method(0);
1551 options.set_attribute_int(0, "quantization_bits", 14);
1552 options.set_threads(threads);
1553 options.set_prediction_search(search);
1554 let mut encoder = PointCloudEncoder::new();
1555 encoder.set_point_cloud(cloud.clone());
1556 let mut buffer = EncoderBuffer::new();
1557 encoder.encode(&options, &mut buffer).expect("encodes");
1558 buffer.data().to_vec()
1559 };
1560 for search in [false, true] {
1561 let single = encode(1, search);
1562 for threads in [4, 16] {
1563 assert!(
1564 encode(threads, search) == single,
1565 "{threads} threads, search {search}"
1566 );
1567 }
1568 }
1569 }
1570
1571 #[test]
1572 fn a_stream_too_small_to_repay_the_threads_is_decoded_in_place() {
1573 let bytes = stream(500, 2);
1574 let (parallel, engaged) = engaged_by(|| decode(&bytes, 16));
1575 assert_eq!(engaged, 0);
1576 assert_eq!(parallel, decode_in_order(&bytes));
1577 }
1578
1579 #[test]
1580 fn a_damaged_stream_is_read_by_the_threads_as_by_one() {
1581 let bytes = stream(40_000, 3);
1582 let mut rng = Xorshift(77);
1583 for trial in 0..40 {
1584 let mut damaged = bytes.clone();
1585 let at = if trial % 2 == 0 {
1588 rng.next() as usize % damaged.len()
1589 } else {
1590 rng.next() as usize % 400
1591 };
1592 damaged[at] ^= 1 << (rng.next() % 8);
1593 let serial = decode_in_order(&damaged);
1594 for threads in [1, 16] {
1595 let parallel = decode(&damaged, threads);
1596 assert_eq!(
1597 parallel, serial,
1598 "trial {trial}: byte {at} damaged, {threads} threads"
1599 );
1600 }
1601 }
1602 for cut in [bytes.len() / 2, bytes.len() - 1, bytes.len() - 40, 5000] {
1604 for threads in [1, 16] {
1605 assert_eq!(
1606 decode(&bytes[..cut], threads),
1607 decode_in_order(&bytes[..cut]),
1608 "cut at {cut}, {threads} threads"
1609 );
1610 }
1611 }
1612 }
1613
1614 #[test]
1620 fn constant_attributes_in_a_big_cloud_do_not_exhaust_what_threads_share() {
1621 let points = 7_000_000;
1622 let mut rng = Xorshift(9);
1623 let mut cloud = PointCloud::new();
1624 cloud.set_num_points(points);
1625 let mut options = EncoderOptions::new();
1626 options.set_encoding_method(0);
1627 options.set_threads(1);
1628 let mut next_id = 0;
1629 let mut add = |cloud: &mut PointCloud, attribute: PointAttribute| {
1630 cloud.add_attribute(attribute);
1631 options.set_attribute_int(next_id, "quantization_bits", 8);
1632 next_id += 1;
1633 };
1634 for _ in 0..3 {
1635 let noise: Vec<f32> = (0..points).map(|_| rng.unit()).collect();
1636 add(
1637 &mut cloud,
1638 float_attribute(GeometryAttributeType::Generic, 1, &noise),
1639 );
1640 }
1641 for _ in 0..4 {
1642 let zeros = vec![0.0f32; points];
1643 add(
1644 &mut cloud,
1645 float_attribute(GeometryAttributeType::Generic, 1, &zeros),
1646 );
1647 }
1648 let mut encoder = PointCloudEncoder::new();
1649 encoder.set_point_cloud(cloud);
1650 let mut buffer = EncoderBuffer::new();
1651 encoder.encode(&options, &mut buffer).expect("encodes");
1652 let bytes = buffer.data();
1653 assert!(bytes.len() >= PARALLEL_MIN_STREAM_BYTES);
1654 let serial = decode_in_order(bytes).expect("decodes");
1655 for threads in [1, 4, 12, 16] {
1656 let (parallel, engaged) = engaged_by(|| decode(bytes, threads));
1657 assert_eq!(
1658 engaged,
1659 if threads > 1 { 7 } else { 0 },
1660 "{threads} threads: the parallel path ran for {engaged} of 7 streams"
1661 );
1662 assert_eq!(parallel.expect("decodes"), serial, "{threads} threads");
1663 }
1664 }
1665
1666 #[test]
1673 fn constant_attributes_past_the_backstop_decode_side_by_side() {
1674 const CONSTANT: usize = 4;
1675 const COMPONENTS: usize = 3;
1676 const CEILING_IN_VALUES: usize =
1677 crate::decode_budget::MAX_UNBACKED_BYTES / std::mem::size_of::<u32>();
1678 const POINTS: usize = CEILING_IN_VALUES / (CONSTANT * COMPONENTS) * 5 / 4;
1681 let mut cloud = PointCloud::new();
1682 cloud.set_num_points(POINTS);
1683 for _ in 0..CONSTANT {
1684 let colour = vec![0u8; POINTS * COMPONENTS];
1685 cloud.add_attribute(byte_attribute(
1686 GeometryAttributeType::Color,
1687 COMPONENTS as u8,
1688 &colour,
1689 ));
1690 }
1691 let mut rng = Xorshift(21);
1692 let noise: Vec<u8> = (0..POINTS).map(|_| (rng.next() >> 56) as u8).collect();
1693 cloud.add_attribute(byte_attribute(GeometryAttributeType::Generic, 1, &noise));
1694 let mut options = EncoderOptions::new();
1695 options.set_encoding_method(0);
1696 options.set_threads(1);
1697 let mut encoder = PointCloudEncoder::new();
1698 encoder.set_point_cloud(cloud);
1699 let mut buffer = EncoderBuffer::new();
1700 encoder.encode(&options, &mut buffer).expect("encodes");
1701 let bytes = buffer.data();
1702 assert!(bytes.len() >= PARALLEL_MIN_STREAM_BYTES);
1703 let serial = decode_in_order(bytes).expect("decodes");
1704 for threads in [1, 16] {
1705 let (parallel, engaged) = engaged_by(|| decode(bytes, threads));
1706 assert_eq!(
1707 engaged,
1708 if threads > 1 { CONSTANT + 1 } else { 0 },
1709 "{threads} threads: {engaged} streams decoded aside"
1710 );
1711 assert_eq!(parallel.expect("decodes"), serial, "{threads} threads");
1712 }
1713 }
1714
1715 #[test]
1721 fn synthetic_clouds_come_back_the_same_by_every_path() {
1722 use crate::synthetic_cloud::{self, Cloud};
1723 let mut clouds: Vec<Cloud> = [0, 1, 2, 3, 97]
1724 .into_iter()
1725 .flat_map(|points| synthetic_cloud::all(points, 1))
1726 .collect();
1727 let big = synthetic_cloud::all(400_000, 2);
1730 clouds.extend(big.iter().map(|cloud| cloud.shuffled(3)));
1731 clouds.extend(big);
1732 std::thread::scope(|scope| {
1733 for cloud in &clouds {
1734 scope.spawn(|| every_path_agrees(cloud));
1735 }
1736 });
1737 }
1738
1739 fn every_path_agrees(cloud: &crate::synthetic_cloud::Cloud) {
1740 let name = format!("{} of {} points", cloud.name, cloud.points);
1741 let encode = |threads: i32| {
1742 let mut options = EncoderOptions::new();
1743 options.set_encoding_method(0);
1744 options.set_threads(threads);
1745 for (id, column) in cloud.columns.iter().enumerate() {
1746 if column.quantization_bits > 0 {
1747 options.set_attribute_int(
1748 id as i32,
1749 "quantization_bits",
1750 column.quantization_bits,
1751 );
1752 }
1753 }
1754 let mut encoder = PointCloudEncoder::new();
1755 encoder.set_point_cloud(cloud.to_point_cloud());
1756 let mut buffer = EncoderBuffer::new();
1757 encoder
1758 .encode(&options, &mut buffer)
1759 .map(|()| buffer.data().to_vec())
1760 .map_err(|e| e.to_string())
1761 };
1762 let single = encode(1);
1763 for threads in [4, 16] {
1764 assert!(
1765 encode(threads) == single,
1766 "{name}: the stream on {threads} threads differs"
1767 );
1768 }
1769 let bytes = match single {
1772 Err(e) if cloud.points == 0 && e.contains("empty attribute") => return,
1773 result => result.unwrap_or_else(|e| panic!("{name}: {e}")),
1774 };
1775 let decoded = decode_in_order(&bytes).unwrap_or_else(|e| panic!("{name}: {e}"));
1776 let big = cloud.points >= 400_000 && !cloud.name.starts_with("constant");
1777 assert!(
1778 !big || bytes.len() >= PARALLEL_MIN_STREAM_BYTES,
1779 "{name}: {} bytes are too few to send the decode to threads",
1780 bytes.len()
1781 );
1782 for threads in [1, 4, 16] {
1783 let paired = PAIRED.with(|paired| paired.get());
1784 let (parallel, engaged) = engaged_by(|| decode(&bytes, threads));
1785 let paired = PAIRED.with(|paired| paired.get()) - paired;
1786 assert!(
1787 parallel.as_ref() == Ok(&decoded),
1788 "{name}: the decode on {threads} threads differs"
1789 );
1790 assert!(
1791 !big || if threads > 1 {
1792 engaged > 0
1793 } else {
1794 engaged == 0
1795 },
1796 "{name}: {threads} threads decoded {engaged} streams aside, {paired} paired"
1797 );
1798 }
1799 for (column, got) in cloud.columns.iter().zip(&decoded) {
1800 let source = column.values.bytes();
1801 let got = &got[..source.len()];
1802 if column.quantization_bits == 0 {
1803 assert!(got == source, "{name}: {} changed", column.name);
1804 continue;
1805 }
1806 let values = |bytes: &[u8]| -> Vec<f32> {
1807 bytes
1808 .as_chunks::<4>()
1809 .0
1810 .iter()
1811 .map(|&b| f32::from_le_bytes(b))
1812 .collect()
1813 };
1814 let (source, got) = (values(&source), values(got));
1815 if column.kind == GeometryAttributeType::Normal {
1816 let cell = 4.0 / (1u32 << column.quantization_bits) as f32;
1820 let pairs = source.as_chunks::<3>().0.iter().zip(got.as_chunks::<3>().0);
1821 for (i, (want, have)) in pairs.enumerate() {
1822 if want.iter().all(|&v| v == 0.0) {
1823 continue;
1824 }
1825 let error = want
1826 .iter()
1827 .zip(have)
1828 .map(|(w, h)| (w - h).abs())
1829 .fold(0.0, f32::max);
1830 assert!(
1831 error <= 2.0 * cell,
1832 "{name}: normal {i} came back {have:?} for {want:?}"
1833 );
1834 }
1835 continue;
1836 }
1837 let range = (0..column.components)
1839 .map(|c| {
1840 let component = source.iter().skip(c).step_by(column.components);
1841 let (lo, hi) =
1842 component.fold((f32::MAX, f32::MIN), |(lo, hi), &v| (lo.min(v), hi.max(v)));
1843 hi - lo
1844 })
1845 .fold(0.0f32, f32::max);
1846 let step = range / ((1u32 << column.quantization_bits) - 1) as f32;
1847 for (i, (&want, &have)) in source.iter().zip(&got).enumerate() {
1848 let slack = (want.abs() + range) * f32::EPSILON * 4.0;
1849 assert!(
1850 (want - have).abs() <= step * 0.5 + slack,
1851 "{name}: {} value {i} came back {have} for {want}, step {step}",
1852 column.name
1853 );
1854 }
1855 }
1856 }
1857}
1858
1859#[cfg(all(test, feature = "point_cloud_decode", feature = "encoder"))]
1860mod pair_tests {
1861 use super::*;
1862 use crate::encoder_buffer::EncoderBuffer;
1863 use crate::encoder_options::EncoderOptions;
1864 use crate::point_cloud_encoder::PointCloudEncoder;
1865
1866 struct Xorshift(u64);
1867
1868 impl Xorshift {
1869 fn next(&mut self) -> u64 {
1870 self.0 ^= self.0 << 13;
1871 self.0 ^= self.0 >> 7;
1872 self.0 ^= self.0 << 17;
1873 self.0
1874 }
1875
1876 fn unit(&mut self) -> f32 {
1877 (self.next() >> 40) as f32 / (1u64 << 24) as f32
1878 }
1879 }
1880
1881 fn float_attribute(
1882 kind: GeometryAttributeType,
1883 components: u8,
1884 values: &[f32],
1885 ) -> PointAttribute {
1886 let points = values.len() / components as usize;
1887 let mut attribute = PointAttribute::new();
1888 attribute.init(kind, components, DataType::Float32, false, points);
1889 for (i, value) in values.iter().enumerate() {
1890 attribute.buffer_mut().write(i * 4, &value.to_le_bytes());
1891 }
1892 attribute
1893 }
1894
1895 fn byte_attribute(
1896 kind: GeometryAttributeType,
1897 components: u8,
1898 values: &[u8],
1899 ) -> PointAttribute {
1900 let points = values.len() / components as usize;
1901 let mut attribute = PointAttribute::new();
1902 attribute.init(kind, components, DataType::Uint8, false, points);
1903 for (i, value) in values.iter().enumerate() {
1904 attribute.buffer_mut().write(i, &[*value]);
1905 }
1906 attribute
1907 }
1908
1909 fn stream(points: usize, seed: u64) -> Vec<u8> {
1914 let mut rng = Xorshift(seed);
1915 let mut cloud = PointCloud::new();
1916 cloud.set_num_points(points);
1917 let mut options = EncoderOptions::new();
1918 options.set_encoding_method(0);
1919 let mut next_id = 0;
1920 let mut add = |cloud: &mut PointCloud, attribute: PointAttribute, bits: i32| {
1921 cloud.add_attribute(attribute);
1922 if bits > 0 {
1923 options.set_attribute_int(next_id, "quantization_bits", bits);
1924 }
1925 next_id += 1;
1926 };
1927 let positions: Vec<f32> = (0..points * 3).map(|_| rng.unit() * 100.0).collect();
1928 add(
1929 &mut cloud,
1930 float_attribute(GeometryAttributeType::Position, 3, &positions),
1931 16,
1932 );
1933 let normals: Vec<f32> = (0..points * 3).map(|_| rng.unit() * 2.0 - 1.0).collect();
1934 add(
1935 &mut cloud,
1936 float_attribute(GeometryAttributeType::Normal, 3, &normals),
1937 8,
1938 );
1939 for _ in 0..40 {
1940 let values: Vec<f32> = (0..points).map(|_| rng.unit()).collect();
1941 add(
1942 &mut cloud,
1943 float_attribute(GeometryAttributeType::Generic, 1, &values),
1944 8,
1945 );
1946 }
1947 let colours: Vec<u8> = (0..points * 3).map(|_| (rng.next() >> 56) as u8).collect();
1948 add(
1949 &mut cloud,
1950 byte_attribute(GeometryAttributeType::Color, 3, &colours),
1951 0,
1952 );
1953 let raw: Vec<f32> = (0..points).map(|_| rng.unit()).collect();
1955 add(
1956 &mut cloud,
1957 float_attribute(GeometryAttributeType::Generic, 1, &raw),
1958 0,
1959 );
1960 for _ in 0..3 {
1961 let zeros = vec![0.0f32; points];
1962 add(
1963 &mut cloud,
1964 float_attribute(GeometryAttributeType::Generic, 1, &zeros),
1965 8,
1966 );
1967 }
1968 let tags: Vec<u8> = (0..points).map(|_| (rng.next() >> 56) as u8).collect();
1969 add(
1970 &mut cloud,
1971 byte_attribute(GeometryAttributeType::Generic, 1, &tags),
1972 0,
1973 );
1974 let mut encoder = PointCloudEncoder::new();
1975 encoder.set_point_cloud(cloud);
1976 let mut buffer = EncoderBuffer::new();
1977 encoder.encode(&options, &mut buffer).expect("encodes");
1978 buffer.data().to_vec()
1979 }
1980
1981 fn decode(
1984 bytes: &[u8],
1985 pairs: bool,
1986 ) -> (Result<Vec<Vec<u8>>, String>, crate::decoder_buffer::Budget) {
1987 PAIRS_OFF.with(|off| off.set(!pairs));
1988 let mut buffer = DecoderBuffer::new(bytes);
1989 let mut cloud = PointCloud::new();
1990 let mut decoder = PointCloudDecoder::new();
1992 decoder.set_threads(1);
1993 let decoded = decoder
1994 .decode(&mut buffer, &mut cloud)
1995 .map_err(|e| e.to_string())
1996 .map(|()| {
1997 (0..cloud.num_attributes())
1998 .map(|id| cloud.attribute(id).buffer().data().to_vec())
1999 .collect()
2000 });
2001 PAIRS_OFF.with(|off| off.set(false));
2002 (decoded, buffer.budget())
2003 }
2004
2005 fn paired_by<T>(f: impl FnOnce() -> T) -> (T, usize) {
2006 let before = PAIRED.with(|paired| paired.get());
2007 let out = f();
2008 (out, PAIRED.with(|paired| paired.get()) - before)
2009 }
2010
2011 #[test]
2016 fn paired_streams_decode_the_same_cloud_value_for_value() {
2017 let bytes = stream(40_000, 1);
2018 let ((alone, budget_alone), paired) = paired_by(|| decode(&bytes, false));
2019 assert_eq!(paired, 0, "the reference paired");
2020 let ((together, budget_together), paired) = paired_by(|| decode(&bytes, true));
2021 assert!(paired >= 44, "{paired} attributes paired");
2022 assert_eq!(together.expect("decodes"), alone.expect("decodes"));
2023 assert_eq!(budget_together, budget_alone, "the budget");
2024 }
2025
2026 #[test]
2027 fn a_damaged_stream_is_read_paired_as_alone() {
2028 let bytes = stream(20_000, 3);
2029 let mut rng = Xorshift(77);
2030 for trial in 0..40 {
2031 let mut damaged = bytes.clone();
2032 let at = if trial % 2 == 0 {
2035 rng.next() as usize % damaged.len()
2036 } else {
2037 rng.next() as usize % 400
2038 };
2039 damaged[at] ^= 1 << (rng.next() % 8);
2040 assert_eq!(
2041 decode(&damaged, true),
2042 decode(&damaged, false),
2043 "trial {trial}: byte {at} damaged"
2044 );
2045 }
2046 for cut in [bytes.len() / 2, bytes.len() - 1, bytes.len() - 40, 5000] {
2048 assert_eq!(
2049 decode(&bytes[..cut], true),
2050 decode(&bytes[..cut], false),
2051 "cut at {cut}"
2052 );
2053 }
2054 }
2055
2056 #[test]
2063 fn synthetic_clouds_come_back_the_same_by_every_path() {
2064 use crate::synthetic_cloud::{self, Cloud};
2065 let mut clouds: Vec<Cloud> = [0, 1, 2, 3, 97]
2066 .into_iter()
2067 .flat_map(|points| synthetic_cloud::all(points, 1))
2068 .collect();
2069 let big = synthetic_cloud::all(400_000, 2);
2070 clouds.extend(big.iter().map(|cloud| cloud.shuffled(3)));
2071 clouds.extend(big);
2072 let paired: usize = std::thread::scope(|scope| {
2073 let runs: Vec<_> = clouds
2074 .iter()
2075 .map(|cloud| scope.spawn(|| paired_by(|| every_path_agrees(cloud)).1))
2076 .collect();
2077 runs.into_iter()
2078 .map(|run| run.join().expect("a cloud's check panicked"))
2079 .sum()
2080 });
2081 assert!(paired > 0, "no synthetic cloud paired a single stream");
2082 }
2083
2084 fn every_path_agrees(cloud: &crate::synthetic_cloud::Cloud) {
2085 let name = format!("{} of {} points", cloud.name, cloud.points);
2086 let mut options = EncoderOptions::new();
2087 options.set_encoding_method(0);
2088 for (id, column) in cloud.columns.iter().enumerate() {
2089 if column.quantization_bits > 0 {
2090 options.set_attribute_int(id as i32, "quantization_bits", column.quantization_bits);
2091 }
2092 }
2093 let mut encoder = PointCloudEncoder::new();
2094 encoder.set_point_cloud(cloud.to_point_cloud());
2095 let mut buffer = EncoderBuffer::new();
2096 match encoder.encode(&options, &mut buffer) {
2099 Err(e) if cloud.points == 0 && e.to_string().contains("empty attribute") => return,
2100 result => result.unwrap_or_else(|e| panic!("{name}: {e}")),
2101 }
2102 let bytes = buffer.data().to_vec();
2103 let (alone, budget_alone) = decode(&bytes, false);
2104 let decoded = alone.unwrap_or_else(|e| panic!("{name}: {e}"));
2105 let (together, budget_together) = decode(&bytes, true);
2106 assert!(
2107 together.as_ref() == Ok(&decoded),
2108 "{name}: the paired decode differs"
2109 );
2110 assert_eq!(budget_together, budget_alone, "{name}: the budget");
2111 for (column, got) in cloud.columns.iter().zip(&decoded) {
2112 let source = column.values.bytes();
2113 let got = &got[..source.len()];
2114 if column.quantization_bits == 0 {
2115 assert!(got == source, "{name}: {} changed", column.name);
2116 continue;
2117 }
2118 let values = |bytes: &[u8]| -> Vec<f32> {
2119 bytes
2120 .as_chunks::<4>()
2121 .0
2122 .iter()
2123 .map(|&b| f32::from_le_bytes(b))
2124 .collect()
2125 };
2126 let (source, got) = (values(&source), values(got));
2127 if column.kind == GeometryAttributeType::Normal {
2128 let cell = 4.0 / (1u32 << column.quantization_bits) as f32;
2132 let pairs = source.as_chunks::<3>().0.iter().zip(got.as_chunks::<3>().0);
2133 for (i, (want, have)) in pairs.enumerate() {
2134 if want.iter().all(|&v| v == 0.0) {
2135 continue;
2136 }
2137 let error = want
2138 .iter()
2139 .zip(have)
2140 .map(|(w, h)| (w - h).abs())
2141 .fold(0.0, f32::max);
2142 assert!(
2143 error <= 2.0 * cell,
2144 "{name}: normal {i} came back {have:?} for {want:?}"
2145 );
2146 }
2147 continue;
2148 }
2149 let range = (0..column.components)
2151 .map(|c| {
2152 let component = source.iter().skip(c).step_by(column.components);
2153 let (lo, hi) =
2154 component.fold((f32::MAX, f32::MIN), |(lo, hi), &v| (lo.min(v), hi.max(v)));
2155 hi - lo
2156 })
2157 .fold(0.0f32, f32::max);
2158 let step = range / ((1u32 << column.quantization_bits) - 1) as f32;
2159 for (i, (&want, &have)) in source.iter().zip(&got).enumerate() {
2160 let slack = (want.abs() + range) * f32::EPSILON * 4.0;
2161 assert!(
2162 (want - have).abs() <= step * 0.5 + slack,
2163 "{name}: {} value {i} came back {have} for {want}, step {step}",
2164 column.name
2165 );
2166 }
2167 }
2168 }
2169}