Skip to main content

draco_core/
point_cloud_decoder.rs

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/// Whether a prediction transform byte follows this prediction method byte.
41///
42/// Upstream writes the transform only when the method is not `PREDICTION_NONE`,
43/// which is `-2` and reaches the stream as `0xFE`. `0xFF` is `-1`, which this
44/// crate once wrote for the same meaning, so both are read as "nothing follows"
45/// -- the same pair `SequentialIntegerAttributeDecoder` accepts.
46///
47/// The pre-1.2 shims below need this because they walk the prediction header by
48/// hand to reach the quantization parameters behind it. Testing `0xFF` alone
49/// made them step one byte into a `PREDICTION_NONE` stream and read the
50/// parameters shifted: the range came out zero and every position dequantized to
51/// the origin, with the point and face counts still right and the decode still
52/// reporting success. Draco writes `PREDICTION_NONE` at compression level 0.
53/// Gated on the feature its callers live behind, and only that one: both of
54/// them are now the pre-2.0 shims inside the shared normal and quantization
55/// decoders, so a `point_cloud_decode` build without legacy support compiles
56/// neither and would carry this as dead code.
57#[cfg(feature = "legacy_bitstream_decode")]
58pub(crate) fn carries_transform_byte(method_byte: u8) -> bool {
59    method_byte != 0xFF && method_byte != 0xFE
60}
61
62/// Decoder for Draco point cloud bitstreams.
63///
64/// `PointCloudDecoder` reads a point-cloud `.drc` bitstream and reconstructs a
65/// [`PointCloud`] with its attributes and metadata. Both
66/// KD-tree and sequential attribute encodings are supported (the actual decode
67/// requires the `point_cloud_decode` feature).
68///
69/// A round trip is shown on the `PointCloudEncoder` type docs.
70pub 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    /// Ungated, unlike the fields above: `bitstream_version` is read by the
77    /// attribute decoders on both the mesh and the point-cloud path, and the
78    /// mesh path exists without `point_cloud_decode`.
79    version_major: u8,
80    version_minor: u8,
81    #[cfg(feature = "point_cloud_decode")]
82    threads: i32,
83}
84
85/// The stream length below which a point cloud stays on the calling thread. A
86/// stream this small claiming this many values is what the allocation budget
87/// exists for, and it is read by one thread and one budget.
88///
89/// A fuzzing build takes it down with the other thread gates; see
90/// `parallel::ATTRIBUTES_MIN_VALUES`.
91#[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/// The components an attribute's integer stream codes, for the sequential
97/// decoder types that have one (1 to 3): a normal's are the two of its
98/// octahedral coordinates, not the attribute's three.
99#[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// How many attribute streams this thread has decoded side by side, for the
108// tests that must see the parallel path run rather than fall back.
109#[cfg(all(test, feature = "point_cloud_decode"))]
110thread_local! {
111    static ENGAGED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
112    // How many jobs this thread ran on symbols decoded beside another's.
113    static PAIRED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
114    // Set by the tests that need the decode in order, attribute by attribute,
115    // as the reference the side-by-side decode is held to.
116    static IN_ORDER: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
117    // Set by the tests that hold the decode in order with its pairs to the one
118    // without them.
119    static PAIRS_OFF: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
120}
121
122/// A quantized attribute whose values are decoded and whose inverse transform
123/// has not yet run.
124#[cfg(feature = "point_cloud_decode")]
125struct PendingQuant {
126    att_id: i32,
127    portable: PointAttribute,
128    transform: AttributeQuantizationTransform,
129}
130
131/// A normal attribute whose octahedral values are decoded and whose inverse
132/// transform has not yet run.
133#[cfg(feature = "point_cloud_decode")]
134struct PendingNormal {
135    att_id: i32,
136    portable: PointAttribute,
137    quantization_bits: u8,
138}
139
140/// Runs the inverse quantization of every pending attribute, one thread each at
141/// most. Each writes only its own attribute, so the attributes are taken out of
142/// the cloud for the duration and put back. They are taken in attribute order,
143/// whatever order the decode left them pending in, so the first error in that
144/// order is the one reported, on any number of threads.
145#[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/// [`dequantize_in_parallel`] for the octahedral normals.
175#[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    // Each attribute must have at least type, data type, component count,
227    // normalized flag, unique id, and a decoder type byte. Reject impossible
228    // counts before reserving vectors from untrusted input.
229    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    /// Creates a point cloud decoder with default state.
252    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    /// Caps the threads a decode may use: `1`, the default, keeps everything on
267    /// the calling thread, `0` is as many as the machine has, and no count is
268    /// taken past sixteen. One by default because a library cannot see what
269    /// its caller already runs: a server decoding on a pool of its own would
270    /// have every decode multiply its threads by up to sixteen, and a hostile
271    /// stream would have the machine's cores by default. A thread the system
272    /// refuses to start is one fewer rather than an error.
273    /// [`MeshDecoder::set_threads`] passes the same cap to the point-cloud
274    /// streams a mesh decoder reads.
275    ///
276    /// [`MeshDecoder::set_threads`]: crate::MeshDecoder::set_threads
277    ///
278    /// A sequential point cloud with many attributes decodes them side by side
279    /// once the stream is large enough to repay it, and the decoded cloud is the
280    /// one a single thread produces, value for value. A large cloud of many
281    /// attributes, a Gaussian splat for one, decodes several times faster on
282    /// many threads. A cloud of two attributes, a scan's position and colour,
283    /// gains next to nothing: its decode is the position's one chain. On one
284    /// thread the attributes are decoded in order. On WebAssembly there are no
285    /// threads and the value is ignored.
286    #[cfg(feature = "point_cloud_decode")]
287    pub fn set_threads(&mut self, threads: i32) {
288        self.threads = threads;
289    }
290
291    /// The packed bitstream version (`0xMMmm`), `0` before a header was read.
292    ///
293    /// The attribute decoders read it when they bind prediction parents, which
294    /// is upstream's `decoder_->bitstream_version()` inside
295    /// `InitPredictionScheme`.
296    pub(crate) fn bitstream_version(&self) -> u16 {
297        crate::version::bitstream_version(self.version_major, self.version_minor)
298    }
299
300    /// Carries the version to a decoder that did not read the header itself.
301    ///
302    /// The mesh path parses its own header and then hands attributes to these
303    /// decoders through a `PointCloudDecoder` it constructs on the spot, so
304    /// without this that decoder reports version zero -- and every parent
305    /// binding on the fallback path would read as pre-2.0.
306    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    /// Decodes a Draco point cloud from `in_buffer` into `out_pc`.
313    ///
314    /// `out_pc` need not be empty: whatever it held is replaced, the same way
315    /// [`MeshDecoder::decode`](crate::mesh_decoder::MeshDecoder::decode)
316    /// replaces its mesh. `decode_after_header` does not clear, because its
317    /// caller has already done so and has decoded metadata since.
318    ///
319    /// # Errors
320    ///
321    /// Returns an error if the header is invalid, the bitstream version is
322    /// unsupported, or the encoded attributes are malformed.
323    pub fn decode(&mut self, in_buffer: &mut DecoderBuffer, out_pc: &mut PointCloud) -> Status {
324        out_pc.clear();
325
326        // 1. Decode Header
327        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        // 2. Decode Geometry Data
341        self.decode_geometry_data(in_buffer, out_pc)
342    }
343
344    /// Decode point cloud data when header + metadata have already been parsed.
345    /// Used by MeshDecoder to delegate point cloud streams.
346    #[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        // Flags field is always present in the binary header (C++ reads unconditionally).
390        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        // Note: Draco point cloud bitstreams encode the number of points as a
402        // fixed-width int32 for both sequential (method=0) and KD-tree
403        // (method=1) encodings (see C++ PointCloudSequentialDecoder and
404        // PointCloudKdTreeDecoder). It is NOT varint encoded, even for v2.x.
405        // Read as the `int32_t` upstream reads, and refused when negative for
406        // the same reason `PointCloudSequentialDecoder` and
407        // `PointCloudKdTreeDecoder` refuse it: no encoder writes a count with
408        // the sign bit set, and C++ Draco stops on one before it allocates
409        // anything. Taking the same bytes unsigned is how a header claiming
410        // 2,147,483,652 points reached the KD-tree walk, which then expanded a
411        // single run-length node into 2.4 GB -- an OOM the `decode_drc` soak
412        // found, on a file upstream rejects in a fifth of a second having
413        // touched no memory at all.
414        //
415        // Past this the count is used but not guarded, which is also upstream's
416        // shape: what bounds the work is the allocation budget applied where
417        // the buffers are sized -- see `decode_budget`.
418        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            // KD-tree encoding.
432            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            // Sequential encoding.
443            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                    // Nothing is charged against the *budget* for this
526                    // attribute, because nothing is taken for it: the buffer is
527                    // left unreserved and sized by whichever decoder writes the
528                    // values, once they exist. A charge there would be for an
529                    // allocation that no longer happens, and it is not free --
530                    // the budget is a backstop against unbacked reservations,
531                    // and billing it for backed ones is what made it refuse
532                    // files this crate writes. The caller's ceiling above is
533                    // the other question and is charged: it bounds what the
534                    // decode may produce at all, backed or not.
535                    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                // The identity, and not written out. Entry `i` is point `i`
548                // here, so materializing it bought nothing and cost four bytes
549                // per point of a count the header supplies -- 134 MB from a
550                // 9 KB stream on one fuzz artifact, and gigabytes on a bigger
551                // claim. See `EntryToPointIdMap::Identity`.
552                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                // On threads, attributes whose streams can be stepped over are
559                // decoded side by side, two at a time on each thread; on one,
560                // and where that cannot be done or goes wrong in any way, every
561                // attribute is decoded in order, two consecutive ones' symbols
562                // decoded together where they pair. On one thread the order is
563                // the faster of the two: the side-by-side decode gives each
564                // attribute a cloud and a buffer of its own, which the decode in
565                // order does not pay.
566                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                    // Symbols decoded beside the attribute before, for the one
596                    // at hand.
597                    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    /// Returns the encoded geometry type handled by this decoder.
708    pub fn get_geometry_type(&self) -> EncodedGeometryType {
709        self.geometry_type
710    }
711}
712
713#[cfg(feature = "point_cloud_decode")]
714impl PointCloudDecoder {
715    /// The symbols of two consecutive attributes' integer streams, decoded
716    /// side by side ([`decode_raw_symbol_pair_in`]), for each attribute's
717    /// decode to take instead of decoding its own. The buffer is left where it
718    /// stood, at the first attribute's stream, which is then decoded as ever.
719    ///
720    /// Where the second stream starts is found by stepping over the first
721    /// without decoding it. `None` where either attribute's stream is not one
722    /// that can be stepped over or paired -- tagged symbols, symbols too wide
723    /// for the paired loop, a prediction other than none or a plain
724    /// difference, a decoder that is not an integer one -- or where anything
725    /// goes wrong; the budget is then put back as it was and the two are
726    /// decoded one after the other, which is what reports the error if there
727    /// is one.
728    ///
729    /// A normal's stream codes its two octahedral coordinates, not the
730    /// attribute's three components.
731    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        // The paired loop runs on steps, which only a table of at most 2^16
750        // slots has: raw symbols of at most 11 bits. Past that each run would
751        // be decoded alone after its table had been read twice, so the two
752        // bytes that say so are read first, from a buffer of their own.
753        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        // `start` is a position this buffer held a moment ago.
784        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    /// Decodes one attribute's values from `buffer`, in place, the way the
796    /// sequential decoder always has.
797    #[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                // The identity map costs nothing to build and is all this
882                // decoder reads off it -- the values are copied verbatim, in
883                // order -- so the arm does not need the shared `point_ids`,
884                // which is `None` when every attribute is generic.
885                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    /// Decodes the attributes of a 2.0 or later sequential stream side by side.
900    ///
901    /// The streams are laid end to end with no table of where each starts, so
902    /// the main thread walks them once, stepping over the ones whose symbols it
903    /// can step over (the raw scheme, which is nearly all of them) and decoding
904    /// the rest in place. Each stretch it steps over becomes a job that decodes
905    /// it as a buffer of its own. A stream that does not come out exactly the
906    /// length it was stepped over as is a stream this did not understand.
907    ///
908    /// Jobs are run two at a time, their symbols decoded in one loop
909    /// (`Job::run_pair`): the rANS run is a chain each symbol waits on, and two
910    /// of them overlap. Each job goes to a worker as soon as the walk reaches
911    /// it, so the walk, and the in-place decoding that is most of it for a
912    /// cloud of a few attributes, goes on while the workers run. Only for more
913    /// than one thread: on one the decode in order, which pairs too, is the
914    /// faster.
915    ///
916    /// `Ok(true)` when every attribute is decoded and `buffer` stands after the
917    /// last of them. `Ok(false)` when anything at all went wrong -- the position
918    /// and the pending lists are put back and the caller decodes in order, which
919    /// is what reports the error if there is one. So the result is the serial
920    /// decode's, value for value, and only the time differs.
921    #[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                    // Two at a time where two are waiting, taken together so
947                    // that no other worker splits them. A worker never waits
948                    // for the second: the walk may be decoding a long stream in
949                    // place, and the first job is better started alone than
950                    // held until the walk moves on.
951                    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            // The walk, on this thread, and the workers beside it, no more of
975            // them than there are attributes to hand out. A failure in the walk
976            // is not returned from inside the scope: the workers have to be let
977            // go of first, and their results, if any, thrown away.
978            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                // Nothing more is coming: the workers run out of jobs and
996                // finish, with this thread working beside them now that its
997                // walk is done -- alone, if the system started none of them.
998                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            // The budget as it stood before the walk, which the decode in
1023            // order that follows charges the walk's own attributes to again.
1024            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        // Every charge, the walk's and the pieces', was checked against what
1046        // all of them had spent, so what is left to check is the count.
1047        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    /// The walk `decode_attributes_in_parallel` makes: each attribute stream
1080    /// stepped over becomes a job handed to `send`, in order, and each one that
1081    /// cannot be is decoded in place. `Ok(Some(jobs))` when the walk reached
1082    /// the end; `Ok(None)` when a stream decoded in place failed or `send`
1083    /// refused a job, which the caller answers by decoding everything in order.
1084    #[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/// One attribute's stream, stepped over, waiting for a thread: where it lies in
1149/// the stream, and the attribute to decode it into.
1150#[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/// What a thread made of a [`Job`].
1161#[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    /// Decodes the stretch of `stream` this job names, as a buffer of its own
1182    /// over an attribute of its own, and returns what it made and what it spent
1183    /// of the allocation budget.
1184    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    /// Runs two jobs, their symbols decoded side by side first
1270    /// (`decode_raw_symbol_pair`) and each job then decoded with its own. Where
1271    /// the pair cannot be decoded together the jobs run as they would alone,
1272    /// so what comes of each is what `run` makes of it.
1273    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    /// Every kind of attribute stream a sequential cloud carries -- a position
1363    /// the coder writes tagged, normals, quantized floats, bytes, raw floats --
1364    /// with noise in all of them, so the stream is as large as the values.
1365    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        // No quantization on a float: the coder copies it as raw bytes.
1417        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        // Attributes that never vary, as a 3DGS file's normals do not: their
1424        // alphabet is one symbol and no payload backs the count, so each run
1425        // is drawn from the allowance the limits admitted, which the threads
1426        // draw on together.
1427        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    /// What a decode makes of `bytes`: every attribute's bytes, or the error.
1449    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    /// The decode in order, attribute by attribute, that the side-by-side one
1462    /// is held to.
1463    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            // One thread decodes in order, pairing as it goes; on the pairs'
1491            // own thread the count is kept, and every attribute is paired but
1492            // the ones whose symbols are too wide or tagged.
1493            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    /// The encoder runs each attribute's encoder on its own thread and appends
1508    /// their buffers in attribute order. That is the stream one buffer written in
1509    /// that order holds, so the bytes may not depend on the number of threads.
1510    #[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    /// A cloud of few attributes, each past a million values, gives each
1523    /// attribute threads of its own for its quantization, gather, prediction
1524    /// and symbol plan. The stream may not depend on how many, with or without
1525    /// the prediction search, which plans twice.
1526    #[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        // Coarse positions, so equal values and both zeros turn up in the
1533        // bounds a quantization folds.
1534        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            // Bytes anywhere, and bytes in the header and the attribute
1586            // descriptions, where the lengths come from.
1587            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        // And cut short.
1603        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    /// Attributes that never vary have runs no payload backs, one symbol's
1615    /// worth of stream for millions of values, and a big cloud has several.
1616    /// The allowance and the budget are one for the whole decode, so what the
1617    /// threads draw together is what one thread would, and such a cloud
1618    /// decodes on threads as it does in order.
1619    #[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    /// Constant attributes past what the allocation backstop holds, beside one
1667    /// that makes the stream long enough for the threads. A constant run is
1668    /// drawn from the values the caller's limits admitted, and the pieces
1669    /// decoded side by side draw on that one allowance as the decode in order
1670    /// does: the threads and the pairing run on every stream, rather than
1671    /// failing on the budget and leaving it all to one thread.
1672    #[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        // A quarter past the ceiling, so no payload the runs carry brings them
1679        // back under it.
1680        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    /// Every synthetic shape, in its own order and shuffled, from no points to
1716    /// past the thresholds that hand a cloud to threads: the same bytes encoded
1717    /// on one thread and on several, the same values decoded in order, paired
1718    /// and side by side, and those values the source's -- to the bit where
1719    /// nothing was quantized, within half a step where it was.
1720    #[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        // Big enough that every stream but the constant cloud's passes
1728        // `PARALLEL_MIN_STREAM_BYTES`, so the threads and the pairing run.
1729        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        // An empty attribute has no range to quantize over, and the encoder
1770        // says so rather than read a value that is not there.
1771        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                // Octahedral coordinates on a grid of `bits` a side: a unit
1817                // vector comes back within a few cells. A zero one, which the
1818                // octahedron has no point for, comes back as some direction.
1819                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            // The quantizer's grid spans the widest component's range.
1838            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    /// Every kind of attribute stream a sequential cloud carries -- a position
1910    /// the coder writes tagged, normals, quantized floats, bytes, raw floats,
1911    /// attributes that never vary -- with noise in the rest, so the stream is
1912    /// as large as the values.
1913    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        // No quantization on a float: the coder copies it as raw bytes.
1954        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    /// What a decode makes of `bytes` -- every attribute's bytes or the error
1982    /// -- and where it left the allocation budget, with the pairs on or off.
1983    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        // On one thread, which decodes in order and is where the pairs run.
1991        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    /// The pairs decode the cloud the one-after-the-other decode does, value
2012    /// for value, and leave the budget where it leaves it. All but the tagged
2013    /// position, the raw float and an odd one out are paired, the normals'
2014    /// two octahedral coordinates included.
2015    #[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            // Bytes anywhere, and bytes in the header and the attribute
2033            // descriptions, where the lengths come from.
2034            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        // And cut short.
2047        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    /// Every synthetic shape, in its own order and shuffled, from no points to
2057    /// past where the decode's run forms pay: the same values and budget
2058    /// decoded with the pairs and without, and those values the source's -- to
2059    /// the bit where nothing was quantized, within half a step where it was.
2060    /// The large clouds must pair somewhere, and `skewed`'s bimodal attribute
2061    /// is decoded through the rANS buckets.
2062    #[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        // An empty attribute has no range to quantize over, and the encoder
2097        // says so rather than read a value that is not there.
2098        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                // Octahedral coordinates on a grid of `bits` a side: a unit
2129                // vector comes back within a few cells. A zero one, which the
2130                // octahedron has no point for, comes back as some direction.
2131                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            // The quantizer's grid spans the widest component's range.
2150            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}