Parallelise select_best_assets - #1466
Conversation
There was a problem hiding this comment.
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
RctoArc(including IDs and many model maps) to enable safe cross-thread sharing. - Makes
Assetthread-safe by replacingCell<AssetCapacity>withMutex<AssetCapacity>and adding a manualCloneimpl. - 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_capacitytakes&mut self, so it can avoid locking theMutexand useMutex::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.
|
Cool! Lots of small changes but this is actually quite minimal. If I understand, was Would we want an ability to turn parallelisation on/off (e.g. via an option in |
alexdewar
left a comment
There was a problem hiding this comment.
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.
|
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. |
6322e8e to
11575f0
Compare
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #1466 +/- ##
==========================================
- Coverage 89.78% 89.78% -0.01%
==========================================
Files 60 60
Lines 8439 8445 +6
Branches 8439 8445 +6
==========================================
+ Hits 7577 7582 +5
Misses 545 545
- Partials 317 318 +1 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 28 out of 29 changed files in this pull request and generated no new comments.
Suppressed comments (1)
src/simulation/investment/appraisal/optimisation.rs:135
highs_model.set_option("parallel", "off")will panic if the option name/value is invalid for the linked HiGHS version (and it also silently ignores the error channel). Since this is enforcing a correctness/operational constraint (avoid over-subscription), it’s safer to usetry_set_optionand propagate a contextual error instead of panicking at runtime.
// Enforce single-threaded solving: when appraisals run in parallel via Rayon each HiGHS
// instance must not spawn additional worker threads. Setting `parallel="off"` disables
// HiGHS's concurrent simplex strategy without touching the global thread-pool scheduler
// (setting `threads=N` would fail if the scheduler was already initialised on this thread
// by a previous solve with a different count).
highs_model.set_option("parallel", "off");
|
I think this is ready to merge, so going to rebase this onto main, and rebase #1446 onto this Will also open an issue about configuring the number of threads |
5edc8c3 to
160676c
Compare
|
pre-commit is passing locally, but here it's running on a file that doesn't exist in this branch... strange |
tsmbland
left a comment
There was a problem hiding this comment.
Assuming the performance cost of Rc -> Arc is negligible, which it appears to be, then this is all good!
This would be because the tests are triggered on |
Ah ok makes sense. I guess maybe it was confused because I rebased onto |
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→ArcmigrationWhy: 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.rsand its submodules,src/simulation/market.rs,src/simulation/prices.rs,src/simulation/optimisation/constraints.rs,src/graph/investment.rs,src/graph/validate.rs, allsrc/input/files,src/fixture.rs, andbenches/assets.rs.Processalso received#[derive(Clone)], required by the benchmark'sbuild_synthetic_processeshelper.## Step 2 — `Cell` → `Mutex` in `src/asset.rs`
Site
Before
After
Import
Field type
on
Removed; manual
Initialisation (3 sites)
Lock guard + assign
Lock guard +
Single lock, mutate in place
Why: After the
Rc → Arcmigration, Clippy (configureddeny(clippy::all)) rejected the code with:The root cause was
capacity: Cell<AssetCapacity>.Cell<T>is!Syncby design — it provides interior mutability with no synchronisation, which is only sound on a single thread.Assetneeded interior mutability here becausedecrement_unit_count(&self)must mutate the capacity of a parent asset through a sharedArc<Asset>(multiple child assets point to the same parent).Fix: Replace
CellwithMutex. AMutexalso provides interior mutability but with the synchronisation guarantee that makesAsset: 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:use std::cell::Celluse std::sync::Mutexcapacity: Cell<AssetCapacity>capacity: Mutex<AssetCapacity>#[derive(Clone)]Assetimpl CloneaddedCell::new(...)Mutex::new(...)capacity()readerself.capacity.get()*self.capacity.lock().expect(...)set_capacity()self.capacity.set(c)increase_capacity()self.capacity.update(|c| c + cap)*guard = *guard + capdecrement_unit_count()get()thenset(...)A manual
Cloneimpl was required becauseMutex<T>does not implementClone.(No longer needed - see comment from Alex below)
Step 3 — Enforce single-threaded HiGHS solving in
src/simulation/investment/appraisal/optimisation.rsWhy: 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 defaultthreads=0(auto-detecting N threads), callingset_threads(1)later trips this guard inHighs.cpp:cargo testreuses 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_resetGlobalSchedulerexists in C++ but is not exposed throughhighs-sys).Fix: Set
parallel="off"instead:This disables HiGHS's concurrent dual simplex (PAMI) strategy without touching the
threadsoption, sooptions_.threadsstays0and 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.rsWhat changed: Added
use rayon::prelude::*and replaced the sequentialforloop withpar_iter():The loop was a natural fit for Rayon: every iteration reads only shared references (
model,commodity,demand,remaining_capacities,coefficients), andmake_mut()creates a freshArcallocation for each capacity-adjusted clone, so there is no shared mutable state between iterations.anyhow::ErrorisSend + Sync, socollect::<Result<Vec<_>>>()correctly short-circuits on the first solver failure.Benchmark —
benches/assets.rswas updated to compare the two modes side-by-side. The sequential group installs a 1-thread Rayon pool viasequential_pool.install(run), makingpar_iterdegenerate to serial execution through the same code path — a fair like-for-like comparison.Results
Validation
cargo clippy --all-targets— clean.cargo test— 487 tests pass.Fixes # (issue)
Type of change
Key checklist
$ cargo test$ cargo docpresent in the previous release
Further checks