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