directshow之filter开发---ASYNC例程解析

三、ASYNC例程解析

ASYNC例程中有三个工程：asynbase(Base文件夹)、async（Filter文件夹）、memfile（Memfile文件夹）。
Asynbase工程是一个静态库工程，完成了基类、接口的定义和部分功能函数的实现。该工程共定义了CAsyncStream接口类，该类是一个纯虚类，所有接口函数都有用户实现，该类定义了操作文件（读文件等）的接口（由于该工程是一次性将文件的全部内容读到内从中，然后对内存进行操作的，CAsyncStream实际是操作内存的）；CAsyncRequest类，该类封装了对文件读取的需求；CAsyncIo类，该类来管理、完成CAsyncRequest需求，该类还实现工作线程的创建；CAsyncOutputPin类，该类继承自IAsyncReader、CBasePin接口，来实现拉模式source
filter的输出pin；CAsyncReader类，该类继承自CBasefilter类，作为拉模式source filter的基类。

在拉模式下，source filter的下游filter用source filter输出pin上的IAsyncReader接口从sourcefilter请求数据。因此可知拉模式的source
filter上的输出pin需要实现IAsyncReader接口（CAsyncOutputPin类的一个父类便是IAsyncReader），而Asynbase工程中的CAsyncOutputPin实现了IAsyncReader接口的各个方法如：WaiteForNext、Request、SyncRead、Length、BeginFlush、EndFlush等。更多关于IAsyncReader接口说明可以参见MSND。

虽然CAsyncOutputPin实现了IAsyncReader接口函数，但IAsyncReader接口函数真正实现是在CAsyncIo，如

STDMETHODIMPCAsyncOutputPin::Length(LONGLONG* pTotal, LONGLONG* pAvailable)

{

// m_pIo为CAsyncIo类的指针

HRESULThr=m_pIo-> Length(pTotal, pAvailable);

return hr;

}

Async工程是一个拉模式的sourcefilter，其基于Asynbase工程中实现的类实现。该工程有两个类，CMemStream继承自CAsyncStream（Asynbase工程定义的接口类），实现了CAsyncStream的各个接口函数；CAsyncFilter继承自CAsyncReader、IFileSourceFilter,其中CAsyncReader（Asynbase工程定义的filter类）。

1、
Async工程怎么没有有关pin类的定义？

每一个filter都至少要有一个pin，Async工程作为一个拉模式的sourcefilter，其必

须具有一个输出pin，否则ASync工程生成的pin将是一个没有用的filter。Async工程生成的源filter的输出pin在哪里呢？

CAsyncFilter继承自CAsyncReader，CAsyncReader类中申明了CAsyncIo、CAsyncOutputPin的变量，而CAsyncOutputPin正是一个继承了IAsyncReader接口的输出pin类，在CAsyncReader的构造函数中实例化了CAsyncOutputPin变量，这样在构造CAsyncFilter的时候会构造其父类CAsyncReader，CAsyncReader构造的时候又构造了CAsyncOutputPin。因此CAsyncFilter边拥有了一个输出pin（CAsyncOutputPin）。

CAsyncFilter(LPUNKNOWN pUnk, HRESULT *phr) :

CAsyncReader(NAME("MemReader"), pUnk, &m_Stream, phr),// CAsyncReader构造

m_pFileName(NULL),

m_pbData(NULL)

{ }

CAsyncReader::CAsyncReader(TCHAR *pName,LPUNKNOWN pUnk,CAsyncStream*pStream,

HRESULT *phr):CBaseFilter(pName,pUnk,&m_csFilter,CLSID_AsyncReader,NULL),

m_OutputPin(phr,this,&m_Io,&m_csFilter),m_Io(pStream)//CAsyncOutputPin构造

{
}

2、
作为拉模式的sourcefilter，数据源是怎么被下游filter拉出去的呢？

拉模式的sourcefilter后的filter一般是一个spliterfilter或者parser filter

, spliter、parser负责从sourcefilter拿到数据，解析出不同的数据，然后推送给其下游的filter。将拉模式的sourcefilter和spliter（parser）filter看做一个整体，那么它们就组成了一个推模式的sourcefilter。由此可见负责拉数据的就是spliter（parser）filter的输入pin。spliter（parser）filter的输入pin在连接方式上和其他一般的pin是有所不同的，spliter（parser）filter的输入pin在连接时要查询的到IAsyncReader接口。DirectShow的Baseclasses工程实现了一个可用于spliter（parser）filter的输入pin的基类CPullPin。

