devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

std::future, std::promise, std::packaged_task, std::async_std::promise和std::future和std::task

作者：笔触狂放9 | 2024-08-03 09:29:03

赞

踩

std::promise和std::future和std::task

std::promise 进程间通信，std::packaged_task 任务封装，std::async 任务异步执行；std::future 获取结果。

1 std::promise

1.1 线程间同步

std::promise 可以用于线程间通信。

如下代码是 std::promise 中的示例代码。

std::promise - cppreference.com

（1）accumulate_promise 用于在线程间传递数据

（2）barrier 是 void 类型的，可以单纯地做线程之间的等待与唤醒

（3）std::fututre 是一个基本的组成元素，从名字也可以看出来，future 代表未来的结果。std::promise 只能使用一次，promise set_value 只能调用一次；std::future 调用 get() 也只能调用一次，如果再次调用 get()，那么会抛异常，在调用 get() 之前可以使用 valid() 来做判断，如果 valid() 是 true 的话，那么就可以调用 get()；否则不能调用 get()。

  1. #include <chrono>
  2. #include <future>
  3. #include <iostream>
  4. #include <numeric>
  5. #include <thread>
  6. #include <vector>
  7. #include <unistd.h>
  8.  
  9. void accumulate(std::vector<int>::iterator first,
  10.                 std::vector<int>::iterator last,
  11.                 std::promise<int> accumulate_promise)
  12. {
  13.     int sum = std::accumulate(first, last, 0);
  14.     accumulate_promise.set_value(sum); // Notify future
  15. }
  16.  
  17. void do_work(std::promise<void> barrier)
  18. {
  19.     std::cout << "before sleep\n";
  20.     std::this_thread::sleep_for(std::chrono::seconds(1));
  21.     barrier.set_value();
  22.     std::cout << "after promise set value\n";
  23. }
  24.  
  25. int main()
  26. {
  27.     // Demonstrate using promise<int> to transmit a result between threads.
  28.     std::vector<int> numbers = {1, 2, 3, 4, 5, 6};
  29.     std::promise<int> accumulate_promise;
  30.     std::future<int> accumulate_future = accumulate_promise.get_future();
  31.     std::thread work_thread(accumulate, numbers.begin(), numbers.end(),
  32.                             std::move(accumulate_promise));
  33.  
  34.     // future::get() will wait until the future has a valid result and retrieves it.
  35.     // Calling wait() before get() is not needed
  36.     // accumulate_future.wait(); // wait for result
  37.     std::cout << "before get, future valid: " << accumulate_future.valid() << std::endl;
  38.     std::cout << "result=" << accumulate_future.get() << '\n';
  39.     std::cout << "after get, future valid: " << accumulate_future.valid() << std::endl;
  40.     work_thread.join(); // wait for thread completion
  41.  
  42.     // Demonstrate using promise<void> to signal state between threads.
  43.     std::promise<void> barrier;
  44.     std::future<void> barrier_future = barrier.get_future();
  45.     std::thread new_work_thread(do_work, std::move(barrier));
  46.  
  47.     std::cout << "before wait\n";
  48.     barrier_future.wait();
  49.     std::cout << "after wait\n";
  50.     new_work_thread.join();
  51.     return 0;
  52. }

1.2 线程间同步 —— 条件变量

条件变量是一个基础的线程间同步机制，在 c 和 c++ 中都有使用。条件变量经常用于生产者线程和消费者线程之间通信的场景。

1.2.1 std::condition_variable

如下是使用条件变量时经常使用的方式，第一个形参是 std::unique_lock<> 类型的锁，第二个形参是 wait() 返回的条件。

template< class Predicate >
void wait( std::unique_lock<std::mutex>& lock, Predicate pred );

（1）条件满足之后，即使不进行 notify，wait() 也会返回

（2）条件不满足的时候，即使进行 notify()，wait() 也不会返回

这就是使用条件变量的时候最典型的用法，因为条件变量可能存在惊群问题(本人没有复现过)，可能被唤醒的时候，条件还没有满足。这种用法就类似于在 c 语言使用条件变量的时候的如下代码。

