`ba308a5`

Parse object inputs in parallel

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 1 week ago

SHA: ba308a5c41e51374d36db5171da936c476189d4e
Parents: 01ed08d
Tree: aac1dfb

5 changed files

Status	File	+	-
M	`src/lib.rs`	184	10
M	`src/resolve.rs`	34	11
M	`tests/atom_integration.rs`	3	9
M	`tests/determinism.rs`	32	10
M	`tests/perf_baseline.rs`	11	9

src/lib.rsmodified

  use std::os::unix::fs::PermissionsExt;
  use std::path::PathBuf;
 +use std::sync::{mpsc, Arc, Mutex};
 +use std::thread;
  use std::time::{Duration, Instant};
 -use std::{fs, io};
 +use std::{collections::VecDeque, fs, io};
 +use archive::Archive;
  use atom::{atomize_object, backpatch_symbol_atoms, AtomTable};
  use icf::IcfError;
 +use input::ObjectFile;
  use layout::{ExtraLayoutSections, Layout, LayoutInput};
  use macho::dylib::{DylibDependency, DylibFile, DylibLoadKind};
  use macho::reader::ReadError;
          let mut inputs = Inputs::new();
          let mut deferred_dylibs = Vec::new();
 +        let mut initial_loads = Vec::new();
          let phase_started = Instant::now();
          for (load_order, path) in load_paths.iter().enumerate() {
              if matches!(
              if opts.trace_inputs {
                  eprintln!("afs-ld: loading {}", path.display());
+             }
 -            let timings = register_input(&mut inputs, path, load_order, true)?;
 +            initial_loads.push((load_order, path.clone()));
 +        }
 +        for loaded in load_initial_inputs(initial_loads)? {
 +            let timings = register_loaded_initial_input(&mut inputs, loaded);
              phases.add_input_load(timings);
+         }
          let include_tbd_exports = inputs_may_need_dylib_exports(&inputs)?;
          for idx in 0..inputs.objects.len() {
              let input_id = resolve::InputId(idx as u32);
              let obj = inputs.object_file(input_id)?;
 -            let atomization = atomize_object(input_id, &obj, &mut atom_table);
 -            backpatch_symbol_atoms(
 -                &atomization,
 -                input_id,
 -                &obj,
 -                &mut sym_table,
 -                &mut atom_table,
 -            );
 +            let atomization = atomize_object(input_id, obj, &mut atom_table);
 +            backpatch_symbol_atoms(&atomization, input_id, obj, &mut sym_table, &mut atom_table);
              objects.push((input_id, obj));
+         }
          phases.atomization = phase_started.elapsed();
          .unwrap_or_else(|| PathBuf::from("a.out"))
+ }
 +struct LoadedObjectInput {
 +    path: PathBuf,
 +    load_order: usize,
 +    bytes: Vec<u8>,
 +    parsed: ObjectFile,
 +    timings: InputLoadTimings,
 +}
++
 +struct LoadedArchiveInput {
 +    path: PathBuf,
 +    load_order: usize,
 +    bytes: Vec<u8>,
 +    timings: InputLoadTimings,
 +}
++
 +enum LoadedInitialInput {
 +    Object(Box<LoadedObjectInput>),
 +    Archive(LoadedArchiveInput),
 +}
++
 +impl LoadedInitialInput {
 +    fn load_order(&self) -> usize {
 +        match self {
 +            LoadedInitialInput::Object(input) => input.load_order,
 +            LoadedInitialInput::Archive(input) => input.load_order,
 +        }
 +    }
 +}
++
 +struct InitialLoadError {
 +    load_order: usize,
 +    error: LinkError,
 +}
++
 +fn load_initial_inputs(loads: Vec<(usize, PathBuf)>) -> Result<Vec<LoadedInitialInput>, LinkError> {
 +    let mut results = Vec::new();
 +    let mut object_jobs = Vec::new();
 +    for (load_order, path) in loads {
 +        if matches!(path.extension().and_then(|ext| ext.to_str()), Some("a")) {
 +            results.push(load_archive_input(path, load_order));
 +        } else {
 +            object_jobs.push((load_order, path));
 +        }
 +    }
 +    results.extend(load_objects_parallel(object_jobs));
 +    results.sort_by_key(|result| match result {
 +        Ok(input) => input.load_order(),
 +        Err(error) => error.load_order,
 +    });
++
 +    let mut loaded = Vec::with_capacity(results.len());
 +    for result in results {
 +        match result {
 +            Ok(input) => loaded.push(input),
 +            Err(error) => return Err(error.error),
 +        }
 +    }
 +    Ok(loaded)
 +}
