c++ - 多次使用 std::async 对小任务性能友好吗？

Question

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

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

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 库对此负责？)，所以上面的代码仍然可能导致性能提升？

提前致谢!

Answer 1

the C++ runtime is supposed to monitor the number of threads created and schedule std::async appropriately

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

有一个注释:

[ Note: If this policy is specified together with other policies, such as when using a policy value of launch::async | launch::deferred,implementations should defer invocation or the selection of the policywhen no more concurrency can be effectively exploited. —end note ]

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

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

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

(I believe in Microsoft's case their ConcRT library is responsible for this?)

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);
    });
}