Skip to main content

rustfs_erasure_codec/galois_8/
policy.rs

1#[cfg(feature = "std")]
2use crate::core::RuntimeParallelPolicyCache;
3
4#[cfg(feature = "std")]
5const RECONSTRUCT_DATA_MIN_PARALLEL_SHARD_BYTES: usize = 512 * 1024;
6#[cfg(feature = "std")]
7const RECONSTRUCT_FULL_MIN_PARALLEL_SHARD_BYTES: usize = 256 * 1024;
8#[cfg(feature = "std")]
9const RS_RECONSTRUCT_DATA_MIN_PARALLEL_SHARD_BYTES_ENV: &str =
10    "RS_RECONSTRUCT_DATA_MIN_PARALLEL_SHARD_BYTES";
11#[cfg(feature = "std")]
12const RS_RECONSTRUCT_FULL_MIN_PARALLEL_SHARD_BYTES_ENV: &str =
13    "RS_RECONSTRUCT_FULL_MIN_PARALLEL_SHARD_BYTES";
14#[cfg(feature = "std")]
15const RS_RECONSTRUCT_MIN_BYTES_PER_JOB_ENV: &str = "RS_RECONSTRUCT_MIN_BYTES_PER_JOB";
16#[cfg(all(feature = "std", target_arch = "aarch64"))]
17const RS_AARCH64_RECONSTRUCT_MIN_PARALLEL_SHARD_BYTES_ENV: &str =
18    "RS_AARCH64_RECONSTRUCT_MIN_PARALLEL_SHARD_BYTES";
19#[cfg(all(feature = "std", target_arch = "aarch64"))]
20const RS_AARCH64_RECONSTRUCT_MIN_BYTES_PER_JOB_ENV: &str =
21    "RS_AARCH64_RECONSTRUCT_MIN_BYTES_PER_JOB";
22#[cfg(all(feature = "std", target_arch = "aarch64"))]
23const RS_AARCH64_RECONSTRUCT_MAX_JOBS_ENV: &str = "RS_AARCH64_RECONSTRUCT_MAX_JOBS";
24#[cfg(all(feature = "std", target_arch = "aarch64"))]
25const RS_AARCH64_RECONSTRUCT_DATA_MIN_BYTES_PER_JOB_ENV: &str =
26    "RS_AARCH64_RECONSTRUCT_DATA_MIN_BYTES_PER_JOB";
27#[cfg(all(feature = "std", target_arch = "aarch64"))]
28const RS_AARCH64_RECONSTRUCT_PARITY_MIN_BYTES_PER_JOB_ENV: &str =
29    "RS_AARCH64_RECONSTRUCT_PARITY_MIN_BYTES_PER_JOB";
30
31#[cfg(feature = "std")]
32#[derive(Clone)]
33struct OptionVecReconstructPlan {
34    shard_len: usize,
35    valid_indices: smallvec::SmallVec<[usize; 32]>,
36    invalid_indices: smallvec::SmallVec<[usize; 32]>,
37    number_present: usize,
38    data_decode_matrix: Option<std::sync::Arc<crate::matrix::Matrix<super::Field>>>,
39    required_missing_data_indices: smallvec::SmallVec<[usize; 32]>,
40}
41
42#[cfg(feature = "std")]
43/// Reusable reconstruct workspace for `Option<Vec<u8>>` shard sets.
44///
45/// This workspace caches the shard-size and missing-index planning derived from a
46/// specific `Option<Vec<u8>>` shape so repeated reconstruct calls can skip the
47/// per-call plan build. It is currently intended for repeated serial classic
48/// reconstruct workloads where the missing pattern stays stable across calls.
49pub struct OptionVecReconstructWorkspace(OptionVecReconstructPlan);
50
51impl crate::ReedSolomon<super::Field> {
52    #[cfg(feature = "std")]
53    fn encode_leopard_gf8_opt<T, U>(&self, data: &[T], parity: &mut [U]) -> Result<(), crate::Error>
54    where
55        T: AsRef<[u8]> + Sync,
56        U: AsRef<[u8]> + AsMut<[u8]> + Send,
57    {
58        crate::core::leopard_gf8::encode_with_tables(
59            self.data_shard_count(),
60            self.parity_shard_count(),
61            data,
62            parity,
63        )
64        .map(|_| ())
65    }
66
67    #[cfg(feature = "std")]
68    fn decode_idx_accumulate_reduced_outputs(
69        &self,
70        matrix_rows: &[smallvec::SmallVec<[u8; 32]>],
71        inputs: &[&[u8]],
72        outputs: &mut [&mut [u8]],
73    ) {
74        debug_assert!(!outputs.is_empty());
75
76        let shard_len = inputs.first().map(|input| input.len()).unwrap_or(0);
77        if shard_len == 0 {
78            return;
79        }
80
81        let chunk_len = self.code_chunk_len(shard_len);
82        if outputs.len() == 1 {
83            let matrix_row = matrix_rows[0].as_slice();
84            outputs[0]
85                .chunks_mut(chunk_len)
86                .enumerate()
87                .for_each(|(chunk_idx, output_chunk)| {
88                    let start = chunk_idx * chunk_len;
89                    let end = start + output_chunk.len();
90                    for i_input in 0..inputs.len() {
91                        super::mul_slice_xor(
92                            matrix_row[i_input],
93                            &inputs[i_input][start..end],
94                            output_chunk,
95                        );
96                    }
97                });
98            return;
99        }
100
101        if outputs.len() == 2 {
102            let (first, second) = outputs.split_at_mut(1);
103            let output0 = &mut first[0];
104            let output1 = &mut second[0];
105            let row0 = matrix_rows[0].as_slice();
106            let row1 = matrix_rows[1].as_slice();
107            output0
108                .chunks_mut(chunk_len)
109                .zip(output1.chunks_mut(chunk_len))
110                .enumerate()
111                .for_each(|(chunk_idx, (output0_chunk, output1_chunk))| {
112                    let start = chunk_idx * chunk_len;
113                    let end = start + output0_chunk.len();
114                    let input0 = &inputs[0][start..end];
115                    super::mul_slice_xor(row0[0], input0, output0_chunk);
116                    super::mul_slice_xor(row1[0], input0, output1_chunk);
117                    for i_input in 1..inputs.len() {
118                        let input_chunk = &inputs[i_input][start..end];
119                        super::mul_slice_xor(row0[i_input], input_chunk, output0_chunk);
120                        super::mul_slice_xor(row1[i_input], input_chunk, output1_chunk);
121                    }
122                });
123            return;
124        }
125
126        for (row, output) in matrix_rows.iter().zip(outputs.iter_mut()) {
127            output
128                .chunks_mut(chunk_len)
129                .enumerate()
130                .for_each(|(chunk_idx, output_chunk)| {
131                    let start = chunk_idx * chunk_len;
132                    let end = start + output_chunk.len();
133                    for (coefficient, input) in row.iter().copied().zip(inputs.iter().copied()) {
134                        super::mul_slice_xor(coefficient, &input[start..end], output_chunk);
135                    }
136                });
137        }
138    }
139
140    #[cfg(feature = "std")]
141    fn execute_option_vec_required_data_plan(
142        &self,
143        shards: &mut [Option<Vec<u8>>],
144        plan: OptionVecReconstructPlan,
145    ) -> Result<(), crate::Error> {
146        if plan.required_missing_data_indices.is_empty() {
147            return Ok(());
148        }
149
150        let data_decode_matrix = plan
151            .data_decode_matrix
152            .as_ref()
153            .expect("non-empty plan must include decode matrix");
154        let mut matrix_rows: smallvec::SmallVec<[&[u8]; 32]> =
155            smallvec::SmallVec::with_capacity(plan.required_missing_data_indices.len());
156        for &idx in &plan.required_missing_data_indices {
157            matrix_rows.push(data_decode_matrix.get_row(idx));
158        }
159
160        let mut recovered_data: Vec<Vec<u8>> = plan
161            .required_missing_data_indices
162            .iter()
163            .map(|_| vec![0u8; plan.shard_len])
164            .collect();
165        {
166            let sub_shards_snapshot: Vec<&[u8]> = plan
167                .valid_indices
168                .iter()
169                .map(|&idx| {
170                    shards[idx]
171                        .as_deref()
172                        .expect("valid shard index must be present")
173                })
174                .collect();
175            let sub_shards: smallvec::SmallVec<[&[u8]; 32]> =
176                sub_shards_snapshot.into_iter().collect();
177
178            let mut outputs: smallvec::SmallVec<[&mut [u8]; 32]> = recovered_data
179                .iter_mut()
180                .map(|shard| shard.as_mut_slice())
181                .collect();
182            let use_parallel = self
183                .parallel_policy(plan.shard_len, plan.required_missing_data_indices.len())
184                .use_parallel;
185            if use_parallel {
186                self.code_some_slices_par_raw(&matrix_rows, &sub_shards, &mut outputs);
187            } else {
188                self.code_some_slices_chunked(&matrix_rows, &sub_shards, &mut outputs);
189            }
190        }
191
192        for (idx, recovered) in plan
193            .required_missing_data_indices
194            .into_iter()
195            .zip(recovered_data)
196        {
197            shards[idx] = Some(recovered);
198        }
199
200        Ok(())
201    }
202
203    #[cfg(feature = "std")]
204    fn execute_option_vec_reconstruct_plan_serial(
205        &self,
206        shards: &mut [Option<Vec<u8>>],
207        plan: &OptionVecReconstructPlan,
208        data_only: bool,
209    ) -> Result<(), crate::Error> {
210        let data_decode_matrix = plan
211            .data_decode_matrix
212            .as_ref()
213            .expect("non-empty reconstruct plan must include decode matrix");
214
215        let mut missing_data_indices: smallvec::SmallVec<[usize; 32]> = smallvec::SmallVec::new();
216        let mut missing_parity_indices: smallvec::SmallVec<[usize; 32]> = smallvec::SmallVec::new();
217        for &idx in &plan.invalid_indices {
218            if idx < self.data_shard_count() {
219                missing_data_indices.push(idx);
220            } else if !data_only {
221                missing_parity_indices.push(idx);
222            }
223        }
224
225        self.record_reconstruct_runtime(
226            data_only,
227            missing_data_indices.len(),
228            missing_parity_indices.len(),
229            false,
230        );
231
232        if !missing_data_indices.is_empty() {
233            self.record_reconstruct_data_stage_runtime(plan.shard_len, missing_data_indices.len());
234
235            let mut matrix_rows: smallvec::SmallVec<[&[u8]; 32]> =
236                smallvec::SmallVec::with_capacity(missing_data_indices.len());
237            for &idx in &missing_data_indices {
238                matrix_rows.push(data_decode_matrix.get_row(idx));
239            }
240
241            {
242                let sub_shard_ptrs: smallvec::SmallVec<[(*const u8, usize); 32]> = plan
243                    .valid_indices
244                    .iter()
245                    .map(|&idx| {
246                        let shard = shards[idx]
247                            .as_deref()
248                            .expect("valid shard index must be present");
249                        (shard.as_ptr(), shard.len())
250                    })
251                    .collect();
252                let sub_shards: smallvec::SmallVec<[&[u8]; 32]> = sub_shard_ptrs
253                    .iter()
254                    .map(|&(ptr, len)| {
255                        // SAFETY: each pointer/len pair comes from a present shard
256                        // and remains valid for the duration of this compute phase.
257                        unsafe { core::slice::from_raw_parts(ptr, len) }
258                    })
259                    .collect();
260                let mut output_ptrs: smallvec::SmallVec<[(*mut u8, usize); 32]> =
261                    smallvec::SmallVec::with_capacity(missing_data_indices.len());
262                for &idx in &missing_data_indices {
263                    let shard = shards[idx]
264                        .get_or_insert_with(|| vec![0u8; plan.shard_len])
265                        .as_mut_slice();
266                    output_ptrs.push((shard.as_mut_ptr(), shard.len()));
267                }
268                let mut outputs: smallvec::SmallVec<[&mut [u8]; 32]> = output_ptrs
269                    .iter()
270                    .map(|&(ptr, len)| {
271                        // SAFETY: each pointer/len pair was taken from a distinct
272                        // missing shard slot initialized to `plan.shard_len`.
273                        unsafe { core::slice::from_raw_parts_mut(ptr, len) }
274                    })
275                    .collect();
276                self.code_some_slices_chunked(&matrix_rows, &sub_shards, &mut outputs);
277            }
278        }
279
280        if data_only || missing_parity_indices.is_empty() {
281            return Ok(());
282        }
283
284        self.record_reconstruct_parity_stage_runtime(plan.shard_len, missing_parity_indices.len());
285
286        let parity_rows = self.get_parity_rows();
287        let mut matrix_rows: smallvec::SmallVec<[&[u8]; 32]> =
288            smallvec::SmallVec::with_capacity(missing_parity_indices.len());
289        for &idx in &missing_parity_indices {
290            matrix_rows.push(parity_rows[idx - self.data_shard_count()]);
291        }
292
293        {
294            let all_data_ptrs: smallvec::SmallVec<[(*const u8, usize); 32]> = shards
295                .iter()
296                .take(self.data_shard_count())
297                .map(|shard| {
298                    let data = shard.as_deref().expect("data shard must be present");
299                    (data.as_ptr(), data.len())
300                })
301                .collect();
302            let all_data: smallvec::SmallVec<[&[u8]; 32]> = all_data_ptrs
303                .iter()
304                .map(|&(ptr, len)| {
305                    // SAFETY: each pointer/len pair comes from a present data shard
306                    // and remains valid for the duration of this compute phase.
307                    unsafe { core::slice::from_raw_parts(ptr, len) }
308                })
309                .collect();
310            let mut output_ptrs: smallvec::SmallVec<[(*mut u8, usize); 32]> =
311                smallvec::SmallVec::with_capacity(missing_parity_indices.len());
312            for &idx in &missing_parity_indices {
313                let shard = shards[idx]
314                    .get_or_insert_with(|| vec![0u8; plan.shard_len])
315                    .as_mut_slice();
316                output_ptrs.push((shard.as_mut_ptr(), shard.len()));
317            }
318            let mut outputs: smallvec::SmallVec<[&mut [u8]; 32]> = output_ptrs
319                .iter()
320                .map(|&(ptr, len)| {
321                    // SAFETY: each pointer/len pair was taken from a distinct
322                    // missing parity shard slot initialized to `plan.shard_len`.
323                    unsafe { core::slice::from_raw_parts_mut(ptr, len) }
324                })
325                .collect();
326            self.code_some_slices_chunked(&matrix_rows, &all_data, &mut outputs);
327        }
328
329        Ok(())
330    }
331
332    #[cfg(feature = "std")]
333    fn plan_option_vec_reconstruct(
334        &self,
335        shards: &[Option<Vec<u8>>],
336        required: Option<&[bool]>,
337    ) -> Result<OptionVecReconstructPlan, crate::Error> {
338        let mut number_present = 0;
339        let mut shard_len = None;
340        for shard in shards.iter() {
341            if let Some(shard) = shard.as_ref() {
342                if shard.is_empty() {
343                    return Err(crate::Error::EmptyShard);
344                }
345                number_present += 1;
346                if let Some(old_len) = shard_len
347                    && shard.len() != old_len
348                {
349                    return Err(crate::Error::IncorrectShardSize);
350                }
351                shard_len = Some(shard.len());
352            }
353        }
354
355        if number_present == self.total_shard_count() {
356            return Ok(OptionVecReconstructPlan {
357                shard_len: 0,
358                valid_indices: smallvec::SmallVec::new(),
359                invalid_indices: smallvec::SmallVec::new(),
360                number_present,
361                data_decode_matrix: None,
362                required_missing_data_indices: smallvec::SmallVec::new(),
363            });
364        }
365        if number_present < self.data_shard_count() {
366            return Err(crate::Error::TooFewShardsPresent);
367        }
368
369        let shard_len = shard_len.expect("at least one shard present; qed");
370        let mut valid_indices =
371            smallvec::SmallVec::<[usize; 32]>::with_capacity(self.data_shard_count());
372        let mut invalid_indices =
373            smallvec::SmallVec::<[usize; 32]>::with_capacity(self.total_shard_count());
374        for (idx, shard) in shards.iter().enumerate() {
375            if shard.is_some() {
376                if valid_indices.len() < self.data_shard_count() {
377                    valid_indices.push(idx);
378                }
379            } else {
380                invalid_indices.push(idx);
381            }
382        }
383
384        let data_decode_matrix = self.get_data_decode_matrix(&valid_indices, &invalid_indices)?;
385        let required_missing_data_indices = required
386            .map(|required| {
387                (0..self.data_shard_count())
388                    .filter(|&i| required[i] && shards[i].is_none())
389                    .collect()
390            })
391            .unwrap_or_default();
392
393        Ok(OptionVecReconstructPlan {
394            shard_len,
395            valid_indices,
396            invalid_indices,
397            number_present,
398            data_decode_matrix: Some(data_decode_matrix),
399            required_missing_data_indices,
400        })
401    }
402
403    #[cfg(feature = "std")]
404    fn first_shard_len<T: AsRef<[u8]>>(slices: &[T]) -> usize {
405        slices
406            .first()
407            .map(|shard| shard.as_ref().len())
408            .unwrap_or(0)
409    }
410
411    #[cfg(feature = "std")]
412    fn first_present_shard_len(shards: &[Option<Vec<u8>>]) -> usize {
413        shards
414            .iter()
415            .find_map(|shard| shard.as_ref().map(|shard| shard.len()))
416            .unwrap_or(0)
417    }
418
419    #[cfg(feature = "std")]
420    fn summarize_option_vec_reconstruct_shape(
421        &self,
422        shards: &[Option<Vec<u8>>],
423    ) -> Result<(usize, usize, usize), crate::Error> {
424        let mut number_present = 0usize;
425        let mut shard_len = None;
426        let mut missing_total = 0usize;
427        let mut missing_data = 0usize;
428
429        for (idx, shard) in shards.iter().enumerate() {
430            if let Some(shard) = shard.as_ref() {
431                if shard.is_empty() {
432                    return Err(crate::Error::EmptyShard);
433                }
434                number_present += 1;
435                if let Some(old_len) = shard_len
436                    && shard.len() != old_len
437                {
438                    return Err(crate::Error::IncorrectShardSize);
439                }
440                shard_len = Some(shard.len());
441            } else {
442                missing_total += 1;
443                if idx < self.data_shard_count() {
444                    missing_data += 1;
445                }
446            }
447        }
448
449        if missing_total == 0 {
450            return Ok((0, 0, 0));
451        }
452        if number_present < self.data_shard_count() {
453            return Err(crate::Error::TooFewShardsPresent);
454        }
455
456        Ok((shard_len.unwrap_or(0), missing_data, missing_total))
457    }
458
459    #[cfg(feature = "std")]
460    fn should_parallel_data_path(&self, shard_len: usize, output_shards: usize) -> bool {
461        self.parallel_policy(shard_len, output_shards).use_parallel
462    }
463
464    #[cfg(feature = "std")]
465    pub(crate) fn reconstruct_parallel_decision_with(
466        &self,
467        shard_len: usize,
468        missing_data: usize,
469        missing_total: usize,
470        data_only: bool,
471        available_parallelism: usize,
472    ) -> crate::ParallelDecision {
473        let output_shards = if data_only {
474            missing_data
475        } else {
476            missing_total
477        };
478        let tuned = self.policy_cache.reconstruct_policy(data_only);
479        tuned.decide(
480            shard_len,
481            self.data_shard_count(),
482            output_shards,
483            available_parallelism,
484        )
485    }
486
487    #[cfg(feature = "std")]
488    fn reconstruct_parallel_decision(
489        &self,
490        shard_len: usize,
491        missing_data: usize,
492        missing_total: usize,
493        data_only: bool,
494    ) -> crate::ParallelDecision {
495        self.reconstruct_parallel_decision_with(
496            shard_len,
497            missing_data,
498            missing_total,
499            data_only,
500            self.policy_cache.available_parallelism,
501        )
502    }
503
504    #[cfg(feature = "std")]
505    fn reconstruct_stage_policies(
506        &self,
507        data_only: bool,
508    ) -> (crate::ParallelPolicy, crate::ParallelPolicy) {
509        self.policy_cache.reconstruct_stage_policies(data_only)
510    }
511
512    #[cfg(feature = "std")]
513    #[doc(hidden)]
514    pub fn reconstruct_stage_policies_for_bench(
515        &self,
516        data_only: bool,
517    ) -> (crate::ParallelPolicy, crate::ParallelPolicy) {
518        self.reconstruct_stage_policies(data_only)
519    }
520
521    #[cfg(feature = "std")]
522    #[doc(hidden)]
523    pub fn reconstruct_execution_context_for_bench(
524        &self,
525        shard_len: usize,
526        missing_data: usize,
527        missing_total: usize,
528        data_only: bool,
529        available_parallelism: usize,
530    ) -> (
531        crate::ParallelDecision,
532        crate::ParallelPolicy,
533        crate::ParallelPolicy,
534    ) {
535        let decision = self.reconstruct_parallel_decision_with(
536            shard_len,
537            missing_data,
538            missing_total,
539            data_only,
540            available_parallelism,
541        );
542        let (data_policy, parity_policy) = self.reconstruct_stage_policies(data_only);
543        (decision, data_policy, parity_policy)
544    }
545
546    #[cfg(feature = "std")]
547    #[doc(hidden)]
548    pub fn plan_option_vec_reconstruct_for_bench(
549        &self,
550        shards: &[Option<Vec<u8>>],
551        required: Option<&[bool]>,
552    ) -> Result<(usize, usize, usize, usize), crate::Error> {
553        let plan = self.plan_option_vec_reconstruct(shards, required)?;
554        Ok((
555            plan.shard_len,
556            plan.number_present,
557            plan.valid_indices.len(),
558            plan.invalid_indices.len(),
559        ))
560    }
561
562    #[cfg(feature = "std")]
563    /// Prepare a reusable reconstruct workspace for an `Option<Vec<u8>>` shard layout.
564    ///
565    /// The workspace is only valid for subsequent calls that keep the same
566    /// missing-shard pattern and shard length.
567    pub fn prepare_reconstruct_opt_workspace(
568        &self,
569        shards: &[Option<Vec<u8>>],
570    ) -> Result<OptionVecReconstructWorkspace, crate::Error> {
571        Ok(OptionVecReconstructWorkspace(
572            self.plan_option_vec_reconstruct(shards, None)?,
573        ))
574    }
575
576    #[cfg(feature = "std")]
577    /// Execute reconstruct using a previously prepared workspace.
578    ///
579    /// This skips the per-call option-vec planning pass, but the caller must
580    /// provide shards with the same missing-index layout and shard length as the
581    /// one used to prepare `workspace`.
582    pub fn reconstruct_opt_with_workspace(
583        &self,
584        shards: &mut [Option<Vec<u8>>],
585        workspace: &OptionVecReconstructWorkspace,
586    ) -> Result<(), crate::Error> {
587        if workspace.0.shard_len == 0 {
588            return Ok(());
589        }
590
591        let mut invalid_indices = smallvec::SmallVec::<[usize; 32]>::new();
592        let mut shard_len = None;
593        let mut number_present = 0usize;
594        for (idx, shard) in shards.iter().enumerate() {
595            if let Some(shard) = shard.as_ref() {
596                if shard.is_empty() {
597                    return Err(crate::Error::EmptyShard);
598                }
599                number_present += 1;
600                if let Some(old_len) = shard_len
601                    && shard.len() != old_len
602                {
603                    return Err(crate::Error::IncorrectShardSize);
604                }
605                shard_len = Some(shard.len());
606            } else {
607                invalid_indices.push(idx);
608            }
609        }
610
611        if invalid_indices != workspace.0.invalid_indices {
612            return Err(crate::Error::InvalidShardFlags);
613        }
614        if number_present != workspace.0.number_present {
615            return Err(crate::Error::TooFewShardsPresent);
616        }
617        if shard_len.unwrap_or(0) != workspace.0.shard_len {
618            return Err(crate::Error::IncorrectShardSize);
619        }
620
621        self.execute_option_vec_reconstruct_plan_serial(shards, &workspace.0, false)
622    }
623
624    #[cfg(feature = "std")]
625    #[doc(hidden)]
626    pub fn execute_option_vec_reconstruct_plan_serial_for_bench(
627        &self,
628        shards: &mut [Option<Vec<u8>>],
629        data_only: bool,
630    ) -> Result<(), crate::Error> {
631        let plan = self.plan_option_vec_reconstruct(shards, None)?;
632        self.execute_option_vec_reconstruct_plan_serial(shards, &plan, data_only)
633    }
634
635    #[cfg(feature = "std")]
636    pub fn encode_opt<T, U>(&self, mut shards: T) -> Result<(), crate::Error>
637    where
638        T: AsRef<[U]> + AsMut<[U]>,
639        U: AsRef<[u8]> + AsMut<[u8]> + Send + Sync,
640    {
641        if self.is_leopard_gf8_family() {
642            let slices = shards.as_mut();
643            if slices.len() != self.total_shard_count() {
644                return Err(crate::Error::TooFewShards);
645            }
646            if slices.is_empty() {
647                return Err(crate::Error::TooFewShards);
648            }
649            let shard_len = slices[0].as_ref().len();
650            if shard_len == 0 {
651                return Err(crate::Error::EmptyShard);
652            }
653            for shard in slices.iter() {
654                if shard.as_ref().len() != shard_len {
655                    return Err(crate::Error::IncorrectShardSize);
656                }
657            }
658            let (data, parity) = slices.split_at_mut(self.data_shard_count());
659            return self.encode_leopard_gf8_opt(&*data, parity);
660        }
661
662        let shard_len = Self::first_shard_len(shards.as_ref());
663        if self.should_parallel_data_path(shard_len, self.parity_shard_count()) {
664            self.encode_par(shards)
665        } else {
666            self.encode(shards)
667        }
668    }
669
670    #[cfg(feature = "std")]
671    pub fn encode_sep_opt<T, U>(&self, data: &[T], parity: &mut [U]) -> Result<(), crate::Error>
672    where
673        T: AsRef<[u8]> + Sync,
674        U: AsRef<[u8]> + AsMut<[u8]> + Send,
675    {
676        if self.is_leopard_gf8_family() {
677            return self.encode_leopard_gf8_opt(data, parity);
678        }
679
680        let shard_len = Self::first_shard_len(data);
681        if self.should_parallel_data_path(shard_len, parity.len()) {
682            self.encode_sep_par(data, parity)
683        } else {
684            self.encode_sep(data, parity)
685        }
686    }
687
688    #[cfg(feature = "std")]
689    pub fn verify_opt<T>(&self, slices: &[T]) -> Result<bool, crate::Error>
690    where
691        T: AsRef<[u8]> + Sync,
692    {
693        self.ensure_classic_family_execution()?;
694        if self.is_leopard_gf8_family() {
695            return self.verify(slices);
696        }
697        let shard_len = Self::first_shard_len(slices);
698        if self.should_parallel_data_path(shard_len, self.parity_shard_count()) {
699            self.verify_par(slices)
700        } else {
701            self.verify(slices)
702        }
703    }
704
705    #[cfg(feature = "std")]
706    pub fn verify_with_buffer_opt<T, U>(
707        &self,
708        slices: &[T],
709        buffer: &mut [U],
710    ) -> Result<bool, crate::Error>
711    where
712        T: AsRef<[u8]> + Sync,
713        U: AsRef<[u8]> + AsMut<[u8]> + Send,
714    {
715        self.ensure_classic_family_execution()?;
716        if self.is_leopard_gf8_family() {
717            return self.verify_with_buffer(slices, buffer);
718        }
719        let shard_len = Self::first_shard_len(slices);
720        if self.should_parallel_data_path(shard_len, buffer.len()) {
721            self.verify_with_buffer_par(slices, buffer)
722        } else {
723            self.verify_with_buffer(slices, buffer)
724        }
725    }
726
727    #[cfg(feature = "std")]
728    pub fn verify_with_workspace_opt(
729        &self,
730        slices: &[Vec<u8>],
731        workspace: &mut crate::VerifyWorkspace<crate::galois_8::Field>,
732    ) -> Result<bool, crate::Error> {
733        self.ensure_classic_family_execution()?;
734        if self.is_leopard_gf8_family() {
735            return self.verify_with_workspace(slices, workspace);
736        }
737        let shard_len = Self::first_shard_len(slices);
738        workspace.resize(self, shard_len);
739        if self.should_parallel_data_path(shard_len, self.parity_shard_count()) {
740            self.verify_with_buffer_par(slices, workspace.as_mut_shards())
741        } else {
742            self.verify_with_buffer(slices, workspace.as_mut_shards())
743        }
744    }
745
746    #[cfg(feature = "std")]
747    pub fn reconstruct_opt(&self, shards: &mut [Option<Vec<u8>>]) -> Result<(), crate::Error> {
748        if self.is_leopard_gf8_family() {
749            return self.reconstruct(shards);
750        }
751        let (shard_len, missing_data, missing) =
752            self.summarize_option_vec_reconstruct_shape(shards)?;
753        if missing == 0 {
754            return Ok(());
755        }
756        let decision = self.reconstruct_parallel_decision(shard_len, missing_data, missing, false);
757        self.record_reconstruct_entry_path(decision.use_parallel);
758        if decision.use_parallel {
759            let (data_policy, parity_policy) = self.reconstruct_stage_policies(false);
760            self.reconstruct_internal_option_vec_par_with_stage_policies(
761                shards,
762                false,
763                data_policy,
764                parity_policy,
765            )
766        } else {
767            self.record_reconstruct_opt_fallback_serial_path();
768            self.reconstruct(shards)
769        }
770    }
771
772    #[cfg(feature = "std")]
773    pub fn reconstruct_data_opt(&self, shards: &mut [Option<Vec<u8>>]) -> Result<(), crate::Error> {
774        if self.is_leopard_gf8_family() {
775            return self.reconstruct_data(shards);
776        }
777        let (shard_len, missing_data, missing) =
778            self.summarize_option_vec_reconstruct_shape(shards)?;
779        if missing == 0 {
780            return Ok(());
781        }
782        let decision = self.reconstruct_parallel_decision(shard_len, missing_data, missing, true);
783        self.record_reconstruct_entry_path(decision.use_parallel);
784        if decision.use_parallel {
785            let (data_policy, _parity_policy) = self.reconstruct_stage_policies(true);
786            self.reconstruct_internal_option_vec_par_with_policy(shards, true, data_policy)
787        } else {
788            self.reconstruct_data(shards)
789        }
790    }
791
792    #[cfg(feature = "std")]
793    pub fn reconstruct_some_opt(
794        &self,
795        shards: &mut [Option<Vec<u8>>],
796        required: &[bool],
797    ) -> Result<(), crate::Error> {
798        if self.is_leopard_gf8_family() {
799            return self.reconstruct_some(shards, required);
800        }
801        if required.len() != self.total_shard_count() {
802            return Err(crate::Error::InvalidShardFlags);
803        }
804
805        let data_only = required
806            .iter()
807            .enumerate()
808            .all(|(idx, required)| !*required || idx < self.data_shard_count());
809
810        if data_only {
811            let plan = self.plan_option_vec_reconstruct(shards, Some(required))?;
812            if plan.number_present == self.total_shard_count() {
813                return Ok(());
814            }
815            return self.execute_option_vec_required_data_plan(shards, plan);
816        }
817
818        self.reconstruct_opt(shards)?;
819        Ok(())
820    }
821
822    #[cfg(feature = "std")]
823    #[allow(clippy::needless_range_loop)]
824    pub fn decode_idx(
825        &self,
826        dst: &mut [Option<Vec<u8>>],
827        expect_input: Option<&[bool]>,
828        input: &[Option<Vec<u8>>],
829    ) -> Result<(), crate::Error> {
830        if self.is_leopard_gf8_family() {
831            return Err(crate::Error::UnsupportedCodecFamily);
832        }
833        self.ensure_classic_family_execution()?;
834        if dst.len() != self.total_shard_count() || input.len() != self.total_shard_count() {
835            return Err(crate::Error::TooFewShards);
836        }
837
838        if let Some(expect_input) = expect_input {
839            if expect_input.len() != self.total_shard_count() {
840                return Err(crate::Error::InvalidShardFlags);
841            }
842
843            let mut valid_indices =
844                smallvec::SmallVec::<[usize; 32]>::with_capacity(self.data_shard_count());
845            let mut invalid_indices =
846                smallvec::SmallVec::<[usize; 32]>::with_capacity(self.total_shard_count());
847
848            for (idx, expected) in expect_input.iter().copied().enumerate() {
849                if expected {
850                    valid_indices.push(idx);
851                } else {
852                    invalid_indices.push(idx);
853                }
854            }
855
856            if valid_indices.len() < self.data_shard_count() {
857                return Err(crate::Error::TooFewShardsPresent);
858            }
859
860            let shard_len = input
861                .iter()
862                .chain(dst.iter())
863                .find_map(|shard| shard.as_ref().map(|shard| shard.len()))
864                .ok_or(crate::Error::TooFewShardsPresent)?;
865
866            for shard in input.iter().flatten() {
867                if shard.len() != shard_len {
868                    return Err(crate::Error::IncorrectShardSize);
869                }
870            }
871            for shard in dst.iter().flatten() {
872                if shard.len() != shard_len {
873                    return Err(crate::Error::IncorrectShardSize);
874                }
875            }
876
877            let mut output_indices: smallvec::SmallVec<[usize; 32]> = smallvec::SmallVec::new();
878
879            for (idx, shard) in dst.iter().enumerate() {
880                let Some(_dst_shard) = shard.as_ref() else {
881                    continue;
882                };
883                output_indices.push(idx);
884            }
885
886            if output_indices.is_empty() {
887                return Ok(());
888            }
889
890            let mut input_positions = smallvec::SmallVec::<[usize; 32]>::new();
891            let mut input_refs = smallvec::SmallVec::<[&[u8]; 32]>::new();
892            for (col, &idx) in valid_indices
893                .iter()
894                .take(self.data_shard_count())
895                .enumerate()
896            {
897                if let Some(shard) = input[idx].as_deref() {
898                    input_positions.push(col);
899                    input_refs.push(shard);
900                }
901            }
902
903            if input_refs.is_empty() {
904                return Ok(());
905            }
906
907            let data_decode_matrix =
908                self.get_data_decode_matrix(&valid_indices, &invalid_indices)?;
909            let parity_rows = self.get_parity_rows();
910            let mut reduced_rows: smallvec::SmallVec<[smallvec::SmallVec<[u8; 32]>; 32]> =
911                smallvec::SmallVec::with_capacity(output_indices.len());
912            for &idx in &output_indices {
913                let mut row = smallvec::SmallVec::<[u8; 32]>::with_capacity(input_positions.len());
914                if idx < self.data_shard_count() {
915                    for &col in &input_positions {
916                        row.push(data_decode_matrix.get(idx, col));
917                    }
918                } else {
919                    let parity_row = parity_rows[idx - self.data_shard_count()];
920                    for &col in &input_positions {
921                        let mut acc = 0u8;
922                        for i in 0..self.data_shard_count() {
923                            acc ^= super::mul(parity_row[i], data_decode_matrix.get(i, col));
924                        }
925                        row.push(acc);
926                    }
927                }
928                reduced_rows.push(row);
929            }
930
931            let mut output_ptrs: smallvec::SmallVec<[(*mut u8, usize); 32]> =
932                smallvec::SmallVec::with_capacity(output_indices.len());
933            for &idx in &output_indices {
934                let dst_shard = dst[idx]
935                    .as_deref_mut()
936                    .expect("output index was collected only for present destinations");
937                if dst_shard.len() != shard_len {
938                    return Err(crate::Error::IncorrectShardSize);
939                }
940                output_ptrs.push((dst_shard.as_mut_ptr(), dst_shard.len()));
941            }
942
943            {
944                let mut output_refs: smallvec::SmallVec<[&mut [u8]; 32]> = output_ptrs
945                    .iter()
946                    .map(|&(ptr, len)| {
947                        // SAFETY: each pointer/length pair was captured from a distinct
948                        // destination shard collected in `output_indices`. Those shards
949                        // remain alive for this scope and are only mutated through these
950                        // temporary slices.
951                        unsafe { core::slice::from_raw_parts_mut(ptr, len) }
952                    })
953                    .collect();
954
955                self.decode_idx_accumulate_reduced_outputs(
956                    &reduced_rows,
957                    &input_refs,
958                    &mut output_refs,
959                );
960            }
961
962            return Ok(());
963        }
964
965        for (dst_shard, input_shard) in dst.iter_mut().zip(input.iter()) {
966            match (dst_shard.as_deref_mut(), input_shard.as_deref()) {
967                (Some(dst), Some(input)) => {
968                    if dst.len() != input.len() {
969                        return Err(crate::Error::IncorrectShardSize);
970                    }
971                    for (dst_byte, input_byte) in dst.iter_mut().zip(input.iter()) {
972                        *dst_byte ^= *input_byte;
973                    }
974                }
975                (None, Some(_)) => return Err(crate::Error::TooFewShards),
976                _ => {}
977            }
978        }
979
980        Ok(())
981    }
982}
983
984#[cfg(feature = "std")]
985fn reconstruct_parallel_policy_default(
986    base: crate::ParallelPolicy,
987    data_only: bool,
988) -> crate::ParallelPolicy {
989    let data_only_min = parse_positive_env_usize(RS_RECONSTRUCT_DATA_MIN_PARALLEL_SHARD_BYTES_ENV)
990        .unwrap_or(RECONSTRUCT_DATA_MIN_PARALLEL_SHARD_BYTES);
991    let full_min = parse_positive_env_usize(RS_RECONSTRUCT_FULL_MIN_PARALLEL_SHARD_BYTES_ENV)
992        .unwrap_or(RECONSTRUCT_FULL_MIN_PARALLEL_SHARD_BYTES);
993    let min_bytes_per_job = parse_positive_env_usize(RS_RECONSTRUCT_MIN_BYTES_PER_JOB_ENV)
994        .unwrap_or(base.min_bytes_per_job);
995    if data_only {
996        crate::ParallelPolicy {
997            min_parallel_shard_bytes: core::cmp::max(base.min_parallel_shard_bytes, data_only_min),
998            min_bytes_per_job,
999            max_jobs: base.max_jobs,
1000            l2_cache_bytes: base.l2_cache_bytes,
1001        }
1002    } else {
1003        crate::ParallelPolicy {
1004            min_parallel_shard_bytes: core::cmp::max(base.min_parallel_shard_bytes / 2, full_min),
1005            min_bytes_per_job,
1006            max_jobs: base.max_jobs,
1007            l2_cache_bytes: base.l2_cache_bytes,
1008        }
1009    }
1010}
1011
1012#[cfg(feature = "std")]
1013fn parse_positive_env_usize(name: &str) -> Option<usize> {
1014    std::env::var(name)
1015        .ok()
1016        .and_then(|value| value.parse::<usize>().ok())
1017        .filter(|value| *value > 0)
1018}
1019
1020#[cfg(all(feature = "std", target_arch = "aarch64"))]
1021fn reconstruct_policy_cache_aarch64(base: crate::ParallelPolicy) -> RuntimeParallelPolicyCache {
1022    let mut reconstruct_full_data = reconstruct_parallel_policy_default(base, false);
1023    if let Some(value) =
1024        parse_positive_env_usize(RS_AARCH64_RECONSTRUCT_MIN_PARALLEL_SHARD_BYTES_ENV)
1025    {
1026        reconstruct_full_data.min_parallel_shard_bytes = value;
1027    }
1028    if let Some(value) = parse_positive_env_usize(RS_AARCH64_RECONSTRUCT_MIN_BYTES_PER_JOB_ENV) {
1029        reconstruct_full_data.min_bytes_per_job = value;
1030    }
1031    if let Some(value) = std::env::var(RS_AARCH64_RECONSTRUCT_MAX_JOBS_ENV)
1032        .ok()
1033        .and_then(|value| value.parse::<usize>().ok())
1034    {
1035        reconstruct_full_data.max_jobs = value;
1036    }
1037
1038    let mut reconstruct_data = reconstruct_parallel_policy_default(base, true);
1039    reconstruct_data.min_parallel_shard_bytes = reconstruct_full_data.min_parallel_shard_bytes;
1040    reconstruct_data.min_bytes_per_job = reconstruct_full_data.min_bytes_per_job;
1041    reconstruct_data.max_jobs = reconstruct_full_data.max_jobs;
1042
1043    let mut reconstruct_full_parity = reconstruct_parallel_policy_default(base, false);
1044    if let Some(value) =
1045        parse_positive_env_usize(RS_AARCH64_RECONSTRUCT_PARITY_MIN_BYTES_PER_JOB_ENV)
1046    {
1047        reconstruct_full_parity.min_bytes_per_job = value;
1048    }
1049
1050    if let Some(value) = parse_positive_env_usize(RS_AARCH64_RECONSTRUCT_DATA_MIN_BYTES_PER_JOB_ENV)
1051    {
1052        reconstruct_data.min_bytes_per_job = value;
1053    }
1054
1055    RuntimeParallelPolicyCache {
1056        available_parallelism: std::thread::available_parallelism()
1057            .map(|parallelism| parallelism.get())
1058            .unwrap_or(1),
1059        data: base,
1060        reconstruct_data,
1061        reconstruct_full_data,
1062        reconstruct_full_parity,
1063    }
1064}
1065
1066#[cfg(all(feature = "std", not(target_arch = "aarch64")))]
1067pub(crate) fn resolve_runtime_parallel_policy_cache(
1068    base: crate::ParallelPolicy,
1069) -> RuntimeParallelPolicyCache {
1070    let reconstruct_data = reconstruct_parallel_policy_default(base, true);
1071    let reconstruct_full = reconstruct_parallel_policy_default(base, false);
1072    RuntimeParallelPolicyCache {
1073        available_parallelism: std::thread::available_parallelism()
1074            .map(|parallelism| parallelism.get())
1075            .unwrap_or(1),
1076        data: base,
1077        reconstruct_data,
1078        reconstruct_full_data: reconstruct_full,
1079        reconstruct_full_parity: reconstruct_full,
1080    }
1081}
1082
1083#[cfg(all(feature = "std", target_arch = "aarch64"))]
1084pub(crate) fn resolve_runtime_parallel_policy_cache(
1085    base: crate::ParallelPolicy,
1086) -> RuntimeParallelPolicyCache {
1087    reconstruct_policy_cache_aarch64(base)
1088}