(C++20) jthread中stop_token的基础使用

C++20 jthread

std::jthread - cppreference.com

std::stop_token - cppreference.com

std::stop_source - cppreference.com

在C++20中推出了jthread。其两个核心功能是：

• 自动合并joining thread
• 线程取消stop token

下面有两个重点：

在std::jthread中有request_stop()操作能够原子的将关联的std::stop_source停止标志进行切换。

而std::stop_token是对于关联的std::stop_source视图。

---

这里辅助打印使用一个基于RAII的技巧：

视频讲解：【细莲】(C++) 基于RAII的多线程类原子输出_哔哩哔哩_bilibili

class Writer {
    std::ostringstream buffer;

public:
    Writer() {
        buffer << std::this_thread::get_id();
    }
    ~Writer() {
        std::cout << buffer.str();
    }
    Writer& operator<<(auto input) {
        buffer << input;
        return *this;
    }
};

使用方式

循环判断

最普通的方式就是循环判断。这对于常见的线程封装来说很常见。

std::jthread能够接受一个第一个参数为std::stop_token stoken的函数。该sstoken与线程对象绑定。

#include <chrono>
#include <iostream>
#include <sstream>
#include <thread>

class Writer {
    std::ostringstream buffer;

public:
    Writer() {
        buffer << std::this_thread::get_id();
    }
    ~Writer() {
        std::cout << buffer.str();
    }
    Writer& operator<<(auto input) {
        buffer << input;
        return *this;
    }
};

void fun_jth_pass(std::stop_token stoken, long secCnt) {
    Writer() << "Test While Start" << '\n';
    // 直到请求停止
    do {
        std::this_thread::sleep_for(std::chrono::milliseconds(secCnt));
        Writer() << "Test While Running" << '\n';
    } while (!stoken.stop_requested());

    Writer() << "Test While End" << '\n';
}

int main() {
    std::cout << std::boolalpha;
    Writer() << ">>>Main Start\n";

    /**
     * 可以手动显示操作
     * 显示的请求结束和join
     */
    {
        std::jthread jth_pass(fun_jth_pass, 300);
        std::this_thread::sleep_for(std::chrono::seconds(1));
        if (jth_pass.joinable()) {
            Writer() << "request_stop() = " << jth_pass.request_stop() << '\n';
            jth_pass.join();
        }
    }

    Writer() << ">>>Main End\n";
}

11124>>>Main Start
20148Test While Start
20148Test While Running
20148Test While Running
20148Test While Running
11124request_stop() = 1
20148Test While Running
20148Test While End
11124>>>Main End

条件变量condition_variable_any

在多线程中单单使用循环判断一个标志是效率比较低的。如果能将一个线程进行挂起那CPU的效果会提升很多。

这里就需要介绍std::condition_variable_any。

这里的wait()第二个参数就是std::stop_token，当外部停止请求stoken停止时，会自动唤醒cv_any。

template< class Lock, class Predicate >
bool wait( Lock& lock, std::stop_token stoken, Predicate pred );

#include <chrono>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <sstream>
#include <thread>

class Writer {
    std::ostringstream buffer;

public:
    Writer() {
        buffer << std::this_thread::get_id();
    }
    ~Writer() {
        std::cout << buffer.str();
    }
    Writer& operator<<(auto input) {
        buffer << input;
        return *this;
    }
};
void fun_jth_cv_any(std::stop_token stoken) {
    std::mutex       mutex;
    std::unique_lock lock(mutex);

    // `cv_any::wait`有`std::stop_token`的重载
    // `stoken`停止时，会自动唤醒`cv_any`
    std::condition_variable_any().wait(lock, stoken, [] {
        Writer() << "Test condition_variable_any::wait() Running" << '\n';
        return false;
    });
    Writer() << "Test condition_variable_any::wait() End" << '\n';
}

int main() {
    std::cout << std::boolalpha;
    Writer() << ">>>Main Start\n";

    /**
     * 基于RAII会自动停止stoken和进行join
     */
    {
        std::jthread jth_cv_any(fun_jth_cv_any);
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    Writer() << ">>>Main End\n";
}

20416>>>Main Start
19876Test condition_variable_any::wait() Running
19876Test condition_variable_any::wait() Running
19876Test condition_variable_any::wait() Running
19876Test condition_variable_any::wait() End
20416>>>Main End

stop回调 std::stop_callback

可以对stop_token对象注册一个监控停止的回调。

当外部请求停止request_stop()时。会自动调用注册的callback。

其中回调函数的声明周期，跟随回调对象。

template< class Callback >
class stop_callback;

template<class C>
explicit stop_callback( std::stop_token&& st, C&& cb ) noexcept(/*see below*/);

template<class C>
explicit stop_callback( const std::stop_token& st, C&& cb ) noexcept(/*see below*/);

#include <chrono>
#include <iostream>
#include <sstream>
#include <thread>

class Writer {
    std::ostringstream buffer;

public:
    Writer() {
        buffer << std::this_thread::get_id();
    }
    ~Writer() {
        std::cout << buffer.str();
    }
    Writer& operator<<(auto input) {
        buffer << input;
        return *this;
    }
};

void fun(std::stop_token stoken) {
    Writer() << "Start worker thread\n";
    while (!stoken.stop_requested()) {
        Writer() << "is sleeping and need stop\n";
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    Writer() << "End worker thread\n";
}

int main() {
    Writer() << ">>>Main End\n";

    {
        std::jthread       jth(fun);
        std::stop_callback callback(jth.get_stop_token(),
                                    [] { Writer() << "$$$Stop callback executed\n"; });

        std::this_thread::sleep_for(std::chrono::seconds(3));
        Writer() << ">>>Main before request\n";
        jth.request_stop();
        Writer() << ">>>Main after request\n";
    }

    Writer() << ">>>Main End\n";
}

19312>>>Main End
6856Start worker thread
6856is sleeping and need stop
6856is sleeping and need stop
6856is sleeping and need stop
19312>>>Main before request
6856is sleeping and need stop
19312$$$Stop callback executed
19312>>>Main after request
6856End worker thread
19312>>>Main End

---

---

---

END

关注我，学习更多C/C++，算法，计算机知识