编写异步方法

本主题介绍如何在 Microsoft Media Foundation 中实现异步方法。

异步方法在媒体基础管道中无处不在。 使用异步方法可以更轻松地在多个线程之间分配工作。 异步执行 I/O 尤其重要,以便从文件或网络读取不会阻止管道的其余部分。

如果要编写媒体源或媒体接收器,正确处理异步操作至关重要,因为组件的性能对整个管道都有影响。

注意

默认情况下,Media Foundation 转换 (MRT) 同步方法。

 

异步操作的工作队列

在 Media Foundation 中,异步回调方法和工作队列之间存在密切的关系。 工作队列是用于将工作从调用方的线程移动到工作线程的抽象。 若要对工作队列执行工作,请执行以下操作:

  1. 实现 IMFAsyncCallback 接口。

  2. 调用 MFCreateAsyncResult 来创建 结果 对象。 结果对象公开 IMFAsyncResult。 结果对象包含三个指针:

    • 指向调用方 IMFAsyncCallback 接口的指针。
    • 指向状态对象的可选指针。 如果指定,则状态对象必须实现 IUnknown
    • 指向私有对象的可选指针。 如果指定,此对象还必须实现 IUnknown

    最后两个指针可以为 NULL。 否则,请使用它们来保存有关异步操作的信息。

  3. 调用 MFPutWorkItemEx 以排队到工作项。

  4. 工作队列线程调用 IMFAsyncCallback::Invoke 方法。

  5. Invoke 方法中执行工作。 此方法的 pAsyncResult 参数是步骤 2 中的 IMFAsyncResult 指针。 使用此指针获取状态对象和私有对象:

或者,可以通过调用 MFPutWorkItem 函数来组合步骤 2 和 3。 在内部,此函数调用 MFCreateAsyncResult 来创建结果对象。

下图显示了调用方、结果对象、状态对象和私有对象之间的关系。

显示异步结果对象的示意图

以下序列图显示了对象如何对工作项进行排队。 当工作队列线程调用 Invoke 时,对象在该线程上执行异步操作。

显示对象如何对工作项进行排队的示意图

请务必记住 ,Invoke 是从工作队列拥有的线程调用的。 Invoke 的实现必须是线程安全的。 此外,如果使用平台工作队列 (MFASYNC_CALLBACK_QUEUE_STANDARD) ,则切勿阻止线程至关重要,因为这会阻止整个 Media Foundation 管道处理数据。 如果需要执行将阻止或需要很长时间才能完成的操作,请使用专用工作队列。 若要创建专用工作队列,请调用 MFAllocateWorkQueue。 执行 I/O 操作的任何管道组件都应避免出于相同原因阻止 I/O 调用。 IMFByteStream 接口为异步文件 I/O 提供了有用的抽象。

正在实现 Begin.../End...模式

调用异步方法中所述,Media Foundation 中的异步方法通常使用 Begin.../结束。。。。 模式。 在此模式中,异步操作使用两个签名方法,如下所示:

// Starts the asynchronous operation.
HRESULT BeginX(IMFAsyncCallback *pCallback, IUnknown *punkState);

// Completes the asynchronous operation. 
// Call this method from inside the caller's Invoke method.
HRESULT EndX(IMFAsyncResult *pResult);

若要使 方法真正异步, BeginX 的实现必须在另一个线程上执行实际工作。 这是工作队列进入图片的位置。 在后续步骤中, 调用方 是调用 BeginXEndX 的代码。 这可能是应用程序或媒体基础管道。 组件是实现 BeginXEndX 的代码。

  1. 调用方调用 Begin...,传入指向调用方 IMFAsyncCallback 接口的指针。
  2. 组件创建新的异步结果对象。 此对象存储调用方回调接口和状态对象。 通常,它还存储组件完成操作所需的任何专用状态信息。 下一个关系图中,此步骤的结果对象标记为“Result 1”。
  3. 组件创建第二个结果对象。 此结果对象存储两个指针:第一个结果对象和被调用方的回调接口。 此结果对象在下一个关系图中标记为“Result 2”。
  4. 组件调用 MFPutWorkItemEx 以将新工作项排队。
  5. Invoke 方法中,组件执行异步工作。
  6. 组件调用 MFInvokeCallback 来调用调用方回调方法。
  7. 调用方调用 EndX 方法。

显示对象如何实现开始/结束模式的示意图

异步方法示例

为了说明此讨论,我们将使用一个经过尝试的示例。 请考虑使用异步方法来计算平方根:

    HRESULT BeginSquareRoot(double x, IMFAsyncCallback *pCB, IUnknown *pState);
    HRESULT EndSquareRoot(IMFAsyncResult *pResult, double *pVal);

BeginSquareRootx 参数是将计算其平方根的值。 平方根在 的 EndSquareRootpVal 参数中返回。

