Move video encode accelerator IPC messages to GPU IO thread

media/gpu/ipc/service/gpu_video_encode_accelerator.cc

 // Copyright 2013 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 "media/gpu/ipc/service/gpu_video_encode_accelerator.h"
 
 #include <memory>
 
 #include "base/callback.h"
 #include "base/command_line.h"
@@ skipping 81 matching lines @@
 
 #if defined(OS_WIN)
 std::unique_ptr<VideoEncodeAccelerator> CreateMediaFoundationVEA() {
   return base::WrapUnique<media::VideoEncodeAccelerator>(
       new MediaFoundationVideoEncodeAccelerator());
 }
 #endif
 
 }  // anonymous namespace
 
+class GpuVideoEncodeAccelerator::MessageFilter : public IPC::MessageFilter {
+ public:
+  MessageFilter(GpuVideoEncodeAccelerator* owner, int32_t host_route_id)
+      : owner_(owner), host_route_id_(host_route_id) {}
+
+  void OnChannelError() override { sender_ = NULL; }
+
+  void OnChannelClosing() override { sender_ = NULL; }
+
+  void OnFilterAdded(IPC::Channel* channel) override { sender_ = channel; }
+
+  void OnFilterRemoved() override {
+    // This will delete |owner_| and |this|.

[sandersd (OOO until July 31), 2016/10/21 18:41:03]

Misleading comment, since it makes the (cross-class) assumption that
RemoveFilter() was called by the destructor. I would prefer the simplicity of a
model where the filter lifecycle is not so directly tied to GVEA. (No specific
suggestions to achieve that though, the existing callback model certainly
doesn't make it easy.)

[emircan, 2016/10/24 19:44:52]

I agree that it sounds like it is coming from dtor, so I removed the comment. I
actually got this model from GPUVDA directly.
OnWillDestroyStub() calls OnFilterRemoved(), then destroys GPUVEA and filter as
well.

https://cs.chromium.org/chromium/src/media/gpu/ipc/service/gpu_video_decode_a...

+    owner_->OnFilterRemoved();
+  }
+
+  bool OnMessageReceived(const IPC::Message& msg) override {
+    if (msg.routing_id() != host_route_id_)
+      return false;
+
+    IPC_BEGIN_MESSAGE_MAP(MessageFilter, msg)
+      IPC_MESSAGE_FORWARD(AcceleratedVideoEncoderMsg_Encode, owner_,
+                          GpuVideoEncodeAccelerator::OnEncode)
+      IPC_MESSAGE_FORWARD(AcceleratedVideoEncoderMsg_UseOutputBitstreamBuffer,
+                          owner_,
+                          GpuVideoEncodeAccelerator::OnUseOutputBitstreamBuffer)
+      IPC_MESSAGE_FORWARD(
+          AcceleratedVideoEncoderMsg_RequestEncodingParametersChange, owner_,
+          GpuVideoEncodeAccelerator::OnRequestEncodingParametersChange)
+      IPC_MESSAGE_UNHANDLED(return false)
+    IPC_END_MESSAGE_MAP()
+    return true;
+  }
+
+  bool SendOnIOThread(IPC::Message* message) {
+    CHECK(!message->is_sync());
+    if (!sender_) {
+      delete message;
+      return false;
+    }
+    const bool rv = sender_->Send(message);
+    CHECK(rv);

[sandersd (OOO until July 31), 2016/10/21 18:41:03]

?

[emircan, 2016/10/24 19:44:52]

It was paranioa check. Removing it.

+    return rv;
+  }
+
+ protected:
+  ~MessageFilter() override {}
+
+ private:
+  GpuVideoEncodeAccelerator* const owner_;
+  const int32_t host_route_id_;
+  // The sender to which this filter was added.
+  IPC::Sender* sender_;
+};
+
 GpuVideoEncodeAccelerator::GpuVideoEncodeAccelerator(
     int32_t host_route_id,
-    gpu::GpuCommandBufferStub* stub)
+    gpu::GpuCommandBufferStub* stub,
+    const scoped_refptr<base::SingleThreadTaskRunner>& io_task_runner)
     : host_route_id_(host_route_id),
       stub_(stub),
       input_format_(PIXEL_FORMAT_UNKNOWN),
       output_buffer_size_(0),
