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

Question

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

我目前的想法是使用并行性来获得一点性能提升，因为加载每个对象是彼此独立的。所以我要定义一个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库对此负责？），所以上面的代码仍然可能导致性能提升？

提前致谢！

Answer 1

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

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

Answer 2

C++ 运行时应该监控创建的线程数并适当地调度 std::async

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

有一个注释：

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

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

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

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

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

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

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

#include <future>
#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) {
    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);
    });
}