23. Concurrency Processing - C++

Concurrency Processing

---

Prerequisites

C++

---

1. Concurrency Processing

It is very similar before posting about concurrency, But usually python support many libraries. So It is also meaningful if I build a concurrency environments with C++

There’re some elements.

• Worker functions (it is main worker each thread)
• Capturing data from cam independently
• Create lock and unlock when input or output memory should be protected from other that it can be revising

2. Concurrency Processing Code

#include <iostream>
#include <opencv2/opencv.hpp>
#include "CImageViewer.cpp"
#include "COperation.h"
#include <chrono>
#include <filesystem>
#include <string>

#include <thread>
#include <mutex>
#include <atomic>
#include <chrono>
#include <string>

struct SSharedFrame
{
    cv::Mat imgFrameGray;
    std::mutex mtx;
    std::atomic<int> version{0};
};

struct SWorkerResult
{
    cv::Mat dst;
    double timeMs = 0.0;
    bool success = false;
    int processedVersion = -1;
    std::mutex mtx;
};

void ConvolutionWorker(SSharedFrame& sFrameInput, SWorkerResult& sResult, int i32Type, std::atomic<bool>& bRun)
{
    COperation algmOper;

    int i32LastVersion = -1;

    while (bRun.load())
    {
        cv::Mat imgFrameSrc;
        int i32CurrentVersion = -1;

        {
            std::lock_guard<std::mutex> lock(sFrameInput.mtx);

            i32CurrentVersion = sFrameInput.version.load();

            if (!(sFrameInput.imgFrameGray.empty() || i32CurrentVersion == i32LastVersion))
                imgFrameSrc = sFrameInput.imgFrameGray.clone();
        }

        if (imgFrameSrc.empty())
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
            continue;
        }

        cv::Mat imgFramDst(imgFrameSrc.rows, imgFrameSrc.cols, CV_8UC1, cv::Scalar(0));

        auto start = std::chrono::high_resolution_clock::now();

        bool bResult = algmOper.Convolution5x5(imgFrameSrc, imgFramDst, i32Type);

        auto end = std::chrono::high_resolution_clock::now();
        double elapsedMs = std::chrono::duration<double, std::milli>(end - start).count();

        std::string strText = "Type " + std::to_string(i32Type) + " | " + std::to_string(elapsedMs) + " ms | " + (bResult ? "OK" : "FAIL");
        cv::putText(imgFramDst,strText,cv::Point(10, 30),cv::FONT_HERSHEY_SIMPLEX,0.7,cv::Scalar(255),2,cv::LINE_AA);

        {
            std::lock_guard<std::mutex> lock(sResult.mtx);

            sResult.dst = imgFramDst.clone();
            sResult.timeMs = elapsedMs;
            sResult.success = bResult;
            sResult.processedVersion = i32CurrentVersion;
        }

        i32LastVersion = i32CurrentVersion;
    }
}

int main()
{
    bool bResult = false;

    do
    {
        cv::VideoCapture cap(0);

        if (!cap.isOpened())
            break;

        cap.set(cv::CAP_PROP_FRAME_WIDTH, 640);
        cap.set(cv::CAP_PROP_FRAME_HEIGHT, 480);

        cv::Mat imgFrame;
        cap.read(imgFrame);

        if (imgFrame.empty())
            break;

        cv::Mat imgFrameGray;
        cv::cvtColor(imgFrame, imgFrameGray, cv::COLOR_BGR2GRAY);

        SSharedFrame SSharedFrame;
        {
            std::lock_guard<std::mutex> lock(SSharedFrame.mtx);
            SSharedFrame.imgFrameGray = imgFrameGray.clone();
            SSharedFrame.version = 1;
        }

        constexpr int CONV_COUNT = 7;

        std::vector<std::thread> vctThread;
        std::vector<SWorkerResult> vctResults(CONV_COUNT);
        std::atomic<bool> bRun{true};

        for (int i32Type = 1; i32Type <= CONV_COUNT; ++i32Type)
            vctThread.emplace_back(ConvolutionWorker,std::ref(SSharedFrame), std::ref(vctResults[i32Type - 1]), i32Type, std::ref(bRun));

        std::vector<cv::Mat> images(CONV_COUNT + 1, cv::Mat(imgFrameGray.rows, imgFrameGray.cols, CV_8UC1, cv::Scalar(0)));
        CImageViewer viewer("Convolution Compare - Concurrent", images);

        int frameVersion = 1;

        while (true)
        {
            cap.read(imgFrame);

            if (imgFrame.empty())
                break;

            cv::cvtColor(imgFrame, imgFrameGray, cv::COLOR_BGR2GRAY);

            {
                std::lock_guard<std::mutex> lock(SSharedFrame.mtx);
                SSharedFrame.imgFrameGray = imgFrameGray.clone();
                SSharedFrame.version = ++frameVersion;
            }

            images.clear();

            cv::Mat imgOrg = imgFrameGray.clone();
            cv::putText(imgOrg,"Original",cv::Point(10, 30),cv::FONT_HERSHEY_SIMPLEX,0.8,cv::Scalar(255),2,cv::LINE_AA);
            images.push_back(imgOrg);

            for (int i = 0; i < CONV_COUNT; ++i)
            {
                cv::Mat imgDisplay;

                {
                    std::lock_guard<std::mutex> lock(vctResults[i].mtx);

                    if (!vctResults[i].dst.empty())
                        imgDisplay = vctResults[i].dst.clone();
                }

                if (imgDisplay.empty())
                {
                    imgDisplay = cv::Mat(imgFrameGray.rows, imgFrameGray.cols, CV_8UC1, cv::Scalar(0));
                    cv::putText(imgDisplay,"Type " + std::to_string(i + 1) + " | Waiting...",cv::Point(10, 30),cv::FONT_HERSHEY_SIMPLEX,0.7,cv::Scalar(255),2,cv::LINE_AA);
                }

                images.push_back(imgDisplay);
            }

            viewer.UpdateImages(images);

            if (!viewer.ShowOnce(1))
                break;
        }

        bRun = false;

        for (auto& worker : vctThread)
        {
            if (worker.joinable())
                worker.join();
        }

        cap.release();
        cv::destroyAllWindows();

        bResult = true;
    }
    while(false);

    return 0;
}

Result: Concureency processing