while(!condition) {

  wait();

}

  1. #include <condition_variable>
  2. #include <iostream>
  3. #include <mutex>
  4. #include <string>
  5. #include <thread>
  6. #include <unistd.h>
  7.  
  8. std::mutex m;
  9. std::condition_variable cv;
  10. std::string data;
  11. bool ready = false;
  12. bool processed = false;
  13.  
  14. void worker_thread()
  15. {
  16.     // wait until main() sends data
  17.     std::unique_lock<std::mutex> lk(m);
  18.     cv.wait(lk, []{ return ready; });
  19.  
  20.     // after the wait, we own the lock
  21.     std::cout << "Worker thread is processing data\n";
  22.     data += " after processing";
  23.  
  24.     // send data back to main()
  25.     processed = true;
  26.     std::cout << "Worker thread signals data processing completed\n";
  27.  
  28.     // manual unlocking is done before notifying, to avoid waking up
  29.     // the waiting thread only to block again (see notify_one for details)
  30.     lk.unlock();
  31.     cv.notify_one();
  32. }
  33.  
  34. int main()
  35. {
  36.     std::thread worker(worker_thread);
  37.  
  38.     data = "Example data";
  39.     // send data to the worker thread
  40.     {
  41.         std::lock_guard<std::mutex> lk(m);
  42.         sleep(2);
  43.         std::cout << "after sleep\n";
  44.         ready = true;
  45.         std::cout << "after set value\n";
  46.         std::cout << "main() signals data ready for processing\n";
  47.     }
  48.  
  49.     std::cout << "before notify one\n";
  50.     sleep(2);
  51.     cv.notify_one();
  52.     std::cout << "after notify one\n";
  53.     // wait for the worker
  54.     {
  55.         std::unique_lock<std::mutex> lk(m);
  56.         cv.wait(lk, []{ return processed; });
  57.     }
  58.     std::cout << "Back in main(), data = " << data << '\n';
  59.     worker.join();
  60. }

1.2.2 pthread_cond_t

pthread 中的条件变量，使用方式如下。也是需要一个 mutex 和条件变量在一块使用。

wait 端调用的函数：

        pthread_mutex_lock(&mutex);
        while (!condition) {
            pthread_cond_wait(&cond, &mutex);
        }
        pthread_mutex_unlock(&mutex);

唤醒端需要调用的函数：

        pthread_mutex_lock(&mutex);
        pthread_cond_signal(&cond);
        pthread_mutex_unlock(&mutex);

在 wait 端，首先要加锁，然后进行 wait；在 wait 的时候，会释放锁，所以在 signal 端加锁是可以加成功的。

  1. #include <stdio.h>
  2. #include <pthread.h>
  3.  
  4. #define MAX_COUNT 10
  5.  
  6. int shared_resource = 0; // 全局共享资源
  7. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; // 互斥锁
  8. pthread_cond_t cond = PTHREAD_COND_INITIALIZER; // 条件变量
  9.  
  10. void *producer(void *arg) {
  11.     while (1) {
  12.         pthread_mutex_lock(&mutex);
  13.         while (shared_resource >= MAX_COUNT) { // 当共享资源达到最大值时等待
  14.             pthread_cond_wait(&cond, &mutex);
  15.         }
  16.         shared_resource++; // 增加共享资源值
  17.         printf("Produced: %d\n", shared_resource);
  18.         pthread_cond_signal(&cond); // 唤醒消费者线程
  19.         pthread_mutex_unlock(&mutex);
  20.         sleep(1); // 模拟生产过程
  21.     }
  22.     return NULL;
  23. }
  24.  
  25. void *consumer(void *arg) {
  26.     while (1) {
  27.         pthread_mutex_lock(&mutex);
  28.         while (shared_resource <= 0) { // 当共享资源为0时等待
  29.             pthread_cond_wait(&cond, &mutex);
  30.         }
  31.         shared_resource--; // 减少共享资源值
  32.         printf("Consumed: %d\n", shared_resource);
  33.         pthread_cond_signal(&cond); // 唤醒生产者线程
  34.         pthread_mutex_unlock(&mutex);
  35.         sleep(1); // 模拟消费过程
  36.     }
  37.     return NULL;
  38. }
  39.  
  40. int main() {
  41.     pthread_t producer_thread, consumer_thread;
  42.     // 创建生产者线程和消费者线程
  43.     pthread_create(&producer_thread, NULL, producer, NULL);
  44.     pthread_create(&consumer_thread, NULL, consumer, NULL);
  45.     // 等待线程结束
  46.     pthread_join(producer_thread, NULL);
  47.     pthread_join(consumer_thread, NULL);
  48.     // 销毁互斥锁和条件变量
  49.     pthread_mutex_destroy(&mutex);
  50.     pthread_cond_destroy(&cond);
  51.     return 0;
  52. }

