audio_pipe_reader.cc 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211
  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 "remoting/host/linux/audio_pipe_reader.h"
  5. #include <fcntl.h>
  6. #include <stddef.h>
  7. #include <sys/stat.h>
  8. #include <sys/types.h>
  9. #include <unistd.h>
  10. #include <utility>
  11. #include "base/bind.h"
  12. #include "base/logging.h"
  13. #include "base/posix/eintr_wrapper.h"
  14. namespace remoting {
  15. namespace {
  16. const int kSampleBytesPerSecond = int{AudioPipeReader::kSamplingRate} *
  17. AudioPipeReader::kChannels *
  18. AudioPipeReader::kBytesPerSample;
  19. #if !defined(F_SETPIPE_SZ)
  20. // F_SETPIPE_SZ is supported only starting linux 2.6.35, but we want to be able
  21. // to compile this code on machines with older kernel.
  22. #define F_SETPIPE_SZ 1031
  23. #endif // defined(F_SETPIPE_SZ)
  24. } // namespace
  25. // static
  26. scoped_refptr<AudioPipeReader> AudioPipeReader::Create(
  27. scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  28. const base::FilePath& pipe_path) {
  29. // Create a reference to the new AudioPipeReader before posting the
  30. // StartOnAudioThread task, otherwise it may be deleted on the audio
  31. // thread before we return.
  32. scoped_refptr<AudioPipeReader> pipe_reader =
  33. new AudioPipeReader(task_runner, pipe_path);
  34. task_runner->PostTask(
  35. FROM_HERE,
  36. base::BindOnce(&AudioPipeReader::StartOnAudioThread, pipe_reader));
  37. return pipe_reader;
  38. }
  39. AudioPipeReader::AudioPipeReader(
  40. scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  41. const base::FilePath& pipe_path)
  42. : task_runner_(task_runner),
  43. pipe_path_(pipe_path),
  44. observers_(new base::ObserverListThreadSafe<StreamObserver>()) {
  45. }
  46. AudioPipeReader::~AudioPipeReader() = default;
  47. void AudioPipeReader::AddObserver(StreamObserver* observer) {
  48. observers_->AddObserver(observer);
  49. }
  50. void AudioPipeReader::RemoveObserver(StreamObserver* observer) {
  51. observers_->RemoveObserver(observer);
  52. }
  53. void AudioPipeReader::StartOnAudioThread() {
  54. DCHECK(task_runner_->BelongsToCurrentThread());
  55. if (!file_watcher_.Watch(
  56. pipe_path_.DirName(), base::FilePathWatcher::Type::kRecursive,
  57. base::BindRepeating(&AudioPipeReader::OnDirectoryChanged,
  58. base::Unretained(this)))) {
  59. LOG(ERROR) << "Failed to watch pulseaudio directory "
  60. << pipe_path_.DirName().value();
  61. }
  62. TryOpenPipe();
  63. }
  64. void AudioPipeReader::OnDirectoryChanged(const base::FilePath& path,
  65. bool error) {
  66. DCHECK(task_runner_->BelongsToCurrentThread());
  67. if (error) {
  68. LOG(ERROR) << "File watcher returned an error.";
  69. return;
  70. }
  71. TryOpenPipe();
  72. }
  73. void AudioPipeReader::TryOpenPipe() {
  74. DCHECK(task_runner_->BelongsToCurrentThread());
  75. base::File new_pipe(
  76. HANDLE_EINTR(open(pipe_path_.value().c_str(), O_RDONLY | O_NONBLOCK)));
  77. // If both |pipe_| and |new_pipe| are valid then compare inodes for the two
  78. // file descriptors. Don't need to do anything if inode hasn't changed.
  79. if (new_pipe.IsValid() && pipe_.IsValid()) {
  80. struct stat old_stat;
  81. struct stat new_stat;
  82. if (fstat(pipe_.GetPlatformFile(), &old_stat) == 0 &&
  83. fstat(new_pipe.GetPlatformFile(), &new_stat) == 0 &&
  84. old_stat.st_ino == new_stat.st_ino) {
  85. return;
  86. }
  87. }
  88. pipe_watch_controller_.reset();
  89. timer_.Stop();
  90. pipe_ = std::move(new_pipe);
  91. if (pipe_.IsValid()) {
  92. // Get buffer size for the pipe.
  93. pipe_buffer_size_ = fpathconf(pipe_.GetPlatformFile(), _PC_PIPE_BUF);
  94. if (pipe_buffer_size_ < 0) {
  95. PLOG(ERROR) << "fpathconf(_PC_PIPE_BUF)";
  96. pipe_buffer_size_ = 4096;
  97. }
  98. // Read from the pipe twice per buffer length, to avoid starving the stream.
  99. capture_period_ =
  100. base::Seconds(1) * pipe_buffer_size_ / kSampleBytesPerSecond / 2;
  101. WaitForPipeReadable();
  102. }
  103. }
  104. void AudioPipeReader::StartTimer() {
  105. DCHECK(task_runner_->BelongsToCurrentThread());
  106. DCHECK(pipe_watch_controller_);
  107. pipe_watch_controller_.reset();
  108. started_time_ = base::TimeTicks::Now();
  109. last_capture_position_ = 0;
  110. timer_.Start(FROM_HERE, capture_period_, this, &AudioPipeReader::DoCapture);
  111. }
  112. void AudioPipeReader::DoCapture() {
  113. DCHECK(task_runner_->BelongsToCurrentThread());
  114. DCHECK(pipe_.IsValid());
  115. // Calculate how much we need read from the pipe. Pulseaudio doesn't control
  116. // how much data it writes to the pipe, so we need to pace the stream.
  117. base::TimeDelta stream_position = base::TimeTicks::Now() - started_time_;
  118. int64_t stream_position_bytes = stream_position.InMilliseconds() *
  119. kSampleBytesPerSecond /
  120. base::Time::kMillisecondsPerSecond;
  121. int64_t bytes_to_read = stream_position_bytes - last_capture_position_;
  122. std::string data = left_over_bytes_;
  123. size_t pos = data.size();
  124. left_over_bytes_.clear();
  125. data.resize(pos + bytes_to_read);
  126. while (pos < data.size()) {
  127. int read_result =
  128. pipe_.ReadAtCurrentPos(std::data(data) + pos, data.size() - pos);
  129. if (read_result > 0) {
  130. pos += read_result;
  131. } else {
  132. if (read_result < 0 && errno != EWOULDBLOCK && errno != EAGAIN)
  133. PLOG(ERROR) << "read";
  134. break;
  135. }
  136. }
  137. // Stop reading from the pipe if PulseAudio isn't writing anything.
  138. if (pos == 0) {
  139. WaitForPipeReadable();
  140. return;
  141. }
  142. // Save any incomplete samples we've read for later. Each packet should
  143. // contain integer number of samples.
  144. int incomplete_samples_bytes = pos % (int{kChannels} * kBytesPerSample);
  145. left_over_bytes_.assign(data, pos - incomplete_samples_bytes,
  146. incomplete_samples_bytes);
  147. data.resize(pos - incomplete_samples_bytes);
  148. last_capture_position_ += data.size();
  149. // Normally PulseAudio will keep pipe buffer full, so we should always be able
  150. // to read |bytes_to_read| bytes, but in case it's misbehaving we need to make
  151. // sure that |stream_position_bytes| doesn't go out of sync with the current
  152. // stream position.
  153. if (stream_position_bytes - last_capture_position_ > pipe_buffer_size_)
  154. last_capture_position_ = stream_position_bytes - pipe_buffer_size_;
  155. DCHECK_LE(last_capture_position_, stream_position_bytes);
  156. // Dispatch asynchronous notification to the stream observers.
  157. scoped_refptr<base::RefCountedString> data_ref =
  158. base::RefCountedString::TakeString(&data);
  159. observers_->Notify(FROM_HERE, &StreamObserver::OnDataRead, data_ref);
  160. }
  161. void AudioPipeReader::WaitForPipeReadable() {
  162. timer_.Stop();
  163. DCHECK(!pipe_watch_controller_);
  164. pipe_watch_controller_ = base::FileDescriptorWatcher::WatchReadable(
  165. pipe_.GetPlatformFile(), base::BindRepeating(&AudioPipeReader::StartTimer,
  166. base::Unretained(this)));
  167. }
  168. // static
  169. void AudioPipeReaderTraits::Destruct(const AudioPipeReader* audio_pipe_reader) {
  170. audio_pipe_reader->task_runner_->DeleteSoon(FROM_HERE, audio_pipe_reader);
  171. }
  172. } // namespace remoting