Feat/distributed adaptors

[shivansh023023]

Proposed Changes

• Scaffolded distributed_bulk_sender with standard P2300 environment queries and completion signatures.
• Implemented the query hook for ex::bulk to intercept executions on the distributed_scheduler.
• Added a temporary local for loop inside the receiver as a fallback/stub for testing the CPO routing.
• Created a full unit test suite (distributed_bulk_test.cpp) covering shape iteration, value propagation, zero-shape handling, and exception forwarding.

Any background context you want to provide?

This is a Draft PR to lay the foundation for distributed_bulk within the execution_distributed module. The current implementation successfully intercepts the CPO and processes the bulk execution locally. The immediate next step is to replace the local fallback loop with the actual parcelport dispatch logic using hpx::actions to achieve true remote distributed data-parallel execution.

Checklist

Not all points below apply to all pull requests.

• I have added a new feature and have added tests to go along with it.
• I have fixed a bug and have added a regression test.
• I have added a test using random numbers; I have made sure it uses a seed, and that random numbers generated are valid inputs for the tests.

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro Plus

Run ID: b678234e-d545-40d0-894b-9f3ae993a3fa

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

📝 Walkthrough

Walkthrough

Adds a networking-gated distributed bulk sender adaptor, integrates it with distributed_scheduler::query(bulk_t, ...), and adds two-locality unit tests covering local bulk execution, forwarding, errors, scheduler propagation, and zero-shape behavior.

Changes

Distributed bulk execution

