multithreading - 如何使用std::atomic_bool提前完成std::async线程*而无需完成*？

Question

我有一个需要回调的函数，并使用它在10个单独的线程上进行工作。但是，通常情况下并不需要所有的工作。例如，如果在第三个线程上获得了期望的结果，则它应该停止其余 Activity 线程上的所有工作。

答案here表示，除非您让回调函数使用附加的std::atomic_bool参数，否则这是不可能的，这表明该函数是否应过早终止。

此解决方案对我不起作用。 worker 在基类中旋转，而基类的全部目的是抽象出多线程的细节。我怎样才能做到这一点？我预计我将不得不放弃std::async以获得更多的参与。

#include <iostream>
#include <future>
#include <vector>

class ABC{
public:
    std::vector<std::future<int> > m_results;
    ABC() {};
    ~ABC(){};
    virtual int callback(int a) = 0;
    void doStuffWithCallBack();
};


void ABC::doStuffWithCallBack(){

    // start working
    for(int i = 0; i < 10; ++i)
        m_results.push_back(std::async(&ABC::callback, this, i));

    // analyze results and cancel all threads when you get the 1
    for(int j = 0; j < 10; ++j){

        double foo = m_results[j].get();

        if ( foo == 1){
            break;  // but threads continue running
        }

    }
    std::cout << m_results[9].get() << " <- this shouldn't have ever been computed\n";
}

class Derived : public ABC {
public:
    Derived() : ABC() {};
    ~Derived() {};
    int callback(int a){
        std::cout << a << "!\n";
        if (a == 3)
            return 1;
        else
            return 0;
    };
};

int main(int argc, char **argv)
{

    Derived myObj;
    myObj.doStuffWithCallBack();

    return 0;
}

Answer 1

我只是说这可能不应该是“正常”程序的一部分，因为它可能会泄漏资源和/或使程序处于不稳定状态，但出于科学的考虑...

如果您可以控制线程循环，并且不介意使用平台功能，则可以将inject an exception插入线程。使用posix可以为此使用信号，在Windows上必须使用SetThreadContext()。尽管异常通常会释放堆栈并调用析构函数，但发生异常时，您的线程可能位于系统调用或其他“非异常安全的地方”。

免责声明:目前我只有Linux，因此没有测试Windows代码。

#if defined(_WIN32)
#   define ITS_WINDOWS
#else
#   define ITS_POSIX
#endif


#if defined(ITS_POSIX)
#include <signal.h>
#endif

void throw_exception() throw(std::string())
{
    throw std::string();
}

void init_exceptions()
{
    volatile int i = 0;
    if (i)
        throw_exception();
}

bool abort_thread(std::thread &t)
{

#if defined(ITS_WINDOWS)

    bool bSuccess = false;
    HANDLE h = t.native_handle();
    if (INVALID_HANDLE_VALUE == h)
        return false;

    if (INFINITE == SuspendThread(h))
        return false;

    CONTEXT ctx;
    ctx.ContextFlags = CONTEXT_CONTROL;
    if (GetThreadContext(h, &ctx))
    {
#if defined( _WIN64 )
        ctx.Rip = (DWORD)(DWORD_PTR)throw_exception;
#else
        ctx.Eip = (DWORD)(DWORD_PTR)throw_exception;
#endif

        bSuccess = SetThreadContext(h, &ctx) ? true : false;
    }

    ResumeThread(h);

    return bSuccess;

#elif defined(ITS_POSIX)

    pthread_kill(t.native_handle(), SIGUSR2);

#endif

    return false;
}


#if defined(ITS_POSIX)

void worker_thread_sig(int sig)
{
    if(SIGUSR2 == sig)
        throw std::string();
}

#endif

void init_threads()
{
#if defined(ITS_POSIX)

    struct sigaction sa;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = 0;
    sa.sa_handler = worker_thread_sig;
    sigaction(SIGUSR2, &sa, 0);

#endif
}

class tracker
{
public:
    tracker() { printf("tracker()\n"); }
    ~tracker() { printf("~tracker()\n"); }
};

int main(int argc, char *argv[])
{
    init_threads();

    printf("main: starting thread...\n");
    std::thread t([]()
    {
        try
        {
            tracker a;

            init_exceptions();

            printf("thread: started...\n");
            std::this_thread::sleep_for(std::chrono::minutes(1000));
            printf("thread: stopping...\n");
        }
        catch(std::string s)
        {
            printf("thread: exception caught...\n");
        }
    });

    printf("main: sleeping...\n");
    std::this_thread::sleep_for(std::chrono::seconds(2));

    printf("main: aborting...\n");
    abort_thread(t);

    printf("main: joining...\n");
    t.join();

    printf("main: exiting...\n");

    return 0;
}

输出:

main: starting thread...
main: sleeping...
tracker()
thread: started...
main: aborting...
main: joining...
~tracker()
thread: exception caught...
main: exiting...