c++ - OpenCV:如何对带有图像的文件夹进行批处理？

Question

我有图像文件夹，我对它们执行一系列基本操作:

1. 加载源图像。
2. 对图像执行一些图像处理。
3. 保存结果图像。

所以我想在单独的线程中处理每个图像以加速处理。

这是我的示例代码:

ThreadExample.h

#include <thread>


    class ThreadProcessing
    {
         static unsigned int concurentThreadsSupported;

         static void ImageProcessingFunction(const std::string &input_dir, const std::string &filename);
    public:
         void PrintNumberOfCPU();
         void MultithreadingProcessing(const std::string &dir, int N);
         void SingleThreadProcessing(const std::string &dir);
    };

ThreadExample.cpp

    #include "ThreadExample.h"

    unsigned int ThreadProcessing::concurentThreadsSupported = std::thread::hardware_concurrency();

    using namespace std;

    void ThreadProcessing::PrintNumberOfCPU()
    {
        cout << "Number of CPU : " << concurentThreadsSupported << endl;
    }

void ThreadProcessing::ImageProcessingFunction(const string &input_dir, const string &filename)
    {
        Mat src= imread(input_dir+"/"+filename);
        Mat dst;

        for(int i=0; i<10; ++i)
        {
            medianBlur(src, dst, 71);
        }

        boost::filesystem::path name= path(filename).stem();

        string output_filename= (input_dir/name).string()+"_output.png";
        imwrite(output_filename, dst);
    }

    void ThreadProcessing::SingleThreadProcessing(const string &dir)
    {
        time_t SingleThreadProcessingTime = clock();

        vector<string> imageNames= GetAllFilenamesInDir(dir, ".jpg");

        for(int i=0; i<(int)imageNames.size(); ++i)
        {
            ImageProcessingFunction(dir, imageNames[i]);
        }

        SingleThreadProcessingTime = clock() - SingleThreadProcessingTime;
        cout << "SingleThreadProcessingTime : " << (float(SingleThreadProcessingTime) / CLOCKS_PER_SEC) << endl;
    }

    void ThreadProcessing::MultithreadingProcessing(const string &dir, int N)
    {
        time_t MultithreadingProcessingTime = clock();

        std::thread threads[N];

        bool isAllImageProcessed= false;
        vector<string> imageNames= GetAllFilenamesInDir(dir, ".jpg");

        for(int i=0; i<(int)imageNames.size();)
        {
            //Launch a group of threads
            for(int k= 0; k< N; ++k)
            {
                threads[k] = std::thread(ImageProcessingFunction, dir, imageNames[i]);

                i++;

                if(i>=(int)imageNames.size())
                {
                    N= k+1;
                    isAllImageProcessed= true;
                    break;
                }
            }

            //Join the threads with the main thread
            for(int k= 0; k< N; ++k)
            {
                threads[k].join();
            }

            if(isAllImageProcessed)
                break;
        }

        MultithreadingProcessingTime = clock() - MultithreadingProcessingTime;
        cout << "MultithreadingProcessingTime : " << (float(MultithreadingProcessingTime) / CLOCKS_PER_SEC) << endl;
    }

main.cpp

int main(int argc, char** argv)
{
    ThreadProcessing threadProcessing;
    threadProcessing.PrintNumberOfCPU();
    threadProcessing.SingleThreadProcessing("/home/user/Images");
    threadProcessing.MultithreadingProcessing("/home/user/Images", 1);
    cout << "Done." << endl;
    return 0;
}

但是好像没有速度提升:

当我使用 1 个线程时，输出是:

Number of CPU : 8
SingleThreadProcessingTime : 6.54173
MultithreadingProcessingTime : 6.73393
Done.

当我使用 4 个线程时，输出是:

Number of CPU : 8
SingleThreadProcessingTime : 6.39089
MultithreadingProcessingTime : 8.3365
Done.

我的代码有问题还是概念上有问题？

更新:

我还尝试了 2 种变体:

