GTTF: Unify parallel test launching.

base/test/launcher/test_launcher.cc

 // Copyright 2013 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "base/test/launcher/test_launcher.h"
 
 #if defined(OS_POSIX)
 #include <fcntl.h>
 #endif
 
 #include "base/at_exit.h"
 #include "base/bind.h"
 #include "base/command_line.h"
 #include "base/environment.h"
 #include "base/file_util.h"
 #include "base/files/file_path.h"
 #include "base/format_macros.h"
 #include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/message_loop/message_loop.h"
 #include "base/process/kill.h"
 #include "base/process/launch.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/stringprintf.h"
 #include "base/strings/utf_string_conversions.h"
 #include "base/test/launcher/test_results_tracker.h"
+#include "base/test/sequenced_worker_pool_owner.h"
 #include "base/test/test_switches.h"
 #include "base/test/test_timeouts.h"
 #include "base/threading/thread_checker.h"
 #include "base/time/time.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 #if defined(OS_MACOSX)
 #include "base/mac/scoped_nsautorelease_pool.h"
 #endif
 
 namespace base {
 
 // See https://groups.google.com/a/chromium.org/d/msg/chromium-dev/nkdTP7sstSc/uT3FaE_sgkAJ .
 using ::operator<<;
 
 // The environment variable name for the total number of test shards.
 const char kTestTotalShards[] = "GTEST_TOTAL_SHARDS";
 // The environment variable name for the test shard index.
 const char kTestShardIndex[] = "GTEST_SHARD_INDEX";
 
 namespace {
 
+// Maximum time of no output after which we print list of processes still
+// running. This deliberately doesn't use TestTimeouts (which is otherwise
+// a recommended solution), because they can be increased. This would defeat
+// the purpose of this timeout, which is 1) to avoid buildbot "no output for
+// X seconds" timeout killing the process 2) help communicate status of
+// the test launcher to people looking at the output (no output for a long
+// time is mysterious and gives no info about what is happening) 3) help
+// debugging in case the process hangs anyway.
+const int kOutputTimeoutSeconds = 15;
+
 // Set of live launch test processes with corresponding lock (it is allowed
 // for callers to launch processes on different threads).
-LazyInstance<std::set<ProcessHandle> > g_live_process_handles
+LazyInstance<std::map<ProcessHandle, CommandLine> > g_live_processes
     = LAZY_INSTANCE_INITIALIZER;
-LazyInstance<Lock> g_live_process_handles_lock = LAZY_INSTANCE_INITIALIZER;
+LazyInstance<Lock> g_live_processes_lock = LAZY_INSTANCE_INITIALIZER;
 
 #if defined(OS_POSIX)
 // Self-pipe that makes it possible to do complex shutdown handling
 // outside of the signal handler.
 int g_shutdown_pipe[2] = { -1, -1 };
 
 void ShutdownPipeSignalHandler(int signal) {
   HANDLE_EINTR(write(g_shutdown_pipe[1], "q", 1));
 }
 
 void KillSpawnedTestProcesses() {
   // Keep the lock until exiting the process to prevent further processes
   // from being spawned.
-  AutoLock lock(g_live_process_handles_lock.Get());
+  AutoLock lock(g_live_processes_lock.Get());
 
   fprintf(stdout,
           "Sending SIGTERM to %" PRIuS " child processes... ",
-          g_live_process_handles.Get().size());
+          g_live_processes.Get().size());
   fflush(stdout);
 
-  for (std::set<ProcessHandle>::iterator i =
-           g_live_process_handles.Get().begin();
-       i != g_live_process_handles.Get().end();
+  for (std::map<ProcessHandle, CommandLine>::iterator i =
+           g_live_processes.Get().begin();
+       i != g_live_processes.Get().end();
        ++i) {
-    kill((-1) * (*i), SIGTERM);  // Send the signal to entire process group.
+    // Send the signal to entire process group.
+    kill((-1) * (i->first), SIGTERM);
   }
 
   fprintf(stdout,
           "done.\nGiving processes a chance to terminate cleanly... ");
   fflush(stdout);
 
   PlatformThread::Sleep(TimeDelta::FromMilliseconds(500));
 
   fprintf(stdout, "done.\n");
   fflush(stdout);
 
   fprintf(stdout,
           "Sending SIGKILL to %" PRIuS " child processes... ",
-          g_live_process_handles.Get().size());
+          g_live_processes.Get().size());
   fflush(stdout);
 
-  for (std::set<ProcessHandle>::iterator i =
-           g_live_process_handles.Get().begin();
-       i != g_live_process_handles.Get().end();
+  for (std::map<ProcessHandle, CommandLine>::iterator i =
+           g_live_processes.Get().begin();
+       i != g_live_processes.Get().end();
        ++i) {
-    kill((-1) * (*i), SIGKILL);  // Send the signal to entire process group.
+    // Send the signal to entire process group.
+    kill((-1) * (i->first), SIGKILL);
   }
 
   fprintf(stdout, "done.\n");
   fflush(stdout);
 }
 
 // I/O watcher for the reading end of the self-pipe above.
 // Terminates any launched child processes and exits the process.
 class SignalFDWatcher : public MessageLoopForIO::Watcher {
  public:
@@ skipping 76 matching lines @@
     // Returns if no more pattern can be found.
     if (cur_pattern == NULL) {
       return false;
     }
 
     // Skips the pattern separater (the ':' character).
     cur_pattern++;
   }
 }
 
+void RunCallback(
+    const TestLauncher::LaunchChildGTestProcessCallback& callback,
+    int exit_code,
+    const TimeDelta& elapsed_time,
+    bool was_timeout,
+    const std::string& output) {
+  callback.Run(exit_code, elapsed_time, was_timeout, output);
+}
+
+void DoLaunchChildTestProcess(
+    const CommandLine& command_line,
+    base::TimeDelta timeout,
+    scoped_refptr<MessageLoopProxy> message_loop_proxy,
+    const TestLauncher::LaunchChildGTestProcessCallback& callback) {
+  TimeTicks start_time = TimeTicks::Now();
+
+  // Redirect child process output to a file.
+  base::FilePath output_file;
+  CHECK(file_util::CreateTemporaryFile(&output_file));
+
+  LaunchOptions options;
+#if defined(OS_WIN)
+  // Make the file handle inheritable by the child.
+  SECURITY_ATTRIBUTES sa_attr;
+  sa_attr.nLength = sizeof(SECURITY_ATTRIBUTES);
+  sa_attr.lpSecurityDescriptor = NULL;
+  sa_attr.bInheritHandle = TRUE;
+
+  win::ScopedHandle handle(CreateFile(output_file.value().c_str(),
+                                      GENERIC_WRITE,
+                                      FILE_SHARE_READ | FILE_SHARE_DELETE,
+                                      &sa_attr,
+                                      OPEN_EXISTING,
+                                      FILE_ATTRIBUTE_TEMPORARY,
+                                      NULL));
+  CHECK(handle.IsValid());
+  options.inherit_handles = true;
+  options.stdin_handle = INVALID_HANDLE_VALUE;
+  options.stdout_handle = handle.Get();
+  options.stderr_handle = handle.Get();
+#elif defined(OS_POSIX)
+  options.new_process_group = true;
+
+  int output_file_fd = open(output_file.value().c_str(), O_RDWR);
+  CHECK_GE(output_file_fd, 0);
+
+  file_util::ScopedFD output_file_fd_closer(&output_file_fd);
+
+  base::FileHandleMappingVector fds_mapping;
+  fds_mapping.push_back(std::make_pair(output_file_fd, STDOUT_FILENO));
+  fds_mapping.push_back(std::make_pair(output_file_fd, STDERR_FILENO));
+  options.fds_to_remap = &fds_mapping;
+#endif
+
+  bool was_timeout = false;
+  int exit_code = LaunchChildTestProcessWithOptions(
+      command_line, options, timeout, &was_timeout);
+
+#if defined(OS_WIN)
+  FlushFileBuffers(handle.Get());
+  handle.Close();
+#elif defined(OS_POSIX)
+  output_file_fd_closer.reset();
+#endif
+
+  std::string output_file_contents;
+  CHECK(base::ReadFileToString(output_file, &output_file_contents));
+
+  if (!base::DeleteFile(output_file, false)) {
+    // This needs to be non-fatal at least for Windows.
+    LOG(WARNING) << "Failed to delete " << output_file.AsUTF8Unsafe();
+  }
+
+  // Run target callback on the thread it was originating from, not on
+  // a worker pool thread.
+  message_loop_proxy->PostTask(
+      FROM_HERE,
+      Bind(&RunCallback,
+           callback,
+           exit_code,
+           TimeTicks::Now() - start_time,
+           was_timeout,
+           output_file_contents));
+}
+
 }  // namespace
 
 const char kGTestFilterFlag[] = "gtest_filter";
 const char kGTestHelpFlag[]   = "gtest_help";
 const char kGTestListTestsFlag[] = "gtest_list_tests";
 const char kGTestRepeatFlag[] = "gtest_repeat";
 const char kGTestRunDisabledTestsFlag[] = "gtest_also_run_disabled_tests";
 const char kGTestOutputFlag[] = "gtest_output";
 
 TestLauncherDelegate::~TestLauncherDelegate() {
 }
 