在CPullPin（继承自CAMThread）类中定义了一个IAsyncReader接口的指针变量，在其连接函数Connect给IAsyncReader变量赋值。

Async工程生成filter的输出pin实现了IAsyncReader接口，所以当CPullPin和Async工程生成filter的输出pin成功连接在一起时，CPullPin就可以通过调用IAsyncReader接口，从工程生成filter的输出pin拉取数据。这里假设CPullPin为spliter（parser）filter的输入pin和Async工程生成filter的输出pin成功连接在了一起，来分析拉模式下sourcefilter的数据源怎么被拉取的？

（1）spliter（parser）filter拉数据的请求怎么传给上游sourcefilter的呢？

CPullPin的开始运行时在Active函数中会开启一个线程（调用其父类CAMThread的create函数），通过上一章对Ball的分析可以线程函数最终会调用CPullPin的ThreadProc函数，在ThreadProc函数用调用process函数来实现具体拉取数据的任务。Process函数中调用QueueSample函数来发送拉取数据的请求：

HRESULT CPullPin::QueueSample(REFERENCE_TIME& tCurrent,REFERENCE_TIMEtAlignStop,

BOOL bDiscontinuity)

{

IMediaSample* pSample;

//分配一个媒体样本

HRESULT hr =m_pAlloc->GetBuffer(&pSample, NULL, NULL, 0);

if (FAILED(hr)) {

return hr;

}

LONGLONG tStopThis = tCurrent +(pSample->GetSize() * UNITS);

if (tStopThis > tAlignStop) {

tStopThis = tAlignStop;

}

pSample->SetTime(&tCurrent, &tStopThis);

tCurrent = tStopThis;

pSample->SetDiscontinuity(bDiscontinuity);

//调用IAsyncReader接口向sourcefilter发送请求，m_pReader是一个IAsyncReader指针变//量


hr = m_pReader->Request(


pSample,


0);

if (FAILED(hr)) {

pSample->Release();

CleanupCancelled();

OnError(hr);

}

return hr;

}

在拉模式当spliter输入pin调用IasyncReader的相应接口时，也就调用了sourcefilter的输出pin（CAsyncOutputPin）上实现的IasyncReader接口。CAsyncOutputPin上的Request函数，在该函数中有IMediaSample参数的到对其方式和计算请求读取的长度，然后调用CAsyncIo函数的相应函数将请求保存到求情队列中。

STDMETHODIMP CAsyncOutputPin::Request(IMediaSample* pSample,DWORD dwUser)

{

REFERENCE_TIME tStart, tStop;

HRESULT hr =pSample->GetTime(&tStart, &tStop);

if (FAILED(hr)) {

return hr;

}

//UNITS?10^7

LONGLONG llPos = tStart / UNITS;

//计算该请求要读取的数据长度
UNITS为10^7

LONG lLength = (LONG) ((tStop - tStart) / UNITS);

LONGLONG llTotal;

LONGLONG llAvailable;

//m_pIo为CAsyncIo的指针变量

//得到资源数据的总长度和可读长度

hr = m_pIo->Length(&llTotal, &llAvailable);

if (llPos + lLength >llTotal) {

// the end needs to bealigned, but may have been aligned

// on a coarser alignment.

LONG lAlign;

//获取资源数据的对齐方式

m_pIo->Alignment(&lAlign);

//根据资源的对齐方式，总长度和可读取长度计算能否满足请求读取的长度，不满足时//做相应修正

llTotal = (llTotal + lAlign-1) & ~(lAlign-1);

if (llPos + lLength >llTotal) {

lLength = (LONG) (llTotal - llPos);

// must be reducingthis!

ASSERT((llTotal * UNITS)<= tStop);

tStop = llTotal * UNITS;

pSample->SetTime(&tStart, &tStop);

}

}

BYTE* pBuffer;

hr = pSample->GetPointer(&pBuffer);

if (FAILED(hr)) {

return hr;

}

//调用CAsyncIo的Request函数，将请求入队


returnm_pIo->Request(llPos,lLength,TRUE,pBuffer,(LPVOID)pSample,dwUser);

}

CAsyncIo的Request函数会将请求封装到CAsyncRequest类中，并将该类对象保存到一个队列中：

HRESULT

CAsyncIo::Request( LONGLONG llPos,LONG lLength,BOOL bAligned,BYTE*pBuffer,

LPVOID pContext,DWORDdwUser)

