Add threads for computeL and computeR

In case of post-processing, LR check is required
which a left and a right disparities are mandatory.
To reduce excution time, add two threads which compute
the both disparities at the same time.

Signed-off-by: Tae-Young Chung <ty83.chung@samsung.com>

diff --git a/src/dfs_opencv.cpp b/src/dfs_opencv.cpp
index 2c396ce91246069515586acc8ef93e931111db3b..47d5dc054f453cf7c3955aae970a539da05a419b 100644 (file)
--- a/src/dfs_opencv.cpp
+++ b/src/dfs_opencv.cpp
@@ -1,5 +1,5 @@
 /**
- * Copyright (c) 2019 Samsung Electronics Co., Ltd All Rights Reserved
+ * Copyright (c) 2021 Samsung Electronics Co., Ltd All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@
 #define DEFAULT_STEREO_VGA_CALIB_FILE_NAME "stereoCalibZedVGA.yaml"
 #define DEFAULT_STEREO_HD_CALIB_FILE_NAME "stereoCalibZedHD.yaml"
 
+#define MAX_THREADS_NUM 2
 namespace DfsAdaptationImpl
 {
        DfsOCV::DfsOCV() :
@@ -43,7 +44,8 @@ namespace DfsAdaptationImpl
                mIsStereoCalibrated(false),
                mUpdateStereoCalibration(false),
                mImageSize(cv::Size(0,0)),
-               mDownScale(1)
+               mDownScale(1),
+               mIsStopAll(false)
        {
                LOGI("ENTER");
                LOGI("LEAVE");
@@ -52,6 +54,14 @@ namespace DfsAdaptationImpl
        DfsOCV::~DfsOCV()
        {
                LOGI("ENTER");
+               mIsStopAll = true;
+               mControlJob.notify_all();
+
+               for (auto& t : mThreadPool) {
+                       LOGI("thread joinable: %s", t.joinable() ? "true" : "false");
+                       if (t.joinable())
+                               t.join();
+               }
                LOGI("LEAVE");
        }
 
@@ -193,6 +203,14 @@ namespace DfsAdaptationImpl
                mDfsPostOcv->setSigmaColor(10.0);
                mDfsPostOcv->setLambda(8000);
 
+               if (!mDfsOcvExtra) {
+                       mThreadPool.reserve(MAX_THREADS_NUM - 1);
+                       mThreadPool.emplace_back([this](){this->Runner();});
+               } else {
+                       mThreadPool.reserve(MAX_THREADS_NUM);
+                       mThreadPool.emplace_back([this](){this->Runner();});
+                       mThreadPool.emplace_back([this](){this->Runner();});
+               }
                LOGI("LEAVE");
        }
 
@@ -226,6 +244,67 @@ namespace DfsAdaptationImpl
                LOGI("ENTER");
        }
 
+       bool DfsOCV::computeL(const cv::Mat& baseMat, const cv::Mat& extraMat, cv::Mat& disp)
+       {
+               LOGI("ENTER");
+
+               mDfsOcv->compute(baseMat, extraMat, disp);
+
+               LOGI("LEAVE");
+               return true;
+       }
+
+       bool DfsOCV::computeR(const cv::Mat& baseMat, const cv::Mat& extraMat, cv::Mat& disp)
+       {
+               LOGI("ENTER");
+
+               mDfsOcvExtra->compute(baseMat, extraMat, disp);
+
+               LOGI("LEAVE");
+               return true;
+       }
+
+       void DfsOCV::Runner()
+       {
+               while (true) {
+                       std::unique_lock<std::mutex> lock(mMutexJob);
+                       mControlJob.wait(lock, [this]() {return !mJobs.empty() || mIsStopAll;});
+                       if (mIsStopAll && mJobs.empty()) {
+                               LOGI("Stop and all jobs are done");
+                               return;
+                       }
+
+                       std::function<void()> job = std::move(mJobs.front());
+                       mJobs.pop();
+                       lock.unlock();
+
+                       job();
+               }
+       }
+
+       template <class F, class... Args>
+       std::future<bool> DfsOCV::EnqueueJob(F f, Args... args)
+       {
+               LOGE("ENTER");
+               if (mIsStopAll) {
+                       LOGE("Thread pool stopped");
+                       throw std::runtime_error("thread pool stopped");
+               }
+
+               auto job = std::make_shared<std::packaged_task<bool()>>(
+                                       std::bind(f, this, args...));
+
+               std::future<bool> results = job->get_future();
+               {
+                       std::lock_guard<std::mutex> lock(mMutexJob);
+                       mJobs.push([job]() { (*job)(); });
+               }
+               mControlJob.notify_one();
+
+               LOGE("LEAVE");
+               return results;
+       }
+
        void DfsOCV::Run(DfsData& base, DfsData& extra)
        {
                LOGI("ENTER");
@@ -272,12 +351,20 @@ namespace DfsAdaptationImpl
                                        1.0/static_cast<double>((1<<mDownScale)),
                                        1.0/static_cast<double>((1<<mDownScale)));
 
-               mDfsOcv->compute(srcBaseMat, srcExtraMat, dispMat);
+               std::vector<std::future<bool>> results;
+               results.emplace_back(EnqueueJob(&DfsOCV::computeL,
+                                                                               std::cref(srcBaseMat),
+                                                                               std::cref(srcExtraMat),
+                                                                               std::ref(dispMat)));
                if (mDfsPostOcv) {
                        if (mDfsOcvExtra) {
                                cv::Mat dispMatExtra;
-                               mDfsOcvExtra->compute(srcExtraMat, srcBaseMat, dispMatExtra);
-
+                               results.emplace_back(EnqueueJob(&DfsOCV::computeR,
+                                                                               std::cref(srcExtraMat),
+                                                                               std::cref(srcBaseMat),
+                                                                               std::ref(dispMatExtra)));
+                               LOGI("left: %s, right: %s", results[0].get() ? "true" : "false",
+                                                                                       results[1].get() ? "true" : "false");
                                if (mDownScale) {
                                        cv::Mat tmp;
                                        // base
@@ -299,6 +386,7 @@ namespace DfsAdaptationImpl
                                                                        cv::Rect(0,0,rBaseMat.cols, rBaseMat.rows),
                                                                        rExtraMat);
                        } else {
+                               LOGI("left : %s", results[0].get() ? "true" : "false");
                                if (mDownScale) {
                                        cv::Mat tmp;
                                        cv::resize(dispMat, tmp,
@@ -312,6 +400,7 @@ namespace DfsAdaptationImpl
 
                        dispFiltMat.convertTo(mDispMat, CV_8UC1, 1.0/16.0);
                } else {
+                       LOGI("left : %s", results[0].get() ? "true" : "false");
                        if (mDownScale) {
                                cv::Mat tmp;
                                cv::resize(dispMat, tmp,

diff --git a/src/dfs_opencv_private.h b/src/dfs_opencv_private.h
index ebca4fcc8ec5b59dea3858d4735a56520163bcea..eb49d61563344b1a6219efd48cf50d1e86ebd8c7 100644 (file)
--- a/src/dfs_opencv_private.h
+++ b/src/dfs_opencv_private.h
@@ -19,6 +19,12 @@
 
 #include <dfs_adaptation.h>
 #include <dfs_parameter.h>
+#include <thread>
+#include <future>
+#include <queue>
+#include <mutex>
+#include <condition_variable>
+#include <functional>
 
 #include <opencv2/core.hpp>
 #include <opencv2/calib3d.hpp>
@@ -96,11 +102,26 @@ namespace DfsAdaptationImpl
 
                size_t mDownScale;
 
+               bool mIsStopAll;
+               std::vector<std::thread> mThreadPool;
+               std::queue<std::function<void()> > mJobs;
+               std::mutex mMutexJob;
+               std::condition_variable mControlJob;
+
+
+               void Runner();
+
+               template <class F, class... Args>
+               std::future<bool> EnqueueJob(F f, Args... args);
+
                void SetParameters();
                int  ConvertDfsDataTypeToCV(int type);
                void InitializeStereoCalibration();
                void InitRectifyMap();
 
+               bool computeL(const cv::Mat& baseMat, const cv::Mat& extraMat, cv::Mat& disp);
+               bool computeR(const cv::Mat& baseMat, const cv::Mat& extraMat, cv::Mat& disp);
+
        public:
                DfsOCV();
                ~DfsOCV();