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 {
1745 remote += pages.parse::<i64>().unwrap_or(0);
1746 }
1747 }
1748 }
1749 remote
1750}
1751
1752#[cfg(feature = "parallel")]
1759fn byte_balanced_ranges(boundaries: &[usize], len: usize, workers: usize) -> Vec<(usize, usize)> {
1760 let n = boundaries.len();
1761 if n == 0 {
1762 return Vec::new();
1763 }
1764 let workers = workers.max(1).min(n);
1765 let chunks = workers.saturating_mul(4).clamp(workers, n);
1766 let origin = boundaries[0];
1767 let total = len.saturating_sub(origin).max(1) as u128;
1768 let mut ranges = Vec::with_capacity(chunks);
1769 let mut start = 0usize;
1770 for chunk in 0..chunks {
1771 if start >= n {
1772 break;
1773 }
1774 let end = if chunk + 1 == chunks {
1775 n
1776 } else {
1777 let target = origin + ((total * (chunk as u128 + 1)) / chunks as u128) as usize;
1778 let mut idx = if start + 1 < n {
1779 boundaries[start + 1..].partition_point(|&b| b < target) + start + 1
1780 } else {
1781 n
1782 };
1783 let min_end = start + 1;
1784 let reserved = chunks - chunk - 1;
1785 let max_end = n.saturating_sub(reserved);
1786 if max_end >= min_end {
1787 idx = idx.clamp(min_end, max_end);
1788 } else {
1789 idx = min_end.min(n);
1790 }
1791 idx
1792 };
1793 if end > start {
1794 ranges.push((start, end));
1795 start = end;
1796 }
1797 }
1798 if start < n {
1799 match ranges.last_mut() {
1800 Some(last) => last.1 = n,
1801 None => ranges.push((start, n)),
1802 }
1803 }
1804 ranges
1805}
1806
1807#[cfg(feature = "parallel")]
1814fn parse_into_slots<F>(
1815 bytes: &[u8],
1816 len: usize,
1817 boundaries: &[usize],
1818 workers: usize,
1819 parse_span: F,
1820) -> Vec<Result<types::ConFrame, error::ParseError>>
1821where
1822 F: Fn(usize, usize) -> Result<types::ConFrame, error::ParseError> + Sync,
1823{
1824 parse_into_slots_timed(bytes, len, boundaries, workers, parse_span).slots
1825}
1826
1827#[cfg(feature = "parallel")]
1829struct SlotParse {
1830 slots: Vec<Result<types::ConFrame, error::ParseError>>,
1831 alloc_s: f64,
1832 work_s: f64,
1833 sum_s: f64,
1834 max_s: f64,
1835 chunk_s: Vec<f64>,
1837 worker_minflt: Vec<i64>,
1839 share_frames: Vec<u64>,
1841 share_bytes: Vec<u64>,
1843 fail_lines: Vec<(usize, u32)>,
1846}
1847
1848#[cfg(feature = "parallel")]
1856fn parse_into_slots_timed<F>(
1857 bytes: &[u8],
1858 len: usize,
1859 boundaries: &[usize],
1860 workers: usize,
1861 parse_span: F,
1862) -> SlotParse
1863where
1864 F: Fn(usize, usize) -> Result<types::ConFrame, error::ParseError> + Sync,
1865{
1866 let n = boundaries.len();
1867 if n == 0 {
1868 return SlotParse {
1869 slots: Vec::new(),
1870 alloc_s: 0.0,
1871 work_s: 0.0,
1872 sum_s: 0.0,
1873 max_s: 0.0,
1874 chunk_s: Vec::new(),
1875 worker_minflt: Vec::new(),
1876 share_frames: Vec::new(),
1877 share_bytes: Vec::new(),
1878 fail_lines: Vec::new(),
1879 };
1880 }
1881 if workers < 2 {
1882 let t_alloc = std::time::Instant::now();
1883 let mut slots = Vec::with_capacity(n);
1884 let alloc_s = t_alloc.elapsed().as_secs_f64();
1885 let flt0 = thread_minflt();
1886 let t_work = std::time::Instant::now();
1887 for i in 0..n {
1888 slots.push(parse_span(boundaries[i], span_end(boundaries, i, len)));
1889 }
1890 let work_s = t_work.elapsed().as_secs_f64();
1891 let worker_minflt = vec![thread_minflt().saturating_sub(flt0)];
1892 return SlotParse {
1893 slots,
1894 alloc_s,
1895 work_s,
1896 sum_s: work_s,
1897 max_s: work_s,
1898 chunk_s: vec![work_s],
1899 worker_minflt,
1900 share_frames: vec![n as u64],
1901 share_bytes: vec![len as u64],
1902 fail_lines: Vec::new(),
1903 };
1904 }
1905 let t_alloc = std::time::Instant::now();
1906 let mut slots = take_slot_buf(n);
1907 let alloc_s = t_alloc.elapsed().as_secs_f64();
1908 let ranges = byte_balanced_ranges(boundaries, len, workers);
1912 let chunk_ns: Vec<std::sync::atomic::AtomicU64> = (0..ranges.len())
1913 .map(|_| std::sync::atomic::AtomicU64::new(0))
1914 .collect();
1915 let chunk_flt: Vec<std::sync::atomic::AtomicI64> = (0..ranges.len())
1916 .map(|_| std::sync::atomic::AtomicI64::new(0))
1917 .collect();
1918 let slots_ptr = SlotWritePtr(slots.as_mut_ptr());
1919 let fail_lines = std::sync::Mutex::new(Vec::<(usize, u32)>::new());
1920 let t_work = std::time::Instant::now();
1921 {
1922 let parse_range = |range_index: usize, start: usize, end: usize| {
1923 if start >= end {
1924 return;
1925 }
1926 let flt0 = thread_minflt();
1927 let t_chunk = std::time::Instant::now();
1928 let byte_start = boundaries[start];
1929 let byte_end = span_end(boundaries, end - 1, len);
1930 let mut iter = ConFrameIterator::from_bytes(&bytes[byte_start..byte_end]);
1931 for i in start..end {
1932 let frame = match iter.next() {
1933 Some(Ok(frame)) => Ok(frame),
1934 Some(Err(error)) => {
1935 let line = share_cursor(iter.lines.line_no(), bytes, byte_start);
1936 fail_lines.lock().expect("fail lines").push((i, line));
1937 Err(error)
1938 }
1939 None => {
1940 let line = share_cursor(iter.lines.line_no(), bytes, byte_start);
1941 fail_lines.lock().expect("fail lines").push((i, line));
1942 Err(error::ParseError::IncompleteFrame)
1943 }
1944 };
1945 unsafe {
1948 slots_ptr.write_slot(i, frame);
1949 }
1950 }
1951 chunk_ns[range_index].store(
1952 t_chunk.elapsed().as_nanos() as u64,
1953 std::sync::atomic::Ordering::Relaxed,
1954 );
1955 chunk_flt[range_index].store(
1956 thread_minflt().saturating_sub(flt0),
1957 std::sync::atomic::Ordering::Relaxed,
1958 );
1959 };
1960 let run_ranges = || {
1961 use rayon::prelude::*;
1962 ranges
1963 .par_iter()
1964 .enumerate()
1965 .for_each(|(range_index, &(start, end))| {
1966 parse_range(range_index, start, end);
1967 });
1968 };
1969 if rayon::current_thread_index().is_some() {
1970 run_ranges();
1971 } else {
1972 match rayon::ThreadPoolBuilder::new().num_threads(workers).build() {
1973 Ok(pool) => pool.install(run_ranges),
1974 Err(_) => {
1975 for (range_index, &(start, end)) in ranges.iter().enumerate() {
1976 parse_range(range_index, start, end);
1977 }
1978 }
1979 }
1980 }
1981 }
1982 let work_s = t_work.elapsed().as_secs_f64();
1983 let mut worker_sum_s = 0.0_f64;
1984 let mut worker_max_s = 0.0_f64;
1985 let mut chunk_s = Vec::new();
1986 let mut worker_minflt = Vec::new();
1987 let mut share_frames = Vec::new();
1988 let mut share_bytes = Vec::new();
1989 for (range_index, &(start, end)) in ranges.iter().enumerate() {
1990 if start >= end {
1991 continue;
1992 }
1993 let seconds =
1994 chunk_ns[range_index].load(std::sync::atomic::Ordering::Relaxed) as f64 / 1.0e9;
1995 chunk_s.push(seconds);
1996 worker_sum_s += seconds;
1997 worker_max_s = worker_max_s.max(seconds);
1998 worker_minflt.push(chunk_flt[range_index].load(std::sync::atomic::Ordering::Relaxed));
1999 share_frames.push((end - start) as u64);
2000 let nbytes = span_end(boundaries, end - 1, len).saturating_sub(boundaries[start]);
2001 share_bytes.push(nbytes as u64);
2002 }
2003 let slots = unsafe {
2006 let mut raw = std::mem::ManuallyDrop::new(slots);
2007 Vec::from_raw_parts(
2008 raw.as_mut_ptr() as *mut Result<types::ConFrame, error::ParseError>,
2009 n,
2010 raw.capacity(),
2011 )
2012 };
2013 let fail_lines = fail_lines.into_inner().unwrap_or_default();
2014 SlotParse {
2015 slots,
2016 alloc_s,
2017 work_s,
2018 sum_s: worker_sum_s,
2019 max_s: worker_max_s,
2020 chunk_s,
2021 worker_minflt,
2022 share_frames,
2023 share_bytes,
2024 fail_lines,
2025 }
2026}
2027
2028#[cfg(feature = "parallel")]
2029fn parse_byte_span(
2030 bytes: &[u8],
2031 start: usize,
2032 end: usize,
2033) -> Result<types::ConFrame, error::ParseError> {
2034 let mut iter = ConFrameIterator::from_bytes(&bytes[start..end]);
2035 match iter.next() {
2036 Some(result) => result,
2037 None => Err(error::ParseError::IncompleteFrame),
2038 }
2039}
2040
2041#[cfg(feature = "parallel")]
2042fn parse_text_span(
2043 text: &str,
2044 start: usize,
2045 end: usize,
2046) -> Result<types::ConFrame, error::ParseError> {
2047 let mut iter = ConFrameIterator::new(&text[start..end]);
2048 match iter.next() {
2049 Some(result) => result,
2050 None => Err(error::ParseError::IncompleteFrame),
2051 }
2052}
2053
2054#[cfg(feature = "parallel")]
2055struct TimedParse {
2056 slots: Vec<Result<types::ConFrame, error::ParseError>>,
2057 pool: Option<std::sync::Arc<rayon::ThreadPool>>,
2060 pool_s: f64,
2061 pool_join_s: f64,
2063 scan_s: f64,
2064 scan_minflt: i64,
2065 slot_alloc_s: f64,
2066 worker_parse_s: f64,
2067 worker_sum_s: f64,
2068 worker_max_s: f64,
2069 chunk_s: Vec<f64>,
2070 worker_minflt: Vec<i64>,
2071 share_frames: Vec<u64>,
2072 share_bytes: Vec<u64>,
2073 fail_lines: Vec<(usize, u32)>,
2075 llc_misses: i64,
2076 numa_remote: i64,
2077 scan_kind: u8,
2079}
2080
2081#[cfg(feature = "parallel")]
2087fn keep_parse_pages() {
2088 #[cfg(target_os = "linux")]
2092 unsafe {
2093 libc::mallopt(-4, 0);
2094 libc::mallopt(-1, i32::MAX);
2095 }
2096}
2097
2098#[cfg(feature = "parallel")]
2104fn take_slot_buf(
2105 n: usize,
2106) -> Vec<std::mem::MaybeUninit<Result<types::ConFrame, error::ParseError>>> {
2107 let mut buf = slot_cache()
2108 .lock()
2109 .map(|mut slot| std::mem::take(&mut *slot))
2110 .unwrap_or_default();
2111 if buf.capacity() < n {
2112 buf = Vec::with_capacity(n);
2113 }
2114 unsafe { buf.set_len(n) };
2115 buf
2116}
2117
2118#[cfg(feature = "parallel")]
2119fn slot_cache()
2120-> &'static std::sync::Mutex<Vec<std::mem::MaybeUninit<Result<types::ConFrame, error::ParseError>>>>
2121{
2122 static CACHE: std::sync::Mutex<
2123 Vec<std::mem::MaybeUninit<Result<types::ConFrame, error::ParseError>>>,
2124 > = std::sync::Mutex::new(Vec::new());
2125 &CACHE
2126}
2127
2128#[cfg(feature = "parallel")]
2133#[allow(clippy::type_complexity)]
2134fn shared_pool(workers: usize) -> Option<std::sync::Arc<rayon::ThreadPool>> {
2135 use std::sync::{Arc, Mutex, OnceLock};
2136 static CACHE: OnceLock<Mutex<Vec<(usize, Arc<rayon::ThreadPool>)>>> = OnceLock::new();
2137 let cache = CACHE.get_or_init(|| Mutex::new(Vec::new()));
2138 let mut guard = cache.lock().ok()?;
2139 if let Some((_, pool)) = guard.iter().find(|(count, _)| *count == workers) {
2140 return Some(Arc::clone(pool));
2141 }
2142 keep_parse_pages();
2143 let pool = Arc::new(
2144 rayon::ThreadPoolBuilder::new()
2145 .num_threads(workers)
2146 .build()
2147 .ok()?,
2148 );
2149 guard.push((workers, Arc::clone(&pool)));
2150 Some(pool)
2151}
2152
2153#[cfg(feature = "parallel")]
2157pub fn process_minflt() -> i64 {
2158 #[cfg(unix)]
2159 {
2160 unsafe {
2161 let mut usage = std::mem::zeroed::<libc::rusage>();
2162 libc::getrusage(libc::RUSAGE_SELF, &mut usage);
2163 usage.ru_minflt
2164 }
2165 }
2166 #[cfg(not(unix))]
2167 {
2168 0
2169 }
2170}
2171
2172#[cfg(feature = "parallel")]
2174fn host_page_size() -> usize {
2175 #[cfg(unix)]
2176 {
2177 let raw = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
2178 if raw > 0 { raw as usize } else { 4096 }
2179 }
2180 #[cfg(not(unix))]
2181 {
2182 4096
2183 }
2184}
2185
2186#[cfg(feature = "parallel")]
2191fn first_touch_f64(values: &mut [f64], workers: usize) {
2192 use rayon::prelude::*;
2193 if values.is_empty() || workers < 2 {
2194 return;
2195 }
2196 let page = host_page_size().max(std::mem::size_of::<f64>());
2197 let step = (page / std::mem::size_of::<f64>()).max(1);
2198 let pages = values.len().div_ceil(step);
2199 let share = pages.div_ceil(workers);
2200 if rayon::current_thread_index().is_some() {
2201 values.par_chunks_mut(step).for_each(|chunk| {
2204 if let Some(slot) = chunk.first_mut() {
2205 *slot = 0.0;
2206 }
2207 });
2208 return;
2209 }
2210 std::thread::scope(|scope| {
2211 let mut rest = values;
2212 for worker in 0..workers {
2213 if rest.is_empty() {
2214 break;
2215 }
2216 let pages_here = share.min(rest.len().div_ceil(step));
2217 let take = (pages_here * step).min(rest.len());
2218 let (head, tail) = rest.split_at_mut(take);
2219 rest = tail;
2220 let _ = worker;
2221 scope.spawn(move || {
2222 let mut at = 0;
2223 while at < head.len() {
2224 head[at] = 0.0;
2225 at += step;
2226 }
2227 });
2228 }
2229 });
2230}
2231
2232#[cfg(feature = "parallel")]
2238fn finish_results(
2239 pool: Option<&rayon::ThreadPool>,
2240 slots: Vec<Result<types::ConFrame, error::ParseError>>,
2241) -> Result<Vec<types::ConFrame>, error::ParseError> {
2242 if slots.len() < 2 || pool.is_none() {
2243 let mut frames = Vec::with_capacity(slots.len());
2244 for part in slots {
2245 frames.push(part?);
2246 }
2247 return Ok(frames);
2248 }
2249 let pool = pool.expect("pool");
2250 pool.install(|| {
2251 use rayon::prelude::*;
2252 if let Some(index) = slots.par_iter().position_first(|item| item.is_err()) {
2253 return Err(match slots.into_iter().nth(index) {
2254 Some(Err(err)) => err,
2255 _ => error::ParseError::IncompleteFrame,
2256 });
2257 }
2258 Ok(slots
2259 .into_par_iter()
2260 .map(|item| item.expect("ok"))
2261 .collect())
2262 })
2263}
2264
2265#[cfg(feature = "parallel")]
2267pub const SCAN_SERIAL: u8 = 0;
2268#[cfg(feature = "parallel")]
2275pub const SCAN_INDEXED: u8 = 1;
2276#[cfg(feature = "parallel")]
2278pub const SCAN_EQUAL_SPAN: u8 = 2;
2279
2280#[cfg(feature = "parallel")]
2288#[cfg_attr(not(test), allow(dead_code))]
2289fn equal_span_boundaries(bytes: &[u8], workers: usize) -> Option<Vec<usize>> {
2290 if workers < 2 || bytes.len() < 2 {
2291 return None;
2292 }
2293 let mut first = ConFrameIterator::from_bytes(bytes);
2294 match first.forward_fast() {
2295 Some(Ok(())) => {}
2296 _ => return None,
2297 }
2298 let span = first.lines.pos;
2299 if span < 2 {
2300 return None;
2301 }
2302 let n = bytes.len() / span;
2303 if n < 2 {
2304 return None;
2305 }
2306 let tiled = n * span;
2309 if bytes[tiled..].iter().any(|b| !b.is_ascii_whitespace()) {
2310 return None;
2311 }
2312 let byte_share = tiled.div_ceil(workers);
2316 for worker in 0..workers {
2317 let byte_cut = worker.saturating_mul(byte_share).min(tiled);
2318 if byte_cut >= tiled {
2319 continue;
2320 }
2321 let index = if worker == 0 {
2322 0
2323 } else {
2324 byte_cut.div_ceil(span).min(n - 1)
2325 };
2326 let start = index.saturating_mul(span);
2327 if start >= tiled {
2328 continue;
2329 }
2330 let end = (start + span).min(tiled);
2331 let mut iter = ConFrameIterator::from_bytes(&bytes[start..end]);
2332 match iter.forward_fast() {
2333 Some(Ok(())) if iter.lines.pos == end - start => {}
2334 _ => return None,
2335 }
2336 }
2337 Some(arithmetic_starts(span, n))
2338}
2339
2340#[cfg(feature = "parallel")]
2345static START_CACHE: std::sync::Mutex<Vec<usize>> = std::sync::Mutex::new(Vec::new());
2346
2347#[cfg(feature = "parallel")]
2348#[cfg_attr(not(test), allow(dead_code))]
2349fn arithmetic_starts(span: usize, n: usize) -> Vec<usize> {
2350 let mut starts = START_CACHE
2351 .lock()
2352 .map(|mut slot| std::mem::take(&mut *slot))
2353 .unwrap_or_default();
2354 if starts.len() != n {
2355 starts.resize(n, 0);
2356 }
2357 for (index, slot) in starts.iter_mut().enumerate() {
2358 *slot = index * span;
2359 }
2360 starts
2361}
2362
2363#[cfg(feature = "parallel")]
2364fn recycle_starts(starts: Vec<usize>) {
2365 if let Ok(mut slot) = START_CACHE.lock()
2366 && starts.capacity() >= slot.capacity()
2367 {
2368 *slot = starts;
2369 }
2370}
2371
2372#[cfg(feature = "parallel")]
2373fn parse_frames_parallel_bytes(
2374 bytes: &[u8],
2375 num_threads: Option<usize>,
2376) -> Vec<Result<types::ConFrame, error::ParseError>> {
2377 parse_bytes_timed(bytes, num_threads).slots
2378}
2379
2380#[cfg(feature = "parallel")]
2387fn parse_bytes_timed(bytes: &[u8], num_threads: Option<usize>) -> TimedParse {
2388 let parse_workers = match num_threads {
2389 Some(n) if n >= 2 => n,
2390 _ => 1,
2391 };
2392 let run = || {
2393 let fault0 = process_minflt();
2394 let t_scan = std::time::Instant::now();
2395 let (boundaries, scan_kind) =
2399 if let Some(bounds) = equal_span_boundaries(bytes, parse_workers) {
2400 (bounds, SCAN_EQUAL_SPAN)
2401 } else {
2402 (
2403 collect_frame_starts(ConFrameIterator::from_bytes(bytes), true),
2404 SCAN_SERIAL,
2405 )
2406 };
2407 let scan_s = t_scan.elapsed().as_secs_f64();
2408 let scan_minflt = process_minflt().saturating_sub(fault0);
2409 let parsed = parse_into_slots_timed(
2410 bytes,
2411 bytes.len(),
2412 &boundaries,
2413 parse_workers,
2414 |start, end| parse_byte_span(bytes, start, end),
2415 );
2416 recycle_starts(boundaries);
2417 (parsed, scan_s, scan_kind, scan_minflt)
2418 };
2419 let (pool_s, pool, packed) = match num_threads {
2420 None => (0.0, None, run()),
2421 Some(n) if n <= 1 => {
2422 let _ = n;
2423 keep_parse_pages();
2424 let fault0 = process_minflt();
2425 let t_scan = std::time::Instant::now();
2426 let boundaries = collect_frame_starts(ConFrameIterator::from_bytes(bytes), true);
2427 let scan_s = t_scan.elapsed().as_secs_f64();
2428 let scan_minflt = process_minflt().saturating_sub(fault0);
2429 let parsed =
2430 parse_into_slots_timed(bytes, bytes.len(), &boundaries, 1, |start, end| {
2431 parse_byte_span(bytes, start, end)
2432 });
2433 (0.0, None, (parsed, scan_s, SCAN_SERIAL, scan_minflt))
2434 }
2435 Some(n) => {
2436 let t_pool = std::time::Instant::now();
2437 match shared_pool(n) {
2438 Some(pool) => {
2439 let built = t_pool.elapsed().as_secs_f64();
2440 let t_wake = std::time::Instant::now();
2441 let (wake, body) = pool.install(|| {
2442 let wake = t_wake.elapsed().as_secs_f64();
2443 (wake, run())
2444 });
2445 (built + wake, Some(pool), body)
2448 }
2449 None => (t_pool.elapsed().as_secs_f64(), None, run()),
2450 }
2451 }
2452 };
2453 let (parsed, scan_s, scan_kind, scan_minflt) = packed;
2454 TimedParse {
2455 slots: parsed.slots,
2456 pool,
2457 pool_s,
2458 pool_join_s: 0.0,
2459 scan_s,
2460 scan_minflt,
2461 slot_alloc_s: parsed.alloc_s,
2462 worker_parse_s: parsed.work_s,
2463 worker_sum_s: parsed.sum_s,
2464 worker_max_s: parsed.max_s,
2465 chunk_s: parsed.chunk_s,
2466 worker_minflt: parsed.worker_minflt,
2467 share_frames: parsed.share_frames,
2468 share_bytes: parsed.share_bytes,
2469 fail_lines: parsed.fail_lines,
2470 llc_misses: sample_llc_misses(),
2471 numa_remote: numa_remote_pages(),
2472 scan_kind,
2473 }
2474}
2475
2476pub fn count_frames(path: &Path) -> Result<usize, LoadError> {
2483 let contents = load_text(path)?;
2484 let text = contents.as_str().map_err(LoadError::Utf8)?;
2485 let mut n = 0usize;
2486 let mut iter = ConFrameIterator::new(text);
2487 loop {
2488 match iter.forward_fast() {
2489 Some(Ok(())) => n += 1,
2490 Some(Err(e)) => {
2491 return Err(LoadError::Parse(error::LocatedParseError::new(
2492 e,
2493 iter.lines.line_no(),
2494 iter.frame_index(),
2495 )));
2496 }
2497 None => break,
2498 }
2499 }
2500 Ok(n)
2501}
2502
2503pub fn read_first_frame(path: &Path) -> Result<types::ConFrame, Box<dyn std::error::Error>> {
2508 let contents = crate::compression::read_file_contents(path)?;
2509 let text = contents.as_str()?;
2510 let mut iter = ConFrameIterator::new(text);
2511 match iter.next() {
2512 Some(Ok(frame)) => Ok(frame),
2513 Some(Err(e)) => Err(Box::new(e)),
2514 None => Err("No frames found in file".into()),
2515 }
2516}
2517
2518pub fn read_nth_frame(
2522 path: &Path,
2523 index: usize,
2524) -> Result<types::ConFrame, Box<dyn std::error::Error>> {
2525 let contents = crate::compression::read_file_contents(path)?;
2526 let text = contents.as_str()?;
2527 read_nth_frame_from_text(text, index)
2528}
2529
2530pub fn read_nth_frame_from_text(
2532 text: &str,
2533 index: usize,
2534) -> Result<types::ConFrame, Box<dyn std::error::Error>> {
2535 let mut iter = ConFrameIterator::new(text);
2536 let skipped = iter.skip_frames(index)?;
2537 if skipped < index {
2538 return Err(format!("frame {index} out of range (only {skipped} frames)").into());
2539 }
2540 match iter.next() {
2541 Some(Ok(frame)) => Ok(frame),
2542 Some(Err(e)) => Err(Box::new(e)),
2543 None => Err(format!("frame {index} out of range").into()),
2544 }
2545}
2546
2547#[cfg(feature = "parallel")]
2553pub fn read_all_frames_indexed(
2554 path: &Path,
2555 num_threads: Option<usize>,
2556) -> Result<Vec<types::ConFrame>, Box<dyn std::error::Error>> {
2557 let contents = crate::compression::read_file_contents(path)?;
2558 let text = contents.as_str_sliced()?;
2559 let idx = match contents {
2560 crate::compression::FileContents::Mapped(_) => {
2561 crate::frame_index::load_or_build(path, text, true)?
2562 }
2563 crate::compression::FileContents::Owned(_) => {
2564 let on_disk = std::fs::metadata(path)?.len();
2565 crate::frame_index::load_or_build(path, text, on_disk == text.len() as u64)?
2566 }
2567 };
2568 let offs: Vec<usize> = idx.offsets.iter().map(|&o| o as usize).collect();
2569 parse_frames_at_offsets(text, &offs, num_threads)
2570 .into_iter()
2571 .map(|r| r.map_err(|e| Box::new(e) as Box<dyn std::error::Error>))
2572 .collect()
2573}
2574
2575enum IndexStep {
2576 Stay(usize),
2577 Missing(usize),
2578 Landed(usize, usize),
2579 Short(usize, usize),
2580}
2581
2582struct IndexedSpan {
2583 start: usize,
2584 content_end: usize,
2585 next_pos: usize,
2586}
2587
2588fn line_count(nl: &[usize], len: usize) -> usize {
2590 if len == 0 {
2591 0
2592 } else if nl.last().copied() == Some(len - 1) {
2593 nl.len()
2594 } else {
2595 nl.len() + 1
2596 }
2597}
2598
2599fn line_start_pos(nl: &[usize], len: usize, line_no: usize) -> usize {
2600 if line_no == 0 {
2601 return 0;
2602 }
2603 if line_no >= line_count(nl, len) {
2604 return len;
2605 }
2606 nl[line_no - 1] + 1
2607}
2608
2609fn line_no_for_pos(nl: &[usize], len: usize, pos: usize) -> usize {
2611 if pos >= len {
2612 return line_count(nl, len);
2613 }
2614 nl.partition_point(|&p| p < pos)
2615}
2616
2617fn indexed_span(nl: &[usize], bytes: &[u8], line_no: usize) -> Option<IndexedSpan> {
2618 let len = bytes.len();
2619 if line_no >= line_count(nl, len) {
2620 return None;
2621 }
2622 let start = if line_no == 0 { 0 } else { nl[line_no - 1] + 1 };
2623 let (raw_end, next_pos) = if line_no < nl.len() {
2624 (nl[line_no], nl[line_no] + 1)
2625 } else {
2626 (len, len)
2627 };
2628 let mut content_end = raw_end;
2629 if content_end > start && bytes[content_end - 1] == b'\r' {
2630 content_end -= 1;
2631 }
2632 Some(IndexedSpan {
2633 start,
2634 content_end,
2635 next_pos,
2636 })
2637}
2638
2639#[cfg(feature = "parallel")]
2641fn equal_cuts(len: usize, n: usize) -> Vec<usize> {
2642 let n = n.max(1);
2643 let mut cuts = Vec::with_capacity(n + 1);
2644 for i in 0..=n {
2645 let at = ((i as u128) * (len as u128) / (n as u128)) as usize;
2646 let at = at.min(len);
2647 if cuts.last().copied() != Some(at) {
2648 cuts.push(at);
2649 }
2650 }
2651 if cuts.last().copied() != Some(len) {
2652 cuts.push(len);
2653 }
2654 cuts
2655}
2656
2657#[cfg(feature = "parallel")]
2664fn newline_positions_at_cuts(bytes: &[u8], cuts: &[usize]) -> Vec<usize> {
2665 let spans: Vec<(usize, usize)> = cuts
2666 .windows(2)
2667 .map(|w| (w[0].min(bytes.len()), w[1].min(bytes.len())))
2668 .filter(|(start, end)| start < end)
2669 .collect();
2670 if spans.len() <= 1 {
2671 let (start, end) = spans.first().copied().unwrap_or((0, 0));
2672 return memchr::memchr_iter(b'\n', &bytes[start..end])
2673 .map(|p| start + p)
2674 .collect();
2675 }
2676 use rayon::prelude::*;
2680 let parts: Vec<Vec<usize>> = spans
2681 .par_iter()
2682 .map(|&(start, end)| {
2683 memchr::memchr_iter(b'\n', &bytes[start..end])
2684 .map(|p| start + p)
2685 .collect()
2686 })
2687 .collect();
2688 let mut all = Vec::with_capacity(bytes.len() / 32);
2689 for part in parts {
2690 all.extend(part);
2691 }
2692 all
2693}
2694
2695#[cfg(feature = "parallel")]
2696fn newline_positions(bytes: &[u8], threads: usize) -> Vec<usize> {
2697 newline_positions_at_cuts(bytes, &equal_cuts(bytes.len(), threads))
2698}
2699
2700#[cfg(feature = "parallel")]
2710fn scan_thread_count(num_threads: Option<usize>, len: usize) -> usize {
2711 let requested = match num_threads {
2712 Some(n) if n >= 2 => n,
2713 _ => return 1,
2714 };
2715 if len < 2 { 1 } else { requested }
2716}
2717
2718#[derive(Debug)]
2720pub enum LoadError {
2721 Io(std::io::Error),
2722 Utf8(std::str::Utf8Error),
2723 Parse(error::LocatedParseError),
2724 Empty,
2726}
2727
2728impl std::fmt::Display for LoadError {
2729 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2730 match self {
2731 LoadError::Io(e) => write!(f, "{e}"),
2732 LoadError::Utf8(e) => write!(f, "{e}"),
2733 LoadError::Parse(e) => write!(f, "{e}"),
2734 LoadError::Empty => write!(f, "no frames found in file"),
2735 }
2736 }
2737}
2738
2739impl std::error::Error for LoadError {}
2740
2741fn load_error_from_box(err: Box<dyn std::error::Error>) -> LoadError {
2742 if let Some(io) = err.downcast_ref::<std::io::Error>() {
2743 return LoadError::Io(std::io::Error::new(io.kind(), io.to_string()));
2744 }
2745 LoadError::Io(std::io::Error::other(err.to_string()))
2746}
2747
2748fn load_text(path: &Path) -> Result<crate::compression::FileContents, LoadError> {
2749 crate::compression::read_file_contents(path).map_err(load_error_from_box)
2750}
2751
2752#[cfg(feature = "parallel")]
2754fn file_line_at(text: &str, byte: usize) -> u32 {
2755 let end = byte.min(text.len());
2756 1 + text.as_bytes()[..end]
2757 .iter()
2758 .filter(|&&b| b == b'\n')
2759 .count() as u32
2760}
2761
2762pub fn frames_from_text_located(
2764 text: &str,
2765 num_threads: Option<usize>,
2766) -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
2767 #[cfg(feature = "parallel")]
2768 {
2769 let use_parallel = matches!(num_threads, Some(n) if n >= 2);
2772 if use_parallel {
2773 let pool = num_threads.filter(|&n| n >= 2);
2774 return collect_parallel_located(text, pool);
2775 }
2776 }
2777 #[cfg(not(feature = "parallel"))]
2778 {
2779 let _ = num_threads;
2780 }
2781 let mut frames = Vec::new();
2782 let mut iter = ConFrameIterator::new(text);
2783 let mut index = 0u64;
2784 while let Some(item) = iter.next() {
2785 match item {
2786 Ok(frame) => {
2787 frames.push(frame);
2788 index += 1;
2789 }
2790 Err(e) => {
2791 return Err(error::LocatedParseError::new(
2792 e,
2793 iter.lines.line_no(),
2794 index,
2795 ));
2796 }
2797 }
2798 }
2799 Ok(frames)
2800}
2801
2802pub fn read_all_frames_located(
2804 path: &Path,
2805 num_threads: Option<usize>,
2806) -> Result<Vec<types::ConFrame>, LoadError> {
2807 let contents = load_text(path)?;
2808 let text = contents.as_str().map_err(LoadError::Utf8)?;
2809 frames_from_text_located(text, num_threads).map_err(LoadError::Parse)
2810}
2811
2812pub fn read_first_frame_located(path: &Path) -> Result<types::ConFrame, LoadError> {
2814 let contents = load_text(path)?;
2815 let text = contents.as_str().map_err(LoadError::Utf8)?;
2816 let mut iter = ConFrameIterator::new(text);
2817 match iter.next() {
2818 Some(Ok(frame)) => Ok(frame),
2819 Some(Err(e)) => Err(LoadError::Parse(error::LocatedParseError::new(
2820 e,
2821 iter.lines.line_no(),
2822 0,
2823 ))),
2824 None => Err(LoadError::Empty),
2825 }
2826}
2827
2828pub fn read_nth_frame_located(path: &Path, index: usize) -> Result<types::ConFrame, LoadError> {
2830 let contents = load_text(path)?;
2831 let text = contents.as_str().map_err(LoadError::Utf8)?;
2832 let mut iter = ConFrameIterator::new(text);
2833 match iter.skip_frames_counted(index) {
2834 Ok(skipped) if skipped == index => {}
2835 Ok(skipped) => {
2836 return Err(LoadError::Parse(error::LocatedParseError::new(
2837 error::ParseError::IncompleteFrame,
2838 iter.lines.line_no(),
2839 skipped as u64,
2840 )));
2841 }
2842 Err((failed_at, e)) => {
2843 return Err(LoadError::Parse(error::LocatedParseError::new(
2844 e,
2845 iter.lines.line_no(),
2846 failed_at as u64,
2847 )));
2848 }
2849 }
2850 match iter.next() {
2851 Some(Ok(frame)) => Ok(frame),
2852 Some(Err(e)) => Err(LoadError::Parse(error::LocatedParseError::new(
2853 e,
2854 iter.lines.line_no(),
2855 index as u64,
2856 ))),
2857 None => Err(LoadError::Parse(error::LocatedParseError::new(
2858 error::ParseError::IncompleteFrame,
2859 iter.lines.line_no(),
2860 index as u64,
2861 ))),
2862 }
2863}
2864
2865fn collect_frame_starts(scanner: ConFrameIterator<'_>, skip_whitespace_tail: bool) -> Vec<usize> {
2866 #[cfg(target_arch = "x86_64")]
2867 {
2868 if std::is_x86_feature_detected!("avx2") {
2869 return unsafe { collect_frame_starts_avx2(scanner, skip_whitespace_tail) };
2873 }
2874 }
2875 collect_frame_starts_inner(scanner, skip_whitespace_tail)
2876}
2877
2878#[cfg(target_arch = "x86_64")]
2879#[target_feature(enable = "avx2,popcnt")]
2880unsafe fn collect_frame_starts_avx2(
2881 scanner: ConFrameIterator<'_>,
2882 skip_whitespace_tail: bool,
2883) -> Vec<usize> {
2884 collect_frame_starts_inner(scanner, skip_whitespace_tail)
2885}
2886
2887#[inline(always)]
2888fn collect_frame_starts_inner(
2889 mut scanner: ConFrameIterator<'_>,
2890 skip_whitespace_tail: bool,
2891) -> Vec<usize> {
2892 let mut boundaries = Vec::new();
2893 loop {
2894 scanner.lines.clear_peek();
2895 let start = scanner.lines.pos;
2896 if start >= scanner.lines.bytes.len() {
2897 break;
2898 }
2899 if skip_whitespace_tail {
2900 let rest = &scanner.lines.bytes[start..];
2901 if rest.iter().all(u8::is_ascii_whitespace) {
2902 break;
2903 }
2904 }
2905 boundaries.push(start);
2906 match scanner.forward_fast_to(None) {
2907 Some(Ok(())) => {}
2908 Some(Err(_)) | None => break,
2909 }
2910 }
2911 boundaries
2912}
2913
2914pub fn frame_start_offsets_serial(file_contents: &str) -> Vec<usize> {
2916 collect_frame_starts(ConFrameIterator::new(file_contents), false)
2917}
2918
2919#[cfg(feature = "parallel")]
2920fn scanner_with_newlines<'a>(text: &'a str, newlines: Vec<usize>) -> ConFrameIterator<'a> {
2921 let mut scanner = ConFrameIterator::new(text);
2922 scanner.lines.index_line = 0;
2923 scanner.lines.newlines = Some(newlines);
2924 scanner
2925}
2926
2927#[cfg(feature = "parallel")]
2928fn frame_start_offsets_indexed(text: &str, newlines: Vec<usize>) -> Vec<usize> {
2929 collect_frame_starts(scanner_with_newlines(text, newlines), false)
2930}
2931
2932#[cfg(all(feature = "parallel", test))]
2938fn frame_start_offsets_at_cuts(text: &str, cuts: &[usize]) -> Vec<usize> {
2939 let newlines = newline_positions_at_cuts(text.as_bytes(), cuts);
2940 frame_start_offsets_indexed(text, newlines)
2941}
2942
2943pub fn frame_start_offsets(file_contents: &str) -> Vec<usize> {
2955 frame_start_offsets_with_threads(file_contents, None)
2956}
2957
2958pub fn frame_start_offsets_with_threads(
2966 file_contents: &str,
2967 num_threads: Option<usize>,
2968) -> Vec<usize> {
2969 #[cfg(feature = "parallel")]
2970 {
2971 let threads = scan_thread_count(num_threads, file_contents.len());
2972 if threads >= 2 {
2973 let newlines = newline_positions(file_contents.as_bytes(), threads);
2974 return frame_start_offsets_indexed(file_contents, newlines);
2975 }
2976 }
2977 #[cfg(not(feature = "parallel"))]
2978 {
2979 let _ = num_threads;
2980 }
2981 frame_start_offsets_serial(file_contents)
2982}
2983
2984#[cfg(feature = "parallel")]
2993pub fn parse_frames_parallel(
2994 file_contents: &str,
2995) -> Vec<Result<types::ConFrame, error::ParseError>> {
2996 parse_frames_parallel_with_threads(file_contents, None)
2997}
2998
2999#[cfg(feature = "parallel")]
3009pub fn parse_frames_parallel_with_threads(
3010 file_contents: &str,
3011 num_threads: Option<usize>,
3012) -> Vec<Result<types::ConFrame, error::ParseError>> {
3013 let parse_workers = match num_threads {
3014 Some(n) if n >= 2 => n,
3015 _ => 1,
3016 };
3017 let run = || {
3018 let boundaries = collect_frame_starts(ConFrameIterator::new(file_contents), true);
3022 parse_into_slots(
3023 file_contents.as_bytes(),
3024 file_contents.len(),
3025 &boundaries,
3026 parse_workers,
3027 |start, end| parse_text_span(file_contents, start, end),
3028 )
3029 };
3030 match num_threads {
3031 Some(n) if n >= 2 => match rayon::ThreadPoolBuilder::new().num_threads(n).build() {
3032 Ok(pool) => pool.install(run),
3033 Err(_) => run(),
3034 },
3035 _ => run(),
3036 }
3037}
3038
3039#[cfg(feature = "parallel")]
3054pub fn parse_frames_at_offsets(
3055 file_contents: &str,
3056 boundaries: &[usize],
3057 num_threads: Option<usize>,
3058) -> Vec<Result<types::ConFrame, error::ParseError>> {
3059 let workers = match num_threads {
3060 Some(n) if n <= 1 => 1,
3061 Some(n) => n,
3062 None => rayon::current_num_threads().max(1),
3063 };
3064 let run = || {
3065 parse_into_slots(
3066 file_contents.as_bytes(),
3067 file_contents.len(),
3068 boundaries,
3069 workers,
3070 |start, end| parse_text_span(file_contents, start, end),
3071 )
3072 };
3073 match num_threads {
3074 Some(n) if n >= 2 => match rayon::ThreadPoolBuilder::new().num_threads(n).build() {
3075 Ok(pool) => pool.install(run),
3076 Err(_) => run(),
3077 },
3078 _ => run(),
3079 }
3080}
3081
3082#[derive(Clone, Copy, Debug, Default)]
3085pub struct ReadPhaseTimes {
3086 pub open_mmap_s: f64,
3087 pub pool_s: f64,
3089 pub pool_join_s: f64,
3092 pub scan_s: f64,
3093 pub slot_alloc_s: f64,
3094 pub worker_parse_s: f64,
3095 pub finish_s: f64,
3097 pub worker_sum_s: f64,
3099 pub worker_max_s: f64,
3101 pub close_s: f64,
3103 pub schedule_s: f64,
3105 pub scan_minflt: i64,
3107 pub parse_minflt: i64,
3109 pub llc_misses: i64,
3111 pub numa_remote: i64,
3113 pub slab_minflt: i64,
3115 pub touch_minflt: i64,
3117 pub mapped: bool,
3119}
3120
3121pub struct PhasedFrames {
3123 pub frames: Vec<types::ConFrame>,
3124 pub phases: ReadPhaseTimes,
3125 pub n_bytes: usize,
3126 pub chunk_s: Vec<f64>,
3128 pub scan_kind: u8,
3131 pub worker_minflt: Vec<i64>,
3133 pub share_frames: Vec<u64>,
3135 pub share_bytes: Vec<u64>,
3137 #[cfg(feature = "parallel")]
3139 pub pool: Option<std::sync::Arc<rayon::ThreadPool>>,
3140}
3141
3142pub fn read_all_frames_phased(
3152 path: &Path,
3153 num_threads: Option<usize>,
3154) -> Result<PhasedFrames, Box<dyn std::error::Error>> {
3155 let t_all = std::time::Instant::now();
3156 let t_open = std::time::Instant::now();
3157 let contents = crate::compression::read_file_contents(path)?;
3158 let open_mmap_s = t_open.elapsed().as_secs_f64();
3159 let mapped = matches!(contents, crate::compression::FileContents::Mapped(_));
3160 let n_bytes = match &contents {
3161 crate::compression::FileContents::Owned(text) => text.len(),
3162 crate::compression::FileContents::Mapped(map) => map.len(),
3163 };
3164 let bytes: &[u8] = match &contents {
3165 crate::compression::FileContents::Owned(text) => text.as_bytes(),
3166 crate::compression::FileContents::Mapped(map) => map,
3167 };
3168 let phased = frames_from_bytes_phased_located(bytes, num_threads)?;
3169 let workers = match num_threads {
3170 Some(n) if n >= 2 => n,
3171 _ => 1,
3172 };
3173 let t_close = std::time::Instant::now();
3174 if let crate::compression::FileContents::Mapped(map) = &contents {
3175 if workers >= 2 {
3180 #[cfg(feature = "parallel")]
3181 {
3182 if let Some(pool) = phased.pool.as_ref() {
3183 pool.install(|| {
3184 crate::compression::advise_dontneed_parallel(map, workers);
3185 });
3186 }
3187 }
3188 #[cfg(not(feature = "parallel"))]
3189 {
3190 let _ = (map, workers);
3191 }
3192 }
3193 }
3194 drop(contents);
3195 let close_s = t_close.elapsed().as_secs_f64();
3196 let elapsed = t_all.elapsed().as_secs_f64();
3197 let named = open_mmap_s
3198 + phased.pool_s
3199 + phased.pool_join_s
3200 + phased.scan_s
3201 + phased.slot_alloc_s
3202 + phased.worker_parse_s
3203 + phased.finish_s
3204 + close_s;
3205 let schedule_s = (elapsed - named).max(0.0);
3206 let parse_minflt = phased.worker_minflt.iter().sum();
3207 Ok(PhasedFrames {
3208 frames: phased.frames,
3209 n_bytes,
3210 chunk_s: phased.chunk_s,
3211 scan_kind: phased.scan_kind,
3212 worker_minflt: phased.worker_minflt,
3213 share_frames: phased.share_frames,
3214 share_bytes: phased.share_bytes,
3215 phases: ReadPhaseTimes {
3216 open_mmap_s,
3217 pool_s: phased.pool_s,
3218 pool_join_s: phased.pool_join_s,
3219 scan_s: phased.scan_s,
3220 slot_alloc_s: phased.slot_alloc_s,
3221 worker_parse_s: phased.worker_parse_s,
3222 finish_s: phased.finish_s,
3223 worker_sum_s: phased.worker_sum_s,
3224 worker_max_s: phased.worker_max_s,
3225 close_s,
3226 schedule_s,
3227 scan_minflt: phased.scan_minflt,
3228 parse_minflt,
3229 llc_misses: phased.llc_misses,
3230 numa_remote: phased.numa_remote,
3231 mapped,
3232 ..ReadPhaseTimes::default()
3233 },
3234 #[cfg(feature = "parallel")]
3235 pool: phased.pool,
3236 })
3237}
3238
3239pub struct PhasedByteTimes {
3245 pub frames: Vec<types::ConFrame>,
3246 pub pool_s: f64,
3247 pub pool_join_s: f64,
3248 pub scan_s: f64,
3249 pub slot_alloc_s: f64,
3250 pub worker_parse_s: f64,
3251 pub finish_s: f64,
3252 pub worker_sum_s: f64,
3253 pub worker_max_s: f64,
3254 pub chunk_s: Vec<f64>,
3256 pub scan_kind: u8,
3259 pub scan_minflt: i64,
3261 pub worker_minflt: Vec<i64>,
3263 pub share_frames: Vec<u64>,
3264 pub share_bytes: Vec<u64>,
3265 pub llc_misses: i64,
3267 pub numa_remote: i64,
3269 #[cfg(feature = "parallel")]
3271 pub pool: Option<std::sync::Arc<rayon::ThreadPool>>,
3272}
3273
3274pub fn frames_from_bytes_phased(
3275 bytes: &[u8],
3276 num_threads: Option<usize>,
3277) -> Result<PhasedByteTimes, error::ParseError> {
3278 frames_from_bytes_phased_located(bytes, num_threads).map_err(|err| err.error)
3279}
3280
3281fn frames_from_bytes_phased_located(
3282 bytes: &[u8],
3283 num_threads: Option<usize>,
3284) -> Result<PhasedByteTimes, error::LocatedParseError> {
3285 #[cfg(feature = "parallel")]
3286 {
3287 if !matches!(num_threads, Some(n) if n >= 2) {
3290 let flt0 = thread_minflt();
3291 let t_work = std::time::Instant::now();
3292 let frames = collect_bytes_located(bytes)?;
3293 let worker_parse_s = t_work.elapsed().as_secs_f64();
3294 let worker_flt = thread_minflt().saturating_sub(flt0);
3295 let n_frames = frames.len() as u64;
3296 return Ok(PhasedByteTimes {
3297 frames,
3298 pool_s: 0.0,
3299 pool_join_s: 0.0,
3300 scan_s: 0.0,
3301 slot_alloc_s: 0.0,
3302 worker_parse_s,
3303 finish_s: 0.0,
3304 worker_sum_s: worker_parse_s,
3305 worker_max_s: worker_parse_s,
3306 chunk_s: vec![worker_parse_s],
3307 scan_kind: SCAN_SERIAL,
3308 scan_minflt: 0,
3309 worker_minflt: vec![worker_flt],
3310 share_frames: vec![n_frames],
3311 share_bytes: vec![bytes.len() as u64],
3312 llc_misses: sample_llc_misses(),
3313 numa_remote: numa_remote_pages(),
3314 pool: None,
3315 });
3316 }
3317 let requested = Some(num_threads.expect("workers"));
3318 let timed = parse_bytes_timed(bytes, requested);
3319 let t_finish = std::time::Instant::now();
3320 let frames = if timed.slots.iter().any(|item| item.is_err()) {
3321 collect_located_slots(timed.slots, &timed.fail_lines)?
3322 } else {
3323 finish_results(timed.pool.as_deref(), timed.slots)
3324 .map_err(|error| error::LocatedParseError::new(error, 0, 0))?
3325 };
3326 let finish_s = t_finish.elapsed().as_secs_f64();
3327 Ok(PhasedByteTimes {
3328 frames,
3329 pool_s: timed.pool_s,
3330 pool_join_s: timed.pool_join_s,
3331 scan_s: timed.scan_s,
3332 slot_alloc_s: timed.slot_alloc_s,
3333 worker_parse_s: timed.worker_parse_s,
3334 finish_s,
3335 worker_sum_s: timed.worker_sum_s,
3336 worker_max_s: timed.worker_max_s,
3337 chunk_s: timed.chunk_s,
3338 scan_kind: timed.scan_kind,
3339 scan_minflt: timed.scan_minflt,
3340 worker_minflt: timed.worker_minflt,
3341 share_frames: timed.share_frames,
3342 share_bytes: timed.share_bytes,
3343 llc_misses: timed.llc_misses,
3344 numa_remote: timed.numa_remote,
3345 pool: timed.pool,
3346 })
3347 }
3348 #[cfg(not(feature = "parallel"))]
3349 {
3350 let _ = num_threads;
3351 let t_work = std::time::Instant::now();
3352 let frames = collect_bytes_located(bytes)?;
3353 let worker_parse_s = t_work.elapsed().as_secs_f64();
3354 Ok(PhasedByteTimes {
3355 frames,
3356 pool_s: 0.0,
3357 pool_join_s: 0.0,
3358 scan_s: 0.0,
3359 slot_alloc_s: 0.0,
3360 worker_parse_s,
3361 finish_s: 0.0,
3362 worker_sum_s: worker_parse_s,
3363 worker_max_s: worker_parse_s,
3364 chunk_s: vec![worker_parse_s],
3365 scan_kind: 0,
3366 scan_minflt: 0,
3367 worker_minflt: Vec::new(),
3368 share_frames: Vec::new(),
3369 share_bytes: Vec::new(),
3370 llc_misses: -1,
3371 numa_remote: 0,
3372 })
3373 }
3374}
3375
3376pub fn pack_position_slab(frames: &[types::ConFrame], workers: usize) -> Option<Vec<f64>> {
3382 if frames.is_empty() {
3383 return None;
3384 }
3385 let n_atoms = frames[0].positions.nrows();
3386 if frames
3387 .iter()
3388 .any(|frame| frame.positions.nrows() != n_atoms)
3389 {
3390 return None;
3391 }
3392 let stride = n_atoms.saturating_mul(3);
3393 let total = frames.len().saturating_mul(stride);
3394 let mut values = Vec::<f64>::with_capacity(total);
3395 if total > 0 {
3396 unsafe {
3400 values.set_len(total);
3401 }
3402 }
3403 if stride == 0 {
3404 return Some(values);
3405 }
3406 let write = |frame: &types::ConFrame, dest: &mut [f64]| {
3407 if let Some(src) = frame.positions.f64_slice() {
3408 dest.copy_from_slice(src);
3409 } else {
3410 for (row, slot) in dest.as_chunks_mut::<3>().0.iter_mut().enumerate() {
3411 slot.copy_from_slice(&frame.positions.as_f64_row(row));
3412 }
3413 }
3414 };
3415 #[cfg(feature = "parallel")]
3416 {
3417 if workers >= 2 {
3418 let chunk_frames = frames.len().div_ceil(workers).max(1);
3419 let chunk_vals = chunk_frames.saturating_mul(stride);
3420 let copy_chunks = |dest_all: &mut [f64]| {
3421 use rayon::prelude::*;
3422 dest_all
3423 .par_chunks_mut(chunk_vals)
3424 .enumerate()
3425 .for_each(|(chunk_index, dest)| {
3426 let base = chunk_index * chunk_frames;
3427 for (offset, slot) in dest.chunks_exact_mut(stride).enumerate() {
3428 let index = base + offset;
3429 if index < frames.len() {
3430 write(&frames[index], slot);
3431 }
3432 }
3433 });
3434 };
3435 if rayon::current_thread_index().is_some() {
3439 first_touch_f64(&mut values, workers);
3440 copy_chunks(&mut values);
3441 } else {
3442 first_touch_f64(&mut values, workers);
3443 std::thread::scope(|scope| {
3444 let mut rest = &mut values[..];
3445 let mut base = 0usize;
3446 let mut handles = Vec::new();
3447 while base < frames.len() {
3448 let take_frames = chunk_frames.min(frames.len() - base);
3449 let take_vals = take_frames * stride;
3450 let (dest, tail) = rest.split_at_mut(take_vals);
3451 rest = tail;
3452 let frame_slice = &frames[base..base + take_frames];
3453 handles.push(scope.spawn(move || {
3454 for (frame, slot) in
3455 frame_slice.iter().zip(dest.chunks_exact_mut(stride))
3456 {
3457 write(frame, slot);
3458 }
3459 }));
3460 base += take_frames;
3461 }
3462 for handle in handles {
3463 handle.join().expect("slab worker");
3464 }
3465 });
3466 }
3467 return Some(values);
3468 }
3469 }
3470 #[cfg(not(feature = "parallel"))]
3471 {
3472 let _ = workers;
3473 }
3474 for (frame, dest) in frames.iter().zip(values.chunks_exact_mut(stride)) {
3475 write(frame, dest);
3476 }
3477 Some(values)
3478}
3479
3480fn mix_touch(hash: u64, value: u64) -> u64 {
3481 hash.wrapping_mul(0x9E3779B185EBCA87) ^ value
3482}
3483
3484fn one_frame_touch(frame: &types::ConFrame, coords: &[f64], index: u64) -> u64 {
3485 let mut hash = index;
3486 for value in frame.header.boxl {
3487 hash = mix_touch(hash, value.to_bits());
3488 }
3489 for atom in &frame.atom_data {
3490 hash = mix_touch(hash, atom.symbol.len() as u64);
3491 for byte in atom.symbol.as_bytes() {
3492 hash = mix_touch(hash, u64::from(*byte));
3493 }
3494 }
3495 for value in coords {
3496 hash = mix_touch(hash, value.to_bits());
3497 }
3498 hash
3499}
3500
3501pub fn touch_frame_checksum(
3508 frames: &[types::ConFrame],
3509 positions: &[f64],
3510 workers: usize,
3511) -> Option<u64> {
3512 if frames.is_empty() {
3513 return Some(0);
3514 }
3515 let n_atoms = frames[0].atom_data.len();
3516 let stride = n_atoms.saturating_mul(3);
3517 if stride == 0 || positions.len() != frames.len().saturating_mul(stride) {
3518 return None;
3519 }
3520 if frames.iter().any(|frame| frame.atom_data.len() != n_atoms) {
3521 return None;
3522 }
3523 let one = |index: usize| {
3524 let start = index * stride;
3525 one_frame_touch(
3526 &frames[index],
3527 &positions[start..start + stride],
3528 index as u64,
3529 )
3530 };
3531 Some(reduce_touch(frames.len(), workers, one))
3532}
3533
3534pub fn touch_option_checksum(
3538 frames: &[Option<types::ConFrame>],
3539 positions: &[f64],
3540 workers: usize,
3541) -> Option<u64> {
3542 if frames.is_empty() {
3543 return Some(0);
3544 }
3545 if frames.iter().any(Option::is_none) {
3546 return None;
3547 }
3548 let n_atoms = frames[0].as_ref()?.atom_data.len();
3549 let stride = n_atoms.saturating_mul(3);
3550 if stride == 0 || positions.len() != frames.len().saturating_mul(stride) {
3551 return None;
3552 }
3553 if frames
3554 .iter()
3555 .any(|frame| frame.as_ref().map(|item| item.atom_data.len()) != Some(n_atoms))
3556 {
3557 return None;
3558 }
3559 let one = |index: usize| {
3560 let start = index * stride;
3561 one_frame_touch(
3562 frames[index].as_ref().expect("slot"),
3563 &positions[start..start + stride],
3564 index as u64,
3565 )
3566 };
3567 Some(reduce_touch(frames.len(), workers, one))
3568}
3569
3570fn reduce_touch(n: usize, workers: usize, one: impl Fn(usize) -> u64 + Sync) -> u64 {
3571 #[cfg(feature = "parallel")]
3572 if workers >= 2 {
3573 let fold_range = |first: usize, last: usize| {
3574 let mut acc = 0u64;
3575 for index in first..last {
3576 acc ^= one(index);
3577 }
3578 acc
3579 };
3580 let chunk = n.div_ceil(workers).max(1);
3581 if rayon::current_thread_index().is_some() {
3584 use rayon::prelude::*;
3585 return (0..workers)
3586 .into_par_iter()
3587 .map(|worker| {
3588 let first = worker * chunk;
3589 let last = (first + chunk).min(n);
3590 fold_range(first, last)
3591 })
3592 .reduce(|| 0u64, |left, right| left ^ right);
3593 }
3594 let mut acc = 0u64;
3595 std::thread::scope(|scope| {
3596 let mut handles = Vec::with_capacity(workers);
3597 for worker in 0..workers {
3598 handles.push(scope.spawn(move || {
3599 let first = worker * chunk;
3600 let last = (first + chunk).min(n);
3601 fold_range(first, last)
3602 }));
3603 }
3604 for handle in handles {
3605 acc ^= handle.join().unwrap_or(0);
3606 }
3607 });
3608 return acc;
3609 }
3610 #[cfg(not(feature = "parallel"))]
3611 {
3612 let _ = workers;
3613 }
3614 (0..n).map(one).fold(0, |acc, item| acc ^ item)
3615}
3616
3617pub fn measure_newline_scan(text: &str, threads: usize, chunked: bool) -> (Vec<usize>, f64) {
3625 #[cfg(feature = "parallel")]
3626 {
3627 if chunked && threads >= 2 {
3628 let pool = rayon::ThreadPoolBuilder::new().num_threads(threads).build();
3629 let t_scan = std::time::Instant::now();
3630 let offsets = match pool {
3631 Ok(pool) => pool.install(|| {
3632 let newlines = newline_positions(text.as_bytes(), threads);
3633 frame_start_offsets_indexed(text, newlines)
3634 }),
3635 Err(_) => {
3636 let newlines = newline_positions(text.as_bytes(), threads);
3637 frame_start_offsets_indexed(text, newlines)
3638 }
3639 };
3640 return (offsets, t_scan.elapsed().as_secs_f64());
3641 }
3642 }
3643 #[cfg(not(feature = "parallel"))]
3644 {
3645 let _ = (threads, chunked);
3646 }
3647 let t_scan = std::time::Instant::now();
3648 let offsets = frame_start_offsets_serial(text);
3649 (offsets, t_scan.elapsed().as_secs_f64())
3650}
3651
3652#[cfg(feature = "parallel")]
3655fn collect_parallel_located(
3656 file_contents: &str,
3657 num_threads: Option<usize>,
3658) -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
3659 use rayon::prelude::*;
3660
3661 let boundaries = frame_start_offsets(file_contents);
3662 let parse_chunks = || -> Result<Vec<types::ConFrame>, error::LocatedParseError> {
3663 let num_frames = boundaries.len();
3664 let parts: Vec<Result<types::ConFrame, error::LocatedParseError>> = (0..num_frames)
3665 .into_par_iter()
3666 .map(|i| {
3667 let start = boundaries[i];
3668 let end = if i + 1 < num_frames {
3669 boundaries[i + 1]
3670 } else {
3671 file_contents.len()
3672 };
3673 let chunk = &file_contents[start..end];
3674 let mut iter = ConFrameIterator::new(chunk);
3675 match iter.next() {
3676 Some(Ok(frame)) => Ok(frame),
3677 Some(Err(e)) => {
3678 let local = iter.lines.line_no();
3679 let base = file_line_at(file_contents, start);
3680 let cursor = if local == 0 { base } else { base + local - 1 };
3681 Err(error::LocatedParseError::new(e, cursor, i as u64))
3682 }
3683 None => Err(error::LocatedParseError::new(
3684 error::ParseError::IncompleteFrame,
3685 file_line_at(file_contents, start),
3686 i as u64,
3687 )),
3688 }
3689 })
3690 .collect();
3691 let mut frames = Vec::with_capacity(parts.len());
3692 for part in parts {
3693 frames.push(part?);
3694 }
3695 Ok(frames)
3696 };
3697 match num_threads {
3698 None => parse_chunks(),
3699 Some(n) => {
3700 let n = n.max(1);
3701 match rayon::ThreadPoolBuilder::new().num_threads(n).build() {
3702 Ok(pool) => pool.install(parse_chunks),
3703 Err(_) => parse_chunks(),
3704 }
3705 }
3706 }
3707}
3708
3709#[cfg(all(test, feature = "parallel"))]
3710mod parallel_strong_scale_tests {
3711 use super::*;
3712 use std::path::PathBuf;
3713
3714 fn multi_frame_fixture() -> String {
3715 let p = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/test/tiny_cuh2.con");
3716 let one = std::fs::read_to_string(p).expect("fixture");
3717 one.repeat(8)
3719 }
3720
3721 fn sequential_frames(text: &str) -> Vec<types::ConFrame> {
3722 ConFrameIterator::new(text)
3723 .map(|r| r.expect("seq frame"))
3724 .collect()
3725 }
3726
3727 fn frames_payload_key(f: &types::ConFrame) -> (usize, Vec<(String, f64, f64, f64)>) {
3728 let atoms: Vec<_> = f
3729 .atom_data
3730 .iter()
3731 .map(|a| (a.symbol.to_string(), a.x, a.y, a.z))
3732 .collect();
3733 (f.atom_data.len(), atoms)
3734 }
3735
3736 #[test]
3737 fn parallel_workers_match_sequential_payloads() {
3738 let text = multi_frame_fixture();
3739 let seq = sequential_frames(&text);
3740 assert!(seq.len() >= 8);
3741 let seq_keys: Vec<_> = seq.iter().map(frames_payload_key).collect();
3742
3743 for workers in [1usize, 2, 4] {
3744 let par = parse_frames_parallel_with_threads(&text, Some(workers));
3745 assert_eq!(par.len(), seq.len(), "workers={workers}");
3746 let par_keys: Vec<_> = par
3747 .into_iter()
3748 .map(|r| frames_payload_key(&r.expect("par frame")))
3749 .collect();
3750 assert_eq!(par_keys, seq_keys, "workers={workers} frame payloads");
3751 }
3752
3753 let par_default = parse_frames_parallel(&text);
3755 let def_keys: Vec<_> = par_default
3756 .into_iter()
3757 .map(|r| frames_payload_key(&r.expect("par")))
3758 .collect();
3759 assert_eq!(def_keys, seq_keys);
3760 }
3761
3762 fn live_threads() -> usize {
3763 std::fs::read_to_string("/proc/self/status")
3764 .unwrap()
3765 .lines()
3766 .find(|line| line.starts_with("Threads:"))
3767 .unwrap()
3768 .split_whitespace()
3769 .nth(1)
3770 .unwrap()
3771 .parse()
3772 .unwrap()
3773 }
3774
3775 #[test]
3777 fn default_read_of_a_wide_file_stays_on_the_calling_thread() {
3778 const CHILD: &str = "READCON_DEFAULT_THREAD_CHILD";
3779 const NAME: &str = "iterators::parallel_strong_scale_tests::default_read_of_a_wide_file_stays_on_the_calling_thread";
3780 if std::env::var_os(CHILD).is_none() {
3781 let out = std::process::Command::new(std::env::current_exe().unwrap())
3782 .args([NAME, "--exact", "--test-threads=1"])
3783 .env(CHILD, "1")
3784 .output()
3785 .unwrap();
3786 let stdout = String::from_utf8_lossy(&out.stdout);
3787 assert!(
3788 out.status.success() && stdout.contains("test result: ok. 1 passed;"),
3789 "{stdout}{}",
3790 String::from_utf8_lossy(&out.stderr)
3791 );
3792 return;
3793 }
3794 let dir = tempfile::tempdir().unwrap();
3795 let path = dir.path().join("wide.con");
3796 let one = std::fs::read_to_string(
3797 PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/test/tiny_cuh2.con"),
3798 )
3799 .unwrap();
3800 std::fs::write(&path, one.repeat(200)).unwrap();
3801 assert!(std::fs::metadata(&path).unwrap().len() > 64 * 1024);
3802 let before = live_threads();
3803 let frames = super::read_all_frames(&path).unwrap();
3804 let after = live_threads();
3805 assert_eq!(before, after, "default read started worker threads");
3806 assert!(frames.len() > 1);
3807 let text = std::fs::read_to_string(&path).unwrap();
3808 let sequential = super::frames_from_text(&text, None).unwrap();
3809 assert_eq!(sequential.len(), frames.len());
3810 assert_eq!(
3811 sequential[0].atom_data[0].x, frames[0].atom_data[0].x,
3812 "default read changed a coordinate"
3813 );
3814 }
3815}
3816
3817#[cfg(all(test, feature = "parallel"))]
3818mod parallel_header_scan_tests {
3819 use super::*;
3820 use std::path::PathBuf;
3821
3822 fn fixture(name: &str) -> String {
3823 let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
3824 .join("resources/test")
3825 .join(name);
3826 std::fs::read_to_string(path).expect(name)
3827 }
3828
3829 fn assert_crlf_gap_cut(text: &str) {
3831 let bytes = text.as_bytes();
3832 let mut found = 0usize;
3833 let mut i = 0usize;
3834 while i + 1 < bytes.len() && found < 6 {
3835 if bytes[i] == b'\r' && bytes[i + 1] == b'\n' {
3836 let cut = i + 1;
3837 assert_eq!(bytes[cut - 1], b'\r');
3838 assert_eq!(bytes[cut], b'\n');
3839 assert_cuts_match(text, &[0, cut, bytes.len()]);
3840 found += 1;
3841 i += 2;
3842 continue;
3843 }
3844 i += 1;
3845 }
3846 assert!(found > 0, "fixture has no CRLF pair");
3847 }
3848
3849 fn assert_cuts_match(text: &str, cuts: &[usize]) {
3850 let serial_nl: Vec<usize> = memchr::memchr_iter(b'\n', text.as_bytes()).collect();
3851 let stitched = newline_positions_at_cuts(text.as_bytes(), cuts);
3852 assert_eq!(stitched, serial_nl, "newline stitch at {cuts:?}");
3853 let serial = frame_start_offsets_serial(text);
3854 let parallel = frame_start_offsets_at_cuts(text, cuts);
3855 assert_eq!(parallel, serial, "frame offsets at {cuts:?}");
3856 }
3857
3858 fn header_straddle_cuts(text: &str) -> Vec<Vec<usize>> {
3859 let serial = frame_start_offsets_serial(text);
3860 let bytes = text.as_bytes();
3861 let mut out = Vec::new();
3862 for n in [2usize, 3, 4, 8] {
3864 out.push(equal_cuts(bytes.len(), n));
3865 }
3866 for (index, &start) in serial.iter().enumerate() {
3867 let next = serial.get(index + 1).copied().unwrap_or(bytes.len());
3868 let window = super::header_window_end(bytes, start, next);
3871 let mut line = start;
3872 while line < window {
3873 let Some(rel) = memchr::memchr(b'\n', &bytes[line..window]) else {
3874 break;
3875 };
3876 let end = line + rel;
3877 if end - line > 1 {
3878 let mid = line + (end - line) / 2;
3879 assert!(mid > line && mid < end);
3881 out.push(vec![0, mid, bytes.len()]);
3882 }
3883 line = end + 1;
3884 }
3885 }
3886 out
3887 }
3888
3889 fn assert_slab(text: &str) {
3890 assert!(!text.is_empty());
3891 for cuts in header_straddle_cuts(text) {
3892 assert_cuts_match(text, &cuts);
3893 }
3894 let bare = text.trim_end_matches(['\n', '\r']);
3895 if !bare.is_empty() {
3896 assert_cuts_match(bare, &equal_cuts(bare.len(), 4));
3897 let mid = bare.len() / 2;
3898 let line_start = bare[..mid].rfind('\n').map(|i| i + 1).unwrap_or(0);
3899 let line_end = bare[mid..]
3900 .find('\n')
3901 .map(|i| mid + i)
3902 .unwrap_or(bare.len());
3903 if line_end - line_start > 1 {
3904 let cut = line_start + (line_end - line_start) / 2;
3905 assert_cuts_match(bare, &[0, cut, bare.len()]);
3906 }
3907 }
3908 let tailed = format!("{text}\n \n\t\n");
3909 assert_cuts_match(
3910 &tailed,
3911 &[0, 3, tailed.len() / 2, tailed.len() - 1, tailed.len()],
3912 );
3913 let crlf = text.replace('\n', "\r\n");
3914 assert_cuts_match(&crlf, &equal_cuts(crlf.len(), 5));
3915 assert_crlf_gap_cut(&crlf);
3916 let serial = frame_start_offsets_serial(text);
3917 if serial.len() >= 2 {
3918 let second = serial[1];
3919 let next = serial.get(2).copied().unwrap_or(text.len());
3920 let broken_at = second + (next - second) / 2;
3921 let broken = &text[..broken_at.max(1)];
3922 assert_cuts_match(broken, &[0, 1.max(broken.len() / 3), broken.len()]);
3923 }
3924 }
3925
3926 #[test]
3927 fn chunked_header_scan_matches_serial_offsets() {
3928 assert_eq!(frame_start_offsets_serial(""), frame_start_offsets(""));
3929 assert_cuts_match("\n", &[0, 1]);
3930 assert_cuts_match(" \n\n", &[0, 2, 5]);
3931 for name in [
3932 "tiny_cuh2.con",
3933 "tiny_cuh2.convel",
3934 "tiny_cuh2_forces.con",
3935 "tiny_cuh2_vel_forces.con",
3936 "cuh2.con",
3937 ] {
3938 let one = fixture(name);
3939 assert_slab(&one);
3940 assert_slab(&one.repeat(3));
3941 }
3942 let one = fixture("cuh2.con");
3946 for wide in [one.clone(), one.repeat(8)] {
3947 let serial = frame_start_offsets_serial(&wide);
3948 for workers in [2usize, 3, 4, 8] {
3949 let parallel = frame_start_offsets_with_threads(&wide, Some(workers));
3950 assert_eq!(parallel, serial, "workers={workers} bytes={}", wide.len());
3951 let parsed =
3952 frames_from_bytes(wide.as_bytes(), Some(workers)).expect("bytes parse");
3953 assert_eq!(parsed.len(), serial.len());
3954 let seq = frames_from_text(&wide, Some(1)).expect("serial parse");
3955 assert_eq!(parsed.len(), seq.len());
3956 assert_eq!(parsed[0].atom_data[0].x, seq[0].atom_data[0].x);
3957 assert_eq!(
3958 parsed.last().unwrap().atom_data.last().unwrap().z,
3959 seq.last().unwrap().atom_data.last().unwrap().z
3960 );
3961 }
3962 }
3963 }
3964}
3965
3966#[cfg(all(test, feature = "parallel"))]
3971fn header_window_end(bytes: &[u8], start: usize, limit: usize) -> usize {
3972 let limit = limit.min(bytes.len());
3973 let marker = b"Coordinates";
3974 let mut line = start;
3975 while line < limit {
3976 let rest = &bytes[line..limit];
3977 let rel = memchr::memchr(b'\n', rest).unwrap_or(rest.len());
3978 let content_end = line + rel;
3979 let content = &bytes[line..content_end];
3980 let trimmed = match content.iter().position(|byte| !byte.is_ascii_whitespace()) {
3981 Some(index) => &content[index..],
3982 None => b"",
3983 };
3984 if trimmed.starts_with(marker) {
3985 return line;
3986 }
3987 if content_end >= limit {
3988 break;
3989 }
3990 line = content_end + 1;
3991 }
3992 limit
3993}
3994
3995#[cfg(all(test, feature = "parallel"))]
3996mod parallel_frame_parse_tests {
3997 use super::*;
3998 use std::path::PathBuf;
3999
4000 const ZERO_ATOM: &str = "\
4001zero-atom frame
4002{\"con_spec_version\":3,\"con_spec_minor\":1,\"units\":{\"length\":\"angstrom\",\"energy\":\"eV\"}}
400310.0 10.0 10.0
400490.0 90.0 90.0
4005
4006
40070
4008
4009
4010";
4011
4012 fn fixture(name: &str) -> String {
4013 let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
4014 .join("resources/test")
4015 .join(name);
4016 std::fs::read_to_string(path).expect(name)
4017 }
4018
4019 fn serial_results(text: &str) -> Vec<Result<types::ConFrame, error::ParseError>> {
4020 ConFrameIterator::new(text).collect()
4021 }
4022
4023 fn written(frame: &types::ConFrame) -> Vec<u8> {
4024 let mut out = Vec::new();
4025 {
4026 let mut writer = crate::writer::ConFrameWriter::new(&mut out);
4027 writer.write_frame(frame).expect("write");
4028 writer.flush().expect("flush");
4029 }
4030 out
4031 }
4032
4033 fn assert_same_results(
4034 label: &str,
4035 serial: Vec<Result<types::ConFrame, error::ParseError>>,
4036 parallel: Vec<Result<types::ConFrame, error::ParseError>>,
4037 ) {
4038 assert_eq!(serial.len(), parallel.len(), "{label} count");
4039 for (i, (left, right)) in serial.into_iter().zip(parallel).enumerate() {
4040 match (left, right) {
4041 (Ok(a), Ok(b)) => {
4042 assert_eq!(written(&a), written(&b), "{label} bytes frame {i}");
4043 assert_eq!(a, b, "{label} fields frame {i}");
4044 }
4045 (Err(a), Err(b)) => {
4046 assert_eq!(
4047 format!("{a:?}"),
4048 format!("{b:?}"),
4049 "{label} error frame {i}"
4050 );
4051 }
4052 (Ok(_), Err(err)) => panic!("{label} frame {i} serial ok parallel {err:?}"),
4053 (Err(err), Ok(_)) => panic!("{label} frame {i} serial {err:?} parallel ok"),
4054 }
4055 }
4056 }
4057
4058 fn assert_workers(label: &str, text: &str) {
4059 let serial_ok = serial_results(text).iter().all(|item| item.is_ok());
4060 for workers in [1usize, 2, 3, 4, 8] {
4061 let parallel = parse_frames_parallel_with_threads(text, Some(workers));
4062 assert_same_results(
4063 &format!("{label} workers {workers}"),
4064 serial_results(text),
4065 parallel,
4066 );
4067 let bounds = collect_frame_starts(ConFrameIterator::new(text), true);
4068 let at = parse_frames_at_offsets(text, &bounds, Some(workers));
4069 assert_same_results(
4070 &format!("{label} offsets workers {workers}"),
4071 serial_results(text),
4072 at,
4073 );
4074 }
4075 if serial_ok {
4076 for workers in [2usize, 3, 8] {
4077 let bytes = frames_from_bytes(text.as_bytes(), Some(workers)).expect(label);
4078 let seq = frames_from_text(text, Some(1)).expect(label);
4079 assert_eq!(bytes.len(), seq.len(), "{label} byte count");
4080 for (i, (a, b)) in bytes.iter().zip(seq.iter()).enumerate() {
4081 assert_eq!(written(a), written(b), "{label} byte path frame {i}");
4082 assert_eq!(a, b, "{label} byte path fields frame {i}");
4083 }
4084 }
4085 }
4086 }
4087
4088 fn crlf_gap_offsets(text: &str) -> Vec<usize> {
4089 let bytes = text.as_bytes();
4090 let mut cut = None;
4091 let mut i = 0usize;
4092 while i + 1 < bytes.len() {
4093 if bytes[i] == b'\r' && bytes[i + 1] == b'\n' {
4094 cut = Some(i + 1);
4095 break;
4096 }
4097 i += 1;
4098 }
4099 let cut = cut.expect("crlf");
4100 frame_start_offsets_at_cuts(text, &[0, cut, bytes.len()])
4101 }
4102
4103 #[test]
4104 fn parallel_slots_match_serial_frames() {
4105 assert_workers("empty", "");
4106 assert_workers("blank", "\n");
4107 assert_workers("space", " \n\n");
4108 assert_workers("zero", ZERO_ATOM);
4109 assert_workers("zero-repeat", &ZERO_ATOM.repeat(5));
4110 for name in [
4111 "tiny_cuh2.con",
4112 "tiny_cuh2.convel",
4113 "tiny_cuh2_forces.con",
4114 "tiny_cuh2_vel_forces.con",
4115 "tiny_cuh2_charges_spins_magmoms.con",
4116 "tiny_cuh2_displacements.con",
4117 "tiny_cuh2_spreads.con",
4118 "cuh2.con",
4119 ] {
4120 let one = fixture(name);
4121 assert_workers(name, &one);
4122 assert_workers(&format!("{name} x5"), &one.repeat(5));
4123 let bare = one.trim_end_matches(['\n', '\r']);
4124 assert_workers(&format!("{name} bare"), bare);
4125 let tailed = format!("{one}\n \n\t\n");
4126 assert_workers(&format!("{name} tail"), &tailed);
4127 let crlf = one.replace('\n', "\r\n");
4128 assert_workers(&format!("{name} crlf"), &crlf);
4129 let gap = crlf_gap_offsets(&crlf);
4130 assert_eq!(
4131 gap,
4132 frame_start_offsets_serial(&crlf),
4133 "{name} crlf gap offsets"
4134 );
4135 let parsed = parse_frames_at_offsets(&crlf, &gap, Some(4));
4136 assert_same_results(
4137 &format!("{name} crlf gap parse"),
4138 serial_results(&crlf),
4139 parsed,
4140 );
4141 let serial_offs = frame_start_offsets_serial(&one);
4142 let broken: &str = if serial_offs.len() >= 2 {
4143 let second = serial_offs[1];
4144 let next = serial_offs.get(2).copied().unwrap_or(one.len());
4145 &one[..second + (next - second) / 2]
4146 } else if one.len() > 8 {
4147 &one[..one.len() / 2]
4148 } else {
4149 ""
4150 };
4151 if !broken.is_empty() {
4152 let serial_err = frames_from_text(broken, Some(1))
4153 .err()
4154 .map(|e| format!("{e:?}"));
4155 assert!(serial_err.is_some(), "{name} chopped still parsed");
4156 for workers in [1usize, 2, 4, 8] {
4157 let parallel_err = frames_from_text(broken, Some(workers))
4158 .err()
4159 .map(|e| format!("{e:?}"));
4160 assert_eq!(parallel_err, serial_err, "{name} chopped workers {workers}");
4161 let byte_err = frames_from_bytes(broken.as_bytes(), Some(workers))
4162 .err()
4163 .map(|e| format!("{e:?}"));
4164 assert_eq!(
4165 byte_err, serial_err,
4166 "{name} chopped bytes workers {workers}"
4167 );
4168 }
4169 }
4170 }
4171 }
4172}
4173
4174#[cfg(all(test, feature = "parallel"))]
4175mod phased_read_tests {
4176 use super::*;
4177 use std::path::PathBuf;
4178
4179 fn fixture(name: &str) -> String {
4180 let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
4181 .join("resources/test")
4182 .join(name);
4183 std::fs::read_to_string(path).expect(name)
4184 }
4185
4186 #[test]
4187 fn phased_bytes_match_serial_frames() {
4188 let text = fixture("cuh2.con").repeat(4);
4189 let serial = frames_from_text(&text, Some(1)).expect("serial");
4190 for workers in [1usize, 2, 4, 8] {
4191 let phased = frames_from_bytes_phased(text.as_bytes(), Some(workers)).expect("phased");
4192 let parallel = phased.frames;
4193 assert_eq!(parallel.len(), serial.len(), "workers={workers}");
4194 assert!(
4195 phased.scan_s >= 0.0 && phased.slot_alloc_s >= 0.0 && phased.worker_parse_s >= 0.0
4196 );
4197 assert_eq!(parallel[0], serial[0], "workers={workers} first");
4198 assert_eq!(
4199 parallel.last().unwrap(),
4200 serial.last().unwrap(),
4201 "workers={workers} last"
4202 );
4203 let slab = pack_position_slab(&serial, workers).expect("slab");
4204 let stride = serial[0].positions.nrows() * 3;
4205 let src = serial[0].positions.f64_slice().expect("f64");
4206 assert_eq!(&slab[..stride], src);
4207 let (chunked, _) = measure_newline_scan(&text, workers.max(2), true);
4208 let (plain, _) = measure_newline_scan(&text, 1, false);
4209 if workers >= 2 {
4210 assert_eq!(chunked, plain);
4211 }
4212 }
4213 }
4214
4215 #[test]
4216 fn phased_file_matches_serial_and_maps_a_wide_file() {
4217 let text = fixture("cuh2.con").repeat(5);
4218 let dir = std::env::temp_dir();
4219 let path = dir.join("readcon-phased-wide.con");
4220 std::fs::write(&path, &text).expect("write");
4221 let phased = read_all_frames_phased(&path, Some(4)).expect("phased");
4222 let serial = frames_from_text(&text, Some(1)).expect("serial");
4223 assert_eq!(phased.frames, serial);
4224 assert!(phased.phases.mapped, "a file past 64 KiB is mapped");
4225 assert!(phased.phases.worker_parse_s >= 0.0);
4226 assert_eq!(phased.n_bytes, text.len());
4227 let _ = std::fs::remove_file(&path);
4228 }
4229
4230 #[test]
4231 fn equal_span_boundaries_match_the_serial_walk() {
4232 let text = fixture("cuh2.con").repeat(8);
4233 let bytes = text.as_bytes();
4234 let got = equal_span_boundaries(bytes, 4).expect("equal span");
4235 let expect = collect_frame_starts(ConFrameIterator::from_bytes(bytes), true);
4236 assert_eq!(got, expect);
4237 assert_eq!(got.len(), 8);
4238 let phased = frames_from_bytes_phased(bytes, Some(4)).expect("phased");
4239 assert_eq!(phased.scan_kind, SCAN_EQUAL_SPAN);
4240 assert_eq!(
4241 phased.frames,
4242 frames_from_text(&text, Some(1)).expect("serial")
4243 );
4244 let tailed = format!("{text}\n\n \n");
4245 let tail_phased = frames_from_bytes_phased(tailed.as_bytes(), Some(4)).expect("tail");
4246 assert_eq!(tail_phased.scan_kind, SCAN_EQUAL_SPAN);
4247 assert_eq!(
4248 tail_phased.frames,
4249 frames_from_text(&tailed, Some(1)).expect("tail serial")
4250 );
4251 }
4252
4253 #[test]
4254 fn a_whitespace_tail_uses_the_equal_span_walk() {
4255 let mut text = fixture("cuh2.con").repeat(3);
4256 text.push('\n');
4257 assert!(equal_span_boundaries(text.as_bytes(), 4).is_some());
4258 let serial = frames_from_text(&text, Some(1)).expect("serial");
4259 let parallel = frames_from_bytes_phased(text.as_bytes(), Some(4)).expect("phased");
4260 assert_eq!(parallel.scan_kind, SCAN_EQUAL_SPAN);
4261 assert_eq!(parallel.frames, serial);
4262 }
4263
4264 #[test]
4265 fn touch_checksum_matches_across_worker_counts() {
4266 let text = fixture("cuh2.con").repeat(4);
4267 let serial = frames_from_bytes_phased(text.as_bytes(), Some(1)).expect("serial");
4268 let parallel = frames_from_bytes_phased(text.as_bytes(), Some(4)).expect("parallel");
4269 let slab_s = pack_position_slab(&serial.frames, 1).expect("slab");
4270 let slab_p = pack_position_slab(¶llel.frames, 4).expect("slab");
4271 let left = touch_frame_checksum(&serial.frames, &slab_s, 1).expect("touch");
4272 let right = touch_frame_checksum(¶llel.frames, &slab_p, 4).expect("touch");
4273 assert_eq!(left, right);
4274 assert_ne!(left, 0);
4275 }
4276}
4277
4278#[cfg(test)]
4279mod later_frame_location_tests {
4280 use super::*;
4281 use std::fs;
4282
4283 fn one_atom(coord: &str) -> String {
4284 format!(
4285 "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"
4286 )
4287 }
4288
4289 fn bad_count_frame() -> String {
4290 "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()
4291 }
4292
4293 fn two_frames() -> String {
4294 format!("{}{}", one_atom("0 0 0 0 0"), bad_count_frame())
4295 }
4296
4297 fn assert_later_count(
4298 err: &error::ParseError,
4299 line: u32,
4300 frame: u64,
4301 iter: &ConFrameIterator<'_>,
4302 ) {
4303 assert_eq!(err.class_name(), "InvalidCount", "{err}");
4304 assert_eq!(iter.lines.line_no(), line, "{err}");
4305 assert_eq!(iter.frame_index(), frame, "{err}");
4306 }
4307
4308 fn assert_first_pass(err: Box<dyn std::error::Error>) {
4309 let located = err
4310 .downcast_ref::<error::LocatedParseError>()
4311 .unwrap_or_else(|| panic!("first pass lost the location: {err}"));
4312 assert_eq!(located.error.class_name(), "InvalidCount", "{located:?}");
4313 assert_eq!(located.line, 20, "{located:?}");
4314 assert_eq!(located.frame_index, 1, "{located:?}");
4315 }
4316
4317 fn take_err<T>(result: Result<T, Box<dyn std::error::Error>>) -> Box<dyn std::error::Error> {
4318 match result {
4319 Err(err) => err,
4320 Ok(_) => panic!("expected a parse error"),
4321 }
4322 }
4323
4324 #[test]
4327 fn batch_readers_keep_the_later_frame_on_the_first_pass() {
4328 let path = std::env::temp_dir().join(format!(
4329 "readcon-later-frame-first-{}.con",
4330 std::process::id()
4331 ));
4332 fs::write(&path, two_frames()).unwrap();
4333 for threads in [None, Some(1), Some(2)] {
4334 assert_first_pass(take_err(read_all_frames_with_threads(&path, threads)));
4335 assert_first_pass(take_err(read_all_frames_phased(&path, threads)));
4336 assert_first_pass(take_err(read_positions_with_threads(&path, threads)));
4337 }
4338 let _ = fs::remove_file(&path);
4339 }
4340
4341 #[test]
4343 fn next_forward_and_skip_locate_the_later_frame() {
4344 let text = two_frames();
4345
4346 let mut iter = ConFrameIterator::new(&text);
4347 assert!(iter.next().unwrap().is_ok());
4348 let err = iter.next().unwrap().unwrap_err();
4349 assert_later_count(&err, 20, 1, &iter);
4350
4351 let mut iter = ConFrameIterator::new(&text);
4352 assert!(iter.forward().unwrap().is_ok());
4353 let err = iter.forward().unwrap().unwrap_err();
4354 assert_later_count(&err, 20, 1, &iter);
4355
4356 let mut iter = ConFrameIterator::new(&text);
4357 let err = iter.skip_frames(3).unwrap_err();
4358 assert_later_count(&err, 20, 1, &iter);
4359 }
4360
4361 #[test]
4365 fn skip_after_a_velocity_frame_keeps_the_separator_line() {
4366 let text = format!(
4367 "{}\nH\nVelocities of Component 1\n0 0 0 0 0\n{}",
4368 one_atom("0 0 0 0 0"),
4369 bad_count_frame()
4370 );
4371 let mut iter = ConFrameIterator::new(&text);
4372 assert!(iter.forward().unwrap().is_ok());
4373 let err = iter.forward().unwrap().unwrap_err();
4374 assert_later_count(&err, 24, 1, &iter);
4375 }
4376
4377 #[test]
4378 fn count_frames_locates_the_later_frame() {
4379 let path = std::env::temp_dir().join(format!(
4380 "readcon-later-frame-count-{}.con",
4381 std::process::id()
4382 ));
4383 fs::write(&path, two_frames()).unwrap();
4384 let err = count_frames(&path).unwrap_err();
4385 let _ = fs::remove_file(&path);
4386 match err {
4387 LoadError::Parse(located) => {
4388 assert_eq!(located.error.class_name(), "InvalidCount");
4389 assert_eq!(located.line, 20, "{located}");
4390 assert_eq!(located.frame_index, 1, "{located}");
4391 }
4392 other => panic!("expected a located parse error, got {other}"),
4393 }
4394 }
4395
4396 #[test]
4399 fn metadata_in_a_later_frame_reports_that_frames_line() {
4400 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";
4401 let text = format!("{}{}", one_atom("0 0 0 0 0"), bad);
4402 for threads in [Some(1usize), Some(2)] {
4403 let err = frames_from_text_located(&text, threads).unwrap_err();
4404 assert_eq!(
4405 err.error.class_name(),
4406 "InvalidMetadataJson",
4407 "{threads:?}: {err}"
4408 );
4409 assert_eq!(err.frame_index, 1, "{threads:?}");
4410 assert_eq!(err.line, 14, "{threads:?}: {err}");
4411 }
4412 }
4413
4414 #[test]
4422 fn parallel_positions_report_the_earliest_failing_slot() {
4423 const SLOW_ATOMS: usize = 20_000;
4424 let slow = counted_frame(SLOW_ATOMS);
4425 let bad = counted_frame(1);
4426 let text = format!("{slow}{bad}{bad}{bad}");
4427 let earliest_line = 1 + slow.bytes().filter(|byte| *byte == b'\n').count() as u32;
4428 let path =
4429 std::env::temp_dir().join(format!("readcon-earliest-slot-{}.con", std::process::id()));
4430 fs::write(&path, &text).unwrap();
4431 assert_earliest_slot(
4432 take_err(read_positions_with_threads(&path, Some(1))),
4433 earliest_line,
4434 );
4435 assert_earliest_slot(
4436 take_err(read_positions_with_threads(&path, Some(2))),
4437 earliest_line,
4438 );
4439 let _ = fs::remove_file(&path);
4440 }
4441
4442 fn counted_frame(natoms: usize) -> String {
4443 let mut body = String::with_capacity(natoms * 10 + 96);
4444 body.push_str("seed\n{\"con_spec_version\":2}\n10 10 10\n90 90 90\n0 0\n0 0 0\n1\n");
4445 body.push_str(&natoms.to_string());
4446 body.push_str("\n1.0\nH\nCoordinates of Component 1\n");
4447 for _ in 0..natoms {
4448 body.push_str("0 0 0 0 0\n");
4449 }
4450 body
4451 }
4452
4453 fn assert_earliest_slot(err: Box<dyn std::error::Error>, line: u32) {
4454 let located = err
4455 .downcast_ref::<error::LocatedParseError>()
4456 .unwrap_or_else(|| panic!("position read lost the location: {err}"));
4457 assert_eq!(located.error.class_name(), "InvalidCount", "{located:?}");
4458 assert_eq!(located.frame_index, 1, "{located:?}");
4459 assert_eq!(located.line, line, "{located:?}");
4460 }
4461}
4462
4463#[cfg(test)]
4464mod frame_equality_tests {
4465 use super::ConFrameIterator;
4466
4467 fn sample_con(frames: usize, long: bool) -> String {
4468 let mut out = String::new();
4469 for f in 0..frames {
4470 let shift = (f as f64) * 1.0e-4;
4471 out.push_str("Generated by benchmark\n");
4472 out.push_str("{\"con_spec_version\":2}\n");
4473 out.push_str("15.3456 21.702 100.0\n");
4474 out.push_str("90.0 90.0 90.0\n");
4475 out.push_str("0 0\n");
4476 out.push_str(&format!("{f} 0 1\n"));
4477 out.push_str("2\n6 2\n63.546 1.00793\n");
4478 out.push_str("Cu\nCoordinates of Component 1\n");
4479 for i in 0..6 {
4480 let x = (i as f64) * 2.557 + 0.6394 + shift;
4481 let y = 0.9045 + shift * 0.5;
4482 let z = (f as f64) * 1.0e-6 + if i == 3 { -1.25 } else { 0.0 };
4483 if long {
4484 out.push_str(&format!(" {x:.17} {y:.17} {z:.17} 0 {i}\n"));
4485 } else {
4486 out.push_str(&format!(" {x:.6} {y:.6} {z:.6} 0 {i}\n"));
4487 }
4488 }
4489 out.push_str("H\nCoordinates of Component 2\n");
4490 out.push_str(" 8.6823 9.947 11.733 0 6\n");
4491 out.push_str(&format!(" -7.9421 9.947 {shift:.6} 0 7\n"));
4492 }
4493 out
4494 }
4495
4496 fn sample_convel() -> String {
4497 let mut out = String::new();
4498 for f in 0..3 {
4499 let shift = (f as f64) * 1.0e-4;
4500 out.push_str("Generated by benchmark\n");
4501 out.push_str("{\"con_spec_version\":2}\n");
4502 out.push_str("15.3456 21.702 100.0\n");
4503 out.push_str("90.0 90.0 90.0\n");
4504 out.push_str("0 0\n");
4505 out.push_str(&format!("{f} 0 1\n"));
4506 out.push_str("2\n4 1\n63.546 1.00793\n");
4507 out.push_str("Cu\nCoordinates of Component 1\n");
4508 for i in 0..4 {
4509 let x = (i as f64) * 2.557 + shift;
4510 out.push_str(&format!(" {x:.17} 0.9045 1.0 0 {i}\n"));
4511 }
4512 out.push_str("H\nCoordinates of Component 2\n");
4513 out.push_str(" 1.5 2.5 3.5 0 4\n");
4514 out.push_str("\nCu\nVelocities of Component 1\n");
4515 for i in 0..4 {
4516 let vx = (i as f64) * 1.0e-4 - shift;
4517 out.push_str(&format!(" {vx:.17} -0.5 0.25 0 {i}\n"));
4518 }
4519 out.push_str("H\nVelocities of Component 2\n");
4520 out.push_str(" 0.1 -0.2 0.3 0 4\n");
4521 }
4522 out
4523 }
4524
4525 fn reference_f64(token: &str) -> f64 {
4526 let (val, n) = fast_float2::parse_partial::<f64, _>(token).expect(token);
4527 assert_eq!(n, token.len(), "{token}");
4528 val
4529 }
4530
4531 fn assert_rows_match(text: &str, positions: bool) {
4532 let frames: Vec<_> = ConFrameIterator::new(text)
4533 .map(|frame| frame.expect("parse"))
4534 .collect();
4535 assert!(frames.len() >= 2);
4536 let mut frame_i = 0usize;
4537 let mut row = 0usize;
4538 let mut in_section = false;
4539 for line in text.lines() {
4540 let token = line.trim();
4541 if token.is_empty() {
4542 in_section = true;
4543 row = 0;
4544 continue;
4545 }
4546 if token.starts_with("Generated") {
4547 if row > 0 || in_section {
4548 frame_i += 1;
4549 }
4550 row = 0;
4551 in_section = false;
4552 continue;
4553 }
4554 let cols: Vec<&str> = token.split_whitespace().collect();
4555 if cols.len() < 4 || reference_ok(cols[0]).is_none() {
4556 continue;
4557 }
4558 if cols[3].parse::<u64>().is_err() {
4559 continue;
4560 }
4561 let got = if in_section {
4562 frames[frame_i].atom_data[row].velocity().expect("velocity")
4563 } else if positions {
4564 let p = frames[frame_i].positions.as_f64_row(row);
4565 [p[0], p[1], p[2]]
4566 } else {
4567 continue;
4568 };
4569 for (k, token) in cols.iter().take(3).enumerate() {
4570 assert_eq!(
4571 got[k].to_bits(),
4572 reference_f64(token).to_bits(),
4573 "frame {frame_i} row {row} col {k} token {token} section {in_section}"
4574 );
4575 }
4576 let aos = if in_section {
4577 frames[frame_i].atom_data[row].velocity().unwrap()
4578 } else {
4579 let atom = &frames[frame_i].atom_data[row];
4580 [atom.x, atom.y, atom.z]
4581 };
4582 for k in 0..3 {
4583 assert_eq!(aos[k].to_bits(), got[k].to_bits());
4584 }
4585 row += 1;
4586 }
4587 }
4588
4589 fn reference_ok(token: &str) -> Option<f64> {
4590 let (val, n) = fast_float2::parse_partial::<f64, _>(token).ok()?;
4591 (n == token.len()).then_some(val)
4592 }
4593
4594 #[test]
4595 fn frames_match_general_decimal_parser() {
4596 assert_rows_match(&sample_con(4, true), true);
4597 assert_rows_match(&sample_con(4, false), true);
4598 let text = sample_convel();
4599 let frames: Vec<_> = ConFrameIterator::new(&text)
4600 .map(|frame| frame.expect("convel"))
4601 .collect();
4602 assert_eq!(frames.len(), 3);
4603 assert!(
4604 frames
4605 .iter()
4606 .all(|frame| frame.atom_data[0].velocity().is_some())
4607 );
4608 assert_eq!(frames[0].positions.nrows(), 5);
4609 let mut frame_i = 0usize;
4611 let mut coord_row = 0usize;
4612 let mut vel_row = 0usize;
4613 let mut velocities = false;
4614 for line in text.lines() {
4615 let token = line.trim();
4616 if token.starts_with("Generated") && (coord_row > 0 || velocities) {
4617 frame_i += 1;
4618 coord_row = 0;
4619 vel_row = 0;
4620 velocities = false;
4621 }
4622 if token.starts_with("Velocities") {
4623 velocities = true;
4624 continue;
4625 }
4626 let cols: Vec<&str> = token.split_whitespace().collect();
4627 if cols.len() < 5 {
4628 continue;
4629 }
4630 if reference_ok(cols[0]).is_none() || cols[3].parse::<u64>().is_err() {
4631 continue;
4632 }
4633 let row = if velocities { vel_row } else { coord_row };
4634 let got = if velocities {
4635 frames[frame_i].atom_data[row].velocity().unwrap()
4636 } else {
4637 let p = frames[frame_i].positions.as_f64_row(row);
4638 [p[0], p[1], p[2]]
4639 };
4640 for (k, col) in cols.iter().take(3).enumerate() {
4641 assert_eq!(got[k].to_bits(), reference_f64(col).to_bits(), "{token}");
4642 }
4643 if velocities {
4644 let soa = frames[frame_i].velocities.as_f64_row(row);
4645 for k in 0..3 {
4646 assert_eq!(soa[k].to_bits(), got[k].to_bits());
4647 }
4648 vel_row += 1;
4649 } else {
4650 coord_row += 1;
4651 }
4652 }
4653 assert_eq!(frame_i, 2);
4654 assert_eq!(vel_row, 5);
4655 }
4656}
4657
4658#[cfg(test)]
4659mod newline_skip_tests {
4660 use super::ConFrameIterator;
4661
4662 fn naive(bytes: &[u8], n: usize) -> Result<usize, ()> {
4665 let mut pos = 0usize;
4666 for i in 0..n {
4667 let rest = &bytes[pos..];
4668 match memchr::memchr(b'\n', rest) {
4669 Some(at) => pos += at + 1,
4670 None => {
4671 if rest.is_empty() {
4672 return Err(());
4673 }
4674 pos = bytes.len();
4675 if i + 1 == n {
4676 return Ok(pos);
4677 }
4678 return Err(());
4679 }
4680 }
4681 }
4682 Ok(pos)
4683 }
4684
4685 #[test]
4686 fn counted_skip_matches_per_line_memchr() {
4687 let mut samples = vec![
4688 String::new(),
4689 "a".into(),
4690 "a\n".into(),
4691 "a\nb".into(),
4692 "\n".into(),
4693 "\n\n".into(),
4694 "abc\ndef\nghi".into(),
4695 "a\r\nb\r\n".into(),
4696 "x".repeat(100) + "\n" + &"y".repeat(40),
4697 ];
4698 let path =
4699 std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/test/cuh2.con");
4700 let one = std::fs::read_to_string(&path).unwrap();
4701 samples.push(one.repeat(2));
4702 let bare = one.trim_end_matches(['\n', '\r']).to_string();
4703 samples.push(bare);
4704 for text in samples {
4705 let bytes = text.as_bytes();
4706 for n in [0usize, 1, 2, 3, 6, 7, 9, 40, 200, 450, 1000, 5000] {
4707 let expect = naive(bytes, n);
4708 let mut it = ConFrameIterator::from_bytes(bytes);
4709 let got = it.advance_lines(n).map(|_| it.byte_pos()).map_err(|_| ());
4710 assert_eq!(got, expect, "n={n} len={}", bytes.len());
4711 }
4712 }
4713 }
4714}
4715
4716#[cfg(test)]
4717mod cached_unit_table_read_paths {
4718 use super::*;
4719
4720 const FRAME: &str = "\
4721v3
4722{\"con_spec_version\":3,\"con_spec_minor\":1,\"units\":{\"length\":\"angstrom\",\"energy\":\"eV\"}}
472310 10 10
472490 90 90
4725
4726
47271
47281
47291.0
4730H
4731Coordinates of Component 1
47320.5 0.25 0.125 0 7
4733";
4734
4735 fn coords(frame: &types::ConFrame) -> [f64; 3] {
4736 frame.positions.as_f64_row(0)
4737 }
4738
4739 #[test]
4742 fn read_paths_share_the_cached_unit_table() {
4743 let addr = crate::units::cached_unit_table_addr();
4744 let serial = ConFrameIterator::new(FRAME)
4745 .next()
4746 .expect("frame")
4747 .expect("parse");
4748 let from_bytes = frames_from_bytes(FRAME.as_bytes(), Some(1)).expect("bytes");
4749 let phased = frames_from_bytes_phased(FRAME.as_bytes(), Some(1)).expect("phased");
4750 assert_eq!(from_bytes.len(), 1);
4751 assert_eq!(phased.frames.len(), 1);
4752 assert_eq!(coords(&from_bytes[0]), coords(&serial));
4753 assert_eq!(coords(&phased.frames[0]), coords(&serial));
4754 assert_eq!(crate::units::cached_unit_table_addr(), addr);
4755 let bad = FRAME.replace("angstrom", "not-a-unit");
4756 assert!(ConFrameIterator::new(&bad).next().unwrap().is_err());
4757 assert!(frames_from_bytes(bad.as_bytes(), Some(1)).is_err());
4758 assert!(frames_from_bytes_phased(bad.as_bytes(), Some(1)).is_err());
4759 assert_eq!(crate::units::cached_unit_table_addr(), addr);
4760 }
4761}