下面是实现这两种方法的类的声明:

class SqrRoot : public IMFAsyncCallback
{
    LONG    m_cRef;
    double  m_sqrt;

    HRESULT DoCalculateSquareRoot(AsyncOp *pOp);

public:

    SqrRoot() : m_cRef(1)
    {

    }

    HRESULT BeginSquareRoot(double x, IMFAsyncCallback *pCB, IUnknown *pState);
    HRESULT EndSquareRoot(IMFAsyncResult *pResult, double *pVal);

    // IUnknown methods.
    STDMETHODIMP QueryInterface(REFIID riid, void **ppv)
    {
        static const QITAB qit[] = 
        {
            QITABENT(SqrRoot, IMFAsyncCallback),
            { 0 }
        };
        return QISearch(this, qit, riid, ppv);
    }

    STDMETHODIMP_(ULONG) AddRef()
    {
        return InterlockedIncrement(&m_cRef);
    }

    STDMETHODIMP_(ULONG) Release()
    {
        LONG cRef = InterlockedDecrement(&m_cRef);
        if (cRef == 0)
        {
            delete this;
        }
        return cRef;
    }

    // IMFAsyncCallback methods.

    STDMETHODIMP GetParameters(DWORD* pdwFlags, DWORD* pdwQueue)
    {
        // Implementation of this method is optional.
        return E_NOTIMPL;  
    }
    // Invoke is where the work is performed.
    STDMETHODIMP Invoke(IMFAsyncResult* pResult);
};

SqrRoot 实现 IMFAsyncCallback ,以便它可以将平方根操作放在工作队列上。 方法 DoCalculateSquareRoot 是计算平方根的私有类方法。 此方法将从工作队列线程调用。

首先,我们需要一种方法来存储 x 的值,以便在工作队列线程调用 SqrRoot::Invoke时检索它。 下面是存储信息的简单类:

class AsyncOp : public IUnknown
{
    LONG    m_cRef;

public:

    double  m_value;

    AsyncOp(double val) : m_cRef(1), m_value(val) { }

    STDMETHODIMP QueryInterface(REFIID riid, void **ppv)
    {
        static const QITAB qit[] = 
        {
            QITABENT(AsyncOp, IUnknown),
            { 0 }
        };
        return QISearch(this, qit, riid, ppv);
    }

    STDMETHODIMP_(ULONG) AddRef()
    {
        return InterlockedIncrement(&m_cRef);
    }

    STDMETHODIMP_(ULONG) Release()
    {
        LONG cRef = InterlockedDecrement(&m_cRef);
        if (cRef == 0)
        {
            delete this;
        }
        return cRef;
    }
};

此类实现 IUnknown ,以便可以将其存储在结果对象中。

以下代码实现 BeginSquareRoot 方法:

HRESULT SqrRoot::BeginSquareRoot(double x, IMFAsyncCallback *pCB, IUnknown *pState)
{
    AsyncOp *pOp = new (std::nothrow) AsyncOp(x);
    if (pOp == NULL)
    {
        return E_OUTOFMEMORY;
    }

    IMFAsyncResult *pResult = NULL;

    // Create the inner result object. This object contains pointers to:
    // 
    //   1. The caller's callback interface and state object. 
    //   2. The AsyncOp object, which contains the operation data.
    //

    HRESULT hr = MFCreateAsyncResult(pOp, pCB, pState, &pResult);

    if (SUCCEEDED(hr))
    {
        // Queue a work item. The work item contains pointers to:
        // 
        // 1. The callback interface of the SqrRoot object.
        // 2. The inner result object.

        hr = MFPutWorkItem(MFASYNC_CALLBACK_QUEUE_STANDARD, this, pResult);

        pResult->Release();
    }

    return hr;
}

此代码将执行以下操作:

  1. 创建 类的新实例 AsyncOp 来保存 x 的值。
  2. 调用 MFCreateAsyncResult 来创建结果对象。 此对象包含多个指针:
    • 指向调用方 IMFAsyncCallback 接口的指针。
    • 指向调用方状态对象的指针 (pState) 。
    • 指向 AsyncOp 对象的指针。
  3. 调用 MFPutWorkItem 以将新工作项排队。 此调用隐式创建一个包含以下指针的外部结果对象:
    • 指向 SqrRoot 对象的 IMFAsyncCallback 接口的指针。
    • 指向步骤 2 中内部结果对象的指针。

以下代码实现 SqrRoot::Invoke 方法:

// Invoke is called by the work queue. This is where the object performs the
// asynchronous operation.

