在性能关键的并行代码中，我有一个 vector ，其元素是:

• 计算成本很高，结果是确定性的(给定位置的元素值将仅取决于位置)
• 随机访问(通常访问次数大于或远大于 vector 的大小)
• 集群访问(许多访问请求相同的值)
• vector 由不同的线程共享(竞争条件？)
• 为避免堆碎片整理，永远不要重新创建对象，而是尽可能重置和回收
• 放置在 vector 中的值将由多态对象提供

目前，我预先计算了 vector 的所有可能值，因此竞争条件应该不是问题。 为了提高性能，我正在考虑创建一个惰性 vector ，以便代码仅在请求 vector 元素时才执行计算。 在并行区域中，可能会发生多个线程同时请求并可能计算同一元素的情况。 我如何处理这种可能的竞争条件？

下面是我想要实现的示例。它在 Windows 10、Visual Studio 17 下编译和运行正常。我使用 C++17。

// Lazy.cpp : Defines the entry point for the console application.
#include "stdafx.h"
#include <vector>
#include <iostream>
#include <stdlib.h>  
#include <chrono>
#include <math.h>
const double START_SUM = 1;
const double END_SUM = 1000;

//base object responsible for providing the values
class Evaluator
{
public:
    Evaluator() {};
    ~Evaluator() {};
    //Function with deterministic output, depending on the position
    virtual double expensiveFunction(int pos) const = 0;
};
//
class EvaluatorA: public Evaluator
{
public:
    //expensive evaluation
    virtual double expensiveFunction(int pos) const override {
        double t = 0;
        for (int j = START_SUM; j++ < END_SUM; j++)
            t += log(exp(log(exp(log(j + pos)))));
        return t;
    }
    EvaluatorA() {};
    ~EvaluatorA() {};
};
class EvaluatorB : public Evaluator
{
public:
    //even more expensive evaluation
    virtual double expensiveFunction(int pos) const override {
        double t = 0;
        for (int j = START_SUM; j++ < 10*END_SUM; j++)
            t += log(exp(log(exp(log(j + pos)))));
        return t;
    }
    EvaluatorB() {};
    ~EvaluatorB() {};
};

class LazyVectorTest //vector that contains N possible results
{
public:
    LazyVectorTest(int N,const Evaluator & eval) : N(N), innerContainer(N, 0), isThatComputed(N, false), eval_ptr(&eval)
    {};
    ~LazyVectorTest() {};

    //reset, to generate a new table of values
    //the size of the vector stays constant
    void reset(const Evaluator & eval) {
        this->eval_ptr = &eval;
        for (int i = 0; i<N; i++)
            isThatComputed[i] = false;
    }
    int size() { return N; }
    //accessing the same position should yield the same result
    //unless the object is resetted
    const inline double& operator[](int pos) {
        if (!isThatComputed[pos]) {
            innerContainer[pos] = eval_ptr->expensiveFunction(pos);
            isThatComputed[pos] = true;
        }
        return innerContainer[pos];
    }

private:
    const int N;
    const Evaluator* eval_ptr;
    std::vector<double> innerContainer;
    std::vector<bool> isThatComputed;
};
    //the parallel access will take place here

template <typename T>
double accessingFunction(T& A, const std::vector<int>& elementsToAccess) {
    double tsum = 0;
    int size = elementsToAccess.size();
//#pragma omp parallel for
    for (int i = 0; i < size; i++)
        tsum += A[elementsToAccess[i]];
    return tsum;
}

std::vector<int> randomPos(int sizePos, int N) {
    std::vector<int> elementsToAccess;
    for (int i = 0; i < sizePos; i++)
        elementsToAccess.push_back(rand() % N);
    return elementsToAccess;
}
int main()
{
    srand(time(0));
    int minAccessNumber = 1;
    int maxAccessNumber = 100;
    int sizeVector = 50;

    auto start = std::chrono::steady_clock::now();
    double res = 0;
    float numberTest = 100;
    typedef LazyVectorTest container;

    EvaluatorA eval;
    for (int i = 0; i < static_cast<int>(numberTest); i++) {
        res = eval.expensiveFunction(i);
    }
    auto end = std::chrono::steady_clock::now();
    std::chrono::duration<double, std::milli>diff(end - start);
    double benchmark = diff.count() / numberTest;
    std::cout <<"Average time to compute expensive function:" <<benchmark<<" ms"<<std::endl;
    std::cout << "Value of the function:" << res<< std::endl;
    std::vector<std::vector<int>> indexs(numberTest);
    container A(sizeVector, eval);

    for (int accessNumber = minAccessNumber; accessNumber < maxAccessNumber; accessNumber++) {
        indexs.clear();
        for (int i = 0; i < static_cast<int>(numberTest); i++) {
            indexs.emplace_back(randomPos(accessNumber, sizeVector));
        }
        auto start_lazy = std::chrono::steady_clock::now();
        for (int i = 0; i < static_cast<int>(numberTest); i++) {
            A.reset(eval);
            double res_lazy = accessingFunction(A, indexs[i]);
        }
        auto end_lazy = std::chrono::steady_clock::now();
        std::chrono::duration<double, std::milli>diff_lazy(end_lazy - start_lazy);
        std::cout << accessNumber << "," << diff_lazy.count() / numberTest << ", " << diff_lazy.count() / (numberTest* benchmark) << std::endl;
    }
    return 0;
}

最佳答案

我不会使用您自己的锁定，而是首先查看您是否通过 std::call_once 获得可接受的性能。

class LazyVectorTest //vector that contains N possible results
{
    //Function with deterministic output, depending on the position
    void expensiveFunction(int pos) {
        double t = 0;
        for (int j = START_SUM; j++ < END_SUM; j++)
            t += log(exp(log(exp(log(j+pos)))));
        values[pos] = t;
    }

public:
    LazyVectorTest(int N) : values(N), flags(N)
    {};

    int size() { return values.size(); }
    //accessing the same position should yield the same result
    double operator[](int pos) {
        std::call_once(flags[pos], &LazyVectorTest::expensiveFunction, this, pos);
        return values[pos];
    }
private:
    std::vector<double> values;
    std::vector<std::once_flag> flags;
};

call_once 非常透明。它只允许一个线程运行一个函数直到完成。唯一的潜在缺点是它会阻止第二个线程等待可能的异常，而不是立即什么都不做。在这种情况下，这是可取的，因为您希望在读取 return values[pos];

关于c++ - 并行循环中的惰性 vector 访问，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/51356863/