Add size-based pairwise dispatch to all_to_all - #7431
Conversation
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
…nt only) Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Signed-off-by: iemAnshuman <asquare567@gmail.com>
Threshold selection for the 4 KiB defaultI evaluated each sampled global threshold over the 32 cells in the 16-node sweep: four locality counts and eight payload sizes. Cells below the threshold keep the fastest non-pairwise result and contribute 1.0x to the overall geometric mean. Cells at or above the threshold use the measured pairwise result. For each cell, the non-pairwise baseline is the fastest result among standard one-shot, standard reused, forced flat, and tree arities 2, 3, and 4.
The 1 KiB threshold has the highest aggregate geometric mean, but it introduces two regressions:
The 4 KiB threshold wins all 12 cells it selects and produces a 2.08x whole-matrix geometric mean, 3.2% below the maximum at 1 KiB. At the 4 KiB boundary, pairwise is 1.14x, 2.79x, 5.25x, and 13.82x faster at 16, 32, 64, and 128 localities, respectively. This makes 4096 bytes the smallest no-regression threshold in this sweep. The crossover can change with the transport and topology, so callers can override the default. Full measurements and methodology: 16-node pairwise |
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Up to standards ✅🟢 Issues
|
Signed-off-by: iemAnshuman <asquare567@gmail.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
libs/full/collectives/tests/performance/benchmark_collectives.cpp (1)
1655-1661: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueHoist the threshold computation out of the timed region.
benchmark_pairwise_threshold(block_size)runs after the timer is constructed at line 1656.block_sizeis constant for the whole loop, so the value never changes. Compute it once before the loop and keep the timed region limited to the collective call.♻️ Proposed change
+ pairwise_threshold_arg const threshold = + benchmark_pairwise_threshold(block_size); for (std::size_t i = 0; i != warmup_iterations + iterations; ++i) {recv_data = all_to_all(all_to_all_direct_basename, std::move(iter_data), num_sites_arg(num_localities), this_site_arg(this_locality), - generation_arg(i + 1), root_site_arg(0), - benchmark_pairwise_threshold(block_size)) + generation_arg(i + 1), root_site_arg(0), threshold) .get();🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@libs/full/collectives/tests/performance/benchmark_collectives.cpp` around lines 1655 - 1661, Compute benchmark_pairwise_threshold(block_size) once before the loop, store the result in a reusable local, and pass that value to all_to_all in the timed benchmark block. Keep the timer and collective call unchanged otherwise, ensuring threshold computation is excluded from timing.libs/full/collectives/src/channel_communicator.cpp (1)
190-195: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueConsider bounding the cache or documenting the growth characteristic.
Every distinct name adds one permanent entry. Each entry keeps a channel communicator and its AGAS registration alive until
reset_cached_channel_communicatorsruns at shutdown. A caller that uses a fresh basename per phase, per iteration, or per site-count therefore accumulates communicators and AGAS names for the whole run. The dispatcher inall_to_all.hppderives the name from the user basename, so the number of entries follows user input.Add a metric or a documented upper bound, or expose a way to drop one entry, so an application can control the growth.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@libs/full/collectives/src/channel_communicator.cpp` around lines 190 - 195, Update the cached_channel_communicators cache and its public management API to support removing an individual entry by basename, using cached_channel_communicators_mtx for synchronized lookup and erase. Ensure the removed communicator and its AGAS registration are released, while preserving reset_cached_channel_communicators for clearing the entire cache.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@libs/full/collectives/include/hpx/collectives/all_to_all.hpp`:
- Around line 669-718: Ensure the all_to_all_by_name overload documents or
enforces a stable routing decision for each basename across generations. Update
the relevant documentation near the direct pairwise/routed path selection to
require consistent threshold and row-type choices, or make that selection
reproducible from the basename so generations cannot alternate paths and leave
the collective generation gate unsynchronized.
---
Nitpick comments:
In `@libs/full/collectives/src/channel_communicator.cpp`:
- Around line 190-195: Update the cached_channel_communicators cache and its
public management API to support removing an individual entry by basename, using
cached_channel_communicators_mtx for synchronized lookup and erase. Ensure the
removed communicator and its AGAS registration are released, while preserving
reset_cached_channel_communicators for clearing the entire cache.
In `@libs/full/collectives/tests/performance/benchmark_collectives.cpp`:
- Around line 1655-1661: Compute benchmark_pairwise_threshold(block_size) once
before the loop, store the result in a reusable local, and pass that value to
all_to_all in the timed benchmark block. Keep the timer and collective call
unchanged otherwise, ensuring threshold computation is excluded from timing.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: da7df8e0-0d83-4abc-9ac3-423c82e9386e
📒 Files selected for processing (11)
libs/full/collectives/CMakeLists.txtlibs/full/collectives/include/hpx/collectives/all_to_all.hpplibs/full/collectives/include/hpx/collectives/argument_types.hpplibs/full/collectives/include/hpx/collectives/channel_communicator.hpplibs/full/collectives/include/hpx/collectives/detail/pairwise_all_to_all_helpers.hpplibs/full/collectives/src/channel_communicator.cpplibs/full/collectives/tests/performance/benchmark_collectives.cpplibs/full/collectives/tests/unit/CMakeLists.txtlibs/full/collectives/tests/unit/pairwise_all_to_all.cpplibs/full/collectives/tests/unit/pairwise_all_to_all_dispatch.cpplibs/full/init_runtime/src/pre_main.cpp
| if (basename != nullptr && basename[0] != '\0' && | ||
| this_site < num_sites && root_site < num_sites && | ||
| !generation.is_default() && | ||
| exchange_pairwise(static_cast<std::size_t>(num_sites), | ||
| pairwise_type_bytes<T>(), threshold)) | ||
| { | ||
| // The channel communicator carries a name of its own so it | ||
| // cannot collide with the collective communicator registered | ||
| // under this basename. That name spans a group of sites rather | ||
| // than a single call: one communicator serves every | ||
| // generation, and the exchange tag is what keeps the | ||
| // generations apart. Registering a name per generation instead | ||
| // would put an AGAS registration and a full peer lookup back | ||
| // into every call, which is the cost this path exists to | ||
| // remove. | ||
| // | ||
| // The site count belongs in the name because two groups of | ||
| // different size are two different communicators, and the | ||
| // cache creates one only on the first call that names it. | ||
| std::string channel_basename(basename); | ||
| HPX_ASSERT(!channel_basename.empty()); | ||
| if (channel_basename.back() != '/') | ||
| { | ||
| channel_basename += '/'; | ||
| } | ||
| channel_basename += "pairwise/" + | ||
| std::to_string(static_cast<std::size_t>(num_sites)) + "/"; | ||
|
|
||
| std::size_t const num_sites_value = num_sites; | ||
| std::size_t const this_site_value = this_site; | ||
| tag_arg const tag(static_cast<std::size_t>(generation)); | ||
|
|
||
| // The communicator is chained onto rather than waited for. | ||
| // This function backs an overload that hands the caller a | ||
| // future, so it must not spend the caller's thread on the | ||
| // first call's AGAS registration before returning one. The | ||
| // communicator type is spelled out because the unqualified | ||
| // name resolves to the detail class in this namespace. | ||
| return get_cached_channel_communicator( | ||
| HPX_MOVE(channel_basename), num_sites, this_site) | ||
| .then(hpx::launch::sync, | ||
| [local_result = HPX_MOVE(local_result), num_sites_value, | ||
| this_site_value, tag](hpx::shared_future< | ||
| hpx::collectives::channel_communicator>&& | ||
| f) mutable { | ||
| return pairwise_all_to_all(f.get(), | ||
| HPX_MOVE(local_result), num_sites_value, | ||
| this_site_value, tag); | ||
| }); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect how the collective communicator gate handles generation numbers.
fd -t f 'communicator.hpp' libs/full/collectives/include -x ast-grep outline {} --items all
rg -nP -C6 'generation' libs/full/collectives/include/hpx/collectives/detail/communicator.hppRepository: TheHPXProject/hpx
Length of output: 20008
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the direct/routed dispatch, all_to_all operations, and documentation around basename overloads.
fd -t f 'all_to_all.hpp' libs/full/collectives/include/hpx/collectives -x sh -c '
echo "--- $1 ---"
wc -l "$1"
ast-grep outline "$1" --items all 2>/dev/null | sed -n "1,220p"
' sh {}
echo "--- relevant all_to_all.hpp sections ---"
sed -n '1,110p' libs/full/collectives/include/hpx/collectives/all_to_all.hpp
sed -n '450,760p' libs/full/collectives/include/hpx/collectives/all_to_all.hpp
sed -n '760,840p' libs.full.collectives /include/hpx/collectives/all_to_all.hpp || trueRepository: TheHPXProject/hpx
Length of output: 23463
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "--- communicator gate definitions ---"
fd -t f 'synchronization\.(hpp|cpp|ipp)$' . | rg 'lcos|synchronization|mutex' || true
fd -t f 'gate.*' libs/full/collectives/include hpx -x sh -c '
echo "--- $1 ---"
wc -l "$1"
ast-grep outline "$1" --items all 2>/dev/null | sed -n "1,160p" || true
rg -nP -C4 'class .*gate|gate\.h|next_generation|synchronize|set_future|generation' "$1" || true
' sh {}
echo "--- all_to_all sync tail ---"
sed -n '800,880p' libs/full/collectives/include/hpx/collectives/all_to_all.hpp
echo "--- synchronize/set definitions if present in source ---"
rg -nP -C4 'struct gate|class.*gate|shared_latch_mutex|next_generation\(|\.synchronize\(|void synchronize' libs hpx 2>/dev/null | sed -n '1,260p'Repository: TheHPXProject/hpx
Length of output: 540
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "--- communicator server synchronization usage ---"
rg -n -C8 'shared_latch|shared_mutex|shared_latch_mutex|\.synchronize\(|next_generation\(|get_future_and_synchronize|generation_arg' libs/full/collectives/include/hpx/collectives/detail/communicator.hpp
echo "--- lcos_local/synchronization files ---"
fd -t f '.*' hpx/libs/full hpx/modules 2>/dev/null | rg 'synchronization|mutex|lcos|future' | sed -n '1,200p' || true
echo "--- git show if needed around gate definitions ---"
rg -n -C8 'class shared_latch_mutex|shared_latch_mutex.*gate|next_generation\(.*generation|set\(.*generation|synchronize\(.*generation' . --glob '*.{hpp,h,cpp}' --glob '*.h' --glob '*.hpp' --glob '*.cpp' 2>/dev/null | sed -n '1,320p'Repository: TheHPXProject/hpx
Length of output: 37032
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "--- and_gate implementation ---"
sed -n '330,560p' libs/core/lcos_local/include/hpx/lcos_local/and_gate.hpp
echo "--- communicator server gate declaration/context ---"
rg -n -C8 'and_gate|gate_|next_generation|set_future|shared_future|future<.*future|get_generation|generation' libs/full/collectives/include/hpx/collectives/detail/communicator.hpp libs/full/collectives/include/hpx/collectives/detail/communicator
echo "--- locate communicator server includes and initialization ---"
rg -n -C4 'gate_|next_generation|get_shared_future|shared_future<|future<.*future' libs/full/collectives/include --glob '*.hpp'
echo "--- focused source files for communicator and future/trigger helpers ---"
fd -t f 'communicator|trigger|future|channel' libs hpx 2>/dev/null | rg 'collectives|core/lcos|modules' | sed -n '1,200p'Repository: TheHPXProject/hpx
Length of output: 50373
Document or reject routing path changes across generations for one basename.
all_to_all_by_name can select the direct pairwise path or the routed collective path based on threshold and the row type. The direct path uses a separate channel communicator and does not advance the collective communicator's generation gate. If generation 1 takes the direct path and generation 2 takes the routed path on the same basename, generation 2 waits for a gate position that generation 1 never moved. Update the overload documentation to require a stable routing decision per basename, or make the path choice reproducible per basename.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@libs/full/collectives/include/hpx/collectives/all_to_all.hpp` around lines
669 - 718, Ensure the all_to_all_by_name overload documents or enforces a stable
routing decision for each basename across generations. Update the relevant
documentation near the direct pairwise/routed path selection to require
consistent threshold and row-type choices, or make that selection reproducible
from the basename so generations cannot alternate paths and leave the collective
generation gate unsynchronized.
|
@iemAnshuman I wouldn't have guessed that the channel communicator setup is actually decently performant. We've never spent any time on optimizing that. This is cool. |
Performance test reportHPX PerformanceComparison
Info
Comparison
Info
Comparison
Info
Explanation of Symbols
|
3 participants
Proposed Changes
all_to_alloverloads.pairwise_threshold_arg, with a default of 4096 bytes, to select between routed and pairwise exchange.--pairwise_thresholdinbenchmark_collectives.Any background context you want to provide?
The basename-based implementation currently routes every contribution through one communicator site. This limits the number of messages sent by each site, but concentrates the exchange on one locality. The pairwise path sends each destination row directly to its peer. This removes the relay when the payload is large enough to offset the additional messages.
Pairwise exchange requires at least three sites and an explicit positive generation. Calls using the default generation remain on the routed path. Automatic dispatch uses
sizeof(T)for trivially copyable, fixed-size row types. Dynamically sized rows remain routed because local size checks could cause different sites to select different algorithms. Callers that know the rows are uniform can passpairwise_threshold_arg(0)at every site to request pairwise exchange.The channel communicator is cached by normalized basename and site count. Reusing it across generations avoids repeated AGAS registration and peer discovery, while generation tags keep exchanges separate. The implementation preserves the existing argument validation and supports
vector<bool>rows and rows that cannot be default-constructed or assigned.The 16-node Rostam/Buran campaign covered 16, 32, 64, and 128 localities, with payloads from 4 B through 64 KiB per destination. All 960 executions completed without retries or failures. A 4 KiB threshold was the smallest sampled threshold for which every selected pairwise result beat the fastest non-pairwise HPX result in the same campaign.
Full measurements: 16-node pairwise
all_to_allresultsVerification
benchmark_collectives.all_to_alltests.intelements.clang-format --dry-run --Werrorandgit diff --check.Checklist
Not all points below apply to all pull requests.