C++20中增加了很多重量級(jí)新特性，它不僅帶來(lái)了ranges、concept和協(xié)程，也為多線程編程帶來(lái)了jthread和stop_source這些強(qiáng)力輔助。利用這些新特性，我們可以更高效地編寫(xiě)并發(fā)程序。

今天要說(shuō)的就是利用jthread和stop_source來(lái)簡(jiǎn)化線程超時(shí)控制的實(shí)現(xiàn)，最終我們可以實(shí)現(xiàn)一個(gè)簡(jiǎn)單高效、可維護(hù)性不輸給號(hào)稱(chēng)“天生支持并發(fā)”的Go語(yǔ)言的版本。

為什么需要超時(shí)控制

超時(shí)控制是很常見(jiàn)的需求，最普遍的場(chǎng)景是為了防止程序卡住或者長(zhǎng)時(shí)間占用資源，程序會(huì)主動(dòng)取消掉一些超過(guò)允許運(yùn)行時(shí)間的或者無(wú)響應(yīng)的線程，比如一些耗時(shí)很長(zhǎng)的網(wǎng)絡(luò)連接處理線程等。當(dāng)然用戶等得不耐煩了手動(dòng)點(diǎn)擊取消任務(wù)執(zhí)行也勉強(qiáng)可以算在內(nèi)。

通常超時(shí)發(fā)生或者用戶點(diǎn)擊取消之后，我們都期待線程能迅速終止執(zhí)行并讓整個(gè)程序保持一個(gè)完整且安全的狀態(tài)。然而現(xiàn)實(shí)是復(fù)雜的，想實(shí)現(xiàn)上述功能對(duì)于線程來(lái)說(shuō)是一件難事，尤其在Linux系統(tǒng)上。

第一個(gè)難點(diǎn)是如何讓線程知道自己要退出。對(duì)于進(jìn)程來(lái)說(shuō)這不是難點(diǎn)，因?yàn)椴还苓M(jìn)程在做什么，我們都可以靠向其發(fā)送信號(hào)來(lái)立即中斷進(jìn)程的執(zhí)行（前提是線程沒(méi)有屏蔽這個(gè)信號(hào)），這樣進(jìn)程的停止請(qǐng)求可以被立即感知到，進(jìn)程從而可以盡快完成善后工作退出執(zhí)行。同樣的招數(shù)對(duì)多線程程序來(lái)說(shuō)就沒(méi)那么好用了——信號(hào)默認(rèn)是發(fā)給整個(gè)進(jìn)程的，為了能讓每個(gè)線程獨(dú)立地接收信號(hào)，我們需要保存線程的標(biāo)識(shí)符并在每個(gè)線程中設(shè)置接收和屏蔽信號(hào)的mask，這大大增加了程序的復(fù)雜性；其次信號(hào)處理函數(shù)是整個(gè)進(jìn)程內(nèi)所有線程共享的，我們需要額外的手段來(lái)保證并發(fā)安全，同時(shí)還得兼顧信號(hào)處理函數(shù)需要可重入、快速執(zhí)行的最佳實(shí)踐，這會(huì)提高程序的開(kāi)發(fā)難度。

第二個(gè)難點(diǎn)在于如何保證線程一定會(huì)退出執(zhí)行。前面說(shuō)到信號(hào)可以打斷進(jìn)程的執(zhí)行，但這只是通知，實(shí)際上進(jìn)程完全可以在信號(hào)處理函數(shù)返回后無(wú)視這個(gè)通知繼續(xù)運(yùn)行，或者有一種更普遍的場(chǎng)景——程序正好卡在某個(gè)系統(tǒng)調(diào)用上，而程序又設(shè)置了系統(tǒng)調(diào)用被信號(hào)中斷后自動(dòng)重啟，這樣即使我們有效通知了進(jìn)程，進(jìn)程也會(huì)在收完通知之后再次進(jìn)入系統(tǒng)調(diào)用從而無(wú)法響應(yīng)停止請(qǐng)求。所以作為保底手段，Linux可以發(fā)送SIGKILL這個(gè)信號(hào)強(qiáng)制終止進(jìn)程，這個(gè)信號(hào)無(wú)法捕獲也無(wú)法屏蔽，是我們貨真價(jià)實(shí)的“底牌”。

上述的情況在多線程中同樣存在，而且我們沒(méi)有“底牌”可用——因?yàn)椴还芙o哪個(gè)線程發(fā)送SIGKILL，都會(huì)殺死整個(gè)進(jìn)程而不是單獨(dú)接收到信號(hào)的那個(gè)線程。另外即使有辦法強(qiáng)制終止線程（比如早期的JVM），我們還會(huì)遇到資源釋放的問(wèn)題。進(jìn)程退出執(zhí)行之后，內(nèi)核會(huì)盡可能釋放進(jìn)程持有的所有資源，打開(kāi)的文件會(huì)被關(guān)閉，緩沖區(qū)的內(nèi)容會(huì)被刷新，文件鎖之類(lèi)的同步機(jī)制也會(huì)正常解鎖；但線程并沒(méi)有這種自動(dòng)清理機(jī)制，清理工作完全需要手動(dòng)執(zhí)行，一旦進(jìn)程沒(méi)有釋放自己持有的資源就退出，系統(tǒng)就會(huì)遇到各種數(shù)據(jù)損壞和死鎖等并發(fā)問(wèn)題，排查和修復(fù)會(huì)極其困難。

