perf(ifc): rvt-ifc writes the IFC as it builds it, after releasing the file handle (#1075)

changelog.d/changed/1075-stream-step.md

@@ -0,0 +1,10 @@
+- **`rvt-ifc` writes the IFC as it builds it.** The STEP text was built
+  whole in memory and then written. It is now handed to the file every MiB,
+  after the file handle and the partitions it keeps inflated are released.
+  The bytes are the same. The text goes to a file beside the output, renamed
+  when whole, so a failed export leaves no truncated file under its name. On
+  the BIM4LCA office the text is 68 MB and the handle 362 MiB of inflated
+  partitions; the saved-graphics pass, which runs before, still sets the
+  peak there (#1075). With `--observation` the text is still built whole,
+  since the witness reads it back. `rvt::ifc::step_writer::write_step_to`
+  streams to any writer.

probes/examples/probe_1075_memory_stages.rs

@@ -0,0 +1,177 @@
+//! #1075: where `rvt-ifc`'s memory goes, stage by stage.
+//!
+//! FACT under test: the export's peak resident memory (10.3 GB on the
+//! reporter's 438 MB model) is held by one stage. Candidates: the inflated
+//! partitions `RevitFile` keeps for the handle's lifetime, the element walk
+//! and export, the export's diagnostics, the saved-graphics pass, and the
+//! STEP text built whole in memory before it is written.
+//!
+//! The probe runs `rvt-ifc`'s default pipeline in its order (open, export,
+//! diagnostics, saved graphics for the default scope, STEP text) and, around
+//! each stage, reads `/proc/self/status`. Before a stage it writes `5` to
+//! `/proc/self/clear_refs`, which resets the peak (`VmHWM`) to the current
+//! resident size, so each stage's `peak` is its own high-water mark. Linux
+//! only; elsewhere the memory columns read 0.
+//!
+//! Last, after the handle is dropped, `step_stream` writes the same text to a
+//! sink as it is built, as `rvt-ifc` does since #1075.
+//!
+//! Output, one line per stage:
+//!   stage <name> ms <elapsed> rss_kib <after> peak_kib <stage peak> start_kib <before> [<detail>]
+//! then
+//!   cache streams <n> bytes <inflated bytes the handle holds>
+//!   peak_stage <name> peak_kib <largest stage peak>
+//!
+//! `cache` calls `RevitFile::inflated_partition` for every partition and
+//! `Global/Latest`, so a stream the export never read is inflated there: the
+//! `cache` stage's own peak says whether it was. RSS after a drop may not fall,
+//! since the allocator keeps freed pages; the stage peaks are what to read.
+//!
+//! Usage:
+//!   cargo run --profile ci -p rvt-probes --example probe_1075_memory_stages -- MODEL.rvt [--large-model]
+
+use std::time::Instant;
+
+use rvt::RevitFile;
+use rvt::ifc::step_writer::{StepOptions, write_step_to, write_step_with_options};
+use rvt::ifc::{ExportDiagnosticsMode, ExportQualityMode, RvtDocExporter};
+
+fn status_kib(field: &str) -> u64 {
+    std::fs::read_to_string("/proc/self/status")
+        .ok()
+        .and_then(|status| {
+            status.lines().find_map(|line| {
+                line.strip_prefix(field)?
+                    .trim()
+                    .trim_end_matches("kB")
+                    .trim()
+                    .parse()
+                    .ok()
+            })
+        })
+        .unwrap_or(0)
+}
+
+struct Stages {
+    started: Instant,
+    start_kib: u64,
+    peak: (String, u64),
+}
+
+impl Stages {
+    fn begin(&mut self) {
+        let _ = std::fs::write("/proc/self/clear_refs", "5");
+        self.start_kib = status_kib("VmRSS:");
+        self.started = Instant::now();
+    }
+
+    fn end(&mut self, name: &str, detail: &str) {
+        let ms = self.started.elapsed().as_millis();
+        let rss = status_kib("VmRSS:");
+        let peak = status_kib("VmHWM:");
+        println!(
+            "stage {name} ms {ms} rss_kib {rss} peak_kib {peak} start_kib {} {detail}",
+            self.start_kib
+        );
+        if peak > self.peak.1 {
+            self.peak = (name.to_string(), peak);
+        }
+    }
+}
+
+fn main() -> anyhow::Result<()> {
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    let path = args.iter().find(|a| !a.starts_with("--")).ok_or_else(|| {
+        anyhow::anyhow!("usage: probe_1075_memory_stages MODEL.rvt [--large-model]")
+    })?;
+    let large_model = args.iter().any(|a| a == "--large-model");
+
+    let mut stages = Stages {
+        started: Instant::now(),
+        start_kib: 0,
+        peak: (String::new(), 0),
+    };
+    println!("start rss_kib {}", status_kib("VmRSS:"));
+
+    stages.begin();
+    let mut rf = RevitFile::open_with_limits(path, rvt::cli::open_limits(None, None, large_model))?;
+    let release = rf.basic_file_info().map(|b| b.version).unwrap_or(0);
+    stages.end("open", &format!("release {release}"));
+
+    let mode = ExportQualityMode::Scaffold;
+    let mut walker_limits = rvt::walker::WalkerLimits::default();
+    if large_model {
+        walker_limits.max_scan_bytes = rvt::cli::LARGE_MODEL_MAX_INFLATE_BYTES;
+    }
+
+    stages.begin();
+    let mut model = RvtDocExporter.export_with_mode_and_limits(&mut rf, mode, walker_limits)?;
+    stages.end("export", &format!("entities {}", model.entities.len()));
+
+    stages.begin();
+    let diagnostics = rvt::ifc::build_export_diagnostics_with_limits(
+        &mut rf,
+        &model,
+        ExportDiagnosticsMode::Default,
+        walker_limits,
+    );
+    stages.end("diagnostics", "");
+    drop(diagnostics);
+
+    stages.begin();
+    let detail = match rvt::ifc::saved_meshes::default_scope(&mut rf) {
+        Some(scope) => {
+            let control = rvt::control::WalkerControl::default();
+            match rvt::ifc::saved_meshes::attach_scoped_with_control(
+                &mut rf, &mut model, scope, &control,
+            ) {
+                Ok(report) => format!(
+                    "replaced {} of {} triangles {}",
+                    report.replaced, report.candidates, report.triangles
+                ),
+                Err(e) => format!("error {e}"),
+            }
+        }
+        None => "none".to_string(),
+    };
+    stages.end("saved_meshes", &detail);
+
+    stages.begin();
+    let step = write_step_with_options(&model, &StepOptions { timestamp: Some(0) });
+    stages.end("step_text", &format!("bytes {}", step.len()));
+    drop(step);
+
+    stages.begin();
+    let mut names = rf.partition_stream_names();
+    names.push(rvt::streams::GLOBAL_LATEST.to_string());
+    let (mut streams, mut bytes) = (0usize, 0usize);
+    for name in &names {
+        if let Ok(inflated) = rf.inflated_partition(name) {
+            streams += 1;
+            bytes += inflated.bytes().len();
+        }
+    }
+    stages.end("cache", "");
+    println!("cache streams {streams} bytes {bytes}");
+
+    stages.begin();
+    drop(rf);
+    stages.end("drop_file", "");
+
+    // What rvt-ifc does since #1075: the text is written as it is built, after
+    // the file handle is released.
+    stages.begin();
+    let streamed = write_step_to(
+        &model,
+        &StepOptions { timestamp: Some(0) },
+        &mut std::io::sink(),
+    )?;
+    stages.end("step_stream", &format!("bytes {streamed}"));
+
+    stages.begin();
+    drop(model);
+    stages.end("drop_model", "");
+
+    println!("peak_stage {} peak_kib {}", stages.peak.0, stages.peak.1);
+    Ok(())
+}

src/bin/rvt_ifc.rs

@@ -12,7 +12,7 @@ use rvt::{
     ifc::{
         DiagnosticRvtDocExporter, ExportDiagnostics, ExportQualityMode, PlaceholderExporter,
         RvtDocExporter,
-        step_writer::{StepOptions, write_step_with_options},
+        step_writer::{StepOptions, write_step_to, write_step_with_options},
     },
     walker::WalkerLimits,
 };
@@ -328,26 +328,49 @@ fn run(args: Args) -> anyhow::Result<()> {
 
     warn_about_export_quality(&diagnostics);
 
-    let step = write_step_with_options(&model, &StepOptions { timestamp });
-    std::fs::write(&out_path, step.as_bytes())?;
-    eprintln!(
-        "rvt-ifc: wrote {} bytes to {}",
-        step.len(),
-        out_path.display()
-    );
+    let options = StepOptions { timestamp };
+    // The witness observation reads the text back; otherwise it is written as
+    // it is built, after the file handle and the partitions it holds inflated
+    // are released (#1075).
+    let mut observed_step = None;
+    drop(rf);
+    let written = if args.observation.is_some() {
+        let step = write_step_with_options(&model, &options);
+        std::fs::write(&out_path, step.as_bytes())?;
+        let written = step.len() as u64;
+        observed_step = Some(step);
+        written
+    } else {
+        // Written beside the output and renamed when whole, so a failed or
+        // killed export never leaves a truncated IFC under the output's name.
+        let mut partial = out_path.clone().into_os_string();
+        partial.push(".partial");
+        let partial = PathBuf::from(partial);
+        let streamed = std::fs::File::create(&partial)
+            .and_then(|file| write_step_to(&model, &options, file))
+            .and_then(|written| std::fs::rename(&partial, &out_path).map(|()| written));
+        match streamed {
+            Ok(written) => written,
+            Err(error) => {
+                let _ = std::fs::remove_file(&partial);
+                return Err(error.into());
+            }
+        }
+    };
+    eprintln!("rvt-ifc: wrote {written} bytes to {}", out_path.display());
     for line in export_summary(&diagnostics, args.diagnostics.is_some()) {
         eprintln!("{line}");
     }
     if let Some(diagnostics_path) = &args.diagnostics {
         write_diagnostics_sidecar(diagnostics_path, &diagnostics, args.anonymize)?;
     }
-    if let Some(observation_path) = &args.observation {
+    if let (Some(observation_path), Some(step)) = (&args.observation, &observed_step) {
         write_observation(
             observation_path,
             &args.input,
             args.artifact_id.as_deref(),
             &diagnostics,
-            &step,
+            step,
         )?;
     }
     if model.project_name.is_none() {

src/ifc/step_writer.rs

@@ -67,6 +67,31 @@ pub fn write_step_with_options(model: &IfcModel, options: &StepOptions) -> Strin
     w.finish()
 }
 
+/// [`write_step_with_options`], writing the text to `out` as it is built
+/// instead of returning it, so a large model's text is never held whole in
+/// memory (#1075). The bytes are the same. Returns how many were written.
+/// Pass `&mut writer` to keep the writer.
+///
+/// # Errors
+///
+/// The first error `out` returns. What was written before it stays written.
+pub fn write_step_to<W: std::io::Write>(
+    model: &IfcModel,
+    options: &StepOptions,
+    mut out: W,
+) -> std::io::Result<u64> {
+    let now = options.timestamp.unwrap_or_else(unix_seconds);
+    let mut w = StepWriter::new(now);
+    w.sink = Some(Sink {
+        writer: &mut out,
+        written: 0,
+        error: None,
+    });
+    w.emit_header(model);
+    w.emit_data(model);
+    w.finish_to_sink()
+}
+
 /// Attribute layout of the tail an IFC4 building-element instance
 /// carries after `IfcProduct`'s seven common attributes
 /// (`GlobalId`, `OwnerHistory`, `Name`, `Description`, `ObjectType`,
@@ -357,16 +382,30 @@ fn element_attribute_tail(
     }
 }
 
-struct StepWriter {
+struct StepWriter<'a> {
     out: String,
+    /// Where [`write_step_to`] sends the text: `out` is handed on whenever it
+    /// passes [`SPILL_BYTES`].
+    sink: Option<Sink<'a>>,
     next_id: usize,
     /// Unix timestamp (seconds) used for FILE_NAME + IfcOwnerHistory.
     /// Injected at construction so the output is a pure function of
     /// `(model, timestamp)` when a fixed timestamp is supplied.
     timestamp: i64,
 }
 
-impl StepWriter {
+/// A writer [`write_step_to`] streams to, the bytes it took and its first
+/// error.
+struct Sink<'a> {
+    writer: &'a mut dyn std::io::Write,
+    written: u64,
+    error: Option<std::io::Error>,
+}
+
+/// How much text [`StepWriter`] holds before handing it to its sink.
+const SPILL_BYTES: usize = 1 << 20;
+
+impl StepWriter<'_> {
     /// The `Position` and depth, metres, of an opening's extruded solid: the
     /// shared `solid_position` and the body's `depth_m`, or, for a cut with
     /// its own height (RE-93), a placement that far above the element's and
@@ -456,6 +495,7 @@ impl StepWriter {
     fn new(timestamp: i64) -> Self {
         Self {
             out: String::new(),
+            sink: None,
             next_id: 1,
             timestamp,
         }
@@ -470,10 +510,30 @@ impl StepWriter {
     fn emit_line<S: AsRef<str>>(&mut self, line: S) {
         self.out.push_str(line.as_ref());
         self.out.push('\n');
+        if self.out.len() >= SPILL_BYTES {
+            self.spill();
+        }
     }
 
     fn emit_entity<S: AsRef<str>>(&mut self, id: usize, body: S) {
         self.out.push_str(&format!("#{id}={};\n", body.as_ref()));
+        if self.out.len() >= SPILL_BYTES {
+            self.spill();
+        }
+    }
+
+    /// Hand the text held so far to the sink, if there is one; after the
+    /// sink's first error the text is dropped unwritten.
+    fn spill(&mut self) {
+        if let Some(sink) = &mut self.sink {
+            if sink.error.is_none() {
+                match sink.writer.write_all(self.out.as_bytes()) {
+                    Ok(()) => sink.written += self.out.len() as u64,
+                    Err(error) => sink.error = Some(error),
+                }
+            }
+            self.out.clear();
+        }
     }
 
     /// Emit `pset` as one `IfcPropertySingleValue` per property, the
@@ -4057,6 +4117,18 @@ impl StepWriter {
         out.push_str("END-ISO-10303-21;\n");
         out
     }
+
+    fn finish_to_sink(mut self) -> std::io::Result<u64> {
+        self.out.push_str("END-ISO-10303-21;\n");
+        self.spill();
+        match self.sink {
+            Some(Sink {
+                error: Some(error), ..
+            }) => Err(error),
+            Some(Sink { written, .. }) => Ok(written),
+            None => Ok(0),
+        }
+    }
 }
 
 /// STEP-style string escape per ISO-10303-21: