Project import generated by Copybara.

GitOrigin-RevId: 9295f8ea2339edb71073695ed4fb3fded2f48c60

mediapipe/framework/calculator.proto

@@ -194,9 +194,8 @@ message ProfilerConfig {
   // to be appended to the TraceBuffer.
   int64 trace_log_margin_usec = 12;
 
-  // False specifies an event for each calculator invocation.
-  // True specifies a separate event for each start and finish time.
-  bool trace_log_duration_events = 13;
+  // Deprecated, replaced by trace_log_instant_events.
+  bool trace_log_duration_events = 13 [deprecated = true];
 
   // The number of trace log intervals per file. The total log duration is:
   // trace_log_interval_usec * trace_log_file_count * trace_log_interval_count.
@@ -209,6 +208,10 @@ message ProfilerConfig {
 
   // If true, tracer timing events are recorded and reported.
   bool trace_enabled = 16;
+
+  // False specifies an event for each calculator invocation.
+  // True specifies a separate event for each start and finish time.
+  bool trace_log_instant_events = 17;
 }
 
 // Describes the topology and function of a MediaPipe Graph.  The graph of

mediapipe/framework/calculator_graph_bounds_test.cc

@@ -1375,5 +1375,132 @@ TEST(CalculatorGraphBoundsTest, ProcessTimestampBounds_Passthrough) {
   MP_ASSERT_OK(graph.WaitUntilDone());
 }
 
+// A Calculator that sends a timestamp bound for every other input.
+class OccasionalBoundCalculator : public CalculatorBase {
+ public:
+  static ::mediapipe::Status GetContract(CalculatorContract* cc) {
+    cc->Inputs().Index(0).Set<int>();
+    cc->Outputs().Index(0).SetSameAs(&cc->Inputs().Index(0));
+    return ::mediapipe::OkStatus();
+  }
+
+  ::mediapipe::Status Process(CalculatorContext* cc) final {
+    absl::SleepFor(absl::Milliseconds(1));
+    if (cc->InputTimestamp().Value() % 20 == 0) {
+      Timestamp bound = cc->InputTimestamp().NextAllowedInStream();
+      cc->Outputs().Index(0).SetNextTimestampBound(
+          std::max(bound, cc->Outputs().Index(0).NextTimestampBound()));
+    }
+    return ::mediapipe::OkStatus();
+  }
+};
+REGISTER_CALCULATOR(OccasionalBoundCalculator);
+
+// This test fails without the fix in CL/324708313, because
+// PropagateUpdatesToMirrors is called with decreasing next_timestamp_bound,
+// because each parallel thread in-flight computes next_timestamp_bound using
+// a separate OutputStreamShard::NextTimestampBound.
+TEST(CalculatorGraphBoundsTest, MaxInFlightWithOccasionalBound) {
+  // OccasionalCalculator runs on parallel threads and sends ts occasionally.
+  std::string config_str = R"(
+            input_stream: "input_0"
+            node {
+              calculator: "OccasionalBoundCalculator"
+              input_stream: "input_0"
+              output_stream: "output_0"
+              max_in_flight: 5
+            }
+            num_threads: 4
+          )";
+  CalculatorGraphConfig config =
+      ::mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(config_str);
+  CalculatorGraph graph;
+  std::vector<Packet> output_0_packets;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.ObserveOutputStream("output_0", [&](const Packet& p) {
+    output_0_packets.push_back(p);
+    return ::mediapipe::OkStatus();
+  }));
+  MP_ASSERT_OK(graph.StartRun({}));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+
+  // Send in packets.
+  for (int i = 0; i < 9; ++i) {
+    const int ts = 10 + i * 10;
+    Packet p = MakePacket<int>(i).At(Timestamp(ts));
+    MP_ASSERT_OK(graph.AddPacketToInputStream("input_0", p));
+  }
+
+  // Only bounds arrive.
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_EQ(output_0_packets.size(), 0);
+
+  // Shutdown the graph.
+  MP_ASSERT_OK(graph.CloseAllPacketSources());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
+// A Calculator that uses both SetTimestampOffset and SetNextTimestampBound.
+class OffsetAndBoundCalculator : public CalculatorBase {
+ public:
+  static ::mediapipe::Status GetContract(CalculatorContract* cc) {
+    cc->Inputs().Index(0).Set<int>();
+    cc->Outputs().Index(0).SetSameAs(&cc->Inputs().Index(0));
+    return ::mediapipe::OkStatus();
+  }
+  ::mediapipe::Status Open(CalculatorContext* cc) final {
+    cc->SetOffset(0);
+    return ::mediapipe::OkStatus();
+  }
+  ::mediapipe::Status Process(CalculatorContext* cc) final {
+    if (cc->InputTimestamp().Value() % 20 == 0) {
+      cc->Outputs().Index(0).SetNextTimestampBound(Timestamp(10000));
+    }
+    return ::mediapipe::OkStatus();
+  }
+};
+REGISTER_CALCULATOR(OffsetAndBoundCalculator);
+
+// This test shows that the bound defined by SetOffset is ignored
+// if it is superseded by SetNextTimestampBound.
+TEST(CalculatorGraphBoundsTest, OffsetAndBound) {
+  // OffsetAndBoundCalculator runs on parallel threads and sends ts
+  // occasionally.
+  std::string config_str = R"(
+            input_stream: "input_0"
+            node {
+              calculator: "OffsetAndBoundCalculator"
+              input_stream: "input_0"
+              output_stream: "output_0"
+            }
+          )";
+  CalculatorGraphConfig config =
+      ::mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(config_str);
+  CalculatorGraph graph;
+  std::vector<Packet> output_0_packets;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.ObserveOutputStream("output_0", [&](const Packet& p) {
+    output_0_packets.push_back(p);
+    return ::mediapipe::OkStatus();
+  }));
+  MP_ASSERT_OK(graph.StartRun({}));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+
+  // Send in packets.
+  for (int i = 0; i < 9; ++i) {
+    const int ts = 10 + i * 10;
+    Packet p = MakePacket<int>(i).At(Timestamp(ts));
+    MP_ASSERT_OK(graph.AddPacketToInputStream("input_0", p));
+  }
+
+  // Only bounds arrive.
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+  EXPECT_EQ(output_0_packets.size(), 0);
+
+  // Shutdown the graph.
+  MP_ASSERT_OK(graph.CloseAllPacketSources());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
 }  // namespace
 }  // namespace mediapipe

