std :: future возвращается из std :: async зависает при выходе из сферы действия

Question

Я использую комбинацию std::async и std::future от C++ 11. Я использую для принудительного выполнения time_out для определенного действия, которое я делаю в своем коде, который может использовать время, когда я пытаюсь подключиться к серверу.

Ниже, как код:

#include <future> 
#include <chrono> 

std::size_t PotentiallyLongRunningActivity() { 
    using namespace std::chrono_literals; 
    std::this_thread::sleep_for(10000s); 
    return 10; 
} 

bool DoActivity() { 

    bool activity_done = false; 
    auto my_future_result(std::async(std::launch::async, []() { 
     return PotentiallyLongRunningActivity(); //returns size_t 
    })); 

    std::future_status my_future_status = my_future_result.wait_for(std::chrono::milliseconds(800)); 
    if (my_future_status == std::future_status::timeout) { 
     activity_done = false; 
    } 
    else if (my_future_status == std::future_status::ready) { 
     if (my_future_result.valid() && my_future_result.get() > 0) { 
      activity_done = true; 
     } 
    } 

    return activity_done; 
    //my_future_result hangs while exiting this method !!! 
} 

int main(int argc, char *argv[]) 
{ 
    DoActivity(); 
    return 0; 
} 

вещь прекрасно работает в большинстве случаев. Ожидается, что будущие тайм-ауты & готовятся во многих случаях. Но, странное поведение, которое я наблюдаю, заключается в том, что в некоторых случаях пользовательский интерфейс зависает, потому что my_future_result при выходе из области зависает. Я подтвердил это, повторив вызов my_future_result.get(), который никогда не возвращается, если вызван непосредственно перед выходом из метода.

Как я могу обойти это? Есть ли способ отменить или удалить или прекратить действие std::future?

Answer 1

Забрав из cppreference sample, только «начало», «f2 закончена» и «конец» и будет напечатан из этого кода (потому что f1 не «висят»):

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

int main() { 
    using namespace std::literals; 

    { 
     std::packaged_task<int()> task([]() { 
      std::this_thread::sleep_for(5s); 
      std::cout << "f1 finished" << std::endl; 
      return 42; 
     }); 
     std::future<int> f1 = task.get_future(); 
     std::thread(std::move(task)).detach(); 

     std::future<int> f2 = std::async(std::launch::async, []() { 
      std::this_thread::sleep_for(3s); 
      std::cout << "f2 finished" << std::endl; 
      return 42; 
     }); 

     f1.wait_for(1s); 
     f2.wait_for(1s); 
     std::cout << "the start" << std::endl; 
    } 

    // std::this_thread::sleep_for(7s); 
    std::cout << "the end" << std::endl; 
} 

Для хорошей дискуссии см.: http://scottmeyers.blogspot.com.br/2013/03/stdfutures-from-stdasync-arent-special.html.

Стандартная библиотека C++ не поддерживает операции по уничтожению нитей.

Позаботься о ниточках detach. Отряд как таковой не является «очень плохим», он может быть полезен, например, для пользователей, которые могут быть завершены, или если у вас есть другая идея оркестровки и срыва. В противном случае стандартная библиотека не должна указывать detach.

Answer 2

Вы покидаете свою функцию до завершения задания std::async. При определенных обстоятельствах деструктор для std::future будет блокироваться до завершения задачи.

http://en.cppreference.com/w/cpp/thread/future/wait_for

Вот в документации для wait_for пример показывает несколько вызовов wait_for после тайм-аута, показывая, что акт тайм-аута не отменяет std::async задачу.

Нет встроенной поддержки (которую я мог обнаружить), которая позволяет убивать потоки извне. Это имеет смысл, поскольку нет способа правильно очистить состояние системных ресурсов, используемых в потоке, если оно завершено таким образом.

Вместо этого лучше поместить логику тайм-аута в сам поток, чтобы он мог закончить себя и правильно очистить.

Answer 3

Как правило, потерять трек нити очень плоха. Наличие кода, выполняющегося в другом потоке, когда выходы main - это рецепт неопределенного поведения.

Таким образом, std::future, который возвращается от std::async, имеет специальное свойство, которое будет ждать завершения std::async, когда оно будет уничтожено.

Это то, что вы описываете как «зависание». Это не зависание - он ждет завершения задачи.

Первичные примитивы на C++ 11 являются примитивами; они не являются готовыми типами для полнофункционального приложения. Вы можете использовать их для написания очередей очередей очередей потоков и т. П .; использование их наивно «в сыром», как правило, приводит к смещению их к правильности, но не дает вам того, чего вы хотите.