{

if (bAligned) {

if (!IsAligned(llPos) ||

!IsAligned(lLength)||

!IsAligned((LONG)pBuffer)) {

return VFW_E_BADALIGN;

}

}

//新建一个CAsyncRequest变量，必将下游filter的请求封装到该变量中


CAsyncRequest* pRequest = newCAsyncRequest;


HRESULT hr = pRequest->Request(this,m_pStream,llPos, lLength,
bAligned,

pBuffer,pContext,dwUser);

if (SUCCEEDED(hr)) {

// might fail if flushing

//将请求放进请求队列中


hr = PutWorkItem(pRequest);

}

if (FAILED(hr)) {

delete pRequest;

}

return hr;

}

这样Async工程的sourcefilter就把CpullPin拉数据的请求保存到了请求队列中，那么Async是怎么处理请求队列中的请求的呢？

（2）sourcefilter怎么完成下游的拉数据请求

在Async工程基类CAsyncIo的PutWorkItem函数中，会开启一个线程来完成放到请求对队列中的各个数据请求。也就是在将下游filter拉数据请求放到请求队列的同时会来开启一个线程来读文件完成请求：

HRESULT CAsyncIo::PutWorkItem(CAsyncRequest* pRequest)

{

CAutoLock lock(&m_csLists);

HRESULT hr;

if (m_bFlushing) {

hr = VFW_E_WRONG_STATE;

}

else if (m_listWork.AddTail(pRequest)) {//将请求放入队列

// event should now be in aset state - force this

m_evWork.Set();

// start the thread now ifnot already started

//开启处理请求的线程，如果线程已经启动，该函数不做任何事情

hr = StartThread();

} else {

hr = E_OUTOFMEMORY;

}

return(hr);

}

在开启的线程函数中会调用ThreadProc函数，该函数中有一个无限循环，不断地调用

ProcessRequests()函数来处理请求队列中的请求：

voidCAsyncIo::ProcessRequests(void)

{

// lockto get the item and increment the outstanding count

CAsyncRequest * preq = NULL;

for (;;) {

{

CAutoLock lock(&m_csLists);

//从请求队列中取出一个请求

preq = GetWorkItem();

if (preq == NULL) {

// done

return;

}

// one more item not onthe done or work list

m_cItemsOut++;

// release critsec

}

//处理请求

preq->Complete();

// regain critsec to replaceon done list

{

CAutoLock l(&m_csLists);

//处理完一个请求后将请求放入已完成请求队列


PutDoneItem(preq);

if (--m_cItemsOut == 0){

if (m_bWaiting) {

m_evAllDone.Set();

}

}

}

}

}

处理请求的具体动作在CAsyncRequest类中的Complete()函数完成，该函数调用了CAsyncStream的Read函数读取数据放入IMediaSample中，在Async工程中CMemStream继承自CAsyncStream接口，实现了该接口的各个函数，因此此时调用的是CMemStream的Read函数。

CAsyncRequest::Complete()

{

m_pStream->Lock();

//设置读取的位置


m_hr = m_pStream->SetPointer(m_llPos);

if (S_OK == m_hr) {

DWORD dwActual;

//读取收放入IMediaSample中


m_hr = m_pStream->Read(m_pBuffer,m_lLength, m_bAligned, &dwActual);

if (m_hr == OLE_S_FIRST) {

if (m_pContext) {

IMediaSample *pSample= reinterpret_cast<IMediaSample *>(m_pContext);

pSample->SetDiscontinuity(TRUE);

m_hr = S_OK;

}

}

if (FAILED(m_hr)) {

} else if (dwActual != (DWORD)m_lLength){

// tell caller sizechanged - probably because of EOF

m_lLength = (LONG)dwActual;

m_hr = S_FALSE;

} else {

m_hr = S_OK;

}

}

m_pStream->Unlock();

return m_hr;

}

至此拉数据的请求已有Async工程的sourcefilter接收并完成，那么CPullPin是怎么把完成的请求拉回去的呢？

（3）sourcefilter完成的下游拉数据请求读取的数据是怎么被下游filter得到的？

在CPullPin的Process函数中调用QueueSample函数发送完拉取数据的请求后，会调用CollectAndDeliver函数来获取完成的请求，并处理接收到的数据：

HRESULT CPullPin::CollectAndDeliver(REFERENCE_TIME tStart, REFERENCE_TIMEtStop)