在使用条件变量的时候要注意，当条件变量被 wait 的时候，对条件变量进行 destroy，那么 destroy 会阻塞住，直到 wait 返回之后，destroy 才会返回。如下代码是 destroy 一个正在被 wait 的条件变量，这个时候 destroy 会阻塞住。

  1. #include <stdio.h>
  2. #include <pthread.h>
  3. #include <unistd.h>
  4.  
  5. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  6. pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
  7.  
  8. void *consumer(void *arg) {
  9.     pthread_mutex_lock(&mutex);
  10.     pthread_cond_wait(&cond, &mutex);
  11.     pthread_mutex_unlock(&mutex);
  12.     return NULL;
  13. }
  14.  
  15. int main() {
  16.     pthread_t consumer_thread;
  17.     pthread_create(&consumer_thread, NULL, consumer, NULL);
  18.     sleep(1);
  19.  
  20.     pthread_mutex_destroy(&mutex);
  21.     printf("before destroy condition\n");
  22.     pthread_cond_destroy(&cond);
  23.     printf("after destroy condition\n");
  24.     return 0;
  25. }

2 std::packaged_task

std::package_task，从名字也可以看出来，表示一个打包的任务。这个任务自己并不能执行，而是需要显式的调用来执行。

如下是官网的示例代码。

std::packaged_task - cppreference.com

（1）task 能够封装的任务类型包括 lambda 表达式，bind 的函数，也可以是一个单纯的函数

（2）与 std::promise 类似，std::packaged_task 也可以获取一个 std::future，std::future 可以获取到 std::package_task 执行的结果

  1. #include <cmath>
  2. #include <functional>
  3. #include <future>
  4. #include <iostream>
  5. #include <thread>
  6.  
  7. // unique function to avoid disambiguating the std::pow overload set
  8. int f(int x, int y) { return std::pow(x, y); }
  9.  
  10. void task_lambda()
  11. {
  12.     std::packaged_task<int(int, int)> task([](int a, int b)
  13.     {
  14.         return std::pow(a, b);
  15.     });
  16.     std::future<int> result = task.get_future();
  17.  
  18.     // 封装一个 lamda 表达式，可以将 task 当成函数名来直接调用
  19.     task(2, 9);
  20.  
  21.     std::cout << "task_lambda:\t" << result.get() << '\n';
  22. }
  23.  
  24. void task_bind()
  25. {
  26.     std::packaged_task<int()> task(std::bind(f, 2, 11));
  27.     std::future<int> result = task.get_future();
  28.  
  29.     // 封装的 bind() 函数，可以直接调用
  30.     task();
  31.  
  32.     std::cout << "task_bind:\t" << result.get() << '\n';
  33. }
  34.  
  35. void task_thread()
  36. {
  37.     std::packaged_task<int(int, int)> task(f);
  38.     std::future<int> result = task.get_future();
  39.     // 非 bind 方式，不能这样直接调用
  40.     // std::cout << "directly call: " << task(2, 10) << std::endl;
  41.     std::thread task_td(std::move(task), 2, 10);
  42.     task_td.join();
  43.  
  44.     std::cout << "task_thread:\t" << result.get() << '\n';
  45. }
  46.  
  47. int main()
  48. {
  49.     task_lambda();
  50.     task_bind();
  51.     task_thread();
  52.     return 0;
  53. }

2.1 异常传递

