User MessageLoopForIO::WatchFileDescriptor in proces_output_watcher

chromeos/process_proxy/process_proxy.cc

 // Copyright (c) 2012 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 "chromeos/process_proxy/process_proxy.h"
 
 #include <stdlib.h>
 #include <sys/ioctl.h>
 
 #include "base/bind.h"
 #include "base/command_line.h"
 #include "base/files/file_util.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/posix/eintr_wrapper.h"
 #include "base/process/launch.h"
 #include "base/process/process.h"
 #include "base/single_thread_task_runner.h"
 #include "base/thread_task_runner_handle.h"
-#include "base/threading/thread.h"
 #include "third_party/cros_system_api/switches/chrome_switches.h"
 
 namespace {
 
-enum PipeEnd {
-  PIPE_END_READ,
-  PIPE_END_WRITE
-};
-
 enum PseudoTerminalFd {
   PT_MASTER_FD,
   PT_SLAVE_FD
 };
 
 const int kInvalidFd = -1;
 
+void StopOutputWatcher(scoped_ptr<chromeos::ProcessOutputWatcher> watcher) {
+  // Just deleting |watcher| if sufficient to stop it.
+}
+
 }  // namespace
 
 namespace chromeos {
 
-ProcessProxy::ProcessProxy(): process_launched_(false),
-                              callback_set_(false),
-                              watcher_started_(false) {
+ProcessProxy::ProcessProxy() : process_launched_(false), callback_set_(false) {
   // Set pipes to initial, invalid value so we can easily know if a pipe was
   // opened by us.
-  ClearAllFdPairs();
+  ClearFdPair(pt_pair_);
 }
 
 bool ProcessProxy::Open(const std::string& command, pid_t* pid) {
   if (process_launched_)
     return false;
 
   if (!CreatePseudoTerminalPair(pt_pair_)) {
     return false;
   }
 
   process_launched_ = LaunchProcess(command, pt_pair_[PT_SLAVE_FD], &pid_);
 
   if (process_launched_) {
     // We won't need these anymore. These will be used by the launched process.
     CloseFd(&pt_pair_[PT_SLAVE_FD]);
     *pid = pid_;
     LOG(WARNING) << "Process launched: " << pid_;
   } else {
     CloseFdPair(pt_pair_);
   }
   return process_launched_;
 }
 
-bool ProcessProxy::StartWatchingOnThread(
-    base::Thread* watch_thread,
+bool ProcessProxy::StartWatchingOutput(
+    const scoped_refptr<base::SingleThreadTaskRunner>& watcher_runner,
     const ProcessOutputCallback& callback) {
   DCHECK(process_launched_);
-  if (watcher_started_)
-    return false;
-  if (pipe(shutdown_pipe_) ||
-      !ProcessOutputWatcher::VerifyFileDescriptor(
-          shutdown_pipe_[PIPE_END_READ])) {
-    return false;
-  }
+  CHECK(!output_watcher_.get());
 
   // We give ProcessOutputWatcher a copy of master to make life easier during
   // tear down.
   // TODO(tbarzic): improve fd managment.
   int master_copy = HANDLE_EINTR(dup(pt_pair_[PT_MASTER_FD]));
-  if (!ProcessOutputWatcher::VerifyFileDescriptor(master_copy))
+  if (master_copy < 0)
     return false;
 
   callback_set_ = true;
   callback_ = callback;
   callback_runner_ = base::ThreadTaskRunnerHandle::Get();
+  watcher_runner_ = watcher_runner;
 
   // This object will delete itself once watching is stopped.
   // It also takes ownership of the passed fds.
-  ProcessOutputWatcher* output_watcher =
-      new ProcessOutputWatcher(master_copy,
-                               shutdown_pipe_[PIPE_END_READ],
-                               base::Bind(&ProcessProxy::OnProcessOutput,
-                                          this));
+  output_watcher_.reset(new ProcessOutputWatcher(
+      master_copy, base::Bind(&ProcessProxy::OnProcessOutput, this)));
 
-  // Output watcher took ownership of the read end of shutdown pipe.
-  shutdown_pipe_[PIPE_END_READ] = -1;
+  watcher_runner_->PostTask(
+      FROM_HERE, base::Bind(&ProcessOutputWatcher::Start,
+                            base::Unretained(output_watcher_.get())));
 
-  // |watch| thread is blocked by |output_watcher| from now on.
-  watch_thread->task_runner()->PostTask(
-      FROM_HERE, base::Bind(&ProcessOutputWatcher::Start,
-                            base::Unretained(output_watcher)));
-  watcher_started_ = true;
   return true;
 }
 
 void ProcessProxy::OnProcessOutput(ProcessOutputType type,
                                    const std::string& output) {
   if (!callback_runner_.get())
     return;
 
   callback_runner_->PostTask(
       FROM_HERE,
       base::Bind(&ProcessProxy::CallOnProcessOutputCallback,
                  this, type, output));
 }
 
 void ProcessProxy::CallOnProcessOutputCallback(ProcessOutputType type,
                                                const std::string& output) {
   // We may receive some output even after Close was called (crosh process does
   // not have to quit instantly, or there may be some trailing data left in
   // output stream fds). In that case owner of the callback may be gone so we
   // don't want to send it anything. |callback_set_| is reset when this gets
   // closed.
   if (callback_set_)
     callback_.Run(type, output);
 }
 
-bool ProcessProxy::StopWatching() {
-  if (!watcher_started_)
-    return true;
-  // Signal Watcher that we are done. We use self-pipe trick to unblock watcher.
-  // Anything may be written to the pipe.
-  const char message[] = "q";
-  return base::WriteFileDescriptor(shutdown_pipe_[PIPE_END_WRITE],
-                                   message, sizeof(message));
+void ProcessProxy::StopWatching() {
+  if (!output_watcher_.get())
+    return;
+
+  watcher_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&StopOutputWatcher, base::Passed(&output_watcher_)));
 }
 
 void ProcessProxy::Close() {
   if (!process_launched_)
     return;
 
   process_launched_ = false;
   callback_set_ = false;
   callback_ = ProcessOutputCallback();
   callback_runner_ = NULL;
 
   base::Process process = base::Process::DeprecatedGetProcessFromHandle(pid_);
   process.Terminate(0, true /* wait */);
 
-  // TODO(tbarzic): What if this fails?
   StopWatching();
-
-  CloseAllFdPairs();
+  CloseFdPair(pt_pair_);
 }
 
 bool ProcessProxy::Write(const std::string& text) {
   if (!process_launched_)
     return false;
 
   // We don't want to write '\0' to the pipe.
   size_t data_size = text.length() * sizeof(*text.c_str());
   return base::WriteFileDescriptor(
              pt_pair_[PT_MASTER_FD], text.c_str(), data_size);
 }
 
 bool ProcessProxy::OnTerminalResize(int width, int height) {
   if (width < 0 || height < 0)
     return false;
 
   winsize ws;
   // Number of rows.
   ws.ws_row = height;
   // Number of columns.
   ws.ws_col = width;
 
   return (HANDLE_EINTR(ioctl(pt_pair_[PT_MASTER_FD], TIOCSWINSZ, &ws)) != -1);
 }
 
 ProcessProxy::~ProcessProxy() {
-  // In case watcher did not started, we may get deleted without calling Close.
-  // In that case we have to clean up created pipes. If watcher had been
-  // started, there will be a callback with our reference owned by
-  // process_output_watcher until Close is called, so we know Close has been
-  // called  by now (and pipes have been cleaned).
-  if (!watcher_started_)
-    CloseAllFdPairs();
+  Close();
 }
 
 bool ProcessProxy::CreatePseudoTerminalPair(int *pt_pair) {
   ClearFdPair(pt_pair);
 
   // Open Master.
   pt_pair[PT_MASTER_FD] = HANDLE_EINTR(posix_openpt(O_RDWR | O_NOCTTY));
   if (pt_pair[PT_MASTER_FD] == -1)
     return false;
 
@@ skipping 36 matching lines @@
   base::Process process =
       base::LaunchProcess(base::CommandLine(base::FilePath(command)), options);
 
   // TODO(rvargas) crbug/417532: This is somewhat wrong but the interface of
   // Open vends pid_t* so ownership is quite vague anyway, and Process::Close
   // doesn't do much in POSIX.
   *pid = process.Pid();
   return process.IsValid();
 }
 
-void ProcessProxy::CloseAllFdPairs() {
-  CloseFdPair(pt_pair_);
-  CloseFdPair(shutdown_pipe_);
-}
-
 void ProcessProxy::CloseFdPair(int* pipe) {
-  CloseFd(&(pipe[PIPE_END_READ]));
-  CloseFd(&(pipe[PIPE_END_WRITE]));
+  CloseFd(&(pipe[PT_MASTER_FD]));
+  CloseFd(&(pipe[PT_SLAVE_FD]));
 }
 
 void ProcessProxy::CloseFd(int* fd) {
   if (*fd != kInvalidFd) {
     if (IGNORE_EINTR(close(*fd)) != 0)
       DPLOG(WARNING) << "close fd failed.";
   }
   *fd = kInvalidFd;
 }
 
-void ProcessProxy::ClearAllFdPairs() {
-  ClearFdPair(pt_pair_);
-  ClearFdPair(shutdown_pipe_);
-}
-
 void ProcessProxy::ClearFdPair(int* pipe) {
-  pipe[PIPE_END_READ] = kInvalidFd;
-  pipe[PIPE_END_WRITE] = kInvalidFd;
+  pipe[PT_MASTER_FD] = kInvalidFd;
+  pipe[PT_SLAVE_FD] = kInvalidFd;
 }
 
 }  // namespace chromeos