Parallelise select_best_assets

[dalonsoa]

Goal

Parallelise the per-asset appraisal loop inside select_best_assets (src/simulation/investment.rs) using Rayon, so that competing investment options are appraised concurrently rather than one at a time.

---

Step 1 — Rc → Arc migration

Why: Rayon requires all data shared across threads to be Send + Sync. std::rc::Rc<T> uses non-atomic reference counting and is neither, so it cannot cross thread boundaries. std::sync::Arc<T> is the drop-in replacement with the same ownership semantics but atomic counts.

Files changed: Every source file that owned or constructed an Rc-wrapped value — src/id.rs, src/commodity.rs, src/agent.rs, src/graph.rs, src/process.rs, src/asset.rs, src/asset/pool.rs, src/simulation.rs, src/simulation/investment.rs, src/simulation/investment/appraisal.rs and its submodules, src/simulation/market.rs, src/simulation/prices.rs, src/simulation/optimisation/constraints.rs, src/graph/investment.rs, src/graph/validate.rs, all src/input/ files, src/fixture.rs, and benches/assets.rs.

Process also received #[derive(Clone)], required by the benchmark's build_synthetic_processes helper.

---

Step 2 — Cell<AssetCapacity> → Mutex<AssetCapacity> in src/asset.rs

Why: After the Rc → Arc migration, Clippy (configured deny(clippy::all)) rejected the code with:

error: usage of an `Arc` that is not `Send` and `Sync`
  --> src/asset.rs
   = note: `Arc<Asset>` is not `Send` and `Sync` as `Asset` is neither `Send` nor `Sync`

The root cause was capacity: Cell<AssetCapacity>. Cell<T> is !Sync by design — it provides interior mutability with no synchronisation, which is only sound on a single thread. Asset needed interior mutability here because decrement_unit_count(&self) must mutate the capacity of a parent asset through a shared Arc<Asset> (multiple child assets point to the same parent).

Fix: Replace Cell with Mutex. A Mutex also provides interior mutability but with the synchronisation guarantee that makes Asset: Send + Sync. The lock is never actually contended (decommissioning is sequential), so there is no runtime cost beyond an uncontended lock/unlock.

Changes within src/asset.rs:

Site	Before	After
Import	use std::cell::Cell	use std::sync::Mutex
Field type	capacity: Cell<AssetCapacity>	capacity: Mutex<AssetCapacity>
#[derive(Clone)]	on Asset	Removed; manual impl Clone added
Initialisation (3 sites)	Cell::new(...)	Mutex::new(...)
capacity() reader	self.capacity.get()	*self.capacity.lock().expect(...)
set_capacity()	self.capacity.set(c)	Lock guard + assign
increase_capacity()	self.capacity.update(|c| c + cap)	Lock guard + *guard = *guard + cap
decrement_unit_count()	get() then set(...)	Single lock, mutate in place

A manual Clone impl was required because Mutex<T> does not implement Clone.

---

Step 3 — Enforce single-threaded HiGHS solving in src/simulation/investment/appraisal/optimisation.rs

Why: When multiple appraisals run in parallel, each HiGHS instance must not spawn its own background worker threads, otherwise N parallel appraisals × M HiGHS threads = severe CPU over-subscription.

What was tried first — set_threads(1) — and why it was rejected:

The highs::Model::set_threads(NonZeroU32::MIN) API was the obvious choice, but caused 5 test failures ("Incoherent model: Error"). HiGHS uses a thread-local task executor, initialised lazily on first use. If any previous HiGHS solve on the same OS thread used the default threads=0 (auto-detecting N threads), calling set_threads(1) later trips this guard in Highs.cpp:

if (options_.threads != 0 && max_threads != options_.threads) {
    return HighsStatus::kError;
}

cargo test reuses threads across tests, so the scheduler was already initialised with the auto-detected count before our test ran. There is no Rust-accessible API to reset it (Highs_resetGlobalScheduler exists in C++ but is not exposed through highs-sys).

Fix: Set parallel="off" instead:

highs_model.set_option("parallel", "off");

This disables HiGHS's concurrent dual simplex (PAMI) strategy without touching the threads option, so options_.threads stays 0 and the scheduler check is never triggered. Since all appraisal problems are LP (not MIP), disabling parallel simplex is sufficient to guarantee single-threaded execution.

---

Step 4 — Rayon parallelisation of the appraisal loop in src/simulation/investment.rs

What changed: Added use rayon::prelude::* and replaced the sequential for loop with par_iter():

