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

webrtc/modules/video_coding/codecs/test/videoprocessor_integrationtest.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.
  */
 
@@ skipping 10 matching lines @@
 #if defined(WEBRTC_ANDROID)
 #include "webrtc/modules/video_coding/codecs/test/android_test_initializer.h"
 #include "webrtc/sdk/android/src/jni/androidmediadecoder_jni.h"
 #include "webrtc/sdk/android/src/jni/androidmediaencoder_jni.h"
 #elif defined(WEBRTC_IOS)
 #include "webrtc/sdk/objc/Framework/Classes/h264_video_toolbox_decoder.h"
 #include "webrtc/sdk/objc/Framework/Classes/h264_video_toolbox_encoder.h"
 #endif
 
 #include "webrtc/base/checks.h"
+#include "webrtc/base/event.h"
 #include "webrtc/base/file.h"
 #include "webrtc/base/logging.h"
+#include "webrtc/base/task_queue.h"
 #include "webrtc/media/engine/webrtcvideodecoderfactory.h"
 #include "webrtc/media/engine/webrtcvideoencoderfactory.h"
 #include "webrtc/modules/video_coding/codecs/h264/include/h264.h"
 #include "webrtc/modules/video_coding/codecs/test/packet_manipulator.h"
 #include "webrtc/modules/video_coding/codecs/test/videoprocessor.h"
 #include "webrtc/modules/video_coding/codecs/vp8/include/vp8.h"
 #include "webrtc/modules/video_coding/codecs/vp8/include/vp8_common_types.h"
 #include "webrtc/modules/video_coding/codecs/vp9/include/vp9.h"
 #include "webrtc/modules/video_coding/include/video_codec_interface.h"
 #include "webrtc/modules/video_coding/include/video_coding.h"
 #include "webrtc/modules/video_coding/utility/ivf_file_writer.h"
+#include "webrtc/system_wrappers/include/sleep.h"
 #include "webrtc/test/gtest.h"
 #include "webrtc/test/testsupport/fileutils.h"
 #include "webrtc/test/testsupport/frame_reader.h"
 #include "webrtc/test/testsupport/frame_writer.h"
 #include "webrtc/test/testsupport/metrics/video_metrics.h"
 #include "webrtc/test/testsupport/packet_reader.h"
 #include "webrtc/typedefs.h"
 
 namespace webrtc {
 namespace test {
 // Maximum number of rate updates (i.e., calls to encoder to change bitrate
 // and/or frame rate) for the current tests.
 const int kMaxNumRateUpdates = 3;
 
 // Maximum number of temporal layers to use in tests.
 const int kMaxNumTemporalLayers = 3;
 
 const int kPercTargetvsActualMismatch = 20;
 const int kBaseKeyFrameInterval = 3000;
 
 // Default sequence is foreman (CIF): may be better to use VGA for resize test.
 const int kCifWidth = 352;
 const int kCifHeight = 288;
 const char kFilenameForemanCif[] = "foreman_cif";
 
+const int kNumMillisecondsPerSecond = 1000;

[sprang_webrtc, 2017/03/20 19:36:17]

use rtc::kNumMillisPerSecond instead?

[brandtr, 2017/03/21 09:40:54]

Done.

+
 // Codec and network settings.
 struct CodecParams {
   VideoCodecType codec_type;
   bool hw_codec;
   bool use_single_core;
 
   int width;
   int height;
 
   int num_temporal_layers;
@@ skipping 150 matching lines @@
         encoder_.reset(VP9Encoder::Create());
         decoder_.reset(VP9Decoder::Create());
         break;
       default:
         RTC_NOTREACHED();
         break;
     }
   }
 
   void SetUpCodecConfig(const CodecParams& process,
+                        const RateProfile& rate_profile,
                         const VisualizationParams* visualization_params) {
     CreateEncoderAndDecoder(process.hw_codec, process.codec_type);
 
     // Configure input filename.
     config_.input_filename = test::ResourcePath(process.filename, "yuv");
     if (process.verbose_logging)
       printf("Filename: %s\n", process.filename.c_str());
     // Generate an output filename in a safe way.
     config_.output_filename = test::TempFilename(
         test::OutputPath(), "videoprocessor_integrationtest");
 
     config_.frame_length_in_bytes =
         CalcBufferSize(kI420, process.width, process.height);
     config_.verbose = process.verbose_logging;
     config_.use_single_core = process.use_single_core;
     // Key frame interval and packet loss are set for each test.
     config_.keyframe_interval = process.key_frame_interval;
     config_.networking_config.packet_loss_probability =
-        packet_loss_probability_;
+        process.packet_loss_probability;
 
     // Configure codec settings.
     VideoCodingModule::Codec(process.codec_type, &codec_settings_);
     config_.codec_settings = &codec_settings_;
-    config_.codec_settings->startBitrate = start_bitrate_;
+    float start_bitrate = rate_profile.target_bit_rate[0];
+    config_.codec_settings->startBitrate = start_bitrate;
     config_.codec_settings->width = process.width;
     config_.codec_settings->height = process.height;
 
     // These features may be set depending on the test.
     switch (config_.codec_settings->codecType) {
       case kVideoCodecH264:
         config_.codec_settings->H264()->frameDroppingOn =
             process.frame_dropper_on;
         config_.codec_settings->H264()->keyFrameInterval =
             kBaseKeyFrameInterval;
@@ skipping 29 matching lines @@
     analysis_frame_reader_.reset(new test::YuvFrameReaderImpl(
         config_.input_filename, config_.codec_settings->width,
         config_.codec_settings->height));
     analysis_frame_writer_.reset(new test::YuvFrameWriterImpl(
         config_.output_filename, config_.codec_settings->width,
         config_.codec_settings->height));
     RTC_CHECK(analysis_frame_reader_->Init());
     RTC_CHECK(analysis_frame_writer_->Init());
 
     if (visualization_params) {
+      int start_frame_rate = rate_profile.input_frame_rate[0];
       // clang-format off
       const std::string output_filename_base =
           test::OutputPath() + process.filename +
           "_cd-" + CodecTypeToPayloadName(process.codec_type).value_or("") +
           "_hw-" + std::to_string(process.hw_codec) +
-          "_fr-" + std::to_string(start_frame_rate_) +
-          "_br-" + std::to_string(static_cast<int>(start_bitrate_));
+          "_fr-" + std::to_string(start_frame_rate) +
+          "_br-" + std::to_string(static_cast<int>(start_bitrate));
       // clang-format on
       if (visualization_params->save_source_y4m) {
         source_frame_writer_.reset(new test::Y4mFrameWriterImpl(
             output_filename_base + "_source.y4m", config_.codec_settings->width,
-            config_.codec_settings->height, start_frame_rate_));
+            config_.codec_settings->height, start_frame_rate));
         RTC_CHECK(source_frame_writer_->Init());
       }
       if (visualization_params->save_encoded_ivf) {
         rtc::File post_encode_file =
             rtc::File::Create(output_filename_base + "_encoded.ivf");
         encoded_frame_writer_ =
             IvfFileWriter::Wrap(std::move(post_encode_file), 0);
       }
       if (visualization_params->save_decoded_y4m) {
         decoded_frame_writer_.reset(new test::Y4mFrameWriterImpl(
             output_filename_base + "_decoded.y4m",
             config_.codec_settings->width, config_.codec_settings->height,
-            start_frame_rate_));
+            start_frame_rate));
         RTC_CHECK(decoded_frame_writer_->Init());
       }
     }
 
     packet_manipulator_.reset(new test::PacketManipulatorImpl(
         &packet_reader_, config_.networking_config, config_.verbose));
     processor_.reset(new test::VideoProcessorImpl(
         encoder_.get(), decoder_.get(), analysis_frame_reader_.get(),
         analysis_frame_writer_.get(), packet_manipulator_.get(), config_,
         &stats_, source_frame_writer_.get(), encoded_frame_writer_.get(),
         decoded_frame_writer_.get()));