為了克服上述難點(diǎn)并安全高效地實(shí)現(xiàn)終止超時(shí)線程的執(zhí)行，我們需要一些額外的控制手段。這也一直都是開(kāi)發(fā)者中的熱門(mén)話題。

在介紹C++20如何簡(jiǎn)化超時(shí)控制之前，我們先來(lái)看看前人的智慧成果。

Golang實(shí)現(xiàn)超時(shí)控制

Golang是天生支持并發(fā)的語(yǔ)言，這一點(diǎn)可謂名副其實(shí)，尤其是在超時(shí)控制上。

我們直接看個(gè)例子，例子里有主線程和工作線程，工作線程超時(shí)時(shí)間為5秒，如果超過(guò)這個(gè)時(shí)間還有線程沒(méi)完成工作，就取消所有線程的執(zhí)行。Golang里沒(méi)有系統(tǒng)級(jí)的線程，但我們可以用goroutine模擬。

在工作線程中我們用sleep代替耗時(shí)的工作，這樣便于測(cè)試：

func Work(ctx context.Context, id int) error {
	for range 10 {
		select {
		case <-ctx.Done():
			fmt.Printf("worker %d: canceled\n", id)
			return ctx.Err()
		default:
		}
		if rand.IntN(2) == 0 {
			time.Sleep(500 * time.Millisecond)
		} else {
			time.Sleep(time.Second)
		}
	}
	fmt.Printf("worker %d: done\n", id)
	return nil
}

超時(shí)控制是ctx參數(shù)實(shí)現(xiàn)的，每次循環(huán)處理前我們都會(huì)主動(dòng)檢查線程是否需要退出，這種協(xié)作式的“請(qǐng)求-檢查-響應(yīng)”是各種語(yǔ)言中取消線程執(zhí)行的常見(jiàn)做法。

這個(gè)工作函數(shù)執(zhí)行時(shí)間在5秒到10秒之間，取值的步長(zhǎng)在0.5秒，加上go標(biāo)準(zhǔn)庫(kù)默認(rèn)隨機(jī)數(shù)是均勻分布的，所以整體執(zhí)行時(shí)間的概率是正態(tài)分布的，在7.5秒左右我們很容易看到超時(shí)和正常運(yùn)行結(jié)束兩種情況。所以我們把超時(shí)時(shí)間分別設(shè)為4秒、7.5秒、11秒，來(lái)進(jìn)行模擬運(yùn)行實(shí)驗(yàn)：

func main() {
    // 從命令行獲取超時(shí)時(shí)間，單位毫秒
    timeout, err := strconv.Atoi(os.Args[1])
    if err != nil {
        panic(err)
    }
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Millisecond)
	now := time.Now()
	defer cancel()
	g := &errgroup.Group{}
	for i := range 3 {
		g.Go(func() error {
			return Work(ctx, i)
		})
	}
	err = g.Wait()
	fmt.Printf("run time: %s\n", time.Since(now))
	if err != nil {
		if errors.Is(err, context.DeadlineExceeded) {
			fmt.Println("Tasks canceled")
			return
		}
		panic(err)
	}
	fmt.Println("All work done!")
}

代碼很簡(jiǎn)單，關(guān)鍵在這行：ctx, cancel := context.WithTimeout(context.Background(), 7500*time.Millisecond)，只要我們?cè)O(shè)定的時(shí)間到了，<-ctx.Done()就會(huì)從阻塞變?yōu)榉亲枞?，循環(huán)開(kāi)始處的檢查會(huì)發(fā)現(xiàn)這個(gè)變化，然后會(huì)退出線程的執(zhí)行。代碼中使用了errgroup，但這不是必須的，實(shí)際上有很多辦法可以通知主線程，這里我選擇了一種最通用的，代價(jià)是代碼會(huì)稍微復(fù)雜一些。

運(yùn)行代碼，會(huì)看到下面這樣的輸出，結(jié)果有很大的隨機(jī)成分，下面只是無(wú)數(shù)種可能中的一種：

$ go build -o test

$ ./test 4000
worker 1: canceled
worker 0: canceled
worker 2: canceled
run time: 4.00431275s
Tasks canceled

$ ./test 7500
worker 0: done
worker 2: done
worker 1: canceled
run time: 7.507776458s
Tasks canceled

$ ./test 11000
worker 1: done
worker 2: done
worker 0: done
run time: 8.509193125s
All work done!

可以看到超時(shí)控制發(fā)揮了作用，盡管內(nèi)置的time計(jì)時(shí)有一些誤差，但程序的總體的運(yùn)行時(shí)間是小于等于超時(shí)時(shí)間的。

