Lock scheme #8: Introduced the new locking scheme

webrtc/modules/audio_processing/audio_processing_impl.h

 /*
  *  Copyright (c) 2012 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
 #ifndef WEBRTC_MODULES_AUDIO_PROCESSING_AUDIO_PROCESSING_IMPL_H_
 #define WEBRTC_MODULES_AUDIO_PROCESSING_AUDIO_PROCESSING_IMPL_H_
 
 #include <list>
 #include <string>
 #include <vector>
 
+#include "webrtc/base/criticalsection.h"
 #include "webrtc/base/scoped_ptr.h"
 #include "webrtc/base/thread_annotations.h"
 #include "webrtc/base/thread_checker.h"
+#include "webrtc/modules/audio_processing/audio_buffer.h"
 #include "webrtc/modules/audio_processing/include/audio_processing.h"
+#include "webrtc/system_wrappers/include/file_wrapper.h"
+
+#ifdef WEBRTC_AUDIOPROC_DEBUG_DUMP
+// Files generated at build-time by the protobuf compiler.
+#ifdef WEBRTC_ANDROID_PLATFORM_BUILD
+#include "external/webrtc/webrtc/modules/audio_processing/debug.pb.h"
+#else
+#include "webrtc/audio_processing/debug.pb.h"
+#endif
+#endif  // WEBRTC_AUDIOPROC_DEBUG_DUMP
 
 namespace webrtc {
 
 class AgcManagerDirect;
-class AudioBuffer;
 class AudioConverter;
 
 template<typename T>
 class Beamformer;
 
-class CriticalSectionWrapper;
+struct ApmPublicSubmodules;
+struct ApmPrivateSubmodules;
 class EchoCancellationImpl;
 class EchoControlMobileImpl;
-class FileWrapper;
 class GainControlImpl;
 class GainControlForNewAgc;
 class HighPassFilterImpl;
 class LevelEstimatorImpl;
 class NoiseSuppressionImpl;
 class ProcessingComponent;
 class TransientSuppressor;
 class VoiceDetectionImpl;
 class IntelligibilityEnhancer;
 
 #ifdef WEBRTC_AUDIOPROC_DEBUG_DUMP
 namespace audioproc {
 
 class Event;
 
 }  // namespace audioproc
+
+// State for the debug dump.
+struct ApmDebugDumpThreadState {
+  ApmDebugDumpThreadState() : event_msg(new audioproc::Event()) {}
+  rtc::scoped_ptr<audioproc::Event> event_msg;  // Protobuf message.
+  std::string event_str;  // Memory for protobuf serialization.
+
+  // Serialized string of last saved APM configuration.
+  std::string last_serialized_config;
+};
+
+struct ApmDebugDumpState {
+  ApmDebugDumpState() : debug_file(FileWrapper::Create()) {}
+  rtc::scoped_ptr<FileWrapper> debug_file;
+  ApmDebugDumpThreadState render;
+  ApmDebugDumpThreadState capture;
+};
+
 #endif
 
 class AudioProcessingImpl : public AudioProcessing {
  public:
+  // Methods forcing APM to run in a single-threaded manner.
+  // Acquires both the render and capture locks.
   explicit AudioProcessingImpl(const Config& config);
-
   // AudioProcessingImpl takes ownership of beamformer.
   AudioProcessingImpl(const Config& config, Beamformer<float>* beamformer);
   virtual ~AudioProcessingImpl();
-
-  // AudioProcessing methods.
   int Initialize() override;
   int Initialize(int input_sample_rate_hz,
                  int output_sample_rate_hz,
                  int reverse_sample_rate_hz,
                  ChannelLayout input_layout,
                  ChannelLayout output_layout,
                  ChannelLayout reverse_layout) override;
   int Initialize(const ProcessingConfig& processing_config) override;
   void SetExtraOptions(const Config& config) override;
-  int proc_sample_rate_hz() const override;
-  int proc_split_sample_rate_hz() const override;
-  int num_input_channels() const override;
-  int num_output_channels() const override;
-  int num_reverse_channels() const override;
-  void set_output_will_be_muted(bool muted) override;
+  void UpdateHistogramsOnCallEnd() override;
+  int StartDebugRecording(const char filename[kMaxFilenameSize]) override;

[kwiberg-webrtc, 2015/11/23 22:15:11]

Why not std::string?

[peah-webrtc, 2015/11/24 21:42:23]

Good point! Not sure. This is something that should be changed, but since it was
there before this work I'd rather do that in the scope of another CL.

+  int StartDebugRecording(FILE* handle) override;
+  int StartDebugRecordingForPlatformFile(rtc::PlatformFile handle) override;
+  int StopDebugRecording() override;
+
+  // Capture-side exclusive methods possibly running APM in a
+  // multi-threaded manner. Acquires the capture lock.

[kwiberg-webrtc, 2015/11/23 22:15:11]

Acquire.

[peah-webrtc, 2015/11/24 21:42:23]

Done.

   int ProcessStream(AudioFrame* frame) override;
   int ProcessStream(const float* const* src,
                     size_t samples_per_channel,
                     int input_sample_rate_hz,
                     ChannelLayout input_layout,
                     int output_sample_rate_hz,
                     ChannelLayout output_layout,
                     float* const* dest) override;
   int ProcessStream(const float* const* src,
                     const StreamConfig& input_config,
                     const StreamConfig& output_config,
                     float* const* dest) override;
+  void set_output_will_be_muted(bool muted) override;
+  int set_stream_delay_ms(int delay) override;
+  void set_delay_offset_ms(int offset) override;
+  int delay_offset_ms() const override;
+  void set_stream_key_pressed(bool key_pressed) override;
+
+  // Render-side exclusive methods possibly running APM in a
+  // multi-threaded manner. Acquires the render lock.

[kwiberg-webrtc, 2015/11/23 22:15:10]

Acquire.

[peah-webrtc, 2015/11/24 21:42:23]

Done.

   int AnalyzeReverseStream(AudioFrame* frame) override;
   int ProcessReverseStream(AudioFrame* frame) override;
   int AnalyzeReverseStream(const float* const* data,
                            size_t samples_per_channel,
                            int sample_rate_hz,
                            ChannelLayout layout) override;
   int ProcessReverseStream(const float* const* src,
                            const StreamConfig& reverse_input_config,
                            const StreamConfig& reverse_output_config,
                            float* const* dest) override;
-  int set_stream_delay_ms(int delay) override;
+
+  // Methods only accessed from APM submodules or
+  // from AudioProcessing tests in a single-threaded manner.
+  // Hence there is no need for locks in these.
+  int proc_sample_rate_hz() const override;
+  int proc_split_sample_rate_hz() const override;
+  int num_input_channels() const override;
+  int num_output_channels() const override;
+  int num_reverse_channels() const override;
   int stream_delay_ms() const override;
-  bool was_stream_delay_set() const override;
-  void set_delay_offset_ms(int offset) override;
-  int delay_offset_ms() const override;
-  void set_stream_key_pressed(bool key_pressed) override;
-  int StartDebugRecording(const char filename[kMaxFilenameSize]) override;
-  int StartDebugRecording(FILE* handle) override;
-  int StartDebugRecordingForPlatformFile(rtc::PlatformFile handle) override;
-  int StopDebugRecording() override;
-  void UpdateHistogramsOnCallEnd() override;
+  bool was_stream_delay_set() const override
+      EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
+
+  // Methods returning pointers to APM submodules.
+  // No locks are aquired in those, as those locks
+  // would offer no protection (the submodules are
+  // created only once in a single-treaded manner
+  // during APM creation).
   EchoCancellation* echo_cancellation() const override;
   EchoControlMobile* echo_control_mobile() const override;
   GainControl* gain_control() const override;
   HighPassFilter* high_pass_filter() const override;
   LevelEstimator* level_estimator() const override;
   NoiseSuppression* noise_suppression() const override;
   VoiceDetection* voice_detection() const override;
 
  protected:
   // Overridden in a mock.
-  virtual int InitializeLocked() EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  virtual int InitializeLocked()
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_, crit_capture_);
 
  private:
+  // Method for modifying the formats struct that are called from both
+  // the render and capture threads. The check for whether modifications
+  // are needed is done while holding the render lock only, thereby avoiding
+  // that the capture thread blocks the render thread.
+  // The struct is modified in a single-threaded manner by holding both the
+  // render and capture locks.
+  int MaybeInitialize(const ProcessingConfig& config)
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_);
+  // Method for checking for the need of conversion. Accesses the formats
+  // structs in a read manner but the requirement for the render lock to be held
+  // was added as it currently anyway is always called in that manner.
+  bool rev_conversion_needed() const EXCLUSIVE_LOCKS_REQUIRED(crit_render_);
+
+  // Methods requiring APM running in a single-threaded manner.
+  // Are called with both the render and capture locks already
+  // acquired.
+  void InitializeExperimentalAgc()
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_, crit_capture_);
+  void InitializeTransient()
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_, crit_capture_);
+  void InitializeBeamformer()
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_, crit_capture_);
+  void InitializeIntelligibility()
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_, crit_capture_);
   int InitializeLocked(const ProcessingConfig& config)
-      EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  int MaybeInitializeLocked(const ProcessingConfig& config)
-      EXCLUSIVE_LOCKS_REQUIRED(crit_);
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_, crit_capture_);
+
+  // Capture-side exclusive methods possibly running APM in a multi-threaded
+  // manner that are called with the render lock already acquired.
+  int ProcessStreamLocked() EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
+  bool output_copy_needed(bool is_data_processed) const
+      EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
+  bool is_data_processed() const EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
+  bool synthesis_needed(bool is_data_processed) const
+      EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
+  bool analysis_needed(bool is_data_processed) const
+      EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
+  void MaybeUpdateHistograms() EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
+
+  // Render-side exclusive methods possibly running APM in a multi-threaded
+  // manner that are called with the render lock already acquired.
   // TODO(ekm): Remove once all clients updated to new interface.
-  int AnalyzeReverseStream(const float* const* src,
-                           const StreamConfig& input_config,
-                           const StreamConfig& output_config);
-  int ProcessStreamLocked() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  int ProcessReverseStreamLocked() EXCLUSIVE_LOCKS_REQUIRED(crit_);
+  int AnalyzeReverseStreamLocked(const float* const* src,
+                                 const StreamConfig& input_config,
+                                 const StreamConfig& output_config)
+      EXCLUSIVE_LOCKS_REQUIRED(crit_render_);
+  bool is_rev_processed() const EXCLUSIVE_LOCKS_REQUIRED(crit_render_);
+  int ProcessReverseStreamLocked() EXCLUSIVE_LOCKS_REQUIRED(crit_render_);
 
-  bool is_data_processed() const;
-  bool output_copy_needed(bool is_data_processed) const;
-  bool synthesis_needed(bool is_data_processed) const;
-  bool analysis_needed(bool is_data_processed) const;
-  bool is_rev_processed() const;
-  bool rev_conversion_needed() const;
-  void InitializeExperimentalAgc() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  void InitializeTransient() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  void InitializeBeamformer() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  void InitializeIntelligibility() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-  void MaybeUpdateHistograms() EXCLUSIVE_LOCKS_REQUIRED(crit_);
-
-  EchoCancellationImpl* echo_cancellation_;
-  EchoControlMobileImpl* echo_control_mobile_;
-  GainControlImpl* gain_control_;
-  HighPassFilterImpl* high_pass_filter_;
-  LevelEstimatorImpl* level_estimator_;
-  NoiseSuppressionImpl* noise_suppression_;
-  VoiceDetectionImpl* voice_detection_;
-  rtc::scoped_ptr<GainControlForNewAgc> gain_control_for_new_agc_;
-
-  std::list<ProcessingComponent*> component_list_;
-  CriticalSectionWrapper* crit_;
-  rtc::ThreadChecker render_thread_checker_;
-  rtc::ThreadChecker capture_thread_checker_;
-  rtc::ThreadChecker signal_thread_checker_;
-  rtc::scoped_ptr<AudioBuffer> render_audio_;
-  rtc::scoped_ptr<AudioBuffer> capture_audio_;
-  rtc::scoped_ptr<AudioConverter> render_converter_;
+// Debug dump methods that are internal and called without locks.
+// TODO(peah): Make thread safe.
 #ifdef WEBRTC_AUDIOPROC_DEBUG_DUMP
   // TODO(andrew): make this more graceful. Ideally we would split this stuff
   // out into a separate class with an "enabled" and "disabled" implementation.
-  int WriteMessageToDebugFile();
-  int WriteInitMessage();
+  static int WriteMessageToDebugFile(FileWrapper* debug_file,
+                                     rtc::CriticalSection* crit_debug,
+                                     ApmDebugDumpThreadState* debug_state);
+  int WriteInitMessage() EXCLUSIVE_LOCKS_REQUIRED(crit_render_, crit_capture_);
 
   // Writes Config message. If not |forced|, only writes the current config if
   // it is different from the last saved one; if |forced|, writes the config
   // regardless of the last saved.
-  int WriteConfigMessage(bool forced);
-
-  rtc::scoped_ptr<FileWrapper> debug_file_;
-  rtc::scoped_ptr<audioproc::Event> event_msg_;  // Protobuf message.
-  std::string event_str_;  // Memory for protobuf serialization.
-
-  // Serialized string of last saved APM configuration.
-  std::string last_serialized_config_;
+  int WriteConfigMessage(bool forced) EXCLUSIVE_LOCKS_REQUIRED(crit_capture_)
+      EXCLUSIVE_LOCKS_REQUIRED(crit_capture_);
 #endif
 
+  // Critical sections and threadcheckers.
+  mutable rtc::CriticalSection crit_render_ ACQUIRED_BEFORE(crit_capture_);
+  mutable rtc::CriticalSection crit_capture_;
+#ifdef WEBRTC_AUDIOPROC_DEBUG_DUMP
+  mutable rtc::CriticalSection crit_debug_;
+#endif
+  rtc::ThreadChecker render_thread_checker_;
+  rtc::ThreadChecker capture_thread_checker_;
+  rtc::ThreadChecker signal_thread_checker_;
+
+  // Structs containing the pointers to the submodules.
+  rtc::scoped_ptr<ApmPublicSubmodules> public_submodules_;
+  rtc::scoped_ptr<ApmPrivateSubmodules> private_submodules_
+      GUARDED_BY(crit_capture_);
+
   // State that is written to while holding both the render and capture locks
-  // but can be read while holding only one of the locks.
-  struct SharedState {
-    SharedState()
+  // but can be read without any lock being held.
+  // As this is only accessed internally of APM, and all internal methods in APM
+  // either are holding the render or capture locks, this construct is safe as
+  // it is not possible to read the variables while writing them.
+  struct ApmFormatState {
+    ApmFormatState()
         :  // Format of processing streams at input/output call sites.
-          api_format_({{{kSampleRate16kHz, 1, false},
-                        {kSampleRate16kHz, 1, false},
-                        {kSampleRate16kHz, 1, false},
-                        {kSampleRate16kHz, 1, false}}}) {}
-    ProcessingConfig api_format_;
-  } shared_state_;
+          api_format({{{kSampleRate16kHz, 1, false},
+                       {kSampleRate16kHz, 1, false},
+                       {kSampleRate16kHz, 1, false},
+                       {kSampleRate16kHz, 1, false}}}),
+          rev_proc_format(kSampleRate16kHz, 1) {}
+    ProcessingConfig api_format;
+    StreamConfig rev_proc_format;
+  } formats_;
 
-  // Only the rate and samples fields of fwd_proc_format_ are used because the
-  // forward processing number of channels is mutable and is tracked by the
-  // capture_audio_.
-  StreamConfig fwd_proc_format_;
-  StreamConfig rev_proc_format_;
-  int split_rate_;
+  // APM constants.
+  const struct ApmConstants {
+    ApmConstants(int agc_startup_min_volume,
+                 const std::vector<Point> array_geometry,
+                 SphericalPointf target_direction,
+                 bool use_new_agc,
+                 bool intelligibility_enabled,
+                 bool beamformer_enabled)
+        :  // Format of processing streams at input/output call sites.
+          agc_startup_min_volume(agc_startup_min_volume),
+          array_geometry(array_geometry),
+          target_direction(target_direction),
+          use_new_agc(use_new_agc),
+          intelligibility_enabled(intelligibility_enabled),
+          beamformer_enabled(beamformer_enabled) {}
+    int agc_startup_min_volume;
+    std::vector<Point> array_geometry;
+    SphericalPointf target_direction;
+    bool use_new_agc;
+    bool intelligibility_enabled;
+    bool beamformer_enabled;
+  } constants_;
 
-  int stream_delay_ms_;
-  int delay_offset_ms_;
-  bool was_stream_delay_set_;
-  int last_stream_delay_ms_;
-  int last_aec_system_delay_ms_;
-  int stream_delay_jumps_;
-  int aec_system_delay_jumps_;
+  struct ApmCaptureState {
+    ApmCaptureState(bool transient_suppressor_enabled)
+        : aec_system_delay_jumps(-1),
+          delay_offset_ms(0),
+          was_stream_delay_set(false),
+          last_stream_delay_ms(0),
+          last_aec_system_delay_ms(0),
+          stream_delay_jumps(-1),
+          output_will_be_muted(false),
+          key_pressed(false),
+          transient_suppressor_enabled(transient_suppressor_enabled),
+          fwd_proc_format(kSampleRate16kHz),
+          split_rate(kSampleRate16kHz) {}
+    int aec_system_delay_jumps;
+    int delay_offset_ms;
+    bool was_stream_delay_set;
+    int last_stream_delay_ms;
+    int last_aec_system_delay_ms;
+    int stream_delay_jumps;
+    bool output_will_be_muted;
+    bool key_pressed;
+    bool transient_suppressor_enabled;
+    rtc::scoped_ptr<AudioBuffer> capture_audio;
+    // Only the rate and samples fields of fwd_proc_format_ are used because the
+    // forward processing number of channels is mutable and is tracked by the
+    // capture_audio_.
+    StreamConfig fwd_proc_format;
+    int split_rate;
+  } capture_ GUARDED_BY(crit_capture_);
 
-  bool output_will_be_muted_ GUARDED_BY(crit_);
+  struct ApmCaptureNonLockedState {
+    ApmCaptureNonLockedState()
+        : fwd_proc_format(kSampleRate16kHz),
+          split_rate(kSampleRate16kHz),
+          stream_delay_ms(0) {}
+    // Only the rate and samples fields of fwd_proc_format_ are used because the
+    // forward processing number of channels is mutable and is tracked by the
+    // capture_audio_.
+    StreamConfig fwd_proc_format;
+    int split_rate;
+    int stream_delay_ms;
+  } capture_nonlocked_;
 
-  bool key_pressed_;
+  struct ApmRenderState {
+    rtc::scoped_ptr<AudioConverter> render_converter;
+    rtc::scoped_ptr<AudioBuffer> render_audio;
+  } render_ GUARDED_BY(crit_render_);
 
-  // Only set through the constructor's Config parameter.
-  const bool use_new_agc_;
-  rtc::scoped_ptr<AgcManagerDirect> agc_manager_ GUARDED_BY(crit_);
-  int agc_startup_min_volume_;
-
-  bool transient_suppressor_enabled_;
-  rtc::scoped_ptr<TransientSuppressor> transient_suppressor_;
-  const bool beamformer_enabled_;
-  rtc::scoped_ptr<Beamformer<float>> beamformer_;
-  const std::vector<Point> array_geometry_;
-  const SphericalPointf target_direction_;
-
-  bool intelligibility_enabled_;
-  rtc::scoped_ptr<IntelligibilityEnhancer> intelligibility_enhancer_;
+// Debug dump state.
+#ifdef WEBRTC_AUDIOPROC_DEBUG_DUMP
+  ApmDebugDumpState debug_dump_;
+#endif
 };
 
 }  // namespace webrtc
 
 #endif  // WEBRTC_MODULES_AUDIO_PROCESSING_AUDIO_PROCESSING_IMPL_H_