[Chromecast] Process streams with different post-processing.

chromecast/media/cma/backend/alsa/stream_mixer_alsa.cc

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "chromecast/media/cma/backend/alsa/stream_mixer_alsa.h"
 
 #include <algorithm>
 #include <cmath>
 #include <limits>
+#include <unordered_set>
 #include <utility>
 
 #include "base/bind_helpers.h"
 #include "base/command_line.h"
 #include "base/lazy_instance.h"
+#include "base/memory/ptr_util.h"
 #include "base/memory/weak_ptr.h"
+#include "base/numerics/saturated_arithmetic.h"
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/threading/platform_thread.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "chromecast/base/chromecast_switches.h"
+#include "chromecast/media/base/audio_device_ids.h"
 #include "chromecast/media/cma/backend/alsa/alsa_wrapper.h"
 #include "chromecast/media/cma/backend/alsa/audio_filter_factory.h"
+#include "chromecast/media/cma/backend/alsa/filter_group.h"
 #include "chromecast/media/cma/backend/alsa/stream_mixer_alsa_input_impl.h"
+#include "media/audio/audio_device_description.h"
 #include "media/base/audio_bus.h"
 #include "media/base/media_switches.h"
 
 #define RETURN_REPORT_ERROR(snd_func, ...)                        \
   do {                                                            \
     int err = alsa_->snd_func(__VA_ARGS__);                       \
     if (err < 0) {                                                \
       LOG(ERROR) << #snd_func " error: " << alsa_->StrError(err); \
       SignalError();                                              \
       return;                                                     \
@@ skipping 71 matching lines @@
 // direction they adjusted the requested parameter in, but since we read the
 // output param and then log the information, this module doesn't need to get
 // the direction explicitly.
 static int* kAlsaDirDontCare = nullptr;
 
 // These sample formats will be tried in order. 32 bit samples is ideal, but
 // some devices do not support 32 bit samples.
 const snd_pcm_format_t kPreferredSampleFormats[] = {SND_PCM_FORMAT_S32,
                                                     SND_PCM_FORMAT_S16};
 
-// How many seconds of silence should be passed to the filters to flush them.
-const float kSilenceSecondsToFilter = 1.0f;
-
 const int64_t kNoTimestamp = std::numeric_limits<int64_t>::min();
 
 int64_t TimespecToMicroseconds(struct timespec time) {
   return static_cast<int64_t>(time.tv_sec) *
              base::Time::kMicrosecondsPerSecond +
          time.tv_nsec / 1000;
 }
 
 bool GetSwitchValueAsInt(const std::string& switch_name,
                          int default_value,
@@ skipping 32 matching lines @@
   }
 
   if (*value < 0) {
     LOG(DFATAL) << "--" << switch_name << " must have a non-negative value";
     *value = default_value;
     return false;
   }
   return true;
 }
 
+void VectorAccumulate(const int32_t* source, size_t size, int32_t* dest) {
+  for (size_t i = 0; i < size; ++i) {
+    dest[i] = base::SaturatedAddition(source[i], dest[i]);
+  }
+}
+
 class StreamMixerAlsaInstance : public StreamMixerAlsa {
  public:
   StreamMixerAlsaInstance() {}
   ~StreamMixerAlsaInstance() override {}
 
  private:
   DISALLOW_COPY_AND_ASSIGN(StreamMixerAlsaInstance);
 };
 
 base::LazyInstance<StreamMixerAlsaInstance> g_mixer_instance =
@@ skipping 58 matching lines @@
     LOG(INFO) << "Setting fixed sample rate to " << fixed_samples_per_second;
   }
 
   fixed_output_samples_per_second_ = fixed_samples_per_second;
 
   low_sample_rate_cutoff_ =
       chromecast::GetSwitchValueBoolean(switches::kAlsaEnableUpsampling, false)
           ? kLowSampleRateCutoff
           : 0;
 
