[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 <utility>
 
 #include "base/bind_helpers.h"
 #include "base/command_line.h"
 #include "base/lazy_instance.h"
 #include "base/memory/weak_ptr.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/cma/backend/alsa/alsa_wrapper.h"
 #include "chromecast/media/cma/backend/alsa/audio_filter_factory.h"
 #include "chromecast/media/cma/backend/alsa/stream_mixer_alsa_input_impl.h"
+#include "chromecast/public/media/audio_device_ids.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 76 matching lines @@
 // 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();
 
+const AudioFilterFactory::FilterType kFilterTypes[kNumFilterGroups] = {
+    AudioFilterFactory::MEDIA_FILTER, AudioFilterFactory::SYSTEM_AUDIO_FILTER};
+
+int GetFilterGroup(std::string stream_name) {
+  if (stream_name == kSystemAudioDeviceId) {
+    return 1;
+  }
+  return 0;
+}
+
 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,
                          int* value) {
   DCHECK(value);
@@ skipping 30 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] += source[i];
+  }
+}
+
 class StreamMixerAlsaInstance : public StreamMixerAlsa {
  public:
   StreamMixerAlsaInstance() {}
   ~StreamMixerAlsaInstance() override {}
 
  private:
   DISALLOW_COPY_AND_ASSIGN(StreamMixerAlsaInstance);
 };
 
 base::LazyInstance<StreamMixerAlsaInstance> g_mixer_instance =
@@ skipping 59 matching lines @@
   }
 
   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);
+  for (int filter = 0; filter < kNumFilterGroups; ++filter) {
+    pre_loopback_filter_[filter] =
+        AudioFilterFactory::MakeAudioFilter(kFilterTypes[filter]);
+  }
   post_loopback_filter_ = AudioFilterFactory::MakeAudioFilter(
       AudioFilterFactory::POST_LOOPBACK_FILTER);
 
   DefineAlsaParameters();
 }
 
 void StreamMixerAlsa::ResetTaskRunnerForTest() {
   mixer_task_runner_ = base::ThreadTaskRunnerHandle::Get();
 }
 
@@ skipping 261 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);
+  for (int filter = 0; filter < kNumFilterGroups; ++filter) {
+    if (pre_loopback_filter_[filter]) {
+      pre_loopback_filter_[filter]->SetSampleRateAndFormat(
+          output_samples_per_second_, ::media::SampleFormat::kSampleFormatS32);
+    }
   }
 
   if (post_loopback_filter_) {
     post_loopback_filter_->SetSampleRateAndFormat(
         output_samples_per_second_, ::media::SampleFormat::kSampleFormatS32);
   }
 
   RETURN_REPORT_ERROR(PcmPrepare, pcm_);
   RETURN_REPORT_ERROR(PcmStatusMalloc, &pcm_status_);
 