1. 每个图像 1 个线程 - 似乎这种方法可以达到最大线程数操作系统限制？而且效率低下？

代码:

void ThreadProcessing::SingleThreadForEachImage(const string &dir)
{
    time_t SingleThreadForEachImageTime = clock();

    vector<string> imageNames= GetAllFilenamesInDir(dir, ".jpg");

    int N= imageNames.size();
    std::thread threads[imageNames.size()];

    for(int i=0; i<N; ++i)
    {

        threads[i] = std::thread(ImageProcessingFunction, dir, imageNames[i]);
    }

    for(int i=0; i<N; ++i)
    {
        threads[i].join();
    }

    SingleThreadForEachImageTime = clock() - SingleThreadForEachImageTime;
    cout << "SingleThreadForEachImageTime : " << (float(SingleThreadForEachImageTime) / CLOCKS_PER_SEC) << endl;
}

1. 将图像拆分为 N 个 block ，并在单独的线程中处理每个 block 。

代码:

vector<vector<string>> ThreadProcessing::SplitNamesVector(const vector<string> &imageNames, int N)
{
    vector<vector<string>> imageNameChunks;

    int K=0; //Number images in chunk
    if(imageNames.size()%N==0)
        K= imageNames.size()/N;
    else
        K= imageNames.size()/N+1;

    vector<string> chunk;
    for(int i=0; i<(int)imageNames.size(); ++i)
    {
        chunk.push_back(imageNames[i]);
        if(i%K==0 && i!=0)
        {
            imageNameChunks.push_back(chunk);
            chunk.clear();
        }
    }

    if(chunk.size()!=0)
        imageNameChunks.push_back(chunk);

    assert((int)imageNameChunks.size()==N);

    return imageNameChunks;
}

void ThreadProcessing::EachThreadProcessChunkOfImages(const std::string &dir, int N)
{
    time_t EachThreadProcessChunkOfImagesTime = clock();

    N= std::min(N, (int)concurentThreadsSupported);
    std::thread threads[N];

    vector<string> imageNames= GetAllFilenamesInDir(dir, ".jpg");
    vector<vector<string>> imageNameChunks= SplitNamesVector(imageNames, N);

    //Launch a group of threads
    for(int k= 0; k< N; ++k)
    {
        threads[k] = std::thread(ImageProcessingFunctionChunk, dir, imageNameChunks[k]);
    }

    for(int k= 0; k< N; ++k)
    {
        threads[k].join();
    }

    EachThreadProcessChunkOfImagesTime = clock() - EachThreadProcessChunkOfImagesTime;
    cout << "EachThreadProcessChunkOfImagesTime : " << (float(EachThreadProcessChunkOfImagesTime) / CLOCKS_PER_SEC) << endl;
}

这是结果(MultithreadingProcessing 和 EachThreadProcessChunkOfImages 使用 4 个线程):

SingleThreadProcessingTime : 13.552
MultithreadingProcessingTime : 15.581
SingleThreadForEachImageTime : 26.7727
EachThreadProcessChunkOfImagesTime : 15.9078

更新 2:

我也做了一个没有IO操作的测试，只有图像处理。

代码:

