使用C++中的futures，異步和線程實現搜索11

Question

我想以多線程方式實現分支和邊界搜索。特別是，我想使用async在每個分支中包裝搜索調用，然後等待某個線程出現答案，然後退出。 （理想情況下，我想取消其他線程，但線程取消不在標準中）。這裏有一些簡化代碼：

#include <iostream> 
#include <random> 
#include <future> 
#include <thread> 

using namespace std; 

mt19937 rng; 
uniform_int_distribution<unsigned> random_binary(0, 1); 

bool search() { 
    return static_cast<bool>(random_binary(rng)); 
} 

#define N 10000 

int main() 
{ 
    rng.seed(42); 

    std::vector<future<bool>> tasks; 

    for (unsigned i=0; i<N; ++i) 
    tasks.push_back(async(launch::async, search)); 

    // Don't want to wait sequentially here. 
    for (unsigned i=0; i<N; ++i) { 
    tasks[i].wait(); 
    if (tasks[i].get()) { 
     cout << "i = " << i << "\n"; 
     break; 
    } 
    } 
    return 0; 
} 

search()是搜索功能。它根據是否找到答案返回true/false。爲了說明，我返回一個隨機答案。但問題的關鍵在於調用tasks[i].wait()的for循環。現在，我正在順序地等待任務完成。相反，我想要做這樣的事情：

auto x = wait_for_any(tasks.begin(), tasks.end()); 
x.get(); 
// cancel other threads. 
// Profit? 

什麼是實現這一目標的好方法？

Answer 1

std::future提供了一個valid()函數，可以讓您檢查結果是否可用而不會阻塞，因此您可以使用該函數，例如，在繁忙的等待循環：

std::future<bool>* res_future = 0; 
for(size_t i = 0; ; i==tasks.size()?i=0:++i){ 
    // could add a timeout period to not completely busy-wait the CPU 
    if(tasks[i].wait_for(std::chrono::seconds(0)) == std::future_status::ready){ 
    res = &tasks[i]; 
    break; 
    } 
} 

bool res = res_future->get(); 

擬議除了std::future，做出這樣的任務更容易，是一個.then(func_obj)方法異步調用func_obj當結果是可用的，在那裏你可以設置標誌或東西。

我遺憾地不知道有一種方法可能以上述以外的其他方式實施wait_for_any。 ：/

template<class FwdIt> 
std::future<bool> wait_for_any(FwdIt first, FwdIt last) 
{ 
    return std::async([=]{ 
    for(FwdIt cur(first); ; cur==last?cur=first:++cur){ 
    // could add a timeout period to not completely busy-wait the CPU 
    if(cur->wait_for(std::chrono::seconds(0)) == std::future_status::ready) 
     return cur->get(); 
    }); 
} 

線程破壞通常通過協作取消來完成。

體育：std::future<T>::get()將自動wait()如果結果不可用。

Answer 2

安排所有任務訪問相同的condition_variable,mutex和bool。這可以通過使這些全局變量或成員數據在每個任務運行成員函數的情況下完成，或者可以通過std::ref作爲參數在任務創建函數中傳遞它們。

在開始任何任務之前將bool初始化爲not_found。主線程然後啓動任務並等待condition_variable。搜索者任務然後搜索。當他們搜索時，他們偶爾會檢查bool（可能是原子負載），看看它是否已設置爲found。如果有，搜索器線程立即返回。

當一個線程找到結果時，它將bool設置爲found並指示condition_variable。這將喚醒主線程並有效地取消其餘的搜索任務。然後主線程可以加入，分離，放棄任何搜索任務。如果您沒有主要顯式加入搜索任務，那麼最好是如果您可以安排所有搜索任務在主要退出之前結束。

沒有輪詢。沒有等待死衚衕的搜索。唯一特別的部分是弄清楚搜索者任務如何以及經常檢查bool。我建議性能測試這部分。

Answer 3

我會採取的方法是讓主函數創建一個std::promise，然後將其作爲參考進行工作的各個線程。然後，每個工作線程將使用他們共享的std::promise在他們得到結果後使用set_value()。

哪個工作線程首先到達那裏將能夠發送結果，其他人將在嘗試使用set_value()時得到異常，並且它已被設置。

在下面的源代碼框架中，我使用的是std::async，但如果我們可以在不創建額外的std::promise而不用任何東西的情況下剝離自治線程，那將會很好。

因此代碼框架看起來是這樣的：

#include <iostream> 
#include <thread> // std::thread is defined here 
#include <future> // std::future and std::promise defined here 

bool search(int args, std::promise<int> &p) { 

    // do whatever needs to be done 

    try { 
     p.set_value(args); 
    } 
    catch (std::future_error & e) { 
     // std::future_errc::promise_already_satisfied 
    } 
    return true; 
} 
int main(int argc, char * argv[]) 
{ 
    std::promise<int> myProm;  
    auto fut = myProm.get_future(); 

    for (int i = 1; i < 4; i++) { 
     std::async(std::launch::async, search, i, std::ref(myProm)); 
    } 

    auto retVal = fut.get(); 
    return 0; 
}