diff --git a/changelog.d/changed/1075-stream-step.md b/changelog.d/changed/1075-stream-step.md new file mode 100644 index 000000000..20112a896 --- /dev/null +++ b/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. diff --git a/probes/examples/probe_1075_memory_stages.rs b/probes/examples/probe_1075_memory_stages.rs new file mode 100644 index 000000000..4ded28aef --- /dev/null +++ b/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 ms rss_kib peak_kib start_kib [] +//! then +//! cache streams bytes +//! peak_stage peak_kib +//! +//! `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 = 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(()) +} diff --git a/src/bin/rvt_ifc.rs b/src/bin/rvt_ifc.rs index 681439400..8ae6eed3c 100644 --- a/src/bin/rvt_ifc.rs +++ b/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() { diff --git a/src/ifc/step_writer.rs b/src/ifc/step_writer.rs index 3a5b8bec3..622935f3f 100644 --- a/src/ifc/step_writer.rs +++ b/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( + model: &IfcModel, + options: &StepOptions, + mut out: W, +) -> std::io::Result { + 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,8 +382,11 @@ 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>, next_id: usize, /// Unix timestamp (seconds) used for FILE_NAME + IfcOwnerHistory. /// Injected at construction so the output is a pure function of @@ -366,7 +394,18 @@ struct StepWriter { 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, +} + +/// 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>(&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>(&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 { + 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: