この記事はおそらく [https://qiita.com/advent-calendar/2017/opencv](OpenCV Advent Calendar 2017) の 12/12 の記事です。

OpenCV で並列パイプライン処理(mutex, boost::lockfree, tbb::pipeline)の比較

動機

• VideoCapture -> 物体追跡とか重い処理 -> VideoWriter な処理がしたい
• パイプライン並列にできそう

int main_naive(string INPUT_VIDEO_PATH, string OUTPUT_VIDEO_PATH){
  auto reader = cv::VideoCapture{ INPUT_VIDEO_PATH };
  if(!reader.isOpened()){
    printf("C++: reader cannot opened\n");
    return EXIT_FAILURE;
  }
  int width = reader.get(CV_CAP_PROP_FRAME_WIDTH);
  int height = reader.get(CV_CAP_PROP_FRAME_HEIGHT);
  double fps = reader.get(CV_CAP_PROP_FPS);
  auto writer = cv::VideoWriter{ "appsrc ! videoconvert ! x264enc ! mp4mux ! filesink location=" + OUTPUT_VIDEO_PATH + "  ", 0, fps, cv::Size{ width, height }, true};
  cv::Mat src;
  cv::Mat mid1, mid2;
  cv::Mat dst;
  while(reader.read(src)){
    cv::cvtColor(src, mid1, cv::COLOR_BGR2GRAY);
    cv::threshold(mid1, mid2, 0, 255, cv::THRESH_BINARY);
    std::this_thread::sleep_for(std::chrono::milliseconds(10)); // なんか重い処理
    cv::cvtColor(mid2, dst, cv::COLOR_GRAY2BGR);
    writer.write(dst);
  }
  return EXIT_SUCCESS;
}

パイプライン並列を作る

mutex と queue を使う

• 古来からの技法
• lock_guard で Scoped Locking Pattern を使った

int main_mutex(cv::VideoCapture& reader, cv::VideoWriter& writer){
  using std::thread;
  using std::mutex;
  using std::lock_guard;
  using std::queue;
  mutex src_que_mutex;
  queue<cv::Mat*> src_que;
  mutex dst_que_mutex;
  queue<cv::Mat*> dst_que;
  auto source_th = thread{[&](){
    while(true){
      auto src = new cv::Mat{};
      if(reader.read(*src)){
        auto lg = lock_guard<mutex>{src_que_mutex};
        src_que.push(src);
      }else{
        delete src;
        auto lg = lock_guard<mutex>{src_que_mutex};
        src_que.push(nullptr);
        break;
      }
    }
  }};

  auto serial_th = thread{[&](){
    while(true){
      cv::Mat* src = nullptr;
      { 
        auto lg = lock_guard<mutex>{src_que_mutex};
        if(src_que.empty()){ continue; }
        src = src_que.front();
        src_que.pop();
      }
      if(src != nullptr){
        auto dst = new cv::Mat{};
        heavy(*src, *dst);
        delete src;
        {
          auto lg = lock_guard<mutex>{dst_que_mutex};
          dst_que.push(dst);
        }
      }else{
        auto lg = lock_guard<mutex>{dst_que_mutex};
        dst_que.push(nullptr);
        break;
      }
    }
  }};
  auto sink_th = thread{[&](){
    while(true){
      cv::Mat* dst = nullptr;
      { 
        auto lg = lock_guard<mutex>{dst_que_mutex};
        if(dst_que.empty()){ continue; }
        dst = dst_que.front();
        dst_que.pop();
      }
      if(dst != nullptr){
         writer.write(*dst);
         delete dst;
      }else{
        break;
      }
    }
  }};
  source_th.join();
  serial_th.join();
  sink_th.join();
  return EXIT_SUCCESS;
}

boost::lockfree::queue を使う

• 並列処理のキューといえばこれ
• mutex を気にしなくてもよいのでかなり気が楽

int main_lockfree(cv::VideoCapture& reader, cv::VideoWriter& writer){
  using std::thread;
  using boost::lockfree::queue;
  queue<cv::Mat*> src_que(128);
  queue<cv::Mat*> dst_que(128);
  auto source_th = thread([&]{
    while(true){
      cv::Mat* src = new cv::Mat{};
      if(reader.read(*src)){
        while(!src_que.push(src)){ "retry"; };
      }else{
        delete src;
        while(!src_que.push(nullptr)){ "retry"; }
        break;
      }
    }
  });
  auto serial_th = thread{[&](){
    while(true){
      cv::Mat* src = nullptr;
      while(!src_que.pop(src)){ "retry"; }
      if(src != nullptr){
        cv::Mat* dst = new cv::Mat{};
        heavy(*src, *dst);
        delete src;
        while(!dst_que.push(dst)){ "retry"; }
      }else{
        while(!dst_que.push(nullptr)){ "retry"; }
        break;
      }
    }
  }};
  auto sink_th = thread{[&](){
    while(true){
      cv::Mat* dst = nullptr;
      while(!dst_que.pop(dst)){ "retry"; }
      if(dst != nullptr){
        writer.write(*dst);
        delete dst;
      }else{
        break;
      }
    }
  }};
  source_th.join();
  serial_th.join();
  sink_th.join();
  return EXIT_SUCCESS;
}

Intel® Threading Building Blocks を使う

• 巨人の肩の上に立つ

class Source: public tbb::filter {
private:
  cv::VideoCapture reader;
public:
  Source(cv::VideoCapture& reader): filter(tbb::filter::mode::serial_in_order), reader{reader} {};
  void* operator()(void*){
    auto src = new cv::Mat{};
    if(reader.read(*src)){
      return src;
    }
    delete src;
    return nullptr;
  };
};

class SerialFilter: public tbb::filter {
public:
  SerialFilter(): filter(tbb::filter::mode::serial_in_order) {};
  void* operator()(void* ptr){
    auto src = static_cast<cv::Mat*>(ptr);
    auto dst = new cv::Mat{};
    heavy(*src, *dst);
    delete src;
    return dst;
  };
};

class Sink: public tbb::filter {
private:
  cv::VideoWriter writer;
public:
  Sink(cv::VideoWriter& writer): filter{tbb::filter::mode::serial_in_order}, writer{writer} {};
  void* operator()(void* ptr){
    auto dst = static_cast<cv::Mat*>(ptr);
    writer.write(*dst);
    delete dst;
    return nullptr;
  };
};

int main_tbb(cv::VideoCapture& reader, cv::VideoWriter& writer){
  tbb::task_scheduler_init init;
  tbb::pipeline pipe;
  auto source = Source{reader};
  auto filter = SerialFilter{};
  auto sink = Sink{writer};
  pipe.add_filter(source);
  pipe.add_filter(filter);
  pipe.add_filter(sink);
  pipe.run(4);
  return EXIT_SUCCESS;
}

結果

root@70ce88779368:/data# time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output naive.mp4 --mode 0


real    0m58.225s
user    1m29.259s
sys     0m8.369s
root@70ce88779368:/data# time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output mutex.mp4 --mode 1

real    0m47.476s
user    2m13.875s
sys     0m12.271s
root@70ce88779368:/data# time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output lockfree.mp4 --mode 2

real    0m49.268s
user    2m15.517s
sys     0m17.469s
root@70ce88779368:/data# time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output tbb.mp4 --mode 3

real    0m48.485s
user    1m33.380s
sys     0m31.989s

考察

• 並列化した結果はほぼ同じ。
• tbb 使った時の user 時間が非パイプライン並列版とあまり変わらないのが謎

まとめ

• 画像処理が重い場合、パイプライン並列でスループットを向上させられる
• 自分でキューとパイプラインを作るよりも tbb 使ったほうが楽
• 処理時間の比較、多段パイプライン処理の比較、スレッド数の比較などは今後の課題とする

reference

• オープンソース化された並列化テンプレートクラスライブラリ「Intel Threading Building Blocks」入門 -
  https://mag.osdn.jp/09/08/21/1128207/6
• Intel® Threading Building Blocks - https://www.threadingbuildingblocks.org/docs/help/index.htm
• Boost.Lockfree - http://www.boost.org/doc/libs/1_63_0/doc/html/lockfree.html

付録

開発環境の Dockerfile

docker build -t cpp-tbb-pipeline ./
wget http://www.sample-videos.com/video/mp4/720/big_buck_bunny_720p_30mb.mp4
docker run -ti -v `pwd`:/data --workdir /data cpp-tbb-pipeline /bin/bash

FROM ubuntu:16.04
ENV DEBIAN_FRONTEND "noninteractive"

RUN apt-get update -y
RUN apt-get -y \
  -o Dpkg::Options::="--force-confdef" \
  -o Dpkg::Options::="--force-confold" dist-upgrade

RUN apt-get install -y --no-install-recommends \
  dconf-tools \
  apt-transport-https software-properties-common ppa-purge apt-utils \
  ca-certificates git curl wget \
  tar zip unzip zlib1g-dev bzip2 libbz2-dev \
  openssl libssl-dev \
  zsh vim screen tree htop \
  sudo

# Install gcc and clang
RUN wget -O - https://apt.llvm.org/llvm-snapshot.gpg.key | apt-key add -
RUN add-apt-repository "deb http://apt.llvm.org/xenial/ llvm-toolchain-xenial-5.0 main"
RUN add-apt-repository ppa:ubuntu-toolchain-r/test
RUN apt-get update -y
RUN apt-get install -y --no-install-recommends \
  build-essential binutils cmake autoconf automake autogen pkg-config libtool \
  gcc-6 g++-6 gcc-7 g++-7 gdb \
  clang-5.0 lldb-5.0 lld-5.0
RUN update-alternatives --install /usr/bin/gcc gcc /usr/bin/gcc-6 20
RUN update-alternatives --install /usr/bin/g++ g++ /usr/bin/g++-6 20

RUN apt-get install -y libboost-all-dev
RUN apt-get install -y libtbb2 libtbb-dev
RUN apt-get install -y libopenblas-dev liblapacke-dev
RUN apt-get install -y libeigen3-dev
RUN apt-get install -y ffmpeg libavcodec-dev libavformat-dev libswscale-dev
RUN apt-get install -y \
  gstreamer1.0-tools \
  gstreamer1.0-libav \
  gstreamer1.0-libav-dbg \
  gstreamer1.0-plugins-base \
  gstreamer1.0-plugins-good \
  gstreamer1.0-plugins-ugly \
  gstreamer1.0-plugins-bad \
  libgstreamer1.0-dev \
  libgstreamer-plugins-base1.0-dev \
  libgstreamer-plugins-good1.0-dev \
  libgstreamer-plugins-bad1.0-dev
WORKDIR /opt
RUN git clone --recursive --depth 1 https://github.com/opencv/opencv.git && \
cd opencv && \
mkdir -p build && \
cd build && \
cmake \
  -DCMAKE_BUILD_TYPE=RELEASE \
  -DBUILD_DOCS=OFF \
  -DBUILD_EXAMPLES=OFF \
  -DBUILD_TESTS=OFF \
  -DBUILD_PERF_TESTS=OFF \
  -DBUILD_WITH_DEBUG_INFO=OFF \
  -DBUILD_SHARED_LIBS=ON \
  -DWITH_OPENCL=ON \
  -DWITH_OPENGL=ON \
  -DENABLE_FAST_MATH=ON \
  -DWITH_EIGEN=ON \
  -DWITH_TBB=ON \
  -DWITH_PTHREADS_PF=ON \
  -DWITH_IPP=ON \
  -DWITH_FFMPEG=ON \
  -DWITH_GSTREAMER=ON \
  ../ && \
make -j && \
make install && \
ldconfig

RUN apt-get install -y -f
RUN apt-get update -y
RUN apt-get upgrade -y
RUN apt-get dist-upgrade -y

RUN apt-get clean -y
RUN apt-get autoremove -y
RUN apt-get autoclean -y
RUN rm -rf /var/lib/apt/lists/* /var/cache/apt/archives/*

ソースコード

#include <iostream>
#include <cstdio>
#include <chrono>
#include <queue>
#include <mutex>
#include <thread>
#include <boost/program_options.hpp>
#include <boost/lockfree/queue.hpp>
#include <tbb/task_scheduler_init.h>
#include <tbb/pipeline.h>
#include <opencv2/opencv.hpp>
#include <opencv2/highgui/highgui.hpp>
#include <opencv2/imgproc.hpp>

using std::string;

void heavy(cv::Mat& src, cv::Mat& dst){
  cv::Mat mid1, mid2;
  cv::cvtColor(src, mid1, cv::COLOR_BGR2GRAY);
  cv::threshold(mid1, mid2, 0, 255, cv::THRESH_BINARY);
  cv::cvtColor(mid2, dst, cv::COLOR_GRAY2BGR);
  std::this_thread::sleep_for(std::chrono::milliseconds(10));
}

int main_naive(cv::VideoCapture& reader, cv::VideoWriter& writer){
  cv::Mat src;
  cv::Mat dst;
  while(reader.read(src)){
    heavy(src, dst);
    writer.write(dst);
  }
  return EXIT_SUCCESS;
}

int main_mutex(cv::VideoCapture& reader, cv::VideoWriter& writer){
  using std::thread;
  using std::mutex;
  using std::lock_guard;
  using std::queue;
  mutex src_que_mutex;
  queue<cv::Mat*> src_que;
  mutex dst_que_mutex;
  queue<cv::Mat*> dst_que;
  auto source_th = thread{[&](){
    while(true){
      auto src = new cv::Mat{};
      if(reader.read(*src)){
        auto lg = lock_guard<mutex>{src_que_mutex};
        src_que.push(src);
      }else{
        delete src;
        auto lg = lock_guard<mutex>{src_que_mutex};
        src_que.push(nullptr);
        break;
      }
    }
  }};
  auto serial_th = thread{[&](){
    while(true){
      cv::Mat* src = nullptr;
      { 
        auto lg = lock_guard<mutex>{src_que_mutex};
        if(src_que.empty()){ continue; }
        src = src_que.front();
        src_que.pop();
      }
      if(src != nullptr){
        auto dst = new cv::Mat{};
        heavy(*src, *dst);
        delete src;
        { 
          auto lg = lock_guard<mutex>{dst_que_mutex};
          dst_que.push(dst);
        }
      }else{
        auto lg = lock_guard<mutex>{dst_que_mutex};
        dst_que.push(nullptr);
        break;
      }
    }
  }};
  auto sink_th = thread{[&](){
    while(true){
      cv::Mat* dst = nullptr;
      { 
        auto lg = lock_guard<mutex>{dst_que_mutex};
        if(dst_que.empty()){ continue; }
        dst = dst_que.front();
        dst_que.pop();
      }
      if(dst != nullptr){
         writer.write(*dst);
         delete dst;
      }else{
        break;
      }
    }
  }};
  source_th.join();
  serial_th.join();
  sink_th.join();
  return EXIT_SUCCESS;
}

int main_lockfree(cv::VideoCapture& reader, cv::VideoWriter& writer){
  using std::thread;
  using boost::lockfree::queue;
  queue<cv::Mat*> src_que(128);
  queue<cv::Mat*> dst_que(128);
  auto source_th = thread([&]{
    while(true){
      cv::Mat* src = new cv::Mat{};
      if(reader.read(*src)){
        while(!src_que.push(src)){ "retry"; };
      }else{
        delete src;
        while(!src_que.push(nullptr)){ "retry"; }
        break;
      }
    }
  });
  auto serial_th = thread{[&](){
    while(true){
      cv::Mat* src = nullptr;
      while(!src_que.pop(src)){ "retry"; }
      if(src != nullptr){
        cv::Mat* dst = new cv::Mat{};
        heavy(*src, *dst);
        delete src;
        while(!dst_que.push(dst)){ "retry"; }
      }else{
        while(!dst_que.push(nullptr)){ "retry"; }
        break;
      }
    }
  }};
  auto sink_th = thread{[&](){
    while(true){
      cv::Mat* dst = nullptr;
      while(!dst_que.pop(dst)){ "retry"; }
      if(dst != nullptr){
        writer.write(*dst);
        delete dst;
      }else{
        break;
      }
    }
  }};
  source_th.join();
  serial_th.join();
  sink_th.join();
  return EXIT_SUCCESS;
}

class Source: public tbb::filter {
private:
  cv::VideoCapture reader;
public:
  Source(cv::VideoCapture& reader): filter(tbb::filter::mode::serial_in_order), reader{reader} {};
  void* operator()(void*){
    auto src = new cv::Mat{};
    if(reader.read(*src)){
      return src;
    }
    delete src;
    return nullptr;
  };
};

class SerialFilter: public tbb::filter {
public:
  SerialFilter(): filter(tbb::filter::mode::serial_in_order) {};
  void* operator()(void* ptr){
    auto src = static_cast<cv::Mat*>(ptr);
    auto dst = new cv::Mat{};
    heavy(*src, *dst);
    delete src;
    return dst;
  };
};

class Sink: public tbb::filter {
private:
  cv::VideoWriter writer;
public:
  Sink(cv::VideoWriter& writer): filter{tbb::filter::mode::serial_in_order}, writer{writer} {};
  void* operator()(void* ptr){
    auto dst = static_cast<cv::Mat*>(ptr);
    writer.write(*dst);
    delete dst;
    return nullptr;
  };
};

int main_tbb(cv::VideoCapture& reader, cv::VideoWriter& writer){
  tbb::task_scheduler_init init;
  tbb::pipeline pipe;
  auto source = Source{reader};
  auto filter = SerialFilter{};
  auto sink = Sink{writer};
  pipe.add_filter(source);
  pipe.add_filter(filter);
  pipe.add_filter(sink);
  pipe.run(4);
  return EXIT_SUCCESS;
}

int main(int argc, char* argv[]){
  string INPUT_VIDEO_PATH;
  string OUTPUT_VIDEO_PATH;
  int PIPELINE_MODE;
  namespace po = boost::program_options;
  po::options_description opt("option");
  opt.add_options()
    ("input,i", po::value<string>(&INPUT_VIDEO_PATH)->required(), "input video file path")
    ("output,o", po::value<string>(&OUTPUT_VIDEO_PATH)->required(), "output mp4 video file path if you need")
    ("mode", po::value<int>(&PIPELINE_MODE)->default_value(0), "0: naive, 1: mutex, 2: lockfree, 3: tbb mode")
    ("help,h", "show this menu");
  auto vm = po::variables_map{};
  try{
    po::store(po::parse_command_line(argc, argv, opt), vm);
    po::notify(vm);
  }catch(std::exception& e){
    std::cerr << "Error: " << e.what() << "\n";
    std::cout << opt << std::endl; // put help
    return EXIT_FAILURE;
  }
  if( vm.count("help")  ){
    std::cout << opt << std::endl; // put help
    return EXIT_FAILURE;
  }

  auto reader = cv::VideoCapture{ INPUT_VIDEO_PATH };
  if(!reader.isOpened()){
    printf("C++: reader cannot opened\n");
    return EXIT_FAILURE;
  }
  int width = reader.get(CV_CAP_PROP_FRAME_WIDTH);
  int height = reader.get(CV_CAP_PROP_FRAME_HEIGHT);
  double fps = reader.get(CV_CAP_PROP_FPS);

  auto writer = cv::VideoWriter{ "appsrc ! videoconvert ! x264enc ! mp4mux ! filesink location=" + OUTPUT_VIDEO_PATH + "  ", 0, fps, cv::Size{ width, height }, true};

  switch(PIPELINE_MODE){
    case 0: return main_naive(reader, writer);
    case 1: return main_mutex(reader, writer);
    case 2: return main_lockfree(reader, writer);
    case 3: return main_tbb(reader, writer);
  }
}

g++-7 main.cpp -std=c++1z -O3 -Wall -v `pkg-config --libs opencv` -lboost_program_options -ltbb -lpthread

time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output naive.mp4 --mode 0
time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output mutex.mp4 --mode 1
time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output lockfree.mp4 --mode 2
time ./a.out --input big_buck_bunny_720p_30mb.mp4 --output tbb.mp4 --mode 3