++
 +fn load_objects_parallel(
 +    jobs: Vec<(usize, PathBuf)>,
 +) -> Vec<Result<LoadedInitialInput, InitialLoadError>> {
 +    if jobs.is_empty() {
 +        return Vec::new();
 +    }
 +    let job_count = thread::available_parallelism()
 +        .map(usize::from)
 +        .unwrap_or(1)
 +        .min(jobs.len())
 +        .max(1);
 +    if job_count == 1 {
 +        return jobs
 +            .into_iter()
 +            .map(|(load_order, path)| load_object_input(path, load_order))
 +            .collect();
 +    }
++
 +    let queue = Arc::new(Mutex::new(VecDeque::from(jobs)));
 +    let (tx, rx) = mpsc::channel();
 +    thread::scope(|scope| {
 +        for _ in 0..job_count {
 +            let queue = Arc::clone(&queue);
 +            let tx = tx.clone();
 +            scope.spawn(move || loop {
 +                let Some((load_order, path)) = queue
 +                    .lock()
 +                    .expect("input load queue mutex poisoned")
 +                    .pop_front()
 +                else {
 +                    break;
 +                };
 +                tx.send(load_object_input(path, load_order))
 +                    .expect("input load receiver should stay live");
 +            });
 +        }
 +        drop(tx);
 +        rx.into_iter().collect()
 +    })
 +}
++
 +fn load_object_input(
 +    path: PathBuf,
 +    load_order: usize,
 +) -> Result<LoadedInitialInput, InitialLoadError> {
 +    let mut timings = InputLoadTimings::default();
 +    let phase_started = Instant::now();
 +    let bytes = fs::read(&path).map_err(|error| InitialLoadError {
 +        load_order,
 +        error: LinkError::Io(error),
 +    })?;
 +    timings.read = phase_started.elapsed();
++
 +    let phase_started = Instant::now();
 +    let parsed = ObjectFile::parse(&path, &bytes).map_err(|error| InitialLoadError {
 +        load_order,
 +        error: LinkError::from(error),
 +    })?;
 +    timings.object_parse = phase_started.elapsed();
++
 +    Ok(LoadedInitialInput::Object(Box::new(LoadedObjectInput {
 +        path,
 +        load_order,
 +        bytes,
 +        parsed,
 +        timings,
 +    })))
 +}
++
 +fn load_archive_input(
 +    path: PathBuf,
 +    load_order: usize,
 +) -> Result<LoadedInitialInput, InitialLoadError> {
 +    let mut timings = InputLoadTimings::default();
 +    let phase_started = Instant::now();
 +    let bytes = fs::read(&path).map_err(|error| InitialLoadError {
 +        load_order,
 +        error: LinkError::Io(error),
 +    })?;
 +    timings.read = phase_started.elapsed();
++
 +    let phase_started = Instant::now();
 +    Archive::open(&path, &bytes).map_err(|error| InitialLoadError {
 +        load_order,
 +        error: LinkError::from(InputAddError::from(error)),
 +    })?;
 +    timings.archive_parse = phase_started.elapsed();
++
 +    Ok(LoadedInitialInput::Archive(LoadedArchiveInput {
 +        path,
 +        load_order,
 +        bytes,
 +        timings,
 +    }))
 +}
++
 +fn register_loaded_initial_input(
 +    inputs: &mut Inputs,
 +    loaded: LoadedInitialInput,
 +) -> InputLoadTimings {
 +    match loaded {
 +        LoadedInitialInput::Object(input) => {
 +            inputs.add_parsed_object(input.path, input.bytes, input.parsed, input.load_order);
 +            input.timings
 +        }
 +        LoadedInitialInput::Archive(input) => {
 +            inputs.add_validated_archive(input.path, input.bytes, input.load_order);
 +            input.timings
 +        }
 +    }
 +}
++
  fn register_input(
      inputs: &mut Inputs,
      path: &std::path::Path,

src/resolve.rsmodified

      pub path: PathBuf,
      pub load_order: usize,
      pub archive_member_offset: Option<u32>,
 -    /// Raw bytes; `ObjectFile::parse` re-runs cheaply against this on
 -    /// demand. We don't cache a parsed view because `ObjectFile` copies
 -    /// the fields it needs on construction, so re-parse is idempotent.
 +    /// Raw bytes retained for diagnostics and future low-level readers.
      pub bytes: Vec<u8>,
 +    /// Parsed object view. It owns section/relocation/string-table buffers,
 +    /// so it is safe to build off-thread and then borrow during the link.
 +    pub parsed: ObjectFile,
+ }
  #[derive(Debug)]
          load_order: usize,
      ) -> Result<InputId, InputAddError> {
          // Validate now — we'd rather catch a bad object at the add site.
 -        ObjectFile::parse(&path, &bytes)?;
 +        let parsed = ObjectFile::parse(&path, &bytes)?;
 +        Ok(self.add_parsed_object(path, bytes, parsed, load_order))
 +    }
