Написание асинхронного метода

  • Статья

В этом разделе описывается реализация асинхронного метода в Microsoft Media Foundation.

Асинхронные методы являются повсеместными в конвейере Media Foundation. Асинхронные методы упрощают распределение работы между несколькими потоками. Особенно важно выполнять операции ввода-вывода асинхронно, чтобы чтение из файла или сети не блокировать остальную часть конвейера.

При написании источника или приемника мультимедиа крайне важно правильно обрабатывать асинхронные операции, так как производительность компонента влияет на весь конвейер.

Примечание

Преобразования Media Foundation (MFT) по умолчанию используют синхронные методы.

 

Рабочие очереди для асинхронных операций

В Media Foundation существует тесная связь между асинхронными методами обратного вызова и рабочими очередями. Рабочая очередь — это абстракция для перемещения работы из потока вызывающего объекта в рабочий поток. Чтобы выполнить работу в рабочей очереди, сделайте следующее:

  1. Реализуйте интерфейс IMFAsyncCallback .

  2. Вызовите MFCreateAsyncResult , чтобы создать результирующий объект. Результирующий объект предоставляет объект IMFAsyncResult. Результирующий объект содержит три указателя:

    • Указатель на интерфейс IMFAsyncCallback вызывающего абонента.
    • Необязательный указатель на объект состояния. Если этот параметр задан, объект состояния должен реализовывать IUnknown.
    • Необязательный указатель на частный объект. Если он указан, этот объект также должен реализовывать IUnknown.

    Последние два указателя могут иметь значение NULL. В противном случае используйте их для хранения сведений об асинхронной операции.

  3. Вызовите MFPutWorkItemEx для постановки в очередь к рабочему элементу.

  4. Поток рабочей очереди вызывает метод IMFAsyncCallback::Invoke .

  5. Выполните работу в методе Invoke . Параметр pAsyncResult этого метода является указателем IMFAsyncResult из шага 2. Используйте этот указатель, чтобы получить объект состояния и частный объект:

В качестве альтернативы можно объединить шаги 2 и 3, вызвав функцию MFPutWorkItem . Внутри этой функции вызывается MFCreateAsyncResult , чтобы создать результирующий объект.

На следующей схеме показаны связи между вызывающим объектом, результирующим объектом, объектом состояния и частным объектом.

Схема, показывающая асинхронный результирующий объект

На следующей схеме последовательностей показано, как объект помещает рабочий элемент в очередь. Когда поток рабочей очереди вызывает Invoke, объект выполняет асинхронную операцию с этим потоком.

схема, показывающая, как объект помещает рабочий элемент в очередь

Важно помнить, что вызов Invoke вызывается из потока, который принадлежит рабочей очереди. Реализация Invoke должна быть потокобезопасной. Кроме того, если вы используете рабочую очередь платформы (MFASYNC_CALLBACK_QUEUE_STANDARD), очень важно никогда не блокировать поток, так как это может блокировать обработку данных во всем конвейере Media Foundation. Если вам нужно выполнить операцию, которая будет блокировать или займет много времени, используйте частную рабочую очередь. Чтобы создать частную рабочую очередь, вызовите MFAllocateWorkQueue. Любой компонент конвейера, выполняющий операции ввода-вывода, не должен блокировать вызовы ввода-вывода по той же причине. Интерфейс IMFByteStream предоставляет полезную абстракцию для асинхронных операций ввода-вывода файлов.

Реализация 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. Это может быть приложение или конвейер Media Foundation. Компонент — это код, реализующий BeginX и EndX.

  1. Вызывающий вызывает Begin..., передавая указатель на интерфейс IMFAsyncCallback вызывающего абонента.
  2. Компонент создает новый асинхронный результирующий объект. Этот объект хранит интерфейс обратного вызова вызывающего объекта и объект состояния. Как правило, в нем также хранятся все частные сведения о состоянии, необходимые компоненту для завершения операции. Результирующий объект на этом шаге помечается как "Результат 1" на следующей схеме.
  3. Компонент создает второй результирующий объект. В этом результирующем объекте хранятся два указателя: первый результирующий объект и интерфейс обратного вызова вызываемого объекта. На следующей схеме этот результирующий объект помечен как "Результат 2".
  4. Компонент вызывает MFPutWorkItemEx для постановки нового рабочего элемента в очередь.
  5. В методе Invoke компонент выполняет асинхронную работу.
  6. Компонент вызывает MFInvokeCallback для вызова метода обратного вызова вызывающего элемента.
  7. Вызывающий объект вызывает метод EndX .

Схема, показывающая, как объект реализует шаблон начала и конца

Пример асинхронного метода

Чтобы проиллюстрировать это обсуждение, мы будем использовать надуманый пример. Рассмотрим асинхронный метод вычисления квадратного корня:

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

Параметр BeginSquareRootx — это значение, для которого будет вычисляться квадратный корень. Квадратный корень возвращается в параметре pVal для EndSquareRoot.

Ниже приведено объявление класса, реализующего эти два метода:

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 объект . Затем объект передается в AsyncOpDoCalculateSquareRoot. Наконец, он вызывает 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 , объект не должен помещать рабочий элемент в очередь для B , пока операция A не завершится. Объект может соответствовать этому требованию, имея собственную очередь ожидающих операций. Когда для объекта вызывается асинхронный метод, объект помещает запрос в собственную очередь. По мере завершения каждой асинхронной операции объект извлекает следующий запрос из очереди. В примере MPEG1Source показан пример реализации такой очереди.

Один метод может включать несколько асинхронных операций, особенно при использовании вызовов ввода-вывода. При реализации асинхронных методов тщательно продумайте требования к сериализации. Например, допустимо ли для объекта начать новую операцию, пока предыдущий запрос ввода-вывода все еще находится в состоянии ожидания? Если новая операция изменяет внутреннее состояние объекта, что происходит, когда предыдущий запрос ввода-вывода завершается и возвращает данные, которые теперь могут оказаться устаревшими? Хорошая схема состояния может помочь определить допустимые переходы состояния.

Рекомендации по межпотоковой и межпроцессной обработке

Рабочие очереди не используют маршалинг COM для маршалинга указателей интерфейса через границы потока. Таким образом, даже если объект зарегистрирован как поток с потоком размещения или поток приложения вошел в однопотоковый объект (STA), обратные вызовы IMFAsyncCallback будут вызываться из другого потока. В любом случае все компоненты конвейера Media Foundation должны использовать модель потоков "Оба".

Некоторые интерфейсы в Media Foundation определяют удаленные версии некоторых асинхронных методов. При вызове одного из этих методов через границы процесса библиотека-посредник или заглушка Media Foundation вызывает удаленную версию метода, которая выполняет настраиваемое маршалинг параметров метода. В удаленном процессе заглушка преобразует вызов обратно в локальный метод объекта . Этот процесс прозрачен как для приложения, так и для удаленного объекта. Эти пользовательские методы маршалинга предоставляются в основном для объектов, загруженных в защищенный путь к мультимедиа (PMP). Дополнительные сведения о PMP см. в разделе Защищенный путь к мультимедиа.

Асинхронные методы обратного вызова

Рабочие очереди