STDMETHODIMP SqrRoot::Invoke(IMFAsyncResult* pResult)
{
    HRESULT hr = S_OK;

    IUnknown *pState = NULL;
    IUnknown *pUnk = NULL;
    IMFAsyncResult *pCallerResult = NULL;

    AsyncOp *pOp = NULL; 

    // Get the asynchronous result object for the application callback. 

    hr = pResult->GetState(&pState);
    if (FAILED(hr))
    {
        goto done;
    }

    hr = pState->QueryInterface(IID_PPV_ARGS(&pCallerResult));
    if (FAILED(hr))
    {
        goto done;
    }

    // Get the object that holds the state information for the asynchronous method.
    hr = pCallerResult->GetObject(&pUnk);
    if (FAILED(hr))
    {
        goto done;
    }

    pOp = static_cast<AsyncOp*>(pUnk);

    // Do the work.

    hr = DoCalculateSquareRoot(pOp);

done:
    // Signal the application.
    if (pCallerResult)
    {
        pCallerResult->SetStatus(hr);
        MFInvokeCallback(pCallerResult);
    }

    SafeRelease(&pState);
    SafeRelease(&pUnk);
    SafeRelease(&pCallerResult);
    return S_OK;
}

此方法获取内部结果对象和 AsyncOp 对象。 然后, AsyncOp 它将 对象传递给 DoCalculateSquareRoot。 最后,它调用 IMFAsyncResult::SetStatus 来设置状态代码,并调用 MFInvokeCallback 来调用调用方回调方法。

方法 DoCalculateSquareRoot 完全按照预期执行:

HRESULT SqrRoot::DoCalculateSquareRoot(AsyncOp *pOp)
{
    pOp->m_value = sqrt(pOp->m_value);

    return S_OK;
}

调用调用方回调方法时,调用方负责调用 End... 方法,在本例中为 EndSquareRoot。 调用 EndSquareRoot 方检索异步操作的结果的方式,在此示例中为计算平方根。 此信息存储在结果对象中:

HRESULT SqrRoot::EndSquareRoot(IMFAsyncResult *pResult, double *pVal)
{
    *pVal = 0;

    IUnknown *pUnk = NULL;

    HRESULT hr = pResult->GetStatus();

    if (FAILED(hr))
    {
        goto done;
    }

    hr = pResult->GetObject(&pUnk);
    if (FAILED(hr))
    {
        goto done;
    }

    AsyncOp *pOp = static_cast<AsyncOp*>(pUnk);

    // Get the result.
    *pVal = pOp->m_value;

done:
    SafeRelease(&pUnk);
    return hr;
}

操作队列

到目前为止,一直默契地假定,无论对象的当前状态如何,都可以随时执行异步操作。 例如,如果应用程序调用相同方法的早期调用 BeginSquareRoot 仍处于挂起状态,则会发生什么情况。 类 SqrRoot 可能会在完成上一个工作项之前对新工作项进行排队。 但是,无法保证工作队列能够序列化工作项。 回想一下,一个工作队列可以使用多个线程来调度工作项。 在多线程环境中,可以在上一个工作项完成之前调用工作项。 如果上下文切换恰好在调用回调之前发生,则甚至可以无序调用工作项。

出于此原因,对象负责根据需要对自身执行操作进行序列化。 换句话说,如果对象要求操作 A 在操作 B 启动之前完成操作则在操作 A 完成之前,该对象不得将 B 的工作项排队。 对象可以通过有自己的挂起操作队列来满足此要求。 在 对象上调用异步方法时,对象会将请求置于其自己的队列中。 完成每个异步操作后,对象将从队列中拉取下一个请求。 MPEG1Source 示例演示了如何实现此类队列的示例。

单个方法可能涉及多个异步操作,尤其是在使用 I/O 调用时。 实现异步方法时,请仔细考虑序列化要求。 例如,在以前的 I/O 请求仍处于挂起状态时,对象启动新操作是否有效? 如果新操作更改了对象的内部状态,当以前的 I/O 请求完成并返回现在可能过时的数据时会发生什么情况? 良好的状态图有助于识别有效的状态转换。

跨线程和跨进程注意事项

工作队列不使用 COM 封送处理跨线程边界封送接口指针。 因此,即使对象注册为单元线程或应用程序线程已进入单线程单元 (STA) ,也会从其他线程调用 IMFAsyncCallback 回调。 在任何情况下,所有 Media Foundation 管道组件都应使用“两者”线程模型。

Media Foundation 中的某些接口定义某些异步方法的远程版本。 当跨进程边界调用其中一种方法时,Media Foundation 代理/存根 DLL 将调用方法的远程版本,该方法执行方法参数的自定义封送处理。 在远程进程中,存根将调用转换回 对象上的本地方法。 此过程对应用程序和远程对象都是透明的。 这些自定义封送处理方法主要用于在受保护媒体路径 (PMP) 加载的对象。 有关 PMP 的详细信息,请参阅 受保护的媒体路径

异步回调方法

工作队列