{

IMediaSample* pSample =NULL; // better be sure pSample is set

DWORD_PTR dwUnused;

//获取请求的样本数据，m_pReader是一个IAsyncReader指针变量


HRESULT hr =m_pReader->WaitForNext(INFINITE,&pSample,&dwUnused);

if (FAILED(hr)) {

if (pSample) {

pSample->Release();

}

} else {

//处理获取的样本数据


hr= DeliverSample(pSample, tStart, tStop);

}

if (FAILED(hr)) {

CleanupCancelled();

OnError(hr);

}

return hr;

}

调用IasyncReader的接口WaitForNext时，也就调用了sourcefilter的输出pin（CAsyncOutputPin）上实现的IasyncReader接口的WaitForNext。同样的CAsyncOutputPin实现的WaitForNext函数调用CAsyncIo的WaitForNext函数来完成具体功能：

STDMETHODIMP CAsyncOutputPin::WaitForNext(DWORD dwTimeout,IMediaSample**ppSample,

DWORD * pdwUser) // user context

{

LONG cbActual;

IMediaSample* pSample;

//调用CAsyncIo的WaitForNext函数来完成具体功能

HRESULT hr = m_pIo->WaitForNext(dwTimeout,(LPVOID*)&pSample,pdwUser,&cbActual);

if (SUCCEEDED(hr)) {

pSample->SetActualDataLength(cbActual);

}

*ppSample = pSample;

return hr;

}

在CAsyncIo::WaitForNext函数中变量ppContext就是接口IasyncReader函数WaitForNext中的IMediaSample变量：

HRESULT CAsyncIo::WaitForNext(DWORD dwTimeout,LPVOID *ppContext,DWORD *pdwUser,

LONG* pcbActual)

{

// some errors find a sample,others don't. Ensure that

// *ppContext is NULL if nosample found

*ppContext = NULL;

// wait until the event is set,but since we are not

// holding the critsec whenwaiting, we may need to re-wait

for (;;) {

//等待同步事件

if(!m_evDone.Wait(dwTimeout)) {

// timeout occurred

return VFW_E_TIMEOUT;

}

// get next event from list

//从已完成的请求队列取出一个已完成的请求


CAsyncRequest* pRequest =GetDoneItem();

if (pRequest) {

// found a completedrequest

// check if ok

HRESULT hr = pRequest->GetHResult();

if (hr == S_FALSE) {

// this means theactual length was less than

// requested - maybe ok if he aligned the end of file

if((pRequest->GetActualLength() +

pRequest->GetStart())== Size()) {

hr = S_OK;

} else {

// it was anactual read error

hr = E_FAIL;

}

}

// return actual bytesread

*pcbActual =pRequest->GetActualLength();

// return his context

//从完成的请求中得到IMediaSample，由于ppContext实际是IasyncReader函数//WaitForNext中的IMediaSample变量,这样CPullPin通过IasyncReader接口的WaitForNext函//数就得到了其请求的数据

*ppContext = pRequest->GetContext();


*pdwUser = pRequest->GetUser();

delete pRequest;

return hr;

} else {

// Hold the critical section while checking the

//
list state

CAutoLock lck(&m_csLists);

if (m_bFlushing&& !m_bWaiting) {

// can't block as weare between BeginFlush and EndFlush

// but note that ifm_bWaiting is set, then there are some

// items not yetcomplete that we should block for.

returnVFW_E_WRONG_STATE;

}

}

// done item was grabbedbetween completion and

// us locking m_csLists.

}

}

以上便是拉模式下的sourcefilter的数据被拉进spliter（parser）filter的简要过程，当spliter（parser）filter得到请求的数据后，便可以进行处理，并将处理后的数据推送给下游的Filter了。

3、
Async工程中的Sourcefilter是怎么读取文件的？

CAsyncFilter继承了IFileSourceFilter接口，通过实现IFileSourceFilter接口的Load

函数来加载文件。在Load函数中，实现了打开文件，读取文件的内容到内存，并将文件信息和内容传递给CMemStream的实例，在此后的拉数据中，就有CMemStream类处理数据的读取。

STDMETHODIMP Load(LPCOLESTR lpwszFileName, const AM_MEDIA_TYPE *pmt)