@@ skipping 217 matching lines @@
 
 bool StreamMixerAlsa::TryWriteFrames() {
   DCHECK(mixer_task_runner_->BelongsToCurrentThread());
   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;
+  std::vector<InputQueue*> active_inputs[kNumFilterGroups];
+  bool is_silence = true;
   for (auto&& input : inputs_) {
     int read_size = input->MaxReadSize();
     if (read_size > 0) {
-      active_inputs.push_back(input.get());
+      active_inputs[GetFilterGroup(input->name())].push_back(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;
+    ResizeBuffersIfNecessary(chunk_size);
+    memset(interleaved_.data(), 0,
+           static_cast<size_t>(chunk_size * kNumOutputChannels) *
+               BytesPerOutputFormatSample());
+  } else {
+    ResizeBuffersIfNecessary(chunk_size);
   }
 
-  // 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 |mixed_|, |temp_|, |interleaved_|, or |interleaved_intermediate_| have
+  // not been allocated, or are too small, allocate a buffer.
+
+  // Mix each group, pass through corresponding audio filter, and
+  // accumulate into |interleaved_|.
+  bool filter_frames = false;
+  for (int filter_group = 0; filter_group < kNumFilterGroups; ++filter_group) {
+    filter_frames =
+        filter_frames ||
+        MixAndFilterGroup(active_inputs[filter_group], filter_group, chunk_size,
+                          filter_frames /* accumulate */);
   }
 
-  // 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);
-  }
+  WriteMixedPcm(chunk_size, filter_frames);
+  return true;
+}
 
-  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));
+bool StreamMixerAlsa::MixAndFilterGroup(
+    const std::vector<InputQueue*>& active_inputs,
+    int filter_group,
+    int frames,
+    bool accumulate) {
+  // Mix into group buffer, |mixed_|.
+  mixed_->ZeroFramesPartial(0, frames);
+  bool is_silence = true;
+  if (!active_inputs.empty()) {
+    is_silence = false;
+    // Loop through active inputs, polling them for data, and mixing them.
+    for (InputQueue* input : active_inputs) {
+      input->GetResampledData(temp_.get(), frames);
+      for (int c = 0; c < kNumOutputChannels; ++c) {
+        input->VolumeScaleAccumulate(c, temp_->channel(c), frames,
+                                     mixed_->channel(c));
+      }
     }
   }
 
-  WriteMixedPcm(*mixed_, chunk_size, false /* is_silence */);
-  return true;
+  // If no data has been written to |interleaved_| thus far,
+  // we can write to |interleaved_| directly, skipping a vector addition step.
+  std::vector<uint8_t>* interleaved_group = &interleaved_intermediate_;
+  if (!accumulate) {
+    interleaved_group = &interleaved_;
+  }
+
+  // Convert to interleaved for post-processing.
+  mixed_->ToInterleaved(frames, BytesPerOutputFormatSample(),
+                        interleaved_group->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;

[kmackay, 2017/02/17 06:11:25]

The design for mixing seems a bit clunky. I'd rather have something like:
 * Have a FIlterGroup class that holds state for each filter group.
StreamMixerAlsa will have a vector of these.
 * When iterating through the active inputs, loop through the FilterGroups
calling AddActiveInput() until one of them returns true (ie, the FilterGroup
determines which inputs it cares about, based on the name).
 * Once all inputs have been iterated through, call Execute() or something on
every FilterGroup. This will mix all of the input that that group accepted to a
buffer that is internal to the FilterGroup, and then run the filter on it (if
any). Execute() returns false if there is no output from that filter group.
 * If no filter groups have output, push silence. If only 1 filter group has
output, push that group's buffer. Otherwise, mix the buffers of all filter
groups that have output together, and push the result.

Then it's easy to also do volume control at the same time.

[bshaya, 2017/02/17 18:26:53]

Hmm, how about that, and:
* Have an InputGroup class that contains a list of inputs, a filter, filter
state, and necessary buffers.
* When an InputQueue is added, it is added to the appropriate group directly
(similar to the Alsa implementation).
* Call InputGroup->Update to check for ChunkSize and blocked Primary inputs.
* Execute and mix as you suggested.
At the same time as what? If you mean stream volume control, I'd rather not,
since that either a) adds an additional level of mixing (streams->groups->filter
groups->everything) or requires duplicate filters (since we have more groups
than filter types).

I guess the ideal would be for the architecture to be completely modular
(InputGroup implements InputQueue so that they can be fed into each other, and
you just call GetInterleavedData from the final InputGroup when you want data.
That might be too much re-designing considering we want to remove all of this...

[kmackay, 2017/02/17 19:42:42]

Your suggestions sound good. I meant for the per-stream-type volume control that
I am planning to implement. I want to be able to implement that cleanly here in
case we aren't able to get the ALSA plugins working quickly enough.

[bshaya, 2017/02/21 23:30:13]

Done.

+  if (is_silence) {
+    int silence_frames_to_filter =
+        output_samples_per_second_ * kSilenceSecondsToFilter;
+    if (silence_frames_filtered_[filter_group] < silence_frames_to_filter) {
+      silence_frames_filtered_[filter_group] += frames;
+    } else {
+      return false;  // Output will be silence, no need to mix.
+    }
+  } else {
+    silence_frames_filtered_[filter_group] = 0;
+  }
+
+  // Filter the mixed group.
+  if (pre_loopback_filter_[filter_group] && filter_frames) {
+    pre_loopback_filter_[filter_group]->ProcessInterleaved(
+        interleaved_group->data(), frames);
+  }
+
+  // Exit if we already wrote to |interleaved_|.
+  if (!accumulate) {
+    return filter_frames;
+  }
+
+  // Mix into final output buffer, |interleaved_|
+  DCHECK_EQ(4, BytesPerOutputFormatSample());
+  VectorAccumulate(reinterpret_cast<int32_t*>(interleaved_group->data()),
+                   frames * kNumOutputChannels,
+                   reinterpret_cast<int32_t*>(interleaved_.data()));
+  return filter_frames;
 }
 
 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(int frames, bool filter_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);
-  }
-
   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);
-  }
-
+  size_t interleaved_size = static_cast<size_t>(frames * kNumOutputChannels) *

[kmackay, 2017/02/17 06:11:25]

use int instead of size_t

[bshaya, 2017/02/17 18:26:54]

See the first point in "Types" in
https://chromium.googlesource.com/chromium/src/+/master/styleguide/c++/c++.md

size_t is correct for vector size/indices, int is considered unsafe.

+                            BytesPerOutputFormatSample();
   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) {

[kmackay, 2017/02/17 06:11:25]

Do we ever use post_loopback_filter? If not, let's get rid of it. Then
filter_frames isn't necessary either.

[bshaya, 2017/02/17 18:26:54]

Done.

     post_loopback_filter_->ProcessInterleaved(interleaved_.data(), 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;
@@ skipping 50 matching lines @@
     CastMediaShlib::LoopbackAudioObserver* observer) {
   RUN_ON_MIXER_THREAD(&StreamMixerAlsa::RemoveLoopbackAudioObserver, observer);
   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();
 }
 
+void StreamMixerAlsa::ResizeBuffersIfNecessary(int chunk_size) {
+  if (!mixed_ || mixed_->frames() < chunk_size) {
+    mixed_ = ::media::AudioBus::Create(kNumOutputChannels, chunk_size);
+  }
+  if (!temp_ || temp_->frames() < chunk_size) {
+    temp_ = ::media::AudioBus::Create(kNumOutputChannels, chunk_size);
+  }
+
+  size_t interleaved_size =
+      static_cast<size_t>(chunk_size * kNumOutputChannels) *
+      BytesPerOutputFormatSample();
+  if (interleaved_.size() < interleaved_size) {
+    interleaved_.resize(interleaved_size);
+  }
+  if (interleaved_intermediate_.size() < interleaved_size) {
+    interleaved_intermediate_.resize(interleaved_size);
+  }
+}
+
 }  // namespace media
 }  // namespace chromecast