-TestLauncher::TestLauncher(TestLauncherDelegate* launcher_delegate)
+TestLauncher::TestLauncher(TestLauncherDelegate* launcher_delegate,
+                           size_t parallel_jobs)
     : launcher_delegate_(launcher_delegate),
       total_shards_(1),
       shard_index_(0),
       cycles_(1),
       test_started_count_(0),
       test_finished_count_(0),
       test_success_count_(0),
       test_broken_count_(0),
       retry_count_(0),
       retry_limit_(3),  // TODO(phajdan.jr): Make a flag control this.
-      run_result_(true) {
+      run_result_(true),
+      watchdog_timer_(FROM_HERE,
+                      TimeDelta::FromSeconds(kOutputTimeoutSeconds),
+                      this,
+                      &TestLauncher::OnOutputTimeout),
+      worker_pool_owner_(
+          new SequencedWorkerPoolOwner(parallel_jobs, "test_launcher")) {
 }
 
 TestLauncher::~TestLauncher() {
+  worker_pool_owner_->pool()->Shutdown();
 }
 
 bool TestLauncher::Run(int argc, char** argv) {
   if (!Init())
     return false;
 
 #if defined(OS_POSIX)
   CHECK_EQ(0, pipe(g_shutdown_pipe));
 
   struct sigaction action;
   memset(&action, 0, sizeof(action));
   sigemptyset(&action.sa_mask);
   action.sa_handler = &ShutdownPipeSignalHandler;
 
   CHECK_EQ(0, sigaction(SIGINT, &action, NULL));
   CHECK_EQ(0, sigaction(SIGQUIT, &action, NULL));
   CHECK_EQ(0, sigaction(SIGTERM, &action, NULL));
 
   MessageLoopForIO::FileDescriptorWatcher controller;
   SignalFDWatcher watcher;
 
   CHECK(MessageLoopForIO::current()->WatchFileDescriptor(
             g_shutdown_pipe[0],
             true,
             MessageLoopForIO::WATCH_READ,
             &controller,
             &watcher));
 #endif  // defined(OS_POSIX)
 
+  // Start the watchdog timer.
+  watchdog_timer_.Reset();
+
   MessageLoop::current()->PostTask(
       FROM_HERE,
       Bind(&TestLauncher::RunTestIteration, Unretained(this)));
 
   MessageLoop::current()->Run();
 
   if (cycles_ != 1)
     results_tracker_.PrintSummaryOfAllIterations();
 
   MaybeSaveSummaryAsJSON();
 
   return run_result_;
 }
 
+void TestLauncher::LaunchChildGTestProcess(
+    const CommandLine& command_line,
+    const std::string& wrapper,
+    base::TimeDelta timeout,
+    const LaunchChildGTestProcessCallback& callback) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  // Record the exact command line used to launch the child.
+  CommandLine new_command_line(
+      PrepareCommandLineForGTest(command_line, wrapper));
+
+  worker_pool_owner_->pool()->PostWorkerTask(
+      FROM_HERE,
+      Bind(&DoLaunchChildTestProcess,
+           new_command_line,
+           timeout,
+           MessageLoopProxy::current(),
+           Bind(&TestLauncher::OnLaunchTestProcessFinished,
+                Unretained(this),
+                callback)));
+}
+
 void TestLauncher::OnTestFinished(const TestResult& result) {
   ++test_finished_count_;
 
   bool print_snippet = false;
   std::string print_test_stdio("auto");
   if (CommandLine::ForCurrentProcess()->HasSwitch(
           switches::kTestLauncherPrintTestStdio)) {
     print_test_stdio = CommandLine::ForCurrentProcess()->GetSwitchValueASCII(
         switches::kTestLauncherPrintTestStdio);
   }
@@ skipping 37 matching lines @@
     status_line.append("(SKIPPED)");
   } else if (result.status == TestResult::TEST_UNKNOWN) {
     status_line.append("(UNKNOWN)");
   } else {
     // Fail very loudly so it's not ignored.
     CHECK(false) << "Unhandled test result status: " << result.status;
   }
   fprintf(stdout, "%s\n", status_line.c_str());
   fflush(stdout);
 