Golang的超時(shí)控制可以通過(guò)context簡(jiǎn)單實(shí)現(xiàn)，但需要工作線程主動(dòng)檢查主動(dòng)配合，前文我們也提到了強(qiáng)制終止工作線程很可能會(huì)造成并發(fā)問(wèn)題，因此所有的線程超時(shí)控制中都是采用的這種協(xié)作式退出機(jī)制，即使天生并發(fā)的語(yǔ)言也不能免俗。作為代價(jià)，我們需要謹(jǐn)慎編碼以免工作線程無(wú)法響應(yīng)退出請(qǐng)求，同時(shí)還需要付出一點(diǎn)在循環(huán)里檢查是否需要退出執(zhí)行的性能損失。

C++中的典型超時(shí)控制實(shí)現(xiàn)

c++沒(méi)有方便好用的context，想要實(shí)現(xiàn)協(xié)作式退出得自己造輪子。

Golang好用是因?yàn)闃?biāo)準(zhǔn)庫(kù)和運(yùn)行時(shí)調(diào)度器隱藏了實(shí)現(xiàn)的細(xì)節(jié)：WithTimeout實(shí)際上會(huì)創(chuàng)建一個(gè)定時(shí)器，到時(shí)間后調(diào)度器會(huì)執(zhí)行定時(shí)器的回調(diào)函數(shù)主動(dòng)關(guān)閉ctx內(nèi)部的channel，這樣<-ctx.Done()就會(huì)從阻塞變成非阻塞，協(xié)程就能檢查到這一變化從而退出執(zhí)行。

核心只在于兩點(diǎn)，以合適的方法標(biāo)記線程已被取消和異步地在超時(shí)后設(shè)置取消標(biāo)記。

第一點(diǎn)很容易解決，使用原子變量即可。第二點(diǎn)的異步通知有些棘手，但我們還是有幾種選擇：

• 使用alarm和信號(hào): alarm會(huì)注冊(cè)一個(gè)定時(shí)器，到時(shí)間后給進(jìn)程發(fā)送SIGALRM信號(hào)，雖說(shuō)多線程程序里不推薦用信號(hào)，但在這個(gè)場(chǎng)景下在信號(hào)處理函數(shù)里設(shè)置原子變量是合適的，另外使用alarm(0)可以取消之前注冊(cè)的定時(shí)器。
• 使用多線程：我們可以另外創(chuàng)建一個(gè)線程，并在其中等待到超時(shí)時(shí)間過(guò)去之后設(shè)置標(biāo)志，這樣主線程也不會(huì)阻塞。

當(dāng)然兩個(gè)方案各有缺點(diǎn)：

• alarm是整個(gè)進(jìn)程共享的，且同時(shí)只能設(shè)置一個(gè)定時(shí)器，最后它只能設(shè)置秒級(jí)精度的超時(shí)時(shí)間；使用setitimer可以解決上面這些問(wèn)題，但會(huì)出現(xiàn)不知道信號(hào)是哪個(gè)超時(shí)的定時(shí)器發(fā)送的問(wèn)題。
• 多線程方案問(wèn)題比較少，集中在變量生命周期和任務(wù)正常完成如何取消超時(shí)控制線程這兩點(diǎn)上。

綜合來(lái)看使用多線程方案才能真正解決問(wèn)題，跨平臺(tái)性也更強(qiáng)。知道原理后我們就可以寫(xiě)實(shí)驗(yàn)代碼了。

轉(zhuǎn)換后的工作函數(shù)是這樣的：

namespace {
    std::atomic<int> canceled_flag{0};
    std::atomic<int> is_canceled{0};
}

void Work(int id)
{
    std::mt19937 rng{std::random_device{}()};
    std::uniform_int_distribution<int> dist(0, 1);

    for (int i = 0; i < 10; ++i) {
        if (canceled_flag.load(std::memory_order_acquire) == 1) {
            std::osyncstream{std::cout} << "worker: " << id << " canceled\n";
            is_canceled.store(1, std::memory_order_release);
            return;
        }
        if (dist(rng) == 0) {
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        } else {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }

    std::osyncstream{std::cout} << "worker: " << id << " done\n";
}

代碼和go版本的沒(méi)有太大差異，唯一的區(qū)別是我們不靠返回值而是is_canceled標(biāo)志來(lái)區(qū)分線程是否因?yàn)楸蝗∠顺?。只使?code>canceled_flag會(huì)導(dǎo)致競(jìng)態(tài)條件并導(dǎo)致誤判，你可以想想是為什么，這算是課后練習(xí)。另外我在這還使用了memory_order，這不是必須的，但默認(rèn)的cst內(nèi)存序多少有些殺雞用牛刀。

下面是主線程和超時(shí)控制線程的邏輯：

