并非所有 std::packaged_tasks 在内部 std::async 调用时执行

Question

我有一个相当复杂的代码，其中包含 std::async 个调用和 std::packaged_task，无法执行到最后。我将其简化为最小的可重现示例。

先后调用了两个异步函数，其中有packaged_tasks使用std::async异步执行。然后我们都等待两个异步函数完成，使用相应的 future.wait() 方法。执行在 futY.wait(); 处停止，第二个 packaged_task 永远不会执行（没有第二个 Inside handler func 日志）。

#include <iostream>     // std::cout
#include <future>       // std::packaged_task, std::future
#include <exception>
#include <vector>
#include <list>
#include <memory>
#include <functional>

std::list<std::function<bool(const std::vector<int> &data)>> handlers_;

std::future<std::vector<int>> countup(int from, int to) {
    std::function<std::vector<int>(std::exception_ptr, std::vector<int>)> func = [=] (std::exception_ptr ex, std::vector<int> data) {
        std::cout << "Inside handler func " << from << " " << to << std::endl;
        if (ex != nullptr) {
            std::rethrow_exception(ex);
        }
        return data;
    };
    auto packageP = std::make_shared<std::packaged_task<std::vector<int>(std::exception_ptr, std::vector<int>)>>(func);

    auto fut = packageP->get_future();
    handlers_.push_back([packageP] (const std::vector<int> &data) mutable -> bool {
            std::cout << "Calling handler with data, size: " << data.size() << std::endl;

            (*packageP)(nullptr, data);
            return data.size();
        }
    );

    auto fut2 = std::async(std::launch::async, [=, &handlers_] {
            std::cout << "Before handler called " << from << to << std::endl;

            std::vector<int> vec ( to, from );

            auto res = (*handlers_.begin())(vec);

            std::cout << "Handler result " << res << " for " << from << " " << to << std::endl;
        });

    std::cout << "Called async in countup for " << from << " " << to << std::endl;

    return fut;
}

int main ()
{
  auto futX = std::async(std::launch::async, [] {
      auto fut1 = std::async(std::launch::async, [] {
        auto fut2 = countup(0, 2);

        std::cout << "Called X countup and waiting to finish" << std::endl;

        fut2.wait();

        auto vec = fut2.get();
        std::cout << "The X countup returned" << std::endl;
      });
      std::cout << "Called X async internal and waiting to finish" << std::endl;
      fut1.wait();

      return 2;
  });

  std::cout << "Called async X and waiting to finish" << std::endl;

  auto futY = std::async(std::launch::async, [] {
      auto fut1 = std::async(std::launch::async, [] {
        auto fut2 = countup(0, 3);

        std::cout << "Called Y countup and waiting to finish" << std::endl;
        fut2.wait();

        auto vec = fut2.get();
        std::cout << "The Y countup returned " << std::endl;
      });
      std::cout << "Called Y async internal and waiting to finish" << std::endl;
      fut1.wait();

      return 3;
  });

  std::cout << "Called async Y and waiting to finish" << std::endl;

  futX.wait();
  std::cout << "After async X and waiting to finish" << std::endl;

  futY.wait();
  std::cout << "After async Y and waiting to finish" << std::endl;

  int valueX = futX.get();                  // wait for the task to finish and get result
  int valueY = futY.get();                  // wait for the task to finish and get result

  std::cout << "The countdown lasted for " << valueX  << " " << valueY << " seconds" << std::endl;

  return 0;
}

日志如下：

Called async X and waiting to finish
Called async Y and waiting to finish
Called X async internal and waiting to finish
Called Y async internal and waiting to finish
Called async in countup for Before handler called 02
0 2
Calling handler with data, size: 2
Inside handler func 0Called async in countup for  2
Handler result 01 for 0 2
 Before handler called 03
3Calling handler with data, size: 
Called X countup and waiting to finish
The X countup returned
3
After async X and waiting to finish
Called Y countup and waiting to finish

能否解释一下为什么代码执行到最后才执行？

Answer 1

您在调用 (*handlers_.begin()) 时正在重复使用相同的打包任务，我认为您在 handlers_.push_back.

时存在数据竞争

您不需要全局 handlers_，您可以只捕获一个本地处理程序。

#include <iostream>     // std::cout
#include <future>       // std::packaged_task, std::future
#include <exception>
#include <vector>
#include <list>
#include <memory>
#include <functional>

std::future<std::vector<int>> countup(int from, int to) {
    auto func = [=] (std::exception_ptr ex, std::vector<int> data) {
        std::cout << "Inside handler func " << from << " " << to << std::endl;
        if (ex != nullptr) {
            std::rethrow_exception(ex);
        }
        return data;
    };
    auto packageP = std::make_shared<std::packaged_task<std::vector<int>(std::exception_ptr, std::vector<int>)>>(func);

    auto fut = packageP->get_future();
    auto handler = [packageP] (const std::vector<int> &data) mutable -> bool {
        std::cout << "Calling handler with data, size: " << data.size() << std::endl;

        (*packageP)(nullptr, data);
        return data.size();
    };

    auto fut2 = std::async(std::launch::async, [=]() mutable {
            std::cout << "Before handler called " << from << to << std::endl;

            std::vector<int> vec ( to, from );

            auto res = handler(vec);

            std::cout << "Handler result " << res << " for " << from << " " << to << std::endl;
        });

    std::cout << "Called async in countup for " << from << " " << to << std::endl;

    return fut;
}

int main ()
{
  auto futX = std::async(std::launch::async, [] {
      auto fut1 = std::async(std::launch::async, [] {
        auto fut2 = countup(0, 2);

        std::cout << "Called X countup and waiting to finish" << std::endl;

        fut2.wait();

        auto vec = fut2.get();
        std::cout << "The X countup returned" << std::endl;
      });
      std::cout << "Called X async internal and waiting to finish" << std::endl;
      fut1.wait();

      return 2;
  });

  std::cout << "Called async X and waiting to finish" << std::endl;

  auto futY = std::async(std::launch::async, [] {
      auto fut1 = std::async(std::launch::async, [] {
        auto fut2 = countup(0, 3);

        std::cout << "Called Y countup and waiting to finish" << std::endl;
        fut2.wait();

        auto vec = fut2.get();
        std::cout << "The Y countup returned " << std::endl;
      });
      std::cout << "Called Y async internal and waiting to finish" << std::endl;
      fut1.wait();

      return 3;
  });

  std::cout << "Called async Y and waiting to finish" << std::endl;

  futX.wait();
  std::cout << "After async X and waiting to finish" << std::endl;

  futY.wait();
  std::cout << "After async Y and waiting to finish" << std::endl;

  int valueX = futX.get();                  // wait for the task to finish and get result
  int valueY = futY.get();                  // wait for the task to finish and get result

  std::cout << "The countdown lasted for " << valueX  << " " << valueY << " seconds" << std::endl;

  return 0;
}

See it on coliru

并非所有 std::packaged_tasks 在内部 std::async 调用时执行

Not all std::packaged_tasks executed when inside std::async calls

c++

multithreading

packaged-task

c++11

std-future