1use crate::parser::{LineStream, parse_declared_sections};
6use crate::{error, types};
7use std::path::Path;
8
9pub struct MemchrLines<'a> {
16 bytes: &'a [u8],
17 pos: usize,
18 peeked: Option<&'a str>,
19 check_utf8: bool,
22 utf8_bad: bool,
23 newlines: Option<Vec<usize>>,
26 index_line: usize,
28 line_no: u32,
31}
32
33impl<'a> MemchrLines<'a> {
34 pub fn new(text: &'a str) -> Self {
35 Self::from_parts(text.as_bytes(), false)
36 }
37
38 pub fn from_bytes(bytes: &'a [u8]) -> Self {
44 Self::from_parts(bytes, true)
45 }
46
47 fn from_parts(bytes: &'a [u8], check_utf8: bool) -> Self {
48 Self {
49 bytes,
50 pos: 0,
51 peeked: None,
52 check_utf8,
53 utf8_bad: false,
54 newlines: None,
55 index_line: 0,
56 line_no: 0,
57 }
58 }
59
60 #[inline]
62 pub fn line_no(&self) -> u32 {
63 self.line_no
64 }
65
66 #[inline]
67 fn read_one(&mut self) -> Option<&'a str> {
68 if self.newlines.is_some() {
69 return self.read_one_indexed();
70 }
71 if self.utf8_bad || self.pos >= self.bytes.len() {
72 return None;
73 }
74 let rest = &self.bytes[self.pos..];
75 let (line_bytes, advance) = match memchr::memchr(b'\n', rest) {
76 Some(i) => (&rest[..i], i + 1),
77 None => (rest, rest.len()),
78 };
79 let trimmed = if line_bytes.last() == Some(&b'\r') {
80 &line_bytes[..line_bytes.len() - 1]
81 } else {
82 line_bytes
83 };
84 if self.check_utf8 {
85 match std::str::from_utf8(trimmed) {
86 Ok(s) => {
87 self.accept_line(advance);
88 Some(s)
89 }
90 Err(_) => {
91 self.line_no = self.line_no.saturating_add(1);
95 self.utf8_bad = true;
96 None
97 }
98 }
99 } else {
100 self.accept_line(advance);
101 Some(unsafe { std::str::from_utf8_unchecked(trimmed) })
103 }
104 }
105
106 #[inline]
107 fn accept_line(&mut self, advance: usize) {
108 self.pos += advance;
109 self.line_no = self.line_no.saturating_add(1);
110 }
111
112 #[inline]
113 pub fn next_line(&mut self) -> Option<&'a str> {
114 if let Some(p) = self.peeked.take() {
115 return Some(p);
116 }
117 self.read_one()
118 }
119
120 #[inline]
121 pub fn peek_line(&mut self) -> Option<&'a str> {
122 if self.peeked.is_none() {
123 self.peeked = self.read_one();
124 }
125 self.peeked
126 }
127
128 fn clear_peek(&mut self) {
130 if let Some(p) = self.peeked.take() {
131 let start = p.as_ptr() as usize - self.bytes.as_ptr() as usize;
133 self.pos = start;
134 self.line_no = self.line_no.saturating_sub(1);
138 if let Some(nl) = self.newlines.as_deref() {
139 self.index_line = line_no_for_pos(nl, self.bytes.len(), start);
140 }
141 }
142 }
143
144 fn read_one_indexed(&mut self) -> Option<&'a str> {
150 if self.utf8_bad || self.pos >= self.bytes.len() {
151 return None;
152 }
153 let len = self.bytes.len();
154 let line_no = {
155 let nl = self.newlines.as_deref().unwrap();
156 if self.pos == line_start_pos(nl, len, self.index_line) {
157 self.index_line
158 } else {
159 line_no_for_pos(nl, len, self.pos)
160 }
161 };
162 let span = indexed_span(self.newlines.as_deref().unwrap(), self.bytes, line_no)?;
163 let trimmed = &self.bytes[span.start..span.content_end];
164 if self.check_utf8 {
165 match std::str::from_utf8(trimmed) {
166 Ok(s) => {
167 self.pos = span.next_pos;
168 self.index_line = line_no + 1;
169 self.line_no = self.index_line as u32;
170 Some(s)
171 }
172 Err(_) => {
173 self.utf8_bad = true;
174 self.index_line = line_no;
175 self.line_no = (line_no as u32).saturating_add(1);
178 None
179 }
180 }
181 } else {
182 self.pos = span.next_pos;
183 self.index_line = line_no + 1;
184 self.line_no = self.index_line as u32;
185 Some(unsafe { std::str::from_utf8_unchecked(trimmed) })
187 }
188 }
189
190 fn utf8_error(&self) -> Option<error::ParseError> {
191 self.utf8_bad.then(invalid_utf8)
192 }
193
194 fn prefer_utf8(&self, err: error::ParseError) -> error::ParseError {
196 if self.utf8_bad { invalid_utf8() } else { err }
197 }
198}
199
200fn invalid_utf8() -> error::ParseError {
201 error::ParseError::ValidationError("invalid UTF-8".into())
202}
203
204impl<'a> Iterator for MemchrLines<'a> {
205 type Item = &'a str;
206 fn next(&mut self) -> Option<&'a str> {
207 self.next_line()
208 }
209}
210
211impl<'a> LineStream<'a> for MemchrLines<'a> {
212 #[inline]
213 fn next_line(&mut self) -> Option<&'a str> {
214 MemchrLines::next_line(self)
215 }
216 #[inline]
217 fn peek_line(&mut self) -> Option<&'a str> {
218 MemchrLines::peek_line(self)
219 }
220 #[inline]
221 fn next_coord_row(
222 &mut self,
223 spec_version: u32,
224 default_id: u64,
225 ) -> Result<([f64; 3], [bool; 3], u64), crate::error::ParseError> {
226 if let Some(p) = self.peeked.take() {
227 return crate::parser::parse_coord_row_at(p.as_bytes(), 0, spec_version, default_id)
228 .map(|(row, _)| row);
229 }
230 self.line_no = self.line_no.saturating_add(1);
233 let (row, pos) =
234 crate::parser::parse_coord_row_at(self.bytes, self.pos, spec_version, default_id)?;
235 self.pos = pos;
236 Ok(row)
237 }
238 #[inline]
239 fn next_vector_row(
240 &mut self,
241 spec_version: u32,
242 ) -> Result<([f64; 3], [bool; 3], u64), crate::error::ParseError> {
243 if let Some(p) = self.peeked.take() {
244 return crate::parser::parse_vector_row_at(p.as_bytes(), 0, spec_version)
245 .map(|(row, _)| row);
246 }
247 let (row, pos) = crate::parser::parse_vector_row_at(self.bytes, self.pos, spec_version)?;
248 self.pos = pos;
249 Ok(row)
250 }
251 fn remainder_is_whitespace(&mut self) -> bool {
252 if self.utf8_bad {
253 return false;
254 }
255 if let Some(line) = self.peek_line()
256 && !line.trim().is_empty()
257 {
258 return false;
259 }
260 if self.utf8_bad {
263 return false;
264 }
265 self.bytes[self.pos..]
266 .iter()
267 .all(|b| b.is_ascii_whitespace())
268 }
269 fn blank_line_opens_section(&mut self) -> bool {
270 let Some(peeked) = self.peeked else {
272 return false;
273 };
274 if !peeked.trim().is_empty() {
275 return false;
276 }
277 match probe_section(&self.bytes[self.pos..]) {
278 SectionProbe::Opens => true,
279 SectionProbe::Closed => false,
280 SectionProbe::InvalidUtf8 => {
283 self.utf8_bad = true;
284 false
285 }
286 }
287 }
288
289 fn blank_opens_velocity_block(&mut self) -> bool {
290 match self.peek_line() {
291 Some(line) if line.trim().is_empty() => self.blank_line_opens_section(),
292 _ => true,
293 }
294 }
295}
296
297fn split_raw_line(bytes: &[u8]) -> Option<(&[u8], &[u8])> {
298 if bytes.is_empty() {
299 return None;
300 }
301 match memchr::memchr(b'\n', bytes) {
302 Some(i) => Some((&bytes[..i], &bytes[i + 1..])),
303 None => Some((bytes, &[])),
304 }
305}
306
307fn strip_cr(line: &[u8]) -> &[u8] {
308 if line.last() == Some(&b'\r') {
309 &line[..line.len() - 1]
310 } else {
311 line
312 }
313}
314
315enum SectionProbe {
316 Opens,
317 Closed,
318 InvalidUtf8,
319}
320
321fn probe_section(after_blank: &[u8]) -> SectionProbe {
324 let Some((raw1, after1)) = split_raw_line(after_blank) else {
325 return SectionProbe::Closed;
326 };
327 let line1 = strip_cr(raw1);
328 if line1.iter().all(|b| matches!(b, b' ' | b'\t')) {
329 return SectionProbe::Closed;
330 }
331 let Ok(text1) = std::str::from_utf8(line1) else {
332 return SectionProbe::InvalidUtf8;
333 };
334 if text1.trim_start().starts_with('{') {
335 return SectionProbe::Closed;
336 }
337 let Some((raw2, _)) = split_raw_line(after1) else {
338 return SectionProbe::Closed;
339 };
340 let Ok(text2) = std::str::from_utf8(strip_cr(raw2)) else {
341 return SectionProbe::InvalidUtf8;
342 };
343 if crate::parser::line_has_component_label(text2) {
344 SectionProbe::Opens
345 } else {
346 SectionProbe::Closed
347 }
348}
349
350fn blank_opens_section(after_blank: &[u8]) -> bool {
351 matches!(probe_section(after_blank), SectionProbe::Opens)
352}
353
354pub struct ConFrameIterator<'a> {
365 pub(crate) lines: MemchrLines<'a>,
366 frame_index: u64,
370}
371
372#[cfg(target_arch = "x86_64")]
380#[target_feature(enable = "avx2,popcnt")]
381#[inline]
382unsafe fn skip_newlines_avx2(hay: &[u8], n: usize) -> (usize, usize) {
383 use std::arch::x86_64::{
384 __m256i, _mm256_cmpeq_epi8, _mm256_loadu_si256, _mm256_movemask_epi8, _mm256_set1_epi8,
385 _popcnt32,
386 };
387 if n == 0 {
388 return (0, 0);
389 }
390 let mut need = n;
391 let mut offset = 0usize;
392 let ptr = hay.as_ptr();
393 let len = hay.len();
394 let nl = _mm256_set1_epi8(b'\n' as i8);
395 while offset + 32 <= len {
396 let chunk = unsafe { _mm256_loadu_si256(ptr.add(offset) as *const __m256i) };
398 let eq = _mm256_cmpeq_epi8(chunk, nl);
399 let mask = _mm256_movemask_epi8(eq) as u32;
400 let count = _popcnt32(mask as i32) as usize;
401 if count >= need {
402 let mut bits = mask;
403 for _ in 1..need {
404 bits &= bits - 1;
405 }
406 let bit = bits.trailing_zeros() as usize;
407 return (n, offset + bit + 1);
408 }
409 need -= count;
410 offset += 32;
411 }
412 while offset < len {
413 if hay[offset] == b'\n' {
414 need -= 1;
415 if need == 0 {
416 return (n, offset + 1);
417 }
418 }
419 offset += 1;
420 }
421 (n - need, len)
422}
423
424impl<'a> ConFrameIterator<'a> {
425 pub fn new(file_contents: &'a str) -> Self {
431 ConFrameIterator {
432 lines: MemchrLines::new(file_contents),
433 frame_index: 0,
434 }
435 }
436
437 pub(crate) fn starting_at(file_contents: &'a str, pos: usize) -> Self {
442 debug_assert!(pos == 0 || file_contents.as_bytes()[pos - 1] == b'\n');
443 let mut it = Self::new(file_contents);
444 it.lines.pos = pos;
445 it
446 }
447
448 pub fn from_bytes(bytes: &'a [u8]) -> Self {
454 ConFrameIterator {
455 lines: MemchrLines::from_bytes(bytes),
456 frame_index: 0,
457 }
458 }
459
460 pub(crate) fn frame_index(&self) -> u64 {
465 self.frame_index
466 }
467
468 fn note_skipped_frame(&mut self) -> Option<Result<(), error::ParseError>> {
469 self.frame_index = self.frame_index.saturating_add(1);
470 Some(Ok(()))
471 }
472
473 #[inline(always)]
483 fn advance_lines(&mut self, n: usize) -> Result<(), error::ParseError> {
484 self.lines.clear_peek();
485 if self.lines.newlines.is_some() {
486 return self.advance_lines_indexed(n);
487 }
488 #[cfg(target_arch = "x86_64")]
489 {
490 if n > 1 && std::is_x86_feature_detected!("avx2") {
491 return self.advance_lines_avx2(n);
492 }
493 }
494 self.advance_lines_memchr(n)
495 }
496
497 fn advance_lines_memchr(&mut self, n: usize) -> Result<(), error::ParseError> {
498 for i in 0..n {
499 let rest = &self.lines.bytes[self.lines.pos..];
500 match memchr::memchr(b'\n', rest) {
501 Some(pos) => {
502 self.lines.pos += pos + 1;
503 self.lines.line_no = self.lines.line_no.saturating_add(1);
504 }
505 None => {
506 if rest.is_empty() {
507 return Err(error::ParseError::IncompleteFrame);
508 }
509 self.lines.pos = self.lines.bytes.len();
510 self.lines.line_no = self.lines.line_no.saturating_add(1);
514 if i + 1 == n {
515 return Ok(());
516 }
517 return Err(error::ParseError::IncompleteFrame);
518 }
519 }
520 }
521 Ok(())
522 }
523
524 #[cfg(target_arch = "x86_64")]
530 #[inline]
531 fn advance_lines_avx2(&mut self, n: usize) -> Result<(), error::ParseError> {
532 let rest = &self.lines.bytes[self.lines.pos..];
533 let (found, end) = unsafe { skip_newlines_avx2(rest, n) };
534 if found == n {
535 self.lines.pos += end;
536 self.lines.line_no = self
539 .lines
540 .line_no
541 .saturating_add(u32::try_from(n).unwrap_or(u32::MAX));
542 return Ok(());
543 }
544 if rest.is_empty() {
545 return Err(error::ParseError::IncompleteFrame);
546 }
547 let has_tail = rest.last().copied() != Some(b'\n');
548 self.lines.pos = self.lines.bytes.len();
549 let counted = found + usize::from(has_tail);
550 self.lines.line_no = self
551 .lines
552 .line_no
553 .saturating_add(u32::try_from(counted).unwrap_or(u32::MAX));
554 if counted >= n {
555 Ok(())
556 } else {
557 Err(error::ParseError::IncompleteFrame)
558 }
559 }
560
561 fn advance_lines_to(&mut self, n: usize, end: Option<usize>) -> Result<(), error::ParseError> {
567 if self.lines.newlines.is_some() {
570 return self.advance_lines(n);
571 }
572 self.lines.clear_peek();
573 let pos = self.lines.pos;
574 let bytes = self.lines.bytes;
575 if let Some(end) = end
576 && pos < end
577 && end <= bytes.len()
578 && bytes[end - 1] == b'\n'
579 && memchr::memchr_iter(b'\n', &bytes[pos..end]).count() == n
580 {
581 self.lines.pos = end;
582 self.lines.line_no = self
585 .lines
586 .line_no
587 .saturating_add(u32::try_from(n).unwrap_or(u32::MAX));
588 return Ok(());
589 }
590 self.advance_lines(n)
591 }
592
593 fn advance_lines_indexed(&mut self, n: usize) -> Result<(), error::ParseError> {
598 let len = self.lines.bytes.len();
599 let step = {
600 let nl = self.lines.newlines.as_deref().unwrap();
601 let nlines = line_count(nl, len);
602 let line_no = if self.lines.pos == line_start_pos(nl, len, self.lines.index_line) {
603 self.lines.index_line
604 } else {
605 line_no_for_pos(nl, len, self.lines.pos)
606 };
607 if n == 0 {
608 IndexStep::Stay(line_no)
609 } else if line_no >= nlines {
610 IndexStep::Missing(line_no)
611 } else {
612 match line_no.checked_add(n) {
613 Some(target) if target <= nlines => {
614 let pos = if target == nlines {
615 len
616 } else {
617 line_start_pos(nl, len, target)
618 };
619 IndexStep::Landed(target, pos)
620 }
621 _ => IndexStep::Short(nlines, len),
622 }
623 }
624 };
625 match step {
626 IndexStep::Stay(line_no) => {
627 self.lines.index_line = line_no;
628 self.lines.line_no = u32::try_from(line_no).unwrap_or(u32::MAX);
629 Ok(())
630 }
631 IndexStep::Missing(line_no) => {
632 self.lines.index_line = line_no;
633 self.lines.line_no = u32::try_from(line_no).unwrap_or(u32::MAX);
634 Err(error::ParseError::IncompleteFrame)
635 }
636 IndexStep::Landed(line_no, pos) => {
637 self.lines.index_line = line_no;
638 self.lines.line_no = u32::try_from(line_no).unwrap_or(u32::MAX);
639 self.lines.pos = pos;
640 Ok(())
641 }
642 IndexStep::Short(line_no, pos) => {
643 self.lines.index_line = line_no;
644 self.lines.line_no = u32::try_from(line_no).unwrap_or(u32::MAX);
645 self.lines.pos = pos;
646 Err(error::ParseError::IncompleteFrame)
647 }
648 }
649 }
650
651 fn read_line_str(&mut self) -> Option<&'a str> {
653 self.lines.clear_peek();
654 self.lines.next_line()
655 }
656
657 pub fn forward_fast(&mut self) -> Option<Result<(), error::ParseError>> {
661 self.forward_fast_to(None)
662 }
663
664 #[inline(always)]
669 pub(crate) fn forward_fast_to(
670 &mut self,
671 end: Option<usize>,
672 ) -> Option<Result<(), error::ParseError>> {
673 self.lines.clear_peek();
674 if self.lines.pos >= self.lines.bytes.len() {
675 return None;
676 }
677 if let Err(e) = self.advance_lines(6) {
679 return Some(Err(e));
680 }
681 let natm_types: usize = match self.read_line_str() {
683 Some(line) => match crate::parser::parse_count_line(line, 1) {
684 Ok(v) => v[0],
685 Err(e) => return Some(Err(e)),
686 },
687 None => return Some(Err(error::ParseError::IncompleteHeader)),
688 };
689 if natm_types == 0 {
690 return match crate::parser::read_zero_atom_tail(|| self.read_line_str()) {
693 Ok(()) => self.note_skipped_frame(),
694 Err(e) => Some(Err(e)),
695 };
696 }
697 let natms_per_type: Vec<usize> = match self.read_line_str() {
699 Some(line) => match crate::parser::parse_count_line(line, natm_types) {
700 Ok(v) => v,
701 Err(e) => return Some(Err(e)),
702 },
703 None => return Some(Err(error::ParseError::IncompleteHeader)),
704 };
705 if natms_per_type.contains(&0) {
706 return Some(Err(error::ParseError::InvalidCount(
707 "each per-type count must be >= 1".into(),
708 )));
709 }
710 if let Err(e) = self.advance_lines(1) {
712 return Some(Err(e));
713 }
714 let total_atoms: usize = natms_per_type.iter().sum();
715 let coord_block_lines = total_atoms + natm_types * 2;
716 if let Err(e) = self.advance_lines_to(coord_block_lines, end) {
717 return Some(Err(e));
718 }
719 self.lines.clear_peek();
721 loop {
722 let rest = &self.lines.bytes[self.lines.pos..];
723 if rest.is_empty() {
724 break;
725 }
726 let next_eol = memchr::memchr(b'\n', rest);
727 let line = match next_eol {
728 Some(pos) => &rest[..pos],
729 None => rest,
730 };
731 let is_blank = line.iter().all(|b| matches!(b, b' ' | b'\t' | b'\r'));
732 if !is_blank {
733 break;
734 }
735 let after = match next_eol {
736 Some(pos) => &rest[pos + 1..],
737 None => &[][..],
738 };
739 if after.iter().all(|b| b.is_ascii_whitespace()) {
742 self.lines.pos = self.lines.bytes.len();
743 break;
744 }
745 if !blank_opens_section(after) {
748 break;
749 }
750 self.lines.pos += next_eol.map(|p| p + 1).unwrap_or(rest.len());
754 self.lines.line_no = self.lines.line_no.saturating_add(1);
757 if let Err(e) = self.advance_lines_to(coord_block_lines, end) {
758 return Some(Err(e));
759 }
760 }
761 self.note_skipped_frame()
762 }
763
764 pub fn forward(&mut self) -> Option<Result<(), error::ParseError>> {
776 self.forward_fast()
778 }
779
780 pub fn byte_pos(&self) -> usize {
782 self.lines.pos
783 }
784
785 pub fn skip_frames(&mut self, n: usize) -> Result<usize, error::ParseError> {
788 self.skip_frames_counted(n).map_err(|(_, err)| err)
789 }
790
791 pub(crate) fn skip_frames_counted(
795 &mut self,
796 n: usize,
797 ) -> Result<usize, (usize, error::ParseError)> {
798 let mut skipped = 0usize;
799 for _ in 0..n {
800 match self.forward() {
801 None => break,
802 Some(Err(e)) => return Err((skipped, e)),
803 Some(Ok(())) => skipped += 1,
804 }
805 }
806 Ok(skipped)
807 }
808
809 pub fn next_with_raw_span(
822 &mut self,
823 file_contents: &'a str,
824 ) -> Option<Result<(types::ConFrame, &'a str), error::ParseError>> {
825 let base = file_contents.as_ptr() as usize;
826 let start = {
827 let line = self.lines.peek_line()?;
828 line.as_ptr() as usize - base
829 };
830 let frame = match self.next()? {
831 Ok(f) => f,
832 Err(e) => return Some(Err(e)),
833 };
834 let end = match self.lines.peek_line() {
835 Some(line) => line.as_ptr() as usize - base,
836 None => file_contents.len(),
837 };
838 debug_assert!(end >= start && end <= file_contents.len());
839 Some(Ok((frame, &file_contents[start..end])))
840 }
841
842 pub(crate) fn next_with_end(
844 &mut self,
845 ) -> Option<Result<(types::ConFrame, usize), error::ParseError>> {
846 let frame = match self.next()? {
847 Ok(f) => f,
848 Err(e) => return Some(Err(e)),
849 };
850 let end = match self.lines.peek_line() {
851 Some(line) => line.as_ptr() as usize - self.lines.bytes.as_ptr() as usize,
852 None => self.lines.pos,
853 };
854 Some(Ok((frame, end)))
855 }
856}
857
858impl<'a> Iterator for ConFrameIterator<'a> {
859 type Item = Result<types::ConFrame, error::ParseError>;
864
865 fn next(&mut self) -> Option<Self::Item> {
872 if let Some(err) = self.lines.utf8_error() {
875 return Some(Err(err));
876 }
877 if self.lines.remainder_is_whitespace() {
881 if let Some(err) = self.lines.utf8_error() {
882 return Some(Err(err));
883 }
884 while self.lines.next_line().is_some() {}
885 return None;
886 }
887 if let Some(err) = self.lines.utf8_error() {
888 return Some(Err(err));
889 }
890 let mut frame = match crate::parser::parse_single_frame_ls(&mut self.lines) {
892 Ok(f) => f,
893 Err(e) => return Some(Err(self.lines.prefer_utf8(e))),
894 };
895 let sections =
899 match parse_declared_sections(&mut self.lines, &mut frame.header, &mut frame.atom_data)
900 {
901 Ok(n) => n,
902 Err(e) => return Some(Err(self.lines.prefer_utf8(e))),
905 };
906 if sections > 0 {
907 frame.install_declared_section_soa();
911 }
912 self.frame_index = self.frame_index.saturating_add(1);
913 Some(Ok(frame))
914 }
915}
916
917#[cfg(test)]
918mod aos_soa_agreement_tests {
919 use super::*;
920 use std::path::PathBuf;
921
922 #[test]
923 fn iterator_vel_forces_soa_matches_aos() {
924 let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
925 .join("resources/test/tiny_cuh2_vel_forces.con");
926 let text = std::fs::read_to_string(&p).expect("fixture");
927 let fr = ConFrameIterator::new(&text)
928 .next()
929 .expect("frame")
930 .expect("parse");
931 let n = fr.atom_data.len();
932 assert!(n > 0);
933 assert_eq!(fr.positions.nrows(), n);
934 let has_vel = fr.atom_data.iter().any(|a| a.velocity().is_some());
935 let has_frc = fr.atom_data.iter().any(|a| a.force().is_some());
936 if has_vel {
937 assert_eq!(
938 fr.velocities.nrows(),
939 n,
940 "SoA velocities must match AoS after section parse"
941 );
942 }
943 if has_frc {
944 assert_eq!(fr.forces.nrows(), n, "SoA forces must match AoS");
945 }
946 for (i, a) in fr.atom_data.iter().enumerate() {
947 let p = fr.positions.as_f64_row(i);
948 assert_eq!([a.x, a.y, a.z], p);
949 if let Some(v) = a.velocity() {
950 assert_eq!(v, fr.velocities.as_f64_row(i));
951 }
952 if let Some(f) = a.force() {
953 assert_eq!(f, fr.forces.as_f64_row(i));
954 }
955 }
956 }
957
958 #[test]
961 fn sync_skips_pos_when_nrows_matches_keeps_force_soa() {
962 let p =
963 PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/test/tiny_cuh2_forces.con");
964 let text = std::fs::read_to_string(&p).expect("fixture");
965 let fr = ConFrameIterator::new(&text)
966 .next()
967 .expect("frame")
968 .expect("parse");
969 let n = fr.atom_data.len();
970 assert_eq!(fr.positions.nrows(), n);
971 assert_eq!(fr.forces.nrows(), n);
972 let p0 = fr.positions.as_f64_row(0);
975 let mut fr2 = fr.clone();
976 fr2.sync_arrays_from_atom_data();
977 assert_eq!(fr2.positions.as_f64_row(0), p0);
978 assert_eq!(fr2.forces.nrows(), n);
979 assert_eq!(
980 fr2.forces.as_f64_row(0),
981 fr2.atom_data[0].force().expect("force")
982 );
983 }
984
985 #[test]
988 fn declared_sections_soa_matches_every_atom() {
989 let n = 400;
990 let mut text = String::new();
991 text.push_str("soa fill\n");
992 text.push_str(
993 "{\"con_spec_version\":2,\"sections\":[\"velocities\",\"forces\",\"energies\"]}\n",
994 );
995 text.push_str("10.0 10.0 10.0\n90.0 90.0 90.0\n0 0\n0 0 0\n1\n");
996 text.push_str(&format!("{n}\n63.546\nCu\nCoordinates of Component 1\n"));
997 for i in 0..n {
998 let x = (i as f64) * 0.125 - 3.5;
999 let y = (i as f64) * -0.05 + 1.0;
1000 let z = (i % 17) as f64 * 0.333;
1001 text.push_str(&format!("{x:.8} {y:.8} {z:.8} 0 {i}\n"));
1002 }
1003 text.push_str("\nCu\nVelocities of Component 1\n");
1004 for i in 0..n {
1005 let vx = (i as f64) * 1.0e-4 - 0.02;
1006 let vy = ((i * 3) % 11) as f64 * -1.0e-3;
1007 let vz = (i as f64).sin() * 0.01;
1008 text.push_str(&format!("{vx:.8} {vy:.8} {vz:.8} 0 {i}\n"));
1009 }
1010 text.push_str("\nCu\nForces of Component 1\n");
1011 for i in 0..n {
1012 let fx = (i as f64) * 0.01;
1013 let fy = -fx / 2.0;
1014 let fz = (i % 5) as f64;
1015 text.push_str(&format!("{fx:.8} {fy:.8} {fz:.8} 0 {i}\n"));
1016 }
1017 text.push_str("\nCu\nEnergies of Component 1\n");
1018 for i in 0..n {
1019 let e = (i as f64) * 0.02 - 4.0;
1020 text.push_str(&format!("{e:.8} 0 {i}\n"));
1021 }
1022 let frame = ConFrameIterator::new(&text)
1023 .next()
1024 .expect("frame")
1025 .expect("parse");
1026 assert_eq!(frame.atom_data.len(), n);
1027 assert_eq!(frame.velocities.nrows(), n);
1028 assert_eq!(frame.forces.nrows(), n);
1029 assert_eq!(frame.atom_energies.len(), n);
1030 assert_eq!(frame.positions.nrows(), n);
1031 assert!(frame.magmoms.nrows() == 0);
1032 for i in 0..n {
1033 let atom = &frame.atom_data[i];
1034 let pos = frame.positions.as_f64_row(i);
1035 assert_eq!(pos[0].to_bits(), atom.x.to_bits());
1036 assert_eq!(pos[1].to_bits(), atom.y.to_bits());
1037 assert_eq!(pos[2].to_bits(), atom.z.to_bits());
1038 let vel = atom.velocity().unwrap();
1039 let soa_v = frame.velocities.as_f64_row(i);
1040 assert_eq!(soa_v[0].to_bits(), vel[0].to_bits());
1041 assert_eq!(soa_v[1].to_bits(), vel[1].to_bits());
1042 assert_eq!(soa_v[2].to_bits(), vel[2].to_bits());
1043 let frc = atom.force().unwrap();
1044 let soa_f = frame.forces.as_f64_row(i);
1045 assert_eq!(soa_f[0].to_bits(), frc[0].to_bits());
1046 assert_eq!(soa_f[1].to_bits(), frc[1].to_bits());
1047 assert_eq!(soa_f[2].to_bits(), frc[2].to_bits());
1048 assert_eq!(
1049 frame.atom_energies.get_f64(i).to_bits(),
1050 atom.energy().unwrap().to_bits()
1051 );
1052 assert_eq!(frame.atom_ids[i], atom.atom_id);
1053 }
1054 }
1055}
1056
1057pub fn read_all_frames(path: &Path) -> Result<Vec<types::ConFrame>, Box<dyn std::error::Error>> {
1063 read_all_frames_with_threads(path, None)
1064}
1065
1066pub fn frames_from_text(
1073 text: &str,
1074 num_threads: Option<usize>,
1075) -> Result<Vec<types::ConFrame>, error::ParseError> {
1076 #[cfg(feature = "parallel")]
1077 {
1078 if let Some(n) = num_threads.filter(|n| *n >= 2) {
1079 let parts = parse_frames_parallel_with_threads(text, Some(n));
1080 let mut frames = Vec::with_capacity(parts.len());
1081 for r in parts {
1082 frames.push(r?);
1083 }
1084 return Ok(frames);
1085 }
1086 }
1087 #[cfg(not(feature = "parallel"))]
1088 {
1089 let _ = num_threads;
1090 }
1091 ConFrameIterator::new(text).collect()
1092}
1093
1094pub fn read_all_frames_with_threads(
1106 path: &Path,
1107 num_threads: Option<usize>,
1108) -> Result<Vec<types::ConFrame>, Box<dyn std::error::Error>> {
1109 let bytes = crate::compression::read_bytes_owned(path)?;
1110 Ok(frames_from_bytes_located(bytes.as_slice(), num_threads)?)
1111}
1112
1113#[derive(Debug)]
1115pub struct PositionBlock {
1125 pub n_frames: usize,
1126 pub n_atoms: usize,
1127 pub values: Vec<f64>,
1128}
1129
1130pub fn read_positions_with_threads(
1137 path: &Path,
1138 num_threads: Option<usize>,
1139) -> Result<PositionBlock, Box<dyn std::error::Error>> {
1140 let contents = crate::compression::read_file_contents(path)?;
1141 let block = match &contents {
1142 crate::compression::FileContents::Owned(text) => {
1143 fill_position_block(text.as_bytes(), num_threads)
1144 }
1145 crate::compression::FileContents::Mapped(map) => fill_position_block(map, num_threads),
1146 }?;
1147 Ok(block)
1148}
1149
1150pub struct CoordColumns {
1155 pub block: PositionBlock,
1156 pub masses: Vec<f64>,
1157 pub cells: Vec<f64>,
1158}
1159
1160pub fn read_coord_columns(
1165 path: &Path,
1166 num_threads: Option<usize>,
1167) -> Result<CoordColumns, Box<dyn std::error::Error>> {
1168 let contents = crate::compression::read_file_contents(path)?;
1169 let bytes = contents.as_bytes();
1170 if let Ok(Some(columns)) = certified_columns(bytes, num_threads) {
1171 return Ok(columns);
1172 }
1173 let frames = frames_from_bytes_located(bytes, num_threads)?;
1174 Ok(columns_from_frames(&frames))
1175}
1176
1177fn certified_columns(
1178 bytes: &[u8],
1179 num_threads: Option<usize>,
1180) -> Result<Option<CoordColumns>, error::LocatedParseError> {
1181 let parsed = match crate::columns::positions_from_con_bytes(bytes) {
1182 Ok(parsed) if parsed.stock => parsed,
1183 _ => return Ok(None),
1184 };
1185 if parsed.n_frames == 0 {
1186 return Ok(Some(CoordColumns {
1187 block: PositionBlock {
1188 n_frames: 0,
1189 n_atoms: 0,
1190 values: Vec::new(),
1191 },
1192 masses: Vec::new(),
1193 cells: Vec::new(),
1194 }));
1195 }
1196 let Some(n_atoms) = parsed.rectangular_atoms() else {
1197 let first = parsed.counts[0];
1198 let index = parsed
1199 .counts
1200 .iter()
1201 .position(|count| *count != first)
1202 .unwrap_or(0);
1203 let line = file_line_at_bytes(
1204 bytes,
1205 parsed.spans.get(index).map(|span| span.start).unwrap_or(0),
1206 );
1207 return Err(error::LocatedParseError::new(
1208 error::ParseError::InvalidCount(
1209 "frame atom count does not match the first frame".into(),
1210 ),
1211 line,
1212 index as u64,
1213 ));
1214 };
1215 let values = if parallel_column_threads(num_threads) {
1216 parallel_column_values(bytes, &parsed, n_atoms, num_threads)
1217 .map_err(|err| error::LocatedParseError::new(err, 0, 0))?
1218 } else {
1219 parsed.positions
1220 };
1221 Ok(Some(CoordColumns {
1222 block: PositionBlock {
1223 n_frames: parsed.n_frames,
1224 n_atoms,
1225 values,
1226 },
1227 masses: parsed.masses,
1228 cells: parsed.cells,
1229 }))
1230}
1231
1232fn parallel_column_threads(num_threads: Option<usize>) -> bool {
1233 #[cfg(feature = "parallel")]
1236 {
1237 matches!(num_threads, Some(n) if n >= 2)
1238 }
1239 #[cfg(not(feature = "parallel"))]
1240 {
1241 let _ = num_threads;
1242 false
1243 }
1244}
1245
1246fn parallel_column_values(
1247 bytes: &[u8],
1248 parsed: &crate::columns::ConPositions,
1249 n_atoms: usize,
1250 num_threads: Option<usize>,
1251) -> Result<Vec<f64>, error::ParseError> {
1252 let stride = n_atoms * 3;
1253 if stride == 0 {
1254 return Ok(Vec::new());
1255 }
1256 #[cfg(not(feature = "parallel"))]
1257 {
1258 let _ = (bytes, num_threads);
1259 Ok(parsed.positions.clone())
1260 }
1261 #[cfg(feature = "parallel")]
1262 {
1263 let mut values = vec![0.0_f64; parsed.n_frames * stride];
1264 let fill_one = |index: usize, dest: &mut [f64]| {
1265 let span = &parsed.spans[index];
1266 let slice = crate::columns::positions_from_con_bytes(&bytes[span.start..span.end])?;
1267 if slice.counts.first().copied() != Some(n_atoms) || slice.positions.len() != stride {
1268 return Err(error::ParseError::InvalidCount(
1269 "frame atom count does not match the first frame".into(),
1270 ));
1271 }
1272 dest.copy_from_slice(&slice.positions);
1273 Ok(())
1274 };
1275 use rayon::prelude::*;
1276 let mut run = || match values
1277 .par_chunks_mut(stride)
1278 .enumerate()
1279 .find_map_first(|(index, dest)| fill_one(index, dest).err())
1280 {
1281 Some(err) => Err(err),
1282 None => Ok(()),
1283 };
1284 match num_threads {
1285 Some(n) if n >= 2 => {
1286 let pool = rayon::ThreadPoolBuilder::new()
1287 .num_threads(n)
1288 .build()
1289 .map_err(|err| error::ParseError::ValidationError(err.to_string()))?;
1290 pool.install(run)?;
1291 }
1292 _ => run()?,
1293 }
1294 Ok(values)
1295 }
1296}
1297
1298fn columns_from_frames(frames: &[types::ConFrame]) -> CoordColumns {
1299 let n_frames = frames.len();
1300 let n_atoms = frames
1301 .first()
1302 .map(|frame| frame.positions.nrows())
1303 .unwrap_or(0);
1304 let mut values = Vec::with_capacity(n_frames * n_atoms * 3);
1305 let mut masses = Vec::with_capacity(n_frames * n_atoms);
1306 let mut cells = Vec::with_capacity(n_frames * 6);
1307 for frame in frames {
1308 cells.extend_from_slice(&frame.header.boxl);
1309 cells.extend_from_slice(&frame.header.angles);
1310 if let Some(src) = frame.positions.f64_slice() {
1311 values.extend_from_slice(src);
1312 } else {
1313 for row in 0..frame.positions.nrows() {
1314 values.extend_from_slice(&frame.positions.as_f64_row(row));
1315 }
1316 }
1317 for (type_index, &count) in frame.header.natms_per_type.iter().enumerate() {
1318 let mass = frame
1319 .header
1320 .masses_per_type
1321 .get(type_index)
1322 .copied()
1323 .unwrap_or(0.0);
1324 masses.extend(std::iter::repeat_n(mass, count));
1325 }
1326 }
1327 CoordColumns {
1328 block: PositionBlock {
1329 n_frames,
1330 n_atoms,
1331 values,
1332 },
1333 masses,
1334 cells,
1335 }
1336}
1337
1338fn file_line_at_bytes(bytes: &[u8], byte: usize) -> u32 {
1340 let end = byte.min(bytes.len());
1341 1 + memchr::memchr_iter(b'\n', &bytes[..end]).count() as u32
1342}
1343
1344fn share_cursor(local: u32, bytes: &[u8], share_start: usize) -> u32 {
1349 let base = file_line_at_bytes(bytes, share_start);
1350 if local == 0 {
1351 base
1352 } else {
1353 base.saturating_add(local.saturating_sub(1))
1354 }
1355}
1356
1357fn frame_starts(bytes: &[u8]) -> Result<Vec<usize>, error::LocatedParseError> {
1358 let mut scanner = ConFrameIterator::from_bytes(bytes);
1359 let mut boundaries = Vec::new();
1360 loop {
1361 scanner.lines.clear_peek();
1362 let start = scanner.lines.pos;
1363 if start >= bytes.len() || bytes[start..].iter().all(u8::is_ascii_whitespace) {
1364 break;
1365 }
1366 boundaries.push(start);
1367 match scanner.forward_fast() {
1368 Some(Ok(())) => {}
1369 Some(Err(err)) => {
1370 return Err(error::LocatedParseError::new(
1371 err,
1372 scanner.lines.line_no(),
1373 scanner.frame_index(),
1374 ));
1375 }
1376 None => break,
1377 }
1378 }
1379 Ok(boundaries)
1380}
1381
1382fn parse_span_located(
1384 bytes: &[u8],
1385 start: usize,
1386 end: usize,
1387 frame_index: u64,
1388) -> Result<types::ConFrame, error::LocatedParseError> {
1389 let mut iter = ConFrameIterator::from_bytes(&bytes[start..end]);
1390 match iter.next() {
1391 Some(Ok(frame)) => Ok(frame),
1392 Some(Err(err)) => Err(error::LocatedParseError::new(
1393 err,
1394 share_cursor(iter.lines.line_no(), bytes, start),
1395 frame_index,
1396 )),
1397 None => Err(error::LocatedParseError::new(
1398 error::ParseError::IncompleteFrame,
1399 file_line_at_bytes(bytes, start),
1400 frame_index,
1401 )),
1402 }
1403}
1404
1405fn copy_positions(frame: &types::ConFrame, dest: &mut [f64]) -> Result<(), error::ParseError> {
1406 if frame.positions.nrows().saturating_mul(3) != dest.len() {
1407 return Err(error::ParseError::InvalidCount(
1408 "frame atom count does not match the first frame".into(),
1409 ));
1410 }
1411 if let Some(src) = frame.positions.f64_slice() {
1412 dest.copy_from_slice(src);
1413 return Ok(());
1414 }
1415 for (i, slot) in dest.as_chunks_mut::<3>().0.iter_mut().enumerate() {
1416 slot.copy_from_slice(&frame.positions.as_f64_row(i));
1417 }
1418 Ok(())
1419}
1420
1421fn fill_position_block(
1422 bytes: &[u8],
1423 num_threads: Option<usize>,
1424) -> Result<PositionBlock, error::LocatedParseError> {
1425 match certified_columns(bytes, num_threads) {
1426 Ok(Some(columns)) => return Ok(columns.block),
1427 Ok(None) => {}
1428 Err(err) => return Err(err),
1429 }
1430 let boundaries = frame_starts(bytes)?;
1431 if boundaries.is_empty() {
1432 return Ok(PositionBlock {
1433 n_frames: 0,
1434 n_atoms: 0,
1435 values: Vec::new(),
1436 });
1437 }
1438 let end0 = boundaries.get(1).copied().unwrap_or(bytes.len());
1439 let first = parse_span_located(bytes, boundaries[0], end0, 0)?;
1440 let n_atoms = first.positions.nrows();
1441 let n_frames = boundaries.len();
1442 let stride = n_atoms.saturating_mul(3);
1443 let mut values = vec![0.0_f64; n_frames.saturating_mul(stride)];
1444 if stride == 0 {
1445 return Ok(PositionBlock {
1446 n_frames,
1447 n_atoms,
1448 values,
1449 });
1450 }
1451 fill_position_spans(bytes, &boundaries, n_atoms, num_threads, &mut values)?;
1452 Ok(PositionBlock {
1453 n_frames,
1454 n_atoms,
1455 values,
1456 })
1457}
1458
1459fn fill_position_spans(
1460 bytes: &[u8],
1461 boundaries: &[usize],
1462 n_atoms: usize,
1463 num_threads: Option<usize>,
1464 values: &mut [f64],
1465) -> Result<(), error::LocatedParseError> {
1466 let stride = n_atoms * 3;
1467 let write_one = |i: usize, dest: &mut [f64]| {
1468 let start = boundaries[i];
1469 let end = boundaries.get(i + 1).copied().unwrap_or(bytes.len());
1470 let frame = parse_span_located(bytes, start, end, i as u64)?;
1471 copy_positions(&frame, dest).map_err(|error| {
1472 error::LocatedParseError::new(error, file_line_at_bytes(bytes, start), i as u64)
1473 })
1474 };
1475 #[cfg(feature = "parallel")]
1476 {
1477 let parallel = !matches!(num_threads, Some(n) if n <= 1);
1478 if parallel {
1479 use rayon::prelude::*;
1480 let mut run = || match values
1485 .par_chunks_mut(stride)
1486 .enumerate()
1487 .find_map_first(|(i, dest)| write_one(i, dest).err())
1488 {
1489 Some(err) => Err(err),
1490 None => Ok(()),
1491 };
1492 return match num_threads {
1493 Some(n) if n >= 2 => {
1494 let pool = rayon::ThreadPoolBuilder::new()
1495 .num_threads(n)
1496 .build()
1497 .map_err(|err| {
1498 error::LocatedParseError::new(
1499 error::ParseError::ValidationError(err.to_string()),
1500 0,
1501 0,
1502 )
1503 })?;
1504 pool.install(run)
1505 }
1506 _ => run(),
1507 };
1508 }
1509 }
1510 #[cfg(not(feature = "parallel"))]
1511 {
1512 let _ = num_threads;
1513 }
1514 for (i, dest) in values.chunks_mut(stride).enumerate() {
1515 write_one(i, dest)?;
1516 }
1517 Ok(())
1518}
1519
1520pub fn frames_from_bytes(
1522 bytes: &[u8],
1523 num_threads: Option<usize>,
1524) -> Result<Vec<types::ConFrame>, error::ParseError> {
1525 #[cfg(feature = "parallel")]
1526 {
1527 if let Some(n) = num_threads.filter(|n| *n >= 2) {
1528 let parts = parse_frames_parallel_bytes(bytes, Some(n));
1529 let mut frames = Vec::with_capacity(parts.len());
1530 for r in parts {
1531 frames.push(r?);
1532 }
1533 return Ok(frames);
1534 }
1535 }
1536 #[cfg(not(feature = "parallel"))]
1537 {
1538 let _ = num_threads;
1539 }
1540 let mut frames = Vec::new();
1541 for r in ConFrameIterator::from_bytes(bytes) {
1542 frames.push(r?);
1543 }
1544 Ok(frames)
1545}
1546
1547fn frames_from_bytes_located(
1549 bytes: &[u8],
1550 num_threads: Option<usize>,
1551) -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
1552 #[cfg(feature = "parallel")]
1553 {
1554 if let Some(n) = num_threads.filter(|n| *n >= 2) {
1555 let timed = parse_bytes_timed(bytes, Some(n));
1556 return collect_located_slots(timed.slots, &timed.fail_lines);
1557 }
1558 }
1559 #[cfg(not(feature = "parallel"))]
1560 {
1561 let _ = num_threads;
1562 }
1563 collect_bytes_located(bytes)
1564}
1565
1566fn collect_bytes_located(bytes: &[u8]) -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
1568 let mut frames = Vec::new();
1569 let mut iter = ConFrameIterator::from_bytes(bytes);
1570 while let Some(item) = iter.next() {
1571 match item {
1572 Ok(frame) => frames.push(frame),
1573 Err(error) => {
1574 return Err(error::LocatedParseError::new(
1575 error,
1576 iter.lines.line_no(),
1577 iter.frame_index(),
1578 ));
1579 }
1580 }
1581 }
1582 Ok(frames)
1583}
1584
1585#[cfg(feature = "parallel")]
1587fn collect_located_slots(
1588 slots: Vec<Result<types::ConFrame, error::ParseError>>,
1589 fail_lines: &[(usize, u32)],
1590) -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
1591 let mut frames = Vec::with_capacity(slots.len());
1592 for (index, part) in slots.into_iter().enumerate() {
1593 match part {
1594 Ok(frame) => frames.push(frame),
1595 Err(error) => {
1596 let line = fail_lines
1597 .iter()
1598 .find(|(slot, _)| *slot == index)
1599 .map(|(_, line)| *line)
1600 .unwrap_or(0);
1601 return Err(error::LocatedParseError::new(error, line, index as u64));
1602 }
1603 }
1604 }
1605 Ok(frames)
1606}
1607
1608#[cfg(feature = "parallel")]
1609fn span_end(boundaries: &[usize], index: usize, len: usize) -> usize {
1610 if index + 1 < boundaries.len() {
1611 boundaries[index + 1]
1612 } else {
1613 len
1614 }
1615}
1616
1617#[cfg(feature = "parallel")]
1622#[derive(Copy, Clone)]
1623struct SlotWritePtr(*mut std::mem::MaybeUninit<Result<types::ConFrame, error::ParseError>>);
1624
1625#[cfg(feature = "parallel")]
1626impl SlotWritePtr {
1627 unsafe fn write_slot(self, index: usize, value: Result<types::ConFrame, error::ParseError>) {
1630 unsafe {
1631 (*self.0.add(index)).write(value);
1632 }
1633 }
1634}
1635
1636#[cfg(feature = "parallel")]
1640unsafe impl Send for SlotWritePtr {}
1641
1642#[cfg(feature = "parallel")]
1645unsafe impl Sync for SlotWritePtr {}
1646
1647#[cfg(feature = "parallel")]
1652fn thread_minflt() -> i64 {
1653 #[cfg(target_os = "linux")]
1654 {
1655 unsafe {
1656 let mut usage = std::mem::zeroed::<libc::rusage>();
1657 libc::getrusage(libc::RUSAGE_THREAD, &mut usage);
1658 usage.ru_minflt
1659 }
1660 }
1661 #[cfg(not(target_os = "linux"))]
1662 {
1663 0
1664 }
1665}
1666
1667#[cfg(feature = "parallel")]
1673fn sample_llc_misses() -> i64 {
1674 #[cfg(not(target_os = "linux"))]
1676 {
1677 return -1;
1678 }
1679 #[cfg(target_os = "linux")]
1680 {
1681 #[repr(C)]
1682 struct PerfEventAttr {
1683 typ: u32,
1684 size: u32,
1685 config: u64,
1686 rest: [u64; 14],
1687 }
1688 let attr = PerfEventAttr {
1689 typ: 3,
1690 size: std::mem::size_of::<PerfEventAttr>() as u32,
1691 config: 2 | (1 << 16),
1693 rest: [0; 14],
1694 };
1695 let fd = unsafe {
1696 libc::syscall(
1697 libc::SYS_perf_event_open,
1698 &attr as *const PerfEventAttr,
1699 0i32,
1700 -1i32,
1701 -1i32,
1702 0u64,
1703 )
1704 };
1705 if fd < 0 {
1706 return -1;
1707 }
1708 let fd = fd as libc::c_int;
1709 let mut value = 0u64;
1710 let n = unsafe { libc::read(fd, &mut value as *mut u64 as *mut libc::c_void, 8) };
1711 unsafe {
1712 libc::close(fd);
1713 }
1714 if n == 8 { value as i64 } else { -1 }
1715 }
1716}
1717
1718#[cfg(feature = "parallel")]
1724fn numa_remote_pages() -> i64 {
1725 let online = std::fs::read_to_string("/sys/devices/system/node/online").unwrap_or_default();
1726 let online = online.trim();
1727 let nodes = if let Some((start, end)) = online.split_once('-') {
1728 let start: i64 = start.parse().unwrap_or(0);
1729 let end: i64 = end.parse().unwrap_or(0);
1730 end - start + 1
1731 } else {
1732 1
1733 };
1734 if nodes <= 1 {
1735 return 0;
1736 }
1737 let text = std::fs::read_to_string("/proc/self/numa_maps").unwrap_or_default();
1738 let mut remote = 0i64;
1739 for line in text.lines() {
1740 for token in line.split_whitespace() {
1741 if let Some(rest) = token.strip_prefix('N')
1742 && let Some((node, pages)) = rest.split_once('=')
1743 && node != "0" {
1744 remote += pages.parse::<i64>().unwrap_or(0);
1745 }
1746 }
1747 }
1748 remote
1749}
1750
1751#[cfg(feature = "parallel")]
1758fn byte_balanced_ranges(boundaries: &[usize], len: usize, workers: usize) -> Vec<(usize, usize)> {
1759 let n = boundaries.len();
1760 if n == 0 {
1761 return Vec::new();
1762 }
1763 let workers = workers.max(1).min(n);
1764 let chunks = workers.saturating_mul(4).clamp(workers, n);
1765 let origin = boundaries[0];
1766 let total = len.saturating_sub(origin).max(1) as u128;
1767 let mut ranges = Vec::with_capacity(chunks);
1768 let mut start = 0usize;
1769 for chunk in 0..chunks {
1770 if start >= n {
1771 break;
1772 }
1773 let end = if chunk + 1 == chunks {
1774 n
1775 } else {
1776 let target = origin + ((total * (chunk as u128 + 1)) / chunks as u128) as usize;
1777 let mut idx = if start + 1 < n {
1778 boundaries[start + 1..].partition_point(|&b| b < target) + start + 1
1779 } else {
1780 n
1781 };
1782 let min_end = start + 1;
1783 let reserved = chunks - chunk - 1;
1784 let max_end = n.saturating_sub(reserved);
1785 if max_end >= min_end {
1786 idx = idx.clamp(min_end, max_end);
1787 } else {
1788 idx = min_end.min(n);
1789 }
1790 idx
1791 };
1792 if end > start {
1793 ranges.push((start, end));
1794 start = end;
1795 }
1796 }
1797 if start < n {
1798 match ranges.last_mut() {
1799 Some(last) => last.1 = n,
1800 None => ranges.push((start, n)),
1801 }
1802 }
1803 ranges
1804}
1805
1806#[cfg(feature = "parallel")]
1813fn parse_into_slots<F>(
1814 bytes: &[u8],
1815 len: usize,
1816 boundaries: &[usize],
1817 workers: usize,
1818 parse_span: F,
1819) -> Vec<Result<types::ConFrame, error::ParseError>>
1820where
1821 F: Fn(usize, usize) -> Result<types::ConFrame, error::ParseError> + Sync,
1822{
1823 parse_into_slots_timed(bytes, len, boundaries, workers, parse_span).slots
1824}
1825
1826#[cfg(feature = "parallel")]
1828struct SlotParse {
1829 slots: Vec<Result<types::ConFrame, error::ParseError>>,
1830 alloc_s: f64,
1831 work_s: f64,
1832 sum_s: f64,
1833 max_s: f64,
1834 chunk_s: Vec<f64>,
1836 worker_minflt: Vec<i64>,
1838 share_frames: Vec<u64>,
1840 share_bytes: Vec<u64>,
1842 fail_lines: Vec<(usize, u32)>,
1845}
1846
1847#[cfg(feature = "parallel")]
1855fn parse_into_slots_timed<F>(
1856 bytes: &[u8],
1857 len: usize,
1858 boundaries: &[usize],
1859 workers: usize,
1860 parse_span: F,
1861) -> SlotParse
1862where
1863 F: Fn(usize, usize) -> Result<types::ConFrame, error::ParseError> + Sync,
1864{
1865 let n = boundaries.len();
1866 if n == 0 {
1867 return SlotParse {
1868 slots: Vec::new(),
1869 alloc_s: 0.0,
1870 work_s: 0.0,
1871 sum_s: 0.0,
1872 max_s: 0.0,
1873 chunk_s: Vec::new(),
1874 worker_minflt: Vec::new(),
1875 share_frames: Vec::new(),
1876 share_bytes: Vec::new(),
1877 fail_lines: Vec::new(),
1878 };
1879 }
1880 if workers < 2 {
1881 let t_alloc = std::time::Instant::now();
1882 let mut slots = Vec::with_capacity(n);
1883 let alloc_s = t_alloc.elapsed().as_secs_f64();
1884 let flt0 = thread_minflt();
1885 let t_work = std::time::Instant::now();
1886 for i in 0..n {
1887 slots.push(parse_span(boundaries[i], span_end(boundaries, i, len)));
1888 }
1889 let work_s = t_work.elapsed().as_secs_f64();
1890 let worker_minflt = vec![thread_minflt().saturating_sub(flt0)];
1891 return SlotParse {
1892 slots,
1893 alloc_s,
1894 work_s,
1895 sum_s: work_s,
1896 max_s: work_s,
1897 chunk_s: vec![work_s],
1898 worker_minflt,
1899 share_frames: vec![n as u64],
1900 share_bytes: vec![len as u64],
1901 fail_lines: Vec::new(),
1902 };
1903 }
1904 let t_alloc = std::time::Instant::now();
1905 let mut slots = take_slot_buf(n);
1906 let alloc_s = t_alloc.elapsed().as_secs_f64();
1907 let ranges = byte_balanced_ranges(boundaries, len, workers);
1911 let chunk_ns: Vec<std::sync::atomic::AtomicU64> = (0..ranges.len())
1912 .map(|_| std::sync::atomic::AtomicU64::new(0))
1913 .collect();
1914 let chunk_flt: Vec<std::sync::atomic::AtomicI64> = (0..ranges.len())
1915 .map(|_| std::sync::atomic::AtomicI64::new(0))
1916 .collect();
1917 let slots_ptr = SlotWritePtr(slots.as_mut_ptr());
1918 let fail_lines = std::sync::Mutex::new(Vec::<(usize, u32)>::new());
1919 let t_work = std::time::Instant::now();
1920 {
1921 let parse_range = |range_index: usize, start: usize, end: usize| {
1922 if start >= end {
1923 return;
1924 }
1925 let flt0 = thread_minflt();
1926 let t_chunk = std::time::Instant::now();
1927 let byte_start = boundaries[start];
1928 let byte_end = span_end(boundaries, end - 1, len);
1929 let mut iter = ConFrameIterator::from_bytes(&bytes[byte_start..byte_end]);
1930 for i in start..end {
1931 let frame = match iter.next() {
1932 Some(Ok(frame)) => Ok(frame),
1933 Some(Err(error)) => {
1934 let line = share_cursor(iter.lines.line_no(), bytes, byte_start);
1935 fail_lines.lock().expect("fail lines").push((i, line));
1936 Err(error)
1937 }
1938 None => {
1939 let line = share_cursor(iter.lines.line_no(), bytes, byte_start);
1940 fail_lines.lock().expect("fail lines").push((i, line));
1941 Err(error::ParseError::IncompleteFrame)
1942 }
1943 };
1944 unsafe {
1947 slots_ptr.write_slot(i, frame);
1948 }
1949 }
1950 chunk_ns[range_index].store(
1951 t_chunk.elapsed().as_nanos() as u64,
1952 std::sync::atomic::Ordering::Relaxed,
1953 );
1954 chunk_flt[range_index].store(
1955 thread_minflt().saturating_sub(flt0),
1956 std::sync::atomic::Ordering::Relaxed,
1957 );
1958 };
1959 let run_ranges = || {
1960 use rayon::prelude::*;
1961 ranges
1962 .par_iter()
1963 .enumerate()
1964 .for_each(|(range_index, &(start, end))| {
1965 parse_range(range_index, start, end);
1966 });
1967 };
1968 if rayon::current_thread_index().is_some() {
1969 run_ranges();
1970 } else {
1971 match rayon::ThreadPoolBuilder::new().num_threads(workers).build() {
1972 Ok(pool) => pool.install(run_ranges),
1973 Err(_) => {
1974 for (range_index, &(start, end)) in ranges.iter().enumerate() {
1975 parse_range(range_index, start, end);
1976 }
1977 }
1978 }
1979 }
1980 }
1981 let work_s = t_work.elapsed().as_secs_f64();
1982 let mut worker_sum_s = 0.0_f64;
1983 let mut worker_max_s = 0.0_f64;
1984 let mut chunk_s = Vec::new();
1985 let mut worker_minflt = Vec::new();
1986 let mut share_frames = Vec::new();
1987 let mut share_bytes = Vec::new();
1988 for (range_index, &(start, end)) in ranges.iter().enumerate() {
1989 if start >= end {
1990 continue;
1991 }
1992 let seconds =
1993 chunk_ns[range_index].load(std::sync::atomic::Ordering::Relaxed) as f64 / 1.0e9;
1994 chunk_s.push(seconds);
1995 worker_sum_s += seconds;
1996 worker_max_s = worker_max_s.max(seconds);
1997 worker_minflt.push(chunk_flt[range_index].load(std::sync::atomic::Ordering::Relaxed));
1998 share_frames.push((end - start) as u64);
1999 let nbytes = span_end(boundaries, end - 1, len).saturating_sub(boundaries[start]);
2000 share_bytes.push(nbytes as u64);
2001 }
2002 let slots = unsafe {
2005 let mut raw = std::mem::ManuallyDrop::new(slots);
2006 Vec::from_raw_parts(
2007 raw.as_mut_ptr() as *mut Result<types::ConFrame, error::ParseError>,
2008 n,
2009 raw.capacity(),
2010 )
2011 };
2012 let fail_lines = fail_lines.into_inner().unwrap_or_default();
2013 SlotParse {
2014 slots,
2015 alloc_s,
2016 work_s,
2017 sum_s: worker_sum_s,
2018 max_s: worker_max_s,
2019 chunk_s,
2020 worker_minflt,
2021 share_frames,
2022 share_bytes,
2023 fail_lines,
2024 }
2025}
2026
2027#[cfg(feature = "parallel")]
2028fn parse_byte_span(
2029 bytes: &[u8],
2030 start: usize,
2031 end: usize,
2032) -> Result<types::ConFrame, error::ParseError> {
2033 let mut iter = ConFrameIterator::from_bytes(&bytes[start..end]);
2034 match iter.next() {
2035 Some(result) => result,
2036 None => Err(error::ParseError::IncompleteFrame),
2037 }
2038}
2039
2040#[cfg(feature = "parallel")]
2041fn parse_text_span(
2042 text: &str,
2043 start: usize,
2044 end: usize,
2045) -> Result<types::ConFrame, error::ParseError> {
2046 let mut iter = ConFrameIterator::new(&text[start..end]);
2047 match iter.next() {
2048 Some(result) => result,
2049 None => Err(error::ParseError::IncompleteFrame),
2050 }
2051}
2052
2053#[cfg(feature = "parallel")]
2054struct TimedParse {
2055 slots: Vec<Result<types::ConFrame, error::ParseError>>,
2056 pool: Option<std::sync::Arc<rayon::ThreadPool>>,
2059 pool_s: f64,
2060 pool_join_s: f64,
2062 scan_s: f64,
2063 scan_minflt: i64,
2064 slot_alloc_s: f64,
2065 worker_parse_s: f64,
2066 worker_sum_s: f64,
2067 worker_max_s: f64,
2068 chunk_s: Vec<f64>,
2069 worker_minflt: Vec<i64>,
2070 share_frames: Vec<u64>,
2071 share_bytes: Vec<u64>,
2072 fail_lines: Vec<(usize, u32)>,
2074 llc_misses: i64,
2075 numa_remote: i64,
2076 scan_kind: u8,
2078}
2079
2080#[cfg(feature = "parallel")]
2086fn keep_parse_pages() {
2087 #[cfg(target_os = "linux")]
2091 unsafe {
2092 libc::mallopt(-4, 0);
2093 libc::mallopt(-1, i32::MAX);
2094 }
2095}
2096
2097#[cfg(feature = "parallel")]
2103fn take_slot_buf(
2104 n: usize,
2105) -> Vec<std::mem::MaybeUninit<Result<types::ConFrame, error::ParseError>>> {
2106 let mut buf = slot_cache()
2107 .lock()
2108 .map(|mut slot| std::mem::take(&mut *slot))
2109 .unwrap_or_default();
2110 if buf.capacity() < n {
2111 buf = Vec::with_capacity(n);
2112 }
2113 unsafe { buf.set_len(n) };
2114 buf
2115}
2116
2117#[cfg(feature = "parallel")]
2118fn slot_cache()
2119-> &'static std::sync::Mutex<Vec<std::mem::MaybeUninit<Result<types::ConFrame, error::ParseError>>>>
2120{
2121 static CACHE: std::sync::Mutex<
2122 Vec<std::mem::MaybeUninit<Result<types::ConFrame, error::ParseError>>>,
2123 > = std::sync::Mutex::new(Vec::new());
2124 &CACHE
2125}
2126
2127#[cfg(feature = "parallel")]
2132#[allow(clippy::type_complexity)]
2133fn shared_pool(workers: usize) -> Option<std::sync::Arc<rayon::ThreadPool>> {
2134 use std::sync::{Arc, Mutex, OnceLock};
2135 static CACHE: OnceLock<Mutex<Vec<(usize, Arc<rayon::ThreadPool>)>>> = OnceLock::new();
2136 let cache = CACHE.get_or_init(|| Mutex::new(Vec::new()));
2137 let mut guard = cache.lock().ok()?;
2138 if let Some((_, pool)) = guard.iter().find(|(count, _)| *count == workers) {
2139 return Some(Arc::clone(pool));
2140 }
2141 keep_parse_pages();
2142 let pool = Arc::new(
2143 rayon::ThreadPoolBuilder::new()
2144 .num_threads(workers)
2145 .build()
2146 .ok()?,
2147 );
2148 guard.push((workers, Arc::clone(&pool)));
2149 Some(pool)
2150}
2151
2152#[cfg(feature = "parallel")]
2156pub fn process_minflt() -> i64 {
2157 #[cfg(unix)]
2158 {
2159 unsafe {
2160 let mut usage = std::mem::zeroed::<libc::rusage>();
2161 libc::getrusage(libc::RUSAGE_SELF, &mut usage);
2162 usage.ru_minflt
2163 }
2164 }
2165 #[cfg(not(unix))]
2166 {
2167 0
2168 }
2169}
2170
2171#[cfg(feature = "parallel")]
2173fn host_page_size() -> usize {
2174 #[cfg(unix)]
2175 {
2176 let raw = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
2177 if raw > 0 { raw as usize } else { 4096 }
2178 }
2179 #[cfg(not(unix))]
2180 {
2181 4096
2182 }
2183}
2184
2185#[cfg(feature = "parallel")]
2190fn first_touch_f64(values: &mut [f64], workers: usize) {
2191 use rayon::prelude::*;
2192 if values.is_empty() || workers < 2 {
2193 return;
2194 }
2195 let page = host_page_size().max(std::mem::size_of::<f64>());
2196 let step = (page / std::mem::size_of::<f64>()).max(1);
2197 let pages = values.len().div_ceil(step);
2198 let share = pages.div_ceil(workers);
2199 if rayon::current_thread_index().is_some() {
2200 values.par_chunks_mut(step).for_each(|chunk| {
2203 if let Some(slot) = chunk.first_mut() {
2204 *slot = 0.0;
2205 }
2206 });
2207 return;
2208 }
2209 std::thread::scope(|scope| {
2210 let mut rest = values;
2211 for worker in 0..workers {
2212 if rest.is_empty() {
2213 break;
2214 }
2215 let pages_here = share.min(rest.len().div_ceil(step));
2216 let take = (pages_here * step).min(rest.len());
2217 let (head, tail) = rest.split_at_mut(take);
2218 rest = tail;
2219 let _ = worker;
2220 scope.spawn(move || {
2221 let mut at = 0;
2222 while at < head.len() {
2223 head[at] = 0.0;
2224 at += step;
2225 }
2226 });
2227 }
2228 });
2229}
2230
2231#[cfg(feature = "parallel")]
2237fn finish_results(
2238 pool: Option<&rayon::ThreadPool>,
2239 slots: Vec<Result<types::ConFrame, error::ParseError>>,
2240) -> Result<Vec<types::ConFrame>, error::ParseError> {
2241 if slots.len() < 2 || pool.is_none() {
2242 let mut frames = Vec::with_capacity(slots.len());
2243 for part in slots {
2244 frames.push(part?);
2245 }
2246 return Ok(frames);
2247 }
2248 let pool = pool.expect("pool");
2249 pool.install(|| {
2250 use rayon::prelude::*;
2251 if let Some(index) = slots.par_iter().position_first(|item| item.is_err()) {
2252 return Err(match slots.into_iter().nth(index) {
2253 Some(Err(err)) => err,
2254 _ => error::ParseError::IncompleteFrame,
2255 });
2256 }
2257 Ok(slots
2258 .into_par_iter()
2259 .map(|item| item.expect("ok"))
2260 .collect())
2261 })
2262}
2263
2264#[cfg(feature = "parallel")]
2266pub const SCAN_SERIAL: u8 = 0;
2267#[cfg(feature = "parallel")]
2274pub const SCAN_INDEXED: u8 = 1;
2275#[cfg(feature = "parallel")]
2277pub const SCAN_EQUAL_SPAN: u8 = 2;
2278
2279#[cfg(feature = "parallel")]
2287#[cfg_attr(not(test), allow(dead_code))]
2288fn equal_span_boundaries(bytes: &[u8], workers: usize) -> Option<Vec<usize>> {
2289 if workers < 2 || bytes.len() < 2 {
2290 return None;
2291 }
2292 let mut first = ConFrameIterator::from_bytes(bytes);
2293 match first.forward_fast() {
2294 Some(Ok(())) => {}
2295 _ => return None,
2296 }
2297 let span = first.lines.pos;
2298 if span < 2 {
2299 return None;
2300 }
2301 let n = bytes.len() / span;
2302 if n < 2 {
2303 return None;
2304 }
2305 let tiled = n * span;
2308 if bytes[tiled..].iter().any(|b| !b.is_ascii_whitespace()) {
2309 return None;
2310 }
2311 let byte_share = tiled.div_ceil(workers);
2315 for worker in 0..workers {
2316 let byte_cut = worker.saturating_mul(byte_share).min(tiled);
2317 if byte_cut >= tiled {
2318 continue;
2319 }
2320 let index = if worker == 0 {
2321 0
2322 } else {
2323 byte_cut.div_ceil(span).min(n - 1)
2324 };
2325 let start = index.saturating_mul(span);
2326 if start >= tiled {
2327 continue;
2328 }
2329 let end = (start + span).min(tiled);
2330 let mut iter = ConFrameIterator::from_bytes(&bytes[start..end]);
2331 match iter.forward_fast() {
2332 Some(Ok(())) if iter.lines.pos == end - start => {}
2333 _ => return None,
2334 }
2335 }
2336 Some(arithmetic_starts(span, n))
2337}
2338
2339#[cfg(feature = "parallel")]
2344static START_CACHE: std::sync::Mutex<Vec<usize>> = std::sync::Mutex::new(Vec::new());
2345
2346#[cfg(feature = "parallel")]
2347#[cfg_attr(not(test), allow(dead_code))]
2348fn arithmetic_starts(span: usize, n: usize) -> Vec<usize> {
2349 let mut starts = START_CACHE
2350 .lock()
2351 .map(|mut slot| std::mem::take(&mut *slot))
2352 .unwrap_or_default();
2353 if starts.len() != n {
2354 starts.resize(n, 0);
2355 }
2356 for (index, slot) in starts.iter_mut().enumerate() {
2357 *slot = index * span;
2358 }
2359 starts
2360}
2361
2362#[cfg(feature = "parallel")]
2363fn recycle_starts(starts: Vec<usize>) {
2364 if let Ok(mut slot) = START_CACHE.lock()
2365 && starts.capacity() >= slot.capacity() {
2366 *slot = starts;
2367 }
2368}
2369
2370#[cfg(feature = "parallel")]
2371fn parse_frames_parallel_bytes(
2372 bytes: &[u8],
2373 num_threads: Option<usize>,
2374) -> Vec<Result<types::ConFrame, error::ParseError>> {
2375 parse_bytes_timed(bytes, num_threads).slots
2376}
2377
2378#[cfg(feature = "parallel")]
2385fn parse_bytes_timed(bytes: &[u8], num_threads: Option<usize>) -> TimedParse {
2386 let parse_workers = match num_threads {
2387 Some(n) if n >= 2 => n,
2388 _ => 1,
2389 };
2390 let run = || {
2391 let fault0 = process_minflt();
2392 let t_scan = std::time::Instant::now();
2393 let (boundaries, scan_kind) =
2397 if let Some(bounds) = equal_span_boundaries(bytes, parse_workers) {
2398 (bounds, SCAN_EQUAL_SPAN)
2399 } else {
2400 (
2401 collect_frame_starts(ConFrameIterator::from_bytes(bytes), true),
2402 SCAN_SERIAL,
2403 )
2404 };
2405 let scan_s = t_scan.elapsed().as_secs_f64();
2406 let scan_minflt = process_minflt().saturating_sub(fault0);
2407 let parsed = parse_into_slots_timed(
2408 bytes,
2409 bytes.len(),
2410 &boundaries,
2411 parse_workers,
2412 |start, end| parse_byte_span(bytes, start, end),
2413 );
2414 recycle_starts(boundaries);
2415 (parsed, scan_s, scan_kind, scan_minflt)
2416 };
2417 let (pool_s, pool, packed) = match num_threads {
2418 None => (0.0, None, run()),
2419 Some(n) if n <= 1 => {
2420 let _ = n;
2421 keep_parse_pages();
2422 let fault0 = process_minflt();
2423 let t_scan = std::time::Instant::now();
2424 let boundaries = collect_frame_starts(ConFrameIterator::from_bytes(bytes), true);
2425 let scan_s = t_scan.elapsed().as_secs_f64();
2426 let scan_minflt = process_minflt().saturating_sub(fault0);
2427 let parsed =
2428 parse_into_slots_timed(bytes, bytes.len(), &boundaries, 1, |start, end| {
2429 parse_byte_span(bytes, start, end)
2430 });
2431 (0.0, None, (parsed, scan_s, SCAN_SERIAL, scan_minflt))
2432 }
2433 Some(n) => {
2434 let t_pool = std::time::Instant::now();
2435 match shared_pool(n) {
2436 Some(pool) => {
2437 let built = t_pool.elapsed().as_secs_f64();
2438 let t_wake = std::time::Instant::now();
2439 let (wake, body) = pool.install(|| {
2440 let wake = t_wake.elapsed().as_secs_f64();
2441 (wake, run())
2442 });
2443 (built + wake, Some(pool), body)
2446 }
2447 None => (t_pool.elapsed().as_secs_f64(), None, run()),
2448 }
2449 }
2450 };
2451 let (parsed, scan_s, scan_kind, scan_minflt) = packed;
2452 TimedParse {
2453 slots: parsed.slots,
2454 pool,
2455 pool_s,
2456 pool_join_s: 0.0,
2457 scan_s,
2458 scan_minflt,
2459 slot_alloc_s: parsed.alloc_s,
2460 worker_parse_s: parsed.work_s,
2461 worker_sum_s: parsed.sum_s,
2462 worker_max_s: parsed.max_s,
2463 chunk_s: parsed.chunk_s,
2464 worker_minflt: parsed.worker_minflt,
2465 share_frames: parsed.share_frames,
2466 share_bytes: parsed.share_bytes,
2467 fail_lines: parsed.fail_lines,
2468 llc_misses: sample_llc_misses(),
2469 numa_remote: numa_remote_pages(),
2470 scan_kind,
2471 }
2472}
2473
2474pub fn count_frames(path: &Path) -> Result<usize, LoadError> {
2481 let contents = load_text(path)?;
2482 let text = contents.as_str().map_err(LoadError::Utf8)?;
2483 let mut n = 0usize;
2484 let mut iter = ConFrameIterator::new(text);
2485 loop {
2486 match iter.forward_fast() {
2487 Some(Ok(())) => n += 1,
2488 Some(Err(e)) => {
2489 return Err(LoadError::Parse(error::LocatedParseError::new(
2490 e,
2491 iter.lines.line_no(),
2492 iter.frame_index(),
2493 )));
2494 }
2495 None => break,
2496 }
2497 }
2498 Ok(n)
2499}
2500
2501pub fn read_first_frame(path: &Path) -> Result<types::ConFrame, Box<dyn std::error::Error>> {
2506 let contents = crate::compression::read_file_contents(path)?;
2507 let text = contents.as_str()?;
2508 let mut iter = ConFrameIterator::new(text);
2509 match iter.next() {
2510 Some(Ok(frame)) => Ok(frame),
2511 Some(Err(e)) => Err(Box::new(e)),
2512 None => Err("No frames found in file".into()),
2513 }
2514}
2515
2516pub fn read_nth_frame(
2520 path: &Path,
2521 index: usize,
2522) -> Result<types::ConFrame, Box<dyn std::error::Error>> {
2523 let contents = crate::compression::read_file_contents(path)?;
2524 let text = contents.as_str()?;
2525 read_nth_frame_from_text(text, index)
2526}
2527
2528pub fn read_nth_frame_from_text(
2530 text: &str,
2531 index: usize,
2532) -> Result<types::ConFrame, Box<dyn std::error::Error>> {
2533 let mut iter = ConFrameIterator::new(text);
2534 let skipped = iter.skip_frames(index)?;
2535 if skipped < index {
2536 return Err(format!("frame {index} out of range (only {skipped} frames)").into());
2537 }
2538 match iter.next() {
2539 Some(Ok(frame)) => Ok(frame),
2540 Some(Err(e)) => Err(Box::new(e)),
2541 None => Err(format!("frame {index} out of range").into()),
2542 }
2543}
2544
2545#[cfg(feature = "parallel")]
2551pub fn read_all_frames_indexed(
2552 path: &Path,
2553 num_threads: Option<usize>,
2554) -> Result<Vec<types::ConFrame>, Box<dyn std::error::Error>> {
2555 let contents = crate::compression::read_file_contents(path)?;
2556 let text = contents.as_str_sliced()?;
2557 let idx = match contents {
2558 crate::compression::FileContents::Mapped(_) => {
2559 crate::frame_index::load_or_build(path, text, true)?
2560 }
2561 crate::compression::FileContents::Owned(_) => {
2562 let on_disk = std::fs::metadata(path)?.len();
2563 crate::frame_index::load_or_build(path, text, on_disk == text.len() as u64)?
2564 }
2565 };
2566 let offs: Vec<usize> = idx.offsets.iter().map(|&o| o as usize).collect();
2567 parse_frames_at_offsets(text, &offs, num_threads)
2568 .into_iter()
2569 .map(|r| r.map_err(|e| Box::new(e) as Box<dyn std::error::Error>))
2570 .collect()
2571}
2572
2573enum IndexStep {
2574 Stay(usize),
2575 Missing(usize),
2576 Landed(usize, usize),
2577 Short(usize, usize),
2578}
2579
2580struct IndexedSpan {
2581 start: usize,
2582 content_end: usize,
2583 next_pos: usize,
2584}
2585
2586fn line_count(nl: &[usize], len: usize) -> usize {
2588 if len == 0 {
2589 0
2590 } else if nl.last().copied() == Some(len - 1) {
2591 nl.len()
2592 } else {
2593 nl.len() + 1
2594 }
2595}
2596
2597fn line_start_pos(nl: &[usize], len: usize, line_no: usize) -> usize {
2598 if line_no == 0 {
2599 return 0;
2600 }
2601 if line_no >= line_count(nl, len) {
2602 return len;
2603 }
2604 nl[line_no - 1] + 1
2605}
2606
2607fn line_no_for_pos(nl: &[usize], len: usize, pos: usize) -> usize {
2609 if pos >= len {
2610 return line_count(nl, len);
2611 }
2612 nl.partition_point(|&p| p < pos)
2613}
2614
2615fn indexed_span(nl: &[usize], bytes: &[u8], line_no: usize) -> Option<IndexedSpan> {
2616 let len = bytes.len();
2617 if line_no >= line_count(nl, len) {
2618 return None;
2619 }
2620 let start = if line_no == 0 { 0 } else { nl[line_no - 1] + 1 };
2621 let (raw_end, next_pos) = if line_no < nl.len() {
2622 (nl[line_no], nl[line_no] + 1)
2623 } else {
2624 (len, len)
2625 };
2626 let mut content_end = raw_end;
2627 if content_end > start && bytes[content_end - 1] == b'\r' {
2628 content_end -= 1;
2629 }
2630 Some(IndexedSpan {
2631 start,
2632 content_end,
2633 next_pos,
2634 })
2635}
2636
2637#[cfg(feature = "parallel")]
2639fn equal_cuts(len: usize, n: usize) -> Vec<usize> {
2640 let n = n.max(1);
2641 let mut cuts = Vec::with_capacity(n + 1);
2642 for i in 0..=n {
2643 let at = ((i as u128) * (len as u128) / (n as u128)) as usize;
2644 let at = at.min(len);
2645 if cuts.last().copied() != Some(at) {
2646 cuts.push(at);
2647 }
2648 }
2649 if cuts.last().copied() != Some(len) {
2650 cuts.push(len);
2651 }
2652 cuts
2653}
2654
2655#[cfg(feature = "parallel")]
2662fn newline_positions_at_cuts(bytes: &[u8], cuts: &[usize]) -> Vec<usize> {
2663 let spans: Vec<(usize, usize)> = cuts
2664 .windows(2)
2665 .map(|w| (w[0].min(bytes.len()), w[1].min(bytes.len())))
2666 .filter(|(start, end)| start < end)
2667 .collect();
2668 if spans.len() <= 1 {
2669 let (start, end) = spans.first().copied().unwrap_or((0, 0));
2670 return memchr::memchr_iter(b'\n', &bytes[start..end])
2671 .map(|p| start + p)
2672 .collect();
2673 }
2674 use rayon::prelude::*;
2678 let parts: Vec<Vec<usize>> = spans
2679 .par_iter()
2680 .map(|&(start, end)| {
2681 memchr::memchr_iter(b'\n', &bytes[start..end])
2682 .map(|p| start + p)
2683 .collect()
2684 })
2685 .collect();
2686 let mut all = Vec::with_capacity(bytes.len() / 32);
2687 for part in parts {
2688 all.extend(part);
2689 }
2690 all
2691}
2692
2693#[cfg(feature = "parallel")]
2694fn newline_positions(bytes: &[u8], threads: usize) -> Vec<usize> {
2695 newline_positions_at_cuts(bytes, &equal_cuts(bytes.len(), threads))
2696}
2697
2698#[cfg(feature = "parallel")]
2708fn scan_thread_count(num_threads: Option<usize>, len: usize) -> usize {
2709 let requested = match num_threads {
2710 Some(n) if n >= 2 => n,
2711 _ => return 1,
2712 };
2713 if len < 2 { 1 } else { requested }
2714}
2715
2716#[derive(Debug)]
2718pub enum LoadError {
2719 Io(std::io::Error),
2720 Utf8(std::str::Utf8Error),
2721 Parse(error::LocatedParseError),
2722 Empty,
2724}
2725
2726impl std::fmt::Display for LoadError {
2727 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2728 match self {
2729 LoadError::Io(e) => write!(f, "{e}"),
2730 LoadError::Utf8(e) => write!(f, "{e}"),
2731 LoadError::Parse(e) => write!(f, "{e}"),
2732 LoadError::Empty => write!(f, "no frames found in file"),
2733 }
2734 }
2735}
2736
2737impl std::error::Error for LoadError {}
2738
2739fn load_error_from_box(err: Box<dyn std::error::Error>) -> LoadError {
2740 if let Some(io) = err.downcast_ref::<std::io::Error>() {
2741 return LoadError::Io(std::io::Error::new(io.kind(), io.to_string()));
2742 }
2743 LoadError::Io(std::io::Error::other(err.to_string()))
2744}
2745
2746fn load_text(path: &Path) -> Result<crate::compression::FileContents, LoadError> {
2747 crate::compression::read_file_contents(path).map_err(load_error_from_box)
2748}
2749
2750#[cfg(feature = "parallel")]
2752fn file_line_at(text: &str, byte: usize) -> u32 {
2753 let end = byte.min(text.len());
2754 1 + text.as_bytes()[..end]
2755 .iter()
2756 .filter(|&&b| b == b'\n')
2757 .count() as u32
2758}
2759
2760pub fn frames_from_text_located(
2762 text: &str,
2763 num_threads: Option<usize>,
2764) -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
2765 #[cfg(feature = "parallel")]
2766 {
2767 let use_parallel = matches!(num_threads, Some(n) if n >= 2);
2770 if use_parallel {
2771 let pool = num_threads.filter(|&n| n >= 2);
2772 return collect_parallel_located(text, pool);
2773 }
2774 }
2775 #[cfg(not(feature = "parallel"))]
2776 {
2777 let _ = num_threads;
2778 }
2779 let mut frames = Vec::new();
2780 let mut iter = ConFrameIterator::new(text);
2781 let mut index = 0u64;
2782 while let Some(item) = iter.next() {
2783 match item {
2784 Ok(frame) => {
2785 frames.push(frame);
2786 index += 1;
2787 }
2788 Err(e) => {
2789 return Err(error::LocatedParseError::new(
2790 e,
2791 iter.lines.line_no(),
2792 index,
2793 ));
2794 }
2795 }
2796 }
2797 Ok(frames)
2798}
2799
2800pub fn read_all_frames_located(
2802 path: &Path,
2803 num_threads: Option<usize>,
2804) -> Result<Vec<types::ConFrame>, LoadError> {
2805 let contents = load_text(path)?;
2806 let text = contents.as_str().map_err(LoadError::Utf8)?;
2807 frames_from_text_located(text, num_threads).map_err(LoadError::Parse)
2808}
2809
2810pub fn read_first_frame_located(path: &Path) -> Result<types::ConFrame, LoadError> {
2812 let contents = load_text(path)?;
2813 let text = contents.as_str().map_err(LoadError::Utf8)?;
2814 let mut iter = ConFrameIterator::new(text);
2815 match iter.next() {
2816 Some(Ok(frame)) => Ok(frame),
2817 Some(Err(e)) => Err(LoadError::Parse(error::LocatedParseError::new(
2818 e,
2819 iter.lines.line_no(),
2820 0,
2821 ))),
2822 None => Err(LoadError::Empty),
2823 }
2824}
2825
2826pub fn read_nth_frame_located(path: &Path, index: usize) -> Result<types::ConFrame, LoadError> {
2828 let contents = load_text(path)?;
2829 let text = contents.as_str().map_err(LoadError::Utf8)?;
2830 let mut iter = ConFrameIterator::new(text);
2831 match iter.skip_frames_counted(index) {
2832 Ok(skipped) if skipped == index => {}
2833 Ok(skipped) => {
2834 return Err(LoadError::Parse(error::LocatedParseError::new(
2835 error::ParseError::IncompleteFrame,
2836 iter.lines.line_no(),
2837 skipped as u64,
2838 )));
2839 }
2840 Err((failed_at, e)) => {
2841 return Err(LoadError::Parse(error::LocatedParseError::new(
2842 e,
2843 iter.lines.line_no(),
2844 failed_at as u64,
2845 )));
2846 }
2847 }
2848 match iter.next() {
2849 Some(Ok(frame)) => Ok(frame),
2850 Some(Err(e)) => Err(LoadError::Parse(error::LocatedParseError::new(
2851 e,
2852 iter.lines.line_no(),
2853 index as u64,
2854 ))),
2855 None => Err(LoadError::Parse(error::LocatedParseError::new(
2856 error::ParseError::IncompleteFrame,
2857 iter.lines.line_no(),
2858 index as u64,
2859 ))),
2860 }
2861}
2862
2863fn collect_frame_starts(scanner: ConFrameIterator<'_>, skip_whitespace_tail: bool) -> Vec<usize> {
2864 #[cfg(target_arch = "x86_64")]
2865 {
2866 if std::is_x86_feature_detected!("avx2") {
2867 return unsafe { collect_frame_starts_avx2(scanner, skip_whitespace_tail) };
2871 }
2872 }
2873 collect_frame_starts_inner(scanner, skip_whitespace_tail)
2874}
2875
2876#[cfg(target_arch = "x86_64")]
2877#[target_feature(enable = "avx2,popcnt")]
2878unsafe fn collect_frame_starts_avx2(
2879 scanner: ConFrameIterator<'_>,
2880 skip_whitespace_tail: bool,
2881) -> Vec<usize> {
2882 collect_frame_starts_inner(scanner, skip_whitespace_tail)
2883}
2884
2885#[inline(always)]
2886fn collect_frame_starts_inner(
2887 mut scanner: ConFrameIterator<'_>,
2888 skip_whitespace_tail: bool,
2889) -> Vec<usize> {
2890 let mut boundaries = Vec::new();
2891 loop {
2892 scanner.lines.clear_peek();
2893 let start = scanner.lines.pos;
2894 if start >= scanner.lines.bytes.len() {
2895 break;
2896 }
2897 if skip_whitespace_tail {
2898 let rest = &scanner.lines.bytes[start..];
2899 if rest.iter().all(u8::is_ascii_whitespace) {
2900 break;
2901 }
2902 }
2903 boundaries.push(start);
2904 match scanner.forward_fast_to(None) {
2905 Some(Ok(())) => {}
2906 Some(Err(_)) | None => break,
2907 }
2908 }
2909 boundaries
2910}
2911
2912pub fn frame_start_offsets_serial(file_contents: &str) -> Vec<usize> {
2914 collect_frame_starts(ConFrameIterator::new(file_contents), false)
2915}
2916
2917#[cfg(feature = "parallel")]
2918fn scanner_with_newlines<'a>(text: &'a str, newlines: Vec<usize>) -> ConFrameIterator<'a> {
2919 let mut scanner = ConFrameIterator::new(text);
2920 scanner.lines.index_line = 0;
2921 scanner.lines.newlines = Some(newlines);
2922 scanner
2923}
2924
2925#[cfg(feature = "parallel")]
2926fn frame_start_offsets_indexed(text: &str, newlines: Vec<usize>) -> Vec<usize> {
2927 collect_frame_starts(scanner_with_newlines(text, newlines), false)
2928}
2929
2930#[cfg(all(feature = "parallel", test))]
2936fn frame_start_offsets_at_cuts(text: &str, cuts: &[usize]) -> Vec<usize> {
2937 let newlines = newline_positions_at_cuts(text.as_bytes(), cuts);
2938 frame_start_offsets_indexed(text, newlines)
2939}
2940
2941pub fn frame_start_offsets(file_contents: &str) -> Vec<usize> {
2953 frame_start_offsets_with_threads(file_contents, None)
2954}
2955
2956pub fn frame_start_offsets_with_threads(
2964 file_contents: &str,
2965 num_threads: Option<usize>,
2966) -> Vec<usize> {
2967 #[cfg(feature = "parallel")]
2968 {
2969 let threads = scan_thread_count(num_threads, file_contents.len());
2970 if threads >= 2 {
2971 let newlines = newline_positions(file_contents.as_bytes(), threads);
2972 return frame_start_offsets_indexed(file_contents, newlines);
2973 }
2974 }
2975 #[cfg(not(feature = "parallel"))]
2976 {
2977 let _ = num_threads;
2978 }
2979 frame_start_offsets_serial(file_contents)
2980}
2981
2982#[cfg(feature = "parallel")]
2991pub fn parse_frames_parallel(
2992 file_contents: &str,
2993) -> Vec<Result<types::ConFrame, error::ParseError>> {
2994 parse_frames_parallel_with_threads(file_contents, None)
2995}
2996
2997#[cfg(feature = "parallel")]
3007pub fn parse_frames_parallel_with_threads(
3008 file_contents: &str,
3009 num_threads: Option<usize>,
3010) -> Vec<Result<types::ConFrame, error::ParseError>> {
3011 let parse_workers = match num_threads {
3012 Some(n) if n >= 2 => n,
3013 _ => 1,
3014 };
3015 let run = || {
3016 let boundaries = collect_frame_starts(ConFrameIterator::new(file_contents), true);
3020 parse_into_slots(
3021 file_contents.as_bytes(),
3022 file_contents.len(),
3023 &boundaries,
3024 parse_workers,
3025 |start, end| parse_text_span(file_contents, start, end),
3026 )
3027 };
3028 match num_threads {
3029 Some(n) if n >= 2 => match rayon::ThreadPoolBuilder::new().num_threads(n).build() {
3030 Ok(pool) => pool.install(run),
3031 Err(_) => run(),
3032 },
3033 _ => run(),
3034 }
3035}
3036
3037#[cfg(feature = "parallel")]
3052pub fn parse_frames_at_offsets(
3053 file_contents: &str,
3054 boundaries: &[usize],
3055 num_threads: Option<usize>,
3056) -> Vec<Result<types::ConFrame, error::ParseError>> {
3057 let workers = match num_threads {
3058 Some(n) if n <= 1 => 1,
3059 Some(n) => n,
3060 None => rayon::current_num_threads().max(1),
3061 };
3062 let run = || {
3063 parse_into_slots(
3064 file_contents.as_bytes(),
3065 file_contents.len(),
3066 boundaries,
3067 workers,
3068 |start, end| parse_text_span(file_contents, start, end),
3069 )
3070 };
3071 match num_threads {
3072 Some(n) if n >= 2 => match rayon::ThreadPoolBuilder::new().num_threads(n).build() {
3073 Ok(pool) => pool.install(run),
3074 Err(_) => run(),
3075 },
3076 _ => run(),
3077 }
3078}
3079
3080#[derive(Clone, Copy, Debug, Default)]
3083pub struct ReadPhaseTimes {
3084 pub open_mmap_s: f64,
3085 pub pool_s: f64,
3087 pub pool_join_s: f64,
3090 pub scan_s: f64,
3091 pub slot_alloc_s: f64,
3092 pub worker_parse_s: f64,
3093 pub finish_s: f64,
3095 pub worker_sum_s: f64,
3097 pub worker_max_s: f64,
3099 pub close_s: f64,
3101 pub schedule_s: f64,
3103 pub scan_minflt: i64,
3105 pub parse_minflt: i64,
3107 pub llc_misses: i64,
3109 pub numa_remote: i64,
3111 pub slab_minflt: i64,
3113 pub touch_minflt: i64,
3115 pub mapped: bool,
3117}
3118
3119pub struct PhasedFrames {
3121 pub frames: Vec<types::ConFrame>,
3122 pub phases: ReadPhaseTimes,
3123 pub n_bytes: usize,
3124 pub chunk_s: Vec<f64>,
3126 pub scan_kind: u8,
3129 pub worker_minflt: Vec<i64>,
3131 pub share_frames: Vec<u64>,
3133 pub share_bytes: Vec<u64>,
3135 #[cfg(feature = "parallel")]
3137 pub pool: Option<std::sync::Arc<rayon::ThreadPool>>,
3138}
3139
3140pub fn read_all_frames_phased(
3150 path: &Path,
3151 num_threads: Option<usize>,
3152) -> Result<PhasedFrames, Box<dyn std::error::Error>> {
3153 let t_all = std::time::Instant::now();
3154 let t_open = std::time::Instant::now();
3155 let contents = crate::compression::read_file_contents(path)?;
3156 let open_mmap_s = t_open.elapsed().as_secs_f64();
3157 let mapped = matches!(contents, crate::compression::FileContents::Mapped(_));
3158 let n_bytes = match &contents {
3159 crate::compression::FileContents::Owned(text) => text.len(),
3160 crate::compression::FileContents::Mapped(map) => map.len(),
3161 };
3162 let bytes: &[u8] = match &contents {
3163 crate::compression::FileContents::Owned(text) => text.as_bytes(),
3164 crate::compression::FileContents::Mapped(map) => map,
3165 };
3166 let phased = frames_from_bytes_phased_located(bytes, num_threads)?;
3167 let workers = match num_threads {
3168 Some(n) if n >= 2 => n,
3169 _ => 1,
3170 };
3171 let t_close = std::time::Instant::now();
3172 if let crate::compression::FileContents::Mapped(map) = &contents {
3173 if workers >= 2 {
3178 #[cfg(feature = "parallel")]
3179 {
3180 if let Some(pool) = phased.pool.as_ref() {
3181 pool.install(|| {
3182 crate::compression::advise_dontneed_parallel(map, workers);
3183 });
3184 }
3185 }
3186 #[cfg(not(feature = "parallel"))]
3187 {
3188 let _ = (map, workers);
3189 }
3190 }
3191 }
3192 drop(contents);
3193 let close_s = t_close.elapsed().as_secs_f64();
3194 let elapsed = t_all.elapsed().as_secs_f64();
3195 let named = open_mmap_s
3196 + phased.pool_s
3197 + phased.pool_join_s
3198 + phased.scan_s
3199 + phased.slot_alloc_s
3200 + phased.worker_parse_s
3201 + phased.finish_s
3202 + close_s;
3203 let schedule_s = (elapsed - named).max(0.0);
3204 let parse_minflt = phased.worker_minflt.iter().sum();
3205 Ok(PhasedFrames {
3206 frames: phased.frames,
3207 n_bytes,
3208 chunk_s: phased.chunk_s,
3209 scan_kind: phased.scan_kind,
3210 worker_minflt: phased.worker_minflt,
3211 share_frames: phased.share_frames,
3212 share_bytes: phased.share_bytes,
3213 phases: ReadPhaseTimes {
3214 open_mmap_s,
3215 pool_s: phased.pool_s,
3216 pool_join_s: phased.pool_join_s,
3217 scan_s: phased.scan_s,
3218 slot_alloc_s: phased.slot_alloc_s,
3219 worker_parse_s: phased.worker_parse_s,
3220 finish_s: phased.finish_s,
3221 worker_sum_s: phased.worker_sum_s,
3222 worker_max_s: phased.worker_max_s,
3223 close_s,
3224 schedule_s,
3225 scan_minflt: phased.scan_minflt,
3226 parse_minflt,
3227 llc_misses: phased.llc_misses,
3228 numa_remote: phased.numa_remote,
3229 mapped,
3230 ..ReadPhaseTimes::default()
3231 },
3232 #[cfg(feature = "parallel")]
3233 pool: phased.pool,
3234 })
3235}
3236
3237pub struct PhasedByteTimes {
3243 pub frames: Vec<types::ConFrame>,
3244 pub pool_s: f64,
3245 pub pool_join_s: f64,
3246 pub scan_s: f64,
3247 pub slot_alloc_s: f64,
3248 pub worker_parse_s: f64,
3249 pub finish_s: f64,
3250 pub worker_sum_s: f64,
3251 pub worker_max_s: f64,
3252 pub chunk_s: Vec<f64>,
3254 pub scan_kind: u8,
3257 pub scan_minflt: i64,
3259 pub worker_minflt: Vec<i64>,
3261 pub share_frames: Vec<u64>,
3262 pub share_bytes: Vec<u64>,
3263 pub llc_misses: i64,
3265 pub numa_remote: i64,
3267 #[cfg(feature = "parallel")]
3269 pub pool: Option<std::sync::Arc<rayon::ThreadPool>>,
3270}
3271
3272pub fn frames_from_bytes_phased(
3273 bytes: &[u8],
3274 num_threads: Option<usize>,
3275) -> Result<PhasedByteTimes, error::ParseError> {
3276 frames_from_bytes_phased_located(bytes, num_threads).map_err(|err| err.error)
3277}
3278
3279fn frames_from_bytes_phased_located(
3280 bytes: &[u8],
3281 num_threads: Option<usize>,
3282) -> Result<PhasedByteTimes, error::LocatedParseError> {
3283 #[cfg(feature = "parallel")]
3284 {
3285 if !matches!(num_threads, Some(n) if n >= 2) {
3288 let flt0 = thread_minflt();
3289 let t_work = std::time::Instant::now();
3290 let frames = collect_bytes_located(bytes)?;
3291 let worker_parse_s = t_work.elapsed().as_secs_f64();
3292 let worker_flt = thread_minflt().saturating_sub(flt0);
3293 let n_frames = frames.len() as u64;
3294 return Ok(PhasedByteTimes {
3295 frames,
3296 pool_s: 0.0,
3297 pool_join_s: 0.0,
3298 scan_s: 0.0,
3299 slot_alloc_s: 0.0,
3300 worker_parse_s,
3301 finish_s: 0.0,
3302 worker_sum_s: worker_parse_s,
3303 worker_max_s: worker_parse_s,
3304 chunk_s: vec![worker_parse_s],
3305 scan_kind: SCAN_SERIAL,
3306 scan_minflt: 0,
3307 worker_minflt: vec![worker_flt],
3308 share_frames: vec![n_frames],
3309 share_bytes: vec![bytes.len() as u64],
3310 llc_misses: sample_llc_misses(),
3311 numa_remote: numa_remote_pages(),
3312 pool: None,
3313 });
3314 }
3315 let requested = Some(num_threads.expect("workers"));
3316 let timed = parse_bytes_timed(bytes, requested);
3317 let t_finish = std::time::Instant::now();
3318 let frames = if timed.slots.iter().any(|item| item.is_err()) {
3319 collect_located_slots(timed.slots, &timed.fail_lines)?
3320 } else {
3321 finish_results(timed.pool.as_deref(), timed.slots)
3322 .map_err(|error| error::LocatedParseError::new(error, 0, 0))?
3323 };
3324 let finish_s = t_finish.elapsed().as_secs_f64();
3325 Ok(PhasedByteTimes {
3326 frames,
3327 pool_s: timed.pool_s,
3328 pool_join_s: timed.pool_join_s,
3329 scan_s: timed.scan_s,
3330 slot_alloc_s: timed.slot_alloc_s,
3331 worker_parse_s: timed.worker_parse_s,
3332 finish_s,
3333 worker_sum_s: timed.worker_sum_s,
3334 worker_max_s: timed.worker_max_s,
3335 chunk_s: timed.chunk_s,
3336 scan_kind: timed.scan_kind,
3337 scan_minflt: timed.scan_minflt,
3338 worker_minflt: timed.worker_minflt,
3339 share_frames: timed.share_frames,
3340 share_bytes: timed.share_bytes,
3341 llc_misses: timed.llc_misses,
3342 numa_remote: timed.numa_remote,
3343 pool: timed.pool,
3344 })
3345 }
3346 #[cfg(not(feature = "parallel"))]
3347 {
3348 let _ = num_threads;
3349 let t_work = std::time::Instant::now();
3350 let frames = collect_bytes_located(bytes)?;
3351 let worker_parse_s = t_work.elapsed().as_secs_f64();
3352 Ok(PhasedByteTimes {
3353 frames,
3354 pool_s: 0.0,
3355 pool_join_s: 0.0,
3356 scan_s: 0.0,
3357 slot_alloc_s: 0.0,
3358 worker_parse_s,
3359 finish_s: 0.0,
3360 worker_sum_s: worker_parse_s,
3361 worker_max_s: worker_parse_s,
3362 chunk_s: vec![worker_parse_s],
3363 scan_kind: 0,
3364 scan_minflt: 0,
3365 worker_minflt: Vec::new(),
3366 share_frames: Vec::new(),
3367 share_bytes: Vec::new(),
3368 llc_misses: -1,
3369 numa_remote: 0,
3370 })
3371 }
3372}
3373
3374pub fn pack_position_slab(frames: &[types::ConFrame], workers: usize) -> Option<Vec<f64>> {
3380 if frames.is_empty() {
3381 return None;
3382 }
3383 let n_atoms = frames[0].positions.nrows();
3384 if frames
3385 .iter()
3386 .any(|frame| frame.positions.nrows() != n_atoms)
3387 {
3388 return None;
3389 }
3390 let stride = n_atoms.saturating_mul(3);
3391 let total = frames.len().saturating_mul(stride);
3392 let mut values = Vec::<f64>::with_capacity(total);
3393 if total > 0 {
3394 unsafe {
3398 values.set_len(total);
3399 }
3400 }
3401 if stride == 0 {
3402 return Some(values);
3403 }
3404 let write = |frame: &types::ConFrame, dest: &mut [f64]| {
3405 if let Some(src) = frame.positions.f64_slice() {
3406 dest.copy_from_slice(src);
3407 } else {
3408 for (row, slot) in dest.as_chunks_mut::<3>().0.iter_mut().enumerate() {
3409 slot.copy_from_slice(&frame.positions.as_f64_row(row));
3410 }
3411 }
3412 };
3413 #[cfg(feature = "parallel")]
3414 {
3415 if workers >= 2 {
3416 let chunk_frames = frames.len().div_ceil(workers).max(1);
3417 let chunk_vals = chunk_frames.saturating_mul(stride);
3418 let copy_chunks = |dest_all: &mut [f64]| {
3419 use rayon::prelude::*;
3420 dest_all
3421 .par_chunks_mut(chunk_vals)
3422 .enumerate()
3423 .for_each(|(chunk_index, dest)| {
3424 let base = chunk_index * chunk_frames;
3425 for (offset, slot) in dest.chunks_exact_mut(stride).enumerate() {
3426 let index = base + offset;
3427 if index < frames.len() {
3428 write(&frames[index], slot);
3429 }
3430 }
3431 });
3432 };
3433 if rayon::current_thread_index().is_some() {
3437 first_touch_f64(&mut values, workers);
3438 copy_chunks(&mut values);
3439 } else {
3440 first_touch_f64(&mut values, workers);
3441 std::thread::scope(|scope| {
3442 let mut rest = &mut values[..];
3443 let mut base = 0usize;
3444 let mut handles = Vec::new();
3445 while base < frames.len() {
3446 let take_frames = chunk_frames.min(frames.len() - base);
3447 let take_vals = take_frames * stride;
3448 let (dest, tail) = rest.split_at_mut(take_vals);
3449 rest = tail;
3450 let frame_slice = &frames[base..base + take_frames];
3451 handles.push(scope.spawn(move || {
3452 for (frame, slot) in
3453 frame_slice.iter().zip(dest.chunks_exact_mut(stride))
3454 {
3455 write(frame, slot);
3456 }
3457 }));
3458 base += take_frames;
3459 }
3460 for handle in handles {
3461 handle.join().expect("slab worker");
3462 }
3463 });
3464 }
3465 return Some(values);
3466 }
3467 }
3468 #[cfg(not(feature = "parallel"))]
3469 {
3470 let _ = workers;
3471 }
3472 for (frame, dest) in frames.iter().zip(values.chunks_exact_mut(stride)) {
3473 write(frame, dest);
3474 }
3475 Some(values)
3476}
3477
3478fn mix_touch(hash: u64, value: u64) -> u64 {
3479 hash.wrapping_mul(0x9E3779B185EBCA87) ^ value
3480}
3481
3482fn one_frame_touch(frame: &types::ConFrame, coords: &[f64], index: u64) -> u64 {
3483 let mut hash = index;
3484 for value in frame.header.boxl {
3485 hash = mix_touch(hash, value.to_bits());
3486 }
3487 for atom in &frame.atom_data {
3488 hash = mix_touch(hash, atom.symbol.len() as u64);
3489 for byte in atom.symbol.as_bytes() {
3490 hash = mix_touch(hash, u64::from(*byte));
3491 }
3492 }
3493 for value in coords {
3494 hash = mix_touch(hash, value.to_bits());
3495 }
3496 hash
3497}
3498
3499pub fn touch_frame_checksum(
3506 frames: &[types::ConFrame],
3507 positions: &[f64],
3508 workers: usize,
3509) -> Option<u64> {
3510 if frames.is_empty() {
3511 return Some(0);
3512 }
3513 let n_atoms = frames[0].atom_data.len();
3514 let stride = n_atoms.saturating_mul(3);
3515 if stride == 0 || positions.len() != frames.len().saturating_mul(stride) {
3516 return None;
3517 }
3518 if frames.iter().any(|frame| frame.atom_data.len() != n_atoms) {
3519 return None;
3520 }
3521 let one = |index: usize| {
3522 let start = index * stride;
3523 one_frame_touch(
3524 &frames[index],
3525 &positions[start..start + stride],
3526 index as u64,
3527 )
3528 };
3529 Some(reduce_touch(frames.len(), workers, one))
3530}
3531
3532pub fn touch_option_checksum(
3536 frames: &[Option<types::ConFrame>],
3537 positions: &[f64],
3538 workers: usize,
3539) -> Option<u64> {
3540 if frames.is_empty() {
3541 return Some(0);
3542 }
3543 if frames.iter().any(Option::is_none) {
3544 return None;
3545 }
3546 let n_atoms = frames[0].as_ref()?.atom_data.len();
3547 let stride = n_atoms.saturating_mul(3);
3548 if stride == 0 || positions.len() != frames.len().saturating_mul(stride) {
3549 return None;
3550 }
3551 if frames
3552 .iter()
3553 .any(|frame| frame.as_ref().map(|item| item.atom_data.len()) != Some(n_atoms))
3554 {
3555 return None;
3556 }
3557 let one = |index: usize| {
3558 let start = index * stride;
3559 one_frame_touch(
3560 frames[index].as_ref().expect("slot"),
3561 &positions[start..start + stride],
3562 index as u64,
3563 )
3564 };
3565 Some(reduce_touch(frames.len(), workers, one))
3566}
3567
3568fn reduce_touch(n: usize, workers: usize, one: impl Fn(usize) -> u64 + Sync) -> u64 {
3569 #[cfg(feature = "parallel")]
3570 if workers >= 2 {
3571 let fold_range = |first: usize, last: usize| {
3572 let mut acc = 0u64;
3573 for index in first..last {
3574 acc ^= one(index);
3575 }
3576 acc
3577 };
3578 let chunk = n.div_ceil(workers).max(1);
3579 if rayon::current_thread_index().is_some() {
3582 use rayon::prelude::*;
3583 return (0..workers)
3584 .into_par_iter()
3585 .map(|worker| {
3586 let first = worker * chunk;
3587 let last = (first + chunk).min(n);
3588 fold_range(first, last)
3589 })
3590 .reduce(|| 0u64, |left, right| left ^ right);
3591 }
3592 let mut acc = 0u64;
3593 std::thread::scope(|scope| {
3594 let mut handles = Vec::with_capacity(workers);
3595 for worker in 0..workers {
3596 handles.push(scope.spawn(move || {
3597 let first = worker * chunk;
3598 let last = (first + chunk).min(n);
3599 fold_range(first, last)
3600 }));
3601 }
3602 for handle in handles {
3603 acc ^= handle.join().unwrap_or(0);
3604 }
3605 });
3606 return acc;
3607 }
3608 #[cfg(not(feature = "parallel"))]
3609 {
3610 let _ = workers;
3611 }
3612 (0..n).map(one).fold(0, |acc, item| acc ^ item)
3613}
3614
3615pub fn measure_newline_scan(text: &str, threads: usize, chunked: bool) -> (Vec<usize>, f64) {
3623 #[cfg(feature = "parallel")]
3624 {
3625 if chunked && threads >= 2 {
3626 let pool = rayon::ThreadPoolBuilder::new().num_threads(threads).build();
3627 let t_scan = std::time::Instant::now();
3628 let offsets = match pool {
3629 Ok(pool) => pool.install(|| {
3630 let newlines = newline_positions(text.as_bytes(), threads);
3631 frame_start_offsets_indexed(text, newlines)
3632 }),
3633 Err(_) => {
3634 let newlines = newline_positions(text.as_bytes(), threads);
3635 frame_start_offsets_indexed(text, newlines)
3636 }
3637 };
3638 return (offsets, t_scan.elapsed().as_secs_f64());
3639 }
3640 }
3641 #[cfg(not(feature = "parallel"))]
3642 {
3643 let _ = (threads, chunked);
3644 }
3645 let t_scan = std::time::Instant::now();
3646 let offsets = frame_start_offsets_serial(text);
3647 (offsets, t_scan.elapsed().as_secs_f64())
3648}
3649
3650#[cfg(feature = "parallel")]
3653fn collect_parallel_located(
3654 file_contents: &str,
3655 num_threads: Option<usize>,
3656) -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
3657 use rayon::prelude::*;
3658
3659 let boundaries = frame_start_offsets(file_contents);
3660 let parse_chunks = || -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
3661 let num_frames = boundaries.len();
3662 let parts: Vec<Result<types::ConFrame, error::LocatedParseError>> = (0..num_frames)
3663 .into_par_iter()
3664 .map(|i| {
3665 let start = boundaries[i];
3666 let end = if i + 1 < num_frames {
3667 boundaries[i + 1]
3668 } else {
3669 file_contents.len()
3670 };
3671 let chunk = &file_contents[start..end];
3672 let mut iter = ConFrameIterator::new(chunk);
3673 match iter.next() {
3674 Some(Ok(frame)) => Ok(frame),
3675 Some(Err(e)) => {
3676 let local = iter.lines.line_no();
3677 let base = file_line_at(file_contents, start);
3678 let cursor = if local == 0 { base } else { base + local - 1 };
3679 Err(error::LocatedParseError::new(e, cursor, i as u64))
3680 }
3681 None => Err(error::LocatedParseError::new(
3682 error::ParseError::IncompleteFrame,
3683 file_line_at(file_contents, start),
3684 i as u64,
3685 )),
3686 }
3687 })
3688 .collect();
3689 let mut frames = Vec::with_capacity(parts.len());
3690 for part in parts {
3691 frames.push(part?);
3692 }
3693 Ok(frames)
3694 };
3695 match num_threads {
3696 None => parse_chunks(),
3697 Some(n) => {
3698 let n = n.max(1);
3699 match rayon::ThreadPoolBuilder::new().num_threads(n).build() {
3700 Ok(pool) => pool.install(parse_chunks),
3701 Err(_) => parse_chunks(),
3702 }
3703 }
3704 }
3705}
3706
3707#[cfg(all(test, feature = "parallel"))]
3708mod parallel_strong_scale_tests {
3709 use super::*;
3710 use std::path::PathBuf;
3711
3712 fn multi_frame_fixture() -> String {
3713 let p = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/test/tiny_cuh2.con");
3714 let one = std::fs::read_to_string(p).expect("fixture");
3715 one.repeat(8)
3717 }
3718
3719 fn sequential_frames(text: &str) -> Vec<types::ConFrame> {
3720 ConFrameIterator::new(text)
3721 .map(|r| r.expect("seq frame"))
3722 .collect()
3723 }
3724
3725 fn frames_payload_key(f: &types::ConFrame) -> (usize, Vec<(String, f64, f64, f64)>) {
3726 let atoms: Vec<_> = f
3727 .atom_data
3728 .iter()
3729 .map(|a| (a.symbol.to_string(), a.x, a.y, a.z))
3730 .collect();
3731 (f.atom_data.len(), atoms)
3732 }
3733
3734 #[test]
3735 fn parallel_workers_match_sequential_payloads() {
3736 let text = multi_frame_fixture();
3737 let seq = sequential_frames(&text);
3738 assert!(seq.len() >= 8);
3739 let seq_keys: Vec<_> = seq.iter().map(frames_payload_key).collect();
3740
3741 for workers in [1usize, 2, 4] {
3742 let par = parse_frames_parallel_with_threads(&text, Some(workers));
3743 assert_eq!(par.len(), seq.len(), "workers={workers}");
3744 let par_keys: Vec<_> = par
3745 .into_iter()
3746 .map(|r| frames_payload_key(&r.expect("par frame")))
3747 .collect();
3748 assert_eq!(par_keys, seq_keys, "workers={workers} frame payloads");
3749 }
3750
3751 let par_default = parse_frames_parallel(&text);
3753 let def_keys: Vec<_> = par_default
3754 .into_iter()
3755 .map(|r| frames_payload_key(&r.expect("par")))
3756 .collect();
3757 assert_eq!(def_keys, seq_keys);
3758 }
3759
3760 fn live_threads() -> usize {
3761 std::fs::read_to_string("/proc/self/status")
3762 .unwrap()
3763 .lines()
3764 .find(|line| line.starts_with("Threads:"))
3765 .unwrap()
3766 .split_whitespace()
3767 .nth(1)
3768 .unwrap()
3769 .parse()
3770 .unwrap()
3771 }
3772
3773 #[test]
3775 fn default_read_of_a_wide_file_stays_on_the_calling_thread() {
3776 const CHILD: &str = "READCON_DEFAULT_THREAD_CHILD";
3777 const NAME: &str = "iterators::parallel_strong_scale_tests::default_read_of_a_wide_file_stays_on_the_calling_thread";
3778 if std::env::var_os(CHILD).is_none() {
3779 let out = std::process::Command::new(std::env::current_exe().unwrap())
3780 .args([NAME, "--exact", "--test-threads=1"])
3781 .env(CHILD, "1")
3782 .output()
3783 .unwrap();
3784 let stdout = String::from_utf8_lossy(&out.stdout);
3785 assert!(
3786 out.status.success() && stdout.contains("test result: ok. 1 passed;"),
3787 "{stdout}{}",
3788 String::from_utf8_lossy(&out.stderr)
3789 );
3790 return;
3791 }
3792 let dir = tempfile::tempdir().unwrap();
3793 let path = dir.path().join("wide.con");
3794 let one = std::fs::read_to_string(
3795 PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/test/tiny_cuh2.con"),
3796 )
3797 .unwrap();
3798 std::fs::write(&path, one.repeat(200)).unwrap();
3799 assert!(std::fs::metadata(&path).unwrap().len() > 64 * 1024);
3800 let before = live_threads();
3801 let frames = super::read_all_frames(&path).unwrap();
3802 let after = live_threads();
3803 assert_eq!(before, after, "default read started worker threads");
3804 assert!(frames.len() > 1);
3805 let text = std::fs::read_to_string(&path).unwrap();
3806 let sequential = super::frames_from_text(&text, None).unwrap();
3807 assert_eq!(sequential.len(), frames.len());
3808 assert_eq!(
3809 sequential[0].atom_data[0].x, frames[0].atom_data[0].x,
3810 "default read changed a coordinate"
3811 );
3812 }
3813}
3814
3815#[cfg(all(test, feature = "parallel"))]
3816mod parallel_header_scan_tests {
3817 use super::*;
3818 use std::path::PathBuf;
3819
3820 fn fixture(name: &str) -> String {
3821 let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
3822 .join("resources/test")
3823 .join(name);
3824 std::fs::read_to_string(path).expect(name)
3825 }
3826
3827 fn assert_crlf_gap_cut(text: &str) {
3829 let bytes = text.as_bytes();
3830 let mut found = 0usize;
3831 let mut i = 0usize;
3832 while i + 1 < bytes.len() && found < 6 {
3833 if bytes[i] == b'\r' && bytes[i + 1] == b'\n' {
3834 let cut = i + 1;
3835 assert_eq!(bytes[cut - 1], b'\r');
3836 assert_eq!(bytes[cut], b'\n');
3837 assert_cuts_match(text, &[0, cut, bytes.len()]);
3838 found += 1;
3839 i += 2;
3840 continue;
3841 }
3842 i += 1;
3843 }
3844 assert!(found > 0, "fixture has no CRLF pair");
3845 }
3846
3847 fn assert_cuts_match(text: &str, cuts: &[usize]) {
3848 let serial_nl: Vec<usize> = memchr::memchr_iter(b'\n', text.as_bytes()).collect();
3849 let stitched = newline_positions_at_cuts(text.as_bytes(), cuts);
3850 assert_eq!(stitched, serial_nl, "newline stitch at {cuts:?}");
3851 let serial = frame_start_offsets_serial(text);
3852 let parallel = frame_start_offsets_at_cuts(text, cuts);
3853 assert_eq!(parallel, serial, "frame offsets at {cuts:?}");
3854 }
3855
3856 fn header_straddle_cuts(text: &str) -> Vec<Vec<usize>> {
3857 let serial = frame_start_offsets_serial(text);
3858 let bytes = text.as_bytes();
3859 let mut out = Vec::new();
3860 for n in [2usize, 3, 4, 8] {
3862 out.push(equal_cuts(bytes.len(), n));
3863 }
3864 for (index, &start) in serial.iter().enumerate() {
3865 let next = serial.get(index + 1).copied().unwrap_or(bytes.len());
3866 let window = super::header_window_end(bytes, start, next);
3869 let mut line = start;
3870 while line < window {
3871 let Some(rel) = memchr::memchr(b'\n', &bytes[line..window]) else {
3872 break;
3873 };
3874 let end = line + rel;
3875 if end - line > 1 {
3876 let mid = line + (end - line) / 2;
3877 assert!(mid > line && mid < end);
3879 out.push(vec![0, mid, bytes.len()]);
3880 }
3881 line = end + 1;
3882 }
3883 }
3884 out
3885 }
3886
3887 fn assert_slab(text: &str) {
3888 assert!(!text.is_empty());
3889 for cuts in header_straddle_cuts(text) {
3890 assert_cuts_match(text, &cuts);
3891 }
3892 let bare = text.trim_end_matches(['\n', '\r']);
3893 if !bare.is_empty() {
3894 assert_cuts_match(bare, &equal_cuts(bare.len(), 4));
3895 let mid = bare.len() / 2;
3896 let line_start = bare[..mid].rfind('\n').map(|i| i + 1).unwrap_or(0);
3897 let line_end = bare[mid..]
3898 .find('\n')
3899 .map(|i| mid + i)
3900 .unwrap_or(bare.len());
3901 if line_end - line_start > 1 {
3902 let cut = line_start + (line_end - line_start) / 2;
3903 assert_cuts_match(bare, &[0, cut, bare.len()]);
3904 }
3905 }
3906 let tailed = format!("{text}\n \n\t\n");
3907 assert_cuts_match(
3908 &tailed,
3909 &[0, 3, tailed.len() / 2, tailed.len() - 1, tailed.len()],
3910 );
3911 let crlf = text.replace('\n', "\r\n");
3912 assert_cuts_match(&crlf, &equal_cuts(crlf.len(), 5));
3913 assert_crlf_gap_cut(&crlf);
3914 let serial = frame_start_offsets_serial(text);
3915 if serial.len() >= 2 {
3916 let second = serial[1];
3917 let next = serial.get(2).copied().unwrap_or(text.len());
3918 let broken_at = second + (next - second) / 2;
3919 let broken = &text[..broken_at.max(1)];
3920 assert_cuts_match(broken, &[0, 1.max(broken.len() / 3), broken.len()]);
3921 }
3922 }
3923
3924 #[test]
3925 fn chunked_header_scan_matches_serial_offsets() {
3926 assert_eq!(frame_start_offsets_serial(""), frame_start_offsets(""));
3927 assert_cuts_match("\n", &[0, 1]);
3928 assert_cuts_match(" \n\n", &[0, 2, 5]);
3929 for name in [
3930 "tiny_cuh2.con",
3931 "tiny_cuh2.convel",
3932 "tiny_cuh2_forces.con",
3933 "tiny_cuh2_vel_forces.con",
3934 "cuh2.con",
3935 ] {
3936 let one = fixture(name);
3937 assert_slab(&one);
3938 assert_slab(&one.repeat(3));
3939 }
3940 let one = fixture("cuh2.con");
3944 for wide in [one.clone(), one.repeat(8)] {
3945 let serial = frame_start_offsets_serial(&wide);
3946 for workers in [2usize, 3, 4, 8] {
3947 let parallel = frame_start_offsets_with_threads(&wide, Some(workers));
3948 assert_eq!(parallel, serial, "workers={workers} bytes={}", wide.len());
3949 let parsed =
3950 frames_from_bytes(wide.as_bytes(), Some(workers)).expect("bytes parse");
3951 assert_eq!(parsed.len(), serial.len());
3952 let seq = frames_from_text(&wide, Some(1)).expect("serial parse");
3953 assert_eq!(parsed.len(), seq.len());
3954 assert_eq!(parsed[0].atom_data[0].x, seq[0].atom_data[0].x);
3955 assert_eq!(
3956 parsed.last().unwrap().atom_data.last().unwrap().z,
3957 seq.last().unwrap().atom_data.last().unwrap().z
3958 );
3959 }
3960 }
3961 }
3962}
3963
3964#[cfg(all(test, feature = "parallel"))]
3969fn header_window_end(bytes: &[u8], start: usize, limit: usize) -> usize {
3970 let limit = limit.min(bytes.len());
3971 let marker = b"Coordinates";
3972 let mut line = start;
3973 while line < limit {
3974 let rest = &bytes[line..limit];
3975 let rel = memchr::memchr(b'\n', rest).unwrap_or(rest.len());
3976 let content_end = line + rel;
3977 let content = &bytes[line..content_end];
3978 let trimmed = match content.iter().position(|byte| !byte.is_ascii_whitespace()) {
3979 Some(index) => &content[index..],
3980 None => b"",
3981 };
3982 if trimmed.starts_with(marker) {
3983 return line;
3984 }
3985 if content_end >= limit {
3986 break;
3987 }
3988 line = content_end + 1;
3989 }
3990 limit
3991}
3992
3993#[cfg(all(test, feature = "parallel"))]
3994mod parallel_frame_parse_tests {
3995 use super::*;
3996 use std::path::PathBuf;
3997
3998 const ZERO_ATOM: &str = "\
3999zero-atom frame
4000{\"con_spec_version\":3,\"con_spec_minor\":1,\"units\":{\"length\":\"angstrom\",\"energy\":\"eV\"}}
400110.0 10.0 10.0
400290.0 90.0 90.0
4003
4004
40050
4006
4007
4008";
4009
4010 fn fixture(name: &str) -> String {
4011 let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
4012 .join("resources/test")
4013 .join(name);
4014 std::fs::read_to_string(path).expect(name)
4015 }
4016
4017 fn serial_results(text: &str) -> Vec<Result<types::ConFrame, error::ParseError>> {
4018 ConFrameIterator::new(text).collect()
4019 }
4020
4021 fn written(frame: &types::ConFrame) -> Vec<u8> {
4022 let mut out = Vec::new();
4023 {
4024 let mut writer = crate::writer::ConFrameWriter::new(&mut out);
4025 writer.write_frame(frame).expect("write");
4026 writer.flush().expect("flush");
4027 }
4028 out
4029 }
4030
4031 fn assert_same_results(
4032 label: &str,
4033 serial: Vec<Result<types::ConFrame, error::ParseError>>,
4034 parallel: Vec<Result<types::ConFrame, error::ParseError>>,
4035 ) {
4036 assert_eq!(serial.len(), parallel.len(), "{label} count");
4037 for (i, (left, right)) in serial.into_iter().zip(parallel).enumerate() {
4038 match (left, right) {
4039 (Ok(a), Ok(b)) => {
4040 assert_eq!(written(&a), written(&b), "{label} bytes frame {i}");
4041 assert_eq!(a, b, "{label} fields frame {i}");
4042 }
4043 (Err(a), Err(b)) => {
4044 assert_eq!(
4045 format!("{a:?}"),
4046 format!("{b:?}"),
4047 "{label} error frame {i}"
4048 );
4049 }
4050 (Ok(_), Err(err)) => panic!("{label} frame {i} serial ok parallel {err:?}"),
4051 (Err(err), Ok(_)) => panic!("{label} frame {i} serial {err:?} parallel ok"),
4052 }
4053 }
4054 }
4055
4056 fn assert_workers(label: &str, text: &str) {
4057 let serial_ok = serial_results(text).iter().all(|item| item.is_ok());
4058 for workers in [1usize, 2, 3, 4, 8] {
4059 let parallel = parse_frames_parallel_with_threads(text, Some(workers));
4060 assert_same_results(
4061 &format!("{label} workers {workers}"),
4062 serial_results(text),
4063 parallel,
4064 );
4065 let bounds = collect_frame_starts(ConFrameIterator::new(text), true);
4066 let at = parse_frames_at_offsets(text, &bounds, Some(workers));
4067 assert_same_results(
4068 &format!("{label} offsets workers {workers}"),
4069 serial_results(text),
4070 at,
4071 );
4072 }
4073 if serial_ok {
4074 for workers in [2usize, 3, 8] {
4075 let bytes = frames_from_bytes(text.as_bytes(), Some(workers)).expect(label);
4076 let seq = frames_from_text(text, Some(1)).expect(label);
4077 assert_eq!(bytes.len(), seq.len(), "{label} byte count");
4078 for (i, (a, b)) in bytes.iter().zip(seq.iter()).enumerate() {
4079 assert_eq!(written(a), written(b), "{label} byte path frame {i}");
4080 assert_eq!(a, b, "{label} byte path fields frame {i}");
4081 }
4082 }
4083 }
4084 }
4085
4086 fn crlf_gap_offsets(text: &str) -> Vec<usize> {
4087 let bytes = text.as_bytes();
4088 let mut cut = None;
4089 let mut i = 0usize;
4090 while i + 1 < bytes.len() {
4091 if bytes[i] == b'\r' && bytes[i + 1] == b'\n' {
4092 cut = Some(i + 1);
4093 break;
4094 }
4095 i += 1;
4096 }
4097 let cut = cut.expect("crlf");
4098 frame_start_offsets_at_cuts(text, &[0, cut, bytes.len()])
4099 }
4100
4101 #[test]
4102 fn parallel_slots_match_serial_frames() {
4103 assert_workers("empty", "");
4104 assert_workers("blank", "\n");
4105 assert_workers("space", " \n\n");
4106 assert_workers("zero", ZERO_ATOM);
4107 assert_workers("zero-repeat", &ZERO_ATOM.repeat(5));
4108 for name in [
4109 "tiny_cuh2.con",
4110 "tiny_cuh2.convel",
4111 "tiny_cuh2_forces.con",
4112 "tiny_cuh2_vel_forces.con",
4113 "tiny_cuh2_charges_spins_magmoms.con",
4114 "tiny_cuh2_displacements.con",
4115 "tiny_cuh2_spreads.con",
4116 "cuh2.con",
4117 ] {
4118 let one = fixture(name);
4119 assert_workers(name, &one);
4120 assert_workers(&format!("{name} x5"), &one.repeat(5));
4121 let bare = one.trim_end_matches(['\n', '\r']);
4122 assert_workers(&format!("{name} bare"), bare);
4123 let tailed = format!("{one}\n \n\t\n");
4124 assert_workers(&format!("{name} tail"), &tailed);
4125 let crlf = one.replace('\n', "\r\n");
4126 assert_workers(&format!("{name} crlf"), &crlf);
4127 let gap = crlf_gap_offsets(&crlf);
4128 assert_eq!(
4129 gap,
4130 frame_start_offsets_serial(&crlf),
4131 "{name} crlf gap offsets"
4132 );
4133 let parsed = parse_frames_at_offsets(&crlf, &gap, Some(4));
4134 assert_same_results(
4135 &format!("{name} crlf gap parse"),
4136 serial_results(&crlf),
4137 parsed,
4138 );
4139 let serial_offs = frame_start_offsets_serial(&one);
4140 let broken: &str = if serial_offs.len() >= 2 {
4141 let second = serial_offs[1];
4142 let next = serial_offs.get(2).copied().unwrap_or(one.len());
4143 &one[..second + (next - second) / 2]
4144 } else if one.len() > 8 {
4145 &one[..one.len() / 2]
4146 } else {
4147 ""
4148 };
4149 if !broken.is_empty() {
4150 let serial_err = frames_from_text(broken, Some(1))
4151 .err()
4152 .map(|e| format!("{e:?}"));
4153 assert!(serial_err.is_some(), "{name} chopped still parsed");
4154 for workers in [1usize, 2, 4, 8] {
4155 let parallel_err = frames_from_text(broken, Some(workers))
4156 .err()
4157 .map(|e| format!("{e:?}"));
4158 assert_eq!(parallel_err, serial_err, "{name} chopped workers {workers}");
4159 let byte_err = frames_from_bytes(broken.as_bytes(), Some(workers))
4160 .err()
4161 .map(|e| format!("{e:?}"));
4162 assert_eq!(
4163 byte_err, serial_err,
4164 "{name} chopped bytes workers {workers}"
4165 );
4166 }
4167 }
4168 }
4169 }
4170}
4171
4172#[cfg(all(test, feature = "parallel"))]
4173mod phased_read_tests {
4174 use super::*;
4175 use std::path::PathBuf;
4176
4177 fn fixture(name: &str) -> String {
4178 let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
4179 .join("resources/test")
4180 .join(name);
4181 std::fs::read_to_string(path).expect(name)
4182 }
4183
4184 #[test]
4185 fn phased_bytes_match_serial_frames() {
4186 let text = fixture("cuh2.con").repeat(4);
4187 let serial = frames_from_text(&text, Some(1)).expect("serial");
4188 for workers in [1usize, 2, 4, 8] {
4189 let phased = frames_from_bytes_phased(text.as_bytes(), Some(workers)).expect("phased");
4190 let parallel = phased.frames;
4191 assert_eq!(parallel.len(), serial.len(), "workers={workers}");
4192 assert!(
4193 phased.scan_s >= 0.0 && phased.slot_alloc_s >= 0.0 && phased.worker_parse_s >= 0.0
4194 );
4195 assert_eq!(parallel[0], serial[0], "workers={workers} first");
4196 assert_eq!(
4197 parallel.last().unwrap(),
4198 serial.last().unwrap(),
4199 "workers={workers} last"
4200 );
4201 let slab = pack_position_slab(&serial, workers).expect("slab");
4202 let stride = serial[0].positions.nrows() * 3;
4203 let src = serial[0].positions.f64_slice().expect("f64");
4204 assert_eq!(&slab[..stride], src);
4205 let (chunked, _) = measure_newline_scan(&text, workers.max(2), true);
4206 let (plain, _) = measure_newline_scan(&text, 1, false);
4207 if workers >= 2 {
4208 assert_eq!(chunked, plain);
4209 }
4210 }
4211 }
4212
4213 #[test]
4214 fn phased_file_matches_serial_and_maps_a_wide_file() {
4215 let text = fixture("cuh2.con").repeat(5);
4216 let dir = std::env::temp_dir();
4217 let path = dir.join("readcon-phased-wide.con");
4218 std::fs::write(&path, &text).expect("write");
4219 let phased = read_all_frames_phased(&path, Some(4)).expect("phased");
4220 let serial = frames_from_text(&text, Some(1)).expect("serial");
4221 assert_eq!(phased.frames, serial);
4222 assert!(phased.phases.mapped, "a file past 64 KiB is mapped");
4223 assert!(phased.phases.worker_parse_s >= 0.0);
4224 assert_eq!(phased.n_bytes, text.len());
4225 let _ = std::fs::remove_file(&path);
4226 }
4227
4228 #[test]
4229 fn equal_span_boundaries_match_the_serial_walk() {
4230 let text = fixture("cuh2.con").repeat(8);
4231 let bytes = text.as_bytes();
4232 let got = equal_span_boundaries(bytes, 4).expect("equal span");
4233 let expect = collect_frame_starts(ConFrameIterator::from_bytes(bytes), true);
4234 assert_eq!(got, expect);
4235 assert_eq!(got.len(), 8);
4236 let phased = frames_from_bytes_phased(bytes, Some(4)).expect("phased");
4237 assert_eq!(phased.scan_kind, SCAN_EQUAL_SPAN);
4238 assert_eq!(
4239 phased.frames,
4240 frames_from_text(&text, Some(1)).expect("serial")
4241 );
4242 let tailed = format!("{text}\n\n \n");
4243 let tail_phased = frames_from_bytes_phased(tailed.as_bytes(), Some(4)).expect("tail");
4244 assert_eq!(tail_phased.scan_kind, SCAN_EQUAL_SPAN);
4245 assert_eq!(
4246 tail_phased.frames,
4247 frames_from_text(&tailed, Some(1)).expect("tail serial")
4248 );
4249 }
4250
4251 #[test]
4252 fn a_whitespace_tail_uses_the_equal_span_walk() {
4253 let mut text = fixture("cuh2.con").repeat(3);
4254 text.push('\n');
4255 assert!(equal_span_boundaries(text.as_bytes(), 4).is_some());
4256 let serial = frames_from_text(&text, Some(1)).expect("serial");
4257 let parallel = frames_from_bytes_phased(text.as_bytes(), Some(4)).expect("phased");
4258 assert_eq!(parallel.scan_kind, SCAN_EQUAL_SPAN);
4259 assert_eq!(parallel.frames, serial);
4260 }
4261
4262 #[test]
4263 fn touch_checksum_matches_across_worker_counts() {
4264 let text = fixture("cuh2.con").repeat(4);
4265 let serial = frames_from_bytes_phased(text.as_bytes(), Some(1)).expect("serial");
4266 let parallel = frames_from_bytes_phased(text.as_bytes(), Some(4)).expect("parallel");
4267 let slab_s = pack_position_slab(&serial.frames, 1).expect("slab");
4268 let slab_p = pack_position_slab(¶llel.frames, 4).expect("slab");
4269 let left = touch_frame_checksum(&serial.frames, &slab_s, 1).expect("touch");
4270 let right = touch_frame_checksum(¶llel.frames, &slab_p, 4).expect("touch");
4271 assert_eq!(left, right);
4272 assert_ne!(left, 0);
4273 }
4274}
4275
4276#[cfg(test)]
4277mod later_frame_location_tests {
4278 use super::*;
4279 use std::fs;
4280
4281 fn one_atom(coord: &str) -> String {
4282 format!(
4283 "seed\n{{\"con_spec_version\":2}}\n10 10 10\n90 90 90\n0 0\n0 0 0\n1\n1\n1.0\nH\nCoordinates of Component 1\n{coord}\n"
4284 )
4285 }
4286
4287 fn bad_count_frame() -> String {
4288 "seed\n{\"con_spec_version\":2}\n10 10 10\n90 90 90\n0 0\n0 0 0\n1\n-1\n1.0\nH\nCoordinates of Component 1\n0 0 0 0 0\n".into()
4289 }
4290
4291 fn two_frames() -> String {
4292 format!("{}{}", one_atom("0 0 0 0 0"), bad_count_frame())
4293 }
4294
4295 fn assert_later_count(
4296 err: &error::ParseError,
4297 line: u32,
4298 frame: u64,
4299 iter: &ConFrameIterator<'_>,
4300 ) {
4301 assert_eq!(err.class_name(), "InvalidCount", "{err}");
4302 assert_eq!(iter.lines.line_no(), line, "{err}");
4303 assert_eq!(iter.frame_index(), frame, "{err}");
4304 }
4305
4306 fn assert_first_pass(err: Box<dyn std::error::Error>) {
4307 let located = err
4308 .downcast_ref::<error::LocatedParseError>()
4309 .unwrap_or_else(|| panic!("first pass lost the location: {err}"));
4310 assert_eq!(located.error.class_name(), "InvalidCount", "{located:?}");
4311 assert_eq!(located.line, 20, "{located:?}");
4312 assert_eq!(located.frame_index, 1, "{located:?}");
4313 }
4314
4315 fn take_err<T>(result: Result<T, Box<dyn std::error::Error>>) -> Box<dyn std::error::Error> {
4316 match result {
4317 Err(err) => err,
4318 Ok(_) => panic!("expected a parse error"),
4319 }
4320 }
4321
4322 #[test]
4325 fn batch_readers_keep_the_later_frame_on_the_first_pass() {
4326 let path = std::env::temp_dir().join(format!(
4327 "readcon-later-frame-first-{}.con",
4328 std::process::id()
4329 ));
4330 fs::write(&path, two_frames()).unwrap();
4331 for threads in [None, Some(1), Some(2)] {
4332 assert_first_pass(take_err(read_all_frames_with_threads(&path, threads)));
4333 assert_first_pass(take_err(read_all_frames_phased(&path, threads)));
4334 assert_first_pass(take_err(read_positions_with_threads(&path, threads)));
4335 }
4336 let _ = fs::remove_file(&path);
4337 }
4338
4339 #[test]
4341 fn next_forward_and_skip_locate_the_later_frame() {
4342 let text = two_frames();
4343
4344 let mut iter = ConFrameIterator::new(&text);
4345 assert!(iter.next().unwrap().is_ok());
4346 let err = iter.next().unwrap().unwrap_err();
4347 assert_later_count(&err, 20, 1, &iter);
4348
4349 let mut iter = ConFrameIterator::new(&text);
4350 assert!(iter.forward().unwrap().is_ok());
4351 let err = iter.forward().unwrap().unwrap_err();
4352 assert_later_count(&err, 20, 1, &iter);
4353
4354 let mut iter = ConFrameIterator::new(&text);
4355 let err = iter.skip_frames(3).unwrap_err();
4356 assert_later_count(&err, 20, 1, &iter);
4357 }
4358
4359 #[test]
4363 fn skip_after_a_velocity_frame_keeps_the_separator_line() {
4364 let text = format!(
4365 "{}\nH\nVelocities of Component 1\n0 0 0 0 0\n{}",
4366 one_atom("0 0 0 0 0"),
4367 bad_count_frame()
4368 );
4369 let mut iter = ConFrameIterator::new(&text);
4370 assert!(iter.forward().unwrap().is_ok());
4371 let err = iter.forward().unwrap().unwrap_err();
4372 assert_later_count(&err, 24, 1, &iter);
4373 }
4374
4375 #[test]
4376 fn count_frames_locates_the_later_frame() {
4377 let path = std::env::temp_dir().join(format!(
4378 "readcon-later-frame-count-{}.con",
4379 std::process::id()
4380 ));
4381 fs::write(&path, two_frames()).unwrap();
4382 let err = count_frames(&path).unwrap_err();
4383 let _ = fs::remove_file(&path);
4384 match err {
4385 LoadError::Parse(located) => {
4386 assert_eq!(located.error.class_name(), "InvalidCount");
4387 assert_eq!(located.line, 20, "{located}");
4388 assert_eq!(located.frame_index, 1, "{located}");
4389 }
4390 other => panic!("expected a located parse error, got {other}"),
4391 }
4392 }
4393
4394 #[test]
4397 fn metadata_in_a_later_frame_reports_that_frames_line() {
4398 let bad = "seed\n{not-json\n10 10 10\n90 90 90\n0 0\n0 0 0\n1\n1\n1.0\nH\nCoordinates of Component 1\n0 0 0 0 0\n";
4399 let text = format!("{}{}", one_atom("0 0 0 0 0"), bad);
4400 for threads in [Some(1usize), Some(2)] {
4401 let err = frames_from_text_located(&text, threads).unwrap_err();
4402 assert_eq!(
4403 err.error.class_name(),
4404 "InvalidMetadataJson",
4405 "{threads:?}: {err}"
4406 );
4407 assert_eq!(err.frame_index, 1, "{threads:?}");
4408 assert_eq!(err.line, 14, "{threads:?}: {err}");
4409 }
4410 }
4411
4412 #[test]
4420 fn parallel_positions_report_the_earliest_failing_slot() {
4421 const SLOW_ATOMS: usize = 20_000;
4422 let slow = counted_frame(SLOW_ATOMS);
4423 let bad = counted_frame(1);
4424 let text = format!("{slow}{bad}{bad}{bad}");
4425 let earliest_line = 1 + slow.bytes().filter(|byte| *byte == b'\n').count() as u32;
4426 let path =
4427 std::env::temp_dir().join(format!("readcon-earliest-slot-{}.con", std::process::id()));
4428 fs::write(&path, &text).unwrap();
4429 assert_earliest_slot(
4430 take_err(read_positions_with_threads(&path, Some(1))),
4431 earliest_line,
4432 );
4433 assert_earliest_slot(
4434 take_err(read_positions_with_threads(&path, Some(2))),
4435 earliest_line,
4436 );
4437 let _ = fs::remove_file(&path);
4438 }
4439
4440 fn counted_frame(natoms: usize) -> String {
4441 let mut body = String::with_capacity(natoms * 10 + 96);
4442 body.push_str("seed\n{\"con_spec_version\":2}\n10 10 10\n90 90 90\n0 0\n0 0 0\n1\n");
4443 body.push_str(&natoms.to_string());
4444 body.push_str("\n1.0\nH\nCoordinates of Component 1\n");
4445 for _ in 0..natoms {
4446 body.push_str("0 0 0 0 0\n");
4447 }
4448 body
4449 }
4450
4451 fn assert_earliest_slot(err: Box<dyn std::error::Error>, line: u32) {
4452 let located = err
4453 .downcast_ref::<error::LocatedParseError>()
4454 .unwrap_or_else(|| panic!("position read lost the location: {err}"));
4455 assert_eq!(located.error.class_name(), "InvalidCount", "{located:?}");
4456 assert_eq!(located.frame_index, 1, "{located:?}");
4457 assert_eq!(located.line, line, "{located:?}");
4458 }
4459}
4460
4461#[cfg(test)]
4462mod frame_equality_tests {
4463 use super::ConFrameIterator;
4464
4465 fn sample_con(frames: usize, long: bool) -> String {
4466 let mut out = String::new();
4467 for f in 0..frames {
4468 let shift = (f as f64) * 1.0e-4;
4469 out.push_str("Generated by benchmark\n");
4470 out.push_str("{\"con_spec_version\":2}\n");
4471 out.push_str("15.3456 21.702 100.0\n");
4472 out.push_str("90.0 90.0 90.0\n");
4473 out.push_str("0 0\n");
4474 out.push_str(&format!("{f} 0 1\n"));
4475 out.push_str("2\n6 2\n63.546 1.00793\n");
4476 out.push_str("Cu\nCoordinates of Component 1\n");
4477 for i in 0..6 {
4478 let x = (i as f64) * 2.557 + 0.6394 + shift;
4479 let y = 0.9045 + shift * 0.5;
4480 let z = (f as f64) * 1.0e-6 + if i == 3 { -1.25 } else { 0.0 };
4481 if long {
4482 out.push_str(&format!(" {x:.17} {y:.17} {z:.17} 0 {i}\n"));
4483 } else {
4484 out.push_str(&format!(" {x:.6} {y:.6} {z:.6} 0 {i}\n"));
4485 }
4486 }
4487 out.push_str("H\nCoordinates of Component 2\n");
4488 out.push_str(" 8.6823 9.947 11.733 0 6\n");
4489 out.push_str(&format!(" -7.9421 9.947 {shift:.6} 0 7\n"));
4490 }
4491 out
4492 }
4493
4494 fn sample_convel() -> String {
4495 let mut out = String::new();
4496 for f in 0..3 {
4497 let shift = (f as f64) * 1.0e-4;
4498 out.push_str("Generated by benchmark\n");
4499 out.push_str("{\"con_spec_version\":2}\n");
4500 out.push_str("15.3456 21.702 100.0\n");
4501 out.push_str("90.0 90.0 90.0\n");
4502 out.push_str("0 0\n");
4503 out.push_str(&format!("{f} 0 1\n"));
4504 out.push_str("2\n4 1\n63.546 1.00793\n");
4505 out.push_str("Cu\nCoordinates of Component 1\n");
4506 for i in 0..4 {
4507 let x = (i as f64) * 2.557 + shift;
4508 out.push_str(&format!(" {x:.17} 0.9045 1.0 0 {i}\n"));
4509 }
4510 out.push_str("H\nCoordinates of Component 2\n");
4511 out.push_str(" 1.5 2.5 3.5 0 4\n");
4512 out.push_str("\nCu\nVelocities of Component 1\n");
4513 for i in 0..4 {
4514 let vx = (i as f64) * 1.0e-4 - shift;
4515 out.push_str(&format!(" {vx:.17} -0.5 0.25 0 {i}\n"));
4516 }
4517 out.push_str("H\nVelocities of Component 2\n");
4518 out.push_str(" 0.1 -0.2 0.3 0 4\n");
4519 }
4520 out
4521 }
4522
4523 fn reference_f64(token: &str) -> f64 {
4524 let (val, n) = fast_float2::parse_partial::<f64, _>(token).expect(token);
4525 assert_eq!(n, token.len(), "{token}");
4526 val
4527 }
4528
4529 fn assert_rows_match(text: &str, positions: bool) {
4530 let frames: Vec<_> = ConFrameIterator::new(text)
4531 .map(|frame| frame.expect("parse"))
4532 .collect();
4533 assert!(frames.len() >= 2);
4534 let mut frame_i = 0usize;
4535 let mut row = 0usize;
4536 let mut in_section = false;
4537 for line in text.lines() {
4538 let token = line.trim();
4539 if token.is_empty() {
4540 in_section = true;
4541 row = 0;
4542 continue;
4543 }
4544 if token.starts_with("Generated") {
4545 if row > 0 || in_section {
4546 frame_i += 1;
4547 }
4548 row = 0;
4549 in_section = false;
4550 continue;
4551 }
4552 let cols: Vec<&str> = token.split_whitespace().collect();
4553 if cols.len() < 4 || reference_ok(cols[0]).is_none() {
4554 continue;
4555 }
4556 if cols[3].parse::<u64>().is_err() {
4557 continue;
4558 }
4559 let got = if in_section {
4560 frames[frame_i].atom_data[row].velocity().expect("velocity")
4561 } else if positions {
4562 let p = frames[frame_i].positions.as_f64_row(row);
4563 [p[0], p[1], p[2]]
4564 } else {
4565 continue;
4566 };
4567 for (k, token) in cols.iter().take(3).enumerate() {
4568 assert_eq!(
4569 got[k].to_bits(),
4570 reference_f64(token).to_bits(),
4571 "frame {frame_i} row {row} col {k} token {token} section {in_section}"
4572 );
4573 }
4574 let aos = if in_section {
4575 frames[frame_i].atom_data[row].velocity().unwrap()
4576 } else {
4577 let atom = &frames[frame_i].atom_data[row];
4578 [atom.x, atom.y, atom.z]
4579 };
4580 for k in 0..3 {
4581 assert_eq!(aos[k].to_bits(), got[k].to_bits());
4582 }
4583 row += 1;
4584 }
4585 }
4586
4587 fn reference_ok(token: &str) -> Option<f64> {
4588 let (val, n) = fast_float2::parse_partial::<f64, _>(token).ok()?;
4589 (n == token.len()).then_some(val)
4590 }
4591
4592 #[test]
4593 fn frames_match_general_decimal_parser() {
4594 assert_rows_match(&sample_con(4, true), true);
4595 assert_rows_match(&sample_con(4, false), true);
4596 let text = sample_convel();
4597 let frames: Vec<_> = ConFrameIterator::new(&text)
4598 .map(|frame| frame.expect("convel"))
4599 .collect();
4600 assert_eq!(frames.len(), 3);
4601 assert!(
4602 frames
4603 .iter()
4604 .all(|frame| frame.atom_data[0].velocity().is_some())
4605 );
4606 assert_eq!(frames[0].positions.nrows(), 5);
4607 let mut frame_i = 0usize;
4609 let mut coord_row = 0usize;
4610 let mut vel_row = 0usize;
4611 let mut velocities = false;
4612 for line in text.lines() {
4613 let token = line.trim();
4614 if token.starts_with("Generated") && (coord_row > 0 || velocities) {
4615 frame_i += 1;
4616 coord_row = 0;
4617 vel_row = 0;
4618 velocities = false;
4619 }
4620 if token.starts_with("Velocities") {
4621 velocities = true;
4622 continue;
4623 }
4624 let cols: Vec<&str> = token.split_whitespace().collect();
4625 if cols.len() < 5 {
4626 continue;
4627 }
4628 if reference_ok(cols[0]).is_none() || cols[3].parse::<u64>().is_err() {
4629 continue;
4630 }
4631 let row = if velocities { vel_row } else { coord_row };
4632 let got = if velocities {
4633 frames[frame_i].atom_data[row].velocity().unwrap()
4634 } else {
4635 let p = frames[frame_i].positions.as_f64_row(row);
4636 [p[0], p[1], p[2]]
4637 };
4638 for (k, col) in cols.iter().take(3).enumerate() {
4639 assert_eq!(got[k].to_bits(), reference_f64(col).to_bits(), "{token}");
4640 }
4641 if velocities {
4642 let soa = frames[frame_i].velocities.as_f64_row(row);
4643 for k in 0..3 {
4644 assert_eq!(soa[k].to_bits(), got[k].to_bits());
4645 }
4646 vel_row += 1;
4647 } else {
4648 coord_row += 1;
4649 }
4650 }
4651 assert_eq!(frame_i, 2);
4652 assert_eq!(vel_row, 5);
4653 }
4654}
4655
4656#[cfg(test)]
4657mod newline_skip_tests {
4658 use super::ConFrameIterator;
4659
4660 fn naive(bytes: &[u8], n: usize) -> Result<usize, ()> {
4663 let mut pos = 0usize;
4664 for i in 0..n {
4665 let rest = &bytes[pos..];
4666 match memchr::memchr(b'\n', rest) {
4667 Some(at) => pos += at + 1,
4668 None => {
4669 if rest.is_empty() {
4670 return Err(());
4671 }
4672 pos = bytes.len();
4673 if i + 1 == n {
4674 return Ok(pos);
4675 }
4676 return Err(());
4677 }
4678 }
4679 }
4680 Ok(pos)
4681 }
4682
4683 #[test]
4684 fn counted_skip_matches_per_line_memchr() {
4685 let mut samples = vec![
4686 String::new(),
4687 "a".into(),
4688 "a\n".into(),
4689 "a\nb".into(),
4690 "\n".into(),
4691 "\n\n".into(),
4692 "abc\ndef\nghi".into(),
4693 "a\r\nb\r\n".into(),
4694 "x".repeat(100) + "\n" + &"y".repeat(40),
4695 ];
4696 let path =
4697 std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/test/cuh2.con");
4698 let one = std::fs::read_to_string(&path).unwrap();
4699 samples.push(one.repeat(2));
4700 let bare = one.trim_end_matches(['\n', '\r']).to_string();
4701 samples.push(bare);
4702 for text in samples {
4703 let bytes = text.as_bytes();
4704 for n in [0usize, 1, 2, 3, 6, 7, 9, 40, 200, 450, 1000, 5000] {
4705 let expect = naive(bytes, n);
4706 let mut it = ConFrameIterator::from_bytes(bytes);
4707 let got = it.advance_lines(n).map(|_| it.byte_pos()).map_err(|_| ());
4708 assert_eq!(got, expect, "n={n} len={}", bytes.len());
4709 }
4710 }
4711 }
4712}
4713
4714#[cfg(test)]
4715mod cached_unit_table_read_paths {
4716 use super::*;
4717
4718 const FRAME: &str = "\
4719v3
4720{\"con_spec_version\":3,\"con_spec_minor\":1,\"units\":{\"length\":\"angstrom\",\"energy\":\"eV\"}}
472110 10 10
472290 90 90
4723
4724
47251
47261
47271.0
4728H
4729Coordinates of Component 1
47300.5 0.25 0.125 0 7
4731";
4732
4733 fn coords(frame: &types::ConFrame) -> [f64; 3] {
4734 frame.positions.as_f64_row(0)
4735 }
4736
4737 #[test]
4740 fn read_paths_share_the_cached_unit_table() {
4741 let addr = crate::units::cached_unit_table_addr();
4742 let serial = ConFrameIterator::new(FRAME)
4743 .next()
4744 .expect("frame")
4745 .expect("parse");
4746 let from_bytes = frames_from_bytes(FRAME.as_bytes(), Some(1)).expect("bytes");
4747 let phased = frames_from_bytes_phased(FRAME.as_bytes(), Some(1)).expect("phased");
4748 assert_eq!(from_bytes.len(), 1);
4749 assert_eq!(phased.frames.len(), 1);
4750 assert_eq!(coords(&from_bytes[0]), coords(&serial));
4751 assert_eq!(coords(&phased.frames[0]), coords(&serial));
4752 assert_eq!(crate::units::cached_unit_table_addr(), addr);
4753 let bad = FRAME.replace("angstrom", "not-a-unit");
4754 assert!(ConFrameIterator::new(&bad).next().unwrap().is_err());
4755 assert!(frames_from_bytes(bad.as_bytes(), Some(1)).is_err());
4756 assert!(frames_from_bytes_phased(bad.as_bytes(), Some(1)).is_err());
4757 assert_eq!(crate::units::cached_unit_table_addr(), addr);
4758 }
4759}