Layer / File(s)	Summary
Distributed bulk sender adaptor libs/full/execution_distributed/include/hpx/execution_distributed/distributed_bulk_sender.hpp	Defines the sender, completion environment, receiver wrapper, local shape iteration, exception forwarding, and connect operations.
Scheduler bulk query integration libs/full/execution_distributed/include/hpx/execution_distributed/distributed_scheduler.hpp	Adds the bulk_t query overload that constructs a distributed_bulk_sender with decayed arguments and the current scheduler.
Build registration and behavioral validation libs/full/execution_distributed/CMakeLists.txt, libs/full/execution_distributed/tests/unit/*	Adds the header to the module and registers tests using two localities for bulk invocation, value forwarding, errors, scheduler propagation, and zero-shape behavior.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Caller
  participant distributed_scheduler
  participant distributed_bulk_sender
  participant UpstreamSender
  participant DownstreamReceiver

  Caller->>distributed_scheduler: query(bulk_t, sender, shape, f)
  distributed_scheduler->>distributed_bulk_sender: construct with sender, shape, f, scheduler
  Caller->>distributed_bulk_sender: connect(receiver)
  distributed_bulk_sender->>UpstreamSender: connect(bulk_receiver)
  UpstreamSender->>distributed_bulk_sender: set_value(Ts...)
  distributed_bulk_sender->>distributed_bulk_sender: invoke f(index, Ts...) for each shape index
  distributed_bulk_sender->>DownstreamReceiver: forward set_value(Ts...) or set_error(exception_ptr)

Loading

Possibly related PRs

• TheHPXProject/hpx#7368: Refactors the bulk_t completion-scheduler customization path used by the new scheduler query hook.

Suggested labels: category: algorithms, category: executors

Suggested reviewers: hkaiser

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 inconclusive)

Check name	Status	Explanation	Resolution
Title check	❓ Inconclusive	The title is too generic and doesn't identify the new distributed bulk sender, scheduler hook, or tests.	Rename it to something specific like 'Add distributed bulk sender support to execution_distributed'.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description check	✅ Passed	The description matches the changeset, describing the bulk sender scaffold, bulk hook, local fallback, and unit tests.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[StellarBot]

Can one of the admins verify this patch?

[codacy-production]

Up to standards ✅

🟢 Issues 0 issues

Results:
0 new issues

View in Codacy

_{NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.}

[coderabbitai]

Actionable comments posted: 2

🤖 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/execution_distributed/include/hpx/execution_distributed/distributed_bulk_sender.hpp`:
- Around line 195-198: Update the bulk invocation loop in the distributed
sender’s shape-handling code to support integral shapes by iterating indices
from zero up to shape_. Preserve the existing HPX_INVOKE(f_, ..., ts...) call
for each generated index, while retaining the range-based iteration behavior for
non-integral shapes if applicable.

In `@libs/full/execution_distributed/tests/unit/distributed_bulk_test.cpp`:
- Around line 103-112: Update the catch-all handler in the bulk exception test
to explicitly fail via the test framework instead of setting caught_exception to
true. Keep success limited to the std::runtime_error("bulk_error") branch in the
existing exception handling.

🪄 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: b7e4f987-daa2-4eee-9ae3-1d44bcac0afb

📥 Commits

Reviewing files that changed from the base of the PR and between 5a19320 and feb46f3.

📒 Files selected for processing (5)

• libs/full/execution_distributed/CMakeLists.txt
• libs/full/execution_distributed/include/hpx/execution_distributed/distributed_bulk_sender.hpp
• libs/full/execution_distributed/include/hpx/execution_distributed/distributed_scheduler.hpp
• libs/full/execution_distributed/tests/unit/CMakeLists.txt
• libs/full/execution_distributed/tests/unit/distributed_bulk_test.cpp

[hkaiser]

Please pay attention to the comments. Also, please rebase onto master to resolve the conflicts.

[shivansh023023]

Please pay attention to the comments. Also, please rebase onto master to resolve the conflicts.

Done

[iemAnshuman]

@shivansh023023 Had a read through this. The part that is easy to get wrong is right: bulk_t in libs/core/execution/include/hpx/execution/algorithms/bulk.hpp:223 dispatches on has_query_v<Scheduler, bulk_t, Sender, Shape const&, F&&>, and your distributed_scheduler::query signature matches it exactly, so the interception works for the right reason rather than by accident. Completion signatures look correct too, adding set_error_t(exception_ptr) while keeping set_stopped, and try_catch_exception_ptr is the right idiom here.

One note on wording: the description says you implemented "the tag_invoke overload for ex::bulk", but the code customizes through a scheduler query. Worth correcting so future readers look in the right place.

My main comment is about the tests.

All five construct distributed_scheduler{hpx::find_here()}, so every one runs against the local locality, and the implementation is a local for loop regardless. The CMake entry asks for LOCALITIES 2 but the second locality is never used. What the suite actually proves is that the CPO is intercepted and that a local loop runs the right number of times. It does not exercise anything distributed.

That would be fine for a stub, except the file comment opens with "executes data-parallel work on a remote HPX locality via the distributed_scheduler" and the "Current Status" note saying it is a local fallback sits about twenty lines further down. PR descriptions get forgotten, doc comments do not, so I would move the stub caveat to the first paragraph of the file comment. Someone reading the header a month from now should not have to reach the bottom to learn the parcelport dispatch is not there yet.

Test 4 is weaker than it looks:

catch (std::runtime_error const& e) { caught_exception = true; ... }
catch (...)                         { caught_exception = true; }
HPX_TEST(caught_exception);

The catch-all means any exception passes, including one thrown before the bulk loop ever runs. If the loop silently stopped working, this test would still go green. I would drop the catch-all, or set a separate flag inside it and assert that flag stayed false, so the test can only pass for the reason it claims.

Smaller things:

distributed_scheduler.hpp no longer stands on its own. query() returns detail::distributed_bulk_sender<...>, which is only forward declared there, with the definition in distributed_bulk_sender.hpp, which includes it back. That resolves through the generated umbrella header, but anyone including only distributed_scheduler.hpp and calling bulk gets an incomplete type. Worth a comment at the forward declaration saying which header completes it.

P2300 allows bulk to invoke f concurrently, and this runs strictly sequentially. Harmless today, but the remote version will change observable behaviour for functions that are not thread safe, so it is worth writing down now while the reason is fresh.

There is no constraint that f is invocable with (index, Args&...), so a wrong arity shows up as a deep template error instead of a clear one.

Last, a process point rather than a code one: the description says "This is a Draft PR" but the PR is not marked as draft, and Fixes # is empty. Reviewers seeing a normal open PR will hold a 245 line header to merge standards, which is probably not the feedback you are after at this stage.

[shivansh023023]

@shivansh023023 Had a read through this. The part that is easy to get wrong is right: bulk_t in libs/core/execution/include/hpx/execution/algorithms/bulk.hpp:223 dispatches on has_query_v<Scheduler, bulk_t, Sender, Shape const&, F&&>, and your distributed_scheduler::query signature matches it exactly, so the interception works for the right reason rather than by accident. Completion signatures look correct too, adding set_error_t(exception_ptr) while keeping set_stopped, and try_catch_exception_ptr is the right idiom here.

One note on wording: the description says you implemented "the tag_invoke overload for ex::bulk", but the code customizes through a scheduler query. Worth correcting so future readers look in the right place.

My main comment is about the tests.

All five construct distributed_scheduler{hpx::find_here()}, so every one runs against the local locality, and the implementation is a local for loop regardless. The CMake entry asks for LOCALITIES 2 but the second locality is never used. What the suite actually proves is that the CPO is intercepted and that a local loop runs the right number of times. It does not exercise anything distributed.

That would be fine for a stub, except the file comment opens with "executes data-parallel work on a remote HPX locality via the distributed_scheduler" and the "Current Status" note saying it is a local fallback sits about twenty lines further down. PR descriptions get forgotten, doc comments do not, so I would move the stub caveat to the first paragraph of the file comment. Someone reading the header a month from now should not have to reach the bottom to learn the parcelport dispatch is not there yet.

Test 4 is weaker than it looks:

catch (std::runtime_error const& e) { caught_exception = true; ... }
catch (...)                         { caught_exception = true; }
HPX_TEST(caught_exception);

The catch-all means any exception passes, including one thrown before the bulk loop ever runs. If the loop silently stopped working, this test would still go green. I would drop the catch-all, or set a separate flag inside it and assert that flag stayed false, so the test can only pass for the reason it claims.

Smaller things:

distributed_scheduler.hpp no longer stands on its own. query() returns detail::distributed_bulk_sender<...>, which is only forward declared there, with the definition in distributed_bulk_sender.hpp, which includes it back. That resolves through the generated umbrella header, but anyone including only distributed_scheduler.hpp and calling bulk gets an incomplete type. Worth a comment at the forward declaration saying which header completes it.

P2300 allows bulk to invoke f concurrently, and this runs strictly sequentially. Harmless today, but the remote version will change observable behaviour for functions that are not thread safe, so it is worth writing down now while the reason is fresh.

There is no constraint that f is invocable with (index, Args&...), so a wrong arity shows up as a deep template error instead of a clear one.

Last, a process point rather than a code one: the description says "This is a Draft PR" but the PR is not marked as draft, and Fixes # is empty. Reviewers seeing a normal open PR will hold a 245 line header to merge standards, which is probably not the feedback you are after at this stage.

@iemAnshuman Thanks! I've updated the PR description to use the correct query terminology and marked this as a Draft. I also applied all your code feedback: the stub warning is now at the very top of the file, I added static_assert checks for invocability to catch arity errors cleanly, and added the concurrency warning to the local loop. (The exception catch-all test was also fixed in a recent push!). Let me know if the baseline looks solid to start building the parcelport dispatch.

[iemAnshuman]

@shivansh023023 Had a read through this. The part that is easy to get wrong is right: bulk_t in libs/core/execution/include/hpx/execution/algorithms/bulk.hpp:223 dispatches on has_query_v<Scheduler, bulk_t, Sender, Shape const&, F&&>, and your distributed_scheduler::query signature matches it exactly, so the interception works for the right reason rather than by accident. Completion signatures look correct too, adding set_error_t(exception_ptr) while keeping set_stopped, and try_catch_exception_ptr is the right idiom here.
One note on wording: the description says you implemented "the tag_invoke overload for ex::bulk", but the code customizes through a scheduler query. Worth correcting so future readers look in the right place.
My main comment is about the tests.
All five construct distributed_scheduler{hpx::find_here()}, so every one runs against the local locality, and the implementation is a local for loop regardless. The CMake entry asks for LOCALITIES 2 but the second locality is never used. What the suite actually proves is that the CPO is intercepted and that a local loop runs the right number of times. It does not exercise anything distributed.
That would be fine for a stub, except the file comment opens with "executes data-parallel work on a remote HPX locality via the distributed_scheduler" and the "Current Status" note saying it is a local fallback sits about twenty lines further down. PR descriptions get forgotten, doc comments do not, so I would move the stub caveat to the first paragraph of the file comment. Someone reading the header a month from now should not have to reach the bottom to learn the parcelport dispatch is not there yet.
Test 4 is weaker than it looks:

catch (std::runtime_error const& e) { caught_exception = true; ... }
catch (...)                         { caught_exception = true; }
HPX_TEST(caught_exception);

The catch-all means any exception passes, including one thrown before the bulk loop ever runs. If the loop silently stopped working, this test would still go green. I would drop the catch-all, or set a separate flag inside it and assert that flag stayed false, so the test can only pass for the reason it claims.
Smaller things:
distributed_scheduler.hpp no longer stands on its own. query() returns detail::distributed_bulk_sender<...>, which is only forward declared there, with the definition in distributed_bulk_sender.hpp, which includes it back. That resolves through the generated umbrella header, but anyone including only distributed_scheduler.hpp and calling bulk gets an incomplete type. Worth a comment at the forward declaration saying which header completes it.
P2300 allows bulk to invoke f concurrently, and this runs strictly sequentially. Harmless today, but the remote version will change observable behaviour for functions that are not thread safe, so it is worth writing down now while the reason is fresh.
There is no constraint that f is invocable with (index, Args&...), so a wrong arity shows up as a deep template error instead of a clear one.
Last, a process point rather than a code one: the description says "This is a Draft PR" but the PR is not marked as draft, and Fixes # is empty. Reviewers seeing a normal open PR will hold a 245 line header to merge standards, which is probably not the feedback you are after at this stage.

@iemAnshuman Thanks! I've updated the PR description to use the correct query terminology and marked this as a Draft. I also applied all your code feedback: the stub warning is now at the very top of the file, I added static_assert checks for invocability to catch arity errors cleanly, and added the concurrency warning to the local loop. (The exception catch-all test was also fixed in a recent push!). Let me know if the baseline looks solid to start building the parcelport dispatch.

baseline looks good!

One thing did not make it into the push. The catch-all is still there:

catch (...)
{
caught_exception = true;
}

Before the dispatch work, I would point one test at a remote locality. Every test still uses find_here(), so once the local loop is gone you will have nothing that can tell you whether real dispatch works. LOCALITIES 2 is already in the CMake entry.

[shivansh023023]

@iemAnshuman I've fixed it to explicitly fail the test now. I also updated test_bulk_integral_shape to target hpx::find_remote_localities()[0] instead of the local node, so it's ready to actually test the network dispatch once the stub is replaced.

[shivansh023023]

@hkaiser @iemAnshuman I got the remote network dispatch working for distributed_bulk. Since HPX_DEFINE_PLAIN_ACTION struggles with the variadic Ts... arguments, I ended up packing the upstream values into an hpx::tuple and manually defining the action struct using hpx::actions::make_action_t. Inside set_value, it now checks target_locality_ and either executes directly or dispatches over the network via hpx::async. Tested it locally with --hpx:localities=2 and it passes cleanly. Let me know if this architecture looks good to you!

[iemAnshuman]

@shivansh023023 i would not move the whole receiver yet. The errors are all in 'get_env': its 'auto' return type forces Clang to instantiate the body while 'operation_state' is incomplete

give 'get_env' an explicit return type and, if needed, define it after 'operation_state' is complete. The other cpo can remain inline. Moving the receiver while it still holds an 'operation_state&' would not break the cycle.