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

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

如果其中一个工作线程发生错误，我希望剩余的工作线程优雅地停止，以便主线程可以很快退出。

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

下面是一个简单的系统草图，没有错误处理:

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() on仍在运行的任务。因此，它不能简单地等待每个工人join()在原始实现中 到目前为止，我使用的方法是在以下结构之间共享主线程和每个工作线程:

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

当一个worker成功完成时，减少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

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, 
};

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;
    }
}

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