void ThreadProcessing::ImageProcessingFunction(const cv::Mat &img)
{
    Mat dst;

    for(int i=0; i<10; ++i)
    {
        medianBlur(img, dst, 71);
    }
}
    vector<Mat> ThreadProcessing::LoadBatchOfImages(const std::string &dir, int batchSize)
    {
        vector<string> imageNames= GetAllFilenamesInDir(dir, ".jpg");

        vector<Mat> imageVec;
        for(int i=0; i<N; ++i)
        {
            string filename= dir+"/"+imageNames[i];
            Mat img= imread(filename);
            imageVec.push_back(img);
        }

        return imageVec;
    }

    void ThreadProcessing::OnlyProcessingTimeSequential(const std::string &dir, int batchSize)
    {
        //Load batch of images
        vector<Mat> imageVec= LoadBatchOfImages(dir, batchSize);

        assert((int)imageVec.size() == batchSize);
        cout << "imageVec.size() : " << imageVec.size() << endl;

        time_t OnlyProcessingTimeSequentialTime = clock();

        for(int i=0; i<batchSize; ++i)
        {
            ImageProcessingFunction(imageVec[i]);
        }

        OnlyProcessingTimeSequentialTime = clock() - OnlyProcessingTimeSequentialTime;
        cout << "OnlyProcessingTimeSequentialTime : " << (float(OnlyProcessingTimeSequentialTime) / CLOCKS_PER_SEC) << endl;
    }

    void ThreadProcessing::OnlyProcessingTimeMultithread(const std::string &dir, int batchSize)
    {
        //Load batch of images
        vector<Mat> imageVec= LoadBatchOfImages(dir, batchSize);

        assert((int)imageVec.size() == batchSize);
        cout << "imageVec.size() : " << imageVec.size() << endl;

        time_t OnlyProcessingTimeMultithread = clock();

        std::thread threads[batchSize];
        for(int i=0; i<batchSize; ++i)
        {
            threads[i] = std::thread(ImageProcessingFunction, imageVec[i]);
        }

        for(int i=0; i<batchSize; ++i)
        {
            threads[i].join();
        }

        OnlyProcessingTimeMultithread = clock() - OnlyProcessingTimeMultithread;
        cout << "OnlyProcessingTimeMultithread : " << (float(OnlyProcessingTimeMultithread) / CLOCKS_PER_SEC) << endl;
    }

我发现 clock() 在使用多线程代码时会给出错误的结果，所以我使用 time ./MyBinary

结果如下:

imageVec.size() : 8
OnlyProcessingTimeSequentialTime : 2.34174
Done.

real    0m2.551s
user    0m2.640s
sys 0m0.316s


imageVec.size() : 8
OnlyProcessingTimeMultithread : 4.36681
Done.

real    0m0.861s
user    0m4.564s
sys 0m0.404s

正如我们所见，真实时间更小。

所以之前的结果应该是:

SingleThreadProcessingTime : 13.6235

real    0m13.845s
user    0m13.932s
sys 0m0.280s

MultithreadingProcessingTime : 21.0902

real    0m3.584s
user    0m20.356s
sys 0m1.316s

SingleThreadForEachImageTime : 23.961

real    0m3.370s
user    0m22.584s
sys 0m1.976s

EachThreadProcessChunkOfImagesTime : 20.7885

real    0m3.433s
user    0m20.292s
sys 0m1.116s

那么多线程代码执行时间应该如何衡量呢？

更新:找到答案 here

我需要使用 wallclock time 而不是 cpu time

这是正确的结果:

SingleThreadProcessing :  WALLCLOCK TIME: 13.8245 seconds
MultithreadingProcessing :  WALLCLOCK TIME: 4.1977 seconds
SingleThreadForEachImage :  WALLCLOCK TIME: 3.25084 seconds
EachThreadProcessChunkOfImages :  WALLCLOCK TIME: 3.36626 seconds
OnlyProcessingTimeSequential :  WALLCLOCK TIME: 2.36041 seconds
OnlyProcessingTimeMultithread :  WALLCLOCK TIME: 0.706921 seconds

Answer 1

正如 this question 中明确指出的那样，多线程在单盘I/O操作上确实效率不高。您的线程所做的大部分工作都是 I/O 操作。

您可能受到磁盘速度的限制，并且由于创建线程和 join() 操作而导致开销增加，这会导致函数的多线程版本变慢。

编辑:

正如@Dan Mašek 在他的评论中所说，大部分时间实际上都花在了压缩上。改进流程的一种方法是创建一个线程，从磁盘读取图像并将它们提供给其他工作线程(可能通过 queue，参见 Thread pool)。这样，您可以按顺序阅读，但繁重的工作由很多 worker 完成。