chromeos/process_proxy/process_output_watcher.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "chromeos/process_proxy/process_output_watcher.h"
   6
   7 #include <algorithm>
   8 #include <cstdio>
   9 #include <cstring>
  10 #include "base/bind.h"
  11 #include "base/location.h"
  12 #include "base/logging.h"
  13 #include "base/posix/eintr_wrapper.h"
  14 #include "base/single_thread_task_runner.h"
  15 #include "base/third_party/icu/icu_utf.h"
  16 #include "base/thread_task_runner_handle.h"
  17 #include "base/time/time.h"
  18
  19 namespace {
  20
  21 // Gets byte size for a UTF8 character given it's leading byte. The character
  22 // size is encoded as number of leading '1' bits in the character's leading
  23 // byte. If the most significant bit is '0', the character is a valid ASCII
  24 // and it's byte size is 1.
  25 // The method returns 1 if the provided byte is invalid leading byte.
  26 size_t UTF8SizeFromLeadingByte(uint8 leading_byte) {
  27   size_t byte_count = 0;
  28   uint8 mask = 1 << 7;
  29   uint8 error_mask = 1 << (7 - CBU8_MAX_LENGTH);
  30   while (leading_byte & mask) {
  31     if (mask & error_mask)
  32       return 1;
  33     mask >>= 1;
  34     ++byte_count;
  35   }
  36   return byte_count ? byte_count : 1;
  37 }
  38
  39 }  // namespace
  40
  41 namespace chromeos {
  42
  43 ProcessOutputWatcher::ProcessOutputWatcher(
  44     int out_fd,
  45     const ProcessOutputCallback& callback)
  46     : read_buffer_size_(0),
  47       process_output_file_(out_fd),
  48       on_read_callback_(callback),
  49       weak_factory_(this) {
  50   CHECK_GE(out_fd, 0);
  51   // We want to be sure we will be able to add 0 at the end of the input, so -1.
  52   read_buffer_capacity_ = arraysize(read_buffer_) - 1;
  53 }
  54
  55 ProcessOutputWatcher::~ProcessOutputWatcher() {}
  56
  57 void ProcessOutputWatcher::Start() {
  58   WatchProcessOutput();
  59 }
  60
  61 void ProcessOutputWatcher::OnFileCanReadWithoutBlocking(int fd) {
  62   DCHECK_EQ(process_output_file_.GetPlatformFile(), fd);
  63
  64   output_file_watcher_.StopWatchingFileDescriptor();
  65   ReadFromFd(fd);
  66 }
  67
  68 void ProcessOutputWatcher::OnFileCanWriteWithoutBlocking(int fd) {
  69   NOTREACHED();
  70 }
  71
  72 void ProcessOutputWatcher::WatchProcessOutput() {
  73   base::MessageLoopForIO::current()->WatchFileDescriptor(
  74       process_output_file_.GetPlatformFile(), false,
  75       base::MessageLoopForIO::WATCH_READ, &output_file_watcher_, this);
  76 }
  77
  78 void ProcessOutputWatcher::ReadFromFd(int fd) {
  79   // We don't want to necessary read pipe until it is empty so we don't starve
  80   // other streams in case data is written faster than we read it. If there is
  81   // more than read_buffer_size_ bytes in pipe, it will be read in the next
  82   // iteration.
  83   DCHECK_GT(read_buffer_capacity_, read_buffer_size_);
  84   ssize_t bytes_read =
  85       HANDLE_EINTR(read(fd, &read_buffer_[read_buffer_size_],
  86                         read_buffer_capacity_ - read_buffer_size_));
  87
  88   if (bytes_read > 0) {
  89     ReportOutput(PROCESS_OUTPUT_TYPE_OUT, bytes_read);
  90
  91     // Delay next read to make the process less likely to flood IPC channel
  92     // when output is reported to terminal extension via terminalPrivate API
  93     // (which is the only client of this code).
  94     // TODO(tbarzic): Properly fix this!! Provide a mechanism for clients to
  95     //     ack reported output and continue watching the process when ack is
  96     //     received. https://crbug.com/398901
  97     base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
  98         FROM_HERE, base::Bind(&ProcessOutputWatcher::WatchProcessOutput,
  99                               weak_factory_.GetWeakPtr()),
 100         base::TimeDelta::FromMilliseconds(10));
 101
 102     return;
 103   }
 104
 105   if (bytes_read < 0)
 106     DPLOG(WARNING) << "read from buffer failed";
 107
 108   // If there is nothing on the output the watched process has exited (slave end
 109   // of pty is closed).
 110   on_read_callback_.Run(PROCESS_OUTPUT_TYPE_EXIT, "");
 111
 112   // Cancel pending |WatchProcessOutput| calls.
 113   weak_factory_.InvalidateWeakPtrs();
 114 }
 115
 116 size_t ProcessOutputWatcher::OutputSizeWithoutIncompleteUTF8() {
 117   // Find the last non-trailing character byte. This byte should be used to
 118   // infer the last UTF8 character length.
 119   int last_lead_byte = read_buffer_size_ - 1;
 120   while (true) {
 121     // If the series of trailing bytes is too long, something's not right.
 122     // Report the whole output, without waiting for further character bytes.
 123     if (read_buffer_size_ - last_lead_byte > CBU8_MAX_LENGTH)
 124       return read_buffer_size_;
 125
 126     // If there are trailing characters, there must be a leading one in the
 127     // buffer for a valid UTF8 character. Getting past the buffer begining
 128     // signals something's wrong, or the buffer is empty. In both cases return
 129     // the whole current buffer.
 130     if (last_lead_byte < 0)
 131       return read_buffer_size_;
 132
 133     // Found the starting character byte; stop searching.
 134     if (!CBU8_IS_TRAIL(read_buffer_[last_lead_byte]))
 135       break;
 136
 137     --last_lead_byte;
 138   }
 139
 140   size_t last_length = UTF8SizeFromLeadingByte(read_buffer_[last_lead_byte]);
 141
 142   // Note that if |last_length| == 0 or
 143   // |last_length| + |last_read_byte| < |read_buffer_size_|, the string is
 144   // invalid UTF8. In that case, send the whole read buffer to the observer
 145   // immediately, just as if there is no trailing incomplete UTF8 bytes.
 146   if (!last_length || last_length + last_lead_byte <= read_buffer_size_)
 147     return read_buffer_size_;
 148
 149   return last_lead_byte;
 150 }
 151
 152 void ProcessOutputWatcher::ReportOutput(ProcessOutputType type,
 153                                         size_t new_bytes_count) {
 154   read_buffer_size_ += new_bytes_count;
 155   size_t output_to_report = OutputSizeWithoutIncompleteUTF8();
 156
 157   on_read_callback_.Run(type, std::string(read_buffer_, output_to_report));
 158
 159   // Move the bytes that were left behind to the beginning of the buffer and
 160   // update the buffer size accordingly.
 161   if (output_to_report < read_buffer_size_) {
 162     for (size_t i = output_to_report; i < read_buffer_size_; ++i) {
 163       read_buffer_[i - output_to_report] = read_buffer_[i];
 164     }
 165   }
 166   read_buffer_size_ -= output_to_report;
 167 }
 168
 169 }  // namespace chromeos