std::packaged_task中抛出的异常，保存在 std::future 中，可以通过 get() 来获取。

  1. #include <cmath>
  2. #include <functional>
  3. #include <future>
  4. #include <iostream>
  5. #include <thread>
  6.  
  7. int f(int x, int y) {
  8.   throw std::runtime_error("An error occurred in the task!");
  9.   return std::pow(x, y);
  10. }
  11.  
  12. void task_lambda()
  13. {
  14.     std::packaged_task<int(int, int)> task([](int a, int b)
  15.     {
  16.         throw std::runtime_error("An error occurred in the task!");
  17.         return std::pow(a, b);
  18.     });
  19.     std::future<int> result = task.get_future();
  20.  
  21.     task(2, 9);
  22.  
  23.     try {
  24.       std::cout << "task_lambda:\t" << result.get() << '\n';
  25.     } catch (std::exception &e) {
  26.       std::cout << "lambda exception: " << e.what() << std::endl;
  27.     }
  28. }
  29.  
  30. void task_bind()
  31. {
  32.     std::packaged_task<int()> task(std::bind(f, 2, 11));
  33.     std::future<int> result = task.get_future();
  34.  
  35.     task();
  36.  
  37.     try {
  38.       std::cout << "task_bind:\t" << result.get() << '\n';
  39.     } catch (std::exception &e) {
  40.       std::cout << "bind exception: " << e.what() << std::endl;
  41.     }
  42. }
  43.  
  44. void task_thread()
  45. {
  46.     std::packaged_task<int(int, int)> task(f);
  47.     std::future<int> result = task.get_future();
  48.     std::thread task_td(std::move(task), 2, 10);
  49.     task_td.join();
  50.  
  51.     try {
  52.       std::cout << "task_thread:\t" << result.get() << '\n';
  53.     } catch (std::exception &e) {
  54.       std::cout << "task thread exception: " << e.what() << std::endl;
  55.     }
  56. }
  57.  
  58. int main()
  59. {
  60.     task_lambda();
  61.     task_bind();
  62.     task_thread();
  63.     return 0;
  64. }

2.2 void 类型的 std::future

valid 是 true，也可以 get，不过 get 之后也是 void。

  1. #include <cmath>
  2. #include <functional>
  3. #include <future>
  4. #include <iostream>
  5. #include <thread>
  6.  
  7. void f(int x, int y) {
  8.   std::cout << "x " << x << ", y " << y << std::endl;
  9. }
  10.  
  11. void task_lambda()
  12. {
  13.     std::packaged_task<void(int, int)> task([](int a, int b)
  14.     {
  15.       std::cout << "a " << a << ", b " << b << std::endl;
  16.     });
  17.     auto result = task.get_future();
  18.  
  19.     task(2, 9);
  20.  
  21.     try {
  22.       std::cout << "task_lambda:\t" << result.valid() << '\n';
  23.       result.get();
  24.     } catch (std::exception &e) {
  25.       std::cout << "lambda exception: " << e.what() << std::endl;
  26.     }
  27. }
  28.  
  29. void task_bind()
  30. {
  31.     std::packaged_task<void()> task(std::bind(f, 2, 11));
  32.     std::future<void> result = task.get_future();
  33.  
  34.     task();
  35.  
  36.     try {
  37.       std::cout << "task_bind:\t" << result.valid() << '\n';
  38.       result.get();
  39.     } catch (std::exception &e) {
  40.       std::cout << "bind exception: " << e.what() << std::endl;
  41.     }
  42. }
  43.  
  44. void task_thread()
  45. {
  46.     std::packaged_task<void(int, int)> task(f);
  47.     std::future<void> result = task.get_future();
  48.     std::thread task_td(std::move(task), 2, 10);
  49.     task_td.join();
  50.  
  51.     try {
  52.       std::cout << "task_thread:\t" << result.valid() << '\n';
  53.       result.get();
  54.     } catch (std::exception &e) {
  55.       std::cout << "task thread exception: " << e.what() << std::endl;
  56.     }
  57. }
  58.  
  59. int main()
  60. {
  61.     task_lambda();
  62.     task_bind();
  63.     task_thread();
  64.     return 0;
  65. }

3 std::async

std::async 提供了异步执行的方式。上一节的 std::packaged_task 只是封装了一个 task，但是这个 task 自己不会执行，std::async 相当于在 std::packaged_task 的基础上增加了自动执行的机制。

