Có sử dụng std :: async nhiều lần cho các tác vụ nhỏ hiệu năng thân thiện không?

Question

Để cung cấp một số thông tin cơ bản, tôi đang xử lý tệp đã lưu và sau khi sử dụng biểu thức chính quy để chia tệp thành đối tượng thành phần, tôi cần xử lý dữ liệu của đối tượng dựa trên loại đối tượng đó.

Suy nghĩ hiện tại của tôi là sử dụng tính song song để có được một chút hiệu suất đạt được khi tải từng đối tượng độc lập với nhau. Vì vậy, tôi sẽ định nghĩa một hàm LoadObject chấp nhận một std::string đối với từng loại đối tượng tôi sẽ được xử lý và sau đó gọi std::async như sau:

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

Lưu ý rằng sẽ có nhiều hơn 2 loại trong ứng dụng (đây chỉ là một ví dụ ngắn) và có khả năng hàng ngàn đối tượng trong tệp được đọc.

Tôi biết rằng việc tạo quá nhiều chuỗi thường là điều xấu cho hiệu suất khi vượt quá phần cứng đồng thời tối đa do chuyển ngữ cảnh, nhưng nếu bộ nhớ của tôi phục vụ chính xác thời gian chạy C++ và lên lịch std::async một cách thích hợp (tôi tin rằng trong trường hợp của Microsoft thư viện ConcRT của họ chịu trách nhiệm về điều này?), vì vậy mã trên có thể vẫn dẫn đến cải thiện hiệu suất?

Cảm ơn trước!

Answer 1

cáC++ runtime C là vụ phải theo dõi số lượng đề tạo và tiến độ std :: async một cách thích hợp

số Nếu các nhiệm vụ không đồng bộ là trên thực tế chạy không đồng bộ (chứ không phải hoãn lại), sau đó tất cả những gì yêu cầu là chúng được chạy như thể trên một chủ đề mới. Nó là hoàn toàn hợp lệ cho một chủ đề mới được tạo ra và bắt đầu cho mỗi nhiệm vụ, mà không có bất kỳ liên quan đến khả năng hạn chế của phần cứng cho song song.

Có một lưu ý:

[Ghi chú: Nếu chính sách này được xác định cùng với các chính sách khác, chẳng hạn như khi sử dụng một giá trị chủ trương ra mắt :: async | triển khai :: hoãn lại, triển khai nên trì hoãn yêu cầu hoặc việc lựa chọn chính sách khi không thể đồng thời khai thác hiệu quả hơn nữa. -end note]

Tuy nhiên, đây là không quy chuẩn và trong mọi trường hợp nó chỉ ra rằng một khi không có đồng thời nhiều hơn có thể khai thác các nhiệm vụ có thể bị hoãn lại, và do đó được thực hiện khi có người chờ đợi vào kết quả, thay vì vẫn không đồng bộ và chạy ngay lập tức sau khi một trong các tác vụ không đồng bộ trước đó được hoàn thành, như mong muốn cho tính song song tối đa.

Tức là, nếu chúng ta có 10 nhiệm vụ chạy dài và việc triển khai chỉ có thể thực hiện 4 song song, thì 4 đầu tiên sẽ không đồng bộ và sau đó 6 lần cuối có thể được trì hoãn. Việc chờ tương lai theo thứ tự sẽ thực hiện các nhiệm vụ trì hoãn trên một chuỗi đơn lẻ, loại bỏ việc thực thi song song cho các tác vụ đó.

Lưu ý cũng nói rằng thay vì trì hoãn yêu cầu, việc lựa chọn chính sách có thể được hoãn lại. Nghĩa là, hàm vẫn có thể chạy không đồng bộ nhưng quyết định đó có thể bị trì hoãn, giả sử, cho đến khi một trong các tác vụ trước đó hoàn tất, giải phóng cốt lõi cho một nhiệm vụ mới.Nhưng một lần nữa, điều này là không cần thiết, lưu ý là không quy phạm, và theo như tôi biết thực hiện của Microsoft là người duy nhất cư xử theo cách này. Khi tôi xem xét một triển khai khác, libC++, nó chỉ đơn giản bỏ qua lưu ý này để sử dụng chính sách std::launch::async hoặc std::launch::any dẫn đến việc thực thi không đồng bộ trên một chuỗi mới.

(Tôi tin rằng trong trường hợp của Microsoft thư viện ConcRT họ chịu trách nhiệm cho việc này?)

thực hiện của Microsoft không thực sự cư xử như bạn mô tả, tuy nhiên điều này là không cần thiết và một chương trình di động không thể dựa vào hành vi đó .

Một cách để portably hạn chế bao nhiêu chủ đề đang thực sự chạy là sử dụng một cái gì đó giống như một semaphore:

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