当一个工作线程失败时，如何中止剩余的工人？

2018-07-03 00:44:45

我有一个程序产生多个线程，每个线程执行长时间运行的任务。主线程然后等待所有工作线程加入，收集结果并退出。

如果其中一位工作人员发生错误，我希望其余的工作人员能够优雅地停下来，这样主线程可以在不久之后退出。

我的问题是如何最好地做到这一点，当长期运行的任务的实现是由我无法修改代码的库提供的。

这是系统的一个简单的草图，没有错误处理：

void threadFunc()
{
    // Do long-running stuff
}

void mainFunc()
{
    std::vector<std::thread> threads;

    for (int i = 0; i < 3; ++i) {
        threads.push_back(std::thread(&threadFunc));
    }

    for (auto &t : threads) {
        t.join();
    }
}

如果长时间运行的函数执行一个循环并且可以访问代码，那么只需通过检查每次迭代顶部的共享“继续运行”标志就可以中止执行。

std::mutex mutex;
bool error;

void threadFunc()
{
    try {
        for (...) {
            {
                std::unique_lock<std::mutex> lock(mutex);
                if (error) {
                    break;
                }
            }
        }
    } catch (std::exception &) {
        std::unique_lock<std::mutex> lock(mutex);
        error = true;
    }
}

现在考虑图书馆提供长时间运行的情况：

std::mutex mutex;
bool error;

class Task
{
public:
    // Blocks until completion, error, or stop() is called
    void run();

    void stop();
};

void threadFunc(Task &task)
{
    try {
        task.run();
    } catch (std::exception &) {
        std::unique_lock<std::mutex> lock(mutex);
        error = true;
    }
}

在这种情况下，主线程必须处理该错误，并在仍在运行的任务上调用stop() 。因此，它不能简单地等待每个worker join()就像在原始实现中一样。

我迄今使用的方法是在主线程和每个工作者之间共享以下结构：

struct SharedData
{
    std::mutex mutex;
    std::condition_variable condVar;
    bool error;
    int running;
}

当工人成功完成时，它会减少running计数。如果发现异常，则工作人员设置error标志。在这两种情况下，它都会调用condVar.notify_one() 。

主线程然后等待条件变量，如果设置了error或者running到零，则唤醒。在唤醒时，如果error已被设置，主线程将在所有任务上调用stop() 。

这种方法很有效，但我觉得应该有一个更清晰的解决方案，使用标准并发库中的一些更高级别的基元。任何人都可以提出改进的实施？

以下是我当前解决方案的完整代码：

// main.cpp

#include <chrono>
#include <mutex>
#include <thread>
#include <vector>

#include "utils.h"

// Class which encapsulates long-running task, and provides a mechanism for aborting it
class Task
{
public:
    Task(int tidx, bool fail)
    :   tidx(tidx)
    ,   fail(fail)
    ,   m_run(true)
    {

    }

    void run()
    {
        static const int NUM_ITERATIONS = 10;

        for (int iter = 0; iter < NUM_ITERATIONS; ++iter) {
            {
                std::unique_lock<std::mutex> lock(m_mutex);
                if (!m_run) {
                    out() << "thread " << tidx << " aborting";
                    break;
                }
            }

            out() << "thread " << tidx << " iter " << iter;
            std::this_thread::sleep_for(std::chrono::milliseconds(100));

            if (fail) {
                throw std::exception();
            }
        }
    }

    void stop()
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_run = false;
    }

    const int tidx;
    const bool fail;

private:
    std::mutex m_mutex;
    bool m_run;
};

// Data shared between all threads
struct SharedData
{
    std::mutex mutex;
    std::condition_variable condVar;
    bool error;
    int running;

    SharedData(int count)
    :   error(false)
    ,   running(count)
    {

    }
};

void threadFunc(Task &task, SharedData &shared)
{
    try {
        out() << "thread " << task.tidx << " starting";

        task.run(); // Blocks until task completes or is aborted by main thread

        out() << "thread " << task.tidx << " ended";
    } catch (std::exception &) {
        out() << "thread " << task.tidx << " failed";

        std::unique_lock<std::mutex> lock(shared.mutex);
        shared.error = true;
    }

    {
        std::unique_lock<std::mutex> lock(shared.mutex);
        --shared.running;
    }

    shared.condVar.notify_one();
}

