process_output_watcher.cc 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168
  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. #include "chromeos/process_proxy/process_output_watcher.h"
  5. #include <stddef.h>
  6. #include <stdint.h>
  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/task/single_thread_task_runner.h"
  15. #include "base/third_party/icu/icu_utf.h"
  16. #include "base/threading/thread_task_runner_handle.h"
  17. namespace {
  18. // Number of `unacked_outputs_` we allow before pausing. Bigger windows are
  19. // faster but cause jank in the renderer if we flood it.
  20. // Tuned with `time cat big.txt` using https://norvig.com/big.txt.
  21. constexpr int kAckWindow = 30;
  22. // Gets byte size for a UTF8 character given it's leading byte. The character
  23. // size is encoded as number of leading '1' bits in the character's leading
  24. // byte. If the most significant bit is '0', the character is a valid ASCII
  25. // and it's byte size is 1.
  26. // The method returns 1 if the provided byte is invalid leading byte.
  27. size_t UTF8SizeFromLeadingByte(uint8_t leading_byte) {
  28. size_t byte_count = 0;
  29. uint8_t mask = 1 << 7;
  30. uint8_t error_mask = 1 << (7 - CBU8_MAX_LENGTH);
  31. while (leading_byte & mask) {
  32. if (mask & error_mask)
  33. return 1;
  34. mask >>= 1;
  35. ++byte_count;
  36. }
  37. return byte_count ? byte_count : 1;
  38. }
  39. } // namespace
  40. namespace chromeos {
  41. ProcessOutputWatcher::ProcessOutputWatcher(
  42. int out_fd,
  43. const ProcessOutputCallback& callback)
  44. : read_buffer_size_(0),
  45. process_output_file_(out_fd),
  46. on_read_callback_(callback) {
  47. CHECK_GE(out_fd, 0);
  48. // We want to be sure we will be able to add 0 at the end of the input, so -1.
  49. read_buffer_capacity_ = std::size(read_buffer_) - 1;
  50. }
  51. ProcessOutputWatcher::~ProcessOutputWatcher() = default;
  52. void ProcessOutputWatcher::Start() {
  53. WatchProcessOutput();
  54. }
  55. void ProcessOutputWatcher::OnProcessOutputCanReadWithoutBlocking() {
  56. output_file_watcher_.reset();
  57. ReadFromFd(process_output_file_.GetPlatformFile());
  58. }
  59. void ProcessOutputWatcher::WatchProcessOutput() {
  60. output_file_watcher_ = base::FileDescriptorWatcher::WatchReadable(
  61. process_output_file_.GetPlatformFile(),
  62. base::BindRepeating(
  63. &ProcessOutputWatcher::OnProcessOutputCanReadWithoutBlocking,
  64. base::Unretained(this)));
  65. }
  66. void ProcessOutputWatcher::ReadFromFd(int fd) {
  67. // We don't want to necessary read pipe until it is empty so we don't starve
  68. // other streams in case data is written faster than we read it. If there is
  69. // more than read_buffer_size_ bytes in pipe, it will be read in the next
  70. // iteration.
  71. DCHECK_GT(read_buffer_capacity_, read_buffer_size_);
  72. ssize_t bytes_read =
  73. HANDLE_EINTR(read(fd, &read_buffer_[read_buffer_size_],
  74. read_buffer_capacity_ - read_buffer_size_));
  75. if (bytes_read > 0) {
  76. ReportOutput(PROCESS_OUTPUT_TYPE_OUT, bytes_read);
  77. return;
  78. }
  79. if (bytes_read < 0)
  80. DPLOG(WARNING) << "read from buffer failed";
  81. // If there is nothing on the output the watched process has exited (slave end
  82. // of pty is closed).
  83. on_read_callback_.Run(PROCESS_OUTPUT_TYPE_EXIT, "");
  84. }
  85. size_t ProcessOutputWatcher::OutputSizeWithoutIncompleteUTF8() {
  86. // Find the last non-trailing character byte. This byte should be used to
  87. // infer the last UTF8 character length.
  88. int last_lead_byte = read_buffer_size_ - 1;
  89. while (true) {
  90. // If the series of trailing bytes is too long, something's not right.
  91. // Report the whole output, without waiting for further character bytes.
  92. if (read_buffer_size_ - last_lead_byte > CBU8_MAX_LENGTH)
  93. return read_buffer_size_;
  94. // If there are trailing characters, there must be a leading one in the
  95. // buffer for a valid UTF8 character. Getting past the buffer begining
  96. // signals something's wrong, or the buffer is empty. In both cases return
  97. // the whole current buffer.
  98. if (last_lead_byte < 0)
  99. return read_buffer_size_;
  100. // Found the starting character byte; stop searching.
  101. if (!CBU8_IS_TRAIL(read_buffer_[last_lead_byte]))
  102. break;
  103. --last_lead_byte;
  104. }
  105. size_t last_length = UTF8SizeFromLeadingByte(read_buffer_[last_lead_byte]);
  106. // Note that if |last_length| == 0 or
  107. // |last_length| + |last_read_byte| < |read_buffer_size_|, the string is
  108. // invalid UTF8. In that case, send the whole read buffer to the observer
  109. // immediately, just as if there is no trailing incomplete UTF8 bytes.
  110. if (!last_length || last_length + last_lead_byte <= read_buffer_size_)
  111. return read_buffer_size_;
  112. return last_lead_byte;
  113. }
  114. void ProcessOutputWatcher::ReportOutput(ProcessOutputType type,
  115. size_t new_bytes_count) {
  116. read_buffer_size_ += new_bytes_count;
  117. size_t output_to_report = OutputSizeWithoutIncompleteUTF8();
  118. on_read_callback_.Run(type, std::string(read_buffer_, output_to_report));
  119. // Move the bytes that were left behind to the beginning of the buffer and
  120. // update the buffer size accordingly.
  121. if (output_to_report < read_buffer_size_) {
  122. for (size_t i = output_to_report; i < read_buffer_size_; ++i) {
  123. read_buffer_[i - output_to_report] = read_buffer_[i];
  124. }
  125. }
  126. read_buffer_size_ -= output_to_report;
  127. // Continue watching immediately if we don't have too many unacked outputs.
  128. if (++unacked_outputs_ <= kAckWindow) {
  129. WatchProcessOutput();
  130. }
  131. }
  132. void ProcessOutputWatcher::AckOutput() {
  133. if (--unacked_outputs_ == kAckWindow) {
  134. WatchProcessOutput();
  135. }
  136. }
  137. } // namespace chromeos