c++ - 防止一段代码在协程中并发执行

Question

我需要保护一段代码不被协程并发执行。在多线程环境中防止并发执行是使用 std::lock_guard 的简单问题。类模板。但是，我的协程是从单个线程调用的，因此该解决方案不适用。

以下是我要完成的(伪)代码:

future<http_response> send_req_async(http_request req) {
    while (true) {
        // Attempt to send an HTTP request
        auto const& access_token{ token_store::access_token() };
        auto const response{ co_await impl::send_req_async(req, access_token) };
        if (response.status_code() == http_code::ok) {
            co_return response;
        }

        // Attempt to refresh access token
        if (response.status_code() == http_code::unauthorized) {
            // The following scope needs to be guarded against concurrent execution.
            // To guard against concurrent execution from multiple threads I would use:
            // lock_guard<mutex> guard(refresh_token_mutex);
            if (access_token != token_store::access_token()) {
                continue;
            }
            auto const& token{ co_await refresh_token(token_store::refresh_token()) };
            token_store::save_access_token(token);
            // End of section that needs to be guarded.
        }
    }
}

该代码旨在允许并行发出多个请求，同时仅允许单个协程调用尝试刷新过期的访问 token 。理想情况下，解决方案应该在 token 刷新操作正在进行时暂停并发协程调用，并在之后自动恢复它(即多线程环境中 std::lock_guard 的相同语义)。

协程机制或 C++ 标准库中是否有任何内置的东西可以让我以干净的方式实现它，还是我必须自己动手？

---

注意:我使用的是 Visual Studio 2017 15.7.2，因此您可以假设完全支持 C++17 及其协程 TS 实现。

Answer 1

C++ 或标准库没有提供基础设施来获得所需的功能。然而，Coroutine TS提供构建 block 来实现 co_await-able async mutex。

总体思路是实现一个可等待对象，它会在评估 await_suspend 表达式时尝试获取合适的互斥体。如果无法获取锁，协程将被挂起并添加到等待者队列中，否则将立即继续执行(并持有锁)。

mutex 的 unlock 方法从队列中恢复等待者，除非等待者队列为空。

网络上有预建的解决方案。我和 Lewis Baker 一起去了 async_mutex出于多种原因实现:

• 没有外部或内部依赖。只需将编译单元和头文件放入您的项目中即可。
• 锁由协程拥有，而不是线程拥有。该实现允许协程在不同的线程上恢复。
• 这是一个无锁的实现。

此实现的使用与 std::lock_guard 的使用非常相似:

#include <cppcoro/async_mutex.hpp>

namespace {
    cppcoro::async_mutex refresh_mutex;
}

future<http_response> send_req_async(http_request req) {
    while (true) {
        // Attempt to send an HTTP request
        auto const& access_token{ token_store::access_token() };
        auto const response{ co_await impl::send_req_async(req, access_token) };
        if (response.status_code() == http_code::ok) {
            co_return response;
        }

        // Attempt to refresh access token
        if (response.status_code() == http_code::unauthorized) {
            // The following scope needs to be guarded against concurrent execution.
            auto const refresh_guard{ co_await refresh_mutex.scoped_lock_async() };
            if (access_token != token_store::access_token()) {
                continue;
            }
            auto const& token{ co_await refresh_token(token_store::refresh_token()) };
            token_store::save_access_token(token);
            // refresh_guard falls out of scope, unlocking the mutex.
            // If there are any suspended coroutines, the oldest one gets resumed.
        }
    }
}