+      filter_removed_(base::WaitableEvent::ResetPolicy::MANUAL,
+                      base::WaitableEvent::InitialState::NOT_SIGNALED),
+      main_task_runner_(base::ThreadTaskRunnerHandle::Get()),
+      io_task_runner_(io_task_runner),
       weak_this_factory_(this) {
   stub_->AddDestructionObserver(this);
   make_context_current_ =
       base::Bind(&MakeDecoderContextCurrent, stub_->AsWeakPtr());
 }
 
 GpuVideoEncodeAccelerator::~GpuVideoEncodeAccelerator() {
   // This class can only be self-deleted from OnWillDestroyStub(), which means
   // the VEA has already been destroyed in there.
   DCHECK(!encoder_);
 }
 
 bool GpuVideoEncodeAccelerator::Initialize(VideoPixelFormat input_format,
                                            const gfx::Size& input_visible_size,
                                            VideoCodecProfile output_profile,
                                            uint32_t initial_bitrate) {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
   DVLOG(1) << __FUNCTION__
            << " input_format=" << VideoPixelFormatToString(input_format)
            << ", input_visible_size=" << input_visible_size.ToString()
            << ", output_profile=" << GetProfileName(output_profile)
            << ", initial_bitrate=" << initial_bitrate;
   DCHECK(!encoder_);
 
   if (!stub_->channel()->AddRoute(host_route_id_, stub_->stream_id(), this)) {
     DLOG(ERROR) << __FUNCTION__ << " failed to add route";
     return false;
@@ skipping 11 matching lines @@
       stub_->channel()->gpu_channel_manager()->gpu_preferences();
 
   // Try all possible encoders and use the first successful encoder.
   for (const auto& factory_function : GetVEAFactoryFunctions(gpu_preferences)) {
     encoder_ = factory_function.Run();
     if (encoder_ &&
         encoder_->Initialize(input_format, input_visible_size, output_profile,
                              initial_bitrate, this)) {
       input_format_ = input_format;
       input_visible_size_ = input_visible_size;
+      // Attempt to set up performing encoding tasks on IO thread, if supported
+      // by the VEA.
+      if (encoder_->TryToSetupEncodeOnSeparateThread(
+              weak_this_factory_.GetWeakPtr(), io_task_runner_)) {
+        filter_ = new MessageFilter(this, host_route_id_);
+        stub_->channel()->AddFilter(filter_.get());
+      }
       return true;
     }
   }
   encoder_.reset();
   DLOG(ERROR) << __FUNCTION__ << " VEA initialization failed";
   return false;
 }
 
 bool GpuVideoEncodeAccelerator::OnMessageReceived(const IPC::Message& message) {
   bool handled = true;
   IPC_BEGIN_MESSAGE_MAP(GpuVideoEncodeAccelerator, message)
     IPC_MESSAGE_HANDLER(AcceleratedVideoEncoderMsg_Encode, OnEncode)
     IPC_MESSAGE_HANDLER(AcceleratedVideoEncoderMsg_UseOutputBitstreamBuffer,
                         OnUseOutputBitstreamBuffer)
     IPC_MESSAGE_HANDLER(
         AcceleratedVideoEncoderMsg_RequestEncodingParametersChange,
         OnRequestEncodingParametersChange)
     IPC_MESSAGE_HANDLER(AcceleratedVideoEncoderMsg_Destroy, OnDestroy)
     IPC_MESSAGE_UNHANDLED(handled = false)
   IPC_END_MESSAGE_MAP()
   return handled;
 }
 
 void GpuVideoEncodeAccelerator::RequireBitstreamBuffers(
     unsigned int input_count,
     const gfx::Size& input_coded_size,
     size_t output_buffer_size) {
-  Send(new AcceleratedVideoEncoderHostMsg_RequireBitstreamBuffers(
-      host_route_id_, input_count, input_coded_size, output_buffer_size));
+  if (!Send(new AcceleratedVideoEncoderHostMsg_RequireBitstreamBuffers(
+          host_route_id_, input_count, input_coded_size, output_buffer_size))) {
+    DLOG(ERROR) << __func__ << " failed.";

[sandersd (OOO until July 31), 2016/10/21 18:41:03]

Should GVEA be entering some sort of error state on these kinds of failures?

[emircan, 2016/10/24 19:44:52]

I added these to have logs if any error occurs but it looks like we generally do
not expect these to fail at all. There was no error handling here, and VDA seems
to also just log. I think the policy is to try sending the next message even if
one fails.
https://cs.chromium.org/chromium/src/media/gpu/ipc/service/gpu_video_decode_a...

+  }
   input_coded_size_ = input_coded_size;
   output_buffer_size_ = output_buffer_size;
 }
 
 void GpuVideoEncodeAccelerator::BitstreamBufferReady(
     int32_t bitstream_buffer_id,
     size_t payload_size,
     bool key_frame,
     base::TimeDelta timestamp) {
-  Send(new AcceleratedVideoEncoderHostMsg_BitstreamBufferReady(
-      host_route_id_, bitstream_buffer_id, payload_size, key_frame, timestamp));
+  DCHECK(CheckCalledOnMessageFilterThread());
+  if (!Send(new AcceleratedVideoEncoderHostMsg_BitstreamBufferReady(
+          host_route_id_, bitstream_buffer_id, payload_size, key_frame,
+          timestamp))) {
+    DLOG(ERROR) << __func__ << " failed.";
+  }
 }
 
 void GpuVideoEncodeAccelerator::NotifyError(
     VideoEncodeAccelerator::Error error) {
-  Send(new AcceleratedVideoEncoderHostMsg_NotifyError(host_route_id_, error));
+  DCHECK(CheckCalledOnMessageFilterThread());
+  if (!Send(new AcceleratedVideoEncoderHostMsg_NotifyError(host_route_id_,
+                                                           error))) {
+    DLOG(ERROR) << __func__ << " failed.";
+  }
 }
 
 void GpuVideoEncodeAccelerator::OnWillDestroyStub() {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
   DCHECK(stub_);
+
+  if (filter_) {
+    stub_->channel()->RemoveFilter(filter_.get());
+    filter_removed_.Wait();
+  }
+
   stub_->channel()->RemoveRoute(host_route_id_);
   stub_->RemoveDestructionObserver(this);
   encoder_.reset();
   delete this;
 }
 
 // static
 gpu::VideoEncodeAcceleratorSupportedProfiles
 GpuVideoEncodeAccelerator::GetSupportedProfiles(
     const gpu::GpuPreferences& gpu_preferences) {
   VideoEncodeAccelerator::SupportedProfiles profiles;
 
   for (const auto& factory_function : GetVEAFactoryFunctions(gpu_preferences)) {
     std::unique_ptr<VideoEncodeAccelerator> encoder = factory_function.Run();
     if (!encoder)
       continue;
     VideoEncodeAccelerator::SupportedProfiles vea_profiles =
         encoder->GetSupportedProfiles();
     GpuVideoAcceleratorUtil::InsertUniqueEncodeProfiles(vea_profiles,
                                                         &profiles);
   }
   return GpuVideoAcceleratorUtil::ConvertMediaToGpuEncodeProfiles(profiles);
 }
 
+void GpuVideoEncodeAccelerator::OnFilterRemoved() {
+  // We're destroying; cancel all callbacks.

[sandersd (OOO until July 31), 2016/10/21 18:41:03]

Are there no GPU shutdown paths where the channel filter can be removed without
GVEA being fully destructed first?

[emircan, 2016/10/24 19:44:52]

Filter is actually removed before GVEA is destructed. Only shutdown path here is
OnWillDestroyStub(), lifetime of all these objects are tied to the stub as it
is.

+  weak_this_factory_.InvalidateWeakPtrs();
+  filter_removed_.Signal();
+}
+
 // static
 std::vector<GpuVideoEncodeAccelerator::VEAFactoryFunction>
 GpuVideoEncodeAccelerator::GetVEAFactoryFunctions(
     const gpu::GpuPreferences& gpu_preferences) {
   std::vector<VEAFactoryFunction> vea_factory_functions;
 #if defined(OS_CHROMEOS) && defined(USE_V4L2_CODEC)
   vea_factory_functions.push_back(base::Bind(&CreateV4L2VEA));
 #endif
 #if defined(OS_CHROMEOS) && defined(ARCH_CPU_X86_FAMILY)
   if (!gpu_preferences.disable_vaapi_accelerated_video_encode)
     vea_factory_functions.push_back(base::Bind(&CreateVaapiVEA));
 #endif
 #if defined(OS_ANDROID) && defined(ENABLE_WEBRTC)
   if (!gpu_preferences.disable_web_rtc_hw_encoding)
     vea_factory_functions.push_back(base::Bind(&CreateAndroidVEA));
 #endif
 #if defined(OS_MACOSX)
   vea_factory_functions.push_back(base::Bind(&CreateVTVEA));
 #endif
 #if defined(OS_WIN)
   if (base::FeatureList::IsEnabled(kMediaFoundationH264Encoding))
     vea_factory_functions.push_back(base::Bind(&CreateMediaFoundationVEA));
 #endif
   return vea_factory_functions;
 }
 
 void GpuVideoEncodeAccelerator::OnEncode(
     const AcceleratedVideoEncoderMsg_Encode_Params& params) {
+  DCHECK(CheckCalledOnMessageFilterThread());
   DVLOG(3) << __FUNCTION__ << " frame_id = " << params.frame_id
            << ", buffer_size=" << params.buffer_size
            << ", force_keyframe=" << params.force_keyframe;
   DCHECK_EQ(PIXEL_FORMAT_I420, input_format_);
 
   // Wrap into a SharedMemory in the beginning, so that |params.buffer_handle|
   // is cleaned properly in case of an early return.
   std::unique_ptr<base::SharedMemory> shm(
       new base::SharedMemory(params.buffer_handle, true));
 
@@ skipping 40 matching lines @@
   frame->AddDestructionObserver(BindToCurrentLoop(base::Bind(
       &GpuVideoEncodeAccelerator::EncodeFrameFinished,
       weak_this_factory_.GetWeakPtr(), params.frame_id, base::Passed(&shm))));
   encoder_->Encode(frame, params.force_keyframe);
 }
 
 void GpuVideoEncodeAccelerator::OnUseOutputBitstreamBuffer(
     int32_t buffer_id,
     base::SharedMemoryHandle buffer_handle,
     uint32_t buffer_size) {
+  DCHECK(CheckCalledOnMessageFilterThread());
   DVLOG(3) << __FUNCTION__ << " buffer_id=" << buffer_id
            << ", buffer_size=" << buffer_size;
   if (!encoder_)
     return;
   if (buffer_id < 0) {
     DLOG(ERROR) << __FUNCTION__ << " invalid buffer_id=" << buffer_id;
     NotifyError(VideoEncodeAccelerator::kPlatformFailureError);
     return;
   }
   if (buffer_size < output_buffer_size_) {
     DLOG(ERROR) << __FUNCTION__
                 << " buffer too small for buffer_id=" << buffer_id;
     NotifyError(VideoEncodeAccelerator::kPlatformFailureError);
     return;
   }
   encoder_->UseOutputBitstreamBuffer(
       BitstreamBuffer(buffer_id, buffer_handle, buffer_size));
 }
 
 void GpuVideoEncodeAccelerator::OnDestroy() {
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
   DVLOG(2) << __FUNCTION__;
   OnWillDestroyStub();
 }
 
 void GpuVideoEncodeAccelerator::OnRequestEncodingParametersChange(
     uint32_t bitrate,
     uint32_t framerate) {
+  DCHECK(CheckCalledOnMessageFilterThread());
   DVLOG(2) << __FUNCTION__ << " bitrate=" << bitrate
            << ", framerate=" << framerate;
   if (!encoder_)
     return;
   encoder_->RequestEncodingParametersChange(bitrate, framerate);
 }
 
 void GpuVideoEncodeAccelerator::EncodeFrameFinished(
     int32_t frame_id,
     std::unique_ptr<base::SharedMemory> shm) {
-  Send(new AcceleratedVideoEncoderHostMsg_NotifyInputDone(host_route_id_,
-                                                          frame_id));
+  DCHECK(CheckCalledOnMessageFilterThread());
+  if (!Send(new AcceleratedVideoEncoderHostMsg_NotifyInputDone(host_route_id_,
+                                                               frame_id))) {
+    DLOG(ERROR) << __func__ << " failed.";
+  }
   // Just let |shm| fall out of scope.
 }
 
-void GpuVideoEncodeAccelerator::Send(IPC::Message* message) {
-  stub_->channel()->Send(message);
+bool GpuVideoEncodeAccelerator::Send(IPC::Message* message) {
+  if (filter_ && io_task_runner_->BelongsToCurrentThread()) {
+    DCHECK(io_task_runner_->BelongsToCurrentThread());

[sandersd (OOO until July 31), 2016/10/21 18:41:03]

DCHECK() is redundant.

[emircan, 2016/10/24 19:44:52]

Done.

+    return filter_->SendOnIOThread(message);
+  }
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  return stub_->channel()->Send(message);
+}
+
+bool GpuVideoEncodeAccelerator::CheckCalledOnMessageFilterThread() {
+  return (io_task_runner_->BelongsToCurrentThread() && filter_) ||
+         (main_task_runner_->BelongsToCurrentThread() && !filter_);

[sandersd (OOO until July 31), 2016/10/21 18:41:03]

Nit: Putting |filter_| condition first makes more sense.

[emircan, 2016/10/24 19:44:52]

Done.

 }
 
 }  // namespace media