Project import generated by Copybara.

GitOrigin-RevId: bbbbcb4f5174dea33525729ede47c770069157cd

mediapipe/framework/calculator_graph.cc

@@ -427,7 +427,8 @@ absl::Status CalculatorGraph::Initialize(
     const std::map<std::string, Packet>& side_packets) {
   auto validated_graph = absl::make_unique<ValidatedGraphConfig>();
   MP_RETURN_IF_ERROR(validated_graph->Initialize(
-      input_config, /*graph_registry=*/nullptr, &service_manager_));
+      input_config, /*graph_registry=*/nullptr, /*graph_options=*/nullptr,
+      &service_manager_));
   return Initialize(std::move(validated_graph), side_packets);
 }
 

mediapipe/framework/calculator_runner.h

@@ -122,7 +122,7 @@ class CalculatorRunner {
   const StreamContentsSet& Outputs() const { return *outputs_; }
 
   // Returns the access to the output side packets.
-  const PacketSet& OutputSidePackets() { return *output_side_packets_.get(); }
+  const PacketSet& OutputSidePackets() { return *output_side_packets_; }
 
   // Returns a graph counter.
   mediapipe::Counter* GetCounter(const std::string& name);

mediapipe/framework/formats/image.cc

@@ -77,13 +77,6 @@ bool Image::ConvertToGpu() const {
 #else
   // GlCalculatorHelperImpl::MakeGlTextureBuffer (CreateSourceTexture)
   auto buffer = mediapipe::GlTextureBuffer::Create(*image_frame_);
-  glBindTexture(GL_TEXTURE_2D, buffer->name());
-  // See GlCalculatorHelperImpl::SetStandardTextureParams
-  glTexParameteri(buffer->target(), GL_TEXTURE_MIN_FILTER, GL_LINEAR);
-  glTexParameteri(buffer->target(), GL_TEXTURE_MAG_FILTER, GL_LINEAR);
-  glTexParameteri(buffer->target(), GL_TEXTURE_WRAP_S, GL_CLAMP_TO_EDGE);
-  glTexParameteri(buffer->target(), GL_TEXTURE_WRAP_T, GL_CLAMP_TO_EDGE);
-  glBindTexture(GL_TEXTURE_2D, 0);
   glFlush();
   gpu_buffer_ = mediapipe::GpuBuffer(std::move(buffer));
 #endif  //  MEDIAPIPE_GPU_BUFFER_USE_CV_PIXEL_BUFFER

mediapipe/framework/stream_handler/BUILD

@@ -244,6 +244,8 @@ cc_test(
     srcs = ["mux_input_stream_handler_test.cc"],
     deps = [
         ":mux_input_stream_handler",
+        "//mediapipe/calculators/core:gate_calculator",
+        "//mediapipe/calculators/core:make_pair_calculator",
         "//mediapipe/calculators/core:mux_calculator",
         "//mediapipe/calculators/core:pass_through_calculator",
         "//mediapipe/calculators/core:round_robin_demux_calculator",

mediapipe/framework/stream_handler/mux_input_stream_handler.cc

@@ -75,13 +75,30 @@ class MuxInputStreamHandler : public InputStreamHandler {
     int control_value = control_packet.Get<int>();
     CHECK_LE(0, control_value);
     CHECK_LT(control_value, input_stream_managers_.NumEntries() - 1);
-
     const auto& data_stream = input_stream_managers_.Get(
         input_stream_managers_.BeginId() + control_value);
+
+    // Data stream may contain some outdated packets which failed to be popped
+    // out during "FillInputSet". (This handler doesn't sync input streams,
+    // hence "FillInputSet" can be triggerred before every input stream is
+    // filled with packets corresponding to the same timestamp.)
+    data_stream->ErasePacketsEarlierThan(*min_stream_timestamp);
     Timestamp stream_timestamp = data_stream->MinTimestampOrBound(&empty);
     if (empty) {
-      CHECK_LE(stream_timestamp, *min_stream_timestamp);
-      return NodeReadiness::kNotReady;
+      if (stream_timestamp <= *min_stream_timestamp) {
+        // "data_stream" didn't receive a packet corresponding to the current
+        // "control_stream" packet yet.
+        return NodeReadiness::kNotReady;
+      }
+      // "data_stream" timestamp bound update detected.
+      return NodeReadiness::kReadyForProcess;
+    }
+    if (stream_timestamp > *min_stream_timestamp) {
+      // The earliest packet "data_stream" holds corresponds to a control packet
+      // yet to arrive, which means there won't be a "data_stream" packet
+      // corresponding to the current "control_stream" packet, which should be
+      // indicated as timestamp boun update.
+      return NodeReadiness::kReadyForProcess;
     }
     CHECK_EQ(stream_timestamp, *min_stream_timestamp);
     return NodeReadiness::kReadyForProcess;

mediapipe/framework/stream_handler/mux_input_stream_handler_test.cc

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "absl/status/status.h"
 #include "mediapipe/framework/calculator_framework.h"
 #include "mediapipe/framework/port/gmock.h"
 #include "mediapipe/framework/port/gtest.h"
@@ -19,9 +20,10 @@
 #include "mediapipe/framework/port/status_matchers.h"
 
 namespace mediapipe {
-
 namespace {
 
+using ::testing::ElementsAre;
+
 // A regression test for b/31620439. MuxInputStreamHandler's accesses to the
 // control and data streams should be atomic so that it has a consistent view
 // of the two streams. None of the CHECKs in the GetNodeReadiness() method of
@@ -87,5 +89,561 @@ TEST(MuxInputStreamHandlerTest, AtomicAccessToControlAndDataStreams) {
   MP_ASSERT_OK(graph.WaitUntilDone());
 }
 
+MATCHER_P2(IntPacket, value, ts, "") {
+  return arg.template Get<int>() == value && arg.Timestamp() == ts;
+}
+
+struct GateAndMuxGraphInput {
+  int input0;
+  int input1;
+  int input2;
+  int select;
+  bool allow0;
+  bool allow1;
+  bool allow2;
+  Timestamp at;
+};
+
+constexpr char kGateAndMuxGraph[] = R"pb(
+  input_stream: "input0"
+  input_stream: "input1"
+  input_stream: "input2"
+  input_stream: "select"
+  input_stream: "allow0"
+  input_stream: "allow1"
+  input_stream: "allow2"
+  node {
+    calculator: "GateCalculator"
+    input_stream: "ALLOW:allow0"
+    input_stream: "input0"
+    output_stream: "output0"
+  }
+  node {
+    calculator: "GateCalculator"
+    input_stream: "ALLOW:allow1"
+    input_stream: "input1"
+    output_stream: "output1"
+  }
+  node {
+    calculator: "GateCalculator"
+    input_stream: "ALLOW:allow2"
+    input_stream: "input2"
+    output_stream: "output2"
+  }
+  node {
+    calculator: "MuxCalculator"
+    input_stream: "INPUT:0:output0"
+    input_stream: "INPUT:1:output1"
+    input_stream: "INPUT:2:output2"
+    input_stream: "SELECT:select"
+    output_stream: "OUTPUT:output"
+    input_stream_handler { input_stream_handler: "MuxInputStreamHandler" }
+  })pb";
+
+absl::Status SendInput(GateAndMuxGraphInput in, CalculatorGraph& graph) {
+  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
+      "input0", MakePacket<int>(in.input0).At(in.at)));
+  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
+      "input1", MakePacket<int>(in.input1).At(in.at)));
+  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
+      "input2", MakePacket<int>(in.input2).At(in.at)));
+  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(in.select).At(in.at)));
+  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
+      "allow0", MakePacket<bool>(in.allow0).At(in.at)));
+  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
+      "allow1", MakePacket<bool>(in.allow1).At(in.at)));
+  MP_RETURN_IF_ERROR(graph.AddPacketToInputStream(
+      "allow2", MakePacket<bool>(in.allow2).At(in.at)));
+  return graph.WaitUntilIdle();
+}
+
+TEST(MuxInputStreamHandlerTest, BasicMuxing) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(kGateAndMuxGraph);
+  std::vector<Packet> output_packets;
+  tool::AddVectorSink("output", &config, &output_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 0,
+                          .allow0 = true,
+                          .allow1 = false,
+                          .allow2 = false,
+                          .at = Timestamp(1)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(1000, Timestamp(1))));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 1,
+                          .allow0 = false,
+                          .allow1 = true,
+                          .allow2 = false,
+                          .at = Timestamp(2)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(1000, Timestamp(1)),
+                                          IntPacket(900, Timestamp(2))));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 2,
+                          .allow0 = false,
+                          .allow1 = false,
+                          .allow2 = true,
+                          .at = Timestamp(3)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(1000, Timestamp(1)),
+                                          IntPacket(900, Timestamp(2)),
+                                          IntPacket(800, Timestamp(3))));
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+TEST(MuxInputStreamHandlerTest, MuxingNonEmptyInputs) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(kGateAndMuxGraph);
+  std::vector<Packet> output_packets;
+  tool::AddVectorSink("output", &config, &output_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 0,
+                          .allow0 = true,
+                          .allow1 = true,
+                          .allow2 = true,
+                          .at = Timestamp(1)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(1000, Timestamp(1))));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 1,
+                          .allow0 = true,
+                          .allow1 = true,
+                          .allow2 = true,
+                          .at = Timestamp(2)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(1000, Timestamp(1)),
+                                          IntPacket(900, Timestamp(2))));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 2,
+                          .allow0 = true,
+                          .allow1 = true,
+                          .allow2 = true,
+                          .at = Timestamp(3)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(1000, Timestamp(1)),
+                                          IntPacket(900, Timestamp(2)),
+                                          IntPacket(800, Timestamp(3))));
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+TEST(MuxInputStreamHandlerTest, MuxingAllTimestampBoundUpdates) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(kGateAndMuxGraph);
+  std::vector<Packet> output_packets;
+  tool::AddVectorSink("output", &config, &output_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 0,
+                          .allow0 = false,
+                          .allow1 = false,
+                          .allow2 = false,
+                          .at = Timestamp(1)},
+                         graph));
+  EXPECT_TRUE(output_packets.empty());
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 1,
+                          .allow0 = false,
+                          .allow1 = false,
+                          .allow2 = false,
+                          .at = Timestamp(2)},
+                         graph));
+  EXPECT_TRUE(output_packets.empty());
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 2,
+                          .allow0 = false,
+                          .allow1 = false,
+                          .allow2 = false,
+                          .at = Timestamp(3)},
+                         graph));
+  EXPECT_TRUE(output_packets.empty());
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+TEST(MuxInputStreamHandlerTest, MuxingSlectedTimestampBoundUpdates) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(kGateAndMuxGraph);
+  std::vector<Packet> output_packets;
+  tool::AddVectorSink("output", &config, &output_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 0,
+                          .allow0 = false,
+                          .allow1 = true,
+                          .allow2 = true,
+                          .at = Timestamp(1)},
+                         graph));
+  EXPECT_TRUE(output_packets.empty());
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 1,
+                          .allow0 = true,
+                          .allow1 = false,
+                          .allow2 = true,
+                          .at = Timestamp(2)},
+                         graph));
+  EXPECT_TRUE(output_packets.empty());
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 2,
+                          .allow0 = true,
+                          .allow1 = true,
+                          .allow2 = false,
+                          .at = Timestamp(3)},
+                         graph));
+  EXPECT_TRUE(output_packets.empty());
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+TEST(MuxInputStreamHandlerTest, MuxingSometimesTimestampBoundUpdates) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(kGateAndMuxGraph);
+  std::vector<Packet> output_packets;
+  tool::AddVectorSink("output", &config, &output_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 0,
+                          .allow0 = false,
+                          .allow1 = false,
+                          .allow2 = false,
+                          .at = Timestamp(1)},
+                         graph));
+  EXPECT_TRUE(output_packets.empty());
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 1,
+                          .allow0 = false,
+                          .allow1 = true,
+                          .allow2 = false,
+                          .at = Timestamp(2)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(900, Timestamp(2))));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 2,
+                          .allow0 = true,
+                          .allow1 = true,
+                          .allow2 = false,
+                          .at = Timestamp(3)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(900, Timestamp(2))));
+
+  MP_ASSERT_OK(SendInput({.input0 = 1000,
+                          .input1 = 900,
+                          .input2 = 800,
+                          .select = 2,
+                          .allow0 = true,
+                          .allow1 = true,
+                          .allow2 = true,
+                          .at = Timestamp(4)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(900, Timestamp(2)),
+                                          IntPacket(800, Timestamp(4))));
+
+  MP_ASSERT_OK(SendInput({.input0 = 700,
+                          .input1 = 600,
+                          .input2 = 500,
+                          .select = 0,
+                          .allow0 = true,
+                          .allow1 = false,
+                          .allow2 = false,
+                          .at = Timestamp(5)},
+                         graph));
+  EXPECT_THAT(output_packets, ElementsAre(IntPacket(900, Timestamp(2)),
+                                          IntPacket(800, Timestamp(4)),
+                                          IntPacket(700, Timestamp(5))));
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+MATCHER_P(EmptyPacket, ts, "") {
+  return arg.IsEmpty() && arg.Timestamp() == ts;
+}
+
+MATCHER_P2(Pair, m1, m2, "") {
+  const auto& p = arg.template Get<std::pair<Packet, Packet>>();
+  return testing::Matches(m1)(p.first) && testing::Matches(m2)(p.second);
+}
+
+TEST(MuxInputStreamHandlerTest,
+     TimestampBoundUpdateWhenControlPacketEarlierThanDataPacket) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"pb(
+        input_stream: "input0"
+        input_stream: "input1"
+        input_stream: "select"
+        node {
+          calculator: "MuxCalculator"
+          input_stream: "INPUT:0:input0"
+          input_stream: "INPUT:1:input1"
+          input_stream: "SELECT:select"
+          output_stream: "OUTPUT:output"
+          input_stream_handler { input_stream_handler: "MuxInputStreamHandler" }
+        }
+        node {
+          calculator: "MakePairCalculator"
+          input_stream: "select"
+          input_stream: "output"
+          output_stream: "pair"
+        }
+      )pb");
+  std::vector<Packet> pair_packets;
+  tool::AddVectorSink("pair", &config, &pair_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(0).At(Timestamp(1))));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_TRUE(pair_packets.empty());
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input0", MakePacket<int>(1000).At(Timestamp(2))));
+
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(pair_packets, ElementsAre(Pair(IntPacket(0, Timestamp(1)),
+                                             EmptyPacket(Timestamp(1)))));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input1", MakePacket<int>(900).At(Timestamp(1))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input1", MakePacket<int>(800).At(Timestamp(4))));
+
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(pair_packets, ElementsAre(Pair(IntPacket(0, Timestamp(1)),
+                                             EmptyPacket(Timestamp(1)))));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(0).At(Timestamp(2))));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(
+      pair_packets,
+      ElementsAre(
+          Pair(IntPacket(0, Timestamp(1)), EmptyPacket(Timestamp(1))),
+          Pair(IntPacket(0, Timestamp(2)), IntPacket(1000, Timestamp(2)))));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(1).At(Timestamp(3))));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(
+      pair_packets,
+      ElementsAre(
+          Pair(IntPacket(0, Timestamp(1)), EmptyPacket(Timestamp(1))),
+          Pair(IntPacket(0, Timestamp(2)), IntPacket(1000, Timestamp(2))),
+          Pair(IntPacket(1, Timestamp(3)), EmptyPacket(Timestamp(3)))));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(1).At(Timestamp(4))));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(
+      pair_packets,
+      ElementsAre(
+          Pair(IntPacket(0, Timestamp(1)), EmptyPacket(Timestamp(1))),
+          Pair(IntPacket(0, Timestamp(2)), IntPacket(1000, Timestamp(2))),
+          Pair(IntPacket(1, Timestamp(3)), EmptyPacket(Timestamp(3))),
+          Pair(IntPacket(1, Timestamp(4)), IntPacket(800, Timestamp(4)))));
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+TEST(MuxInputStreamHandlerTest,
+     TimestampBoundUpdateWhenControlPacketEarlierThanDataPacketPacketsAtOnce) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"pb(
+        input_stream: "input0"
+        input_stream: "input1"
+        input_stream: "select"
+        node {
+          calculator: "MuxCalculator"
+          input_stream: "INPUT:0:input0"
+          input_stream: "INPUT:1:input1"
+          input_stream: "SELECT:select"
+          output_stream: "OUTPUT:output"
+          input_stream_handler { input_stream_handler: "MuxInputStreamHandler" }
+        }
+        node {
+          calculator: "MakePairCalculator"
+          input_stream: "select"
+          input_stream: "output"
+          output_stream: "pair"
+        }
+      )pb");
+  std::vector<Packet> pair_packets;
+  tool::AddVectorSink("pair", &config, &pair_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(0).At(Timestamp(1))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input0", MakePacket<int>(1000).At(Timestamp(2))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input1", MakePacket<int>(900).At(Timestamp(1))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input1", MakePacket<int>(800).At(Timestamp(4))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(0).At(Timestamp(2))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(1).At(Timestamp(3))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(1).At(Timestamp(4))));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(
+      pair_packets,
+      ElementsAre(
+          Pair(IntPacket(0, Timestamp(1)), EmptyPacket(Timestamp(1))),
+          Pair(IntPacket(0, Timestamp(2)), IntPacket(1000, Timestamp(2))),
+          Pair(IntPacket(1, Timestamp(3)), EmptyPacket(Timestamp(3))),
+          Pair(IntPacket(1, Timestamp(4)), IntPacket(800, Timestamp(4)))));
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+TEST(MuxInputStreamHandlerTest,
+     TimestampBoundUpdateTriggersTimestampBoundUpdate) {
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"pb(
+        input_stream: "input0"
+        input_stream: "input1"
+        input_stream: "select"
+        input_stream: "allow0"
+        input_stream: "allow1"
+        node {
+          calculator: "GateCalculator"
+          input_stream: "ALLOW:allow0"
+          input_stream: "input0"
+          output_stream: "output0"
+        }
+        node {
+          calculator: "GateCalculator"
+          input_stream: "ALLOW:allow1"
+          input_stream: "input1"
+          output_stream: "output1"
+        }
+        node {
+          calculator: "MuxCalculator"
+          input_stream: "INPUT:0:output0"
+          input_stream: "INPUT:1:output1"
+          input_stream: "SELECT:select"
+          output_stream: "OUTPUT:output"
+          input_stream_handler { input_stream_handler: "MuxInputStreamHandler" }
+        }
+        node {
+          calculator: "MakePairCalculator"
+          input_stream: "select"
+          input_stream: "output"
+          output_stream: "pair"
+        }
+      )pb");
+  std::vector<Packet> pair_packets;
+  tool::AddVectorSink("pair", &config, &pair_packets);
+
+  CalculatorGraph graph;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.StartRun({}));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(0).At(Timestamp(1))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input0", MakePacket<int>(1000).At(Timestamp(1))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "allow0", MakePacket<bool>(false).At(Timestamp(1))));
+
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(pair_packets, ElementsAre(Pair(IntPacket(0, Timestamp(1)),
+                                             EmptyPacket(Timestamp(1)))));
+
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "select", MakePacket<int>(0).At(Timestamp(2))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "input0", MakePacket<int>(900).At(Timestamp(2))));
+  MP_ASSERT_OK(graph.AddPacketToInputStream(
+      "allow0", MakePacket<bool>(true).At(Timestamp(2))));
+
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_THAT(
+      pair_packets,
+      ElementsAre(
+          Pair(IntPacket(0, Timestamp(1)), EmptyPacket(Timestamp(1))),
+          Pair(IntPacket(0, Timestamp(2)), IntPacket(900, Timestamp(2)))));
+
+  MP_ASSERT_OK(graph.CloseAllInputStreams());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
 }  // namespace
 }  // namespace mediapipe

