如何在线程之间传播异常？

Question

我们有一个单个线程调用的函数(我们将其命名为主线程).在函数体内,我们生成多个工作线程来进行CPU密集型工作,等待所有线程完成,然后在主线程上返回结果.

结果是调用者可以天真地使用该函数,并且在内部它将使用多个核心.

到目前为止都很好..

我们遇到的问题是处理异常.我们不希望工作线程上的异常使应用程序崩溃.我们希望函数的调用者能够在主线程上捕获它们.我们必须捕获工作线程上的异常,并将它们传播到主线程,让它们继续从那里展开.

我们应该怎么做？

我能想到的最好的是:

1. 在我们的工作线程上捕获各种异常(std :: exception和我们自己的一些).
2. 记录异常的类型和消息.
3. 在主线程上有一个相应的switch语句,它重新抛出工作线程上记录的任何类型的异常.

这有一个明显的缺点,即只支持一组有限的异常类型,并且每当添加新的异常类型时都需要修改.

Answer 1

C++ 11引入了exception_ptr允许在线程之间传输异常的类型:

#include<iostream>
#include<thread>
#include<exception>
#include<stdexcept>

static std::exception_ptr teptr = nullptr;

void f()
{
    try
    {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        throw std::runtime_error("To be passed between threads");
    }
    catch(...)
    {
        teptr = std::current_exception();
    }
}

int main(int argc, char **argv)
{
    std::thread mythread(f);
    mythread.join();

    if (teptr) {
        try{
            std::rethrow_exception(teptr);
        }
        catch(const std::exception &ex)
        {
            std::cerr << "Thread exited with exception: " << ex.what() << "\n";
        }
    }

    return 0;
}

因为在您的情况下,您有多个工作线程,您需要exception_ptr为每个线程保留一个.

请注意,这exception_ptr是一个共享的类ptr指针,因此您需要至少保留一个exception_ptr指向每个异常的指针,否则它们将被释放.

特定于Microsoft:如果您使用SEH Exceptions(/EHa),示例代码也将传输SEH异常,例如访问冲突,这可能不是您想要的.

Answer 2

目前,唯一可移植的方法是为您可能希望在线程之间传输的所有类型的异常编写catch子句,将信息存储在该catch子句的某处,然后在以后使用它来重新抛出异常.这是Boost.Exception采用的方法.

在C++ 0x中,您将能够捕获异常,catch(...)然后将其存储在std::exception_ptrusing 的实例中std::current_exception().然后,您可以从相同或不同的线程中重新抛出它std::rethrow_exception().

如果您使用的是Microsoft Visual Studio 2005或更高版本,那么just :: thread C++ 0x线程库支持std::exception_ptr.(免责声明:这是我的产品).

Answer 3

如果您使用C++ 11,那么std::future可能你正在寻找什么:它可以自动地捕获异常,它作出的工作线程的顶部,并通过在点通过他们的父线程std::future::get是调用.(在幕后,这与@AnthonyWilliams的回答完全一样;它已经为你实现了.)

缺点是没有标准的方法来"停止关心" std::future; 甚至它的析构函数都会阻塞,直到任务完成.[编辑,2017年:阻止析构函数行为仅仅是从返回的伪期货的错误std::async,你不应该使用它.正常的期货不会在他们的析构函数中阻塞.但是如果你正在使用,你仍然无法"取消"任务std::future:即使没有人正在听取答案,承诺 - 履行任务将继续在幕后运行.]这是一个玩具示例,可能会澄清我的内容意思:

#include <atomic>
#include <chrono>
#include <exception>
#include <future>
#include <thread>
#include <vector>
#include <stdio.h>

bool is_prime(int n)
{
    if (n == 1010) {
        puts("is_prime(1010) throws an exception");
        throw std::logic_error("1010");
    }
    /* We actually want this loop to run slowly, for demonstration purposes. */
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    for (int i=2; i < n; ++i) { if (n % i == 0) return false; }
    return (n >= 2);
}

int worker()
{
    static std::atomic<int> hundreds(0);
    const int start = 100 * hundreds++;
    const int end = start + 100;
    int sum = 0;
    for (int i=start; i < end; ++i) {
        if (is_prime(i)) { printf("%d is prime\n", i); sum += i; }
    }
    return sum;
}

int spawn_workers(int N)
{
    std::vector<std::future<int>> waitables;
    for (int i=0; i < N; ++i) {
        std::future<int> f = std::async(std::launch::async, worker);
        waitables.emplace_back(std::move(f));
    }

    int sum = 0;
    for (std::future<int> &f : waitables) {
        sum += f.get();  /* may throw an exception */
    }
    return sum;
    /* But watch out! When f.get() throws an exception, we still need
     * to unwind the stack, which means destructing "waitables" and each
     * of its elements. The destructor of each std::future will block
     * as if calling this->wait(). So in fact this may not do what you
     * really want. */
}

int main()
{
    try {
        int sum = spawn_workers(100);
        printf("sum is %d\n", sum);
    } catch (std::exception &e) {
        /* This line will be printed after all the prime-number output. */
        printf("Caught %s\n", e.what());
    }
}

我只是尝试使用std::thread和编写一个类似于工作的示例std::exception_ptr,但是std::exception_ptr(使用libc ++)出现了问题,所以我还没有实际工作.:(

[编辑,2017年:

int main() {
    std::exception_ptr e;
    std::thread t1([&e](){
        try {
            ::operator new(-1);
        } catch (...) {
            e = std::current_exception();
        }
    });
    t1.join();
    try {
        std::rethrow_exception(e);
    } catch (const std::bad_alloc&) {
        puts("Success!");
    }
}

我不知道2013年我做错了什么,但我确定这是我的错.

Answer 4

您的问题是,您可能会从多个线程收到多个异常,因为每个线程可能会因为不同的原因而失败.

我假设主线程以某种方式等待线程结束以检索结果,或定期检查其他线程的进度,并且同步对共享数据的访问.

简单解决方案

简单的解决方案是捕获每个线程中的所有异常,将它们记录在共享变量中(在主线程中).

完成所有线程后,决定如何处理异常.这意味着所有其他线程继续进行处理,这可能不是您想要的.

复杂解决方案

如果从另一个线程抛出异常,则更复杂的解决方案是让每个线程检查其执行的关键点.

如果一个线程抛出异常,它会在退出线程之前被捕获,异常对象被复制到主线程中的某个容器中(如在简单解决方案中),并且一些共享布尔变量被设置为true.

当另一个线程测试这个布尔值时,它会看到执行将被中止,并以优雅的方式中止.

当所有线程都中止时,主线程可以根据需要处理异常.

Answer 5

从线程抛出的异常将无法在父线程中捕获。线程有不同的上下文和堆栈，通常父线程不需要留在那里等待子线程完成，以便它可以捕获它们的异常。代码中根本没有地方可以捕获该捕获：

try
{
  start thread();
  wait_finish( thread );
}
catch(...)
{
  // will catch exceptions generated within start and wait, 
  // but not from the thread itself
}

您将需要捕获每个线程内的异常并解释主线程中线程的退出状态以重新抛出您可能需要的任何异常。

顺便说一句，在线程中没有 catch 的情况下，如果堆栈展开将完全完成，则它是特定于实现的，即在调用终止之前甚至可能不会调用自动变量的析构函数。有些编译器会这样做，但这不是必需的。