int main(int argc, const char *argv[])
{
    if (argc != 2) {
        std::cerr << "wrong arg\n";
        return 1;
    }
    auto timeout = std::stoi(argv[1]);

    std::vector<std::thread> workers;
    constexpr int worker_num = 3;
    workers.reserve(worker_num);
    for (int i = 0; i < worker_num; ++i) {
        workers.emplace_back(work, i);
    }

    std::atomic<int> timeout_cancel_flag{0};
    std::thread{
        // 超時(shí)控制線程
        [&timeout_cancel_flag](auto timeout){
            std::this_thread::sleep_for(timeout);
            if (timeout_cancel_flag.load(std::memory_order_acquire) == 1) { // 危險(xiǎn)
                return;
            }
            canceled_flag.store(1, std::memory_order_release);
        }, std::chrono::milliseconds(timeout)
    }.detach();

    for (auto &worker: workers) {
        worker.join();
    }
    timeout_cancel_flag.store(1, std::memory_order_release);

    if (is_canceled.load(std::memory_order_acquire) != 1) {
        std::osyncstream{std::cout} << "All works done!\n";
    } else {
        std::osyncstream{std::cout} << "Tasks canceled\n";
    }
}

整體上沒(méi)什么難懂的地方，基本可以看做Golang版本的轉(zhuǎn)譯，如果線程都退出之后不管是否超時(shí)我們都要取消超時(shí)控制線程。整體上只有一點(diǎn)不一樣：超時(shí)控制線程是detach的，因?yàn)槲覀儾荒茏屩骶€程阻塞。

然而這段代碼有很致命的生命周期問(wèn)題，想象一下如果worker都在timeout之前完成工作，且函數(shù)在timeout之前退出，但超時(shí)控制線程仍然需要睡眠到timeout為止，這時(shí)候它醒來(lái)訪問(wèn)到的timeout_cancel_flag將會(huì)是一個(gè)無(wú)效值。

問(wèn)題出在兩個(gè)地方，第一個(gè)是我們用了sleep，這不可中斷，線程必須要等滿timeout時(shí)間才能退出，這會(huì)造成線程泄漏；第二是因?yàn)閟leep不可中斷，導(dǎo)致我們的超時(shí)控制線程生命周期長(zhǎng)于主線程和工作線程，在其中引用的主線程的局部變量很可能會(huì)失效。

解決方案當(dāng)然也很多，最簡(jiǎn)單的就是用std::shared_ptr包裹我們需要跨線程訪問(wèn)的資源，這和在rust中使用Arc是一樣的。但這種方案治標(biāo)不治本，我們的超時(shí)控制線程仍然會(huì)有比其他線程長(zhǎng)的生命周期。

第二種則是使用一種在超時(shí)等待中可以被中斷的機(jī)制，c++20前我們有std::timed_mutex和條件變量可用。

改進(jìn)后的代碼：

auto timeout_cancel_flag = std::make_shared<std::condition_variable>();
std::thread{
    [timeout_cancel_flag](auto timeout){
        std::mutex timeout_lock;
        std::unique_lock u{timeout_lock};
        if (timeout_cancel_flag->wait_for(u, timeout) == std::cv_status::timeout) {
            std::osyncstream{std::cout} << "cancel all threads\n";
            canceled_flag.store(1, std::memory_order_release);
        } else {
            std::osyncstream{std::cout} << "self was canceled by main thread\n";
        }
    }, std::chrono::milliseconds(timeout)
}.detach();

for (auto &worker: workers) {
    worker.join();
}

timeout_cancel_flag->notify_all(); // 取消超時(shí)控制線程

我們可以使用條件變量的wait_for方法，它可以讓當(dāng)前線程阻塞到指定的時(shí)間，或者中途被notify喚醒。這完美實(shí)現(xiàn)了我們既要超時(shí)等待又要中途可被打斷的需求。并且條件變量本身用std::shared_ptr包裹，不會(huì)有任何生命周期問(wèn)題。

然而沒(méi)有了生命周期問(wèn)題，我們還有時(shí)序問(wèn)題，如果主線程中的notify_all()早于控制線程中的wait_for執(zhí)行（概率比較小但不為0），那么這次notify超時(shí)控制線程是收不到的，wait會(huì)一直阻塞到超過(guò)timeout，這時(shí)候再設(shè)置取消標(biāo)志就沒(méi)有意義了。想要解決這種“喚醒丟失”問(wèn)題，我們需要借助wait重載的第三個(gè)參數(shù)，讓它告訴我們超時(shí)控制線程本身是否被取消：

struct TimeoutContext {
    std::atomic<int> canceled{0};
    std::mutex lock;
    std::condition_variable cv;
};

auto timeout_ctx = std::make_shared<TimeoutContext>();
std::thread{
    [timeout_ctx](auto timeout){
        // 必須使用ctx里的鎖才能有效避免競(jìng)態(tài)條件
        std::unique_lock u{timeout_ctx->lock};
        if (!timeout_cancel_flag->wait_for(u, timeout, [&](){ return timeout_ctx->canceled.load(std::memory_order_acquire) == 1; })) {
            // wait_for 返回 false，canceled是值還是0，說(shuō)明是超時(shí)導(dǎo)致的返回
            std::osyncstream{std::cout} << "cancel all threads\n";
            canceled_flag.store(1, std::memory_order_release);
        } else {
            // wait_for 返回 true，canceled被設(shè)置為1，說(shuō)明主線程通知了取消
            std::osyncstream{std::cout} << "self was canceled by main thread\n";
        }
    }, std::chrono::milliseconds(timeout)
}.detach();