-  // Create filters
-  pre_loopback_filter_ = AudioFilterFactory::MakeAudioFilter(
-      AudioFilterFactory::PRE_LOOPBACK_FILTER);
-  post_loopback_filter_ = AudioFilterFactory::MakeAudioFilter(
-      AudioFilterFactory::POST_LOOPBACK_FILTER);
+  // Create filter groups.
+  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+      std::unordered_set<std::string>(
+          {::media::AudioDeviceDescription::kCommunicationsDeviceId}),
+      AudioFilterFactory::COMMUNICATION_AUDIO_FILTER));
+  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+      std::unordered_set<std::string>({kAlarmAudioDeviceId}),
+      AudioFilterFactory::ALARM_AUDIO_FILTER));
+  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+      std::unordered_set<std::string>({kTtsAudioDeviceId}),
+      AudioFilterFactory::TTS_AUDIO_FILTER));
+  filter_groups_.push_back(base::MakeUnique<FilterGroup>(
+      std::unordered_set<std::string>(
+          {::media::AudioDeviceDescription::kDefaultDeviceId,
+           kEarconAudioDeviceId}),
+      AudioFilterFactory::MEDIA_AUDIO_FILTER));
 
   DefineAlsaParameters();
 }
 
 void StreamMixerAlsa::ResetTaskRunnerForTest() {
   mixer_task_runner_ = base::ThreadTaskRunnerHandle::Get();
 }
 
 void StreamMixerAlsa::DefineAlsaParameters() {
   // Get the ALSA output configuration from the command line.
@@ skipping 259 matching lines @@
     int err = SetAlsaPlaybackParams();
     if (err < 0) {
       LOG(ERROR) << "Error setting ALSA playback parameters: "
                  << alsa_->StrError(err);
       SignalError();
       return;
     }
   }
 
   // Initialize filters
-  if (pre_loopback_filter_) {
-    pre_loopback_filter_->SetSampleRateAndFormat(
-        output_samples_per_second_, ::media::SampleFormat::kSampleFormatS32);
-  }
-
-  if (post_loopback_filter_) {
-    post_loopback_filter_->SetSampleRateAndFormat(
-        output_samples_per_second_, ::media::SampleFormat::kSampleFormatS32);
+  for (auto&& filter_group : filter_groups_) {
+    filter_group->Initialize(output_samples_per_second_,
+                             ::media::SampleFormat::kSampleFormatS32);
   }
 
   RETURN_REPORT_ERROR(PcmPrepare, pcm_);
   RETURN_REPORT_ERROR(PcmStatusMalloc, &pcm_status_);
 
   rendering_delay_.timestamp_microseconds = kNoTimestamp;
   rendering_delay_.delay_microseconds = 0;
 
   state_ = kStateNormalPlayback;
 }
