Adding Combination Operators `Zip`, `CombineLatest`, `WithLatestFrom` #330

mdemoret-nv · 2023-06-03T00:12:44Z

Description

This adds a few new operators for combining multiple streams into one. For example, the code below illustrates using the Zip operator

auto source1 = std::make_shared<node::TestSource<int>>();
auto source2 = std::make_shared<node::TestSource<float>>();

auto zip = std::make_shared<node::Zip<int, float>>();

auto sink = std::make_shared<node::TestSink<std::tuple<int, float>>>();

mrc::make_edge(*source1, *zip->get_sink<0>());
mrc::make_edge(*source2, *zip->get_sink<1>());
mrc::make_edge(*zip, *sink);

source1->run();
source2->run();

sink->run();

EXPECT_EQ(sink->get_values(),
            (std::vector<std::tuple<int, float>>{
                std::tuple<int, float>{0, 0},
                std::tuple<int, float>{1, 1},
                std::tuple<int, float>{2, 2},
            }));

Checklist

I am familiar with the Contributing Guidelines.
New or existing tests cover these changes.
The documentation is up to date with these changes.

cwharris

Approved, but I'm not familiar enough with most of the MRC aspects yet to give a thorough review.

cwharris · 2023-09-28T19:08:23Z

cpp/mrc/tests/test_edges.cpp

@@ -824,6 +1012,182 @@ TEST_F(TestEdges, CombineLatest)
 source2->run();

 sink->run();