std::async 的第一个形参有两个可选值：std::launch::async 和 std::launch::deferred，第一个标志会自动异步执行；第二个标志，不会自动异步执行，需要调 wait() 同步执行。

  1. #include <algorithm>
  2. #include <future>
  3. #include <iostream>
  4. #include <mutex>
  5. #include <numeric>
  6. #include <string>
  7. #include <vector>
  8.  
  9. std::mutex m;
  10.  
  11. struct X
  12. {
  13.     void foo(int i, const std::string& str)
  14.     {
  15.         std::lock_guard<std::mutex> lk(m);
  16.         std::cout << str << ' ' << i << '\n';
  17.     }
  18.  
  19.     void bar(const std::string& str)
  20.     {
  21.         std::lock_guard<std::mutex> lk(m);
  22.         std::cout << str << '\n';
  23.     }
  24.  
  25.     int operator()(int i)
  26.     {
  27.         std::lock_guard<std::mutex> lk(m);
  28.         std::cout << i << '\n';
  29.         return i + 10;
  30.     }
  31. };
  32.  
  33. template<typename RandomIt>
  34. int parallel_sum(RandomIt beg, RandomIt end)
  35. {
  36.     auto len = end - beg;
  37.     if (len < 1000)
  38.         return std::accumulate(beg, end, 0);
  39.  
  40.     RandomIt mid = beg + len / 2;
  41.     auto handle = std::async(std::launch::async,
  42.                              parallel_sum<RandomIt>, mid, end);
  43.     int sum = parallel_sum(beg, mid);
  44.     return sum + handle.get();
  45. }
  46.  
  47. int main()
  48. {
  49.     std::vector<int> v(10000, 1);
  50.     std::cout << "The sum is " << parallel_sum(v.begin(), v.end()) << '\n';
  51.  
  52.     X x;
  53.     // Calls (&x)->foo(42, "Hello") with default policy:
  54.     // may print "Hello 42" concurrently or defer execution
  55.     auto a1 = std::async(&X::foo, &x, 42, "Hello");
  56.     // Calls x.bar("world!") with deferred policy
  57.     // prints "world!" when a2.get() or a2.wait() is called
  58.     auto a2 = std::async(std::launch::deferred, &X::bar, x, "world!");
  59.     // Calls X()(43); with async policy
  60.     // prints "43" concurrently
  61.     auto a3 = std::async(std::launch::async, X(), 43);
  62.     a2.wait();                     // prints "world!"
  63.     std::cout << a3.get() << '\n'; // prints "53"
  64. } // if a1 is not done at this point, destructor of a1 prints "Hello 42" here

3.1 使用 std::sync 注意事项

3.1.1 要获取 std::async 的返回值

如下代码，并没有获取 std::async 的返回值，这样的话两个任务是串行执行的，而不是异步并行执行的。

  1. #include <algorithm>
  2. #include <future>
  3. #include <iostream>
  4. #include <mutex>
  5. #include <numeric>
  6. #include <string>
  7. #include <vector>
  8. #include <unistd.h>
  9.  
  10. std::mutex m;
  11.  
  12. struct X
  13. {
  14.     void foo()
  15.     {
  16.         for (int i = 0; i < 10; i++) {
  17.           std::cout << "foo()" << std::endl;
  18.           sleep(1);
  19.         }
  20.     }
  21.  
  22.     void bar()
  23.     {
  24.         for (int i = 0; i < 10; i++) {
  25.           std::cout << "bar()" << std::endl;
  26.           sleep(1);
  27.         }
  28.     }
  29. };
  30.  
  31. int main()
  32. {
  33.     X x;
  34.     std::async(std::launch::async, &X::foo, &x);
  35.     std::async(std::launch::async, &X::bar, &x);
  36.     return 0;
  37. }

官方解释：

3.1.2 异常传递

使用 std::async 的时候，如果任务中抛异常，异常会保存在 std::future 中，可以通过 std::future 来获取。

  1. #include <algorithm>
  2. #include <future>
  3. #include <iostream>
  4. #include <mutex>
  5. #include <numeric>
  6. #include <string>
  7. #include <vector>
  8. #include <unistd.h>
  9.  
  10. std::mutex m;
  11.  
  12. struct X
  13. {
  14.     int bar()
  15.     {
  16.         throw std::runtime_error("An error occurred in the task!");
  17.         return 10;
  18.     }
  19. };
  20.  
  21. int main()
  22. {
  23.     X x;
  24.     auto f1 = std::async(std::launch::async, &X::bar, &x);
  25.  
  26.     try {
  27.       f1.get();
  28.     } catch (std::exception &e) {
  29.       std::cout << "f1 get exception: " << e.what() << std::endl;
  30.     }
  31.     return 0;
  32. }
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/922428
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号