Step #4: Run VideoProcessor integration test batch mode on task queue.

webrtc/modules/video_coding/codecs/test/videoprocessor.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_VIDEO_CODING_CODECS_TEST_VIDEOPROCESSOR_H_
 #define WEBRTC_MODULES_VIDEO_CODING_CODECS_TEST_VIDEOPROCESSOR_H_
 
 #include <memory>
 #include <string>
 #include <vector>
 
 #include "webrtc/api/video/video_frame.h"
 #include "webrtc/base/buffer.h"
 #include "webrtc/base/checks.h"
+#include "webrtc/base/sequenced_task_checker.h"
+#include "webrtc/base/task_queue.h"
 #include "webrtc/common_video/libyuv/include/webrtc_libyuv.h"
 #include "webrtc/modules/video_coding/include/video_codec_interface.h"
 #include "webrtc/modules/video_coding/codecs/test/packet_manipulator.h"
 #include "webrtc/modules/video_coding/codecs/test/stats.h"
 #include "webrtc/modules/video_coding/utility/ivf_file_writer.h"
 #include "webrtc/test/testsupport/frame_reader.h"
 #include "webrtc/test/testsupport/frame_writer.h"
 
 namespace webrtc {
 
@@ skipping 90 matching lines @@
 // Regarding packet loss: Note that keyframes are excluded (first or all
 // depending on the ExcludeFrameTypes setting). This is because if key frames
 // would be altered, all the following delta frames would be pretty much
 // worthless. VP8 has an error-resilience feature that makes it able to handle
 // packet loss in key non-first keyframes, which is why only the first is
 // excluded by default.
 // Packet loss in such important frames is handled on a higher level in the
 // Video Engine, where signaling would request a retransmit of the lost packets,
 // since they're so important.
 //
-// Note this class is not thread safe in any way and is meant for simple testing
-// purposes.
+// This class should be run on a single thread, or a single task queue.
 class VideoProcessor {
  public:
-  virtual ~VideoProcessor() {}
+  virtual ~VideoProcessor();
 
   // Performs initial calculations about frame size, sets up callbacks etc.
   // Returns false if an error has occurred, in addition to printing to stderr.
   virtual bool Init() = 0;
 
+  virtual void DeregisterCallbacks() = 0;

[tommi, 2017/03/21 10:03:27]

Add documentation?
(e.g. if there is a condition that must be met before calling this method, what
callbacks are unregistered, if there's a post condition such as no other methods
should be called etc).

If DeregisterCallbacks matches with Init(), then it would be good to call that
out and even change the name to something like Term, Terminate or
Uninitialize().  Having said that - could all the work done in Init() be done in
the constructor?

[brandtr, 2017/06/28 09:40:24]

Documentation added. Semantics and name have changed to better match the
::Release() semantics of the underlying encoder.

The work done in Init() could be done in the ctor, but the work done in
Release() could not be done in the dtor. The reason for the latter being that we
want to ensure that the encoding/decoding is completely done (i.e., no lingering
frames anywhere) when we access the state of |this|, at the end of a measurement
run.

I'm keeping Init/Release this way, as it matches how we init and release the
codecs.

+
   // Processes a single frame. Returns true as long as there's more frames
   // available in the source clip.
   // Frame number must be an integer >= 0.
   virtual bool ProcessFrame(int frame_number) = 0;
 
   // Updates the encoder with the target bit rate and the frame rate.
   virtual void SetRates(int bit_rate, int frame_rate) = 0;
 
   // Return the size of the encoded frame in bytes. Dropped frames by the
   // encoder are regarded as zero size.
@@ skipping 14 matching lines @@
   VideoProcessorImpl(webrtc::VideoEncoder* encoder,
                      webrtc::VideoDecoder* decoder,
                      FrameReader* analysis_frame_reader,
                      FrameWriter* analysis_frame_writer,
                      PacketManipulator* packet_manipulator,
                      const TestConfig& config,
                      Stats* stats,
                      FrameWriter* source_frame_writer,
                      IvfFileWriter* encoded_frame_writer,
                      FrameWriter* decoded_frame_writer);
-  virtual ~VideoProcessorImpl();
+  ~VideoProcessorImpl() override;
   bool Init() override;
+  void DeregisterCallbacks() override;
   bool ProcessFrame(int frame_number) override;
 
  private:
   // Container that holds per-frame information that needs to be stored between
   // calls to Encode and Decode, as well as the corresponding callbacks. It is
   // not directly used for statistics -- for that, test::FrameStatistic is used.
   struct FrameInfo {
     FrameInfo()
         : timestamp(0),
           encode_start_ns(0),
           decode_start_ns(0),
           encoded_frame_size(0),
           encoded_frame_type(kVideoFrameDelta),
           decoded_width(0),
           decoded_height(0),
           manipulated_length(0) {}
 
     uint32_t timestamp;
     int64_t encode_start_ns;
     int64_t decode_start_ns;
     size_t encoded_frame_size;
     FrameType encoded_frame_type;
     int decoded_width;
     int decoded_height;
     size_t manipulated_length;
   };
 
-  // Callback class required to implement according to the VideoEncoder API.
+  // Callback adapter class for encoder callback.
+  // Will post the callback on the task queue it was created on, if such a
+  // queue exists.
   class VideoProcessorEncodeCompleteCallback
       : public webrtc::EncodedImageCallback {
    public:
-    explicit VideoProcessorEncodeCompleteCallback(VideoProcessorImpl* vp)
-        : video_processor_(vp) {}
+    explicit VideoProcessorEncodeCompleteCallback(
+        VideoProcessorImpl* video_processor)
+        : video_processor_(video_processor),
+          task_queue_(rtc::TaskQueue::Current()) {}

[tommi, 2017/03/21 10:03:27]

if task_queue_ should never be null, please DCHECK that in the ctor so that we
catch the error as early as possible

[brandtr, 2017/06/28 09:40:24]

|task_queue_| can be null when the VideoProcessor is operated on a "normal"
thread. This happens in the SW codec tests on the bots, but not in the manual HW
codec tests on device.

The next step should be to split up these tests, so we don't have to make this
distinction.

+
     Result OnEncodedImage(
         const webrtc::EncodedImage& encoded_image,
         const webrtc::CodecSpecificInfo* codec_specific_info,
         const webrtc::RTPFragmentationHeader* fragmentation) override {
-      // Forward to parent class.
       RTC_CHECK(codec_specific_info);
+
+      if (task_queue_ && !task_queue_->IsCurrent()) {

[tommi, 2017/03/21 10:03:27]

it feels to me that the class has too many ways of operating.  I'd prefer to
either require a task queue or not. If it requires a task queue, we should never
need to check if task_queue_ is null.  It would also be good to either never
have to Post or always, inside this function.  If OnEncodedImage is always
called outside the task queue, we shouldn't have to check if it is, etc. Rather
DCHECK the expected condition.

[brandtr, 2017/06/28 09:40:24]

I agree that it is messy. The underlying reason is our lack of guarantees for
what thread is being used for the codec callbacks. This is currently my
impression of how things are implemented:

Encoders
==
- SW VP8 calls OnEncodedImage synchronously (+)
- Android MediaCodec calls OnEncodedImage from a recurring task running on an
external task queue
- iOS VideoToolbox calls OnEncodedImage from an internal GCD task queue (*)

Decoders
==
- SW VP8 calls Decoded synchronously (+)
- Android MediaCodec calls Decoded from an internal thread (*)
- iOS VideoToolbox calls Decoded from an internal GCD task queue (*)

For the cases marked by (+), this class is not created on a task queue, and the
callbacks are directly forwarded. For the cases marked by (*), this class is
created on a task queue but the callbacks are arriving on some other thread, so
this class posts the callbacks on the correct task queue. For the final case
(Android encode), the callback is arriving on the same task queue as the class
was created on, so it is directly forwarded.

By splitting up the bot tests from the manual tests, I might be able to clean
this up and require the VideoProcessor to run on a task queue at all times.

+        task_queue_->PostTask(std::unique_ptr<rtc::QueuedTask>(
+            new EncodeCallbackTask(video_processor_, encoded_image,
+                                   codec_specific_info, fragmentation)));
+        return Result(Result::OK, 0);
+      }
+
       video_processor_->FrameEncoded(codec_specific_info->codecType,
                                      encoded_image, fragmentation);
       return Result(Result::OK, 0);
     }
 
    private:
+    class EncodeCallbackTask : public rtc::QueuedTask {
+     public:
+      EncodeCallbackTask(VideoProcessorImpl* video_processor,
+                         const webrtc::EncodedImage& encoded_image,
+                         const webrtc::CodecSpecificInfo* codec_specific_info,
+                         const webrtc::RTPFragmentationHeader* fragmentation)
+          : video_processor_(video_processor),
+            buffer_(encoded_image._buffer, encoded_image._length),
+            encoded_image_(encoded_image),
+            codec_specific_info_(*codec_specific_info) {
+        encoded_image_._buffer = buffer_.data();
+        RTC_CHECK(fragmentation);
+        fragmentation_.CopyFrom(*fragmentation);
+      }
+
+      bool Run() override {
+        video_processor_->FrameEncoded(codec_specific_info_.codecType,
+                                       encoded_image_, &fragmentation_);
+        return true;
+      }
+
+     private:
+      VideoProcessorImpl* const video_processor_;
+      rtc::Buffer buffer_;
+      webrtc::EncodedImage encoded_image_;
+      const webrtc::CodecSpecificInfo codec_specific_info_;
+      webrtc::RTPFragmentationHeader fragmentation_;
+    };
+
     VideoProcessorImpl* const video_processor_;
+    rtc::TaskQueue* const task_queue_;
   };
 
-  // Callback class required to implement according to the VideoDecoder API.
+  // Callback adapter class for decoder callback.
+  // Will post the callback on the task queue it was created on, if such a
+  // queue exists.
   class VideoProcessorDecodeCompleteCallback
       : public webrtc::DecodedImageCallback {
    public:
-    explicit VideoProcessorDecodeCompleteCallback(VideoProcessorImpl* vp)
-        : video_processor_(vp) {}
+    explicit VideoProcessorDecodeCompleteCallback(
+        VideoProcessorImpl* video_processor)
+        : video_processor_(video_processor),
+          task_queue_(rtc::TaskQueue::Current()) {}
+
     int32_t Decoded(webrtc::VideoFrame& image) override {
-      // Forward to parent class.
+      if (task_queue_ && !task_queue_->IsCurrent()) {
+        task_queue_->PostTask(std::unique_ptr<rtc::QueuedTask>(
+            new DecodeCallbackTask(video_processor_, image)));
+        return 0;
+      }
+
       video_processor_->FrameDecoded(image);
       return 0;
     }
+
     int32_t Decoded(webrtc::VideoFrame& image,
                     int64_t decode_time_ms) override {
       return Decoded(image);
     }
+
     void Decoded(webrtc::VideoFrame& image,
                  rtc::Optional<int32_t> decode_time_ms,
                  rtc::Optional<uint8_t> qp) override {
-      Decoded(image,
-              decode_time_ms ? static_cast<int32_t>(*decode_time_ms) : -1);
+      Decoded(image);
     }
 
    private:
+    class DecodeCallbackTask : public rtc::QueuedTask {
+     public:
+      DecodeCallbackTask(VideoProcessorImpl* video_processor,
+                         const webrtc::VideoFrame& image)
+          : video_processor_(video_processor), image_(image) {}
+
+      bool Run() override {
+        video_processor_->FrameDecoded(image_);
+        return true;
+      }
+
+     private:
+      VideoProcessorImpl* const video_processor_;
+      webrtc::VideoFrame image_;
+    };
+
     VideoProcessorImpl* const video_processor_;
+    rtc::TaskQueue* const task_queue_;
   };
 
   // Invoked by the callback when a frame has completed encoding.
   void FrameEncoded(webrtc::VideoCodecType codec,
                     const webrtc::EncodedImage& encodedImage,
                     const webrtc::RTPFragmentationHeader* fragmentation);
 
   // Invoked by the callback when a frame has completed decoding.
   void FrameDecoded(const webrtc::VideoFrame& image);
 
   // Updates the encoder with the target bit rate and the frame rate.
   void SetRates(int bit_rate, int frame_rate) override;
 
   // Return the size of the encoded frame in bytes.
   size_t EncodedFrameSize(int frame_number) override;
 
   // Return the encoded frame type (key or delta).
   FrameType EncodedFrameType(int frame_number) override;
 
   // Return the number of dropped frames.
   int NumberDroppedFrames() override;
 
   // Return the number of spatial resizes.
   int NumberSpatialResizes() override;
 
-  webrtc::VideoEncoder* const encoder_;
-  webrtc::VideoDecoder* const decoder_;
-  const std::unique_ptr<VideoBitrateAllocator> bitrate_allocator_;
+  webrtc::VideoEncoder* const encoder_ PT_GUARDED_BY(task_checker_);
+  webrtc::VideoDecoder* const decoder_ PT_GUARDED_BY(task_checker_);
+  const std::unique_ptr<VideoBitrateAllocator> bitrate_allocator_
+      PT_GUARDED_BY(task_checker_);
 
   // Adapters for the codec callbacks.
-  const std::unique_ptr<EncodedImageCallback> encode_callback_;
-  const std::unique_ptr<DecodedImageCallback> decode_callback_;
+  std::unique_ptr<EncodedImageCallback> encode_callback_
+      GUARDED_BY(task_checker_);
+  std::unique_ptr<DecodedImageCallback> decode_callback_
+      GUARDED_BY(task_checker_);
 
-  PacketManipulator* const packet_manipulator_;
+  PacketManipulator* const packet_manipulator_ PT_GUARDED_BY(task_checker_);
   const TestConfig& config_;
 
   // These (mandatory) file manipulators are used for, e.g., objective PSNR and
   // SSIM calculations at the end of a test run.
-  FrameReader* const analysis_frame_reader_;
-  FrameWriter* const analysis_frame_writer_;
+  FrameReader* const analysis_frame_reader_ PT_GUARDED_BY(task_checker_);
+  FrameWriter* const analysis_frame_writer_ PT_GUARDED_BY(task_checker_);
   const int num_frames_;
 
   // These (optional) file writers are used for persistently storing the output
   // of the coding pipeline at different stages: pre encode (source), post
   // encode (encoded), and post decode (decoded). The purpose is to give the
   // experimenter an option to subjectively evaluate the quality of the
   // encoding, given the test settings. Each frame writer is enabled by being
   // non-null.
-  FrameWriter* const source_frame_writer_;
-  IvfFileWriter* const encoded_frame_writer_;
-  FrameWriter* const decoded_frame_writer_;
+  FrameWriter* const source_frame_writer_ PT_GUARDED_BY(task_checker_);
+  IvfFileWriter* const encoded_frame_writer_ PT_GUARDED_BY(task_checker_);
+  FrameWriter* const decoded_frame_writer_ PT_GUARDED_BY(task_checker_);
 
-  bool initialized_;
+  bool initialized_ GUARDED_BY(task_checker_);
 
   // Frame metadata for all frames that have been added through a call to
   // ProcessFrames(). We need to store this metadata over the course of the
   // test run, to support pipelining HW codecs.
-  std::vector<FrameInfo> frame_infos_;
-  int last_encoded_frame_num_;
-  int last_decoded_frame_num_;
+  std::vector<FrameInfo> frame_infos_ GUARDED_BY(task_checker_);
+  int last_encoded_frame_num_ GUARDED_BY(task_checker_);
+  int last_decoded_frame_num_ GUARDED_BY(task_checker_);
 
   // Keep track of if we have excluded the first key frame from packet loss.
-  bool first_key_frame_has_been_excluded_;
+  bool first_key_frame_has_been_excluded_ GUARDED_BY(task_checker_);
 
   // Keep track of the last successfully decoded frame, since we write that
   // frame to disk when decoding fails.
-  rtc::Buffer last_decoded_frame_buffer_;
+  rtc::Buffer last_decoded_frame_buffer_ GUARDED_BY(task_checker_);
 
   // Statistics.
-  Stats* stats_;
-  int num_dropped_frames_;
-  int num_spatial_resizes_;
-  double bit_rate_factor_;  // Multiply frame length with this to get bit rate.
+  Stats* stats_ PT_GUARDED_BY(task_checker_);
+  int num_dropped_frames_ GUARDED_BY(task_checker_);
+  int num_spatial_resizes_ GUARDED_BY(task_checker_);
+  // Multiply frame length with this to get bit rate.
+  double bit_rate_factor_ GUARDED_BY(task_checker_);
+
+  rtc::SequencedTaskChecker task_checker_;
 };
 
 }  // namespace test
 }  // namespace webrtc
 
 #endif  // WEBRTC_MODULES_VIDEO_CODING_CODECS_TEST_VIDEOPROCESSOR_H_