mediapipe/framework/subgraph.h

@@ -56,7 +56,9 @@ class SubgraphContext {
     return options_map_.Get<T>();
   }
 
-  const CalculatorGraphConfig::Node& OriginalNode() { return original_node_; }
+  const CalculatorGraphConfig::Node& OriginalNode() const {
+    return original_node_;
+  }
 
   template <typename T>
   ServiceBinding<T> Service(const GraphService<T>& service) const {

mediapipe/framework/tool/BUILD

@@ -724,6 +724,7 @@ cc_test(
     srcs = ["subgraph_expansion_test.cc"],
     deps = [
         ":node_chain_subgraph_cc_proto",
+        ":node_chain_subgraph_options_lib",
         ":subgraph_expansion",
         "//mediapipe/framework:calculator_cc_proto",
         "//mediapipe/framework:calculator_framework",

mediapipe/framework/tool/proto_util_lite.cc

@@ -23,6 +23,8 @@
 #include "mediapipe/framework/port/ret_check.h"
 #include "mediapipe/framework/type_map.h"
 
+#define RET_CHECK_NO_LOG(cond) RET_CHECK(cond).SetNoLogging()
+
 namespace mediapipe {
 namespace tool {
 
@@ -47,13 +49,13 @@ absl::Status ReadFieldValue(uint32 tag, CodedInputStream* in,
   WireFormatLite::WireType wire_type = WireFormatLite::GetTagWireType(tag);
   if (IsLengthDelimited(wire_type)) {
     uint32 length;
-    RET_CHECK(in->ReadVarint32(&length));
-    RET_CHECK(in->ReadString(result, length));
+    RET_CHECK_NO_LOG(in->ReadVarint32(&length));
+    RET_CHECK_NO_LOG(in->ReadString(result, length));
   } else {
     std::string field_data;
     StringOutputStream sos(&field_data);
     CodedOutputStream cos(&sos);
-    RET_CHECK(WireFormatLite::SkipField(in, tag, &cos));
+    RET_CHECK_NO_LOG(WireFormatLite::SkipField(in, tag, &cos));
     // Skip the tag written by SkipField.
     int tag_size = CodedOutputStream::VarintSize32(tag);
     cos.Trim();
@@ -67,13 +69,13 @@ absl::Status ReadPackedValues(WireFormatLite::WireType wire_type,
                               CodedInputStream* in,
                               std::vector<std::string>* field_values) {
   uint32 data_size;
-  RET_CHECK(in->ReadVarint32(&data_size));
+  RET_CHECK_NO_LOG(in->ReadVarint32(&data_size));
   // fake_tag encodes the wire-type for calls to WireFormatLite::SkipField.
   uint32 fake_tag = WireFormatLite::MakeTag(1, wire_type);
   while (data_size > 0) {
     std::string number;
     MP_RETURN_IF_ERROR(ReadFieldValue(fake_tag, in, &number));
-    RET_CHECK_LE(number.size(), data_size);
+    RET_CHECK_NO_LOG(number.size() <= data_size);
     field_values->push_back(number);
     data_size -= number.size();
   }
@@ -98,7 +100,7 @@ absl::Status GetFieldValues(uint32 field_id, WireFormatLite::WireType wire_type,
         field_values->push_back(value);
       }
     } else {
-      RET_CHECK(WireFormatLite::SkipField(in, tag, out));
+      RET_CHECK_NO_LOG(WireFormatLite::SkipField(in, tag, out));
     }
   }
   return absl::OkStatus();
@@ -157,12 +159,12 @@ absl::Status ProtoUtilLite::ReplaceFieldRange(
   MP_RETURN_IF_ERROR(access.SetMessage(*message));
   std::vector<std::string>& v = *access.mutable_field_values();
   if (!proto_path.empty()) {
-    RET_CHECK(index >= 0 && index < v.size());
+    RET_CHECK_NO_LOG(index >= 0 && index < v.size());
     MP_RETURN_IF_ERROR(ReplaceFieldRange(&v[index], proto_path, length,
                                          field_type, field_values));
   } else {
-    RET_CHECK(index >= 0 && index <= v.size());
-    RET_CHECK(index + length >= 0 && index + length <= v.size());
+    RET_CHECK_NO_LOG(index >= 0 && index <= v.size());
+    RET_CHECK_NO_LOG(index + length >= 0 && index + length <= v.size());
     v.erase(v.begin() + index, v.begin() + index + length);
     v.insert(v.begin() + index, field_values.begin(), field_values.end());
   }
@@ -184,12 +186,12 @@ absl::Status ProtoUtilLite::GetFieldRange(
   MP_RETURN_IF_ERROR(access.SetMessage(message));
   std::vector<std::string>& v = *access.mutable_field_values();
   if (!proto_path.empty()) {
-    RET_CHECK(index >= 0 && index < v.size());
+    RET_CHECK_NO_LOG(index >= 0 && index < v.size());
     MP_RETURN_IF_ERROR(
         GetFieldRange(v[index], proto_path, length, field_type, field_values));
   } else {
-    RET_CHECK(index >= 0 && index <= v.size());
-    RET_CHECK(index + length >= 0 && index + length <= v.size());
+    RET_CHECK_NO_LOG(index >= 0 && index <= v.size());
+    RET_CHECK_NO_LOG(index + length >= 0 && index + length <= v.size());
     field_values->insert(field_values->begin(), v.begin() + index,
                          v.begin() + index + length);
   }

mediapipe/framework/tool/subgraph_expansion.cc

@@ -274,12 +274,14 @@ absl::Status ConnectSubgraphStreams(
 
 absl::Status ExpandSubgraphs(CalculatorGraphConfig* config,
                              const GraphRegistry* graph_registry,
+                             const Subgraph::SubgraphOptions* graph_options,
                              const GraphServiceManager* service_manager) {
   graph_registry =
       graph_registry ? graph_registry : &GraphRegistry::global_graph_registry;
   RET_CHECK(config);
+
   MP_RETURN_IF_ERROR(mediapipe::tool::DefineGraphOptions(
-      CalculatorGraphConfig::Node(), config));
+      graph_options ? *graph_options : CalculatorGraphConfig::Node(), config));
   auto* nodes = config->mutable_node();
   while (1) {
     auto subgraph_nodes_start = std::stable_partition(

mediapipe/framework/tool/subgraph_expansion.h

@@ -72,6 +72,7 @@ absl::Status ConnectSubgraphStreams(
 absl::Status ExpandSubgraphs(
     CalculatorGraphConfig* config,
     const GraphRegistry* graph_registry = nullptr,
+    const Subgraph::SubgraphOptions* graph_options = nullptr,
     const GraphServiceManager* service_manager = nullptr);
 
 // Creates a graph wrapping the provided node and exposing all of its

mediapipe/framework/tool/subgraph_expansion_test.cc

@@ -560,9 +560,111 @@ TEST(SubgraphExpansionTest, GraphServicesUsage) {
   MP_ASSERT_OK(service_manager.SetServiceObject(
       kStringTestService, std::make_shared<std::string>("ExpectedNode")));
   MP_EXPECT_OK(tool::ExpandSubgraphs(&supergraph, /*graph_registry=*/nullptr,
+                                     /*graph_options=*/nullptr,
                                      &service_manager));
   EXPECT_THAT(supergraph, mediapipe::EqualsProto(expected_graph));
 }
 
+// Shows SubgraphOptions consumed by GraphRegistry::CreateByName.
+TEST(SubgraphExpansionTest, SubgraphOptionsUsage) {
+  EXPECT_TRUE(SubgraphRegistry::IsRegistered("NodeChainSubgraph"));
+  GraphRegistry graph_registry;
+
+  // CalculatorGraph::Initialize passes the SubgraphOptions into:
+  // (1) GraphRegistry::CreateByName("NodeChainSubgraph", options)
+  // (2) tool::ExpandSubgraphs(&config, options)
+  auto graph_options =
+      mediapipe::ParseTextProtoOrDie<Subgraph::SubgraphOptions>(R"pb(
+        options {
+          [mediapipe.NodeChainSubgraphOptions.ext] {
+            node_type: "DoubleIntCalculator"
+            chain_length: 3
+          }
+        })pb");
+  SubgraphContext context(&graph_options, /*service_manager=*/nullptr);
+
+  // "NodeChainSubgraph" consumes graph_options only in CreateByName.
+  auto subgraph_status =
+      graph_registry.CreateByName("", "NodeChainSubgraph", &context);
+  MP_ASSERT_OK(subgraph_status);
+  auto subgraph = std::move(subgraph_status).value();
+  MP_ASSERT_OK(
+      tool::ExpandSubgraphs(&subgraph, &graph_registry, &graph_options));
+
+  CalculatorGraphConfig expected_graph =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"pb(
+        node {
+          calculator: "DoubleIntCalculator"
+          input_stream: "stream_0"
+          output_stream: "stream_1"
+        }
+        node {
+          calculator: "DoubleIntCalculator"
+          input_stream: "stream_1"
+          output_stream: "stream_2"
+        }
+        node {
+          calculator: "DoubleIntCalculator"
+          input_stream: "stream_2"
+          output_stream: "stream_3"
+        }
+        input_stream: "INPUT:stream_0"
+        output_stream: "OUTPUT:stream_3"
+      )pb");
+
+  EXPECT_THAT(subgraph, mediapipe::EqualsProto(expected_graph));
+}
+
+// Shows SubgraphOptions consumed by tool::ExpandSubgraphs.
+TEST(SubgraphExpansionTest, SimpleSubgraphOptionsUsage) {
+  EXPECT_TRUE(SubgraphRegistry::IsRegistered("NodeChainSubgraph"));
+  GraphRegistry graph_registry;
+  auto moon_options =
+      mediapipe::ParseTextProtoOrDie<Subgraph::SubgraphOptions>(R"pb(
+        options {
+          [mediapipe.NodeChainSubgraphOptions.ext] {
+            node_type: "DoubleIntCalculator"
+            chain_length: 3
+          }
+        })pb");
+  auto moon_subgraph =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"pb(
+        type: "MoonSubgraph"
+        graph_options: {
+          [type.googleapis.com/mediapipe.NodeChainSubgraphOptions] {}
+        }
+        node: {
+          calculator: "MoonCalculator"
+          node_options: {
+            [type.googleapis.com/mediapipe.NodeChainSubgraphOptions] {}
+          }
+          option_value: "chain_length:options/chain_length"
+        }
+      )pb");
+
+  // The moon_options are copied into the graph_options of moon_subgraph.
+  MP_ASSERT_OK(
+      tool::ExpandSubgraphs(&moon_subgraph, &graph_registry, &moon_options));
+
+  // The field chain_length is copied from moon_options into MoonCalculator.
+  CalculatorGraphConfig expected_graph =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(R"pb(
+        node {
+          calculator: "MoonCalculator"
+          node_options {
+            [type.googleapis.com/mediapipe.NodeChainSubgraphOptions] {
+              chain_length: 3
+            }
+          }
+          option_value: "chain_length:options/chain_length"
+        }
+        type: "MoonSubgraph"
+        graph_options {
+          [type.googleapis.com/mediapipe.NodeChainSubgraphOptions] {}
+        }
+      )pb");
+  EXPECT_THAT(moon_subgraph, mediapipe::EqualsProto(expected_graph));
+}
+
 }  // namespace
 }  // namespace mediapipe

mediapipe/framework/tool/switch_container_test.cc

@@ -150,12 +150,13 @@ void RunTestContainer(CalculatorGraphConfig supergraph,
   const int packet_count = 10;
   // Send int value packets at {10K, 20K, 30K, ..., 100K}.
   for (uint64 t = 1; t <= packet_count; ++t) {
-    MP_EXPECT_OK(graph.AddPacketToInputStream(
-        "foo", MakePacket<int>(t).At(Timestamp(t * 10000))));
     if (send_bounds) {
       MP_EXPECT_OK(graph.AddPacketToInputStream(
           "enable", MakePacket<bool>(true).At(Timestamp(t * 10000))));
+      MP_ASSERT_OK(graph.WaitUntilIdle());
     }
+    MP_EXPECT_OK(graph.AddPacketToInputStream(
+        "foo", MakePacket<int>(t).At(Timestamp(t * 10000))));
     MP_ASSERT_OK(graph.WaitUntilIdle());
     // The inputs are sent to the input stream "foo", they should pass through.
     EXPECT_EQ(out_foo.size(), t);
@@ -175,12 +176,13 @@ void RunTestContainer(CalculatorGraphConfig supergraph,
 
   // Send int value packets at {110K, 120K, ..., 200K}.
   for (uint64 t = 11; t <= packet_count * 2; ++t) {
-    MP_EXPECT_OK(graph.AddPacketToInputStream(
-        "foo", MakePacket<int>(t).At(Timestamp(t * 10000))));
     if (send_bounds) {
       MP_EXPECT_OK(graph.AddPacketToInputStream(
           "enable", MakePacket<bool>(false).At(Timestamp(t * 10000))));
+      MP_ASSERT_OK(graph.WaitUntilIdle());
     }
+    MP_EXPECT_OK(graph.AddPacketToInputStream(
+        "foo", MakePacket<int>(t).At(Timestamp(t * 10000))));
     MP_ASSERT_OK(graph.WaitUntilIdle());
     // The inputs are sent to the input stream "foo", they should pass through.
     EXPECT_EQ(out_foo.size(), t);

mediapipe/framework/validated_graph_config.cc

@@ -143,11 +143,12 @@ absl::Status AddPredefinedExecutorConfigs(CalculatorGraphConfig* graph_config) {
 absl::Status PerformBasicTransforms(
     const CalculatorGraphConfig& input_graph_config,
     const GraphRegistry* graph_registry,
+    const Subgraph::SubgraphOptions* graph_options,
     const GraphServiceManager* service_manager,
     CalculatorGraphConfig* output_graph_config) {
   *output_graph_config = input_graph_config;
   MP_RETURN_IF_ERROR(tool::ExpandSubgraphs(output_graph_config, graph_registry,
-                                           service_manager));
+                                           graph_options, service_manager));
 
   MP_RETURN_IF_ERROR(AddPredefinedExecutorConfigs(output_graph_config));
 
@@ -347,6 +348,7 @@ absl::Status NodeTypeInfo::Initialize(
 absl::Status ValidatedGraphConfig::Initialize(
     const CalculatorGraphConfig& input_config,
     const GraphRegistry* graph_registry,
+    const Subgraph::SubgraphOptions* graph_options,
     const GraphServiceManager* service_manager) {
   RET_CHECK(!initialized_)
       << "ValidatedGraphConfig can be initialized only once.";
@@ -356,8 +358,8 @@ absl::Status ValidatedGraphConfig::Initialize(
           << input_config.DebugString();
 #endif
 
-  MP_RETURN_IF_ERROR(PerformBasicTransforms(input_config, graph_registry,
-                                            service_manager, &config_));
+  MP_RETURN_IF_ERROR(PerformBasicTransforms(
+      input_config, graph_registry, graph_options, service_manager, &config_));
 
   // Initialize the basic node information.
   MP_RETURN_IF_ERROR(InitializeGeneratorInfo());
@@ -431,22 +433,24 @@ absl::Status ValidatedGraphConfig::Initialize(
 }
 
 absl::Status ValidatedGraphConfig::Initialize(
-    const std::string& graph_type, const Subgraph::SubgraphOptions* options,
-    const GraphRegistry* graph_registry,
+    const std::string& graph_type, const GraphRegistry* graph_registry,
+    const Subgraph::SubgraphOptions* graph_options,
     const GraphServiceManager* service_manager) {
   graph_registry =
       graph_registry ? graph_registry : &GraphRegistry::global_graph_registry;
-  SubgraphContext subgraph_context(options, service_manager);
+  SubgraphContext subgraph_context(graph_options, service_manager);
   auto status_or_config =
       graph_registry->CreateByName("", graph_type, &subgraph_context);
   MP_RETURN_IF_ERROR(status_or_config.status());
-  return Initialize(status_or_config.value(), graph_registry, service_manager);
+  return Initialize(status_or_config.value(), graph_registry, graph_options,
+                    service_manager);
 }
 
 absl::Status ValidatedGraphConfig::Initialize(
     const std::vector<CalculatorGraphConfig>& input_configs,
     const std::vector<CalculatorGraphTemplate>& input_templates,
-    const std::string& graph_type, const Subgraph::SubgraphOptions* arguments,
+    const std::string& graph_type,
+    const Subgraph::SubgraphOptions* graph_options,
     const GraphServiceManager* service_manager) {
   GraphRegistry graph_registry;
   for (auto& config : input_configs) {
@@ -455,7 +459,8 @@ absl::Status ValidatedGraphConfig::Initialize(
   for (auto& templ : input_templates) {
     graph_registry.Register(templ.config().type(), templ);
   }
-  return Initialize(graph_type, arguments, &graph_registry, service_manager);
+  return Initialize(graph_type, &graph_registry, graph_options,
+                    service_manager);
 }
 
 absl::Status ValidatedGraphConfig::InitializeCalculatorInfo() {

mediapipe/framework/validated_graph_config.h

@@ -195,18 +195,21 @@ class ValidatedGraphConfig {
   // Initializes the ValidatedGraphConfig.  This function must be called
   // before any other functions.  Subgraphs are specified through the
   // global graph registry or an optional local graph registry.
-  absl::Status Initialize(const CalculatorGraphConfig& input_config,
-                          const GraphRegistry* graph_registry = nullptr,
-                          const GraphServiceManager* service_manager = nullptr);
+  absl::Status Initialize(
+      const CalculatorGraphConfig& input_config,
+      const GraphRegistry* graph_registry = nullptr,
+      const Subgraph::SubgraphOptions* graph_options = nullptr,
+      const GraphServiceManager* service_manager = nullptr);
 
   // Initializes the ValidatedGraphConfig from registered graph and subgraph
   // configs.  Subgraphs are retrieved from the specified graph registry or from
   // the global graph registry.  A subgraph can be instantiated directly by
   // specifying its type in |graph_type|.
-  absl::Status Initialize(const std::string& graph_type,
-                          const Subgraph::SubgraphOptions* options = nullptr,
-                          const GraphRegistry* graph_registry = nullptr,
-                          const GraphServiceManager* service_manager = nullptr);
+  absl::Status Initialize(
+      const std::string& graph_type,
+      const GraphRegistry* graph_registry = nullptr,
+      const Subgraph::SubgraphOptions* graph_options = nullptr,
+      const GraphServiceManager* service_manager = nullptr);
 
   // Initializes the ValidatedGraphConfig from the specified graph and subgraph
   // configs.  Template graph and subgraph configs can be specified through
@@ -218,7 +221,7 @@ class ValidatedGraphConfig {
       const std::vector<CalculatorGraphConfig>& input_configs,
       const std::vector<CalculatorGraphTemplate>& input_templates,
       const std::string& graph_type = "",
-      const Subgraph::SubgraphOptions* arguments = nullptr,
+      const Subgraph::SubgraphOptions* graph_options = nullptr,
       const GraphServiceManager* service_manager = nullptr);
 
   // Returns true if the ValidatedGraphConfig has been initialized.

mediapipe/framework/validated_graph_config_test.cc

@@ -155,6 +155,7 @@ TEST(ValidatedGraphConfigTest, InitializeSubgraphWithServiceCalculatorB) {
         kStringTestService, std::make_shared<std::string>(calculator_name)));
     MP_EXPECT_OK(config.Initialize(graph,
                                    /*graph_registry=*/nullptr,
+                                   /*subgraph_options=*/nullptr,
                                    /*service_manager=*/&service_manager));
     ASSERT_TRUE(config.Initialized());
     EXPECT_THAT(config.Config(), EqualsProto(ExpectedConfigExpandedFromGraph(