asyncio.cpp

Mpeg流的网络客户端实时播放

//------------------------------------------------------------------------------
// File: AsyncIo.cpp
//
// Desc: DirectShow sample code - base library with I/O functionality.
//
// Copyright (c) 1992-2001 Microsoft Corporation.  All rights reserved.
//------------------------------------------------------------------------------

#include "stdafx.h"
#include <streams.h>
#include "asyncio.h"

// --- CAsyncRequest ---


// implementation of CAsyncRequest representing a single
// outstanding request. All the i/o for this object is done
// in the Complete method.


// init the params for this request.
// Read is not issued until the complete call
HRESULT
CAsyncRequest::Request(
    CAsyncIo *pIo,
    CAsyncStream *pStream,
    LONGLONG llPos,
    LONG lLength,
    BOOL bAligned,
    BYTE* pBuffer,
    LPVOID pContext,	// filter's context
    DWORD dwUser)	// downstream filter's context
{
    m_pIo = pIo;
    m_pStream = pStream;
    m_llPos = llPos;
    m_lLength = lLength;
    m_bAligned = bAligned;
    m_pBuffer = pBuffer;
    m_pContext = pContext;
    m_dwUser = dwUser;
    m_hr = VFW_E_TIMEOUT;   // not done yet

    return S_OK;
}


// issue the i/o if not overlapped, and block until i/o complete.
// returns error code of file i/o
//
//
HRESULT
CAsyncRequest::Complete()
{

    m_pStream->Lock();

    m_hr = m_pStream->SetPointer(m_llPos);
    if (S_OK == m_hr) {

        DWORD dwActual;
        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 size changed - probably because of EOF
            m_lLength = (LONG) dwActual;
            m_hr = S_FALSE;
        } else {
            m_hr = S_OK;
        }
    }

    m_pStream->Unlock();
    return m_hr;
}



// --- CAsyncIo ---

// note - all events created manual reset

CAsyncIo::CAsyncIo(CAsyncStream *pStream)
 : m_hThread(NULL),
   m_evWork(TRUE),
   m_evDone(TRUE),
   m_evStop(TRUE),
   m_listWork(NAME("Work list")),
   m_listDone(NAME("Done list")),
   m_bFlushing(FALSE),
   m_cItemsOut(0),
   m_bWaiting(FALSE),
   m_pStream(pStream)
{

}


CAsyncIo::~CAsyncIo()
{
    // move everything to the done list
    BeginFlush();

    // shutdown worker thread
    CloseThread();

    // empty the done list
    POSITION pos = m_listDone.GetHeadPosition();
    while (pos) {
        CAsyncRequest* pRequest = m_listDone.GetNext(pos);
        delete pRequest;
    }
    m_listDone.RemoveAll();
}

// ready for async activity - call this before
// calling Request.
//
// start the worker thread if we need to
//
// !!! use overlapped i/o if possible
HRESULT
CAsyncIo::AsyncActive(void)
{
    return StartThread();
}

// call this when no more async activity will happen before
// the next AsyncActive call
//
// stop the worker thread if active
HRESULT
CAsyncIo::AsyncInactive(void)
{
    return CloseThread();
}


// add a request to the queue.
HRESULT
CAsyncIo::Request(
            LONGLONG llPos,
            LONG lLength,
            BOOL bAligned,
            BYTE* pBuffer,
            LPVOID pContext,
            DWORD dwUser)
{
    if (bAligned) {
        if (!IsAligned(llPos) ||
    	!IsAligned(lLength) ||
    	!IsAligned((LONG) pBuffer)) {
            return VFW_E_BADALIGN;
        }
    }

    CAsyncRequest* pRequest = new CAsyncRequest;

    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;
}


// wait for the next request to complete
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 no sample found
    *ppContext = NULL;

    // wait until the event is set, but since we are not
    // holding the critsec when waiting, 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 completed request

            // check if ok
            HRESULT hr = pRequest->GetHResult();
            if (hr == S_FALSE) {

                // this means the actual length was less than
                // requested - may be ok if he aligned the end of file
                if ((pRequest->GetActualLength() +
                     pRequest->GetStart()) == Size()) {
                        hr = S_OK;
                } else {
                    // it was an actual read error
                    hr = E_FAIL;
                }
            }

            // return actual bytes read
            *pcbActual = pRequest->GetActualLength();

            // return his context
            *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 we are between BeginFlush and EndFlush

                // but note that if m_bWaiting is set, then there are some
                // items not yet complete that we should block for.

                return VFW_E_WRONG_STATE;
            }
        }

        // done item was grabbed between completion and
        // us locking m_csLists.
    }
}

// perform a synchronous read request on this thread.
// Need to hold m_csFile while doing this (done in
// request object)
HRESULT
CAsyncIo::SyncReadAligned(
            LONGLONG llPos,
            LONG lLength,
            BYTE* pBuffer,
            LONG* pcbActual,
            PVOID pvContext
            )
{
    if (!IsAligned(llPos) ||
	!IsAligned(lLength) ||
	!IsAligned((LONG) pBuffer)) {
            return VFW_E_BADALIGN;
    }

    CAsyncRequest request;

    HRESULT hr = request.Request(
                    this,
                    m_pStream,
                    llPos,
                    lLength,
                    TRUE,
                    pBuffer,
                    pvContext,
                    0);

    if (FAILED(hr)) {
        return hr;
    }

    hr = request.Complete();

    // return actual data length
    *pcbActual = request.GetActualLength();
    return hr;
}

HRESULT
CAsyncIo::Length(LONGLONG *pllTotal, LONGLONG* pllAvailable)
{
    *pllTotal = m_pStream->Size(pllAvailable);
    return S_OK;
}

// cancel all items on the worklist onto the done list
// and refuse further requests or further WaitForNext calls
// until the end flush
//
// WaitForNext must return with NULL only if there are no successful requests.
// So Flush does the following:
// 1. set m_bFlushing ensures no more requests succeed
// 2. move all items from work list to the done list.
// 3. If there are any outstanding requests, then we need to release the
//    critsec to allow them to complete. The m_bWaiting as well as ensuring
//    that we are signalled when they are all done is also used to indicate
//    to WaitForNext that it should continue to block.
// 4. Once all outstanding requests are complete, we force m_evDone set and
//    m_bFlushing set and m_bWaiting false. This ensures that WaitForNext will
//    not block when the done list is empty.
HRESULT
CAsyncIo::BeginFlush()
{
    // hold the lock while emptying the work list
    {
        CAutoLock lock(&m_csLists);