编写异步方法
本主题介绍如何在 Microsoft Media Foundation 中实现异步方法。
异步方法在媒体基础管道中无处不在。 使用异步方法可以更轻松地在多个线程之间分配工作。 异步执行 I/O 尤其重要,以便从文件或网络读取不会阻止管道的其余部分。
如果要编写媒体源或媒体接收器,正确处理异步操作至关重要,因为组件的性能对整个管道都有影响。
注意
默认情况下,Media Foundation 转换 (MRT) 同步方法。
异步操作的工作队列
在 Media Foundation 中,异步回调方法和工作队列之间存在密切的关系。 工作队列是用于将工作从调用方的线程移动到工作线程的抽象。 若要对工作队列执行工作,请执行以下操作:
实现 IMFAsyncCallback 接口。
调用 MFCreateAsyncResult 来创建 结果 对象。 结果对象公开 IMFAsyncResult。 结果对象包含三个指针:
- 指向调用方 IMFAsyncCallback 接口的指针。
- 指向状态对象的可选指针。 如果指定,则状态对象必须实现 IUnknown。
- 指向私有对象的可选指针。 如果指定,此对象还必须实现 IUnknown。
最后两个指针可以为 NULL。 否则,请使用它们来保存有关异步操作的信息。
调用 MFPutWorkItemEx 以排队到工作项。
工作队列线程调用 IMFAsyncCallback::Invoke 方法。
在 Invoke 方法中执行工作。 此方法的 pAsyncResult 参数是步骤 2 中的 IMFAsyncResult 指针。 使用此指针获取状态对象和私有对象:
- 若要获取状态对象,请调用 IMFAsyncResult::GetState。
- 若要获取专用对象,请调用 IMFAsyncResult::GetObject。
或者,可以通过调用 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 的实现必须在另一个线程上执行实际工作。 这是工作队列进入图片的位置。 在后续步骤中, 调用方 是调用 BeginX 和 EndX 的代码。 这可能是应用程序或媒体基础管道。 组件是实现 BeginX 和 EndX 的代码。
- 调用方调用 Begin...,传入指向调用方 IMFAsyncCallback 接口的指针。
- 组件创建新的异步结果对象。 此对象存储调用方回调接口和状态对象。 通常,它还存储组件完成操作所需的任何专用状态信息。 下一个关系图中,此步骤的结果对象标记为“Result 1”。
- 组件创建第二个结果对象。 此结果对象存储两个指针:第一个结果对象和被调用方的回调接口。 此结果对象在下一个关系图中标记为“Result 2”。
- 组件调用 MFPutWorkItemEx 以将新工作项排队。
- 在 Invoke 方法中,组件执行异步工作。
- 组件调用 MFInvokeCallback 来调用调用方回调方法。
- 调用方调用 EndX 方法。
异步方法示例
为了说明此讨论,我们将使用一个经过尝试的示例。 请考虑使用异步方法来计算平方根:
HRESULT BeginSquareRoot(double x, IMFAsyncCallback *pCB, IUnknown *pState);
HRESULT EndSquareRoot(IMFAsyncResult *pResult, double *pVal);
的 BeginSquareRoot
x 参数是将计算其平方根的值。 平方根在 的 EndSquareRoot
pVal 参数中返回。
下面是实现这两种方法的类的声明:
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;
}
此代码将执行以下操作:
- 创建 类的新实例
AsyncOp
来保存 x 的值。 - 调用 MFCreateAsyncResult 来创建结果对象。 此对象包含多个指针:
- 指向调用方 IMFAsyncCallback 接口的指针。
- 指向调用方状态对象的指针 (pState) 。
- 指向
AsyncOp
对象的指针。
- 调用 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 的详细信息,请参阅 受保护的媒体路径。
相关主题