From 982dc32032f2f665ca5f35fe5024986c27f4a7b7 Mon Sep 17 00:00:00 2001 From: Thomas James Yurek Date: Thu, 12 Dec 2024 01:51:13 -0800 Subject: [PATCH 1/2] that's it? --- ipa-core/src/query/runner/hybrid.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/ipa-core/src/query/runner/hybrid.rs b/ipa-core/src/query/runner/hybrid.rs index 7ed56c69b..7e87701a3 100644 --- a/ipa-core/src/query/runner/hybrid.rs +++ b/ipa-core/src/query/runner/hybrid.rs @@ -46,6 +46,7 @@ use crate::{ replicated::semi_honest::AdditiveShare as Replicated, BitDecomposed, TransposeFrom, Vectorizable, }, + seq_join::seq_join, sharding::{ShardConfiguration, Sharded}, }; @@ -130,10 +131,12 @@ where })) }) .try_flatten() - .take(sz); + .take(sz) + .map(|v| async move { v }); + let (decrypted_reports, resharded_tags) = reshard_aad( ctx.narrow(&HybridStep::ReshardByTag), - stream, + seq_join(ctx.active_work(), stream), |ctx, _, tag| tag.shard_picker(ctx.shard_count()), ) .await?; From 1f72fb40a405afe3a19983913875ac9a9c290715 Mon Sep 17 00:00:00 2001 From: Thomas James Yurek Date: Thu, 12 Dec 2024 09:36:24 -0800 Subject: [PATCH 2/2] fix text --- ipa-core/src/query/runner/hybrid.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ipa-core/src/query/runner/hybrid.rs b/ipa-core/src/query/runner/hybrid.rs index 7e87701a3..6afaf3f45 100644 --- a/ipa-core/src/query/runner/hybrid.rs +++ b/ipa-core/src/query/runner/hybrid.rs @@ -347,7 +347,7 @@ mod tests { } // cannot test for Err directly because join3v calls unwrap. This should be sufficient. - #[tokio::test] + #[tokio::test(flavor = "multi_thread", worker_threads = 3)] #[should_panic(expected = "DuplicateBytes")] async fn duplicate_encrypted_hybrid_reports() { const SHARDS: usize = 2;