for (auto &worker: workers) {
    worker.join();
}

// 取消超時(shí)控制線程
{
    // 獲取同一把鎖，修改狀態(tài)時(shí)要么超時(shí)控制線程還沒(méi)運(yùn)行，要么已經(jīng)在wait了
    std::lock_guard lk(timeout_ctx->lock);
    timeout_ctx->canceled.store(1, std::memory_order_release);
}
// 解鎖后才能通知
timeout_ctx->cv.notify_all();

因?yàn)橛墟i存在，所以不管怎么樣運(yùn)行順序只有兩種：

• 超時(shí)線程先運(yùn)行，一直到wait方法里解鎖，我們可以保證wait一定在notify之前運(yùn)行
• 主線程設(shè)置超時(shí)線程取消標(biāo)志的代碼先運(yùn)行，這時(shí)wait是晚于notify執(zhí)行的，但我們?cè)O(shè)置取消標(biāo)志是先于wait的，而wait在休眠前會(huì)先檢查謂詞條件，所以條件變量會(huì)馬上退出不會(huì)進(jìn)行等待。
• 會(huì)不會(huì)存在wait中條件變量解除了鎖，在即將進(jìn)入休眠前主線程完成了執(zhí)行？答案是不會(huì)的，標(biāo)準(zhǔn)有明文要求wait和它的兄弟函數(shù)里unlock+wait加在一起是原子的（實(shí)際上分為三部分，解鎖+休眠、被喚醒、重新加鎖，它們各自都是原子的）且和notify之間是全序關(guān)系——要么notify在前他們?cè)诤蠡蛘叻催^(guò)來(lái)，不可能同時(shí)執(zhí)行。簡(jiǎn)單說(shuō)，如果超時(shí)控制線程正在執(zhí)行unlock+wait，這說(shuō)明主線程沒(méi)有拿到鎖，此時(shí)主線程要么還沒(méi)運(yùn)行到notify（這種情況不會(huì)丟失喚醒），要么已經(jīng)設(shè)置了標(biāo)志并釋放了鎖，謂詞會(huì)檢測(cè)到標(biāo)志被設(shè)置（謂詞檢測(cè)在鎖的保護(hù)中），條件變量不進(jìn)入休眠；如果notify在之后運(yùn)行，則notify會(huì)看到超時(shí)控制線程已經(jīng)進(jìn)入wait，會(huì)喚醒它。所以不存在中間可以被打斷的場(chǎng)景。

現(xiàn)在不再有時(shí)序問(wèn)題了。

總體來(lái)說(shuō)這個(gè)實(shí)現(xiàn)還不錯(cuò)，能正常工作性能也尚可，很多框架也選擇了類(lèi)似的方案來(lái)實(shí)現(xiàn)線程的超時(shí)取消，比如Qt。然而它有幾個(gè)顯著的缺點(diǎn)：

• 代碼依賴很多全局狀態(tài)
• 我們需要用join等待所有線程退出，這是因?yàn)闃?biāo)準(zhǔn)庫(kù)的thread不join就析構(gòu)會(huì)導(dǎo)致程序崩潰，然而這是不必要的，我們只關(guān)心工作是否完成，剩下的資源釋放和線程退出無(wú)需去等待
• 線程之間暴露了過(guò)多的實(shí)現(xiàn)細(xì)節(jié)，比如flag具體的值，再比如std::condition_variable用來(lái)通知超時(shí)控制線程
• 代碼真的很復(fù)雜，想正確實(shí)現(xiàn)整體邏輯會(huì)比較麻煩，比如杜絕喚醒丟失

盡管有這些缺點(diǎn)，但在新標(biāo)準(zhǔn)之前我們只能使用這個(gè)方案。當(dāng)然如果放棄跨平臺(tái)的話，Linux上更安全更簡(jiǎn)單的做法其實(shí)是eventfd+epoll_wait，它不需要太多的外部狀態(tài)，且邏輯簡(jiǎn)單易懂，比用條件變量還要預(yù)防喚醒丟失強(qiáng)太多了，使用得當(dāng)?shù)脑捤男阅芤膊粫?huì)比純內(nèi)存操作的條件變量差多少。

C++20帶來(lái)的簡(jiǎn)化

C++20為并發(fā)編程體驗(yàn)帶來(lái)了不少提升。

第一個(gè)就是std::jthread，它會(huì)在析構(gòu)的時(shí)候自動(dòng)join，從而避免了std::thread手動(dòng)操作的麻煩。不僅如此，每個(gè)jthread中還包含一個(gè)std::stop_token，這可以簡(jiǎn)單實(shí)現(xiàn)線程的協(xié)作式取消執(zhí)行。僅僅這一個(gè)新特性就已經(jīng)解決了上一節(jié)說(shuō)的缺點(diǎn)中的前兩條。