Простой пул потоков просто:

template<class T> 
struct threaded_queue { 
    using lock = std::unique_lock<std::mutex>; 
    void push_back(T t) { 
    { 
     lock l(m); 
     data.push_back(std::move(t)); 
    } 
    cv.notify_one(); 
    } 
    boost::optional<T> pop_front() { 
    lock l(m); 
    cv.wait(l, [this]{ return abort || !data.empty(); }); 
    if (abort) return {}; 
    auto r = std::move(data.back()); 
    data.pop_back(); 
    return std::move(r); 
    } 
    void terminate() { 
    { 
     lock l(m); 
     abort = true; 
     data.clear(); 
    } 
    cv.notify_all(); 
    } 
    ~threaded_queue() 
    { 
    terminate(); 
    } 
private: 
    std::mutex m; 
    std::deque<T> data; 
    std::condition_variable cv; 
    bool abort = false; 
}; 
struct thread_pool { 
    thread_pool(std::size_t n = 1) { start_thread(n); } 
    thread_pool(thread_pool&&) = delete; 
    thread_pool& operator=(thread_pool&&) = delete; 
    ~thread_pool() = default; // or `{ terminate(); }` if you want to abandon some tasks 
    template<class F, class R=std::result_of_t<F&()>> 
    std::future<R> queue_task(F task) { 
    std::packaged_task<R()> p(std::move(task)); 
    auto r = p.get_future(); 
    tasks.push_back(std::move(p)); 
    return r; 
    } 
    template<class F, class R=std::result_of_t<F&()>> 
    std::future<R> run_task(F task) { 
    if (threads_active() >= total_threads()) { 
     start_thread(); 
    } 
    return queue_task(std::move(task)); 
    } 
    void terminate() { 
    tasks.terminate(); 
    } 
    std::size_t threads_active() const { 
    return active; 
    } 
    std::size_t total_threads() const { 
    return threads.size(); 
    } 
    void clear_threads() { 
    terminate(); 
    threads.clear(); 
    } 
    void start_thread(std::size_t n = 1) { 
    while(n-->0) { 
     threads.push_back(
     std::async(std::launch::async, 
      [this]{ 
      while(auto task = tasks.pop_front()) { 
       ++active; 
       try{ 
       (*task)(); 
       } catch(...) { 
       --active; 
       throw; 
       } 
       --active; 
      } 
      } 
     ) 
    ); 
    } 
    } 
private: 
    std::vector<std::future<void>> threads; 
    threaded_queue<std::packaged_task<void()>> tasks; 
    std::atomic<std::size_t> active; 
}; 

Live example.

Теперь вы создаете какой-то пул потоков где-то, бросаете на него задачи, и вы можете подождать на интересующих фьючерсах. В пуле есть ограниченное количество потоков.

run_task обеспечит поток для выполнения любой задачи, которую вы ставите в очередь. queue_task будет использовать только существующие потоки, если они доступны.

Возвращенный std::future<void> не блокирует выполнение задачи; но деструктор объекта thread_pool делает.

Обратите внимание, что он прекратит выполнение всех задач в очереди и дождитесь завершения текущих задач запуска по умолчанию при уничтожении.

Что-то, что обертывает unique_ptr<thread_pool>, полезно для удобства передвижения. Перемещение должно быть отключено, так как активные потоки сохраняют указатель на this.

thread_pool не является потокобезопасным по иронии; это потому, что мы не охраняем std::vector<std::future<void>> threads;; Я имею в виду, кроме потоков, безопасных для самих нитей. Он предназначен для прямого доступа только к одной внешней нити.

queue_task и terminate is thread safe главным образом случайно.

Answer 4

Причина ошибки заключается в том, что компилятору не сообщили, что результат вашей функции DoModify() будет доступен асинхронно (следовательно, будет объявлен как std :: future <>), и он ожидал синхронного результата bool типа, который так не поступал. Вы можете использовать std :: future :: is_ready() или std :: future_status. Здесь пример фрагмента кода

std::future<size_t> DoActivity() 
{ 
     return std::async(std::launch::async, []() 
      { 
      return PotentiallyLongRunningActivity(); 
      }); 
} 

int main() 
{ 
    auto result = DoActivity(); 
    if (result. Is_ready()) 
    { 
     auto data = result.get(); 
     //do something with data 
    } 
}