// Before
let mut outputs = Vec::new();
for asset in &opt_assets {
    // ... adjust capacity, appraise ...
    outputs.push(appraise_investment(...)?);
}

// After
let mut outputs: Vec<AppraisalOutput> = opt_assets
    .par_iter()
    .map(|asset| -> Result<Option<AppraisalOutput>> {
        // ... adjust capacity ...
        Ok(Some(appraise_investment(...)?))
    })
    .collect::<Result<Vec<_>>>()?
    .into_iter()
    .flatten()
    .collect();

The loop was a natural fit for Rayon: every iteration reads only shared references (model, commodity, demand, remaining_capacities, coefficients), and make_mut() creates a fresh Arc allocation for each capacity-adjusted clone, so there is no shared mutable state between iterations. anyhow::Error is Send + Sync, so collect::<Result<Vec<_>>>() correctly short-circuits on the first solver failure.

Benchmark — benches/assets.rs was updated to compare the two modes side-by-side. The sequential group installs a 1-thread Rayon pool via sequential_pool.install(run), making par_iter degenerate to serial execution through the same code path — a fair like-for-like comparison.

Results

N technologies	Sequential	Parallel	Speedup
1	5.1 ms	5.4 ms	~1×
5	23.7 ms	14.9 ms	~1.6×
10	48.8 ms	23.3 ms	~2.1×
15	72.7 ms	26.1 ms	~2.8×
20	95.4 ms	40.5 ms	~2.4×

• At N=1 there is a small (~6%) overhead from Rayon task dispatch — expected and acceptable.
• Speedup grows quickly, reaching roughly 2–3× beyond N=10, consistent with the available core count and the short duration of each individual HiGHS LP solve (~4–5 ms).
• Sequential scaling is nearly perfectly linear (~4.7 ms per technology), confirming each appraisal costs roughly the same and the benchmark is exercising the intended code path.
• The parallelism plateau reflects Amdahl's law: the outer greedy selection loop (picking one best asset per round and updating demand) is sequential and cannot be parallelised, so it forms the irreducible serial fraction.

Validation

• cargo clippy --all-targets — clean.
• cargo test — 487 tests pass.

Fixes # (issue)

Type of change

• Bug fix (non-breaking change to fix an issue)
• New feature (non-breaking change to add functionality)
• Refactoring (non-breaking, non-functional change to improve maintainability)
• Optimization (non-breaking change to speed up the code)
• Breaking change (whatever its nature)
• Documentation (improve or add documentation)

Key checklist

• All tests pass: $ cargo test
• The documentation builds and looks OK: $ cargo doc
• Update release notes for the latest release if this PR adds a new feature or fixes a bug
  present in the previous release

Further checks

• Code is commented, particularly in hard-to-understand areas
• Tests added that prove fix is effective or that feature works

[Copilot]

Pull request overview

This PR improves investment-option selection performance by parallelising the per-asset appraisal loop in select_best_assets using Rayon, and makes the supporting data structures thread-safe so appraisals can run concurrently without violating Send + Sync constraints.

Changes:

• Migrates shared ownership across the codebase from Rc to Arc (including IDs and many model maps) to enable safe cross-thread sharing.
• Makes Asset thread-safe by replacing Cell<AssetCapacity> with Mutex<AssetCapacity> and adding a manual Clone impl.
• Ensures each HiGHS solve stays single-threaded during parallel appraisals by setting parallel="off", and adds a benchmark comparing parallel vs forced-single-thread execution.

Reviewed changes

