[PATCH 0/3] MR1904: winegstreamer: Explicitly sleep in the streaming thread when flushing.

[PATCH 1/3] winegstreamer: Factor out a parser_create() helper.

[Zebediah Figura]

From: Zebediah Figura

---
 dlls/winegstreamer/quartz_parser.c | 67 +++++++++++-------------------
 1 file changed, 24 insertions(+), 43 deletions(-)

diff --git a/dlls/winegstreamer/quartz_parser.c b/dlls/winegstreamer/quartz_parser.c
index 6e31f2736fe..8a227145d78 100644
--- a/dlls/winegstreamer/quartz_parser.c
+++ b/dlls/winegstreamer/quartz_parser.c
@@ -1343,29 +1343,34 @@ static HRESULT decodebin_parser_source_get_media_type(struct parser_source *pin,
     return VFW_S_NO_MORE_ITEMS;
 }

-static BOOL parser_init_gstreamer(void)
-{
-    if (!init_gstreamer())
-        return FALSE;
-    return TRUE;
-}
-
-HRESULT decodebin_parser_create(IUnknown *outer, IUnknown **out)
+static HRESULT parser_create(enum wg_parser_type type, struct parser **parser)
 {
     struct parser *object;

-    if (!parser_init_gstreamer())
+    if (!init_gstreamer())
         return E_FAIL;

if (!(object = calloc(1, sizeof(*object))))
         return E_OUTOFMEMORY;

-    if (!(object->wg_parser = wg_parser_create(WG_PARSER_DECODEBIN, false)))
+    if (!(object->wg_parser = wg_parser_create(type, false)))
     {
         free(object);
         return E_OUTOFMEMORY;
     }

+    *parser = object;
+    return S_OK;
+}
+
+HRESULT decodebin_parser_create(IUnknown *outer, IUnknown **out)
+{
+    struct parser *object;
+    HRESULT hr;
+
+    if (FAILED(hr = parser_create(WG_PARSER_DECODEBIN, &object)))
+        return hr;
+
     strmbase_filter_init(&object->filter, outer, &CLSID_decodebin_parser, &filter_ops);
     strmbase_sink_init(&object->sink, &object->filter, L"input pin", &sink_ops, NULL);

@@ -1881,18 +1886,10 @@ static HRESULT wave_parser_source_get_media_type(struct parser_source *pin,
 HRESULT wave_parser_create(IUnknown *outer, IUnknown **out)
 {
     struct parser *object;
+    HRESULT hr;

-    if (!parser_init_gstreamer())
-        return E_FAIL;
-
-    if (!(object = calloc(1, sizeof(*object))))
-        return E_OUTOFMEMORY;
-
-    if (!(object->wg_parser = wg_parser_create(WG_PARSER_WAVPARSE, false)))
-    {
-        free(object);
-        return E_OUTOFMEMORY;
-    }
+    if (FAILED(hr = parser_create(WG_PARSER_WAVPARSE, &object)))
+        return hr;

strmbase_filter_init(&object->filter, outer, &CLSID_WAVEParser, &filter_ops);
     strmbase_sink_init(&object->sink, &object->filter, L"input pin", &wave_parser_sink_ops, NULL);
@@ -1967,18 +1964,10 @@ static HRESULT avi_splitter_source_get_media_type(struct parser_source *pin,
 HRESULT avi_splitter_create(IUnknown *outer, IUnknown **out)
 {
     struct parser *object;
+    HRESULT hr;

-    if (!parser_init_gstreamer())
-        return E_FAIL;
-
-    if (!(object = calloc(1, sizeof(*object))))
-        return E_OUTOFMEMORY;
-
-    if (!(object->wg_parser = wg_parser_create(WG_PARSER_AVIDEMUX, false)))
-    {
-        free(object);
-        return E_OUTOFMEMORY;
-    }
+    if (FAILED(hr = parser_create(WG_PARSER_AVIDEMUX, &object)))
+        return hr;

strmbase_filter_init(&object->filter, outer, &CLSID_AviSplitter, &filter_ops);
     strmbase_sink_init(&object->sink, &object->filter, L"input pin", &avi_splitter_sink_ops, NULL);
@@ -2074,18 +2063,10 @@ static const struct strmbase_filter_ops mpeg_splitter_ops =
 HRESULT mpeg_splitter_create(IUnknown *outer, IUnknown **out)
 {
     struct parser *object;
+    HRESULT hr;

-    if (!parser_init_gstreamer())
-        return E_FAIL;
-
-    if (!(object = calloc(1, sizeof(*object))))
-        return E_OUTOFMEMORY;
-
-    if (!(object->wg_parser = wg_parser_create(WG_PARSER_MPEGAUDIOPARSE, false)))
-    {
-        free(object);
-        return E_OUTOFMEMORY;
-    }
+    if (FAILED(hr = parser_create(WG_PARSER_MPEGAUDIOPARSE, &object)))
+        return hr;

strmbase_filter_init(&object->filter, outer, &CLSID_MPEG1Splitter, &mpeg_splitter_ops);
     strmbase_sink_init(&object->sink, &object->filter, L"Input", &mpeg_splitter_sink_ops, NULL);

[PATCH 2/3] winegstreamer: Protect the "streaming" member of struct parser with a separate lock.

[Zebediah Figura]

From: Zebediah Figura

The code previously relied on inherent atomicity of atomic types, but atomicity
doesn't imply the right memory ordering. Be explicit about the threading model
we want.

Signed-off-by: Zebediah Figura 
---
 dlls/winegstreamer/quartz_parser.c | 27 ++++++++++++++++++++++++++-
 1 file changed, 26 insertions(+), 1 deletion(-)

diff --git a/dlls/winegstreamer/quartz_parser.c b/dlls/winegstreamer/quartz_parser.c
index 8a227145d78..36ca84debba 100644
--- a/dlls/winegstreamer/quartz_parser.c
+++ b/dlls/winegstreamer/quartz_parser.c
@@ -54,6 +54,15 @@ struct parser

struct wg_parser *wg_parser;

+    /* This protects the "streaming" field, accessed by both the application
+     * and streaming threads.
+     * We cannot use the filter lock for this, since that is held while waiting
+     * for the streaming thread, and hence the streaming thread cannot take the
+     * filter lock.
+     * This lock must not be acquired before acquiring the filter lock or
+     * flushing_cs. */
+    CRITICAL_SECTION streaming_cs;
+
     /* FIXME: It would be nice to avoid duplicating these with strmbase.
      * However, synchronization is tricky; we need access to be protected by a
      * separate lock. */
@@ -972,10 +981,18 @@ static DWORD CALLBACK stream_thread(void *arg)

TRACE("Starting streaming thread for pin %p.\n", pin);

-    while (filter->streaming)
+    for (;;)
     {
         struct wg_parser_buffer buffer;

+        EnterCriticalSection(&filter->streaming_cs);
+        if (!filter->streaming)
+        {
+            LeaveCriticalSection(&filter->streaming_cs);
+            break;
+        }
+        LeaveCriticalSection(&filter->streaming_cs);
+
         EnterCriticalSection(&pin->flushing_cs);

if (pin->eos)
@@ -1095,6 +1112,9 @@ static void parser_destroy(struct strmbase_filter *iface)

wg_parser_destroy(filter->wg_parser);

+    filter->streaming_cs.DebugInfo->Spare[0] = 0;
+    DeleteCriticalSection(&filter->streaming_cs);
+
     strmbase_sink_cleanup(&filter->sink);
     strmbase_filter_cleanup(&filter->filter);
     free(filter);
@@ -1167,7 +1187,9 @@ static HRESULT parser_cleanup_stream(struct strmbase_filter *iface)
     if (!filter->sink_connected)
         return S_OK;

+    EnterCriticalSection(&filter->streaming_cs);
     filter->streaming = false;
+    LeaveCriticalSection(&filter->streaming_cs);

for (i = 0; i < filter->source_count; ++i)
     {
@@ -1359,6 +1381,9 @@ static HRESULT parser_create(enum wg_parser_type type, struct parser **parser)
         return E_OUTOFMEMORY;
     }

+    InitializeCriticalSection(&object->streaming_cs);
+    object->streaming_cs.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ": parser.streaming_cs");
+
     *parser = object;
     return S_OK;
 }

[PATCH 3/3] winegstreamer: Explicitly sleep in the streaming thread when flushing.

[Zebediah Figura]

From: Zebediah Figura

The code originally intended to achieve this by simply waiting on flushing_cs,
and having the application thread hold flushing_cs while seeking. Unfortunately,
this can result in starvation of the application thread, since the streaming
thread always reacquires flushing_cs immediately after releasing it. Avoid this
by using a separate condition variable.

Wine-Bug: https://bugs.winehq.org/show_bug.cgi?id=53403
Signed-off-by: Zebediah Figura 
---
 dlls/winegstreamer/quartz_parser.c | 29 +++++++++++++++++++++++++----
 1 file changed, 25 insertions(+), 4 deletions(-)

diff --git a/dlls/winegstreamer/quartz_parser.c b/dlls/winegstreamer/quartz_parser.c
index 36ca84debba..5561b106327 100644
--- a/dlls/winegstreamer/quartz_parser.c
+++ b/dlls/winegstreamer/quartz_parser.c
@@ -54,20 +54,23 @@ struct parser

struct wg_parser *wg_parser;

-    /* This protects the "streaming" field, accessed by both the application
-     * and streaming threads.
+    /* This protects the "streaming" and "flushing" fields, accessed by both
+     * the application and streaming threads.
      * We cannot use the filter lock for this, since that is held while waiting
      * for the streaming thread, and hence the streaming thread cannot take the
      * filter lock.
      * This lock must not be acquired before acquiring the filter lock or
      * flushing_cs. */
     CRITICAL_SECTION streaming_cs;
+    CONDITION_VARIABLE flushing_cv;

/* FIXME: It would be nice to avoid duplicating these with strmbase.
      * However, synchronization is tricky; we need access to be protected by a
      * separate lock. */
     bool streaming, sink_connected;

+    bool flushing;
+
     HANDLE read_thread;

BOOL (*init_gst)(struct parser *filter);
@@ -986,11 +989,16 @@ static DWORD CALLBACK stream_thread(void *arg)
         struct wg_parser_buffer buffer;

EnterCriticalSection(&filter->streaming_cs);
+
+        while (filter->flushing)
+            SleepConditionVariableCS(&filter->flushing_cv, &filter->streaming_cs, INFINITE);
+
         if (!filter->streaming)
         {
             LeaveCriticalSection(&filter->streaming_cs);
             break;
         }
+
         LeaveCriticalSection(&filter->streaming_cs);

EnterCriticalSection(&pin->flushing_cs);
@@ -1384,6 +1392,8 @@ static HRESULT parser_create(enum wg_parser_type type, struct parser **parser)
     InitializeCriticalSection(&object->streaming_cs);
     object->streaming_cs.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ": parser.streaming_cs");

+    InitializeConditionVariable(&object->flushing_cv);
+
     *parser = object;
     return S_OK;
 }
@@ -1532,8 +1542,13 @@ static HRESULT WINAPI GST_Seeking_SetPositions(IMediaSeeking *iface,
             IAsyncReader_BeginFlush(filter->reader);
     }

-    /* Acquire the flushing locks. This blocks the streaming threads, and
-     * ensures the seek is serialized between flushes. */
+    /* Signal the streaming threads to "pause". */
+    EnterCriticalSection(&filter->streaming_cs);
+    filter->flushing = true;
+    LeaveCriticalSection(&filter->streaming_cs);
+
+    /* Acquire the flushing locks, to make sure the streaming threads really
+     * are paused. This ensures the seek is serialized between flushes. */
     for (i = 0; i < filter->source_count; ++i)
     {
         struct parser_source *flush_pin = filter->sources[i];
@@ -1576,6 +1591,12 @@ static HRESULT WINAPI GST_Seeking_SetPositions(IMediaSeeking *iface,
         }
     }

+    /* Signal the streaming threads to resume. */
+    EnterCriticalSection(&filter->streaming_cs);
+    filter->flushing = false;
+    LeaveCriticalSection(&filter->streaming_cs);
+    WakeConditionVariable(&filter->flushing_cv);
+
     return S_OK;
 }