int main(int argc, char **argv)
{
    static const int NUM_THREADS = 3;

    std::vector<std::unique_ptr<Task>> tasks(NUM_THREADS);
    std::vector<std::thread> threads(NUM_THREADS);

    SharedData shared(NUM_THREADS);

    for (int tidx = 0; tidx < NUM_THREADS; ++tidx) {
        const bool fail = (tidx == 1);
        tasks[tidx] = std::make_unique<Task>(tidx, fail);
        threads[tidx] = std::thread(&threadFunc, std::ref(*tasks[tidx]), std::ref(shared));
    }

    {
        std::unique_lock<std::mutex> lock(shared.mutex);

        // Wake up when either all tasks have completed, or any one has failed
        shared.condVar.wait(lock, [&shared](){
            return shared.error || !shared.running;
        });

        if (shared.error) {
            out() << "error occurred - terminating remaining tasks";
            for (auto &t : tasks) {
                t->stop();
            }
        }
    }

    for (int tidx = 0; tidx < NUM_THREADS; ++tidx) {
        out() << "waiting for thread " << tidx << " to join";
        threads[tidx].join();
        out() << "thread " << tidx << " joined";
    }

    out() << "program complete";

    return 0;
}

这里定义了一些实用函数：

// utils.h

#include <iostream>
#include <mutex>
#include <thread>

#ifndef UTILS_H
#define UTILS_H

#if __cplusplus <= 201103L
// Backport std::make_unique from C++14
#include <memory>
namespace std {

template<typename T, typename ...Args>
std::unique_ptr<T> make_unique(
            Args&& ...args)
{
    return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
}

} // namespace std
#endif // __cplusplus <= 201103L

// Thread-safe wrapper around std::cout
class ThreadSafeStdOut
{
public:
    ThreadSafeStdOut()
    :   m_lock(m_mutex)
    {

    }

    ~ThreadSafeStdOut()
    {
        std::cout << std::endl;
    }

    template <typename T>
    ThreadSafeStdOut &operator<<(const T &obj)
    {
        std::cout << obj;
        return *this;
    }

private:
    static std::mutex m_mutex;
    std::unique_lock<std::mutex> m_lock;
};

std::mutex ThreadSafeStdOut::m_mutex;

// Convenience function for performing thread-safe output
ThreadSafeStdOut out()
{
    return ThreadSafeStdOut();
}

#endif // UTILS_H

我一直在考虑你的情况，这可能对你有些帮助。你或许可以尝试使用几种不同的方法来实现你的目标。有2-3个可能使用的选项或三者的组合。我至少会展示第一个选项，因为我仍然在学习并试图掌握模板专业化的概念以及使用Lambdas。

使用管理员类

使用模板专业化封装

使用Lambdas。

经理类的伪代码看起来像这样：

class ThreadManager {
private:
    std::unique_ptr<MainThread> mainThread_;
    std::list<std::shared_ptr<WorkerThread> lWorkers_;  // List to hold finished workers
    std::queue<std::shared_ptr<WorkerThread> qWorkers_; // Queue to hold inactive and waiting threads.
    std::map<unsigned, std::shared_ptr<WorkerThread> mThreadIds_; // Map to associate a WorkerThread with an ID value.
    std::map<unsigned, bool> mFinishedThreads_; // A map to keep track of finished and unfinished threads.

    bool threadError_; // Not needed if using exception handling
public:
    explicit ThreadManager( const MainThread& main_thread );

    void shutdownThread( const unsigned& threadId );
    void shutdownAllThreads();

    void addWorker( const WorkerThread& worker_thread );          
    bool isThreadDone( const unsigned& threadId );

    void spawnMainThread() const; // Method to start main thread's work.

    void spawnWorkerThread( unsigned threadId, bool& error );

    bool getThreadError( unsigned& threadID ); // Returns True If Thread Encountered An Error and passes the ID of that thread, 

};

仅用于演示目的，我是否使用bool值来确定线程是否因简单的结构而失败，当然，如果您更喜欢使用例外或无效的无符号值等，可以将其替换为您喜欢的类型。

现在使用这样的类将是这样的：另请注意，如果这是一个Singleton类型的对象，那么这个类型的类会被认为更好，因为您使用的是共享指针，因此您不需要超过1个ManagerClass 。

SomeClass::SomeClass( ... ) {
    // This class could contain a private static smart pointer of this Manager Class
    // Initialize the smart pointer giving it new memory for the Manager Class and by passing it a pointer of the Main Thread object

   threadManager_ = new ThreadManager( main_thread ); // Wouldn't actually use raw pointers here unless if you had a need to, but just shown for simplicity       
}

SomeClass::addThreads( ... ) {
    for ( unsigned u = 1, u <= threadCount; u++ ) {
         threadManager_->addWorker( some_worker_thread );
    }
}