+
+ EXPECT_EQ(sink->get_values(),


This test doesn't account for all 5 pairs/values that could be yielded in a real scenario where the elements are interleaved, it only accounts for the scenario where source1 is run entirely before source2, meaning we don't get anything except the 2's for source1. Is it worth testing more thoroughly than this?

cwharris · 2023-09-28T19:13:10Z

cpp/mrc/tests/test_edges.cpp

+ source1->run();
+
+ // Should throw when pushing last value
+ EXPECT_THROW(source2->run(), exceptions::MrcRuntimeError);


question: Is this going to cause us to throw when we try to shutdown the pipeline? Is that something we want?

cwharris · 2023-09-28T19:14:27Z

cpp/mrc/tests/test_edges.cpp

+
+ sink->run();
+
+ EXPECT_EQ(sink->get_values(),


question: This differs from the previous test where we throw an exception. Why is that? Is it because we treat sink 0 and sink 1 differently in terms of the number of elements they are allowed to accept?

cwharris · 2023-09-28T19:29:11Z

cpp/mrc/tests/test_edges.cpp

+ // Push 2 from each
+ source2->push(2);
+ source1->push(2);
+ source3->push(2);
+
+ // Push 2 from each
+ source2->push(2);
+ source1->push(2);
+ source3->push(2);
+
+ // Push the rest
+ source3->run();
+ source1->run();
+ source2->run();
+
+ sink->run();
+
+ EXPECT_EQ(sink->get_values(),
+ (std::vector<std::tuple<int, float, std::string>>{
+ std::tuple<int, float, std::string>{0, 1, "a"},
+ std::tuple<int, float, std::string>{1, 1, "a"},
+ std::tuple<int, float, std::string>{2, 3, "b"},
+ std::tuple<int, float, std::string>{3, 3, "b"},
+ std::tuple<int, float, std::string>{4, 3, "e"},
+ }));


remark: the only thing I'm fuzzy on is whether 0, 1, a should be in the output, since 0 _ _ was emitted before we had both _ 1 _ and _ _ a.
question: are we suppose to remember and emit for all values of the source observable, or just the ones that occur after we've received at least one from the others?

cwharris · 2023-09-28T20:13:06Z

cpp/mrc/include/mrc/node/operators/with_latest_from.hpp

+ template <size_t N>
+ channel::Status set_upstream_value(NthTypeOf<N, TypesT...> value)
+ {
+ std::unique_lock<decltype(m_mutex)> lock(m_mutex);


Suggested change

std::unique_lock<decltype(m_mutex)> lock(m_mutex);

auto lock = std::unique_lock(m_mutex);

cwharris · 2023-09-28T20:14:12Z

cpp/mrc/include/mrc/node/operators/with_latest_from.hpp

+
+ void edge_complete()
+ {
+ std::unique_lock<decltype(m_mutex)> lock(m_mutex);


Suggested change

std::unique_lock<decltype(m_mutex)> lock(m_mutex);

auto lock = std::unique_lock(m_mutex);

…ng-operators

codecov · 2024-05-01T00:10:38Z

Codecov Report

Attention: Patch coverage is 86.56716% with 18 lines in your changes are missing coverage. Please review.

❗ No coverage uploaded for pull request base (branch-23.07@5f62b46). Click here to learn what that means.
Report is 7 commits behind head on branch-23.07.

❗ Current head 7cbfe8e differs from pull request most recent head 5c52d5c. Consider uploading reports for the commit 5c52d5c to get more accurate results

Additional details and impacted files

@@               Coverage Diff               @@
##             branch-23.07     #330   +/-   ##
===============================================
  Coverage                ?   73.26%           
===============================================
  Files                   ?      386           
  Lines                   ?    13533           
  Branches                ?     1024           
===============================================
  Hits                    ?     9915           
  Misses                  ?     3618           
  Partials                ?        0

Flag	Coverage Δ
cpp	`69.25% <86.56%> (?)`
py	`42.06% <100.00%> (?)`

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
cpp/mrc/include/mrc/edge/edge_channel.hpp	`85.00% <100.00%> (ø)`
.../mrc/include/mrc/node/operators/combine_latest.hpp	`100.00% <100.00%> (ø)`
cpp/mrc/include/mrc/node/sink_channel_owner.hpp	`100.00% <100.00%> (ø)`
cpp/mrc/include/mrc/node/source_channel_owner.hpp	`90.90% <100.00%> (ø)`
cpp/mrc/include/mrc/utils/tuple_utils.hpp	`100.00% <100.00%> (ø)`
cpp/mrc/include/mrc/node/operators/zip.hpp	`98.21% <98.21%> (ø)`
...rc/include/mrc/node/operators/with_latest_from.hpp	`95.65% <95.65%> (ø)`
cpp/mrc/include/mrc/channel/status.hpp	`0.00% <0.00%> (ø)`

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 5f62b46...5c52d5c. Read the comment docs.

mdemoret-nv added 2 commits June 2, 2023 13:21

Fixing CombineLatest and adding Zip operator

5bc9b35

Adding WithLatestFrom

293c481

mdemoret-nv added non-breaking Non-breaking change feature request New feature or request labels Jun 3, 2023

mdemoret-nv requested review from a team as code owners June 3, 2023 00:12

IWYU fixes

32672c9

jarmak-nv assigned mdemoret-nv Jun 5, 2023

drobison00 approved these changes Jun 5, 2023

View reviewed changes

mdemoret-nv added 4 commits June 21, 2023 12:07

Adding copyright header

34d4767

Merge branch 'branch-23.07' into mdd_combining-operators

cfbafd3

Merge branch 'branch-23.07' into mdd_combining-operators

7cbfe8e

In progress adding children to Object classes

0178730

cwharris approved these changes Sep 28, 2023

View reviewed changes

mdemoret-nv requested a review from a team as a code owner May 1, 2024 00:02

Merge remote-tracking branch 'upstream/branch-24.06' into mdd_combini…

5c52d5c

…ng-operators

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding Combination Operators `Zip`, `CombineLatest`, `WithLatestFrom` #330

Adding Combination Operators `Zip`, `CombineLatest`, `WithLatestFrom` #330

mdemoret-nv commented Jun 3, 2023

cwharris left a comment

cwharris Sep 28, 2023

cwharris Sep 28, 2023

cwharris Sep 28, 2023

cwharris Sep 28, 2023

cwharris Sep 28, 2023

cwharris Sep 28, 2023

codecov bot commented May 1, 2024

	std::unique_lock<decltype(m_mutex)> lock(m_mutex);
	auto lock = std::unique_lock(m_mutex);

Adding Combination Operators Zip, CombineLatest, WithLatestFrom #330

Are you sure you want to change the base?

Adding Combination Operators Zip, CombineLatest, WithLatestFrom #330

Conversation

mdemoret-nv commented Jun 3, 2023

Description

Checklist

cwharris left a comment

Choose a reason for hiding this comment

cwharris Sep 28, 2023

Choose a reason for hiding this comment

cwharris Sep 28, 2023

Choose a reason for hiding this comment

cwharris Sep 28, 2023

Choose a reason for hiding this comment

cwharris Sep 28, 2023

Choose a reason for hiding this comment

cwharris Sep 28, 2023

Choose a reason for hiding this comment

cwharris Sep 28, 2023

Choose a reason for hiding this comment

codecov bot commented May 1, 2024

Codecov Report

Adding Combination Operators `Zip`, `CombineLatest`, `WithLatestFrom` #330

Adding Combination Operators `Zip`, `CombineLatest`, `WithLatestFrom` #330