Skip to main content

readcon_core/
iterators.rs

1//=============================================================================
2// The Public API - A clean iterator for users of our library
3//=============================================================================
4
5use crate::parser::{LineStream, parse_declared_sections};
6use crate::{error, types};
7use std::path::Path;
8
9/// memchr-backed line cursor for the full parse path (not only frame skip).
10///
11/// Profile note: `str::Lines` + `Peekable` showed up under
12/// `ConFrameIterator::next` on multi-atom multi-frame workloads. One cursor
13/// serves `next` / `peek` / `forward_fast` so skip and full parse share the
14/// same O(1) newline scan rather than two desynchronized views of the buffer.
15pub struct MemchrLines<'a> {
16    bytes: &'a [u8],
17    pos: usize,
18    peeked: Option<&'a str>,
19    /// When the buffer was not already a `&str`, header lines are checked
20    /// as they are tokenized. Coordinate rows stay on the byte parser.
21    check_utf8: bool,
22    utf8_bad: bool,
23    /// Absolute offsets of every `\n`. Set only for a frame-start scan.
24    /// `None` keeps the per-line `memchr` path used by a full parse.
25    newlines: Option<Vec<usize>>,
26    /// Index of the line that starts at `pos` when `newlines` is in use.
27    index_line: usize,
28    /// 1-based index of the last line produced by [`Self::read_one`], including
29    /// a line rejected as UTF-8. 0 before that.
30    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    /// Cursor over bytes that are not yet known to be UTF-8.
39    ///
40    /// ASCII coordinate rows are accepted by the numeric scan. Text lines
41    /// are validated when they are materialized, so there is no separate
42    /// whole-buffer UTF-8 pass.
43    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    /// 1-based index of the last line read, including a line that is only peeked.
61    #[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                    // Stay on the rejected line. Moving past it makes a
92                    // whitespace tail look like a clean end of file.
93                    // The line number still names this line.
94                    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            // SAFETY: source was `&str`; line is a UTF-8 prefix cut on ASCII `\n`/`\r`.
102            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    /// Drop any peek buffer (required before bulk cursor advances).
129    fn clear_peek(&mut self) {
130        if let Some(p) = self.peeked.take() {
131            // Rewind pos to the start of the peeked line.
132            let start = p.as_ptr() as usize - self.bytes.as_ptr() as usize;
133            self.pos = start;
134            // A peek already counted this line. Rewinding must drop that
135            // count so the next read names it once, and must put the
136            // newline index back on the line `pos` now starts.
137            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    /// One line from the precomputed newline index.
145    ///
146    /// The cursor must sit on a line start (or at the end). The bytes of
147    /// the line are the same slice `memchr` would return, including a line
148    /// whose body crosses a chunk boundary: only the `\n` is indexed.
149    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                    // The rejected line is 1-based `line_no + 1`. The index
176                    // stays on that line so a later read does not skip it.
177                    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            // SAFETY: source was `&str`; the span is a UTF-8 line cut on ASCII `\n`/`\r`.
186            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    /// Keep `err` unless this cursor already rejected a line as UTF-8.
195    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        // The failing row is this line even when the numeric scan returns Err
231        // before the cursor moves.
232        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        // `peek_line` may have rejected a line. That is a decode error,
261        // including when every byte after it is ASCII whitespace.
262        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        // `peek_line` has already moved `pos` to the line after the peek.
271        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            // Leave the rejected bytes on the cursor. `prefer_utf8` then
281            // reports the decode error instead of a missing section.
282            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
321/// A blank line opens a section when the next line is a symbol and the
322/// line after that is a component label (CON-file.5, CON-header.4).
323fn 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
354/// An iterator that lazily parses simulation frames from a `.con` or `.convel`
355/// file's contents.
356///
357/// This struct wraps a memchr line cursor over the file buffer and, upon each
358/// iteration, attempts to parse a complete `ConFrame`. A blank line after the
359/// coordinates is a velocity section only when it begins a complete velocity
360/// block. Otherwise that line is line 1 of the next frame, and it may be empty.
361///
362/// The iterator yields items of type `Result<ConFrame, ParseError>`, allowing for
363/// robust error handling for each frame.
364pub struct ConFrameIterator<'a> {
365    pub(crate) lines: MemchrLines<'a>,
366    /// 0-based index of the frame the cursor is about to read.
367    /// Successful `next` and `forward_fast` advance it. A failure leaves
368    /// it on the frame that failed.
369    frame_index: u64,
370}
371
372/// Newlines seen and the byte index just past the `n`-th `\n`.
373///
374/// When fewer than `n` newlines exist, the count is how many were seen
375/// and the index is `hay.len()`.
376///
377/// # Safety
378/// The CPU supports AVX2.
379#[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        // SAFETY: `offset + 32 <= len`, so the 32-byte load stays inside `hay`.
397        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    /// Creates a new `ConFrameIterator` from a string slice of the entire file.
426    ///
427    /// # Arguments
428    ///
429    /// * `file_contents` - A string slice containing the text of one or more `.con` frames.
430    pub fn new(file_contents: &'a str) -> Self {
431        ConFrameIterator {
432            lines: MemchrLines::new(file_contents),
433            frame_index: 0,
434        }
435    }
436
437    /// Like [`Self::new`], but reading from byte `pos` of `file_contents`.
438    ///
439    /// `pos` must be 0 or follow a `\n`. The iterator keeps no state from
440    /// earlier frames, so skipping from `pos` sees what a scan sees there.
441    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    /// Like [`Self::new`] for a buffer that has not been UTF-8 validated.
449    ///
450    /// Text lines are checked as they are tokenized. Numeric rows reject a
451    /// non-ASCII byte during the fused parse, which is the UTF-8 check for
452    /// those lines.
453    pub fn from_bytes(bytes: &'a [u8]) -> Self {
454        ConFrameIterator {
455            lines: MemchrLines::from_bytes(bytes),
456            frame_index: 0,
457        }
458    }
459
460    /// 0-based index of the frame the cursor is on.
461    ///
462    /// This is the frame a following error belongs to. It is not the
463    /// number of frames still ahead.
464    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    /// Bulk-skips `n` lines from the shared memchr cursor.
474    ///
475    /// A final line with no trailing newline still counts when it is the
476    /// last line this call needs and its bytes are present. The full
477    /// parser accepts that line. Further lines after it are missing.
478    ///
479    /// The header has already named the line count, so the skip counts
480    /// newlines instead of parsing the rows. On AVX2 that count is one
481    /// pass over the span.
482    #[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                    // Count the unterminated line so a later located error
511                    // names it. It satisfies this skip only when it is the
512                    // last line still required; the full parser accepts it.
513                    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    /// [`Self::advance_lines_memchr`] counting newlines in AVX2 chunks.
525    ///
526    /// The landed position and the incomplete-frame condition match the
527    /// per-line search. `n > 1` is the caller's guard: a one-line skip
528    /// is already a single `memchr`.
529    #[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            // The per-line walk counts each skipped line. The vectorized
537            // jump has to keep the same 1-based file line.
538            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    /// [`Self::advance_lines`] for `n` lines expected to end at `end`.
562    ///
563    /// When `end` follows a `\n` and exactly `n` newlines lie before it,
564    /// the per-line search would stop at `end` too, so one vectorized
565    /// count replaces it. Any other case takes the per-line search.
566    fn advance_lines_to(&mut self, n: usize, end: Option<usize>) -> Result<(), error::ParseError> {
567        // The newline index jumps by line count. `end` does not change
568        // where that jump lands; it only shortcuts the per-line search.
569        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            // The per-line walk would count each of these lines. The
583            // vectorized jump has to keep the same 1-based line.
584            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    /// [`Self::advance_lines`] using the stitched newline index.
594    ///
595    /// A missing line and a final line with no `\n` land on the same byte
596    /// as the `memchr` loop. The index is not consulted for a full parse.
597    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    /// One line from the shared cursor (same as full-parse path).
652    fn read_line_str(&mut self) -> Option<&'a str> {
653        self.lines.clear_peek();
654        self.lines.next_line()
655    }
656
657    /// memchr-backed equivalent of [`Self::forward`]. Skips the next
658    /// frame without fully parsing its atom data. Shares the same line
659    /// cursor as [`Iterator::next`], so skip and full parse interleave safely.
660    pub fn forward_fast(&mut self) -> Option<Result<(), error::ParseError>> {
661        self.forward_fast_to(None)
662    }
663
664    /// [`Self::forward_fast`] for a frame expected to end at byte `end`.
665    ///
666    /// The result and the new position are the same as without `end`.
667    /// Line blocks that end exactly at `end` are counted, not searched.
668    #[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        // Lines 1..=6 of the header are skipped wholesale.
678        if let Err(e) = self.advance_lines(6) {
679            return Some(Err(e));
680        }
681        // Line 7: natm_types.
682        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            // Lines 8 and 9 are whitespace, or absent at EOF. Do not scan
691            // for coordinates or a legacy velocity block (CON-count.1).
692            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        // Line 8: natms_per_type.
698        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        // Line 9: masses_per_type, consumed.
711        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        // Optional sections: blank line + same-shape block, repeated.
720        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            // A blank line and then only whitespace is the end of the file,
740            // not a section (CON-frame.5).
741            if after.iter().all(|b| b.is_ascii_whitespace()) {
742                self.lines.pos = self.lines.bytes.len();
743                break;
744            }
745            // A blank that does not begin a section is line 1 of the next
746            // frame, including an empty user line (CON-frame.7, CON-header.4).
747            if !blank_opens_section(after) {
748                break;
749            }
750            // Consume the blank separator and the section block.
751            // The separator is a file line. Skipping it without counting
752            // shifts every later frame's reported line.
753            self.lines.pos += next_eol.map(|p| p + 1).unwrap_or(rest.len());
754            // The separator is a file line. The section block is counted by
755            // the advance, including the vectorized jump.
756            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    /// Skips the next frame without fully parsing its atomic data.
765    ///
766    /// This is more efficient than `next()` if you only need to advance the
767    /// iterator. It reads the frame's header to determine how many lines to skip,
768    /// including any velocity section if present.
769    ///
770    /// # Returns
771    ///
772    /// * `Some(Ok(()))` on a successful skip.
773    /// * `Some(Err(ParseError::...))` if there's an error parsing the header.
774    /// * `None` if the iterator is already at the end.
775    pub fn forward(&mut self) -> Option<Result<(), error::ParseError>> {
776        // Prefer the shared memchr skip path (same cursor as full parse).
777        self.forward_fast()
778    }
779
780    /// Byte offset of the next frame start in the buffer passed to [`Self::new`].
781    pub fn byte_pos(&self) -> usize {
782        self.lines.pos
783    }
784
785    /// Skip `n` frames without parsing atom data. Returns how many were skipped.
786    /// Named `skip_frames` so it does not collide with [`Iterator::skip`].
787    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    /// Like [`Self::skip_frames`]. The error's first field is how many frames
792    /// were skipped before the failure, which is the 0-based index of the
793    /// failing frame when the iterator started at frame 0.
794    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    /// Next frame plus the exact substring of the buffer passed to [`Self::new`].
810    ///
811    /// **Corpus ingest contract:** successive successful spans from the same
812    /// `file_contents` are contiguous (`end` of frame *i* equals `start` of frame
813    /// *i+1*) and, for a buffer that is only multi-frame CON (no prefix garbage),
814    /// concatenating all spans reproduces the trajectory text. Campaign stores
815    /// (`readcon-db`) must persist these spans as authoritative blobs—do not
816    /// re-serialize on the hot ingest path unless the caller supplied in-memory
817    /// [`types::ConFrame`] values without source text.
818    ///
819    /// See also [`crate::index_proj::frame_byte_spans`] and
820    /// [`crate::index_proj::spans_cover_buffer`].
821    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    /// [`Iterator::next`] plus the byte offset where the frame ends.
843    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    /// The type of item yielded by the iterator.
860    ///
861    /// Each item is a `Result` that contains a successfully parsed `ConFrame` or a
862    /// `ParseError` if the frame's data is malformed.
863    type Item = Result<types::ConFrame, error::ParseError>;
864
865    /// Advances the iterator and attempts to parse the next frame.
866    ///
867    /// This method will return `None` only when there are no more lines to consume.
868    /// If there are lines but they do not form a complete frame, it will return
869    /// `Some(Err(ParseError::...))`. Invalid UTF-8 is a validation error, not
870    /// the end of the stream.
871    fn next(&mut self) -> Option<Self::Item> {
872        // A line rejected earlier in this cursor (for example while probing
873        // for a velocity block) must not be reported as a clean EOF.
874        if let Some(err) = self.lines.utf8_error() {
875            return Some(Err(err));
876        }
877        // Trailing whitespace after the last complete frame is not a frame
878        // (CON-frame.1, D2). An empty line that is followed by more text is
879        // line 1 of the next frame (CON-frame.7).
880        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        // Otherwise, attempt to parse the next frame from the available lines.
891        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        // Optional sections mutate AoS; only re-sync section SoA when needed.
896        // Plain .con assembly already filled positions/ids/masses (no O(N)
897        // post-scan when no velocity/force sections were applied).
898        let sections =
899            match parse_declared_sections(&mut self.lines, &mut frame.header, &mut frame.atom_data)
900            {
901                Ok(n) => n,
902                // A section line that failed UTF-8 makes `next_line` return
903                // `None`. That must not surface as `IncompleteSection`.
904                Err(e) => return Some(Err(self.lines.prefer_utf8(e))),
905            };
906        if sections > 0 {
907            // One flat fill per section that was actually parsed. The full
908            // sync walks every absent section and writes each row through
909            // ArcArray uniqueness checks.
910            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    /// After SoA-primary parse, section sync must not require rewriting positions
959    /// (nrows already equals N); forces SoA still filled from AoS.
960    #[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        // Snapshot first position SoA row then re-sync; coords must stay bit-identical
973        // (no needless rewrite would change nothing but we still require agreement).
974        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    /// Section SoA is filled from one flat buffer. Every value must match the
986    /// per-atom tail, including past a small-vector boundary.
987    #[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
1057/// Reads all frames from a file.
1058///
1059/// The bytes are read into one owned buffer and parsed on the calling
1060/// thread. A worker pool runs only when [`read_all_frames_with_threads`]
1061/// is asked for two or more threads.
1062pub 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
1066/// Collect every frame from `text`.
1067///
1068/// `None`, `Some(0)`, and `Some(1)` parse on the calling thread.
1069/// `Some(n)` with `n >= 2` uses a Rayon pool of that size when the
1070/// `parallel` feature is on. Builds without `parallel` always parse
1071/// sequentially.
1072pub 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
1094/// Like [`read_all_frames`], with an explicit worker count.
1095///
1096/// `None`, `Some(0)`, and `Some(1)` parse on the calling thread.
1097/// `Some(n)` with `n >= 2` pins a Rayon pool when `parallel` is on.
1098///
1099/// A parse failure is the [`error::LocatedParseError`] from that parse.
1100/// Its line is the cursor that failed and its frame is the 0-based index.
1101/// The path is not opened again.
1102///
1103/// A plain file of 64 KiB or more stays mapped. Smaller files and
1104/// compressed input are read into one buffer.
1105pub 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/// Positions of every frame, one row-major `f64` buffer.
1114#[derive(Debug)]
1115///
1116/// A parse failure is the [`error::LocatedParseError`] from the boundary
1117/// scan or the span parse that failed. The path is not opened again.
1118///
1119/// The boundary walk records each frame start. Each worker then parses
1120/// only its span and writes that frame's coordinates at
1121/// `frame * n_atoms * 3`. The returned buffer is the only retained
1122/// coordinate storage. A file whose frames do not share one atom count
1123/// is rejected, because the buffer is rectangular.
1124pub struct PositionBlock {
1125    pub n_frames: usize,
1126    pub n_atoms: usize,
1127    pub values: Vec<f64>,
1128}
1129
1130/// Read every frame's coordinates through the column walker.
1131///
1132/// `num_threads` of `None`, `Some(0)`, and `Some(1)` stay on the calling
1133/// thread. `Some(n)` with `n >= 2` uses that many workers when `parallel`
1134/// is enabled. The buffer is rectangular: frames that do not share one
1135/// atom count are an error. Python `read_positions` returns this buffer.
1136pub 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
1150/// Masses and cell that belong with [`read_positions_with_threads`].
1151///
1152/// The column walker fills these when it certifies the file. Otherwise they
1153/// come from the frame parser, which is the same source as before.
1154pub struct CoordColumns {
1155    pub block: PositionBlock,
1156    pub masses: Vec<f64>,
1157    pub cells: Vec<f64>,
1158}
1159
1160/// Positions, masses, and cells for a stock CON or convel file.
1161///
1162/// A certified file is one coordinate pass. Anything else uses the frame
1163/// reader. Python `read_coords` returns these columns.
1164pub 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    // `None`, `Some(0)`, and `Some(1)` stay on the calling thread. A pool
1234    // runs only for an explicit count of two or more.
1235    #[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
1338/// 1-based file line of the line that contains `byte`.
1339fn 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
1344/// File line of a cursor that counted from the start of a share.
1345///
1346/// `local` is 1-based inside the share. A cursor that never moved still
1347/// names the share's first line.
1348fn 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
1382/// Parse one span. The error keeps this frame's file line.
1383fn 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            // The serial walk reports the earliest failing span. A pool
1481            // that stops on the first finished failure can hand back a
1482            // later span of the same file. `find_map_first` keeps the
1483            // earliest slot and still stops spans to its right.
1484            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
1520/// [`frames_from_text`] for bytes that may not be UTF-8 yet.
1521pub 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
1547/// [`frames_from_bytes`] that keeps the line and frame of the first error.
1548fn 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
1566/// Serial byte parse. The cursor is the iterator's file line.
1567fn 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/// First failed slot, with the file line its worker stored.
1586#[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/// Pointer to slot storage shared by the workers that claim chunks.
1618///
1619/// Each claimed index is written by one thread, so the alias is the
1620/// partition. The pointer stays valid until those workers join.
1621#[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    /// # Safety
1628    /// `index` is in range and this call is the only write of that slot.
1629    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// SAFETY: `SlotWritePtr` is copied into the claim loop, and that loop
1637// writes each slot index from exactly one worker. The allocation
1638// outlives the join.
1639#[cfg(feature = "parallel")]
1640unsafe impl Send for SlotWritePtr {}
1641
1642// SAFETY: same partition as `Send`. Workers do not read a slot another
1643// worker is writing.
1644#[cfg(feature = "parallel")]
1645unsafe impl Sync for SlotWritePtr {}
1646
1647/// Minor faults of the calling thread.
1648///
1649/// `RUSAGE_THREAD` is Linux. Other hosts report 0; the counter is
1650/// diagnostic and the parse result does not depend on it.
1651#[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/// Hardware LLC-miss counter for this process.
1668///
1669/// `-1` means the guest refused the event. This KVM guest returns
1670/// `EPERM` for `perf_event_open`, so the parse explanation uses the
1671/// per-worker minor faults instead.
1672#[cfg(feature = "parallel")]
1673fn sample_llc_misses() -> i64 {
1674    // `perf_event_open` is Linux. `-1` is the existing "counter refused" value.
1675    #[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            // PERF_COUNT_HW_CACHE_LL, OP_READ, RESULT_MISS.
1692            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/// Pages that do not sit on node 0.
1719///
1720/// A guest with one memory node has no remote pages. The count is 0
1721/// in that case. A wider node list sums the pages `numa_maps` records
1722/// on every node other than 0.
1723#[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/// Contiguous frame ranges that cover `boundaries` and split the file
1753/// into similar byte lengths.
1754///
1755/// There are up to four ranges per worker, and never more ranges than
1756/// frames. Each range holds at least one frame. The cuts stay on frame
1757/// starts, so a worker parses a whole frame.
1758#[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/// Parse each frame into a slot allocated before any worker runs.
1808///
1809/// Workers claim chunk indexes from an atomic counter and write only
1810/// those slots. The returned vector is that slot storage, in boundary
1811/// order, which is the serial iterator order. `workers < 2` parses on
1812/// the calling thread.
1813#[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/// Slot allocation and the worker parse, timed apart.
1828#[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    /// One entry per worker range, in worker order. This is CPU time.
1836    chunk_s: Vec<f64>,
1837    /// Minor faults on each worker during its range.
1838    worker_minflt: Vec<i64>,
1839    /// Frames in each worker's share.
1840    share_frames: Vec<u64>,
1841    /// File bytes in each worker's share.
1842    share_bytes: Vec<u64>,
1843    /// `(slot, file line)` for each parse that failed. The line is the
1844    /// cursor of that parse, not a later reread.
1845    fail_lines: Vec<(usize, u32)>,
1846}
1847
1848/// Slot allocation and the worker parse, timed apart.
1849///
1850/// `alloc_s` is the slot vector. `work_s` is the parse wall. `sum_s` is
1851/// the sum of per-chunk parse times and `max_s` is the slowest chunk.
1852/// `chunk_s` keeps every chunk. `workers < 2` still reserves the slot
1853/// vector before the calling thread fills it, and that one chunk's time
1854/// is the sum, the max, and the only entry.
1855#[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    // Several byte-balanced ranges per worker. A slow range can be
1909    // stolen. Each range is still one iterator, not one iterator per
1910    // frame: a fresh iterator per frame faulted a private heap.
1911    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                // SAFETY: this range owns `start..end`, so `i` has one
1946                // writer. `slots` outlives the join.
1947                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    // Each index was written once. Reinterpreting the buffer avoids a
2004    // second serial walk of the frame shells after the workers join.
2005    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    /// Still alive so later phases of this read use the same workers.
2058    /// `None` when the call stayed on the calling thread.
2059    pool: Option<std::sync::Arc<rayon::ThreadPool>>,
2060    pool_s: f64,
2061    /// Zero while `pool` is still alive. The owner times the join.
2062    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    /// `(slot, file line)` recorded by the worker that failed.
2074    fail_lines: Vec<(usize, u32)>,
2075    llc_misses: i64,
2076    numa_remote: i64,
2077    /// [`SCAN_SERIAL`], [`SCAN_INDEXED`], or [`SCAN_EQUAL_SPAN`].
2078    scan_kind: u8,
2079}
2080
2081/// Keep freed parse pages in the heap.
2082///
2083/// Eight workers otherwise return those pages to the kernel, and the
2084/// next file faults them again. One worker already keeps a smaller set.
2085/// The same policy on every width leaves the parse fault count flat.
2086#[cfg(feature = "parallel")]
2087fn keep_parse_pages() {
2088    // glibc `M_MMAP_MAX` and `M_TRIM_THRESHOLD`. macOS and Windows have
2089    // no `mallopt`; leaving the pages to the system allocator is enough
2090    // for the parse to be correct.
2091    #[cfg(target_os = "linux")]
2092    unsafe {
2093        libc::mallopt(-4, 0);
2094        libc::mallopt(-1, i32::MAX);
2095    }
2096}
2097
2098/// Slot storage kept between parses.
2099///
2100/// Eight workers write one `ConFrame` shell per frame into this buffer.
2101/// A fresh buffer is a new mapping, and those writes are the repeating
2102/// minor faults. The next parse reuses the same pages.
2103#[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/// One pool per worker count, kept for the life of the process.
2129///
2130/// A fresh pool on every file made each worker fault its frames again.
2131/// Reusing the workers lets the allocator hand back the pages just freed.
2132#[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/// Minor faults of this process.
2154///
2155/// `getrusage` is Unix. Windows reports 0.
2156#[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/// Page size from `sysconf`, with the POSIX page as the fallback.
2173#[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/// Fault one word on each page of `values`.
2187///
2188/// The page count is split by the worker count. The step is the host
2189/// page, so the slice a worker faults is its own. No extra chunk size.
2190#[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        // One chunk per page. The installed pool hands each page to one
2202        // worker, so that worker faults its own slice.
2203        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/// Move every `Ok` frame on the read's pool.
2233///
2234/// A serial walk of the slot vector grows with the frame count. The
2235/// pool that parsed the frames does this move. One thread still walks
2236/// the slots when that pool is absent.
2237#[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/// Header walk on the calling thread.
2266#[cfg(feature = "parallel")]
2267pub const SCAN_SERIAL: u8 = 0;
2268/// Newline index in chunks, then a header walk on the calling thread.
2269///
2270/// Frames that do not share one byte length take this path. The newline
2271/// bytes are split across the pool. The header walk is not, because a
2272/// later frame start is the end of the skip, and that end is not known
2273/// until the skip runs.
2274#[cfg(feature = "parallel")]
2275pub const SCAN_INDEXED: u8 = 1;
2276/// Equal-length frames. Each worker boundary is checked.
2277#[cfg(feature = "parallel")]
2278pub const SCAN_EQUAL_SPAN: u8 = 2;
2279
2280/// Frame starts when every frame occupies the same number of bytes.
2281///
2282/// The first frame sets the span. Each worker boundary is skipped on
2283/// its own slice. A span that does not tile the
2284/// buffer, or a skip that does not land on the next span, returns
2285/// `None` and the caller keeps the serial walk. The starts are
2286/// `0, span, 2*span, ...` and match [`collect_frame_starts`].
2287#[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    // Bytes after the last full span are a tail, not another frame.
2307    // The serial walk ignores a whitespace tail.
2308    let tiled = n * span;
2309    if bytes[tiled..].iter().any(|b| !b.is_ascii_whitespace()) {
2310        return None;
2311    }
2312    // Equal byte ranges. A cut that falls inside a frame resyncs to the
2313    // next frame start, and only that header is checked. The checks stay
2314    // on this worker: a nested pool wake was larger than the checks.
2315    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/// Starts `0, span, 2*span, ...` from a buffer kept between files.
2341///
2342/// Building that buffer from a fresh allocation stalled for milliseconds
2343/// after a large free. The next file fills the same pages.
2344#[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/// Scan, slot allocation, and worker parse of `bytes`.
2381///
2382/// `Some(n)` with `n >= 2` uses a pool of that size. Every other count
2383/// stays on the calling thread. `None` follows [`scan_thread_count`].
2384/// The pool build is `pool_s`. The pool join is `pool_join_s`.
2385/// Neither one is inside the scan or the parse.
2386#[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        // Equal-length frames skip the serial header walk. Anything else
2396        // still walks headers on this thread. Workers then parse the
2397        // spans in file order.
2398        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                    // The same workers serve later files of this width.
2446                    // Their freed blocks stay warm for the next parse.
2447                    (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
2476/// Count frames without building atom payloads (uses [`ConFrameIterator::forward_fast`]
2477/// when possible, else [`ConFrameIterator::forward`]).
2478///
2479/// Prefer this over `read_all_frames(...).len()` when only the frame count is needed.
2480/// A rejected frame is [`LoadError::Parse`] with that frame's file line and
2481/// 0-based index, including when the failure is after the first frame.
2482pub 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
2503/// Reads only the first frame from a file.
2504///
2505/// More efficient than `read_all_frames` for single-frame access because it
2506/// stops parsing after the first frame rather than collecting all of them.
2507pub 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
2518/// Skip `index` frames (no atom parse), then parse the next one.
2519///
2520/// `index == 0` is [`read_first_frame`]. Out of range is an error.
2521pub 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
2530/// Same as [`read_nth_frame`] on an already-loaded buffer.
2531pub 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/// Reads all frames using the checked `.rcidx` sidecar next to `path`.
2548///
2549/// Uncompressed input uses (and refreshes) `path.rcidx`; compressed
2550/// input is scanned in memory each time. UTF-8 is validated in
2551/// newline-aligned slices.
2552#[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
2588/// Lines in `bytes`, counting a final line that has no `\n`.
2589fn 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
2609/// Line index of the line that contains `pos`, or the line count at `len`.
2610fn 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/// `n` cuts covering `0..=len`. A cut may fall in the middle of a header line.
2640#[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/// Newline offsets in `bytes`.
2658///
2659/// `cuts` splits the buffer into `[cuts[i], cuts[i + 1])`. Each piece is
2660/// scanned on its own and the offsets are concatenated. A `\n` belongs to
2661/// the piece that contains that byte, so a header line whose body crosses
2662/// a cut still contributes one newline.
2663#[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    // The current Rayon pool, or the global pool when the caller has not
2677    // installed one. Chunk count is the worker cap: two chunks do not
2678    // wake every core.
2679    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/// Worker count for a newline scan.
2701///
2702/// `Some(n)` with `n >= 2` uses that many workers at every file length.
2703/// A file shorter than two bytes has nothing to split, and [`equal_cuts`]
2704/// drops a cut that repeats the previous byte, so a short file does not
2705/// invent an empty chunk. `None`, `Some(0)`, and `Some(1)` stay on the
2706/// calling thread. An omitted count must not start the global pool:
2707/// that pool made a one-frame file slower. There is no byte-length
2708/// cutoff taken from one host.
2709#[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/// I/O, UTF-8, or a located parse failure while loading a path.
2719#[derive(Debug)]
2720pub enum LoadError {
2721    Io(std::io::Error),
2722    Utf8(std::str::Utf8Error),
2723    Parse(error::LocatedParseError),
2724    /// The path opened and decoded, and it held no frame.
2725    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/// 1-based line number of the line that contains `byte`.
2753#[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
2762/// Collect every frame, keeping the line and frame index of the first error.
2763pub 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        // A caller that does not ask for a pool stays on this thread.
2770        // Starting the global pool for a long file made a one-frame read slower.
2771        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
2802/// Read every frame from `path`, with the line and frame index of the first error.
2803pub 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
2812/// Read the first frame from `path`, with a line number on failure.
2813pub 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
2828/// Skip `index` frames, then parse one, with a line number on failure.
2829pub 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            // One AVX2 entry so the line skip stays in this function.
2870            // A separate call per block pays for an AVX/SSE transition.
2871            // SAFETY: the runtime check above is the AVX2 and POPCNT contract.
2872            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
2914/// Serial header walk. This is the reference the chunked scan must match.
2915pub 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/// Frame starts for `text` when the newline scan is split at `cuts`.
2933///
2934/// `cuts` is the chunk edges, including 0 and `text.len()`. A cut inside a
2935/// header line is the straddling case: the stitched offsets are still the
2936/// serial offsets.
2937#[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
2943/// Frame start byte offsets via [`ConFrameIterator::forward_fast`] (no atom parse).
2944///
2945/// Stops at the first skip error (that start is still included).
2946///
2947/// With the `parallel` feature and more than one worker, the buffer is
2948/// split into chunks and the newline bytes are scanned together. The
2949/// length of the buffer does not choose the worker count. The chunks
2950/// are not aligned to lines. A header
2951/// that straddles a cut is one line, because each `\n` is recorded by the
2952/// chunk that holds that byte and the lists are stitched in order. The
2953/// returned offsets match the serial walk.
2954pub fn frame_start_offsets(file_contents: &str) -> Vec<usize> {
2955    frame_start_offsets_with_threads(file_contents, None)
2956}
2957
2958/// [`frame_start_offsets`] using `num_threads` chunks.
2959///
2960/// `None` uses the machine's parallelism. `Some(0)` and `Some(1)` stay on
2961/// the calling thread. The offsets do not depend on the thread count.
2962///
2963/// The timed work includes the cut setup, the per-chunk scans, the
2964/// stitch, and the frame walk that reads the stitched newline index.
2965pub 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/// Parses frames in parallel using rayon, splitting on frame boundaries.
2985///
2986/// Phase 1: frame starts from a serial [`ConFrameIterator::forward_fast`]
2987/// walk. The header names each block's line count, so the walk does not
2988/// parse atom rows. Phase 2 parses those spans on a Rayon pool and
2989/// stores the frames in file order. See [`parse_frames_parallel_with_threads`].
2990///
2991/// Requires the `parallel` feature.
2992#[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/// Like [`parse_frames_parallel`], but `Some(n)` with `n >= 2` parses on
3000/// a Rayon pool of that size. `None`, `Some(0)`, and `Some(1)` stay on
3001/// the calling thread.
3002///
3003/// The boundary walk is serial. The parse ranges are byte-balanced and
3004/// the frames come back in file order, field for field with a serial
3005/// iterator.
3006///
3007/// Requires the `parallel` feature.
3008#[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        // A whitespace tail is not a frame. The serial iterator stops
3019        // there. The offset scan used by callers still records that byte
3020        // when a skip error includes it.
3021        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/// Phase 2 of [`parse_frames_parallel_with_threads`]: parse the frames
3040/// that start at `boundaries` (strictly increasing byte offsets into
3041/// `file_contents`; frame `i` ends where frame `i + 1` starts).
3042///
3043/// Offsets may come from [`frame_start_offsets`] or from a checked
3044/// [`crate::frame_index::FrameIndex`]. The output slots are allocated
3045/// first. Each worker then fills one contiguous index range. Results
3046/// stay in frame order.
3047///
3048/// The timer for this function starts after `boundaries` exist. It
3049/// includes slot allocation and the worker parse. It does not include
3050/// the newline scan or the stitch.
3051///
3052/// Requires the `parallel` feature.
3053#[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/// Seconds for one phased read. The open time is the file read or the
3083/// map. Scan, slot allocation, and the worker parse are separate.
3084#[derive(Clone, Copy, Debug, Default)]
3085pub struct ReadPhaseTimes {
3086    pub open_mmap_s: f64,
3087    /// Rayon pool construction. Zero when the call stays on one thread.
3088    pub pool_s: f64,
3089    /// Joining the local pool after the scan and the parse. Zero when
3090    /// the call stays on one thread.
3091    pub pool_join_s: f64,
3092    pub scan_s: f64,
3093    pub slot_alloc_s: f64,
3094    pub worker_parse_s: f64,
3095    /// Moving each parsed `Result` into the frame vector.
3096    pub finish_s: f64,
3097    /// Sum of per-chunk parse times. This is CPU time, not wall time.
3098    pub worker_sum_s: f64,
3099    /// Slowest parse chunk. This is inside `worker_parse_s`.
3100    pub worker_max_s: f64,
3101    /// Drop of the mapped or owned file buffer after the parse.
3102    pub close_s: f64,
3103    /// Wake of the worker pool and the other gaps inside the read.
3104    pub schedule_s: f64,
3105    /// Minor faults during the scan.
3106    pub scan_minflt: i64,
3107    /// Sum of per-worker minor faults during the parse.
3108    pub parse_minflt: i64,
3109    /// LLC read misses, or `-1` when the guest refuses the counter.
3110    pub llc_misses: i64,
3111    /// Pages on a node other than node 0. Zero on a one-node guest.
3112    pub numa_remote: i64,
3113    /// Minor faults while the coordinate slab is filled.
3114    pub slab_minflt: i64,
3115    /// Minor faults during the touch checksum.
3116    pub touch_minflt: i64,
3117    /// True when the bytes came from a map rather than an owned buffer.
3118    pub mapped: bool,
3119}
3120
3121/// Frames from one phased read, plus the byte length of the buffer.
3122pub struct PhasedFrames {
3123    pub frames: Vec<types::ConFrame>,
3124    pub phases: ReadPhaseTimes,
3125    pub n_bytes: usize,
3126    /// Per-chunk parse seconds, in claim order.
3127    pub chunk_s: Vec<f64>,
3128    /// [`SCAN_SERIAL`], [`SCAN_INDEXED`], or [`SCAN_EQUAL_SPAN`] when
3129    /// `parallel` is on. Zero otherwise.
3130    pub scan_kind: u8,
3131    /// Minor faults on each parse worker.
3132    pub worker_minflt: Vec<i64>,
3133    /// Frames in each worker share.
3134    pub share_frames: Vec<u64>,
3135    /// File bytes in each worker share.
3136    pub share_bytes: Vec<u64>,
3137    /// Pool kept for the slab, the touch, and frame teardown.
3138    #[cfg(feature = "parallel")]
3139    pub pool: Option<std::sync::Arc<rayon::ThreadPool>>,
3140}
3141
3142/// Read `path` and time open, scan, slot allocation, and the worker parse.
3143///
3144/// Plain text at or above the map threshold is mapped. Smaller files and
3145/// compressed files are owned buffers. `Some(n)` with `n >= 2` parses on
3146/// that many workers. Every other count parses on the calling thread.
3147/// The frame list matches [`frames_from_bytes`].
3148///
3149/// A parse failure is the [`error::LocatedParseError`] from the parse
3150/// that failed. The path is not opened again.
3151pub 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        // The serial unmap of a fully resident file is the multi-worker
3176        // tail. Advise the pages away on the read's pool first.
3177        // One worker unmaps directly, so a serial read does not walk
3178        // the pages twice.
3179        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
3239/// [`frames_from_bytes`] with scan, slot allocation, and worker-parse times.
3240///
3241/// `Some(n)` with `n >= 2` uses that pool. Every other count, including
3242/// `None`, stays on the calling thread. That is the `read_con` policy:
3243/// an omitted count does not start a pool.
3244pub 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    /// Per-chunk parse seconds, in claim order.
3255    pub chunk_s: Vec<f64>,
3256    /// [`SCAN_SERIAL`], [`SCAN_INDEXED`], or [`SCAN_EQUAL_SPAN`] when
3257    /// `parallel` is on. Zero otherwise.
3258    pub scan_kind: u8,
3259    /// Minor faults during the scan.
3260    pub scan_minflt: i64,
3261    /// Minor faults on each parse worker, in worker order.
3262    pub worker_minflt: Vec<i64>,
3263    pub share_frames: Vec<u64>,
3264    pub share_bytes: Vec<u64>,
3265    /// LLC read misses, or `-1` when the guest refuses the counter.
3266    pub llc_misses: i64,
3267    /// Pages on a node other than node 0.
3268    pub numa_remote: i64,
3269    /// Pool kept for the slab, the touch, and frame teardown.
3270    #[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        // One thread walks the file once. A scan plus a second parse made
3288        // a short file slower than the plain iterator.
3289        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
3376/// Copy every frame's coordinates into one row-major buffer.
3377///
3378/// `None` means the frames do not share one atom count. `workers >= 2`
3379/// writes disjoint frame runs. The copy is bit-identical to reading
3380/// each frame's `f64` positions in order.
3381pub 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        // The parallel writers are the first touch of these pages.
3397        // A zeroed allocation would fault the whole slab on this thread.
3398        // SAFETY: every index is overwritten before the buffer is read.
3399        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            // `par_chunks_mut` follows the installed pool. Outside one,
3436            // a private scope of `workers` threads keeps the global pool
3437            // out of this phase.
3438            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
3501/// XOR of positions, cell lengths, and atom-type symbols.
3502///
3503/// `positions` is frame-major `[frames, atoms, 3]`. `None` means the
3504/// slice does not match the frames. `workers >= 2` folds disjoint
3505/// frame runs on that many threads. Each frame's hash includes its
3506/// index, and the returned word is the XOR of those hashes.
3507pub 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
3534/// [`touch_frame_checksum`] for frames still sitting in optional slots.
3535///
3536/// `None` means a slot is empty or the position slice does not match.
3537pub 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        // Reuse the installed read pool. A fresh pool per touch is the
3582        // build this phase used to pay on every call.
3583        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
3617/// Time a newline scan of `text`.
3618///
3619/// `chunked` with `threads >= 2` splits the buffer into that many
3620/// chunks, stitches the newline index, and walks frame starts. The
3621/// pool is built before the timer starts. The serial walk is
3622/// [`frame_start_offsets_serial`]. The returned offsets are the frame
3623/// starts. The caller chooses the worker count. File length does not.
3624pub 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/// Parallel parse that keeps the file line of the first error.
3653/// Workers return the error. They do not write thread-local error state.
3654#[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        // Enough frames for >1 worker to exercise the pool.
3718        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        // Global pool path agrees too.
3754        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    /// A file past the old map and pool gates stays on this thread.
3776    #[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    /// A cut on the byte between `\r` and `\n`. The `\n` is in the next chunk.
3830    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        // Equal splits, plus a cut forced into a header line of every frame.
3863        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            // The header ends at the coordinate label. The window is that
3869            // header, not a fixed count of bytes past the frame start.
3870            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                    // mid is inside the line, so the header byte sits across two chunks.
3880                    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        // The bytes parser (the read_con path) uses the same stitch.
3943        // The single frame is the short file. The repeat is a wider file.
3944        // Both use the requested worker count. There is no length cutoff.
3945        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/// End of the header window used by straddle cuts.
3967///
3968/// The header runs up to the line that starts the coordinate block.
3969/// When that label is absent the window is the next frame start.
3970#[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(&parallel.frames, 4).expect("slab");
4271        let left = touch_frame_checksum(&serial.frames, &slab_s, 1).expect("touch");
4272        let right = touch_frame_checksum(&parallel.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    /// The batch readers report the later frame from the parse that failed.
4325    /// A second open of the path is not required.
4326    #[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    /// An error in the second frame keeps that frame's index and file line.
4342    #[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    /// The blank line that opens a velocity block is a file line. A later
4362    /// frame's error must include it. 12 coordinate lines, the separator,
4363    /// and 3 velocity lines put the next frame's line 8 at file line 24.
4364    #[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    /// Metadata on line 2 of the second frame is file line 14, not line 2.
4397    /// This is the serial contract the collective reader matches.
4398    #[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    /// Several spans after the first frame fail. Serial collection keeps
4415    /// the earliest slot. The parallel position reader must report that
4416    /// same line and frame index.
4417    ///
4418    /// The first frame is large. A later worker can fail before the
4419    /// calling thread reaches the earliest bad span, which is the case
4420    /// that used to return the later slot.
4421    #[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        // Every stored coordinate and velocity bit matches the general parser.
4610        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    /// The per-line search [`ConFrameIterator::advance_lines`] used before
4663    /// the counted skip. Positions and errors are the oracle.
4664    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    /// The serial iterator, the byte reader behind the C API, and the phased
4740    /// reader behind `read_con` all validate units through the one table.
4741    #[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}