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; }

[Pawel Osciak, 2016/10/25 01:44:04]

sender_ = nullptr;

and similarly below please.

[emircan, 2016/10/25 21:57:51]

Done.

+
+  void OnChannelClosing() override { sender_ = NULL; }
+
+  void OnFilterAdded(IPC::Channel* channel) override { sender_ = channel; }
+
+  void OnFilterRemoved() override { 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());

[Pawel Osciak, 2016/10/25 01:44:05]

Perhaps DCHECK and return false to avoid crashing the process? Do we need to
delete message too?

[emircan, 2016/10/25 21:57:50]

I merged this error case with l.135 and made it a DCHECK crash.

+    if (!sender_) {
+      delete message;
+      return false;
+    }
+    return sender_->Send(message);
+  }
+
+ protected:
+  ~MessageFilter() override {}
+
+ private:
+  GpuVideoEncodeAccelerator* const owner_;
+  const int32_t host_route_id_;
+  // The sender to which this filter was added.
+  IPC::Sender* sender_;

[Pawel Osciak, 2016/10/25 01:44:04]

= nullptr ?

[emircan, 2016/10/25 21:57:50]

Added to the initializer list.

+};

[Pawel Osciak, 2016/10/25 01:44:04]

Should we also have some DISALLOW_COPY_AND_ASSIGN ?

[emircan, 2016/10/25 21:57:50]

Done.

+
 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) {
-  DVLOG(1) << __FUNCTION__
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  DVLOG(1) << __func__
            << " 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";
+    DLOG(ERROR) << __func__ << " failed to add route";
     return false;
   }
 
   if (input_visible_size.width() > limits::kMaxDimension ||
       input_visible_size.height() > limits::kMaxDimension ||
       input_visible_size.GetArea() > limits::kMaxCanvas) {
-    DLOG(ERROR) << __FUNCTION__ << "too large input_visible_size "
+    DLOG(ERROR) << __func__ << "too large input_visible_size "
                 << input_visible_size.ToString();
     return false;
   }
 
   const gpu::GpuPreferences& gpu_preferences =
       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";
+  DLOG(ERROR) << __func__ << " 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(

[Pawel Osciak, 2016/10/25 01:44:04]

DCHECK(CheckCalledOnMessageFilterThread()) if we expect to do this?

[emircan, 2016/10/25 21:57:50]

I moved it to main thread after your comment and added a check.

