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")]
43pub 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 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 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 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 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 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 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 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}