dlls/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 WINE_DEFAULT_DEBUG_CHANNEL(strmbase);
  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     list_init(&This->SampleList);
  81
  82     This->pInputPin = pInputPin;
  83     IPin_AddRef(&pInputPin->pin.IPin_iface);
  84
  85     EnterCriticalSection(&This->csQueue);
  86     if (bAuto && pInputPin->pMemInputPin)
  87         threaded = IMemInputPin_ReceiveCanBlock(pInputPin->pMemInputPin) == S_OK;
  88     else
  89         threaded = bQueue;
  90
  91     if (threaded)
  92     {
  93         This->hThread = CreateThread(NULL, 0, OutputQueue_InitialThreadProc, This, 0, &tid);
  94         if (This->hThread)
  95         {
  96             SetThreadPriority(This->hThread, dwPriority);
  97             This->hProcessQueue = CreateEventW(NULL, 0, 0, NULL);
  98         }
  99     }
 100     LeaveCriticalSection(&This->csQueue);
 101
 102     return S_OK;
 103 }
 104
 105 HRESULT WINAPI OutputQueue_Destroy(OutputQueue *pOutputQueue)
 106 {
 107     EnterCriticalSection(&pOutputQueue->csQueue);
 108     OutputQueue_FreeSamples(pOutputQueue);
 109     pOutputQueue->bTerminate = TRUE;
 110     SetEvent(pOutputQueue->hProcessQueue);
 111     LeaveCriticalSection(&pOutputQueue->csQueue);
 112
 113     pOutputQueue->csQueue.DebugInfo->Spare[0] = 0;
 114     DeleteCriticalSection(&pOutputQueue->csQueue);
 115     CloseHandle(pOutputQueue->hProcessQueue);
 116
 117     IPin_Release(&pOutputQueue->pInputPin->pin.IPin_iface);
 118     HeapFree(GetProcessHeap(),0,pOutputQueue);
 119     return S_OK;
 120 }
 121
 122 HRESULT WINAPI OutputQueue_ReceiveMultiple(OutputQueue *pOutputQueue, IMediaSample **ppSamples, LONG nSamples, LONG *nSamplesProcessed)
 123 {
 124     HRESULT hr = S_OK;
 125     int i;
 126
 127     if (!pOutputQueue->pInputPin->pin.pConnectedTo || !pOutputQueue->pInputPin->pMemInputPin)
 128         return VFW_E_NOT_CONNECTED;
 129
 130     if (!pOutputQueue->hThread)
 131     {
 132         IMemInputPin_AddRef(pOutputQueue->pInputPin->pMemInputPin);
 133         hr = IMemInputPin_ReceiveMultiple(pOutputQueue->pInputPin->pMemInputPin,ppSamples, nSamples, nSamplesProcessed);
 134         IMemInputPin_Release(pOutputQueue->pInputPin->pMemInputPin);
 135     }
 136     else
 137     {
 138         EnterCriticalSection(&pOutputQueue->csQueue);
 139         *nSamplesProcessed = 0;
 140
 141         for (i = 0; i < nSamples; i++)
 142         {
 143             QueuedEvent *qev = HeapAlloc(GetProcessHeap(),0,sizeof(QueuedEvent));
 144             if (!qev)
 145             {
 146                 ERR("Out of Memory\n");
 147                 hr = E_OUTOFMEMORY;
 148                 break;
 149             }
 150             qev->type = SAMPLE_PACKET;
 151             qev->pSample = ppSamples[i];
 152             IMediaSample_AddRef(ppSamples[i]);
 153             list_add_tail(&pOutputQueue->SampleList, &qev->entry);
 154             (*nSamplesProcessed)++;
 155         }
 156
 157         if (!pOutputQueue->bBatchExact || list_count(&pOutputQueue->SampleList) >= pOutputQueue->lBatchSize)
 158             SetEvent(pOutputQueue->hProcessQueue);
 159         LeaveCriticalSection(&pOutputQueue->csQueue);
 160     }
 161     return hr;
 162 }
 163
 164 HRESULT WINAPI OutputQueue_Receive(OutputQueue *pOutputQueue, IMediaSample *pSample)
 165 {
 166     LONG processed;
 167     return OutputQueue_ReceiveMultiple(pOutputQueue,&pSample,1,&processed);
 168 }
 169
 170 VOID WINAPI OutputQueue_SendAnyway(OutputQueue *pOutputQueue)
 171 {
 172     if (pOutputQueue->hThread)
 173     {
 174         EnterCriticalSection(&pOutputQueue->csQueue);
 175         if (!list_empty(&pOutputQueue->SampleList))
 176         {
 177             pOutputQueue->bSendAnyway = TRUE;
 178             SetEvent(pOutputQueue->hProcessQueue);
 179         }
 180         LeaveCriticalSection(&pOutputQueue->csQueue);
 181     }
 182 }
 183
 184 VOID WINAPI OutputQueue_EOS(OutputQueue *pOutputQueue)
 185 {
 186     EnterCriticalSection(&pOutputQueue->csQueue);
 187     if (pOutputQueue->hThread)
 188     {
 189         QueuedEvent *qev = HeapAlloc(GetProcessHeap(),0,sizeof(QueuedEvent));
 190         if (!qev)
 191         {
 192             ERR("Out of Memory\n");
 193             LeaveCriticalSection(&pOutputQueue->csQueue);
 194             return;
 195         }
 196         qev->type = EOS_PACKET;
 197         qev->pSample = NULL;
 198         list_add_tail(&pOutputQueue->SampleList, &qev->entry);
 199     }
 200     else
 201     {
 202         IPin* ppin = NULL;
 203         IPin_ConnectedTo(&pOutputQueue->pInputPin->pin.IPin_iface, &ppin);
 204         if (ppin)
 205         {
 206             IPin_EndOfStream(ppin);
 207             IPin_Release(ppin);
 208         }
 209     }
 210     LeaveCriticalSection(&pOutputQueue->csQueue);
 211     /* Covers sending the Event to the worker Thread */
 212     OutputQueue_SendAnyway(pOutputQueue);
 213 }
 214
 215 DWORD WINAPI OutputQueueImpl_ThreadProc(OutputQueue *pOutputQueue)
 216 {
 217     do
 218     {
 219         EnterCriticalSection(&pOutputQueue->csQueue);
 220         if (!list_empty(&pOutputQueue->SampleList) &&
 221             (!pOutputQueue->bBatchExact ||
 222             list_count(&pOutputQueue->SampleList) >= pOutputQueue->lBatchSize ||
 223             pOutputQueue->bSendAnyway
 224             )
 225            )
 226         {
 227             while (!list_empty(&pOutputQueue->SampleList))
 228             {
 229                 IMediaSample **ppSamples;
 230                 LONG nSamples;
 231                 LONG nSamplesProcessed;
 232                 struct list *cursor, *cursor2;
 233                 int i = 0;
 234
 235                 /* First Pass Process Samples */
 236                 i = list_count(&pOutputQueue->SampleList);
 237                 ppSamples = HeapAlloc(GetProcessHeap(),0,sizeof(IMediaSample*) * i);
 238                 nSamples = 0;
 239                 LIST_FOR_EACH_SAFE(cursor, cursor2, &pOutputQueue->SampleList)
 240                 {
 241                     QueuedEvent *qev = LIST_ENTRY(cursor, QueuedEvent, entry);
 242                     if (qev->type == SAMPLE_PACKET)
 243                         ppSamples[nSamples++] = qev->pSample;
 244                     else
 245                         break;
 246                     list_remove(cursor);
 247                     HeapFree(GetProcessHeap(),0,qev);
 248                 }
 249
 250                 if (pOutputQueue->pInputPin->pin.pConnectedTo && pOutputQueue->pInputPin->pMemInputPin)
 251                 {
 252                     IMemInputPin_AddRef(pOutputQueue->pInputPin->pMemInputPin);
 253                     LeaveCriticalSection(&pOutputQueue->csQueue);
 254                     IMemInputPin_ReceiveMultiple(pOutputQueue->pInputPin->pMemInputPin, ppSamples, nSamples, &nSamplesProcessed);
 255                     EnterCriticalSection(&pOutputQueue->csQueue);
 256                     IMemInputPin_Release(pOutputQueue->pInputPin->pMemInputPin);
 257                 }
 258                 for (i = 0; i < nSamples; i++)
 259                     IMediaSample_Release(ppSamples[i]);
 260                 HeapFree(GetProcessHeap(),0,ppSamples);
 261
 262                 /* Process Non-Samples */
 263                 LIST_FOR_EACH_SAFE(cursor, cursor2, &pOutputQueue->SampleList)
 264                 {
 265                     QueuedEvent *qev = LIST_ENTRY(cursor, QueuedEvent, entry);
 266                     if (qev->type == EOS_PACKET)
 267                     {
 268                         IPin* ppin = NULL;
 269                         IPin_ConnectedTo(&pOutputQueue->pInputPin->pin.IPin_iface, &ppin);
 270                         if (ppin)
 271                         {
 272                             IPin_EndOfStream(ppin);
 273                             IPin_Release(ppin);
 274                         }
 275                     }
 276                     else if (qev->type == SAMPLE_PACKET)
 277                         break;
 278                     else
 279                         FIXME("Unhandled Event type %i\n",qev->type);
 280                     list_remove(cursor);
 281                     HeapFree(GetProcessHeap(),0,qev);
 282                 }
 283             }
 284             pOutputQueue->bSendAnyway = FALSE;
 285         }
 286         LeaveCriticalSection(&pOutputQueue->csQueue);
 287         WaitForSingleObject(pOutputQueue->hProcessQueue, INFINITE);
 288     }
 289     while (!pOutputQueue->bTerminate);
 290     return S_OK;
 291 }