+  // We just printed a status line, reset the watchdog timer.
+  watchdog_timer_.Reset();
+
   if (test_finished_count_ == test_started_count_) {
     if (!tests_to_retry_.empty() && retry_count_ < retry_limit_) {
       retry_count_++;
 
       std::vector<std::string> test_names(tests_to_retry_.begin(),
                                           tests_to_retry_.end());
 
       tests_to_retry_.clear();
 
       size_t retry_started_count =
@@ skipping 196 matching lines @@
   test_broken_count_ = 0;
   retry_count_ = 0;
   tests_to_retry_.clear();
   results_tracker_.OnTestIterationStarting();
   launcher_delegate_->OnTestIterationStarting();
 
   MessageLoop::current()->PostTask(
       FROM_HERE, Bind(&TestLauncher::RunTests, Unretained(this)));
 }
 
-void TestLauncher::OnAllTestsStarted() {
-}
-
 void TestLauncher::MaybeSaveSummaryAsJSON() {
   const CommandLine* command_line = CommandLine::ForCurrentProcess();
   if (command_line->HasSwitch(switches::kTestLauncherSummaryOutput)) {
     FilePath summary_path(command_line->GetSwitchValuePath(
                               switches::kTestLauncherSummaryOutput));
     if (!results_tracker_.SaveSummaryAsJSON(summary_path)) {
       LOG(ERROR) << "Failed to save test launcher output summary.";
     }
   }
 }
 
+void TestLauncher::OnLaunchTestProcessFinished(
+    const LaunchChildGTestProcessCallback& callback,
+    int exit_code,
+    const TimeDelta& elapsed_time,
+    bool was_timeout,
+    const std::string& output) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  callback.Run(exit_code, elapsed_time, was_timeout, output);
+}
+
+void TestLauncher::OnOutputTimeout() {
+  DCHECK(thread_checker_.CalledOnValidThread());
+
+  AutoLock lock(g_live_processes_lock.Get());
+
+  fprintf(stdout, "Still waiting for the following processes to finish:\n");
+
+  for (std::map<ProcessHandle, CommandLine>::iterator i =
+           g_live_processes.Get().begin();
+       i != g_live_processes.Get().end();
+       ++i) {
+#if defined(OS_WIN)
+    fwprintf(stdout, L"\t%s\n", i->second.GetCommandLineString().c_str());
+#else
+    fprintf(stdout, "\t%s\n", i->second.GetCommandLineString().c_str());
+#endif
+  }
+
+  fflush(stdout);
+
+  // Arm the timer again - otherwise it would fire only once.
+  watchdog_timer_.Reset();
+}
+
 std::string GetTestOutputSnippet(const TestResult& result,
                                  const std::string& full_output) {
   size_t run_pos = full_output.find(std::string("[ RUN      ] ") +
                                     result.full_name);
   if (run_pos == std::string::npos)
     return std::string();
 
   size_t end_pos = full_output.find(std::string("[  FAILED  ] ") +
                                     result.full_name,
                                     run_pos);
@@ skipping 94 matching lines @@
 
   new_options.job_handle = job_handle.Get();
 #endif  // defined(OS_WIN)
 
   base::ProcessHandle process_handle;
 
   {
     // Note how we grab the lock before the process possibly gets created.
     // This ensures that when the lock is held, ALL the processes are registered
     // in the set.
-    AutoLock lock(g_live_process_handles_lock.Get());
+    AutoLock lock(g_live_processes_lock.Get());
 
     if (!base::LaunchProcess(command_line, new_options, &process_handle))
       return -1;
 
-    g_live_process_handles.Get().insert(process_handle);
+    g_live_processes.Get().insert(std::make_pair(process_handle, command_line));
   }
 
   int exit_code = 0;
   if (!base::WaitForExitCodeWithTimeout(process_handle,
                                         &exit_code,
                                         timeout)) {
     *was_timeout = true;
     exit_code = -1;  // Set a non-zero exit code to signal a failure.
 
     // Ensure that the process terminates.
     base::KillProcess(process_handle, -1, true);
   }
 
   {
     // Note how we grab the log before issuing a possibly broad process kill.
     // Other code parts that grab the log kill processes, so avoid trying
     // to do that twice and trigger all kinds of log messages.
-    AutoLock lock(g_live_process_handles_lock.Get());
+    AutoLock lock(g_live_processes_lock.Get());
 
 #if defined(OS_POSIX)
     if (exit_code != 0) {
       // On POSIX, in case the test does not exit cleanly, either due to a crash
       // or due to it timing out, we need to clean up any child processes that
       // it might have created. On Windows, child processes are automatically
       // cleaned up using JobObjects.
       base::KillProcessGroup(process_handle);
     }
 #endif
 
-    g_live_process_handles.Get().erase(process_handle);
+    g_live_processes.Get().erase(process_handle);
   }
 
   base::CloseProcessHandle(process_handle);
 
   return exit_code;
 }
 
 }  // namespace base