Küçük görevler için performans std :: async 'yi defalarca kullanabilir misiniz?

Question

Bazı arka plan bilgileri vermek için, kaydedilmiş bir dosyayı işliyorum ve dosyayı bileşen nesnelerine bölmek için normal bir ifade kullandıktan sonra, nesnenin verilerini hangi nesne türüne göre işleme koymam gerekiyor.

Şu andaki düşüncem, her nesnenin birbirinden bağımsız yüklenmesi gibi, bir miktar performans kazancı elde etmek için paralellik kullanmaktır. uygulamada 2'den fazla tipleri olacağına

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

Not olun: Bu yüzden ben şöyle std::async çağırarak sonra işleme ve olacağım her tür nesne için bir std::string kabul eden bir LoadObject işlevi tanımlamak için gidiyordu (bu sadece kısa bir örnektir) ve dosyadaki binlerce obje okunabilir.

Çok fazla iş parçacığı oluşturmanın, içerik anahtarlarından dolayı maksimum donanım eşzamanlılığını aştığında performans açısından genellikle kötü bir şey olduğunun farkındayım, ancak belleğimin doğru şekilde sunulması durumunda C++ çalışma zamanı oluşturulan iş parçacıklarının sayısını izlemelidir. ve std::async'u uygun bir şekilde programlayın (Microsoft'un durumunda ConcRT kitaplıklarının bundan sorumlu olduğuna inanıyorum), bu nedenle yukarıdaki kod hala bir performans iyileştirmesiyle sonuçlanabilir?

Şimdiden teşekkürler!

Answer 1

C++ çalışma zamanı oluşturulan evrelerin ve zamanlama std sayısını izlemek gerekiyordu

:: uygun

sayılı asenkron görevler aslında ise uyumsuz (ertelenmiş yerine) sonra tüm işte çalıştırmak async gerekli olan yeni bir iş parçacığı gibi çalışırlar. Donanımın sınırlı paralellik kapasitesi dikkate alınmaksızın, her görev için oluşturulacak ve başlatılacak yeni bir iş parçacığı için mükemmel bir şekilde geçerlidir. | Bu politika böyle lansman bir politika değerini kullanırken diğer politikaları ile birlikte belirtilirse :: async:

[Not:

bir not var başlatma :: ertelenmiş, uygulamaları, eşzamanlılık etkin bir şekilde kullanılamaz hale geldiğinde, çağrıyı veya ilke seçimini ertelemelidir. -end not]

Ancak bu normatif olmayan ve herhangi bir durumda, artık eşzamanlılık istismar edilebilir bir kez görevleri dolayısıyla birisi sonuca bekler zaman idam ertelenmiş ve hale olabileceğini gösterir , En yüksek eşzamanlılık için istenirse, önceki asenkron görevlerden biri bittikten hemen sonra eşzamanlı olmayan ve hemen çalışmaya devam etmek yerine.

Yani, 10 tane uzun çalışma görevimiz varsa ve uygulama yalnızca 4'ü paralel olarak yürütürse, ilk 4 eşzamansız olur ve son 6 değeri ertelenebilir. Sırasındaki vadeli işlemlerin beklenmesi, ertelenmiş görevleri sırayla tek bir iş parçacığı üzerinde yürütecek ve bu görevler için paralel yürütmeyi ortadan kaldıracaktır. Not, aynı zamanda, çağırma işlemini ertelemek yerine, politikanın seçiminin ertelenebileceğini de söylemektedir. Yani, işlev hala uyumsuz olarak çalışabilir, ancak daha önceki görevlerden biri tamamlanana kadar bu kararın ertelenmesi, yeni bir görev için bir özün serbest bırakılması olabilir.Ama yine de, bu gerekli değildir, not, normatif değildir ve bildiğim kadarıyla Microsoft'un bu şekilde davranan tek uygulamasıdır. Başka bir uygulamaya baktığımda, libC++, bu notu tamamen yok sayar, böylece std::launch::async veya std::launch::any ilkelerini kullanarak yeni bir iş parçacığı üzerinde zaman uyumsuz yürütme sağlar. (Ben Microsoft'un durumda onların ConcRT kütüphane Bunun sorumlusu olduğuna inanıyoruz?)

sen ancak bu gerekli değildir tanımlamak ve taşınabilir bir program o davranışa güvenemez olarak Microsoft'un uygulama gerçekten davranıyor

.

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