在 PlayFab Services SDK 中进行异步调用

项目
07/29/2023

异步 API 指 API 快速返回，但会启动一个异步任务，待任务完成后才返回结果。

过去，游戏很少控制由哪个线程执行异步任务，以及在使用完成回调时由哪个线程返回结果。有些游戏的设计方式是让某个堆集部分仅供某一个线程访问，以避免线程同步的需要。如果未从游戏控件的线程调用完成回调，则使用异步任务的结果更新共享状态需要线程同步。

PlayFab Services SDK 公开一个异步 C API，该 API 在进行异步 API 调用时为开发人员提供直接线程控制，例如 PFAuthenticationLoginWithCustomIDAsync、PFDataGetFilesAsync 或 PFProfilesGetProfileAsync。

下面是调用 PFProfilesGetProfileAsync 的基本示例：

    XAsyncBlock* asyncBlock = new XAsyncBlock();
    asyncBlock->queue = GlobalState()->queue;
    asyncBlock->context = nullptr;
    asyncBlock->callback = [](XAsyncBlock* asyncBlock)
    {
        std::unique_ptr<XAsyncBlock> asyncBlockPtr{ asyncBlock }; // take ownership of XAsyncBlock
        
        size_t bufferSize;
        HRESULT hr = PFProfilesGetProfileGetResultSize(asyncBlock, &bufferSize);
        if (SUCCEEDED(hr))
        {
            std::vector<char> getProfileResultBuffer(bufferSize);
            PFProfilesGetEntityProfileResponse* getProfileResponseResult{ nullptr };
            PFProfilesGetProfileGetResult(asyncBlock, getProfileResultBuffer.size(), getProfileResultBuffer.data(), &getProfileResponseResult, nullptr);
        }
    };

    PFProfilesGetEntityProfileRequest profileRequest{};
    HRESULT hr = PFProfilesGetProfileAsync(GlobalState()->entityHandle, &profileRequest, asyncBlock);

若要了解此调用模式，需要了解如何使用 XAsyncBlock 和 XTaskQueueHandle。

XAsyncBlock 包含与异步任务和完成回调相关的所有信息。

XTaskQueueHandle 可用于确定执行异步任务的线程以及哪些线程调用 XAsyncBlock的完成回调。

XAsyncBlock

我们细看一下 XAsyncBlock。它是一个定义如下的结构：

typedef struct XAsyncBlock
{
    /// <summary>
    /// The queue to queue the call on
    /// </summary>
    XTaskQueueHandle queue;

    /// <summary>
    /// Optional context pointer to pass to the callback
    /// </summary>
    void* context;

    /// <summary>
    /// Optional callback that will be invoked when the call completes
    /// </summary>
    XAsyncCompletionRoutine* callback;

    /// <summary>
    /// Internal use only
    /// </summary>
    unsigned char internal[sizeof(void*) * 4];
};

XAsyncBlock 中包含：

队列 - XTaskQueueHandle ，它是一个句柄，表示有关在何处运行工作的信息。如果未设置此参数，则使用默认队列。
context - 用于向回调函数传递数据。
callback - 一个将在异步工作完成后调用的可选回调函数。如果未指定回调，则可以等待 XAsyncBlock 完成 XAsyncGetStatus，然后获取结果。

应在堆上为每个异步调用创建新的 XAsyncBlock。 XAsyncBlock 必须一直存在，直到调用 XAsyncBlock 的完成回调，然后才能将其删除。

重要事项：

XAsyncBlock 必须一直保留在内存中，直到异步任务完成。如果动态分配，则可以在 XAsyncBlock 的完成回调中删除它。

等待异步任务

可以通过两种不同的方式判断异步任务是否已完成：

调用 XAsyncBlock 的完成回调。
通过 true 值调用 XAsyncGetStatus 一直等到它完成。

使用 XAsyncGetStatus，异步任务被视为在执行 XAsyncBlock的完成回调后完成，但 XAsyncBlock的完成回调是可选的。

一旦异步任务完成，就可以获取结果。

获取异步任务的结果

为了获得结果，大多数异步 API 函数都有一个相应的 Result 函数来接收异步调用的结果。

在示例代码中， PFProfilesGetProfileAsync 具有相应的 PFProfilesGetProfileGetResult 函数。可以使用此函数检索函数结果并相应执行操作。

有关检索结果的完整详细信息，请参阅每个异步 API 函数的文档。

XTaskQueueHandle

XTaskQueueHandle 可用于确定执行异步任务的线程以及哪些线程调用 XAsyncBlock的完成回调。

可以通过设置调度模式控制由哪个线程执行这些操作。有以下三种调度模式：

手动 - 不会自动调度手动队列。开发者负责将它们调度到所需的任何线程。此模式可用于将异步调用的工作端或回调端分配到特定线程。
线程池 - 调度使用线程池。线程池并行调取调用，在线程池线程可用时依次从队列中提取要执行的调用。 线程池 是最容易使用的，但提供对所用线程的最小控制。
序列化线程池 - 调度使用线程池。线程池串行调取调用，在单个线程池线程可用时依次从队列中提取要执行的调用。
Immediate - 立即在从中提交排队的工作的线程上调度此工作。

若要创建新的 XTaskQueueHandle，需要调用 XTaskQueueCreate。例如:

STDAPI XTaskQueueCreate(
    _In_ XTaskQueueDispatchMode workDispatchMode,
    _In_ XTaskQueueDispatchMode completionDispatchMode,
    _Out_ XTaskQueueHandle* queue
    ) noexcept;

此函数采用两个 XTaskQueueDispatchMode 参数。 XTaskQueueDispatchMode 有三个可能的值：

/// <summary>
/// Describes how task queue callbacks are processed.
/// </summary>
enum class XTaskQueueDispatchMode : uint32_t
{
    /// <summary>
    /// Callbacks are invoked manually by XTaskQueueDispatch
    /// </summary>
    Manual,

    /// <summary>
    /// Callbacks are queued to the system thread pool and will
    /// be processed in order by the thread pool across multiple thread
    /// pool threads.
    /// </summary>
    ThreadPool,
    
    /// <summary>
    /// Callbacks are queued to the system thread pool and
    /// will be processed one at a time.
    /// </summary>
    SerializedThreadPool,
    
    /// <summary>
    /// Callbacks are not queued at all but are dispatched
    /// immediately by the thread that submits them.
    /// </summary>
    Immediate
};

workDispatchMode 确定处理异步工作的线程的调度模式。 completionDispatchMode 确定线程的调度模式，该模式处理异步操作的完成。

创建 XTaskQueueHandle后，只需将其添加到 XAsyncBlock 即可控制工作和完成函数的线程处理。使用完 XTaskQueueHandle 时，通常当游戏结束时，可以使用 XTaskQueueCloseHandle 关闭它：

STDAPI_(void) XTaskQueueCloseHandle(
    _In_ XTaskQueueHandle queue
    ) noexcept;

调用示例：

XTaskQueueCloseHandle(queue);

手动调度 XTaskQueueHandle

如果对 XTaskQueueHandle 工作或完成队列使用了手动队列调度模式，则需要手动调度。假设创建了 XTaskQueueHandle，其中工作队列和完成队列都设置为手动调度，如下所示：

XTaskQueueHandle queue = nullptr;
HRESULT hr = XTaskQueueCreate(
    XTaskQueueDispatchMode::Manual,
    XTaskQueueDispatchMode::Manual,
    &queue);

若要调度分配 有 XTaskQueueDispatchMode::Manual 的工作，请调用 XTaskQueueDispatch 函数。

STDAPI_(bool) XTaskQueueDispatch(
    _In_ XTaskQueueHandle queue,
    _In_ XTaskQueuePort port,
    _In_ uint32_t timeoutInMs
    ) noexcept;

调用示例：

HRESULT hr = XTaskQueueDispatch(queue, XTaskQueuePort::Completion, 0);

queue - 在哪个队列上调度工作。
port - XTaskQueuePort 枚举的实例。
timeoutInMs - 表示毫秒超时的 uint32_t 值。

XTaskQueuePort 枚举定义了两种回调类型：

/// <summary>
/// Declares which port of a task queue to dispatch or submit
/// callbacks to.
/// </summary>
enum class XTaskQueuePort : uint32_t
{
    /// <summary>
    /// Work callbacks
    /// </summary>
    Work,

    /// <summary>
    /// Completion callbacks after work is done
    /// </summary>
    Completion
};

何时调用 XTaskQueueDispatch

为了检查队列何时收到新项，可以调用 XTaskQueueRegisterMonitor 来设置事件处理程序，让代码知道工作或完成已准备好调度。

STDAPI XTaskQueueRegisterMonitor(
    _In_ XTaskQueueHandle queue,
    _In_opt_ void* callbackContext,
    _In_ XTaskQueueMonitorCallback* callback,
    _Out_ XTaskQueueRegistrationToken* token
    ) noexcept;

XTaskQueueRegisterMonitor 使用以下参数：

队列 - 要为其提交回调的异步队列。
callbackContext - 一个指向应传递给提交回调的数据的指针。
回调 - 将新回调提交到队列时调用的函数。
令牌 - 稍后调用 XTaskQueueUnregisterMonitor 中用于删除回调的令牌。

例如，下面是 XTaskQueueRegisterMonitor的调用：

XTaskQueueRegisterMonitor(queue, nullptr, HandleAsyncQueueCallback, &m_callbackToken);

相应的 XTaskQueueMonitorCallback 回调可能实现如下：

void CALLBACK HandleAsyncQueueCallback(
    _In_opt_ void* context,
    _In_ XTaskQueueHandle queue,
    _In_ XTaskQueuePort port)
{
    switch (port)
    {
    case XTaskQueuePort::Work:
        {
            std::lock_guard<std::mutex> lock(g_workReadyMutex);
            g_workReady = true;
        }

        g_workReadyConditionVariable.notify_one(); // (std::condition_variable)
        break;
    }
}

然后，在后台线程上，可以侦听此条件变量以唤醒和调用 XTaskQueueDispatch。

void BackgroundWorkThreadProc(XTaskQueueHandle queue)
{
    while (true)
    {
        {
            std::unique_lock<std::mutex> cvLock(g_workReadyMutex);
            g_workReadyConditionVariable.wait(cvLock, [] { return g_workReady; });

            if (g_stopBackgroundWork)
            {
                break;
            }

            g_workReady = false;
        }

        bool workFound = false;
        do
        {
            workFound = XTaskQueueDispatch(queue, XTaskQueuePort::Work, 0);
        } while (workFound);
    }
    
    XTaskQueueCloseHandle(queue);
}

参考

API 参考文档

通过