Skip to content

Speed up multihost exchange #5819

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
blp merged 8 commits into from
Mar 25, 2026
Merged

Speed up multihost exchange #5819

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c270207
[nexmark] Remove unused imports.
blp Mar 19, 2026
d0e131c
[dbsp] Tiny code cleanup in try_send_all() callers.
blp Mar 12, 2026
33d32e0
[adapters] Simplify async_samply_profile with tokio::fs::read.
blp Mar 18, 2026
a1baa04
[adapters] Make the host layout available from the Controller.
blp Mar 24, 2026
63a1aa0
Improve support for custom marker ranges in CPU profiles.
blp Mar 9, 2026
3fdae5a
[dbsp] Use new protocol for exchanging data among hosts.
blp Mar 20, 2026
b5df7a4
[adapters] Indicate host ordinal in multihost profiles.
blp Mar 20, 2026
7f2a613
[adapters] Fix adhoc scan implementation for multihost.
blp Mar 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
[adapters] Indicate host ordinal in multihost profiles. 
Signed-off-by: Ben Pfaff <blp@feldera.com>
  • Loading branch information
@blp
blp committed Mar 24, 2026
commit b5df7a4667b87b647ef148392076c34f4ecbe7a0
8 changes: 6 additions & 2 deletions crates/adapters/src/controller.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -944,7 +944,11 @@ impl Controller {
receiver.await.unwrap()
}

pub async fn async_samply_profile(&self, duration: u64) -> Result<Vec<u8>, AnyError> {
pub async fn async_samply_profile(
&self,
duration: u64,
os_cpu: Option<&str>,
) -> Result<Vec<u8>, AnyError> {
#[cfg(not(unix))]
{
anyhow::bail!(
Expand Down Expand Up @@ -1074,7 +1078,7 @@ impl Controller {
let output = match markers.annotate_profile(
&buf,
self.inner.status.pipeline_config.given_name.as_deref(),
None,
os_cpu,
) {
Ok(output) => output,
Err(error) => {
Expand Down
8 changes: 7 additions & 1 deletion crates/adapters/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1763,8 +1763,14 @@ async fn samply_profile(
.unwrap()
.start_profiling(expected_after);

let layout = controller.layout();
let os_cpu = layout
.is_multihost()
.then(|| format!("host {} of {}", layout.local_host_idx(), layout.n_hosts()));
spawn(async move {
let result = controller.async_samply_profile(duration).await;
let result = controller
.async_samply_profile(duration, os_cpu.as_deref())
.await;
state_samply_state
.lock()
.unwrap()
Expand Down