@@ skipping 78 matching lines @@
 }
 
 void StreamMixerAlsa::AddInput(std::unique_ptr<InputQueue> input) {
   RUN_ON_MIXER_THREAD(&StreamMixerAlsa::AddInput,
                       base::Passed(std::move(input)));
   if (!alsa_) {
     alsa_.reset(new AlsaWrapper());
   }
 
   DCHECK(input);
+
+  LOG(INFO) << __FUNCTION__;

[kmackay, 2017/02/24 18:03:55]

Necessary logging?

[bshaya, 2017/02/27 17:00:15]

Done.

+
   // If the new input is a primary one, we may need to change the output
   // sample rate to match its input sample rate.
   // We only change the output rate if it is not set to a fixed value.
   if (input->primary() &&
       fixed_output_samples_per_second_ == kInvalidSampleRate) {
     CheckChangeOutputRate(input->input_samples_per_second());
   }
 
   check_close_timer_->Stop();
-  if (state_ == kStateUninitialized) {
-    requested_output_samples_per_second_ = input->input_samples_per_second();
-    Start();
-    input->Initialize(rendering_delay_);
-    inputs_.push_back(std::move(input));
-  } else if (state_ == kStateNormalPlayback) {
-    input->Initialize(rendering_delay_);
-    inputs_.push_back(std::move(input));
-  } else {
-    input->SignalError(StreamMixerAlsaInput::MixerError::kInternalError);
-    ignored_inputs_.push_back(std::move(input));
+  bool found_filter_group = false;

[kmackay, 2017/02/24 18:03:55]

Move down into the case statement? (you'll need braces)

[bshaya, 2017/02/27 17:00:15]

Done.

+  switch (state_) {
+    case kStateUninitialized:
+      requested_output_samples_per_second_ = input->input_samples_per_second();
+      Start();
+    // Fallthrough intended
+    case kStateNormalPlayback:
+      input->Initialize(rendering_delay_);
+      for (auto&& filter_group : filter_groups_) {
+        if (filter_group->CanProcessInput(input.get())) {
+          found_filter_group = true;
+          input->set_filter_group(filter_group.get());
+          break;
+        }
+      }
+      DCHECK(found_filter_group) << "Could not find a filter group for "
+                                 << input->device_id();
+      inputs_.push_back(std::move(input));
+      break;
+    case kStateError:
+      input->SignalError(StreamMixerAlsaInput::MixerError::kInternalError);
+      ignored_inputs_.push_back(std::move(input));
+      break;
+    default:
+      NOTREACHED();
   }
 }
 
 void StreamMixerAlsa::CheckChangeOutputRate(int input_samples_per_second) {
   DCHECK(mixer_task_runner_->BelongsToCurrentThread());
   if (!pcm_ ||
       input_samples_per_second == requested_output_samples_per_second_ ||
       input_samples_per_second == output_samples_per_second_ ||
       input_samples_per_second < static_cast<int>(low_sample_rate_cutoff_)) {
     return;
@@ skipping 87 matching lines @@
   retry_write_frames_timer_->Stop();
   if (TryWriteFrames()) {
     retry_write_frames_timer_->Start(
         FROM_HERE, base::TimeDelta(),
         base::Bind(&StreamMixerAlsa::WriteFrames, base::Unretained(this)));
   }
 }
 
 bool StreamMixerAlsa::TryWriteFrames() {
   DCHECK(mixer_task_runner_->BelongsToCurrentThread());
+  DCHECK_GE(filter_groups_.size(), 1u);
+
   if (state_ != kStateNormalPlayback) {
     return false;
   }
 
   const int min_frames_in_buffer =
       output_samples_per_second_ * kMinBufferedDataMs / 1000;
   int chunk_size = output_samples_per_second_ * kMaxWriteSizeMs / 1000;
-  std::vector<InputQueue*> active_inputs;
+  bool is_silence = true;
   for (auto&& input : inputs_) {
     int read_size = input->MaxReadSize();
     if (read_size > 0) {
-      active_inputs.push_back(input.get());
+      DCHECK(input->filter_group());
+      input->filter_group()->AddActiveInput(input.get());
       chunk_size = std::min(chunk_size, read_size);
+      is_silence = false;
     } else if (input->primary()) {
       if (alsa_->PcmStatus(pcm_, pcm_status_) != 0) {
         LOG(ERROR) << "Failed to get status";
         return false;
       }
 
       int frames_in_buffer =
           alsa_buffer_size_ - alsa_->PcmStatusGetAvail(pcm_status_);
       if (alsa_->PcmStatusGetState(pcm_status_) == SND_PCM_STATE_XRUN ||
           frames_in_buffer < min_frames_in_buffer) {
         // If there has been (or soon will be) an underrun, continue without the
         // empty primary input stream.
         input->OnSkipped();
         continue;
       }
 
       // A primary input cannot provide any data, so wait until later.
       retry_write_frames_timer_->Start(
           FROM_HERE, base::TimeDelta::FromMilliseconds(kMinBufferedDataMs / 2),
           base::Bind(&StreamMixerAlsa::WriteFrames, base::Unretained(this)));
       return false;
     } else {
       input->OnSkipped();
     }
   }
 
-  if (active_inputs.empty()) {
-    // No inputs have any data to provide. Fill with silence to avoid underrun.
+  if (is_silence) {
+    // No inputs have any data to provide. Push silence to prevent underrun.
     chunk_size = kPreventUnderrunChunkSize;
-    if (!mixed_ || mixed_->frames() < chunk_size) {
-      mixed_ = ::media::AudioBus::Create(kNumOutputChannels, chunk_size);
-    }
-
-    mixed_->Zero();
-    WriteMixedPcm(*mixed_, chunk_size, true /* is_silence */);
-    return true;
   }
 
-  // If |mixed_| has not been allocated, or it is too small, allocate a buffer.
-  if (!mixed_ || mixed_->frames() < chunk_size) {
-    mixed_ = ::media::AudioBus::Create(kNumOutputChannels, chunk_size);
-  }
-
-  // If |temp_| has not been allocated, or is too small, allocate a buffer.
-  if (!temp_ || temp_->frames() < chunk_size) {
-    temp_ = ::media::AudioBus::Create(kNumOutputChannels, chunk_size);
-  }
-
-  mixed_->ZeroFramesPartial(0, chunk_size);
-
-  // Loop through active inputs, polling them for data, and mixing them.
-  for (InputQueue* input : active_inputs) {
-    input->GetResampledData(temp_.get(), chunk_size);
-    for (int c = 0; c < kNumOutputChannels; ++c) {
-      input->VolumeScaleAccumulate(c, temp_->channel(c), chunk_size,
-                                   mixed_->channel(c));
+  // Mix and filter each group.
+  std::vector<uint8_t>* interleaved = nullptr;
+  for (auto&& filter_group : filter_groups_) {
+    if (filter_group->MixAndFilter(chunk_size)) {
+      if (!interleaved) {
+        interleaved = filter_group->GetInterleaved();
+      } else {
+        DCHECK_EQ(4, BytesPerOutputFormatSample());
+        VectorAccumulate(
+            reinterpret_cast<int32_t*>(filter_group->GetInterleaved()->data()),
+            chunk_size * kNumOutputChannels,
+            reinterpret_cast<int32_t*>(interleaved->data()));
+      }
     }
   }
 
-  WriteMixedPcm(*mixed_, chunk_size, false /* is_silence */);
+  if (!interleaved) {
+    // No group has any data, write empty buffer.
+    filter_groups_[0]->ClearInterleaved(chunk_size);
+    interleaved = filter_groups_[0]->GetInterleaved();
+  }
+
+  WriteMixedPcm(interleaved, chunk_size);
   return true;
 }
 
+size_t StreamMixerAlsa::InterleavedSize(int frames) {
+  return BytesPerOutputFormatSample() *
+         static_cast<size_t>(frames * kNumOutputChannels);
+}
+
 ssize_t StreamMixerAlsa::BytesPerOutputFormatSample() {
   return alsa_->PcmFormatSize(pcm_format_, 1);
 }
 
-void StreamMixerAlsa::WriteMixedPcm(const ::media::AudioBus& mixed,
-                                    int frames, bool is_silence) {
+void StreamMixerAlsa::WriteMixedPcm(std::vector<uint8_t>* interleaved,
+                                    int frames) {
   DCHECK(mixer_task_runner_->BelongsToCurrentThread());
   CHECK_PCM_INITIALIZED();
-
-  size_t interleaved_size = static_cast<size_t>(frames * kNumOutputChannels) *
-                            BytesPerOutputFormatSample();
-  if (interleaved_.size() < interleaved_size) {
-    interleaved_.resize(interleaved_size);
-  }
+  DCHECK(interleaved);
+  DCHECK_GE(interleaved->size(), InterleavedSize(frames));
 
   int64_t expected_playback_time;
   if (rendering_delay_.timestamp_microseconds == kNoTimestamp) {
     expected_playback_time = kNoTimestamp;
   } else {
     expected_playback_time = rendering_delay_.timestamp_microseconds +
                              rendering_delay_.delay_microseconds;
   }
 
-  mixed.ToInterleaved(frames, BytesPerOutputFormatSample(),
-                      interleaved_.data());
-
-  // Ensure that, on onset of silence, at least |kSilenceSecondsToFilter|
-  // second of audio get pushed through the filters to clear any memory.
-  bool filter_frames = true;
-  if (is_silence) {
-    int silence_frames_to_filter =
-        output_samples_per_second_ * kSilenceSecondsToFilter;
-    if (silence_frames_filtered_ < silence_frames_to_filter) {
-      silence_frames_filtered_ += frames;
-    } else {
-      filter_frames = false;
-    }
-  } else {
-    silence_frames_filtered_ = 0;
-  }
-
-  // Filter, send to observers, and post filter
-  if (pre_loopback_filter_ && filter_frames) {
-    pre_loopback_filter_->ProcessInterleaved(interleaved_.data(), frames);
-  }
-
   for (CastMediaShlib::LoopbackAudioObserver* observer : loopback_observers_) {
     observer->OnLoopbackAudio(expected_playback_time, kSampleFormatS32,
                               output_samples_per_second_, kNumOutputChannels,
-                              interleaved_.data(), interleaved_size);
-  }
-
-  if (post_loopback_filter_ && filter_frames) {
-    post_loopback_filter_->ProcessInterleaved(interleaved_.data(), frames);
+                              interleaved->data(), InterleavedSize(frames));
   }
 
   // If the PCM has been drained it will be in SND_PCM_STATE_SETUP and need
   // to be prepared in order for playback to work.
   if (alsa_->PcmState(pcm_) == SND_PCM_STATE_SETUP) {
     RETURN_REPORT_ERROR(PcmPrepare, pcm_);
   }
 
   int frames_left = frames;
-  uint8_t* data = &interleaved_[0];
+  uint8_t* data = interleaved->data();
   while (frames_left) {
     int frames_or_error;
     while ((frames_or_error = alsa_->PcmWritei(pcm_, data, frames_left)) < 0) {
       for (auto* observer : loopback_observers_) {
         observer->OnLoopbackInterrupted();
       }
       RETURN_REPORT_ERROR(PcmRecover, pcm_, frames_or_error,
                           kPcmRecoverIsSilent);
     }
     frames_left -= frames_or_error;
@@ skipping 41 matching lines @@
   DCHECK(std::find(loopback_observers_.begin(), loopback_observers_.end(),
                    observer) != loopback_observers_.end());
   loopback_observers_.erase(std::remove(loopback_observers_.begin(),
                                         loopback_observers_.end(), observer),
                             loopback_observers_.end());
   observer->OnRemoved();
 }
 
 }  // namespace media
 }  // namespace chromecast