assistant: replace most thread task runner with sequenced

Bug: None
Test: locally build and run
Change-Id: I8e77799bcc4f417faf7dff54f9e40ab7a46f3d1f
Reviewed-on: https://chromium-review.googlesource.com/c/1372722Reviewed-by: Tao Wu <wutao@chromium.org>
Commit-Queue: Xiaohui Chen <xiaohuic@chromium.org>
Cr-Commit-Position: refs/heads/master@{#615732}

chromeos/services/assistant/assistant_manager_service_impl.cc

@@ -750,7 +750,7 @@ AssistantManagerServiceImpl::StartAssistantInternal(
 void AssistantManagerServiceImpl::PostInitAssistant(
     base::OnceClosure post_init_callback,
     std::unique_ptr<assistant_client::AssistantManager>* assistant_manager) {
-  DCHECK(service_->main_task_runner()->BelongsToCurrentThread());
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
 
   assistant_manager_ = std::move(*assistant_manager);
   assistant_manager_internal_ =

chromeos/services/assistant/assistant_settings_manager_impl.cc

@@ -38,7 +38,7 @@ void AssistantSettingsManagerImpl::GetSettings(const std::string& selector,
                                                GetSettingsCallback callback) {
   DCHECK(assistant_manager_service_->GetState() ==
          AssistantManagerService::State::RUNNING);
-  DCHECK(service_->main_task_runner()->BelongsToCurrentThread());
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
 
   // Wraps the callback into a repeating callback since the server side
   // interface requires the callback to be copyable.
@@ -70,7 +70,7 @@ void AssistantSettingsManagerImpl::UpdateSettings(
     GetSettingsCallback callback) {
   DCHECK(assistant_manager_service_->GetState() ==
          AssistantManagerService::State::RUNNING);
-  DCHECK(service_->main_task_runner()->BelongsToCurrentThread());
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
   // Wraps the callback into a repeating callback since the server side
   // interface requires the callback to be copyable.
   std::string serialized_proto = SerializeUpdateSettingsUiRequest(update);
@@ -101,7 +101,7 @@ void AssistantSettingsManagerImpl::StartSpeakerIdEnrollment(
     mojom::SpeakerIdEnrollmentClientPtr client) {
   DCHECK(assistant_manager_service_->GetState() ==
          AssistantManagerService::State::RUNNING);
-  DCHECK(service_->main_task_runner()->BelongsToCurrentThread());
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
 
   speaker_id_enrollment_client_ = std::move(client);
 
@@ -126,6 +126,7 @@ void AssistantSettingsManagerImpl::StopSpeakerIdEnrollment(
     StopSpeakerIdEnrollmentCallback callback) {
   DCHECK(assistant_manager_service_->GetState() ==
          AssistantManagerService::State::RUNNING);
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
   assistant_manager_service_->assistant_manager_internal()
       ->StopSpeakerIdEnrollment([repeating_callback =
                                      base::AdaptCallbackForRepeating(
@@ -142,6 +143,7 @@ void AssistantSettingsManagerImpl::StopSpeakerIdEnrollment(
 
 void AssistantSettingsManagerImpl::HandleSpeakerIdEnrollmentUpdate(
     const assistant_client::SpeakerIdEnrollmentUpdate& update) {
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
   switch (update.state) {
     case SpeakerIdEnrollmentState::LISTEN:
       speaker_id_enrollment_client_->OnListeningHotword();
@@ -170,6 +172,7 @@ void AssistantSettingsManagerImpl::HandleSpeakerIdEnrollmentUpdate(
 
 void AssistantSettingsManagerImpl::HandleSpeakerIdEnrollmentStatusSync(
     const assistant_client::SpeakerIdEnrollmentUpdate& update) {
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
   switch (update.state) {
     case SpeakerIdEnrollmentState::LISTEN:
       speaker_id_enrollment_done_ = false;
@@ -194,12 +197,13 @@ void AssistantSettingsManagerImpl::HandleSpeakerIdEnrollmentStatusSync(
 
 void AssistantSettingsManagerImpl::HandleStopSpeakerIdEnrollment(
     base::RepeatingCallback<void()> callback) {
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
   speaker_id_enrollment_client_.reset();
   callback.Run();
 }
 
 void AssistantSettingsManagerImpl::SyncSpeakerIdEnrollmentStatus() {
-  DCHECK(service_->main_task_runner()->BelongsToCurrentThread());
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
   if (speaker_id_enrollment_client_) {
     // Speaker id enrollment is in progress.
     return;
@@ -225,7 +229,7 @@ void AssistantSettingsManagerImpl::SyncSpeakerIdEnrollmentStatus() {
 }
 
 void AssistantSettingsManagerImpl::UpdateServerDeviceSettings() {
-  DCHECK(service_->main_task_runner()->BelongsToCurrentThread());
+  DCHECK(service_->main_task_runner()->RunsTasksInCurrentSequence());
 
   const std::string device_id =
       assistant_manager_service_->assistant_manager()->GetDeviceId();

chromeos/services/assistant/audio_decoder/assistant_audio_decoder.cc

@@ -4,6 +4,9 @@
 
 #include "chromeos/services/assistant/audio_decoder/assistant_audio_decoder.h"
 
+#include <utility>
+#include <vector>
+
 #include "base/threading/thread.h"
 #include "chromeos/services/assistant/audio_decoder/ipc_data_source.h"
 #include "media/base/audio_bus.h"
@@ -32,7 +35,7 @@ AssistantAudioDecoder::AssistantAudioDecoder(
     mojom::AssistantMediaDataSourcePtr data_source)
     : service_ref_(std::move(service_ref)),
       client_(std::move(client)),
-      task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      task_runner_(base::SequencedTaskRunnerHandle::Get()),
       data_source_(std::make_unique<IPCDataSource>(std::move(data_source))),
       media_thread_(std::make_unique<base::Thread>("media_thread")),
       weak_factory_(this) {

chromeos/services/assistant/audio_decoder/ipc_data_source.cc

@@ -4,7 +4,11 @@
 
 #include "chromeos/services/assistant/audio_decoder/ipc_data_source.h"
 
-#include "base/threading/thread_task_runner_handle.h"
+#include <algorithm>
+#include <utility>
+#include <vector>
+
+#include "base/threading/sequenced_task_runner_handle.h"
 #include "mojo/public/cpp/bindings/message.h"
 
 namespace chromeos {
@@ -13,7 +17,7 @@ namespace assistant {
 IPCDataSource::IPCDataSource(
     mojom::AssistantMediaDataSourcePtr media_data_source)
     : media_data_source_(std::move(media_data_source)),
-      utility_task_runner_(base::ThreadTaskRunnerHandle::Get()) {
+      utility_task_runner_(base::SequencedTaskRunnerHandle::Get()) {
   DETACH_FROM_THREAD(data_source_thread_checker_);
 }
 

chromeos/services/assistant/platform/audio_input_impl.cc

@@ -139,7 +139,7 @@ class AudioInputBufferImpl : public assistant_client::AudioBuffer {
 
 AudioInputImpl::AudioInputImpl(service_manager::Connector* connector)
     : connector_(connector),
-      task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      task_runner_(base::SequencedTaskRunnerHandle::Get()),
       weak_factory_(this) {
   DETACH_FROM_SEQUENCE(observer_sequence_checker_);
 

chromeos/services/assistant/platform/audio_output_provider_impl.cc

@@ -88,7 +88,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
       assistant_client::OutputStreamType type,
       assistant_client::OutputStreamFormat format)
       : connector_(connector),
-        main_thread_task_runner_(task_runner),
+        main_task_runner_(task_runner),
         background_thread_task_runner_(background_task_runner),
         audio_decoder_factory_(audio_decoder_factory),
         stream_type_(type),
@@ -101,7 +101,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
 
   ~AudioOutputImpl() override {
     // This ensures that it will be executed after StartOnMainThread.
-    main_thread_task_runner_->PostTask(
+    main_task_runner_->PostTask(
         FROM_HERE,
         base::BindOnce(
             [](std::unique_ptr<AudioDeviceOwner> device_owner,
@@ -110,8 +110,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
               background_runner->DeleteSoon(FROM_HERE, device_owner.release());
             },
             std::move(device_owner_), background_thread_task_runner_));
-    main_thread_task_runner_->DeleteSoon(FROM_HERE,
-                                         audio_stream_handler_.release());
+    main_task_runner_->DeleteSoon(FROM_HERE, audio_stream_handler_.release());
   }
 
   // assistant_client::AudioOutput overrides:
@@ -119,7 +118,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
 
   void Start(assistant_client::AudioOutput::Delegate* delegate) override {
     if (IsEncodedFormat(format_)) {
-      main_thread_task_runner_->PostTask(
+      main_task_runner_->PostTask(
           FROM_HERE,
           base::BindOnce(
               &AudioStreamHandler::StartAudioDecoder,
@@ -129,7 +128,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
                              base::Unretained(device_owner_.get()),
                              audio_stream_handler_.get(), connector_)));
     } else {
-      main_thread_task_runner_->PostTask(
+      main_task_runner_->PostTask(
           FROM_HERE, base::BindOnce(&AudioDeviceOwner::StartOnMainThread,
                                     base::Unretained(device_owner_.get()),
                                     delegate, connector_, format_));
@@ -139,7 +138,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
   void Stop() override {
     if (IsEncodedFormat(format_)) {
       device_owner_->SetDelegate(nullptr);
-      main_thread_task_runner_->PostTask(
+      main_task_runner_->PostTask(
           FROM_HERE,
           base::BindOnce(&AudioStreamHandler::OnStopped,
                          base::Unretained(audio_stream_handler_.get())));
@@ -152,7 +151,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
 
  private:
   service_manager::Connector* connector_;
-  scoped_refptr<base::SequencedTaskRunner> main_thread_task_runner_;
+  scoped_refptr<base::SequencedTaskRunner> main_task_runner_;
   scoped_refptr<base::SequencedTaskRunner> background_thread_task_runner_;
   mojom::AssistantAudioDecoderFactory* audio_decoder_factory_;
 
@@ -170,7 +169,7 @@ class AudioOutputImpl : public assistant_client::AudioOutput {
 
 VolumeControlImpl::VolumeControlImpl(service_manager::Connector* connector)
     : binding_(this),
-      main_thread_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      main_task_runner_(base::SequencedTaskRunnerHandle::Get()),
       weak_factory_(this) {
   connector->BindInterface(ash::mojom::kServiceName, &volume_control_ptr_);
   ash::mojom::VolumeObserverPtr observer;
@@ -188,7 +187,7 @@ float VolumeControlImpl::GetSystemVolume() {
 }
 
 void VolumeControlImpl::SetSystemVolume(float new_volume, bool user_initiated) {
-  main_thread_task_runner_->PostTask(
+  main_task_runner_->PostTask(
       FROM_HERE,
       base::BindOnce(&VolumeControlImpl::SetSystemVolumeOnMainThread,
                      weak_factory_.GetWeakPtr(), new_volume, user_initiated));
@@ -208,7 +207,7 @@ bool VolumeControlImpl::IsSystemMuted() {
 }
 
 void VolumeControlImpl::SetSystemMuted(bool muted) {
-  main_thread_task_runner_->PostTask(
+  main_task_runner_->PostTask(
       FROM_HERE, base::BindOnce(&VolumeControlImpl::SetSystemMutedOnMainThread,
                                 weak_factory_.GetWeakPtr(), muted));
 }
@@ -223,12 +222,12 @@ void VolumeControlImpl::OnMuteStateChanged(bool mute) {
 
 void VolumeControlImpl::SetSystemVolumeOnMainThread(float new_volume,
                                                     bool user_initiated) {
-  DCHECK(main_thread_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
   volume_control_ptr_->SetVolume(new_volume * 100.0, user_initiated);
 }
 
 void VolumeControlImpl::SetSystemMutedOnMainThread(bool muted) {
-  DCHECK(main_thread_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
   volume_control_ptr_->SetMuted(muted);
 }
 
@@ -237,7 +236,7 @@ AudioOutputProviderImpl::AudioOutputProviderImpl(
     scoped_refptr<base::SequencedTaskRunner> background_task_runner)
     : volume_control_impl_(connector),
       connector_(connector),
-      main_thread_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      main_task_runner_(base::SequencedTaskRunnerHandle::Get()),
       background_task_runner_(background_task_runner) {
   connector_->BindInterface(mojom::kAudioDecoderServiceName,
                             mojo::MakeRequest(&audio_decoder_factory_ptr_));
@@ -251,7 +250,7 @@ assistant_client::AudioOutput* AudioOutputProviderImpl::CreateAudioOutput(
     const assistant_client::OutputStreamFormat& stream_format) {
   // Owned by one arbitrary thread inside libassistant. It will be destroyed
   // once assistant_client::AudioOutput::Delegate::OnStopped() is called.
-  return new AudioOutputImpl(connector_, main_thread_task_runner_,
+  return new AudioOutputImpl(connector_, main_task_runner_,
                              background_task_runner_, audio_decoder_factory_,
                              type, stream_format);
 }
@@ -289,7 +288,7 @@ void AudioOutputProviderImpl::RegisterAudioEmittingStateCallback(
 AudioDeviceOwner::AudioDeviceOwner(
     scoped_refptr<base::SequencedTaskRunner> task_runner,
     scoped_refptr<base::SequencedTaskRunner> background_task_runner)
-    : main_thread_task_runner_(task_runner),
+    : main_task_runner_(task_runner),
       background_task_runner_(background_task_runner) {}
 
 AudioDeviceOwner::~AudioDeviceOwner() {
@@ -301,7 +300,7 @@ void AudioDeviceOwner::StartOnMainThread(
     service_manager::Connector* connector,
     const assistant_client::OutputStreamFormat& format) {
   DCHECK(!output_device_);
-  DCHECK(main_thread_task_runner_->RunsTasksInCurrentSequence());
+  DCHECK(main_task_runner_->RunsTasksInCurrentSequence());
 
   delegate_ = delegate;
   format_ = format;

chromeos/services/assistant/platform/audio_output_provider_impl.h

@@ -52,7 +52,7 @@ class VolumeControlImpl : public assistant_client::VolumeControl,
 
   ash::mojom::AssistantVolumeControlPtr volume_control_ptr_;
   mojo::Binding<ash::mojom::VolumeObserver> binding_;
-  scoped_refptr<base::SequencedTaskRunner> main_thread_task_runner_;
+  scoped_refptr<base::SequencedTaskRunner> main_task_runner_;
 
   int volume_ = 100;
   bool mute_ = false;
@@ -89,7 +89,7 @@ class AudioOutputProviderImpl : public assistant_client::AudioOutputProvider {
  private:
   VolumeControlImpl volume_control_impl_;
   service_manager::Connector* connector_;
-  scoped_refptr<base::SequencedTaskRunner> main_thread_task_runner_;
+  scoped_refptr<base::SequencedTaskRunner> main_task_runner_;
   scoped_refptr<base::SequencedTaskRunner> background_task_runner_;
   mojom::AssistantAudioDecoderFactoryPtr audio_decoder_factory_ptr_;
   mojom::AssistantAudioDecoderFactory* audio_decoder_factory_;
@@ -130,7 +130,7 @@ class AudioDeviceOwner : public media::AudioRendererSink::RenderCallback {
   // Callback for assistant to notify that it completes the filling.
   void BufferFillDone(int num_bytes);
 
-  scoped_refptr<base::SequencedTaskRunner> main_thread_task_runner_;
+  scoped_refptr<base::SequencedTaskRunner> main_task_runner_;
   scoped_refptr<base::SequencedTaskRunner> background_task_runner_;
 
   base::Lock lock_;

chromeos/services/assistant/platform/audio_output_provider_impl_unittest.cc

@@ -102,7 +102,8 @@ TEST_F(AudioDeviceOwnerTest, BufferFilling) {
   delegate.set_num_of_bytes_to_fill(200);
   delegate.Reset();
   auto owner = std::make_unique<AudioDeviceOwner>(
-      base::ThreadTaskRunnerHandle::Get(), base::ThreadTaskRunnerHandle::Get());
+      base::SequencedTaskRunnerHandle::Get(),
+      base::SequencedTaskRunnerHandle::Get());
   // Upon start, it will start to fill the buffer.
   owner->StartOnMainThread(&delegate, nullptr, format);
   delegate.Wait();

chromeos/services/assistant/service.cc

@@ -59,7 +59,7 @@ Service::Service(service_manager::mojom::ServiceRequest request,
       platform_binding_(this),
       session_observer_binding_(this),
       token_refresh_timer_(std::make_unique<base::OneShotTimer>()),
-      main_thread_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      main_task_runner_(base::SequencedTaskRunnerHandle::Get()),
       power_manager_observer_(this),
       network_connection_tracker_(network_connection_tracker),
       io_task_runner_(std::move(io_task_runner)),
@@ -184,11 +184,11 @@ void Service::UpdateAssistantManagerState() {
         assistant_manager_service_->Start(
             access_token_.value(), assistant_state_.hotword_enabled().value(),
             base::BindOnce(
-                [](scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+                [](scoped_refptr<base::SequencedTaskRunner> task_runner,
                    base::OnceCallback<void()> callback) {
                   task_runner->PostTask(FROM_HERE, std::move(callback));
                 },
-                main_thread_task_runner_,
+                main_task_runner_,
                 base::BindOnce(&Service::FinalizeAssistantManagerService,
                                weak_ptr_factory_.GetWeakPtr())));
         DVLOG(1) << "Request Assistant start";

chromeos/services/assistant/service.h

@@ -77,12 +77,13 @@ class Service : public service_manager::Service,
   }
 
   ash::AssistantStateBase* assistant_state() { return &assistant_state_; }
+  // net::URLRequestContextGetter requires a base::SingleThreadTaskRunner.
   scoped_refptr<base::SingleThreadTaskRunner> io_task_runner() {
     return io_task_runner_;
   }
 
-  scoped_refptr<base::SingleThreadTaskRunner> main_task_runner() {
-    return main_thread_task_runner_;
+  scoped_refptr<base::SequencedTaskRunner> main_task_runner() {
+    return main_task_runner_;
   }
 
   void RequestAccessToken();
@@ -162,7 +163,7 @@ class Service : public service_manager::Service,
   std::unique_ptr<AssistantManagerService> assistant_manager_service_;
   std::unique_ptr<base::OneShotTimer> token_refresh_timer_;
   int token_refresh_error_backoff_factor = 1;
-  scoped_refptr<base::SingleThreadTaskRunner> main_thread_task_runner_;
+  scoped_refptr<base::SequencedTaskRunner> main_task_runner_;
   ScopedObserver<chromeos::PowerManagerClient,
                  chromeos::PowerManagerClient::Observer>
       power_manager_observer_;