Creating a fix for a threading issue with ffmpeg_lib in third_party.

Bug is at:

#5804

Fix is to add a unique identifier to each temp file name. The id is unique to
the process. Multiple processes could still have a conflict, though even there
the odds do go down somewhat with this fix.

PiperOrigin-RevId: 173261202

3 files changed, 106 insertions, 2 deletions

diff --git a/tensorflow/contrib/ffmpeg/default/BUILD b/tensorflow/contrib/ffmpeg/default/BUILD
index 05fc658d80..949ae9ad9e 100644
--- a/tensorflow/contrib/ffmpeg/default/BUILD
+++ b/tensorflow/contrib/ffmpeg/default/BUILD
@@ -24,6 +24,18 @@ cc_library(
 )
 
 tf_cc_test(
+    name = "ffmpeg_lib_utility_test",
+    srcs = ["ffmpeg_lib_utility_test.cc"],
+    deps = [
+        ":ffmpeg_lib",
+        "//tensorflow/core:framework_internal",
+        "//tensorflow/core:lib",
+        "//tensorflow/core:test",
+        "//tensorflow/core:test_main",
+    ],
+)
+
+tf_cc_test(
     name = "ffmpeg_lib_installed_test",
     srcs = ["ffmpeg_lib_test.cc"],
     args = [
diff --git a/tensorflow/contrib/ffmpeg/default/ffmpeg_lib.cc b/tensorflow/contrib/ffmpeg/default/ffmpeg_lib.cc
index b417a70b6e..545a4386d0 100644
--- a/tensorflow/contrib/ffmpeg/default/ffmpeg_lib.cc
+++ b/tensorflow/contrib/ffmpeg/default/ffmpeg_lib.cc
@@ -198,6 +198,14 @@ string BuildWavFile(int32 samples_per_second, int32 channel_count,
   return data;
 }
 
+// Returns a unique number every time it is called.
+int64 UniqueId() {
+  static mutex mu(LINKER_INITIALIZED);
+  static int64 id = 0;
+  mutex_lock l(mu);
+  return ++id;
+}
+
 }  // namespace
 
 string GetTempFilename(const string& extension) {
@@ -208,8 +216,12 @@ string GetTempFilename(const string& extension) {
     }
     struct stat statbuf;
     if (!stat(dir, &statbuf) && S_ISDIR(statbuf.st_mode)) {
-      string tmp_filepath =
-          io::JoinPath(dir, StrCat("tmp_file_XXXXXX", ".", extension));
+      // UniqueId is added here because mkstemps is not as thread safe as it
+      // looks. https://github.com/tensorflow/tensorflow/issues/5804 shows
+      // the problem.
+      string tmp_filepath = io::JoinPath(
+          dir,
+          StrCat("tmp_file_tensorflow_", UniqueId(), "_XXXXXX.", extension));
       int fd = mkstemps(&tmp_filepath[0], extension.length() + 1);
       if (fd < 0) {
         LOG(FATAL) << "Failed to create temp file.";
diff --git a/tensorflow/contrib/ffmpeg/default/ffmpeg_lib_utility_test.cc b/tensorflow/contrib/ffmpeg/default/ffmpeg_lib_utility_test.cc
new file mode 100644
index 0000000000..7176f3b550
--- /dev/null
+++ b/tensorflow/contrib/ffmpeg/default/ffmpeg_lib_utility_test.cc
@@ -0,0 +1,80 @@
+// Copyright 2017 The TensorFlow Authors. All Rights Reserved.
+//
+// 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 "tensorflow/contrib/ffmpeg/ffmpeg_lib.h"
+
+#include <array>
+#include <set>
+#include <string>
+#include <vector>
+
+#include "tensorflow/core/lib/core/threadpool.h"
+#include "tensorflow/core/platform/env.h"
+#include "tensorflow/core/platform/mutex.h"
+#include "tensorflow/core/platform/test.h"
+
+namespace tensorflow {
+namespace ffmpeg {
+namespace {
+
+TEST(FfmpegLibTest, TestTempDirectoryThreading) {
+  // Testing a fix for a bug that allowed different threads to create
+  // conflicting temp files.
+  // See github.com/tensorflow/tensorflow/issues/5804 for details.
+  const int32 kNumThreads = 10;
+  const int32 kNumWorkItems = 10000;
+  static constexpr size_t kStringsPerItem = 100;
+  Env* environment = Env::Default();
+  thread::ThreadPool pool(environment, "test", kNumThreads);
+
+  mutex mu;
+  std::vector<string> temp_filenames;
+  temp_filenames.reserve(kNumWorkItems * kStringsPerItem);
+
+  // Queue a large number of work items for the threads to process. Each work
+  // item creates a temp file and then deletes it.
+  for (int i = 0; i < kNumWorkItems; ++i) {
+    pool.Schedule([&mu, &temp_filenames, environment]() {
+      std::array<string, kStringsPerItem> buffer;
+      for (int32 j = 0; j < kStringsPerItem; ++j) {
+        buffer[j] = GetTempFilename("mp3");
+        TF_QCHECK_OK(environment->DeleteFile(buffer[j]));
+      }
+      mutex_lock l(mu);
+      for (const auto& fn : buffer) {
+        temp_filenames.push_back(fn);
+      }
+    });
+  }
+
+  // Wait until all work items are complete.
+  while (true) {
+    mutex_lock l(mu);
+    if (temp_filenames.size() == kNumWorkItems * kStringsPerItem) {
+      break;
+    }
+  }
+
+  // Check that no duplicates are created.
+  std::set<string> unique_filenames;
+  mutex_lock l(mu);
+  for (const auto& fn : temp_filenames) {
+    ASSERT_TRUE(unique_filenames.insert(fn).second);
+  }
+}
+
+}  // namespace
+}  // namespace ffmpeg
+}  // namespace tensorflow