reactos/lib/3rdparty/strmbase/outputqueue.c

   1 /*
   2  * Generic Implementation of COutputQueue
   3  *
   4  * Copyright 2011 Aric Stewart, CodeWeavers
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Lesser General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2.1 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General Public
  17  * License along with this library; if not, write to the Free Software
  18  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  19  */
  20
  21 #include "strmbase_private.h"
  22
  23 #include <wine/list.h>
  24
  25 enum {SAMPLE_PACKET, EOS_PACKET};
  26
  27 typedef struct tagQueuedEvent {
  28     int type;
  29     struct list entry;
  30
  31     IMediaSample *pSample;
  32 } QueuedEvent;
  33
  34 static DWORD WINAPI OutputQueue_InitialThreadProc(LPVOID data)
  35 {
  36     OutputQueue *This = (OutputQueue *)data;
  37     return This->pFuncsTable->pfnThreadProc(This);
  38 }
  39
  40 static void OutputQueue_FreeSamples(OutputQueue *pOutputQueue)
  41 {
  42     struct list *cursor, *cursor2;
  43     LIST_FOR_EACH_SAFE(cursor, cursor2, pOutputQueue->SampleList)
  44     {
  45         QueuedEvent *qev = LIST_ENTRY(cursor, QueuedEvent, entry);
  46         list_remove(cursor);
  47         HeapFree(GetProcessHeap(),0,qev);
  48     }
  49 }
  50
  51 HRESULT WINAPI OutputQueue_Construct(
  52     BaseOutputPin *pInputPin,
  53     BOOL bAuto,
  54     BOOL bQueue,
  55     LONG lBatchSize,
  56     BOOL bBatchExact,
  57     DWORD dwPriority,
  58     const OutputQueueFuncTable* pFuncsTable,
  59     OutputQueue **ppOutputQueue )
  60
  61 {
  62     BOOL threaded = FALSE;
  63     DWORD tid;
  64
  65     OutputQueue *This;
  66
  67     if (!pInputPin || !pFuncsTable || !ppOutputQueue)
  68         return E_INVALIDARG;
  69
  70     *ppOutputQueue = HeapAlloc(GetProcessHeap(),HEAP_ZERO_MEMORY,sizeof(OutputQueue));
  71     if (!*ppOutputQueue)
  72         return E_OUTOFMEMORY;
  73
  74     This = *ppOutputQueue;
  75     This->pFuncsTable = pFuncsTable;
  76     This->lBatchSize = lBatchSize;
  77     This->bBatchExact = bBatchExact;
  78     InitializeCriticalSection(&This->csQueue);
  79     This->csQueue.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ": OutputQueue.csQueue");
  80     This->SampleList = HeapAlloc(GetProcessHeap(),0,sizeof(struct list));
  81     if (!This->SampleList)
  82     {
  83         OutputQueue_Destroy(This);
  84         *ppOutputQueue = NULL;
  85         return E_OUTOFMEMORY;
  86     }
  87     list_init(This->SampleList);
  88
  89     This->pInputPin = pInputPin;
  90     IPin_AddRef(&pInputPin->pin.IPin_iface);
  91
  92     EnterCriticalSection(&This->csQueue);
  93     if (bAuto && pInputPin->pMemInputPin)
  94         threaded = IMemInputPin_ReceiveCanBlock(pInputPin->pMemInputPin) == S_OK;
  95     else
  96         threaded = bQueue;
  97
  98     if (threaded)
  99     {
 100         This->hThread = CreateThread(NULL, 0, OutputQueue_InitialThreadProc, This, 0, &tid);
 101         if (This->hThread)
 102         {
 103             SetThreadPriority(This->hThread, dwPriority);
 104             This->hProcessQueue = CreateEventW(NULL, 0, 0, NULL);
 105         }
 106     }
 107     LeaveCriticalSection(&This->csQueue);
 108
 109     return S_OK;
 110 }
 111
 112 HRESULT WINAPI OutputQueue_Destroy(OutputQueue *pOutputQueue)
 113 {
 114     EnterCriticalSection(&pOutputQueue->csQueue);
 115     OutputQueue_FreeSamples(pOutputQueue);
 116     pOutputQueue->bTerminate = TRUE;
 117     SetEvent(pOutputQueue->hProcessQueue);
 118     LeaveCriticalSection(&pOutputQueue->csQueue);
 119
 120     pOutputQueue->csQueue.DebugInfo->Spare[0] = 0;
 121     DeleteCriticalSection(&pOutputQueue->csQueue);
 122     CloseHandle(pOutputQueue->hProcessQueue);
 123
 124     HeapFree(GetProcessHeap(),0,pOutputQueue->SampleList);
 125
 126     IPin_Release(&pOutputQueue->pInputPin->pin.IPin_iface);
 127     HeapFree(GetProcessHeap(),0,pOutputQueue);
 128     return S_OK;
 129 }
 130
 131 HRESULT WINAPI OutputQueue_ReceiveMultiple(OutputQueue *pOutputQueue, IMediaSample **ppSamples, LONG nSamples, LONG *nSamplesProcessed)
 132 {
 133     HRESULT hr = S_OK;
 134     int i;
 135
 136     if (!pOutputQueue->pInputPin->pin.pConnectedTo || !pOutputQueue->pInputPin->pMemInputPin)
 137         return VFW_E_NOT_CONNECTED;
 138
 139     if (!pOutputQueue->hThread)
 140     {
 141         IMemInputPin_AddRef(pOutputQueue->pInputPin->pMemInputPin);
 142         hr = IMemInputPin_ReceiveMultiple(pOutputQueue->pInputPin->pMemInputPin,ppSamples, nSamples, nSamplesProcessed);
 143         IMemInputPin_Release(pOutputQueue->pInputPin->pMemInputPin);
 144     }
 145     else
 146     {
 147         EnterCriticalSection(&pOutputQueue->csQueue);
 148         *nSamplesProcessed = 0;
 149
 150         for (i = 0; i < nSamples; i++)
 151         {
 152             QueuedEvent *qev = HeapAlloc(GetProcessHeap(),0,sizeof(QueuedEvent));
 153             if (!qev)
 154             {
 155                 ERR("Out of Memory\n");
 156                 hr = E_OUTOFMEMORY;
 157                 break;
 158             }
 159             qev->type = SAMPLE_PACKET;
 160             qev->pSample = ppSamples[i];
 161             IMediaSample_AddRef(ppSamples[i]);
 162             list_add_tail(pOutputQueue->SampleList, &qev->entry);
 163             (*nSamplesProcessed)++;
 164         }
 165
 166         if (!pOutputQueue->bBatchExact || list_count(pOutputQueue->SampleList) >= pOutputQueue->lBatchSize)
 167             SetEvent(pOutputQueue->hProcessQueue);
 168         LeaveCriticalSection(&pOutputQueue->csQueue);
 169     }
 170     return hr;
 171 }
 172
 173 HRESULT WINAPI OutputQueue_Receive(OutputQueue *pOutputQueue, IMediaSample *pSample)
 174 {
 175     LONG processed;
 176     return OutputQueue_ReceiveMultiple(pOutputQueue,&pSample,1,&processed);
 177 }
 178
 179 VOID WINAPI OutputQueue_SendAnyway(OutputQueue *pOutputQueue)
 180 {
 181     if (pOutputQueue->hThread)
 182     {
 183         EnterCriticalSection(&pOutputQueue->csQueue);
 184         if (!list_empty(pOutputQueue->SampleList))
 185         {
 186             pOutputQueue->bSendAnyway = TRUE;
 187             SetEvent(pOutputQueue->hProcessQueue);
 188         }
 189         LeaveCriticalSection(&pOutputQueue->csQueue);
 190     }
 191 }
 192
 193 VOID WINAPI OutputQueue_EOS(OutputQueue *pOutputQueue)
 194 {
 195     EnterCriticalSection(&pOutputQueue->csQueue);
 196     if (pOutputQueue->hThread)
 197     {
 198         QueuedEvent *qev = HeapAlloc(GetProcessHeap(),0,sizeof(QueuedEvent));
 199         if (!qev)
 200         {
 201             ERR("Out of Memory\n");
 202             LeaveCriticalSection(&pOutputQueue->csQueue);
 203             return;
 204         }
 205         qev->type = EOS_PACKET;
 206         qev->pSample = NULL;
 207         list_add_tail(pOutputQueue->SampleList, &qev->entry);
 208     }
 209     else
 210     {
 211         IPin* ppin = NULL;
 212         IPin_ConnectedTo(&pOutputQueue->pInputPin->pin.IPin_iface, &ppin);
 213         if (ppin)
 214         {
 215             IPin_EndOfStream(ppin);
 216             IPin_Release(ppin);
 217         }
 218     }
 219     LeaveCriticalSection(&pOutputQueue->csQueue);
 220     /* Covers sending the Event to the worker Thread */
 221     OutputQueue_SendAnyway(pOutputQueue);
 222 }
 223
 224 DWORD WINAPI OutputQueueImpl_ThreadProc(OutputQueue *pOutputQueue)
 225 {
 226     do
 227     {
 228         EnterCriticalSection(&pOutputQueue->csQueue);
 229         if (!list_empty(pOutputQueue->SampleList) &&
 230             (!pOutputQueue->bBatchExact ||
 231             list_count(pOutputQueue->SampleList) >= pOutputQueue->lBatchSize ||
 232             pOutputQueue->bSendAnyway
 233             )
 234            )
 235         {
 236             while (!list_empty(pOutputQueue->SampleList))
 237             {
 238                 IMediaSample **ppSamples;
 239                 LONG nSamples;
 240                 LONG nSamplesProcessed;
 241                 struct list *cursor, *cursor2;
 242                 int i = 0;
 243
 244                 /* First Pass Process Samples */
 245                 i = list_count(pOutputQueue->SampleList);
 246                 ppSamples = HeapAlloc(GetProcessHeap(),0,sizeof(IMediaSample*) * i);
 247                 nSamples = 0;
 248                 LIST_FOR_EACH_SAFE(cursor, cursor2, pOutputQueue->SampleList)
 249                 {
 250                     QueuedEvent *qev = LIST_ENTRY(cursor, QueuedEvent, entry);
 251                     if (qev->type == SAMPLE_PACKET)
 252                         ppSamples[nSamples++] = qev->pSample;
 253                     else
 254                         break;
 255                     list_remove(cursor);
 256                     HeapFree(GetProcessHeap(),0,qev);
 257                 }
 258
 259                 if (pOutputQueue->pInputPin->pin.pConnectedTo && pOutputQueue->pInputPin->pMemInputPin)
 260                 {
 261                     IMemInputPin_AddRef(pOutputQueue->pInputPin->pMemInputPin);
 262                     LeaveCriticalSection(&pOutputQueue->csQueue);
 263                     IMemInputPin_ReceiveMultiple(pOutputQueue->pInputPin->pMemInputPin, ppSamples, nSamples, &nSamplesProcessed);
 264                     EnterCriticalSection(&pOutputQueue->csQueue);
 265                     IMemInputPin_Release(pOutputQueue->pInputPin->pMemInputPin);
 266                 }
 267                 for (i = 0; i < nSamples; i++)
 268                     IMediaSample_Release(ppSamples[i]);
 269                 HeapFree(GetProcessHeap(),0,ppSamples);
 270
 271                 /* Process Non-Samples */
 272                 LIST_FOR_EACH_SAFE(cursor, cursor2, pOutputQueue->SampleList)
 273                 {
 274                     QueuedEvent *qev = LIST_ENTRY(cursor, QueuedEvent, entry);
 275                     if (qev->type == EOS_PACKET)
 276                     {
 277                         IPin* ppin = NULL;
 278                         IPin_ConnectedTo(&pOutputQueue->pInputPin->pin.IPin_iface, &ppin);
 279                         if (ppin)
 280                         {
 281                             IPin_EndOfStream(ppin);
 282                             IPin_Release(ppin);
 283                         }
 284                     }
 285                     else if (qev->type == SAMPLE_PACKET)
 286                         break;
 287                     else
 288                         FIXME("Unhandled Event type %i\n",qev->type);
 289                     list_remove(cursor);
 290                     HeapFree(GetProcessHeap(),0,qev);
 291                 }
 292             }
 293             pOutputQueue->bSendAnyway = FALSE;
 294         }
 295         LeaveCriticalSection(&pOutputQueue->csQueue);
 296         WaitForSingleObject(pOutputQueue->hProcessQueue, INFINITE);
 297     }
 298     while (!pOutputQueue->bTerminate);
 299     return S_OK;
 300 }