++
 +    pub fn add_parsed_object(
 +        &mut self,
 +        path: PathBuf,
 +        bytes: Vec<u8>,
 +        parsed: ObjectFile,
 +        load_order: usize,
 +    ) -> InputId {
          let id = InputId(self.objects.len() as u32);
          self.objects.push(ObjectInput {
              path,
              load_order,
              archive_member_offset: None,
              bytes,
 +            parsed,
          });
 -        Ok(id)
 +        id
+     }
      /// Register an `.a` file.
          load_order: usize,
      ) -> Result<ArchiveId, InputAddError> {
          Archive::open(&path, &bytes)?; // validate
 +        Ok(self.add_validated_archive(path, bytes, load_order))
 +    }
++
 +    pub fn add_validated_archive(
 +        &mut self,
 +        path: PathBuf,
 +        bytes: Vec<u8>,
 +        load_order: usize,
 +    ) -> ArchiveId {
          let id = ArchiveId(self.archives.len() as u32);
          self.archives.push(ArchiveInput {
              path,
              bytes,
              fetched: std::collections::HashSet::new(),
          });
 -        Ok(id)
 +        id
+     }
      /// Register a `.dylib`. TBD-backed dylibs go through
          &self.dylibs[id.0 as usize]
+     }
 -    /// Parse an `ObjectFile` view of a registered object. Fast — `ObjectFile`
 -    /// owns its buffers, so this is just the Mach-O walk cost.
 -    pub fn object_file(&self, id: InputId) -> Result<ObjectFile, ReadError> {
 +    /// Borrow a parsed `ObjectFile` view of a registered object.
 +    pub fn object_file(&self, id: InputId) -> Result<&ObjectFile, ReadError> {
          let o = &self.objects[id.0 as usize];
 -        ObjectFile::parse(&o.path, &o.bytes)
 +        Ok(&o.parsed)
+     }
      /// Open an `Archive` view, borrowing from the registry's bytes for the
          let logical = format!("{}({})", ai.path.display(), member.name);
          (logical, member.body.to_vec())
      };
 +    let logical_path = PathBuf::from(logical_path);
 +    let parsed = ObjectFile::parse(&logical_path, &member_bytes)?;
      inputs.archives[archive_id.0 as usize]
          .fetched
          .insert(member_id.0);
      let input_id = InputId(inputs.objects.len() as u32);
      inputs.objects.push(ObjectInput {
 -        path: PathBuf::from(logical_path),
 +        path: logical_path,
          load_order: archive_load_order,
          archive_member_offset: Some(member_id.0),
          bytes: member_bytes,
 +        parsed,
      });
      report.fetched_members += 1;
      report