第二個(gè)有用的新特性是std::stop_source和std::stop_token。一個(gè)std::stop_source可以產(chǎn)生多個(gè)std::stop_token，當(dāng)一個(gè)source被取消的時(shí)候，每個(gè)從它那派生出來(lái)的token都會(huì)收到通知，這可以上位替代我們之前使用的canceled_flag。

第三個(gè)是std::latch，你可以把它當(dāng)成go的sync.WaitGroup，它可以讓我們?cè)诠ぷ骶€程完成工作之后立即通知主線程，而不用調(diào)用join等待線程完全退出。

最后一個(gè)則是std::condition_variable_any新增了可以用stop_token中斷等待的方法，無(wú)需我們自己手動(dòng)notify，這樣可以盡量隱藏實(shí)現(xiàn)細(xì)節(jié)。

我們首先改造工作函數(shù)：

bool work(std::stop_token stoken, int id)
{
    std::mt19937 rng{std::random_device{}()};
    std::uniform_int_distribution<int> dist(0, 1);

    for (int i = 0; i < 10; ++i) {
        if (stoken.stop_requested()) {
            std::osyncstream{std::cout} << "worker: " << id << " canceled\n";
            return false;
        }
        if (dist(rng) == 0) {
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        } else {
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }

    std::osyncstream{std::cout} << "worker: " << id << " done\n";
    return true;
}

現(xiàn)在我們不依賴全局變量通知線程退出了。并且函數(shù)的返回值改成了bool，以表示線程是否是因?yàn)楸蝗∠顺龅?。我們通過(guò)stoken.stop_requested()判斷線程是否被取消。

接下來(lái)我們利用C++20改造主線程和超時(shí)控制：

int main(int argc, const char *argv[])
{
    if (argc != 2) {
        std::cerr << "wrong arg\n";
        return 1;
    }
    auto timeout = std::stoi(argv[1]);

    std::atomic<int> is_canceled{0};
    std::stop_source source;
    std::vector<std::jthread> workers;
    constexpr int worker_num = 3;
    std::latch wait_group{worker_num};
    workers.reserve(worker_num);
    for (int i = 0; i < worker_num; ++i) {
        workers.emplace_back([&](int id){
            if (!work(source.get_token(), id)) {
                is_canceled.store(1, std::memory_order_release);
            }
            wait_group.count_down(); // 三個(gè)線程都調(diào)用過(guò)之后，wait會(huì)解除阻塞
        }, i);
    }

    // 現(xiàn)在不需要專(zhuān)門(mén)detach了，效果是一樣的
    std::jthread timeout_control{
        [&source](std::stop_token stoken, std::chrono::milliseconds timeout){
            std::mutex timeout_lock;
            std::unique_lock u{timeout_lock};
            std::condition_variable_any timeout_cancel;
            if (!timeout_cancel.wait_for(u, stoken, timeout, [&stoken] { return stoken.stop_requested(); })) {
                std::osyncstream{std::cout} << "cancel all threads\n";
                source.request_stop();
            } else {
                std::osyncstream{std::cout} << "self was canceled by main thread\n";
            }
        }, std::chrono::milliseconds(timeout)
    };

    wait_group.wait(); // 等工作線程完成
    timeout_control.request_stop(); // 工作線程退出后立即取消超時(shí)控制線程，即使線程已經(jīng)執(zhí)行完成也能安全調(diào)用這個(gè)方法

    if (is_canceled.load(std::memory_order_acquire) != 1) {
        std::osyncstream{std::cout} << "All works done!\n";
    } else {
        std::osyncstream{std::cout} << "Tasks canceled\n";
    }
}

可以看到我們現(xiàn)在完全不使用全局變量了，而且線程的協(xié)作式取消是request_stop()和stop_requested()共同完成的，不會(huì)暴露太多實(shí)現(xiàn)細(xì)節(jié)。