-    RTC_CHECK(processor_->Init());
   }
 
   // Reset quantities after each encoder update, update the target
   // per-frame bandwidth.
   void ResetRateControlMetrics(int num_frames_to_hit_target) {
     for (int i = 0; i < num_temporal_layers_; i++) {
       num_frames_per_update_[i] = 0;
       sum_frame_size_mismatch_[i] = 0.0f;
       sum_encoded_frame_size_[i] = 0.0f;
       encoding_bitrate_[i] = 0.0f;
@@ skipping 13 matching lines @@
     sum_encoded_frame_size_total_ = 0.0f;
     encoding_bitrate_total_ = 0.0f;
     perc_encoding_rate_mismatch_ = 0.0f;
     num_frames_to_hit_target_ = num_frames_to_hit_target;
     encoding_rate_within_target_ = false;
     sum_key_frame_size_mismatch_ = 0.0;
     num_key_frames_ = 0;
   }
 
   // For every encoded frame, update the rate control metrics.
-  void UpdateRateControlMetrics(int frame_number) {
+  void UpdateRateControlMetrics(int frame_number,
+                                FrameType frame_type,
+                                size_t encoded_frame_size) {
     RTC_CHECK_GE(frame_number, 0);
     int tl_idx = TemporalLayerIndexForFrame(frame_number);
-    FrameType frame_type = processor_->EncodedFrameType(frame_number);
-    float encoded_size_kbits =
-        processor_->EncodedFrameSize(frame_number) * 8.0f / 1000.0f;
+    float encoded_size_kbits = encoded_frame_size * 8.0f / 1000.0f;
 
     // Update layer data.
     // Update rate mismatch relative to per-frame bandwidth for delta frames.
     if (frame_type == kVideoFrameDelta) {
       // TODO(marpan): Should we count dropped (zero size) frames in mismatch?
       sum_frame_size_mismatch_[tl_idx] +=
           fabs(encoded_size_kbits - per_frame_bandwidth_[tl_idx]) /
           per_frame_bandwidth_[tl_idx];
     } else {
       float target_size = (frame_number == 0) ? target_size_key_frame_initial_
@@ skipping 14 matching lines @@
         sum_encoded_frame_size_total_ * frame_rate_ / num_frames_total_;
     perc_encoding_rate_mismatch_ =
         100 * fabs(encoding_bitrate_total_ - bit_rate_) / bit_rate_;
     if (perc_encoding_rate_mismatch_ < kPercTargetvsActualMismatch &&
         !encoding_rate_within_target_) {
       num_frames_to_hit_target_ = num_frames_total_;
       encoding_rate_within_target_ = true;
     }
   }
 
+  void UpdateRateControlMetrics(int frame_number, rtc::TaskQueue* task_queue) {
+    FrameType frame_type;
+    size_t encoded_frame_size;
+
+    if (task_queue) {
+      rtc::Event sync_event(false, false);
+      task_queue->PostTask([this, &frame_type, &encoded_frame_size,
+                            frame_number, &sync_event]() {
+        frame_type = processor_->EncodedFrameType(frame_number);
+        encoded_frame_size = processor_->EncodedFrameSize(frame_number);
+        sync_event.Set();
+      });
+      sync_event.Wait(rtc::Event::kForever);

[sprang_webrtc, 2017/03/20 19:36:17]

nit: ASSERT_TRUE() maybe?

[brandtr, 2017/03/21 09:40:54]

Done.

+    } else {
+      frame_type = processor_->EncodedFrameType(frame_number);
+      encoded_frame_size = processor_->EncodedFrameSize(frame_number);
+    }
+
+    UpdateRateControlMetrics(frame_number, frame_type, encoded_frame_size);
+  }
+
+  void UpdateRateControlMetrics(int frame_number) {
+    UpdateRateControlMetrics(frame_number, nullptr);
+  }
+
   // Verify expected behavior of rate control and print out data.
   void VerifyRateControlMetrics(int update_index,
-                                const RateControlThresholds& rc_expected) {
-    int num_dropped_frames = processor_->NumberDroppedFrames();
-    int num_resize_actions = processor_->NumberSpatialResizes();
+                                const RateControlThresholds& rc_expected,
+                                int num_dropped_frames,
+                                int num_resize_actions) {
     printf(
         "For update #: %d,\n"
         " Target Bitrate: %d,\n"
         " Encoding bitrate: %f,\n"
         " Frame rate: %d \n",
         update_index, bit_rate_, encoding_bitrate_total_, frame_rate_);
     printf(
         " Number of frames to approach target rate: %d, \n"
         " Number of dropped frames: %d, \n"
         " Number of spatial resizes: %d, \n",
@@ skipping 39 matching lines @@
     EXPECT_LE(num_frames_to_hit_target_, rc_expected.max_time_hit_target);
     EXPECT_LE(num_dropped_frames, rc_expected.max_num_dropped_frames);
     if (rc_expected.num_spatial_resizes >= 0) {
       EXPECT_EQ(rc_expected.num_spatial_resizes, num_resize_actions);
     }
     if (rc_expected.num_key_frames >= 0) {
       EXPECT_EQ(rc_expected.num_key_frames, num_key_frames_);
     }
   }
 
+  void VerifyRateControlMetrics(int update_index,
+                                const RateControlThresholds& rc_expected,
+                                rtc::TaskQueue* task_queue) {
+    int num_dropped_frames;
+    int num_resize_actions;
+
+    if (task_queue) {
+      rtc::Event sync_event(false, false);
+      task_queue->PostTask(
+          [this, &num_dropped_frames, &num_resize_actions, &sync_event]() {
+            num_dropped_frames = processor_->NumberDroppedFrames();
+            num_resize_actions = processor_->NumberSpatialResizes();
+            sync_event.Set();
+          });
+      sync_event.Wait(rtc::Event::kForever);
+    } else {
+      num_dropped_frames = processor_->NumberDroppedFrames();
+      num_resize_actions = processor_->NumberSpatialResizes();
+    }
+
+    VerifyRateControlMetrics(update_index, rc_expected, num_dropped_frames,
+                             num_resize_actions);
+  }
+
+  void VerifyRateControlMetrics(int update_index,
+                                const RateControlThresholds& rc_expected) {
+    VerifyRateControlMetrics(update_index, rc_expected, nullptr);
+  }
+
   void VerifyQuality(const test::QualityMetricsResult& psnr_result,
                      const test::QualityMetricsResult& ssim_result,
                      const QualityThresholds& quality_thresholds) {
     EXPECT_GT(psnr_result.average, quality_thresholds.min_avg_psnr);
     EXPECT_GT(psnr_result.min, quality_thresholds.min_min_psnr);
     EXPECT_GT(ssim_result.average, quality_thresholds.min_avg_ssim);
     EXPECT_GT(ssim_result.min, quality_thresholds.min_min_ssim);
   }
 
   // Temporal layer index corresponding to frame number, for up to 3 layers.
@@ skipping 53 matching lines @@
   // TODO(brandtr): Change the second last argument to be a
   // const std::vector<RateControlThresholds>&, so we can ensure that the user
   // does not expect us to do mid-clip rate updates when we are not able to,
   // e.g., when we are operating in batch mode.
   void ProcessFramesAndVerify(QualityThresholds quality_thresholds,
                               RateProfile rate_profile,
                               CodecParams process,
                               RateControlThresholds* rc_thresholds,
                               const VisualizationParams* visualization_params) {
     // Codec/config settings.
-    start_bitrate_ = rate_profile.target_bit_rate[0];
-    start_frame_rate_ = rate_profile.input_frame_rate[0];
-    packet_loss_probability_ = process.packet_loss_probability;
     num_temporal_layers_ = process.num_temporal_layers;
-    SetUpCodecConfig(process, visualization_params);
+    SetUpCodecConfig(process, rate_profile, visualization_params);
+
     // Update the temporal layers and the codec with the initial rates.
     bit_rate_ = rate_profile.target_bit_rate[0];
     frame_rate_ = rate_profile.input_frame_rate[0];
     SetTemporalLayerRates();
     // Set the initial target size for key frame.
     target_size_key_frame_initial_ =
         0.5 * kInitialBufferSize * bit_rate_layer_[0];
-    processor_->SetRates(bit_rate_, frame_rate_);
 
     // Process each frame, up to |num_frames|.
     int frame_number = 0;
     int update_index = 0;
     int num_frames = rate_profile.num_frames;
     ResetRateControlMetrics(
         rate_profile.frame_index_rate_update[update_index + 1]);
 
     if (process.batch_mode) {
       // In batch mode, we calculate the metrics for all frames after all frames
       // have been sent for encoding.
 
+      // AndroidMediaEncoder must be called on a task queue, so during the batch
+      // run we will call |processor_| on a task queue.
+      std::unique_ptr<rtc::TaskQueue> task_queue(
+          new rtc::TaskQueue("Integration test task queue."));

[sprang_webrtc, 2017/03/20 19:36:17]

might want to choose a shorter name. think some oses has as 12 character limit
or something on the thread names...

[brandtr, 2017/03/21 09:40:54]

Done.

+
+      // Initialize |processor_|, which initializes the encoder and decoder.
+      rtc::Event sync_event(false, false);
+      task_queue->PostTask([this, &sync_event]() {
+        RTC_CHECK(processor_->Init());
+        processor_->SetRates(bit_rate_, frame_rate_);
+        sync_event.Set();
+      });
+      sync_event.Wait(rtc::Event::kForever);
+
+      // Post all frames (in order) to encoder.
+      int start_frame_rate = rate_profile.input_frame_rate[0];
       // TODO(brandtr): Refactor "frame number accounting" so we don't have to
       // call ProcessFrame num_frames+1 times here.
       for (frame_number = 0; frame_number <= num_frames; ++frame_number) {
-        EXPECT_TRUE(processor_->ProcessFrame(frame_number));
+        task_queue->PostTask([this, frame_number]() {
+          EXPECT_TRUE(processor_->ProcessFrame(frame_number));
+        });
+
+        // HW codecs can get stressed out if frames are arriving to quickly,
+        // so we do some rough pacing here.

[sprang_webrtc, 2017/03/20 19:36:17]

How do they get "stressed out"? Any other form of backpressure we can add to
avoid sleeping?

[brandtr, 2017/03/21 09:40:54]

Expanded the comment.

Another approach would be to post delayed tasks with a linearly increasing delay
calculated something like "delay_ms = (1000/frame_rate) * frame_number". (This
assumes that the posting of the tasks is almost instantaneous.) This would leave
the pacing of the encode operations to the task queue but I would have to post a
delayed "synchronization task" at the end, which seems a bit odd. I don't fully
understand what happens when normal and delayed tasks are mixed on the same task
queue and that's why I went with the sleep operation instead.

+        if (process.hw_codec) {
+          SleepMs(kNumMillisecondsPerSecond / start_frame_rate);
+        }
       }
 
+      // Give the task queue some time to finish processing the posted frames,
+      // then shut down processing and synchronize with main thread.
+      SleepMs(1 * kNumMillisecondsPerSecond);

[sprang_webrtc, 2017/03/20 19:36:17]

drop "1 *"

[brandtr, 2017/03/21 09:40:54]

Done.

+      sync_event.Reset();
+      task_queue->PostTask([this, &sync_event]() {
+        EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, encoder_->Release());
+        EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, decoder_->Release());
+        processor_->DeregisterCallbacks();
+        sync_event.Set();
+      });
+      sync_event.Wait(rtc::Event::kForever);
+
+      // Calculate and print rate control metrics.
       for (frame_number = 0; frame_number < num_frames; ++frame_number) {
         ++num_frames_per_update_[TemporalLayerIndexForFrame(frame_number)];
         ++num_frames_total_;
-        UpdateRateControlMetrics(frame_number);
+        UpdateRateControlMetrics(frame_number, task_queue.get());
       }
+
+      VerifyRateControlMetrics(update_index, rc_thresholds[0],
+                               task_queue.get());
     } else {
       // In online mode, we calculate the metrics for a given frame right after
       // it has been sent for encoding.
 
       if (process.hw_codec) {
         LOG(LS_WARNING) << "HW codecs should mostly be run in batch mode, "
                            "since they may be pipelining.";
       }
 
+      // Initialize |processor_| on main thread.
+      RTC_CHECK(processor_->Init());
+      processor_->SetRates(bit_rate_, frame_rate_);
+
       while (frame_number < num_frames) {
         EXPECT_TRUE(processor_->ProcessFrame(frame_number));
 
         ++num_frames_per_update_[TemporalLayerIndexForFrame(frame_number)];
         ++num_frames_total_;
         UpdateRateControlMetrics(frame_number);
 
         ++frame_number;
 
         // If we hit another/next update, verify stats for current state and
         // update layers and codec with new rates.
         if (frame_number ==
             rate_profile.frame_index_rate_update[update_index + 1]) {
           VerifyRateControlMetrics(update_index, rc_thresholds[update_index]);
 
           // Update layer rates and the codec with new rates.
           ++update_index;
           bit_rate_ = rate_profile.target_bit_rate[update_index];
           frame_rate_ = rate_profile.input_frame_rate[update_index];
           SetTemporalLayerRates();
           ResetRateControlMetrics(
               rate_profile.frame_index_rate_update[update_index + 1]);
           processor_->SetRates(bit_rate_, frame_rate_);
         }
       }
       // TODO(brandtr): Refactor "frame number accounting" so we don't have to
       // call ProcessFrame one extra time here.
       EXPECT_TRUE(processor_->ProcessFrame(frame_number));
+
+      // Release codecs to make sure they have finished processing.
+      EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, encoder_->Release());
+      EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, decoder_->Release());
+      processor_->DeregisterCallbacks();
+
+      // Verify for final rate update.
+      VerifyRateControlMetrics(update_index, rc_thresholds[update_index]);
     }
-
-    // Verify rate control metrics for all frames (if in batch mode), or for all
-    // frames since the last rate update (if not in batch mode).
-    VerifyRateControlMetrics(update_index, rc_thresholds[update_index]);
     EXPECT_EQ(num_frames, frame_number);
     EXPECT_EQ(num_frames + 1, static_cast<int>(stats_.stats_.size()));
 
-    // Release encoder and decoder to make sure they have finished processing.
-    EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, encoder_->Release());
-    EXPECT_EQ(WEBRTC_VIDEO_CODEC_OK, decoder_->Release());
-
     // Close the analysis files before we use them for SSIM/PSNR calculations.
     analysis_frame_reader_->Close();
     analysis_frame_writer_->Close();
 
     // Close visualization files.
     if (source_frame_writer_) {
       source_frame_writer_->Close();
     }
     if (encoded_frame_writer_) {
       EXPECT_TRUE(encoded_frame_writer_->Close());
@@ skipping 151 matching lines @@
   float encoding_bitrate_total_;
   float perc_encoding_rate_mismatch_;
   int num_frames_to_hit_target_;
   bool encoding_rate_within_target_;
   int bit_rate_;
   int frame_rate_;
   float target_size_key_frame_initial_;
   float target_size_key_frame_;
   float sum_key_frame_size_mismatch_;
   int num_key_frames_;
-  float start_bitrate_;
-  int start_frame_rate_;
 
   // Codec and network settings.
-  float packet_loss_probability_;
   int num_temporal_layers_;
 };
 
 }  // namespace test
 }  // namespace webrtc
 
 #endif  // WEBRTC_MODULES_VIDEO_CODING_CODECS_TEST_VIDEOPROCESSOR_INTEGRATIONTEST_H_