Project import generated by Copybara.

GitOrigin-RevId: 4cee4a2c2317fb190680c17e31ebbb03bb73b71c

mediapipe/framework/BUILD

@@ -947,6 +947,7 @@ cc_library(
         ],
     }),
     visibility = [
+        "//mediapipe/calculators:__subpackages__",
         "//mediapipe/framework:__subpackages__",
         "//mediapipe/framework/port:__pkg__",
         "//mediapipe/util:__subpackages__",

mediapipe/framework/calculator_graph.cc

@@ -330,6 +330,9 @@ CalculatorGraph::~CalculatorGraph() {
 ::mediapipe::Status CalculatorGraph::InitializeDefaultExecutor(
     const ThreadPoolExecutorOptions* default_executor_options,
     bool use_application_thread) {
+#ifdef __EMSCRIPTEN__
+  use_application_thread = true;
+#endif  // __EMSCRIPTEN__
   // If specified, run synchronously on the calling thread.
   if (use_application_thread) {
     use_application_thread_ = true;

mediapipe/framework/deps/file_helpers.cc

@@ -27,7 +27,9 @@
 
 #include "mediapipe/framework/deps/canonical_errors.h"
 #include "mediapipe/framework/deps/file_path.h"
+#include "mediapipe/framework/deps/status.h"
 #include "mediapipe/framework/deps/status_builder.h"
+#include "mediapipe/framework/deps/status_macros.h"
 
 namespace mediapipe {
 namespace file {
@@ -212,7 +214,7 @@ class DirectoryListing {
 ::mediapipe::Status Exists(absl::string_view file_name) {
   struct stat buffer;
   int status;
-  status = stat(file_name.data(), &buffer);
+  status = stat(std::string(file_name).c_str(), &buffer);
   if (status == 0) {
     return ::mediapipe::OkStatus();
   }
@@ -224,5 +226,30 @@ class DirectoryListing {
   }
 }
 
+#ifndef _WIN32
+int mkdir(std::string path) {
+  return ::mkdir(path.c_str(), S_IRWXU | S_IRWXG | S_IRWXO);
+}
+#else
+int mkdir(std::string path) { return _mkdir(path.c_str()); }
+#endif
+
+::mediapipe::Status RecursivelyCreateDir(absl::string_view path) {
+  if (path.empty() || Exists(path).ok()) {
+    return mediapipe::OkStatus();
+  }
+  auto split_path = file::SplitPath(path);
+  MP_RETURN_IF_ERROR(RecursivelyCreateDir(split_path.first));
+  if (mkdir(std::string(path)) != 0) {
+    switch (errno) {
+      case EACCES:
+        return ::mediapipe::PermissionDeniedError("Insufficient permissions.");
+      default:
+        return ::mediapipe::UnavailableError("Failed to create directory.");
+    }
+  }
+  return mediapipe::OkStatus();
+}
+
 }  // namespace file
 }  // namespace mediapipe

mediapipe/framework/deps/file_helpers.h

@@ -36,6 +36,8 @@ namespace file {
 
 ::mediapipe::Status Exists(absl::string_view file_name);
 
+::mediapipe::Status RecursivelyCreateDir(absl::string_view path);
+
 }  // namespace file
 }  // namespace mediapipe
 

mediapipe/framework/deps/threadpool_pthread_impl.cc

@@ -47,7 +47,8 @@ class ThreadPool::WorkerThread {
 ThreadPool::WorkerThread::WorkerThread(ThreadPool* pool,
                                        const std::string& name_prefix)
     : pool_(pool), name_prefix_(name_prefix) {
-  pthread_create(&thread_, nullptr, ThreadBody, this);
+  int res = pthread_create(&thread_, nullptr, ThreadBody, this);
+  CHECK_EQ(res, 0) << "pthread_create failed";
 }
 
 ThreadPool::WorkerThread::~WorkerThread() {}
@@ -59,9 +60,9 @@ void* ThreadPool::WorkerThread::ThreadBody(void* arg) {
   int nice_priority_level =
       thread->pool_->thread_options().nice_priority_level();
   const std::set<int> selected_cpus = thread->pool_->thread_options().cpu_set();
+#if defined(__linux__)
   const std::string name =
       internal::CreateThreadName(thread->name_prefix_, syscall(SYS_gettid));
-#if defined(__linux__)
   if (nice_priority_level != 0) {
     if (nice(nice_priority_level) != -1 || errno == 0) {
       VLOG(1) << "Changed the nice priority level by " << nice_priority_level;
@@ -94,16 +95,19 @@ void* ThreadPool::WorkerThread::ThreadBody(void* arg) {
                << "Failed to set name for thread: " << name;
   }
 #else
+  const std::string name = internal::CreateThreadName(thread->name_prefix_, 0);
   if (nice_priority_level != 0 || !selected_cpus.empty()) {
     LOG(ERROR) << "Thread priority and processor affinity feature aren't "
                   "supported on the current platform.";
   }
+#if __APPLE__
   int error = pthread_setname_np(name.c_str());
   if (error != 0) {
     LOG(ERROR) << "Error : " << strerror(error) << std::endl
                << "Failed to set name for thread: " << name;
   }
-#endif
+#endif  // __APPLE__
+#endif  // __linux__
   thread->pool_->RunWorker();
   return nullptr;
 }
@@ -178,6 +182,12 @@ const ThreadOptions& ThreadPool::thread_options() const {
 
 namespace internal {
 
+// TODO: revise this:
+// - thread_id is not portable
+// - the 16-byte limit is Linux-specific
+// - the std::thread implementation has a copy of this but doesn't use it
+// - why do we even need the thread id in the name? any thread list should show
+//   the id too.
 std::string CreateThreadName(const std::string& prefix, int thread_id) {
   std::string name = absl::StrCat(prefix, "/", thread_id);
   // 16 is the limit allowed by `pthread_setname_np`, including

mediapipe/framework/encode_binary_proto.bzl

@@ -1,3 +1,17 @@
+# Copyright 2019-2020 The MediaPipe Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 """A rule for encoding a text format protocol buffer into binary.
 
 Example usage:

mediapipe/framework/formats/BUILD

@@ -227,3 +227,28 @@ filegroup(
     srcs = glob(["*.proto"]),
     visibility = ["//mediapipe:__subpackages__"],
 )
+
+cc_library(
+    name = "image_frame_pool",
+    srcs = ["image_frame_pool.cc"],
+    hdrs = ["image_frame_pool.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":image_frame",
+        "@com_google_absl//absl/memory",
+        "@com_google_absl//absl/synchronization",
+    ],
+)
+
+cc_test(
+    name = "image_frame_pool_test",
+    size = "small",
+    srcs = ["image_frame_pool_test.cc"],
+    tags = ["linux"],
+    deps = [
+        ":image_frame_pool",
+        "//mediapipe/framework/port:gtest_main",
+        "//mediapipe/framework/port:status",
+        "@com_google_absl//absl/memory",
+    ],
+)

mediapipe/framework/formats/image_frame_pool.cc

@@ -0,0 +1,88 @@
+// Copyright 2019 The MediaPipe Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "mediapipe/framework/formats/image_frame_pool.h"
+
+#include "absl/synchronization/mutex.h"
+
+namespace mediapipe {
+
+ImageFramePool::ImageFramePool(int width, int height,
+                               ImageFormat::Format format, int keep_count)
+    : width_(width),
+      height_(height),
+      format_(format),
+      keep_count_(keep_count) {}
+
+ImageFrameSharedPtr ImageFramePool::GetBuffer() {
+  std::unique_ptr<ImageFrame> buffer;
+
+  {
+    absl::MutexLock lock(&mutex_);
+    if (available_.empty()) {
+      // Fix alignment at 4 for best compatability with OpenGL.
+      buffer = std::make_unique<ImageFrame>(
+          format_, width_, height_, ImageFrame::kGlDefaultAlignmentBoundary);
+      if (!buffer) return nullptr;
+    } else {
+      buffer = std::move(available_.back());
+      available_.pop_back();
+    }
+
+    ++in_use_count_;
+  }
+
+  // Return a shared_ptr with a custom deleter that adds the buffer back
+  // to our available list.
+  std::weak_ptr<ImageFramePool> weak_pool(shared_from_this());
+  return std::shared_ptr<ImageFrame>(buffer.release(),
+                                     [weak_pool](ImageFrame* buf) {
+                                       auto pool = weak_pool.lock();
+                                       if (pool) {
+                                         pool->Return(buf);
+                                       } else {
+                                         delete buf;
+                                       }
+                                     });
+}
+
+std::pair<int, int> ImageFramePool::GetInUseAndAvailableCounts() {
+  absl::MutexLock lock(&mutex_);
+  return {in_use_count_, available_.size()};
+}
+
+void ImageFramePool::Return(ImageFrame* buf) {
+  std::vector<std::unique_ptr<ImageFrame>> trimmed;
+  {
+    absl::MutexLock lock(&mutex_);
+    --in_use_count_;
+    available_.emplace_back(buf);
+    TrimAvailable(&trimmed);
+  }
+  // The trimmed buffers will be released without holding the lock.
+}
+
+void ImageFramePool::TrimAvailable(
+    std::vector<std::unique_ptr<ImageFrame>>* trimmed) {
+  int keep = std::max(keep_count_ - in_use_count_, 0);
+  if (available_.size() > keep) {
+    auto trim_it = std::next(available_.begin(), keep);
+    if (trimmed) {
+      std::move(trim_it, available_.end(), std::back_inserter(*trimmed));
+    }
+    available_.erase(trim_it, available_.end());
+  }
+}
+
+}  // namespace mediapipe

mediapipe/framework/formats/image_frame_pool.h

@@ -0,0 +1,78 @@
+// Copyright 2019 The MediaPipe Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Consider this file an implementation detail. None of this is part of the
+// public API.
+
+#ifndef MEDIAPIPE_FRAMEWORK_FORMATS_IMAGE_FRAME_POOL_H_
+#define MEDIAPIPE_FRAMEWORK_FORMATS_IMAGE_FRAME_POOL_H_
+
+#include <utility>
+#include <vector>
+
+#include "absl/synchronization/mutex.h"
+#include "mediapipe/framework/formats/image_frame.h"
+
+namespace mediapipe {
+
+using ImageFrameSharedPtr = std::shared_ptr<ImageFrame>;
+
+class ImageFramePool : public std::enable_shared_from_this<ImageFramePool> {
+ public:
+  // Creates a pool. This pool will manage buffers of the specified dimensions,
+  // and will keep keep_count buffers around for reuse.
+  // We enforce creation as a shared_ptr so that we can use a weak reference in
+  // the buffers' deleters.
+  static std::shared_ptr<ImageFramePool> Create(int width, int height,
+                                                ImageFormat::Format format,
+                                                int keep_count) {
+    return std::shared_ptr<ImageFramePool>(
+        new ImageFramePool(width, height, format, keep_count));
+  }
+
+  // Obtains a buffers. May either be reused or created anew.
+  ImageFrameSharedPtr GetBuffer();
+
+  int width() const { return width_; }
+  int height() const { return height_; }
+  ImageFormat::Format format() const { return format_; }
+
+  // This method is meant for testing.
+  std::pair<int, int> GetInUseAndAvailableCounts();
+
+ private:
+  ImageFramePool(int width, int height, ImageFormat::Format format,
+                 int keep_count);
+
+  // Return a buffer to the pool.
+  void Return(ImageFrame* buf);
+
+  // If the total number of buffers is greater than keep_count, destroys any
+  // surplus buffers that are no longer in use.
+  void TrimAvailable(std::vector<std::unique_ptr<ImageFrame>>* trimmed)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+
+  const int width_;
+  const int height_;
+  const ImageFormat::Format format_;
+  const int keep_count_;
+
+  absl::Mutex mutex_;
+  int in_use_count_ ABSL_GUARDED_BY(mutex_) = 0;
+  std::vector<std::unique_ptr<ImageFrame>> available_ ABSL_GUARDED_BY(mutex_);
+};
+
+}  // namespace mediapipe
+
+#endif  // MEDIAPIPE_FRAMEWORK_FORMATS_IMAGE_FRAME_POOL_H_

mediapipe/framework/formats/image_frame_pool_test.cc

@@ -0,0 +1,103 @@
+// Copyright 2019 The MediaPipe Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "mediapipe/framework/formats/image_frame_pool.h"
+
+#include "absl/memory/memory.h"
+#include "mediapipe/framework/port/gmock.h"
+#include "mediapipe/framework/port/gtest.h"
+#include "mediapipe/framework/port/status.h"
+
+namespace mediapipe {
+namespace {
+
+using Pair = std::pair<int, int>;
+
+constexpr int kWidth = 300;
+constexpr int kHeight = 200;
+constexpr ImageFormat::Format kFormat = ImageFormat::SRGBA;
+constexpr int kKeepCount = 2;
+
+class ImageFramePoolTest : public ::testing::Test {
+ protected:
+  ImageFramePoolTest() {
+    pool_ = ImageFramePool::Create(kWidth, kHeight, kFormat, kKeepCount);
+  }
+
+  void SetUp() override {}
+
+  std::shared_ptr<ImageFramePool> pool_;
+};
+
+TEST_F(ImageFramePoolTest, GetBuffer) {
+  EXPECT_EQ(Pair(0, 0), pool_->GetInUseAndAvailableCounts());
+  auto buffer = pool_->GetBuffer();
+  EXPECT_EQ(Pair(1, 0), pool_->GetInUseAndAvailableCounts());
+  buffer = nullptr;
+  EXPECT_EQ(Pair(0, 1), pool_->GetInUseAndAvailableCounts());
+  buffer = pool_->GetBuffer();
+  EXPECT_EQ(Pair(1, 0), pool_->GetInUseAndAvailableCounts());
+}
+
+TEST_F(ImageFramePoolTest, GetMoreBuffers) {
+  EXPECT_EQ(Pair(0, 0), pool_->GetInUseAndAvailableCounts());
+  std::vector<ImageFrameSharedPtr> buffers;
+
+  // Create kKeepCount + 1 buffers
+  for (int i = 0; i <= kKeepCount; i++) {
+    buffers.emplace_back(pool_->GetBuffer());
+  }
+  EXPECT_EQ(Pair(kKeepCount + 1, 0), pool_->GetInUseAndAvailableCounts());
+
+  // Delete one
+  buffers.resize(kKeepCount);
+  EXPECT_EQ(Pair(kKeepCount, 0), pool_->GetInUseAndAvailableCounts());
+
+  // Delete all
+  buffers.resize(0);
+  EXPECT_EQ(Pair(0, kKeepCount), pool_->GetInUseAndAvailableCounts());
+
+  // Create one more
+  buffers.emplace_back(pool_->GetBuffer());
+  EXPECT_EQ(Pair(1, kKeepCount - 1), pool_->GetInUseAndAvailableCounts());
+}
+
+TEST_F(ImageFramePoolTest, DeleteNotLast) {
+  EXPECT_EQ(Pair(0, 0), pool_->GetInUseAndAvailableCounts());
+  std::vector<ImageFrameSharedPtr> buffers;
+
+  // Create kKeepCount + 1 buffers
+  for (int i = 0; i <= kKeepCount; i++) {
+    buffers.emplace_back(pool_->GetBuffer());
+  }
+  EXPECT_EQ(Pair(kKeepCount + 1, 0), pool_->GetInUseAndAvailableCounts());
+
+  // Delete second
+  buffers.erase(buffers.begin() + 1);
+  EXPECT_EQ(Pair(kKeepCount, 0), pool_->GetInUseAndAvailableCounts());
+
+  // Delete first
+  buffers.erase(buffers.begin());
+  EXPECT_EQ(Pair(kKeepCount - 1, 1), pool_->GetInUseAndAvailableCounts());
+}
+
+TEST(ImageFrameBufferPoolStaticTest, BufferCanOutlivePool) {
+  auto pool = ImageFramePool::Create(kWidth, kHeight, kFormat, kKeepCount);
+  auto buffer = pool->GetBuffer();
+  pool = nullptr;
+  buffer = nullptr;
+}
+
+}  // anonymous namespace
+}  // namespace mediapipe

mediapipe/framework/mediapipe_register_type.bzl

@@ -0,0 +1,117 @@
+# Copyright 2020 The MediaPipe Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A rule for registering types with mediapipe.
+
+Example usage:
+    mediapipe_proto_library(
+        name = "foo_proto",
+        srcs = ["foo.proto"],
+    )
+
+    load("//mediapipe/framework:mediapipe_register_type.bzl",
+         "mediapipe_register_type")
+
+    # Creates rules "foo_registration"
+    mediapipe_register_type(
+        base_name = "foo"
+        include_headers = ["mediapipe/framework/formats/foo.proto.h""],
+        types = [
+            "::mediapipe::Foo",
+            "::mediapipe::FooList",
+            "::std::vector<::mediapipe::Foo>",
+        ],
+        deps = [":foo_cc_proto"],
+    )
+
+Args
+  base_name: The base name of the target
+             (name + "_registration" will be created).
+  types: A list of C++ classes to register using MEDIAPIPE_REGISTER_TYPE.
+  deps: A list of cc deps.
+  include_headers: A list of header files that must be included.
+"""
+
+load("//mediapipe/framework/tool:build_defs.bzl", "clean_dep")
+
+def _mediapipe_register_type_generate_cc_impl(ctx):
+    """Generate a cc file that registers types with mediapipe."""
+    file_data_template = '''
+{include_headers}
+#include "mediapipe/framework/type_map.h"
+
+{registration_commands}
+'''
+    header_lines = []
+    for header in ctx.attr.include_headers:
+        header_lines.append('#include "{}"'.format(header))
+    registration_lines = []
+    for registration_type in ctx.attr.types:
+        if " " in registration_type:
+            fail(('registration type "{}" should be fully qualified ' +
+                  "and must not include spaces").format(registration_type))
+        registration_lines.append(
+            "#define TEMP_MP_TYPE {}".format(registration_type),
+        )
+        registration_lines.append(
+            ("MEDIAPIPE_REGISTER_TYPE(\n" +
+             "    TEMP_MP_TYPE,\n" +
+             '    "{}",\n'.format(registration_type) +
+             "    nullptr, nullptr);\n"),
+        )
+        registration_lines.append("#undef TEMP_MP_TYPE")
+
+    file_data = file_data_template.format(
+        include_headers = "\n".join(header_lines),
+        registration_commands = "\n".join(registration_lines),
+    )
+
+    ctx.actions.write(ctx.outputs.output, file_data)
+
+mediapipe_register_type_generate_cc = rule(
+    implementation = _mediapipe_register_type_generate_cc_impl,
+    attrs = {
+        "deps": attr.label_list(),
+        "types": attr.string_list(
+            mandatory = True,
+        ),
+        "include_headers": attr.string_list(
+            mandatory = True,
+        ),
+        "output": attr.output(),
+    },
+)
+
+def mediapipe_register_type(
+        base_name,
+        types,
+        deps = [],
+        include_headers = [],
+        visibility = ["//visibility:public"]):
+    mediapipe_register_type_generate_cc(
+        name = base_name + "_registration_cc",
+        types = types,
+        include_headers = include_headers,
+        output = base_name + "_registration.cc",
+    )
+
+    native.cc_library(
+        name = base_name + "_registration",
+        srcs = [base_name + "_registration.cc"],
+        deps = depset(deps + [
+            clean_dep("//mediapipe/framework:type_map"),
+        ]),
+        visibility = visibility,
+        alwayslink = 1,
+    )

mediapipe/framework/port.h

@@ -52,12 +52,12 @@
 
 // Compile time target platform definitions.
 // Example: #if MEDIAPIPE_OPENGL_ES_VERSION >= MEDIAPIPE_OPENGL_ES_31
-#define MEDIAPIPE_OPENGL_ES_UNSUPPORTED 0
 #define MEDIAPIPE_OPENGL_ES_20 200
+#define MEDIAPIPE_OPENGL_ES_30 300
 #define MEDIAPIPE_OPENGL_ES_31 310
 
 #if defined(MEDIAPIPE_DISABLE_GPU)
-#define MEDIAPIPE_OPENGL_ES_VERSION MEDIAPIPE_OPENGL_ES_UNSUPPORTED
+#define MEDIAPIPE_OPENGL_ES_VERSION 0
 #define MEDIAPIPE_METAL_ENABLED 0
 #else
 #if defined(MEDIAPIPE_ANDROID)
@@ -71,11 +71,11 @@
 #define MEDIAPIPE_OPENGL_ES_VERSION MEDIAPIPE_OPENGL_ES_20
 #define MEDIAPIPE_METAL_ENABLED 1
 #elif defined(MEDIAPIPE_OSX)
-#define MEDIAPIPE_OPENGL_ES_VERSION MEDIAPIPE_OPENGL_ES_UNSUPPORTED
+#define MEDIAPIPE_OPENGL_ES_VERSION 0
 #define MEDIAPIPE_METAL_ENABLED 1
 #else
-// GPU is not supported on Linux yet.
-#define MEDIAPIPE_OPENGL_ES_VERSION MEDIAPIPE_OPENGL_ES_UNSUPPORTED
+// WebGL config.
+#define MEDIAPIPE_OPENGL_ES_VERSION MEDIAPIPE_OPENGL_ES_30
 #define MEDIAPIPE_METAL_ENABLED 0
 #endif
 #endif

mediapipe/framework/profiler/BUILD

@@ -269,7 +269,7 @@ cc_test(
 
 cc_library(
     name = "profiler_resource_util",
-    srcs = select({
+    srcs = ["profiler_resource_util_common.cc"] + select({
         "//conditions:default": ["profiler_resource_util.cc"],
         "//mediapipe:android": ["profiler_resource_util_android.cc"],
         "//mediapipe:ios": ["profiler_resource_util_ios.cc"],
@@ -288,15 +288,22 @@ cc_library(
     deps = [
         "@com_google_absl//absl/strings",
         "//mediapipe/framework/port:logging",
+        "//mediapipe/framework/port:ret_check",
         "//mediapipe/framework/port:statusor",
         "//mediapipe/framework/port:status",
+        "@com_google_absl//absl/flags:flag",
+        "//mediapipe/framework/deps:file_path",
     ] + select({
         "//conditions:default": [
+            "//mediapipe/framework/port:file_helpers",
         ],
         "//mediapipe:android": [
             "//mediapipe/java/com/google/mediapipe/framework/jni:jni_util",
+            "//mediapipe/util/android/file/base",
+        ],
+        "//mediapipe:apple": [
+            "//mediapipe/framework/port:file_helpers",
         ],
-        "//mediapipe:apple": [],
     }),
 )
 

mediapipe/framework/profiler/graph_profiler_test.cc

@@ -22,6 +22,7 @@
 #include "mediapipe/framework/port/gmock.h"
 #include "mediapipe/framework/port/gtest.h"
 #include "mediapipe/framework/port/status_matchers.h"
+#include "mediapipe/framework/port/statusor.h"
 #include "mediapipe/framework/profiler/test_context_builder.h"
 #include "mediapipe/framework/tool/simulation_clock.h"
 #include "mediapipe/framework/tool/tag_map_helper.h"

mediapipe/framework/profiler/profiler_resource_util.h

@@ -27,6 +27,10 @@ namespace mediapipe {
 // error.
 StatusOr<std::string> GetDefaultTraceLogDirectory();
 
+// Given a log file path, this function provides an absolute path with which
+// it can be accessed as a file.  Enclosing directories are created as needed.
+StatusOr<std::string> PathToLogFile(const std::string& path);
+
 }  // namespace mediapipe
 
 #endif  // MEDIAPIPE_FRAMEWORK_PROFILER_PROFILER_RESOURCE_UTIL_H_

mediapipe/framework/profiler/profiler_resource_util_common.cc

@@ -0,0 +1,60 @@
+// Copyright 2020 The MediaPipe Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "absl/flags/flag.h"
+#include "mediapipe/framework/deps/file_path.h"
+#include "mediapipe/framework/port/ret_check.h"
+#include "mediapipe/framework/port/status_macros.h"
+#include "mediapipe/framework/profiler/profiler_resource_util.h"
+
+// TODO: Move this Android include to port/file_helpers.
+// Also move this from resource_util.cc.
+#ifdef __ANDROID__
+#include "mediapipe/util/android/file/base/filesystem.h"
+#else
+#include "mediapipe/framework/port/file_helpers.h"
+#endif
+
+ABSL_FLAG(std::string, log_root_dir, "",
+          "The absolute path to the logging output directory.  If specified, "
+          "log_root_dir will be prepended to each specified log file path.");
+
+#ifdef __ANDROID__
+namespace mediapipe {
+namespace file {
+::mediapipe::Status RecursivelyCreateDir(absl::string_view path) {
+  return RecursivelyCreateDir(path, file::Options());
+}
+}  // namespace file
+}  // namespace mediapipe
+#endif
+
+namespace mediapipe {
+
+::mediapipe::StatusOr<std::string> GetLogDirectory() {
+  if (!FLAGS_log_root_dir.CurrentValue().empty()) {
+    return FLAGS_log_root_dir.CurrentValue();
+  }
+  return GetDefaultTraceLogDirectory();
+}
+
+::mediapipe::StatusOr<std::string> PathToLogFile(const std::string& path) {
+  ASSIGN_OR_RETURN(std::string log_dir, GetLogDirectory());
+  std::string result = file::JoinPath(log_dir, path);
+  MP_RETURN_IF_ERROR(
+      ::mediapipe::file::RecursivelyCreateDir(file::Dirname(result)));
+  return result;
+}
+
+}  // namespace mediapipe

mediapipe/framework/stream_handler/mux_input_stream_handler.cc

@@ -14,6 +14,7 @@
 
 #include "absl/strings/substitute.h"
 #include "absl/synchronization/mutex.h"
+#include "mediapipe/framework/collection_item_id.h"
 #include "mediapipe/framework/input_stream_handler.h"
 #include "mediapipe/framework/port/logging.h"
 
@@ -121,6 +122,16 @@ class MuxInputStreamHandler : public InputStreamHandler {
                             num_packets_dropped, data_stream->Name());
     AddPacketToShard(&input_set->Get(data_stream_id), std::move(data_packet),
                      stream_is_done);
+
+    // Discard old packets on other streams.
+    // Note that control_stream_id is the last valid id.
+    auto next_timestamp = input_timestamp.NextAllowedInStream();
+    for (CollectionItemId id = input_stream_managers_.BeginId();
+         id < control_stream_id; ++id) {
+      if (id == data_stream_id) continue;
+      auto& other_stream = input_stream_managers_.Get(id);
+      other_stream->ErasePacketsEarlierThan(next_timestamp);
+    }
   }
 
  private:

mediapipe/framework/tool/simulation_clock_executor.cc

@@ -19,11 +19,11 @@
 namespace mediapipe {
 
 SimulationClockExecutor::SimulationClockExecutor(int num_threads)
-    : ThreadPoolExecutor(num_threads), clock_(new SimulationClock()) {}
+    : clock_(new SimulationClock()), executor_(num_threads) {}
 
 void SimulationClockExecutor::Schedule(std::function<void()> task) {
   clock_->ThreadStart();
-  ThreadPoolExecutor::Schedule([this, task] {
+  executor_.Schedule([this, task] {
     clock_->Sleep(absl::ZeroDuration());
     task();
     clock_->ThreadFinish();

mediapipe/framework/tool/simulation_clock_executor.h

@@ -23,7 +23,7 @@ namespace mediapipe {
 // Simulation clock multithreaded executor. This is intended to be used with
 // graphs that are using SimulationClock class to emulate various parts of the
 // graph taking specific time to process the incoming packets.
-class SimulationClockExecutor : public ThreadPoolExecutor {
+class SimulationClockExecutor : public Executor {
  public:
   explicit SimulationClockExecutor(int num_threads);
   void Schedule(std::function<void()> task) override;
@@ -36,6 +36,9 @@ class SimulationClockExecutor : public ThreadPoolExecutor {
  private:
   // SimulationClock instance used by this executor.
   std::shared_ptr<SimulationClock> clock_;
+  // The delegate ThreadPoolExecutor.  This is declared after clock_
+  // so that it is destroyed before clock_.
+  ThreadPoolExecutor executor_;
 };
 
 }  // namespace mediapipe

mediapipe/framework/transitive_protos.bzl

@@ -1,3 +1,17 @@
+# Copyright 2019-2020 The MediaPipe Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
 """Extract a cc_library compatible dependency with only the top level proto rules."""
 
 ProtoLibsInfo = provider(fields = ["targets", "out"])