需要注意的是!timeout_cancel.wait_for(u, stoken, timeout, [&stoken] { return stoken.stop_requested(); })這行代碼。這行代碼和普通條件變量的wait_for一樣，只不過(guò)多了一個(gè)參數(shù)stoken，這個(gè)函數(shù)會(huì)阻塞住當(dāng)前線程，直到超時(shí)、被notify或者stoken被取消。最后一個(gè)參數(shù)lambda的返回值會(huì)作為wait_for的返回值，這個(gè)lambda會(huì)在阻塞解除后立即調(diào)用，因此我們?cè)谶@個(gè)函數(shù)里檢查stoken是否被取消。如果stoken沒(méi)有被取消，說(shuō)明是因?yàn)槌瑫r(shí)導(dǎo)致的返回，因?yàn)槲覀冊(cè)谄渌胤秸{(diào)用notify，因此在這時(shí)我們需要取消其他工作線程。

新代碼不僅更簡(jiǎn)潔，而且沒(méi)有上一節(jié)那樣的時(shí)序問(wèn)題：

• 如果stoken的取消發(fā)生在整個(gè)wait之前。wait里的謂詞檢測(cè)會(huì)發(fā)現(xiàn)wait被取消
• stoken的取消發(fā)生在謂詞檢測(cè)之后、unlock+wait之前。condition_variable_any里有內(nèi)部鎖，調(diào)用notify_all（取消stoken會(huì)自動(dòng)調(diào)用）和unlock+wait之前都會(huì)先嘗試獲取這個(gè)鎖，unlock+wait在獲取鎖之后會(huì)檢查stoken是否被取消，所以如果notify先執(zhí)行，那么wait會(huì)檢測(cè)到stoken已取消；如果unlock+wait先執(zhí)行，那么notify一定是在wait之后執(zhí)行的，不存在丟失
• 取消發(fā)生在wait之后，這是最安全的情況，不會(huì)有喚醒丟失。

condition_variable_any內(nèi)部持有的鎖正好對(duì)應(yīng)我們上一節(jié)的TimeoutContext::lock，而stop_token對(duì)應(yīng)TimeoutContext::canceled，request_stop()則對(duì)應(yīng)了加鎖設(shè)置標(biāo)志并調(diào)用notify?，F(xiàn)在所有操作all in one，代碼在保證安全的同時(shí)將復(fù)雜的細(xì)節(jié)全部隱藏。

盡管仍有一定復(fù)雜性，但利用condition_variable_any已經(jīng)是比較理想的超時(shí)控制解決方案了。

condition_variable_any是如何配合stop_token的

最后還有一個(gè)小插曲，condition_variable_any是如何配合stop_token的。

C++20只給condition_variable_any添加了支持stop_token的接口，這是因?yàn)槠胀ǖ臈l件變量和其他的鎖幾乎都是系統(tǒng)接口的簡(jiǎn)單包裝，這些系統(tǒng)接口本身不支持中斷，因此也沒(méi)法為這些包裝接口添加支持，而condition_variable_any為了支持所有種類(lèi)的鎖，幾乎所有標(biāo)準(zhǔn)庫(kù)的實(shí)現(xiàn)都在自己內(nèi)部創(chuàng)建了一個(gè)普通的條件變量和內(nèi)部鎖，并基于這個(gè)條件變量重新實(shí)現(xiàn)了所有接口，因此給支持stop_token留下了余地。

我們以libcxx的實(shí)現(xiàn)為例：

// notify全都要先加內(nèi)部鎖
inline void condition_variable_any::notify_one() _NOEXCEPT {
  { lock_guard<mutex> __lx(*__mut_); }
  // notify時(shí)最好要解鎖，否則wait線程被喚醒后發(fā)現(xiàn)加不了鎖又會(huì)再次休眠，效率很低
  // 鎖只是為了讓notify的調(diào)用和unlock+wait互斥
  __cv_.notify_one();
}

inline void condition_variable_any::notify_all() _NOEXCEPT {
  { lock_guard<mutex> __lx(*__mut_); }
  __cv_.notify_all();
}

// wait_for最終調(diào)用的這個(gè)方法
template <class _Lock, class _Clock, class _Duration, class _Predicate>
bool condition_variable_any::wait_until(
    _Lock& __user_lock,
    stop_token __stoken,
    const chrono::time_point<_Clock, _Duration>& __abs_time,
    _Predicate __pred) {
    // 先檢查一次是否被取消，因?yàn)楹竺娴牟僮鞫己苤亓考?jí)會(huì)浪費(fèi)性能
  if (__stoken.stop_requested())
    return __pred();

  shared_ptr<mutex> __mut = __mut_;
  // 這行是關(guān)鍵，讓底層的普通條件變量可以從wait中蘇醒
  stop_callback __cb(__stoken, [this] { notify_all(); });

  while (true) {
    // 從wait中因?yàn)閚otify醒來(lái)會(huì)回到這里，調(diào)用最后的lambda來(lái)檢查是否應(yīng)該退出等待
    // 初次進(jìn)入循環(huán)也會(huì)檢查，以免進(jìn)行不必要的等待
    // 檢查其實(shí)在臨界區(qū)之外，所以其實(shí)最好我們得持有傳入的外部鎖才能完全避免靜態(tài)條件
    // 只不過(guò)恰好我們的謂詞檢查只檢查了stoken是否被取消，和下面臨界區(qū)內(nèi)的一樣，所以這地方的時(shí)序沒(méi)有影響。我們?cè)谥骶€程中也無(wú)需加外部鎖
    if (__pred())
      return true;

    // 先加內(nèi)部鎖，以免競(jìng)態(tài)條件出現(xiàn)，代碼本身的注釋已經(jīng)解釋清楚了，無(wú)需多言
    // We need to take the internal lock before checking stop_requested,
    // so that the notification cannot come in between the stop_requested
    // check and entering the wait.
    // Note that the stop_callback takes the same internal lock before notifying
    unique_lock<mutex> __internal_lock(*__mut);
    // 檢查stop_token是否被取消
    // notify拿到鎖先運(yùn)行的話循環(huán)就會(huì)在這里退出
    if (__stoken.stop_requested())
      break;

    // 解鎖用戶傳入的外部鎖，和普通的條件變量一樣
    __unlock_guard<_Lock> __unlock(__user_lock);
    unique_lock<mutex> __internal_lock2(
        std::move(__internal_lock)); // switch unlock order between __internal_lock and __user_lock

    // 內(nèi)部鎖的unlock+wait
    if (__cv_.wait_until(__internal_lock2, __abs_time) == cv_status::timeout)
      // 超時(shí)之后直接退出循環(huán)，否則回到循環(huán)開(kāi)始處
      break;
  } // __internal_lock2.unlock(), __user_lock.lock()
  // 還會(huì)調(diào)用謂詞，因此返回值總是能反應(yīng)stoken是否被取消
  return __pred();
}

