ReactOS  0.4.14-dev-552-g2fad488
outputqueue.c
Go to the documentation of this file.
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 #define COBJMACROS
22 
23 #include "dshow.h"
24 #include "wine/debug.h"
25 #include "wine/unicode.h"
26 #include "wine/list.h"
27 #include "wine/strmbase.h"
28 #include "uuids.h"
29 #include "vfwmsgs.h"
30 #include <assert.h>
31 
33 
35 
36 typedef struct tagQueuedEvent {
37  int type;
38  struct list entry;
39 
41 } QueuedEvent;
42 
44 {
46  return This->pFuncsTable->pfnThreadProc(This);
47 }
48 
49 static void OutputQueue_FreeSamples(OutputQueue *pOutputQueue)
50 {
51  struct list *cursor, *cursor2;
52  LIST_FOR_EACH_SAFE(cursor, cursor2, &pOutputQueue->SampleList)
53  {
56  HeapFree(GetProcessHeap(),0,qev);
57  }
58 }
59 
61  BaseOutputPin *pInputPin,
62  BOOL bAuto,
63  BOOL bQueue,
64  LONG lBatchSize,
65  BOOL bBatchExact,
66  DWORD dwPriority,
67  const OutputQueueFuncTable* pFuncsTable,
68  OutputQueue **ppOutputQueue )
69 
70 {
71  BOOL threaded = FALSE;
72  DWORD tid;
73 
75 
76  if (!pInputPin || !pFuncsTable || !ppOutputQueue)
77  return E_INVALIDARG;
78 
79  *ppOutputQueue = HeapAlloc(GetProcessHeap(),HEAP_ZERO_MEMORY,sizeof(OutputQueue));
80  if (!*ppOutputQueue)
81  return E_OUTOFMEMORY;
82 
83  This = *ppOutputQueue;
84  This->pFuncsTable = pFuncsTable;
85  This->lBatchSize = lBatchSize;
86  This->bBatchExact = bBatchExact;
88  This->csQueue.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ": OutputQueue.csQueue");
89  list_init(&This->SampleList);
90 
91  This->pInputPin = pInputPin;
92  IPin_AddRef(&pInputPin->pin.IPin_iface);
93 
94  EnterCriticalSection(&This->csQueue);
95  if (bAuto && pInputPin->pMemInputPin)
96  threaded = IMemInputPin_ReceiveCanBlock(pInputPin->pMemInputPin) == S_OK;
97  else
98  threaded = bQueue;
99 
100  if (threaded)
101  {
103  if (This->hThread)
104  {
105  SetThreadPriority(This->hThread, dwPriority);
106  This->hProcessQueue = CreateEventW(NULL, 0, 0, NULL);
107  }
108  }
109  LeaveCriticalSection(&This->csQueue);
110 
111  return S_OK;
112 }
113 
115 {
116  EnterCriticalSection(&pOutputQueue->csQueue);
117  OutputQueue_FreeSamples(pOutputQueue);
118  pOutputQueue->bTerminate = TRUE;
119  SetEvent(pOutputQueue->hProcessQueue);
120  LeaveCriticalSection(&pOutputQueue->csQueue);
121 
122  pOutputQueue->csQueue.DebugInfo->Spare[0] = 0;
123  DeleteCriticalSection(&pOutputQueue->csQueue);
124  CloseHandle(pOutputQueue->hProcessQueue);
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 
174 {
175  LONG processed;
176  return OutputQueue_ReceiveMultiple(pOutputQueue,&pSample,1,&processed);
177 }
178 
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 
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 
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  {
251  if (qev->type == SAMPLE_PACKET)
252  ppSamples[nSamples++] = qev->pSample;
253  else
254  break;
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  {
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);
290  HeapFree(GetProcessHeap(),0,qev);
291  }
292  }
293  pOutputQueue->bSendAnyway = FALSE;
294  }
295  LeaveCriticalSection(&pOutputQueue->csQueue);
297  }
298  while (!pOutputQueue->bTerminate);
299  return S_OK;
300 }
#define LIST_FOR_EACH_SAFE(cursor, cursor2, list)
Definition: list.h:192
#define TRUE
Definition: types.h:120
#define CloseHandle
Definition: compat.h:406
#define DWORD_PTR
Definition: treelist.c:76
HRESULT hr
Definition: shlfolder.c:183
IPin IPin_iface
Definition: strmbase.h:35
static DWORD WINAPI OutputQueue_InitialThreadProc(LPVOID data)
Definition: outputqueue.c:43
VOID WINAPI OutputQueue_SendAnyway(OutputQueue *pOutputQueue)
Definition: outputqueue.c:179
struct tagQueuedEvent QueuedEvent
BOOL WINAPI DECLSPEC_HOTPATCH SetEvent(IN HANDLE hEvent)
Definition: synch.c:733
HRESULT WINAPI OutputQueue_Receive(OutputQueue *pOutputQueue, IMediaSample *pSample)
Definition: outputqueue.c:173
BOOL bBatchExact
Definition: strmbase.h:354
void WINAPI EnterCriticalSection(LPCRITICAL_SECTION)
HANDLE WINAPI DECLSPEC_HOTPATCH CreateEventW(IN LPSECURITY_ATTRIBUTES lpEventAttributes OPTIONAL, IN BOOL bManualReset, IN BOOL bInitialState, IN LPCWSTR lpName OPTIONAL)
Definition: synch.c:651
HANDLE hThread
Definition: strmbase.h:350
struct list entry
Definition: outputqueue.c:38
__WINE_SERVER_LIST_INLINE void list_add_tail(struct list *list, struct list *elem)
Definition: list.h:102
DWORD WINAPI WaitForSingleObject(IN HANDLE hHandle, IN DWORD dwMilliseconds)
Definition: synch.c:82
GLsizei GLenum const GLvoid GLsizei GLenum GLbyte GLbyte GLbyte GLdouble GLdouble GLdouble GLfloat GLfloat GLfloat GLint GLint GLint GLshort GLshort GLshort GLubyte GLubyte GLubyte GLuint GLuint GLuint GLushort GLushort GLushort GLbyte GLbyte GLbyte GLbyte GLdouble GLdouble GLdouble GLdouble GLfloat GLfloat GLfloat GLfloat GLint GLint GLint GLint GLshort GLshort GLshort GLshort GLubyte GLubyte GLubyte GLubyte GLuint GLuint GLuint GLuint GLushort GLushort GLushort GLushort GLboolean const GLdouble const GLfloat const GLint const GLshort const GLbyte const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLdouble const GLfloat const GLfloat const GLint const GLint const GLshort const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort const GLdouble const GLfloat const GLint const GLshort GLenum GLenum GLenum GLfloat GLenum GLint GLenum GLenum GLenum GLfloat GLenum GLenum GLint GLenum GLfloat GLenum GLint GLint GLushort GLenum GLenum GLfloat GLenum GLenum GLint GLfloat const GLubyte GLenum GLenum GLenum const GLfloat GLenum GLenum const GLint GLenum GLint GLint GLsizei GLsizei GLint GLenum GLenum const GLvoid GLenum GLenum const GLfloat GLenum GLenum const GLint GLenum GLenum const GLdouble GLenum GLenum const GLfloat GLenum GLenum const GLint GLsizei GLuint GLfloat GLuint GLbitfield GLfloat GLint GLuint GLboolean GLenum GLfloat GLenum GLbitfield GLenum GLfloat GLfloat GLint GLint const GLfloat GLenum GLfloat GLfloat GLint GLint GLfloat GLfloat GLint GLint const GLfloat GLint GLfloat GLfloat GLint GLfloat GLfloat GLint GLfloat GLfloat const GLdouble const GLfloat const GLdouble const GLfloat GLint i
Definition: glfuncs.h:248
#define E_OUTOFMEMORY
Definition: ddrawi.h:100
HRESULT WINAPI OutputQueue_Construct(BaseOutputPin *pInputPin, BOOL bAuto, BOOL bQueue, LONG lBatchSize, BOOL bBatchExact, DWORD dwPriority, const OutputQueueFuncTable *pFuncsTable, OutputQueue **ppOutputQueue)
Definition: outputqueue.c:60
unsigned int BOOL
Definition: ntddk_ex.h:94
__WINE_SERVER_LIST_INLINE unsigned int list_count(const struct list *list)
Definition: list.h:155
long LONG
Definition: pedump.c:60
#define FIXME(fmt,...)
Definition: debug.h:110
HRESULT WINAPI OutputQueue_Destroy(OutputQueue *pOutputQueue)
Definition: outputqueue.c:114
#define E_INVALIDARG
Definition: ddrawi.h:101
HANDLE WINAPI DECLSPEC_HOTPATCH CreateThread(IN LPSECURITY_ATTRIBUTES lpThreadAttributes, IN DWORD dwStackSize, IN LPTHREAD_START_ROUTINE lpStartAddress, IN LPVOID lpParameter, IN DWORD dwCreationFlags, OUT LPDWORD lpThreadId)
Definition: thread.c:136
VOID WINAPI InitializeCriticalSection(OUT LPCRITICAL_SECTION lpCriticalSection)
Definition: synch.c:751
smooth NULL
Definition: ftsmooth.c:416
CRITICAL_SECTION csQueue
Definition: strmbase.h:346
BOOL bSendAnyway
Definition: strmbase.h:356
HANDLE hProcessQueue
Definition: strmbase.h:351
#define VFW_E_NOT_CONNECTED
Definition: vfwmsgs.h:48
VOID WINAPI OutputQueue_EOS(OutputQueue *pOutputQueue)
Definition: outputqueue.c:193
BOOL WINAPI SetThreadPriority(IN HANDLE hThread, IN int nPriority)
Definition: thread.c:699
WINE_DEFAULT_DEBUG_CHANNEL(strmbase)
__WINE_SERVER_LIST_INLINE void list_remove(struct list *elem)
Definition: list.h:108
#define GetProcessHeap()
Definition: compat.h:403
PVOID WINAPI HeapAlloc(HANDLE, DWORD, SIZE_T)
struct list SampleList
Definition: strmbase.h:358
void WINAPI DeleteCriticalSection(PCRITICAL_SECTION)
LONG HRESULT
Definition: typedefs.h:77
BasePin pin
Definition: strmbase.h:66
#define WINAPI
Definition: msvc.h:8
IMediaSample * pSample
Definition: outputqueue.c:40
unsigned long DWORD
Definition: ntddk_ex.h:95
GLint GLenum GLsizei GLsizei GLsizei GLint GLsizei const GLvoid * data
Definition: gl.h:1950
PCRITICAL_SECTION_DEBUG DebugInfo
Definition: winbase.h:866
BaseOutputPin * pInputPin
Definition: strmbase.h:348
IPin * pConnectedTo
Definition: strmbase.h:39
uint32_t entry
Definition: isohybrid.c:63
IMemInputPin * pMemInputPin
Definition: strmbase.h:67
Definition: _list.h:228
#define ERR(fmt,...)
Definition: debug.h:109
__WINE_SERVER_LIST_INLINE int list_empty(const struct list *list)
Definition: list.h:143
#define S_OK
Definition: intsafe.h:59
DWORD WINAPI OutputQueueImpl_ThreadProc(OutputQueue *pOutputQueue)
Definition: outputqueue.c:224
HRESULT WINAPI OutputQueue_ReceiveMultiple(OutputQueue *pOutputQueue, IMediaSample **ppSamples, LONG nSamples, LONG *nSamplesProcessed)
Definition: outputqueue.c:131
const char cursor[]
Definition: icontest.c:13
Definition: axcore.idl:91
#define HEAP_ZERO_MEMORY
Definition: compat.h:123
static void OutputQueue_FreeSamples(OutputQueue *pOutputQueue)
Definition: outputqueue.c:49
LONG lBatchSize
Definition: strmbase.h:353
static int processed(const type_t *type)
Definition: typegen.c:2236
__WINE_SERVER_LIST_INLINE void list_init(struct list *list)
Definition: list.h:149
void WINAPI LeaveCriticalSection(LPCRITICAL_SECTION)
#define LIST_ENTRY(type)
Definition: queue.h:175
#define INFINITE
Definition: serial.h:102
static TfClientId tid
#define HeapFree(x, y, z)
Definition: compat.h:402
BOOL bTerminate
Definition: strmbase.h:355
DWORD_PTR Spare[8/sizeof(DWORD_PTR)]
Definition: winbase.h:859