SomeClass::someFunctionThatSpawnsThreads( ... ) {
    threadManager_->spawnMainThread();

    bool error = false;       
    for ( unsigned u = 1; u <= threadCount; u++ ) {
        threadManager_->spawnWorkerThread( u, error );

        if ( error ) { // This Thread Failed To Start, Shutdown All Threads
            threadManager->shutdownAllThreads();
        }
    }

    // If all threads spawn successfully we can do a while loop here to listen if one fails.
    unsigned threadId;
    while ( threadManager_->getThreadError( threadId ) ) {
         // If the function passed to this while loop returns true and we end up here, it will pass the id value of the failed thread.
         // We can now go through a for loop and stop all active threads.
         for ( unsigned u = threadID + 1; u <= threadCount; u++ ) {
             threadManager_->shutdownThread( u );
         }

         // We have successfully shutdown all threads
         break;
    }
}

我喜欢经理类的设计，因为我已经在其他项目中使用过它们，并且它们经常派上用场，尤其是在处理包含许多和多个资源的代码库时，例如具有许多资产的工作游戏引擎，如Sprites，纹理，音频文件，地图，游戏项目等。使用管理员类有助于跟踪和维护所有资产。这个相同的概念可以应用于“管理”活动，非活动，等待线程，并且知道如何正确处理和关闭所有线程。如果你的代码库和库支持异常以及线程安全的异常处理，而不是传递和使用bools来处理错误，我会推荐使用ExceptionHandler。还有一个Logger类对于可以写入日志文件和控制台窗口的位置是很好的，以便给出明确的消息，说明抛出异常的函数以及导致异常的原因是：日志消息可能如下所示：

Exception Thrown: someFunctionNamedThis in ThisFile on Line# (x)
    threadID 021342 failed to execute.

通过这种方式，您可以查看日志文件，并快速找出导致异常的线程，而不是使用传递的布尔变量。

The implementation of the long-running task is provided by a library whose code I cannot modify.

这意味着你无法同步工作线程完成的工作

If an error occurs in one of the workers,

假设您可以真正发现工作人员的错误; 那么一些可以很容易地被检测到，如果使用的库报告其他人不能，也就是说

库代码循环。

库代码过早地退出并带有未捕获的异常。

I want the remaining workers to stop **gracefully**

这是不可能的

最好的办法是编写一个线程管理器来检查工作线程状态，并且如果检测到错误情况，它就会“杀死”所有工作线程并退出。

您还应该考虑检测循环工作线程（通过超时）并向用户提供杀死或继续等待进程完成的选项。

你的问题是长时间运行的功能不是你的代码，你说你不能修改它。因此，除非库开发人员为您完成了这些操作，否则无法对任何类型的外部同步原语（条件变量，信号量，互斥锁，管道等）给予任何关注。

因此，你唯一的选择就是做一些事情，不管它在做什么，都可以控制任何代码。这是什么信号。为此，你将不得不使用pthread_kill（），或者不管现在是什么。

模式就是这样

检测到错误的线程需要以某种方式将该错误传回给主线程。

主线程然后需要为所有其他线程调用pthread_kill（）。不要被这个名字弄糊涂--pthread_kill（）只是一种向线程传递任意信号的方式。请注意，STOP，CONTINUE和TERMINATE等信号即使与pthread_kill（）一起引发，也不是线程相关的，因此不要使用这些信号。

在每一个线程中，你都需要一个信号处理程序。在将信号传递给线程时，无论长时间运行的函数在做什么，线程中的执行路径都会跳转到处理程序。

你现在回到（有限）控制中，并且可以（可能，或许）做一些有限的清理并终止线程。

在此期间，主线程将在所有线程上调用pthread_join（），然后这些线程将返回。

我的想法：

这是一个非常丑陋的做法（并且signal / pthreads非常难以正确，我也不是专家），但是我不确定你有什么其他选择。

在源代码中寻找'优美'将会有很长的路要走，尽管最终用户的体验是可以的。

您将通过运行该库函数中途部分中止执行，因此如果有任何清理，它通常会执行（例如，释放它已分配的内存），这样做不会完成，并且会有内存泄漏。在valgrind之类的东西下运行是一种解决这种情况的方法。

获得库函数清理（如果需要的话）的唯一方法是让信号处理程序将控制权返回给函数，并让它运行到完成状态，就是你不想做的事情。

当然，这在Windows上不起作用（没有pthreads，至少没有值得提及的，尽管可能有一个等价的机制）。

真的最好的方法是重新实现（如果可能的话）该库函数。

链接地址: http://www.djcxy.com/p/92047.html

上一篇: When one worker thread fails, how to abort remaining workers?

下一篇: How to manage worker thread lifecycles when main Java thread terminates?