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/files/file_path.h"
  13 #include "base/logging.h"
  14 #include "base/posix/eintr_wrapper.h"
  15 #include "base/stl_util.h"
  16
  17 namespace remoting {
  18
  19 namespace {
  20
  21 // PulseAudio's module-pipe-sink must be configured to use the following
  22 // parameters for the sink we read from.
  23 const int kSamplesPerSecond = 48000;
  24 const int kChannels = 2;
  25 const int kBytesPerSample = 2;
  26 const int kSampleBytesPerSecond =
  27     kSamplesPerSecond * kChannels * kBytesPerSample;
  28
  29 // Read data from the pipe every 40ms.
  30 const int kCapturingPeriodMs = 40;
  31
  32 // Size of the pipe buffer in milliseconds.
  33 const int kPipeBufferSizeMs = kCapturingPeriodMs * 2;
  34
  35 // Size of the pipe buffer in bytes.
  36 const int kPipeBufferSizeBytes = kPipeBufferSizeMs * kSampleBytesPerSecond /
  37     base::Time::kMillisecondsPerSecond;
  38
  39 #if !defined(F_SETPIPE_SZ)
  40 // F_SETPIPE_SZ is supported only starting linux 2.6.35, but we want to be able
  41 // to compile this code on machines with older kernel.
  42 #define F_SETPIPE_SZ 1031
  43 #endif  // defined(F_SETPIPE_SZ)
  44
  45 }  // namespace
  46
  47 // static
  48 scoped_refptr<AudioPipeReader> AudioPipeReader::Create(
  49     scoped_refptr<base::SingleThreadTaskRunner> task_runner,
  50     const base::FilePath& pipe_name) {
  51   // Create a reference to the new AudioPipeReader before posting the
  52   // StartOnAudioThread task, otherwise it may be deleted on the audio
  53   // thread before we return.
  54   scoped_refptr<AudioPipeReader> pipe_reader =
  55       new AudioPipeReader(task_runner);
  56   task_runner->PostTask(FROM_HERE, base::Bind(
  57       &AudioPipeReader::StartOnAudioThread, pipe_reader, pipe_name));
  58   return pipe_reader;
  59 }
  60
  61 void AudioPipeReader::StartOnAudioThread(const base::FilePath& pipe_name) {
  62   DCHECK(task_runner_->BelongsToCurrentThread());
  63
  64   pipe_fd_ = HANDLE_EINTR(open(
  65       pipe_name.value().c_str(), O_RDONLY | O_NONBLOCK));
  66   if (pipe_fd_ < 0) {
  67     LOG(ERROR) << "Failed to open " << pipe_name.value();
  68     return;
  69   }
  70
  71   // Set buffer size for the pipe.
  72   int result = HANDLE_EINTR(
  73       fcntl(pipe_fd_, F_SETPIPE_SZ, kPipeBufferSizeBytes));
  74   if (result < 0) {
  75     PLOG(ERROR) << "fcntl";
  76   }
  77
  78   WaitForPipeReadable();
  79 }
  80
  81 AudioPipeReader::AudioPipeReader(
  82     scoped_refptr<base::SingleThreadTaskRunner> task_runner)
  83     : task_runner_(task_runner),
  84       observers_(new ObserverListThreadSafe<StreamObserver>()) {
  85 }
  86
  87 AudioPipeReader::~AudioPipeReader() {
  88 }
  89
  90 void AudioPipeReader::AddObserver(StreamObserver* observer) {
  91   observers_->AddObserver(observer);
  92 }
  93 void AudioPipeReader::RemoveObserver(StreamObserver* observer) {
  94   observers_->RemoveObserver(observer);
  95 }
  96
  97 void AudioPipeReader::OnFileCanReadWithoutBlocking(int fd) {
  98   DCHECK_EQ(fd, pipe_fd_);
  99   StartTimer();
 100 }
 101
 102 void AudioPipeReader::OnFileCanWriteWithoutBlocking(int fd) {
 103   NOTREACHED();
 104 }
 105
 106 void AudioPipeReader::StartTimer() {
 107   DCHECK(task_runner_->BelongsToCurrentThread());
 108   started_time_ = base::TimeTicks::Now();
 109   last_capture_position_ = 0;
 110   timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(kCapturingPeriodMs),
 111                this, &AudioPipeReader::DoCapture);
 112 }
 113
 114 void AudioPipeReader::DoCapture() {
 115   DCHECK(task_runner_->BelongsToCurrentThread());
 116   DCHECK_GT(pipe_fd_, 0);
 117
 118   // Calculate how much we need read from the pipe. Pulseaudio doesn't control
 119   // how much data it writes to the pipe, so we need to pace the stream, so
 120   // that we read the exact number of the samples per second we need.
 121   base::TimeDelta stream_position = base::TimeTicks::Now() - started_time_;
 122   int64 stream_position_bytes = stream_position.InMilliseconds() *
 123       kSampleBytesPerSecond / base::Time::kMillisecondsPerSecond;
 124   int64 bytes_to_read = stream_position_bytes - last_capture_position_;
 125
 126   std::string data = left_over_bytes_;
 127   size_t pos = data.size();
 128   left_over_bytes_.clear();
 129   data.resize(pos + bytes_to_read);
 130
 131   while (pos < data.size()) {
 132     int read_result = HANDLE_EINTR(
 133        read(pipe_fd_, string_as_array(&data) + pos, data.size() - pos));
 134     if (read_result > 0) {
 135       pos += read_result;
 136     } else {
 137       if (read_result < 0 && errno != EWOULDBLOCK && errno != EAGAIN)
 138         PLOG(ERROR) << "read";
 139       break;
 140     }
 141   }
 142
 143   // Stop reading from the pipe if PulseAudio isn't writing anything.
 144   if (pos == 0) {
 145     WaitForPipeReadable();
 146     return;
 147   }
 148
 149   // Save any incomplete samples we've read for later. Each packet should
 150   // contain integer number of samples.
 151   int incomplete_samples_bytes = pos % (kChannels * kBytesPerSample);
 152   left_over_bytes_.assign(data, pos - incomplete_samples_bytes,
 153                           incomplete_samples_bytes);
 154   data.resize(pos - incomplete_samples_bytes);
 155
 156   last_capture_position_ += data.size();
 157   // Normally PulseAudio will keep pipe buffer full, so we should always be able
 158   // to read |bytes_to_read| bytes, but in case it's misbehaving we need to make
 159   // sure that |stream_position_bytes| doesn't go out of sync with the current
 160   // stream position.
 161   if (stream_position_bytes - last_capture_position_ > kPipeBufferSizeBytes)
 162     last_capture_position_ = stream_position_bytes - kPipeBufferSizeBytes;
 163   DCHECK_LE(last_capture_position_, stream_position_bytes);
 164
 165   // Dispatch asynchronous notification to the stream observers.
 166   scoped_refptr<base::RefCountedString> data_ref =
 167       base::RefCountedString::TakeString(&data);
 168   observers_->Notify(&StreamObserver::OnDataRead, data_ref);
 169 }
 170
 171 void AudioPipeReader::WaitForPipeReadable() {
 172   timer_.Stop();
 173   base::MessageLoopForIO::current()->WatchFileDescriptor(
 174       pipe_fd_,
 175       false,
 176       base::MessageLoopForIO::WATCH_READ,
 177       &file_descriptor_watcher_,
 178       this);
 179 }
 180
 181 // static
 182 void AudioPipeReaderTraits::Destruct(const AudioPipeReader* audio_pipe_reader) {
 183   audio_pipe_reader->task_runner_->DeleteSoon(FROM_HERE, audio_pipe_reader);
 184 }
 185
 186 }  // namespace remoting