{

CheckPointer(lpwszFileName,E_POINTER);

// lstrlenW is one of thefew Unicode functions that works on win95

//求文件名长度将宽字节转化为多字节


int cch =lstrlenW(lpwszFileName) + 1;

TCHAR *lpszFileName;

#ifndef UNICODE

lpszFileName = new char[cch* 2];

if (!lpszFileName) {

return E_OUTOFMEMORY;

}

WideCharToMultiByte(GetACP(), 0, lpwszFileName, -1,

lpszFileName,cch, NULL, NULL);

#else

lpszFileName =lpwszFileName;

#endif

CAutoLocklck(&m_csFilter);

/* Check the file type */

CMediaType cmt;

if (NULL == pmt) {

cmt.SetType(&MEDIATYPE_Stream);

cmt.SetSubtype(&MEDIASUBTYPE_NULL);

} else {

cmt =*pmt;

}

//读文件内容到内存


if(!ReadTheFile(lpszFileName)) {

#ifndef UNICODE

delete [] lpszFileName;

#endif

return E_FAIL;

}

//CMemStream初始化函数


m_Stream.Init(m_pbData, m_llSize);


//保存文件名


m_pFileName = new WCHAR[cch];


if (m_pFileName!=NULL) {


CopyMemory(m_pFileName, lpwszFileName,cch*sizeof(WCHAR));


}

// this is not a simpleassignment... pointers and format

// block (if any) areintelligently copied


//保存媒体类型


m_mt = cmt;

/* Work out file type */

cmt.bTemporalCompression =TRUE;
//???

cmt.lSampleSize = 1;

return S_OK;

}

至此Aysnc实现的拉模式sourcefilter中数据流具体流动步骤已分析完毕，还有很多其小的细节如IasyncReader其它接口函数的运行，CAsyncIo中线程同步的操作，这里就不在做具体介绍。

memfile（Memfile文件夹）工程实现的基本功能和Async是完成相同的，其和Async唯一不同在于memfile实现了一个可以直接应用的拉模式sourcefilter，省去了生成.ax之后在进行注册使用步骤。

在memfile工程中类CMemReader只继承了CAsyncReader，不在继承IFileSourceFilter接口。CMemReader不在需要注册，因此其重载的注册和反注册函数什么都不做，支持简单的返回S_OK。Memfile功能源文件的的加载在main函数中完成：

int _CRTAPI1 main(int argc, char *argv[])

{

……//省略main参数的解析

/打开文件


HANDLE hFile = CreateFile(argv[1],


GENERIC_READ,


FILE_SHARE_READ,


NULL,


OPEN_EXISTING,


0,


NULL);

if (hFile ==INVALID_HANDLE_VALUE) {

printf("Could not open%s\n", argv[1]);

return 1;

}

//获得文件的长度


ULARGE_INTEGER uliSize;


uliSize.LowPart = GetFileSize(hFile,&uliSize.HighPart);

PBYTE pbMem = newBYTE[uliSize.LowPart];

if (pbMem == NULL) {

printf("Could notallocate %d bytes\n", uliSize.LowPart);

return 1;

}

//读取文件的内容


DWORD dwBytesRead;


if (!ReadFile(hFile,


(LPVOID)pbMem,


uliSize.LowPart,


&dwBytesRead,


NULL) ||


dwBytesRead != uliSize.LowPart) {


printf("Could not readfile\n");


CloseHandle(hFile);


return 1;


}

CloseHandle(hFile);

HRESULT hr = S_OK;

CoInitialize(NULL);

//由读取的文件内容和文件长度构造CMemStream，并由CMemStream构造CMemReader(该源sourcefilter的主filter)


CMemStream Stream(pbMem, (LONGLONG)uliSize.QuadPart,dwKBPerSec);


CMemReader *rdr = newCMemReader(&Stream, &mt, &hr);

if (FAILED(hr) || rdr == NULL) {

delete rdr;

printf("Could notcreate filter HRESULT 0x%8.8X\n", hr);

CoUninitialize();

return 1;

}

//
Make sure we don't accidentally go away!

//增加sourcefilter的引用值，并将其加入Graph中，运行该Graph


rdr->AddRef();


IFilterGraph *pFG = NULL;


hr = SelectAndRender(rdr, &pFG);

if (FAILED(hr)) {

printf("Failed tocreate graph and render file HRESULT 0x%8.8X",

hr);

} else {

// Play the file

//播放文件，程序会阻塞在这里，知道文件播放完毕


HRESULT hr = PlayFileWait(pFG);

if (FAILED(hr)) {

printf("Failed toplay graph HRESULT 0x%8.8X",

hr);

}

}

rdr->Release();

if (pFG) {

ULONG ulRelease =pFG->Release();

if (ulRelease != 0) {

printf("Filtergraph count not 0! was %d",ulRelease);

}

}

CoUninitialize();

return 0;

}