C++11使用多线程（线程池）计算相似度实现性能优化

需求：图像识别中，注册的样本多了会影响计算速度，成为性能瓶颈，其中一个优化方法就是使用多线程。例如，注册了了3000个特征，每个特征4096个float。可以把3000个特征比对放到4个线程中进行计算，然后再把结果进行合并。实现思路：

1. 根据系统性能和需求配置线程池的大小，创建线程池，将比较任务平均分配到各个线程

2. 工作线程启动后在一个condition_variable上wait，注意：锁的范围不能太大了，否则多个线程会变成串行

3. 调用者调用识别接口，接口更新目标特征，通知各个工作线程，在另外一个condition_variable上wait，并且满足完成计数器的值等于线程数

4. 工作线程完成后将计数器加一，并且通知调用线程

5. 调用线程收集到所有线程的结果后再对结果进行合并返回

后续：

1. 代码进行优化，更优雅的实现

测试结果：

线程数	时间
1	71362ms
2	36292ms
4	19420ms
8	18465ms
16	18433ms
32	18842ms
64	19324ms
128	19388ms
256	21853ms
512	26150ms
1024	35593ms

代码如下：

#include <iostream>

#include <string>

#include <cstring>

#include <mutex>

#include <unordered_map>

#include <list>

#include <utility>

#include <algorithm>

#include <string>

#include <vector>

#include <thread>

#include <chrono>

using namespace std;

using namespace chrono;

 

double get_mold(const vector<double> &vec)

{

int n = vec.size();

double sum = 0.0;

for(int i = 0; i < n; ++i)

{

sum += vec[i] * vec[i];

}

return sqrt(sum);

}

double cosine_distance(const vector<double> &base, const vector<double> &target)

{

int n = base.size();

double tmp = 0.0;

for(int i = 0; i < n; ++i)

{

tmp += base[i] * target[i];

}

double simility = tmp / (get_mold(base) * get_mold(target));

return simility;

}

 

class Recognizer

{

public:

Recognizer(int num_threads) :

num_threads_(num_threads),

is_run_calculate_thread_(true),

is_doing_recognize(false),

result_count(0)

{

recognize_result = std::vector<RecognizeResult>(num_threads);

this->load_feature();

this->init_threads();

}

~Recognizer()

{

is_run_calculate_thread_ = false;

cv_.notify_all();

for(std::thread &th : threads_)

th.join();

}

int do_recognize(const vector<double> &feature);

private:

class CigaretteItem

{

public:

int cigarette_id_;

std::string cigarette_name_;

std::vector<double> feature_;

CigaretteItem(int cigarette_id, std::string cigarette_name, const std::vector<double> &cigarette_feature)

{

cigarette_id_ = cigarette_id;

cigarette_name_ = cigarette_name;

feature_ = std::vector<double>(cigarette_feature.size());

for(int i = 0; i < cigarette_feature.size(); i++)

{

feature_[i] = cigarette_feature[i];

}

}

};

class RecognizeResult

{

public:

int cigarette_id_;

std::string cigarette_name_;

double score_;

};

private:

int num_threads_;

bool is_run_calculate_thread_;

bool is_doing_recognize;

std::vector<CigaretteItem> ciagarette_list_;

std::vector<double> target_feature_;

std::mutex cv_mtx_;

std::condition_variable cv_;

std::vector<RecognizeResult> recognize_result;

std::vector<std::thread> threads_;

int result_count;

std::mutex result_count_mtx_;

std::mutex result_cv_mtx_;

std::condition_variable result_cv_;

 

private:

Recognizer(const Recognizer&) = delete;

Recognizer& operator=(const Recognizer&) = delete;

void load_feature();

void init_threads();

void calculate_most_similarity(const int thread_id, const int start_index, const int end_index);

};

 

void Recognizer::load_feature()