mediapipe/framework/output_stream_manager.cc

@@ -118,10 +118,9 @@ Timestamp OutputStreamManager::ComputeOutputTimestampBound(
   // Note that "MaxOutputTimestamp()" must consider both output packet
   // timetstamp and SetNextTimestampBound values.
   // See the timestamp mapping section in go/mediapipe-bounds for details.
-  Timestamp new_bound = output_stream_shard.NextTimestampBound();
+  Timestamp input_bound;
   if (output_stream_spec_.offset_enabled &&
       input_timestamp != Timestamp::Unstarted()) {
-    Timestamp input_bound;
     if (input_timestamp == Timestamp::PreStream()) {
       // Timestamp::PreStream() is a special value that we shouldn't apply any
       // offset.
@@ -144,9 +143,16 @@ Timestamp OutputStreamManager::ComputeOutputTimestampBound(
       input_bound =
           input_timestamp.NextAllowedInStream() + output_stream_spec_.offset;
     }
+  }
+  Timestamp new_bound;
+  // The input_bound if it is greater than the shard bound.
+  if (input_bound > output_stream_shard.NextTimestampBound()) {
     new_bound = std::max(new_bound, input_bound);
   }
-
+  // The bound defined by SetNextTimestampBound.
+  new_bound =
+      std::max(new_bound, output_stream_shard.updated_next_timestamp_bound_);
+  // The bound defined by added packets.
   if (!output_stream_shard.IsEmpty()) {
     new_bound = std::max(
         new_bound,
@@ -160,8 +166,10 @@ void OutputStreamManager::PropagateUpdatesToMirrors(
     Timestamp next_timestamp_bound, OutputStreamShard* output_stream_shard) {
   CHECK(output_stream_shard);
   {
-    absl::MutexLock lock(&stream_mutex_);
-    next_timestamp_bound_ = next_timestamp_bound;
+    if (next_timestamp_bound != Timestamp::Unset()) {
+      absl::MutexLock lock(&stream_mutex_);
+      next_timestamp_bound_ = next_timestamp_bound;
+    }
   }
   std::list<Packet>* packets_to_propagate = output_stream_shard->OutputQueue();
   VLOG(3) << "Output stream: " << Name()
@@ -170,9 +178,10 @@ void OutputStreamManager::PropagateUpdatesToMirrors(
           << " next timestamp: " << next_timestamp_bound;
   bool add_packets = !packets_to_propagate->empty();
   bool set_bound =
-      !add_packets ||
-      packets_to_propagate->back().Timestamp().NextAllowedInStream() !=
-          next_timestamp_bound;
+      (next_timestamp_bound != Timestamp::Unset()) &&
+      (!add_packets ||
+       packets_to_propagate->back().Timestamp().NextAllowedInStream() !=
+           next_timestamp_bound);
   int mirror_count = mirrors_.size();
   for (int idx = 0; idx < mirror_count; ++idx) {
     const Mirror& mirror = mirrors_[idx];

mediapipe/framework/output_stream_manager_test.cc

@@ -123,14 +123,12 @@ TEST_F(OutputStreamManagerTest, Init) {}
 
 TEST_F(OutputStreamManagerTest, ComputeOutputTimestampBoundWithoutOffset) {
   Timestamp input_timestamp = Timestamp(0);
-  Timestamp stream_previous_timestamp_bound =
-      output_stream_manager_->NextTimestampBound();
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
   // If the offset isn't enabled, the input timestamp and the output bound are
   // not related. Since the output stream shard is empty, the output bound is
-  // still equal to the previous timestamp bound of the stream.
-  EXPECT_EQ(stream_previous_timestamp_bound, output_bound);
+  // still Timestamp::Unset().
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_shard_.AddPacket(
       MakePacket<std::string>("Packet 1").At(Timestamp(10)));
@@ -253,11 +251,10 @@ TEST_F(OutputStreamManagerTest, ComputeOutputTimestampBoundWithNegativeOffset) {
 TEST_F(OutputStreamManagerTest,
        ComputeOutputTimestampBoundWithoutOffsetAfterOpenNode) {
   Timestamp input_timestamp = Timestamp::Unstarted();
-  // If the OutputStreamShard is empty, the output_bound is
-  // Timestamp::PreStream().
+  // If the OutputStreamShard is empty, the output_bound is Timestamp::Unset().
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
-  EXPECT_EQ(Timestamp::PreStream(), output_bound);
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_shard_.AddPacket(
       MakePacket<std::string>("Packet 1").At(Timestamp(20)));
@@ -277,10 +274,10 @@ TEST_F(OutputStreamManagerTest,
   output_stream_shard_.SetOffset(0);
   Timestamp input_timestamp = Timestamp::Unstarted();
   // If the OutputStreamShard is empty, the output_bound is always
-  // Timestamp::PreStream() regardless of the offset.
+  // Timestamp::Unset() regardless of the offset.
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
-  EXPECT_EQ(Timestamp::PreStream(), output_bound);
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_shard_.AddPacket(
       MakePacket<std::string>("Packet 1").At(Timestamp(20)));
@@ -298,13 +295,11 @@ TEST_F(OutputStreamManagerTest,
 TEST_F(OutputStreamManagerTest,
        ComputeOutputTimestampBoundWithoutOffsetForPreStream) {
   Timestamp input_timestamp = Timestamp::PreStream();
-  Timestamp stream_previous_timestamp_bound =
-      output_stream_manager_->NextTimestampBound();
   // If the OutputStreamShard is empty, the output bound is equal to the
-  // previous timestamp bound of the stream.
+  // Timestamp::Unset().
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
-  EXPECT_EQ(stream_previous_timestamp_bound, output_bound);
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_shard_.AddPacket(
       MakePacket<std::string>("Packet 1").At(Timestamp(20)));
@@ -370,13 +365,10 @@ TEST_F(OutputStreamManagerTest, AddPacketUnset) {
   ASSERT_EQ(1, errors_.size());
   EXPECT_TRUE(output_stream_shard_.IsEmpty());
 
-  Timestamp stream_previous_timestamp_bound =
-      output_stream_manager_->NextTimestampBound();
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
-  // The output bound is still equal to the previous timestamp bound of the
-  // stream.
-  EXPECT_EQ(stream_previous_timestamp_bound, output_bound);
+  // The output bound is still equal to Timestamp::Unset().
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_manager_->PropagateUpdatesToMirrors(output_bound,
                                                     &output_stream_shard_);
@@ -394,13 +386,10 @@ TEST_F(OutputStreamManagerTest, AddPacketUnstarted) {
   ASSERT_EQ(1, errors_.size());
   EXPECT_TRUE(output_stream_shard_.IsEmpty());
 
-  Timestamp stream_previous_timestamp_bound =
-      output_stream_manager_->NextTimestampBound();
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
-  // The output bound is still equal to the previous timestamp bound of the
-  // stream.
-  EXPECT_EQ(stream_previous_timestamp_bound, output_bound);
+  // The output bound is still equal to Timestamp::Unset().
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_manager_->PropagateUpdatesToMirrors(output_bound,
                                                     &output_stream_shard_);
@@ -419,13 +408,10 @@ TEST_F(OutputStreamManagerTest, AddPacketOneOverPostStream) {
   ASSERT_EQ(1, errors_.size());
   EXPECT_TRUE(output_stream_shard_.IsEmpty());
 
-  Timestamp stream_previous_timestamp_bound =
-      output_stream_manager_->NextTimestampBound();
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
-  // The output bound is still equal to the previous timestamp bound of the
-  // stream.
-  EXPECT_EQ(stream_previous_timestamp_bound, output_bound);
+  // The output bound is still equal to Timestamp::Unset().
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_manager_->PropagateUpdatesToMirrors(output_bound,
                                                     &output_stream_shard_);
@@ -444,13 +430,10 @@ TEST_F(OutputStreamManagerTest, AddPacketdDone) {
   ASSERT_EQ(1, errors_.size());
   EXPECT_TRUE(output_stream_shard_.IsEmpty());
 
-  Timestamp stream_previous_timestamp_bound =
-      output_stream_manager_->NextTimestampBound();
   Timestamp output_bound = output_stream_manager_->ComputeOutputTimestampBound(
       output_stream_shard_, input_timestamp);
-  // The output bound is still equal to the previous timestamp bound of the
-  // stream.
-  EXPECT_EQ(stream_previous_timestamp_bound, output_bound);
+  // The output bound is still equal to Timestamp::Unset().
+  EXPECT_EQ(Timestamp::Unset(), output_bound);
 
   output_stream_manager_->PropagateUpdatesToMirrors(output_bound,
                                                     &output_stream_shard_);

mediapipe/framework/output_stream_shard.cc

@@ -40,11 +40,13 @@ void OutputStreamShard::SetNextTimestampBound(Timestamp bound) {
     return;
   }
   next_timestamp_bound_ = bound;
+  updated_next_timestamp_bound_ = next_timestamp_bound_;
 }
 
 void OutputStreamShard::Close() {
   closed_ = true;
   next_timestamp_bound_ = Timestamp::Done();
+  updated_next_timestamp_bound_ = next_timestamp_bound_;
 }
 
 bool OutputStreamShard::IsClosed() const { return closed_; }
@@ -122,6 +124,7 @@ Status OutputStreamShard::AddPacketInternal(T&& packet) {
   // Otherwise, moves the packet into output_queue_.
   output_queue_.push_back(std::forward<T>(packet));
   next_timestamp_bound_ = timestamp.NextAllowedInStream();
+  updated_next_timestamp_bound_ = next_timestamp_bound_;
 
   // TODO debug log?
 
@@ -152,6 +155,7 @@ Timestamp OutputStreamShard::LastAddedPacketTimestamp() const {
 void OutputStreamShard::Reset(Timestamp next_timestamp_bound, bool close) {
   output_queue_.clear();
   next_timestamp_bound_ = next_timestamp_bound;
+  updated_next_timestamp_bound_ = Timestamp::Unset();
   closed_ = close;
 }
 

mediapipe/framework/output_stream_shard.h

@@ -117,6 +117,11 @@ class OutputStreamShard : public OutputStream {
   std::list<Packet> output_queue_;
   bool closed_;
   Timestamp next_timestamp_bound_;
+  // Equal to next_timestamp_bound_ only if the bound has been explicitly set
+  // by the calculator.  This is needed for parallel Process() calls,
+  // in order to avoid propagating the initial next_timestamp_bound_, which
+  // does not reflect the output of Process() for preceding timestamps.
+  Timestamp updated_next_timestamp_bound_;
 
   // Accesses OutputStreamShard for profiling.
   friend class GraphProfiler;

mediapipe/framework/profiler/graph_profiler.cc

@@ -596,7 +596,7 @@ void AssignNodeNames(GraphProfile* profile) {
       absl::Microseconds(profiler_config_.trace_log_margin_usec());
   GraphProfile profile;
   GraphTrace* trace = profile.add_graph_trace();
-  if (!profiler_config_.trace_log_duration_events()) {
+  if (!profiler_config_.trace_log_instant_events()) {
     tracer()->GetTrace(previous_log_end_time_, end_time, trace);
   } else {
     tracer()->GetLog(previous_log_end_time_, end_time, trace);