tests/atom_integration.rsmodified

      // Atomize + back-patch.
      let obj = inputs.object_file(input_id).unwrap();
      let mut atom_table = AtomTable::new();
 -    let atomization = atomize_object(input_id, &obj, &mut atom_table);
 -    backpatch_symbol_atoms(
 -        &atomization,
 -        input_id,
 -        &obj,
 -        &mut sym_table,
 -        &mut atom_table,
 -    );
 +    let atomization = atomize_object(input_id, obj, &mut atom_table);
 +    backpatch_symbol_atoms(&atomization, input_id, obj, &mut sym_table, &mut atom_table);
      // At least one atom per defined function plus one for data_global.
      assert!(
      let _ = seed_all(&inputs, &mut sym_table).expect("seed_all");
      let obj = inputs.object_file(input_id).unwrap();
      let mut atom_table = AtomTable::new();
 -    let _atomization = atomize_object(input_id, &obj, &mut atom_table);
 +    let _atomization = atomize_object(input_id, obj, &mut atom_table);
      let cstring_atoms: Vec<_> = atom_table
          .iter()

tests/determinism.rsmodified

  //! Sprint 28 determinism guardrails.
  //!
  //! Parallel speedups are only safe if they never perturb the final image. This
 -//! test repeatedly links one synthetic-section-heavy executable and requires
 -//! byte-identical output across concurrent runs.
 +//! test repeatedly links a multi-object executable and requires byte-identical
 +//! output across concurrent runs.
  mod common;
+     }
      let root = unique_temp_dir("determinism").expect("create determinism temp dir");
 -    let obj = root.join("input.o");
 +    let main_obj = root.join("main.o");
      assemble(
          "\
          .section __TEXT,__text,regular,pure_instructions\n\
          .globl _main\n\
          _main:\n\
 +            bl _helper\n\
              adrp x8, _value@GOTPAGE\n\
              ldr x8, [x8, _value@GOTPAGEOFF]\n\
              ldr w0, [x8]\n\
              ret\n\
  \n\
 +        .subsections_via_symbols\n",
 +        &main_obj,
 +    )
 +    .expect("assemble determinism main fixture");
 +    let helper_obj = root.join("helper.o");
 +    assemble(
 +        "\
 +        .section __TEXT,__text,regular,pure_instructions\n\
 +        .globl _helper\n\
 +        _helper:\n\
 +            ret\n\
 +\n\
 +        .subsections_via_symbols\n",
 +        &helper_obj,
 +    )
 +    .expect("assemble determinism helper fixture");
 +    let data_obj = root.join("data.o");
 +    assemble(
 +        "\
          .section __DATA,__data\n\
          .globl _value\n\
          .p2align 2\n\
          _value:\n\
              .long 7\n\
 +\n\
          .subsections_via_symbols\n",
 -        &obj,
 +        &data_obj,
+     )
 -    .expect("assemble determinism fixture");
 +    .expect("assemble determinism data fixture");
 -    let baseline = link_once(&obj, &root, "baseline").expect("baseline deterministic link");
 +    let inputs = vec![main_obj, helper_obj, data_obj];
 +    let baseline = link_once(&inputs, &root, "baseline").expect("baseline deterministic link");
      let run_count = determinism_run_count();
      let jobs = determinism_jobs(run_count);
      let queue = Arc::new(Mutex::new((0..run_count).collect::<VecDeque<_>>()));
              let errors = Arc::clone(&errors);
              let baseline = baseline.clone();
              let root = root.clone();
 -            let obj = obj.clone();
 +            let inputs = inputs.clone();
              scope.spawn(move || loop {
                  let Some(index) = queue
                      .lock()
                  else {
                      break;
                  };
 -                match link_once(&obj, &root, &format!("run-{index:03}")) {
 +                match link_once(&inputs, &root, &format!("run-{index:03}")) {
                      Ok(bytes) if bytes == baseline => {}
                      Ok(bytes) => errors
                          .lock()
      let _ = fs::remove_dir_all(root);
+ }
 -fn link_once(obj: &Path, root: &Path, run_name: &str) -> Result<Vec<u8>, String> {
 +fn link_once(inputs: &[PathBuf], root: &Path, run_name: &str) -> Result<Vec<u8>, String> {
      let dir = root.join(run_name);
      fs::create_dir_all(&dir).map_err(|e| format!("create {}: {e}", dir.display()))?;
      let out = dir.join("deterministic.out");
      let opts = LinkOptions {
 -        inputs: vec![obj.to_path_buf()],
 +        inputs: inputs.to_vec(),
          output: Some(out.clone()),
          kind: OutputKind::Executable,
          ..LinkOptions::default()

tests/perf_baseline.rsmodified

          profile.phases.accounted_total() > Duration::ZERO,
          "{name}: all phase timings were zero"
      );
 +    let input_subphase_total = profile.phases.input_read
 +        + profile.phases.input_object_parse
 +        + profile.phases.input_archive_parse
 +        + profile.phases.input_dylib_parse
 +        + profile.phases.input_tbd_decode
 +        + profile.phases.input_tbd_materialize
 +        + profile.phases.input_reloc_parse;
 +    // Input subphases are summed worker-time once object parsing is parallel,
 +    // so they can legitimately exceed the wall-clock input parsing bucket.
      assert!(
 -        profile.phases.input_parsing
 -            >= profile.phases.input_read
 -                + profile.phases.input_object_parse
 -                + profile.phases.input_archive_parse
 -                + profile.phases.input_dylib_parse
 -                + profile.phases.input_tbd_decode
 -                + profile.phases.input_tbd_materialize
 -                + profile.phases.input_reloc_parse,
 -        "{name}: input subphases exceeded input total"
 +        input_subphase_total > Duration::ZERO,
 +        "{name}: all input subphase timings were zero"
      );
      assert!(
          profile.phases.layout