24

为了提供一些背景信息，我正在处理一个保存的文件，在使用正则表达式将文件拆分为其组件对象之后，我需要根据对象的类型来处理对象的数据。

我目前的想法是使用并行性来获得一点性能提升，因为加载每个对象是相互独立的。所以我要定义一个LoadObject函数，接受std::string我要处理的每种类型的对象，然后调用std::async如下：

void LoadFromFile( const std::string& szFileName )
{
     static const std::regex regexObject( "=== ([^=]+) ===\\n((?:.|\\n)*)\\n=== END \\1 ===", std::regex_constants::ECMAScript | std::regex_constants::optimize );

     std::ifstream inFile( szFileName );
     inFile.exceptions( std::ifstream::failbit | std::ifstream::badbit );

     std::string szFileData( (std::istreambuf_iterator<char>(inFile)), (std::istreambuf_iterator<char>()) );

     inFile.close();

     std::vector<std::future<void>> vecFutures;

     for( std::sregex_iterator itObject( szFileData.cbegin(), szFileData.cend(), regexObject ), end; itObject != end; ++itObject )
     {
          // Determine what type of object we're loading:
          if( (*itObject)[1] == "Type1" )
          {
               vecFutures.emplace_back( std::async( LoadType1, (*itObject)[2].str() ) );
          }
          else if( (*itObject)[1] == "Type2" )
          {
               vecFutures.emplace_back( std::async( LoadType2, (*itObject)[2].str() ) );
          }
          else
          {
               throw std::runtime_error( "Unexpected type encountered whilst reading data file." );
          }
     }

     // Make sure all our tasks completed:
     for( auto& future : vecFutures )
     {
           future.get();
     }
}

请注意，应用程序中将有超过 2 种类型（这只是一个简短的示例），并且文件中可能有数千个要读取的对象。

我知道创建太多线程通常对性能不利，因为上下文切换导致它超过了最大硬件并发，但是如果我的记忆正确地为我服务，C++ 运行时应该监控创建的线程数量并std::async适当地安排（我相信微软的情况是他们的 ConcRT 库对此负责？），所以上面的代码仍然可能导致性能提升？

提前致谢！

4

2 回答 2

20

C++ 运行时应该监视创建的线程数并适当地安排 std::async

不。如果异步任务实际上是异步运行的（而不是延迟），那么所需要的只是它们像在新线程上一样运行。为每个任务创建和启动一个新线程是完全有效的，而无需考虑硬件的并行能力有限。

有一个注释：

[ 注意：如果此策略与其他策略一起指定，例如在使用策略值 launch::async | launch::deferred，当不能有效地利用更多并发时，实现应该推迟调用或策略的选择。——尾注]

但是，这是非规范性的，并且在任何情况下，它都表明一旦不能再利用并发性，任务可能会被延迟，因此在有人等待结果时被执行，而不是仍然是异步的并在其中一个之后立即运行先前的异步任务已完成，这对于最大并行度是可取的。

也就是说，如果我们有 10 个长时间运行的任务，而实现只能并行执行 4 个，那么前 4 个将是异步的，然后后 6 个可能会被延迟。按顺序等待 future 将在单个线程上按顺序执行延迟任务，从而消除这些任务的并行执行。

该说明还说，可以推迟策略的选择，而不是推迟调用。也就是说，该函数可能仍异步运行，但该决定可能会延迟，例如，直到较早的任务之一完成，从而为新任务释放核心。但同样，这不是必需的，该注释是非规范性的，据我所知，微软的实现是唯一一个以这种方式运行的实现。当我查看另一个实现 libc++ 时，它完全忽略了这个注释，因此使用其中一个std::launch::async或std::launch::any策略会导致在新线程上异步执行。

（我相信微软的情况是他们的 ConcRT 库对此负责？）

Microsoft 的实现确实如您所描述的那样运行，但这不是必需的，可移植程序不能依赖该行为。

可移植地限制实际运行的线程数量的一种方法是使用信号量之类的东西：

#include <future>
#include <vector>
#include <mutex>
#include <cstdio>

// a semaphore class
//
// All threads can wait on this object. When a waiting thread
// is woken up, it does its work and then notifies another waiting thread.
// In this way only n threads will be be doing work at any time.
// 
class Semaphore {
private:
    std::mutex m;
    std::condition_variable cv;
    unsigned int count;

public:
    Semaphore(int n) : count(n) {}
    void notify() {
        std::unique_lock<std::mutex> l(m);
        ++count;
        cv.notify_one();
    }
    void wait() {
        std::unique_lock<std::mutex> l(m);
        cv.wait(l, [this]{ return count!=0; });
        --count;
    }
};

// an RAII class to handle waiting and notifying the next thread
// Work is done between when the object is created and destroyed
class Semaphore_waiter_notifier {
    Semaphore &s;
public:
    Semaphore_waiter_notifier(Semaphore &s) : s{s} { s.wait(); }
    ~Semaphore_waiter_notifier() { s.notify(); }
};

// some inefficient work for our threads to do
int fib(int n) {
    if (n<2) return n;
    return fib(n-1) + fib(n-2);
}

// for_each algorithm for iterating over a container but also
// making an integer index available.
//
// f is called like f(index, element)
template<typename Container, typename F>
F for_each(Container &c, F f) {
    typename Container::size_type i = 0;
    for (auto &e : c)
        f(i++, e);
    return f;
}

// global semaphore so that lambdas don't have to capture it
Semaphore thread_limiter(4);

int main() {
    std::vector<int> input(100);
    for_each(input, [](int i, int &e) { e = (i%10) + 35; });

    std::vector<std::future<int>> output;
    for_each(input, [&output](int i, int e) {
        output.push_back(std::async(std::launch::async, [] (int task, int n) -> int {
            Semaphore_waiter_notifier w(thread_limiter);
            std::printf("Starting task %d\n", task);
            int res = fib(n);
            std::printf("\t\t\t\t\t\tTask %d finished\n", task);
            return res;
        }, i, e));
    });

    for_each(output, [](int i, std::future<int> &e) {
        std::printf("\t\t\tWaiting on task %d\n", i);
        int res = e.get();
        std::printf("\t\t\t\t\t\t\t\t\tTask %d result: %d\n", i, res);
    });
}
于 2013-06-19T18:06:32.630 回答
-1

在这里发布到一个旧线程，但一本优秀的书是https://www.amazon.com/C-Concurrency-Action-Practical-Multithreading/dp/1933988770

但是，根据最近的技术趋势，GPU/CPU 的协作和并发性可能会带来更大的性能优势。

于 2020-01-29T17:43:32.960 回答