+          host_route_id_, input_count, input_coded_size, output_buffer_size))) {
+    DLOG(ERROR) << __func__ << " failed.";
+  }
   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() {
+  DVLOG(2) << __func__;
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
   DCHECK(stub_);
+
+  if (filter_) {
+    stub_->channel()->RemoveFilter(filter_.get());
+    filter_removed_.Wait();

[Pawel Osciak, 2016/10/25 01:44:04]

Please add a comment similar to the one in gvda.cc why we are waiting here.

[emircan, 2016/10/25 21:57:51]

Done.

+  }
+
   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() {
+  DVLOG(2) << __func__;
+
+  // We're destroying; cancel all callbacks.
+  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) {
-  DVLOG(3) << __FUNCTION__ << " frame_id = " << params.frame_id
+  DVLOG(3) << __func__ << " frame_id = " << params.frame_id
            << ", buffer_size=" << params.buffer_size
            << ", force_keyframe=" << params.force_keyframe;
+  DCHECK(CheckCalledOnMessageFilterThread());
   DCHECK_EQ(PIXEL_FORMAT_I420, input_format_);
 
   // Wrap into a SharedMemory in the beginning, so that |params.buffer_handle|

[Pawel Osciak, 2016/10/25 01:44:04]

Could we move the reminder of this method off of the IO thread please? Mapping
may be costly.

[emircan, 2016/10/25 21:57:50]

I added a trace to measure how long it takes here. It is 0.058 ms on average on
my Dell E74440 Win laptop. Also, it looks like GpuJpegDecodeAccelerator also
does this on IO thread. I don't think it is worth adding another thread hop
here.
https://cs.chromium.org/chromium/src/media/gpu/ipc/service/gpu_jpeg_decode_ac...

Also, what thread would you suggest to use? Main task runner is pretty busy as
well here. One option would be passing shmemhandle to the encoder's own thread
directly but that would require an interface change. I can add a TODO for a
seperate CL.

[Pawel Osciak, 2016/10/27 01:45:19]

From my understanding, the rule of thumb is that nothing should be done on IO
thread if it can be helped in a reasonable way, it's timing critical for
responsiveness, etc. Also, even if we have small overheads in multiple places,
they add up.

Moreover, I'm concerned that 0.058ms in a particular situation on a particular
device/OS may be a different number on another system, for different source
memory type, under different memory pressure, virtual address space layout, etc.
This may also sleep.

GpuJpegDecodeAccelerator should also be fixed for this.
Making VideoFrame::data() and visible_data() lazy-map the memory on first access
if IsMappable() would be one solution.

The simplest solution though would probably be to have a separate thread in this
class to jump from IO onto, map there and then post back directly to IO to
encoder_->Encode() from there afterwards.

[emircan, 2016/10/27 17:53:30]

I like the idea of letting VideoFrame lazy evaluate. That would help in many
places including GpuJpegDecodeAccelerator. I added a bug, see crbug.com/660082.

I understand that we want to avoid doing work on IO thread. Same principle
applies to the main thread as well, although it is currently used for map. What
I was trying to point out was that thread hopping would be more expensive than
the actual map operation on this platform. Two posted tasks are additional costs
and it would be a clear regression. I just wanted to avoid that.

[Pawel Osciak, 2016/10/31 07:40:21]

From our experience with this, the main performance issue here is not the cost
of thread hops, but the fact that GpuMain is heavily overloaded and thus
Encode() etc. calls have to wait a long time for other tasks on Main before they
can execute.

My suggestion is to use a separate thread here, but not GpuMain, for mapping.
That should still improve latency of these calls - no need to wait for busy
GpuMain to become free, but without doing them on IO.

Also, the mapping cost you benchmarked was in that particular situation and for
that particular platform. This may not be the case always.

[emircan, 2016/10/31 19:45:08]

Sorry if my statement earlier wasn't clear. I am not suggesting posting this
task to GPU main either and I understand the reasons behind. I wanted to point
out that with the current code, map() is happening on GPU main and that is as
bad as doing it on IO.

I think lazy map() you suggested would be the ideal solution. map() would be
delayed until the thread that actually access data-encoder thread in this case-
and would solve the problem in many places including GpuJpeg. We wouldn't need
to spin a thread in each of these classes, use the current ones and avoid thread
hops.

[Pawel Osciak, 2016/11/01 04:59:58]

SGTM, thanks! That would be in this CL, right?

   // is cleaned properly in case of an early return.
   std::unique_ptr<base::SharedMemory> shm(
       new base::SharedMemory(params.buffer_handle, true));
 
   if (!encoder_)
     return;
 
   if (params.frame_id < 0) {
-    DLOG(ERROR) << __FUNCTION__ << " invalid frame_id=" << params.frame_id;
+    DLOG(ERROR) << __func__ << " invalid frame_id=" << params.frame_id;
     NotifyError(VideoEncodeAccelerator::kPlatformFailureError);
     return;
   }
 
   const uint32_t aligned_offset =
       params.buffer_offset % base::SysInfo::VMAllocationGranularity();
   base::CheckedNumeric<off_t> map_offset = params.buffer_offset;
   map_offset -= aligned_offset;
   base::CheckedNumeric<size_t> map_size = params.buffer_size;
   map_size += aligned_offset;
 
   if (!map_offset.IsValid() || !map_size.IsValid()) {
-    DLOG(ERROR) << __FUNCTION__ << "  invalid map_offset or map_size";
+    DLOG(ERROR) << __func__ << "  invalid map_offset or map_size";
     NotifyError(VideoEncodeAccelerator::kPlatformFailureError);
     return;
   }
 
   if (!shm->MapAt(map_offset.ValueOrDie(), map_size.ValueOrDie())) {
-    DLOG(ERROR) << __FUNCTION__
-                << " could not map frame_id=" << params.frame_id;
+    DLOG(ERROR) << __func__ << " could not map frame_id=" << params.frame_id;
     NotifyError(VideoEncodeAccelerator::kPlatformFailureError);
     return;
   }
 
   uint8_t* shm_memory =
       reinterpret_cast<uint8_t*>(shm->memory()) + aligned_offset;
   scoped_refptr<VideoFrame> frame = VideoFrame::WrapExternalSharedMemory(
       input_format_, input_coded_size_, gfx::Rect(input_visible_size_),
       input_visible_size_, shm_memory, params.buffer_size, params.buffer_handle,
       params.buffer_offset, params.timestamp);
   if (!frame) {
-    DLOG(ERROR) << __FUNCTION__ << " could not create a frame";
+    DLOG(ERROR) << __func__ << " could not create a frame";
     NotifyError(VideoEncodeAccelerator::kPlatformFailureError);
     return;
   }
   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) {
-  DVLOG(3) << __FUNCTION__ << " buffer_id=" << buffer_id
+  DVLOG(3) << __func__ << " buffer_id=" << buffer_id
            << ", buffer_size=" << buffer_size;
+  DCHECK(CheckCalledOnMessageFilterThread());
   if (!encoder_)
     return;
   if (buffer_id < 0) {
-    DLOG(ERROR) << __FUNCTION__ << " invalid buffer_id=" << buffer_id;
+    DLOG(ERROR) << __func__ << " 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;
+    DLOG(ERROR) << __func__ << " buffer too small for buffer_id=" << buffer_id;
     NotifyError(VideoEncodeAccelerator::kPlatformFailureError);
     return;
   }
   encoder_->UseOutputBitstreamBuffer(
       BitstreamBuffer(buffer_id, buffer_handle, buffer_size));
 }
 
 void GpuVideoEncodeAccelerator::OnDestroy() {
-  DVLOG(2) << __FUNCTION__;
+  DVLOG(2) << __func__;
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
   OnWillDestroyStub();
 }
 
 void GpuVideoEncodeAccelerator::OnRequestEncodingParametersChange(
     uint32_t bitrate,
     uint32_t framerate) {
-  DVLOG(2) << __FUNCTION__ << " bitrate=" << bitrate
-           << ", framerate=" << framerate;
+  DVLOG(2) << __func__ << " bitrate=" << bitrate << ", framerate=" << framerate;
+  DCHECK(CheckCalledOnMessageFilterThread());
   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()) {
+    return filter_->SendOnIOThread(message);
+  }
+  DCHECK(main_task_runner_->BelongsToCurrentThread());
+  return stub_->channel()->Send(message);
+}
+
+bool GpuVideoEncodeAccelerator::CheckCalledOnMessageFilterThread() {
+  return (filter_ && io_task_runner_->BelongsToCurrentThread()) ||
+         (!filter_ && main_task_runner_->BelongsToCurrentThread());
 }
 
 }  // namespace media