Copilot reviewed 28 out of 29 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
src/simulation/prices.rs	Updates test helpers/fixtures from Rc to Arc.
src/simulation/optimisation/constraints.rs	Updates tests to use Arc for shared data.
src/simulation/market.rs	Updates tests and constraints usage from Rc to Arc.
src/simulation/investment/appraisal/optimisation.rs	Forces single-threaded HiGHS solving via parallel="off" to avoid oversubscription.
src/simulation/investment/appraisal/coefficients.rs	Switches coefficients map storage from Rc to Arc.
src/simulation/investment/appraisal.rs	Converts appraisal outputs/inputs from Rc to Arc and updates tests.
src/simulation/investment.rs	Parallelises appraisal over candidate assets using Rayon par_iter() and updates tests.
src/simulation.rs	Updates candidate asset creation to clone Arc<Process>.
src/process.rs	Converts process-related shared maps and ProcessFlow.commodity from Rc to Arc.
src/input/process/parameter.rs	Stores parsed ProcessParameter in Arc.
src/input/process/investment_constraints.rs	Stores parsed investment constraints in Arc.
src/input/process/flow.rs	Migrates flow parsing/storage to Arc and uses Arc::get_mut where uniqueness is assumed.
src/input/process/availability.rs	Stores ActivityLimits in Arc.
src/input/process.rs	Uses Arc::get_mut when populating processes.
src/input/asset.rs	Updates asset construction to use Arc<Process>.
src/input/agent/search_space.rs	Migrates agent search space types from Rc<Vec<Rc<Process>>> to Arc<Vec<Arc<Process>>> and updates tests.
src/input/agent/commodity_portion.rs	Updates tests/fixtures from Rc to Arc.
src/id.rs	Changes ID backing storage from Rc<str> to Arc<str> for thread-safety.
src/graph/validate.rs	Updates graph validation tests to use Arc commodities.
src/graph/investment.rs	Updates investment graph tests to use Arc commodities.
src/graph.rs	Updates flow retrieval return type to Arc<IndexMap<...>>.
src/fixture.rs	Updates fixtures to construct and return Arc-wrapped values.
src/commodity.rs	Switches CommodityMap values from Rc<Commodity> to Arc<Commodity>.
src/asset/pool.rs	Updates asset pool tests to use Arc<Process>.
src/asset.rs	Replaces Cell capacity with Mutex, migrates AssetRef to Arc, and updates related tests.
src/agent.rs	Migrates AgentSearchSpaceMap to Arc types and updates iterator return type accordingly.
Cargo.toml	Adds Rayon dependency for parallel iteration.
Cargo.lock	Locks in Rayon dependency.
benches/assets.rs	Adds parallel vs sequential (1-thread pool) benchmark groups and migrates to Arc.

Suppressed comments (1)

src/asset.rs:785

• increase_capacity takes &mut self, so it can avoid locking the Mutex and use Mutex::get_mut() instead. This is simpler and avoids an unnecessary lock/unlock in a method that may be called frequently.

        // We require a `&mut self` here to prevent accidental mutation through shared references.
        let mut guard = self.capacity.lock().expect("capacity lock poisoned");
        *guard = *guard + capacity;

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[tsmbland]

Cool! Lots of small changes but this is actually quite minimal. If I understand, was highs already parallelising each optimisation, but it's more efficient to make that single-thread and parallelise the appraisal as a whole?

Would we want an ability to turn parallelisation on/off (e.g. via an option in settings.toml)?

[alexdewar]

This looks cool! Nice to see how it'll all look together 😄 .

Agree with @tsmbland that it would be nice to configure the number of threads used. You can do this by setting the RAYON_NUM_THREADS env var, but it would be good to be able to set it via settings.toml and a command-line flag eventually too.

Some thoughts:

Step 1

Arc is a reasonable abstraction to use and if we have to use Arcs everywhere instead of Rcs, it's not the end of the world, however, it's not free and using atomics has an effect on performance (albeit probably a small one). The issue is that the refcounts have to be synchronised across CPU cores, which can mean flushing caches etc. and can be a little expensive (though atomics are still cheaper than mutexes).

I only mention it because I think the only reason Arcs are needed at all is because we clone assets in order to set their capacities for appraisal, but this might not be especially elegant in any case. If we were to rewrite things so each appraisal (==thread) instead had an immutable reference to an asset (&AssetRef), but could pass a capacity explicitly to appraise_investment rather than using the asset's stored capacity, then we wouldn't need any Arcs (I think). This would be better for performance and probably cleaner.

Step 2

This is no longer needed since #1464 is merged 😄 (capacity isn't wrapped in a Cell anymore). Assets no longer have any interior mutability to worry about.

Step 3

Assuming that limiting HiGHS to a single thread during appraisal doesn't meaningfully hurt performance in other contexts, this is ok. It would be worth digging into a bit more.

If/when we parallelise other parts of the investment code, we will potentially have the same problem with dispatch runs executing in parallel to one another. I suspect HiGHS's parallelisation helps more with dispatch than appraisal, so the trade-off might be a bit more subtle there. Worth exploring.

Step 4

The benchmarking is really cool! The results are tantalising... It would be fun to try running on the HPC with as many threads as we can get our hands on.

[alexdewar]

PS - re HiGHS. If we want to use a feature that's not accessible via the Rust crate, we can add it and submit it upstream. I've done it for a couple of things. It's not hard.