remoting/host/linux/audio_pipe_reader.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 "remoting/host/linux/audio_pipe_reader.h"
   6
   7 #include <fcntl.h>
   8 #include <sys/stat.h>
   9 #include <sys/types.h>
  10 #include <unistd.h>
  11
  12 #include "base/logging.h"
  13 #include "base/posix/eintr_wrapper.h"
  14 #include "base/stl_util.h"
  15
  16 namespace remoting {
  17
  18 namespace {
  19
  20 const int kSampleBytesPerSecond = AudioPipeReader::kSamplingRate *
  21                                   AudioPipeReader::kChannels *
  22                                   AudioPipeReader::kBytesPerSample;
  23
  24 // Read data from the pipe every 40ms.
  25 const int kCapturingPeriodMs = 40;
  26
  27 // Size of the pipe buffer in milliseconds.
  28 const int kPipeBufferSizeMs = kCapturingPeriodMs * 2;
  29
  30 // Size of the pipe buffer in bytes.
  31 const int kPipeBufferSizeBytes = kPipeBufferSizeMs * kSampleBytesPerSecond /
  32     base::Time::kMillisecondsPerSecond;
  33
  34 #if !defined(F_SETPIPE_SZ)
  35 // F_SETPIPE_SZ is supported only starting linux 2.6.35, but we want to be able
  36 // to compile this code on machines with older kernel.
  37 #define F_SETPIPE_SZ 1031
  38 #endif  // defined(F_SETPIPE_SZ)
  39
  40 }  // namespace
  41
  42 // static
  43 scoped_refptr<AudioPipeReader> AudioPipeReader::Create(
  44     scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  45     const base::FilePath& pipe_path) {
  46   // Create a reference to the new AudioPipeReader before posting the
  47   // StartOnAudioThread task, otherwise it may be deleted on the audio
  48   // thread before we return.
  49   scoped_refptr<AudioPipeReader> pipe_reader =
  50       new AudioPipeReader(task_runner, pipe_path);
  51   task_runner->PostTask(
  52       FROM_HERE, base::Bind(&AudioPipeReader::StartOnAudioThread, pipe_reader));
  53   return pipe_reader;
  54 }
  55
  56 AudioPipeReader::AudioPipeReader(
  57     scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  58     const base::FilePath& pipe_path)
  59     : task_runner_(task_runner),
  60       pipe_path_(pipe_path),
  61       observers_(new ObserverListThreadSafe<StreamObserver>()) {
  62 }
  63
  64 AudioPipeReader::~AudioPipeReader() {}
  65
  66 void AudioPipeReader::AddObserver(StreamObserver* observer) {
  67   observers_->AddObserver(observer);
  68 }
  69 void AudioPipeReader::RemoveObserver(StreamObserver* observer) {
  70   observers_->RemoveObserver(observer);
  71 }
  72
  73 void AudioPipeReader::OnFileCanReadWithoutBlocking(int fd) {
  74   DCHECK_EQ(fd, pipe_.GetPlatformFile());
  75   StartTimer();
  76 }
  77
  78 void AudioPipeReader::OnFileCanWriteWithoutBlocking(int fd) {
  79   NOTREACHED();
  80 }
  81
  82 void AudioPipeReader::StartOnAudioThread() {
  83   DCHECK(task_runner_->BelongsToCurrentThread());
  84
  85   if (!file_watcher_.Watch(pipe_path_.DirName(), true,
  86                            base::Bind(&AudioPipeReader::OnDirectoryChanged,
  87                                       base::Unretained(this)))) {
  88     LOG(ERROR) << "Failed to watch pulseaudio directory "
  89                << pipe_path_.DirName().value();
  90   }
  91
  92   TryOpenPipe();
  93 }
  94
  95 void AudioPipeReader::OnDirectoryChanged(const base::FilePath& path,
  96                                          bool error) {
  97   DCHECK(task_runner_->BelongsToCurrentThread());
  98
  99   if (error) {
 100     LOG(ERROR) << "File watcher returned an error.";
 101     return;
 102   }
 103
 104   TryOpenPipe();
 105 }
 106
 107 void AudioPipeReader::TryOpenPipe() {
 108   DCHECK(task_runner_->BelongsToCurrentThread());
 109
 110   base::File new_pipe;
 111   new_pipe.Initialize(
 112       pipe_path_,
 113       base::File::FLAG_OPEN | base::File::FLAG_READ | base::File::FLAG_ASYNC);
 114
 115   // If both |pipe_| and |new_pipe| are valid then compare inodes for the two
 116   // file descriptors. Don't need to do anything if inode hasn't changed.
 117   if (new_pipe.IsValid() && pipe_.IsValid()) {
 118     struct stat old_stat;
 119     struct stat new_stat;
 120     if (fstat(pipe_.GetPlatformFile(), &old_stat) == 0 &&
 121         fstat(new_pipe.GetPlatformFile(), &new_stat) == 0 &&
 122         old_stat.st_ino == new_stat.st_ino) {
 123       return;
 124     }
 125   }
 126
 127   file_descriptor_watcher_.StopWatchingFileDescriptor();
 128   timer_.Stop();
 129
 130   pipe_ = new_pipe.Pass();
 131
 132   if (pipe_.IsValid()) {
 133     // Set O_NONBLOCK flag.
 134     if (HANDLE_EINTR(fcntl(pipe_.GetPlatformFile(), F_SETFL, O_NONBLOCK)) < 0) {
 135       PLOG(ERROR) << "fcntl";
 136       pipe_.Close();
 137       return;
 138     }
 139
 140     // Set buffer size for the pipe.
 141     if (HANDLE_EINTR(fcntl(
 142             pipe_.GetPlatformFile(), F_SETPIPE_SZ, kPipeBufferSizeBytes)) < 0) {
 143       PLOG(ERROR) << "fcntl";
 144     }
 145
 146     WaitForPipeReadable();
 147   }
 148 }
 149
 150 void AudioPipeReader::StartTimer() {
 151   DCHECK(task_runner_->BelongsToCurrentThread());
 152   started_time_ = base::TimeTicks::Now();
 153   last_capture_position_ = 0;
 154   timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(kCapturingPeriodMs),
 155                this, &AudioPipeReader::DoCapture);
 156 }
 157
 158 void AudioPipeReader::DoCapture() {
 159   DCHECK(task_runner_->BelongsToCurrentThread());
 160   DCHECK(pipe_.IsValid());
 161
 162   // Calculate how much we need read from the pipe. Pulseaudio doesn't control
 163   // how much data it writes to the pipe, so we need to pace the stream.
 164   base::TimeDelta stream_position = base::TimeTicks::Now() - started_time_;
 165   int64 stream_position_bytes = stream_position.InMilliseconds() *
 166       kSampleBytesPerSecond / base::Time::kMillisecondsPerSecond;
 167   int64 bytes_to_read = stream_position_bytes - last_capture_position_;
 168
 169   std::string data = left_over_bytes_;
 170   size_t pos = data.size();
 171   left_over_bytes_.clear();
 172   data.resize(pos + bytes_to_read);
 173
 174   while (pos < data.size()) {
 175     int read_result =
 176         pipe_.ReadAtCurrentPos(string_as_array(&data) + pos, data.size() - pos);
 177     if (read_result > 0) {
 178       pos += read_result;
 179     } else {
 180       if (read_result < 0 && errno != EWOULDBLOCK && errno != EAGAIN)
 181         PLOG(ERROR) << "read";
 182       break;
 183     }
 184   }
 185
 186   // Stop reading from the pipe if PulseAudio isn't writing anything.
 187   if (pos == 0) {
 188     WaitForPipeReadable();
 189     return;
 190   }
 191
 192   // Save any incomplete samples we've read for later. Each packet should
 193   // contain integer number of samples.
 194   int incomplete_samples_bytes = pos % (kChannels * kBytesPerSample);
 195   left_over_bytes_.assign(data, pos - incomplete_samples_bytes,
 196                           incomplete_samples_bytes);
 197   data.resize(pos - incomplete_samples_bytes);
 198
 199   last_capture_position_ += data.size();
 200   // Normally PulseAudio will keep pipe buffer full, so we should always be able
 201   // to read |bytes_to_read| bytes, but in case it's misbehaving we need to make
 202   // sure that |stream_position_bytes| doesn't go out of sync with the current
 203   // stream position.
 204   if (stream_position_bytes - last_capture_position_ > kPipeBufferSizeBytes)
 205     last_capture_position_ = stream_position_bytes - kPipeBufferSizeBytes;
 206   DCHECK_LE(last_capture_position_, stream_position_bytes);
 207
 208   // Dispatch asynchronous notification to the stream observers.
 209   scoped_refptr<base::RefCountedString> data_ref =
 210       base::RefCountedString::TakeString(&data);
 211   observers_->Notify(&StreamObserver::OnDataRead, data_ref);
 212 }
 213
 214 void AudioPipeReader::WaitForPipeReadable() {
 215   timer_.Stop();
 216   base::MessageLoopForIO::current()->WatchFileDescriptor(
 217       pipe_.GetPlatformFile(), false, base::MessageLoopForIO::WATCH_READ,
 218       &file_descriptor_watcher_, this);
 219 }
 220
 221 // static
 222 void AudioPipeReaderTraits::Destruct(const AudioPipeReader* audio_pipe_reader) {
 223   audio_pipe_reader->task_runner_->DeleteSoon(FROM_HERE, audio_pipe_reader);
 224 }
 225
 226 }  // namespace remoting