template <class _Lock>
struct __unlock_guard {
  _Lock& __lock_;

  // 對(duì)象創(chuàng)建的時(shí)候解鎖，析構(gòu)的時(shí)候加鎖
  _LIBCPP_HIDE_FROM_ABI __unlock_guard(_Lock& __lock) : __lock_(__lock) { __lock_.unlock(); }

  _LIBCPP_HIDE_FROM_ABI ~__unlock_guard() _NOEXCEPT // turns exception to std::terminate
  {
    __lock_.lock();
  }

  __unlock_guard(const __unlock_guard&)            = delete;
  __unlock_guard& operator=(const __unlock_guard&) = delete;
};

核心在于stop_callback __cb(__stoken, [this] { notify_all(); });這一行，std::stop_callback可以把回調(diào)函數(shù)注冊(cè)給stop_token關(guān)聯(lián)的stop_source，當(dāng)source被要求停止時(shí)，注冊(cè)的回調(diào)會(huì)在調(diào)用request_stop()的那個(gè)線程執(zhí)行，回調(diào)可以注冊(cè)多個(gè)，調(diào)用順序是不確定的。

所以這行代碼等于在我們要求取消stoken的時(shí)候，順手調(diào)用notify_all()把底層的條件變量喚醒了，喚醒之后循環(huán)會(huì)檢查lambda和stoken，然后發(fā)現(xiàn)自己被取消從而從wait中退出。這就是wait可以被立即中斷的秘密。

不過(guò)如果token已經(jīng)取消，stop_callback是不生效的，但這也沒(méi)關(guān)系，因?yàn)樵谶M(jìn)入真正的等待支持，我們至少檢查了兩次stoken是否被取消，且第二次檢查是被鎖保護(hù)的，notify的callback不生效或者喚醒丟失了我們也能檢查到stoken已經(jīng)被取消，不會(huì)進(jìn)入無(wú)意義的等待。

上面的代碼也展現(xiàn)了condition_variable_any的問(wèn)題，相比直接轉(zhuǎn)發(fā)到系統(tǒng)接口的其他標(biāo)準(zhǔn)庫(kù)功能，它的性能通常要差一些，對(duì)于性能要求較高的場(chǎng)景必須謹(jǐn)慎使用。

總結(jié)

為了保證并發(fā)安全，上面的邏輯是有些繞的，但總結(jié)起來(lái)也就幾句話：

• 設(shè)置取消標(biāo)志需要和unlock+wait互斥，這樣超時(shí)控制線程才有機(jī)會(huì)檢測(cè)自己是否被取消。
• 超時(shí)控制線程中的取消標(biāo)志檢查（C++20前的謂詞匿名函數(shù)、C++20里的stoken.stop_requested）需要和unlock+wait在同一塊臨界區(qū)里，這樣才能保證進(jìn)入wait之前取消標(biāo)志被正確檢測(cè)到或者notify在wait之后才被調(diào)用。
• notify通常在設(shè)置完取消標(biāo)志之后執(zhí)行，但不需要在臨界區(qū)里，即使喚醒錯(cuò)過(guò)了我們也有保底措施。

只要想明白這三點(diǎn)你就掌握了這種模式，以后遇到超時(shí)取消或者防止喚醒丟失的場(chǎng)景時(shí)不至于兩眼一黑了。

這就是為什么要用現(xiàn)代C++，新特性有時(shí)候真的可以飛躍式提升開(kāi)發(fā)效率，雖說(shuō)這種程度和Golang相比還稱(chēng)不上優(yōu)雅，但心智負(fù)擔(dān)減輕了很多加班也沒(méi)那么累了。

但話說(shuō)回來(lái)，同樣的需求如果不是追求極致性能/只能用C++，我更樂(lè)意用Golang去實(shí)現(xiàn)。

以上就是源碼解析C++20新特性如何簡(jiǎn)化線程超時(shí)取消的詳細(xì)內(nèi)容，更多關(guān)于C++線程超時(shí)取消的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論