{

for(int i = 0; i < 3000; i++)

{

vector<double> fea = vector<double>(4096);

for(int i = 0; i < 4096; ++i)

fea[i] = (double)(rand() % 998 + 1) / 1000.00;

ciagarette_list_.emplace_back(i+1, "cigarette", fea);

}

}

void Recognizer::init_threads()

{

for(int i = 0; i < num_threads_; i++)

{

int step = this->ciagarette_list_.size() / this->num_threads_;

int start_index = i * step;

int end_index = (i+1) * step;

if(i == num_threads_ - 1){

end_index = ciagarette_list_.size();

}

std::cout << "thread" << i << " starts at " << start_index << "; ends at " << end_index << std::endl;

threads_.emplace_back(&Recognizer::calculate_most_similarity, this, i, start_index, end_index);

}

}

void Recognizer::calculate_most_similarity(const int thread_id, const int start_index, const int end_index)

{

while(is_run_calculate_thread_)

{

{

std::unique_lock<std::mutex> lock(cv_mtx_);

cv_.wait(lock);

}

//cout << "thread" << thread_id << " is running" << endl;

double max_score = -1.00;

int max_score_index = -1;

for(int i = start_index; i < end_index; ++i){

double score = cosine_distance(ciagarette_list_[i].feature_, target_feature_);

if(score > max_score)

{

max_score = score;

max_score_index = i;

}

}

recognize_result[thread_id].cigarette_id_ = ciagarette_list_[max_score_index].cigarette_id_;

recognize_result[thread_id].cigarette_name_ = ciagarette_list_[max_score_index].cigarette_name_;

recognize_result[thread_id].score_ = max_score;

{

std::unique_lock<std::mutex> lock(result_count_mtx_);

result_count += 1;

}

result_cv_.notify_one();

//std::cout << "thread" << thread_id << " finish one task" << endl;

}

//std::cout << "thread" << thread_id << " finished." << std::endl;

}

int Recognizer::do_recognize(const vector<double> &feature)

{

if(is_doing_recognize)

return -1;

is_doing_recognize = true;

this->target_feature_ = feature;

//cout << "cv_.notify_all()" << endl;

cv_.notify_all();

std::unique_lock<std::mutex> lock(result_cv_mtx_);

result_cv_.wait(lock, [this](){return this->num_threads_ == this->result_count;});

//std::cout << "all threads finish computing similarity" << endl;

int max_score_cigarette_id = -1;

std::string max_score_cigarette_name = "";

double max_score = -1.0;

for(int i = 0; i < num_threads_; ++i)

{

if(recognize_result[i].score_ > max_score)

{

max_score_cigarette_id = recognize_result[i].cigarette_id_;

max_score_cigarette_name = recognize_result[i].cigarette_name_;

max_score = recognize_result[i].score_;

}

}

//cout << "cigarette_id=" << max_score_cigarette_id << ", cigarette_name=" << max_score_cigarette_name << ", score=" << max_score << endl;

this->result_count = 0;

is_doing_recognize = false;

return 0;

}

 

int main(void)

{

Recognizer recognizer{1024};

std::this_thread::sleep_for(std::chrono::seconds(1));

const int loops = 400;

auto start_time = system_clock::now();

for(int i = 0; i < loops; i++)

{

//cout << endl;

std::vector<double> target_feature = std::vector<double>(4096);

for(int i = 0; i < 4096; ++i)

{

//target_feature[i] = (double)(rand() % 998 + 1) / 1000.000;

target_feature[i] = (double)(i % 1000 + 1) / 1000.00;

}

recognizer.do_recognize(target_feature);

//if((i+1) % 20 == 0)

// cout << "i=" << i << endl;

}

auto end_time = system_clock::now();

auto duration = duration_cast<milliseconds>(end_time - start_time);

cout << "eplased_time:" << duration.count() << "ms" << endl;

std::this_thread::sleep_for(std::chrono::seconds(2));

return 0;

}