[PATCH 1/3] winegstreamer: Duplicate source shutdown path into constructor with leak fixes.

[Derek Lesho]

Instead of having _Shutdown and the constructors' failures paths mix together creating confusion about what should be released where, we can designate ::Shutdown/::Release to shutting down fully initialized objects without checks, and keep the partially-created object cleanup code in the constructor.

Signed-off-by: Derek Lesho 
---
 dlls/winegstreamer/media_source.c | 99 +++++++++++++++++++------------
 1 file changed, 62 insertions(+), 37 deletions(-)

diff --git a/dlls/winegstreamer/media_source.c b/dlls/winegstreamer/media_source.c
index 01ab626254a..2a0bb83374d 100644
--- a/dlls/winegstreamer/media_source.c
+++ b/dlls/winegstreamer/media_source.c
@@ -733,6 +733,12 @@ static HRESULT new_media_stream(struct media_source *source,
     object->IMFMediaStream_iface.lpVtbl = &media_stream_vtbl;
     object->ref = 1;

+    if (FAILED(hr = MFCreateEventQueue(&object->event_queue)))
+    {
+        free(object);
+        return hr;
+    }
+
     IMFMediaSource_AddRef(&source->IMFMediaSource_iface);
     object->parent_source = source;
     object->stream_id = stream_id;
@@ -741,20 +747,11 @@ static HRESULT new_media_stream(struct media_source *source,
     object->eos = FALSE;
     object->wg_stream = wg_stream;

-    if (FAILED(hr = MFCreateEventQueue(&object->event_queue)))
-        goto fail;
-
     TRACE("Created stream object %p.\n", object);

*out_stream = object;

return S_OK;
-
-fail:
-    WARN("Failed to construct media stream, hr %#x.\n", hr);
-
-    IMFMediaStream_Release(&object->IMFMediaStream_iface);
-    return hr;
 }

static HRESULT media_stream_init_desc(struct media_stream *stream)
@@ -857,6 +854,8 @@ done:
         IMFMediaTypeHandler_Release(type_handler);
     for (i = 0; i < type_count; i++)
         IMFMediaType_Release(stream_types[i]);
+    if (FAILED(hr))
+        IMFStreamDescriptor_Release(stream->descriptor);
     return hr;
 }

@@ -1213,19 +1212,13 @@ static HRESULT WINAPI media_source_Shutdown(IMFMediaSource *iface)

unix_funcs->wg_parser_disconnect(source->wg_parser);

-    if (source->read_thread)
-    {
-        source->read_thread_shutdown = true;
-        WaitForSingleObject(source->read_thread, INFINITE);
-        CloseHandle(source->read_thread);
-    }
+    source->read_thread_shutdown = true;
+    WaitForSingleObject(source->read_thread, INFINITE);
+    CloseHandle(source->read_thread);

-    if (source->pres_desc)
-        IMFPresentationDescriptor_Release(source->pres_desc);
-    if (source->event_queue)
-        IMFMediaEventQueue_Shutdown(source->event_queue);
-    if (source->byte_stream)
-        IMFByteStream_Release(source->byte_stream);
+    IMFPresentationDescriptor_Release(source->pres_desc);
+    IMFMediaEventQueue_Shutdown(source->event_queue);
+    IMFByteStream_Release(source->byte_stream);

for (i = 0; i < source->stream_count; i++)
     {
@@ -1233,23 +1226,18 @@ static HRESULT WINAPI media_source_Shutdown(IMFMediaSource *iface)

stream->state = STREAM_SHUTDOWN;

-        if (stream->event_queue)
-            IMFMediaEventQueue_Shutdown(stream->event_queue);
-        if (stream->descriptor)
-            IMFStreamDescriptor_Release(stream->descriptor);
-        if (stream->parent_source)
-            IMFMediaSource_Release(&stream->parent_source->IMFMediaSource_iface);
+        IMFMediaEventQueue_Shutdown(stream->event_queue);
+        IMFStreamDescriptor_Release(stream->descriptor);
+        IMFMediaSource_Release(&stream->parent_source->IMFMediaSource_iface);

IMFMediaStream_Release(&stream->IMFMediaStream_iface);
     }

unix_funcs->wg_parser_destroy(source->wg_parser);

-    if (source->stream_count)
-        free(source->streams);
+    free(source->streams);

-    if (source->async_commands_queue)
-        MFUnlockWorkQueue(source->async_commands_queue);
+    MFUnlockWorkQueue(source->async_commands_queue);

return S_OK;
 }
@@ -1274,6 +1262,7 @@ static const IMFMediaSourceVtbl IMFMediaSource_vtbl =
 static HRESULT media_source_constructor(IMFByteStream *bytestream, struct media_source **out_media_source)
 {
     IMFStreamDescriptor **descriptors = NULL;
+    unsigned int stream_count = 0;
     struct media_source *object;
     UINT64 total_pres_time = 0;
     struct wg_parser *parser;
@@ -1337,15 +1326,15 @@ static HRESULT media_source_constructor(IMFByteStream *bytestream, struct media_
      * leak occurs with native. */
     unix_funcs->wg_parser_set_unlimited_buffering(parser);

-    object->stream_count = unix_funcs->wg_parser_get_stream_count(parser);
+    stream_count = unix_funcs->wg_parser_get_stream_count(parser);

-    if (!(object->streams = calloc(object->stream_count, sizeof(*object->streams))))
+    if (!(object->streams = calloc(stream_count, sizeof(*object->streams))))
     {
         hr = E_OUTOFMEMORY;
         goto fail;
     }

-    for (i = 0; i < object->stream_count; ++i)
+    for (i = 0; i < stream_count; ++i)
     {
         if (FAILED(hr = new_media_stream(object, unix_funcs->wg_parser_get_stream(parser, i), i, &object->streams[i])))
             goto fail;
@@ -1353,9 +1342,13 @@ static HRESULT media_source_constructor(IMFByteStream *bytestream, struct media_
         if (FAILED(hr = media_stream_init_desc(object->streams[i])))
         {
             ERR("Failed to finish initialization of media stream %p, hr %x.\n", object->streams[i], hr);
-            IMFMediaStream_Release(&object->streams[i]->IMFMediaStream_iface);
+            IMFMediaSource_Release(&object->streams[i]->parent_source->IMFMediaSource_iface);
+            IMFMediaEventQueue_Release(object->streams[i]->event_queue);
+            free(object->streams[i]);
             goto fail;
         }
+
+        object->stream_count++;
     }

/* init presentation descriptor */
@@ -1392,8 +1385,40 @@ static HRESULT media_source_constructor(IMFByteStream *bytestream, struct media_
     fail:
     WARN("Failed to construct MFMediaSource, hr %#x.\n", hr);

-    free(descriptors);
-    IMFMediaSource_Release(&object->IMFMediaSource_iface);
+    if (descriptors)
+    {
+        for(i = 0; i < object->stream_count; i++)
+            IMFStreamDescriptor_Release(descriptors[i]);
+        free(descriptors);
+    }
+    for (i = 0; i < object->stream_count; i++)
+    {
+        struct media_stream *stream = object->streams[i];
+
+        IMFMediaEventQueue_Release(stream->event_queue);
+        IMFStreamDescriptor_Release(stream->descriptor);
+        IMFMediaSource_Release(&stream->parent_source->IMFMediaSource_iface);
+
+        free(stream);
+    }
+    if (object->streams)
+        free(object->streams);
+    if (stream_count)
+        unix_funcs->wg_parser_disconnect(object->wg_parser);
+    if (object->read_thread)
+    {
+        object->read_thread_shutdown = true;
+        WaitForSingleObject(object->read_thread, INFINITE);
+        CloseHandle(object->read_thread);
+    }
+    if (object->wg_parser)
+        unix_funcs->wg_parser_destroy(object->wg_parser);
+    if (object->async_commands_queue)
+        MFUnlockWorkQueue(object->async_commands_queue);
+    if (object->event_queue)
+        IMFMediaEventQueue_Release(object->event_queue);
+    IMFByteStream_Release(object->byte_stream);
+    free(object);
     return hr;
 }

[PATCH v2 2/3] winegstreamer: Allocate source media buffers in the PE components.

[Derek Lesho]

This adds a allocate/blit for the input data, but this is necessary for both WOW64 support and an internal rework of the source path in wg_parser to use GstAppSrc.  Source data is usually compressed anyway, so it shouldn't affect performance too badly.

Signed-off-by: Derek Lesho 
---
v2:
- Move EOS detection to client side.
- Maintain expanding buffer for uploads instead of reallocating on every request.
---
 dlls/winegstreamer/gst_private.h   |  7 ++++---
 dlls/winegstreamer/media_source.c  | 29 +++++++++++++++++++++++++----
 dlls/winegstreamer/quartz_parser.c | 27 ++++++++++++++++++++++++---
 dlls/winegstreamer/wg_parser.c     | 26 +++++++++++++-------------
 4 files changed, 66 insertions(+), 23 deletions(-)

diff --git a/dlls/winegstreamer/gst_private.h b/dlls/winegstreamer/gst_private.h
index c6c99b1dd55..9943682facb 100644
--- a/dlls/winegstreamer/gst_private.h
+++ b/dlls/winegstreamer/gst_private.h
@@ -169,9 +169,10 @@ struct unix_funcs
     void (CDECL *wg_parser_begin_flush)(struct wg_parser *parser);
     void (CDECL *wg_parser_end_flush)(struct wg_parser *parser);

-    bool (CDECL *wg_parser_get_read_request)(struct wg_parser *parser,
-            void **data, uint64_t *offset, uint32_t *size);
-    void (CDECL *wg_parser_complete_read_request)(struct wg_parser *parser, bool ret);
+    bool (CDECL *wg_parser_get_next_read_offset)(struct wg_parser *parser,
+            uint64_t *offset, uint32_t *size);
+    void (CDECL *wg_parser_push_data)(struct wg_parser *parser,
+            const void *data, uint32_t size);

void (CDECL *wg_parser_set_unlimited_buffering)(struct wg_parser *parser);

diff --git a/dlls/winegstreamer/media_source.c b/dlls/winegstreamer/media_source.c
index 2a0bb83374d..0ff0cd73790 100644
--- a/dlls/winegstreamer/media_source.c
+++ b/dlls/winegstreamer/media_source.c
@@ -530,6 +530,11 @@ static DWORD CALLBACK read_thread(void *arg)
 {
     struct media_source *source = arg;
     IMFByteStream *byte_stream = source->byte_stream;
+    uint32_t buffer_size = 0;
+    uint64_t file_size;
+    void *data = NULL;
+
+    IMFByteStream_GetLength(byte_stream, &file_size);

TRACE("Starting read thread for media source %p.\n", source);

@@ -539,18 +544,34 @@ static DWORD CALLBACK read_thread(void *arg)
         ULONG ret_size;
         uint32_t size;
         HRESULT hr;
-        void *data;

-        if (!unix_funcs->wg_parser_get_read_request(source->wg_parser, &data, &offset, &size))
+        if (!unix_funcs->wg_parser_get_next_read_offset(source->wg_parser, &offset, &size))
             continue;

+        if (offset >= file_size)
+            size = 0;
+        else if (offset + size >= file_size)
+            size = file_size - offset;
+
+        if (size > buffer_size)
+        {
+            buffer_size = size;
+            free(data);
+            data = malloc(size);
+        }
+
+        ret_size = 0;
+
         if (SUCCEEDED(hr = IMFByteStream_SetCurrentPosition(byte_stream, offset)))
             hr = IMFByteStream_Read(byte_stream, data, size, &ret_size);
-        if (SUCCEEDED(hr) && ret_size != size)
+        if (FAILED(hr))
+            ERR("Failed to read source stream bytes %p+%u. hr=%#x\n", data, size, hr);
+        else if (ret_size != size)
             ERR("Unexpected short read: requested %u bytes, got %u.\n", size, ret_size);
-        unix_funcs->wg_parser_complete_read_request(source->wg_parser, SUCCEEDED(hr));
+        unix_funcs->wg_parser_push_data(source->wg_parser, SUCCEEDED(hr) ? data : NULL, ret_size);
     }

+    free(data);
     TRACE("Media source is shutting down; exiting.\n");
     return 0;
 }
diff --git a/dlls/winegstreamer/quartz_parser.c b/dlls/winegstreamer/quartz_parser.c
index 09a916d7f5c..5544e30dac9 100644
--- a/dlls/winegstreamer/quartz_parser.c
+++ b/dlls/winegstreamer/quartz_parser.c
@@ -785,6 +785,11 @@ static DWORD CALLBACK stream_thread(void *arg)
 static DWORD CALLBACK read_thread(void *arg)
 {
     struct parser *filter = arg;
+    LONGLONG file_size, unused;
+    uint32_t buffer_size = 0;
+    void *data = NULL;
+
+    IAsyncReader_Length(filter->reader, &file_size, &unused);

TRACE("Starting read thread for filter %p.\n", filter);

@@ -793,14 +798,30 @@ static DWORD CALLBACK read_thread(void *arg)
         uint64_t offset;
         uint32_t size;
         HRESULT hr;
-        void *data;

-        if (!unix_funcs->wg_parser_get_read_request(filter->wg_parser, &data, &offset, &size))
+        if (!unix_funcs->wg_parser_get_next_read_offset(filter->wg_parser, &offset, &size))
             continue;
+
+        if (offset >= file_size)
+            size = 0;
+        else if (offset + size >= file_size)
+            size = file_size - offset;
+
+        if (size > buffer_size)
+        {
+            buffer_size = size;
+            free(data);
+            data = malloc(size);
+        }
+
         hr = IAsyncReader_SyncRead(filter->reader, offset, size, data);
-        unix_funcs->wg_parser_complete_read_request(filter->wg_parser, SUCCEEDED(hr));
+        if (FAILED(hr))
+            ERR("Async Reader failed to failed to read %p+%u. hr=%#x\n", data, size, hr);
+
+        unix_funcs->wg_parser_push_data(filter->wg_parser, SUCCEEDED(hr) ? data : NULL, size);
     }

+    free(data);
     TRACE("Streaming stopped; exiting.\n");
     return 0;
 }
diff --git a/dlls/winegstreamer/wg_parser.c b/dlls/winegstreamer/wg_parser.c
index cd12a23d0c8..fc3ea49d0a7 100644
--- a/dlls/winegstreamer/wg_parser.c
+++ b/dlls/winegstreamer/wg_parser.c
@@ -517,8 +517,8 @@ static void CDECL wg_parser_end_flush(struct wg_parser *parser)
     pthread_mutex_unlock(&parser->mutex);
 }

-static bool CDECL wg_parser_get_read_request(struct wg_parser *parser,
-        void **data, uint64_t *offset, uint32_t *size)
+static bool CDECL wg_parser_get_next_read_offset(struct wg_parser *parser,
+        uint64_t *offset, uint32_t *size)
 {
     pthread_mutex_lock(&parser->mutex);

@@ -531,7 +531,6 @@ static bool CDECL wg_parser_get_read_request(struct wg_parser *parser,
         return false;
     }

-    *data = parser->read_request.data;
     *offset = parser->read_request.offset;
     *size = parser->read_request.size;

@@ -539,11 +538,15 @@ static bool CDECL wg_parser_get_read_request(struct wg_parser *parser,
     return true;
 }

-static void CDECL wg_parser_complete_read_request(struct wg_parser *parser, bool ret)
+static void CDECL wg_parser_push_data(struct wg_parser *parser,
+        const void *data, uint32_t size)
 {
     pthread_mutex_lock(&parser->mutex);
+    parser->read_request.size = size;
     parser->read_request.done = true;
-    parser->read_request.ret = ret;
+    parser->read_request.ret = !!data;
+    if (data)
+        memcpy(parser->read_request.data, data, size);
     parser->read_request.data = NULL;
     pthread_mutex_unlock(&parser->mutex);
     pthread_cond_signal(&parser->read_done_cond);
@@ -1190,10 +1193,6 @@ static GstFlowReturn src_getrange_cb(GstPad *pad, GstObject *parent,
     if (offset == GST_BUFFER_OFFSET_NONE)
         offset = parser->next_pull_offset;
     parser->next_pull_offset = offset + size;
-    if (offset >= parser->file_size)
-        return GST_FLOW_EOS;
-    if (offset + size >= parser->file_size)
-        size = parser->file_size - offset;

if (!*buffer)
         *buffer = new_buffer = gst_buffer_new_and_alloc(size);
@@ -1217,6 +1216,7 @@ static GstFlowReturn src_getrange_cb(GstPad *pad, GstObject *parent,
         pthread_cond_wait(&parser->read_done_cond, &parser->mutex);

ret = parser->read_request.ret;
+    gst_buffer_set_size(*buffer, parser->read_request.size);

pthread_mutex_unlock(&parser->mutex);

@@ -1224,10 +1224,10 @@ static GstFlowReturn src_getrange_cb(GstPad *pad, GstObject *parent,

GST_LOG("Request returned %d.", ret);

-    if (!ret && new_buffer)
+    if ((!ret || !size) && new_buffer)
         gst_buffer_unref(new_buffer);

-    return ret ? GST_FLOW_OK : GST_FLOW_ERROR;
+    return ret ? size ? GST_FLOW_OK : GST_FLOW_EOS : GST_FLOW_ERROR;
 }

static gboolean src_query_cb(GstPad *pad, GstObject *parent, GstQuery *query)
@@ -1918,8 +1918,8 @@ static const struct unix_funcs funcs =
     wg_parser_begin_flush,
     wg_parser_end_flush,

-    wg_parser_get_read_request,
-    wg_parser_complete_read_request,
+    wg_parser_get_next_read_offset,
+    wg_parser_push_data,

wg_parser_set_unlimited_buffering,

[PATCH v2 2/3] winegstreamer: Allocate source media buffers in the PE components.

[Zebediah Figura]

On 9/10/21 12:04 PM, Derek Lesho wrote:
> @@ -539,18 +544,34 @@ static DWORD CALLBACK read_thread(void *arg)
>           ULONG ret_size;
>           uint32_t size;
>           HRESULT hr;
> -
>    void *data;
>
> -
>    if (!unix_funcs->wg_parser_get_read_request(source->wg_parser, &data, &offset, &size))
> -
>    if (!unix_funcs->wg_parser_get_next_read_offset(source->wg_parser, &offset, &size))
>         continue;
>
> -
>    if (offset >= file_size)
> -
>        size = 0;
> -
>    else if (offset + size >= file_size)
> -
>        size = file_size - offset;
> -
> -
>    if (size > buffer_size)
> -
>    {
> -
>        buffer_size = size;
> -
>        free(data);
> -
>        data = malloc(size);
> -
>    }

Any reason not to use realloc() here?
> -
> -
>    ret_size = 0;
> -
>     if (SUCCEEDED(hr = IMFByteStream_SetCurrentPosition(byte_stream, offset)))
>         hr = IMFByteStream_Read(byte_stream, data, size, &ret_size);
> -
>    if (SUCCEEDED(hr) && ret_size != size)
> -
>    if (FAILED(hr))
> -
>        ERR("Failed to read source stream bytes %p+%u. hr=%#x\n", data, size, hr);
> -
>    else if (ret_size != size)
>         ERR("Unexpected short read: requested %u bytes, got %u.\n", size, ret_size);
> -
>    unix_funcs->wg_parser_complete_read_request(source->wg_parser, SUCCEEDED(hr));
> -
>    unix_funcs->wg_parser_push_data(source->wg_parser, SUCCEEDED(hr) ? data : NULL, ret_size);
> }
>
> -
>
> free(data);
>  TRACE("Media source is shutting down; exiting.\n");
>  return 0;
> }

...
> @@ -1224,10 +1224,10 @@ static GstFlowReturn src_getrange_cb(GstPad *pad, GstObject *parent,
>   GST_LOG("Request returned %d.", ret);
>
> - if (!ret && new_buffer)
> - if ((!ret || !size) && new_buffer)
>      gst_buffer_unref(new_buffer);
> - return ret ? GST_FLOW_OK : GST_FLOW_ERROR;
> - return ret ? size ? GST_FLOW_OK : GST_FLOW_EOS : GST_FLOW_ERROR;

Please don't do this; it's hard to read.
> }
>
> static gboolean src_query_cb(GstPad *pad, GstObject *parent, GstQuery *query)

[PATCH v2 3/3] winegstreamer: Replace source pad interface with GstAppSrc.

[Derek Lesho]

Signed-off-by: Derek Lesho 
---
v2:
- Remodel read_request to only store a flag and the request size.
- Fix race condition documented at https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/issues/937
- Send error message to bus to indicate a read error, instead of using an EOS.  (gst_app_src_create will now unblock on shutdown).
---
 dlls/winegstreamer/wg_parser.c | 395 +++++++++------------------------
 1 file changed, 99 insertions(+), 296 deletions(-)

diff --git a/dlls/winegstreamer/wg_parser.c b/dlls/winegstreamer/wg_parser.c
index fc3ea49d0a7..68b26d7093c 100644
--- a/dlls/winegstreamer/wg_parser.c
+++ b/dlls/winegstreamer/wg_parser.c
@@ -33,6 +33,7 @@
 #include <gst/gst.h>
 #include <gst/video/video.h>
 #include <gst/audio/audio.h>
+#include <gst/app/gstappsrc.h>

/* GStreamer callbacks may be called on threads not created by Wine, and
  * therefore cannot access the Wine TEB. This means that we must use GStreamer
@@ -49,28 +50,22 @@ struct wg_parser
     struct wg_parser_stream **streams;
     unsigned int stream_count;

-    GstElement *container, *decodebin;
+    GstElement *container, *appsrc, *decodebin;
     GstBus *bus;
-    GstPad *my_src, *their_sink;

-    guint64 file_size, start_offset, next_offset, stop_offset;
+    guint64 file_size;
     guint64 next_pull_offset;

-    pthread_t push_thread;
-
     pthread_mutex_t mutex;

pthread_cond_t init_cond;
     bool no_more_pads, has_duration, error;

-    pthread_cond_t read_cond, read_done_cond;
+    pthread_cond_t read_cond;
     struct
     {
-        void *data;
-        uint64_t offset;
         uint32_t size;
-        bool done;
-        bool ret;
+        bool pending_read;
     } read_request;

bool flushing, sink_connected;
@@ -522,7 +517,7 @@ static bool CDECL wg_parser_get_next_read_offset(struct wg_parser *parser,
 {
     pthread_mutex_lock(&parser->mutex);

-    while (parser->sink_connected && !parser->read_request.data)
+    while (parser->sink_connected && !parser->read_request.pending_read)
         pthread_cond_wait(&parser->read_cond, &parser->mutex);

if (!parser->sink_connected)
@@ -531,7 +526,7 @@ static bool CDECL wg_parser_get_next_read_offset(struct wg_parser *parser,
         return false;
     }

-    *offset = parser->read_request.offset;
+    *offset = parser->next_pull_offset;
     *size = parser->read_request.size;

pthread_mutex_unlock(&parser->mutex);
@@ -541,15 +536,63 @@ static bool CDECL wg_parser_get_next_read_offset(struct wg_parser *parser,
 static void CDECL wg_parser_push_data(struct wg_parser *parser,
         const void *data, uint32_t size)
 {
+    GstBuffer *buffer;
+    GstFlowReturn ret;
+    GError *error;
+    GstMessage *message;
+
+    if (!data)
+    {
+        pthread_mutex_lock(&parser->mutex);
+
+        error = g_error_new(G_FILE_ERROR, G_FILE_ERROR_FAILED, "WG-Parser client failed to read data at offset %" G_GUINT64_FORMAT, parser->next_pull_offset);
+        message = gst_message_new_error(NULL, error, "");
+        if (!gst_bus_post(parser->bus, message))
+        {
+            GST_ERROR("Failed to post error message to bus!\n");
+            gst_message_unref(message);
+        }
+        parser->read_request.pending_read = false;
+
+        pthread_mutex_unlock(&parser->mutex);
+        return;
+    }
+
+    if (!size)
+    {
+        pthread_mutex_lock(&parser->mutex);
+
+        g_signal_emit_by_name(G_OBJECT(parser->appsrc), "end-of-stream", &ret);
+        parser->read_request.pending_read = false;
+
+        pthread_mutex_unlock(&parser->mutex);
+        return;
+    }
+
+    /*  We could avoid this extra copy using gst_buffer_new_wrapped.
+        However, PE wouldn't know when to release the buffer allocations as the buffer
+        objects are queued, so we'd have to create a ring buffer the size of the gstappsrc
+        queue on the PE side and validate that we don't overrun on the unix side.  I'm not
+        yet convinced that trying to reduce copies of compressed data is worth the
+        complexity.  */
+    buffer = gst_buffer_new_and_alloc(size);
+    gst_buffer_fill(buffer, 0, data, size);
+
     pthread_mutex_lock(&parser->mutex);
-    parser->read_request.size = size;
-    parser->read_request.done = true;
-    parser->read_request.ret = !!data;
-    if (data)
-        memcpy(parser->read_request.data, data, size);
-    parser->read_request.data = NULL;
+    assert(parser->read_request.pending_read);
+
+    GST_BUFFER_OFFSET(buffer) = parser->next_pull_offset;
+    g_signal_emit_by_name(G_OBJECT(parser->appsrc), "push-buffer", buffer, &ret);
+
+    /* In random-access mode, GST_FLOW_EOS shouldn't be returned */
+    assert(ret == GST_FLOW_OK || ret == GST_FLOW_FLUSHING);
+    if (ret == GST_FLOW_OK)
+        parser->next_pull_offset += size;
+    else
+        gst_buffer_unref(buffer);
+
+    parser->read_request.pending_read = false;
     pthread_mutex_unlock(&parser->mutex);
-    pthread_cond_signal(&parser->read_done_cond);
 }

static void CDECL wg_parser_set_unlimited_buffering(struct wg_parser *parser)
@@ -1180,194 +1223,42 @@ static void pad_removed_cb(GstElement *element, GstPad *pad, gpointer user)
     g_free(name);
 }

-static GstFlowReturn src_getrange_cb(GstPad *pad, GstObject *parent,
-        guint64 offset, guint size, GstBuffer **buffer)
+static void src_need_data(GstElement *appsrc, guint length, gpointer user)
 {
-    struct wg_parser *parser = gst_pad_get_element_private(pad);
-    GstBuffer *new_buffer = NULL;
-    GstMapInfo map_info;
-    bool ret;
-
-    GST_LOG("pad %p, offset %" G_GINT64_MODIFIER "u, length %u, buffer %p.", pad, offset, size, *buffer);
-
-    if (offset == GST_BUFFER_OFFSET_NONE)
-        offset = parser->next_pull_offset;
-    parser->next_pull_offset = offset + size;
-
-    if (!*buffer)
-        *buffer = new_buffer = gst_buffer_new_and_alloc(size);
-
-    gst_buffer_map(*buffer, &map_info, GST_MAP_WRITE);
+    struct wg_parser *parser = user;
+    guint64 queued_bytes;

pthread_mutex_lock(&parser->mutex);

-    assert(!parser->read_request.data);
-    parser->read_request.data = map_info.data;
-    parser->read_request.offset = offset;
-    parser->read_request.size = size;
-    parser->read_request.done = false;
-    pthread_cond_signal(&parser->read_cond);
-
-    /* Note that we don't unblock this wait on GST_EVENT_FLUSH_START. We expect
-     * the upstream pin to flush if necessary. We should never be blocked on
-     * read_thread() not running. */
-
-    while (!parser->read_request.done)
-        pthread_cond_wait(&parser->read_done_cond, &parser->mutex);
-
-    ret = parser->read_request.ret;
-    gst_buffer_set_size(*buffer, parser->read_request.size);
-
-    pthread_mutex_unlock(&parser->mutex);
-
-    gst_buffer_unmap(*buffer, &map_info);
-
-    GST_LOG("Request returned %d.", ret);
-
-    if ((!ret || !size) && new_buffer)
-        gst_buffer_unref(new_buffer);
-
-    return ret ? size ? GST_FLOW_OK : GST_FLOW_EOS : GST_FLOW_ERROR;
-}
-
-static gboolean src_query_cb(GstPad *pad, GstObject *parent, GstQuery *query)
-{
-    struct wg_parser *parser = gst_pad_get_element_private(pad);
-    GstFormat format;
-
-    GST_LOG("parser %p, type %s.", parser, GST_QUERY_TYPE_NAME(query));
-
-    switch (GST_QUERY_TYPE(query))
+    /* Workaround for GStreamer bug: https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/issues/937
+       current-level-buffers is a closer fit but it's a recent addition. */
+    g_object_get(G_OBJECT(appsrc), "current-level-bytes", &queued_bytes, NULL);
+    if (queued_bytes)
     {
-        case GST_QUERY_DURATION:
-            gst_query_parse_duration(query, &format, NULL);
-            if (format == GST_FORMAT_PERCENT)
-            {
-                gst_query_set_duration(query, GST_FORMAT_PERCENT, GST_FORMAT_PERCENT_MAX);
-                return TRUE;
-            }
-            else if (format == GST_FORMAT_BYTES)
-            {
-                gst_query_set_duration(query, GST_FORMAT_BYTES, parser->file_size);
-                return TRUE;
-            }
-            return FALSE;
-
-        case GST_QUERY_SEEKING:
-            gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
-            if (format != GST_FORMAT_BYTES)
-            {
-                GST_WARNING("Cannot seek using format "%s".", gst_format_get_name(format));
-                return FALSE;
-            }
-            gst_query_set_seeking(query, GST_FORMAT_BYTES, 1, 0, parser->file_size);
-            return TRUE;
-
-        case GST_QUERY_SCHEDULING:
-            gst_query_set_scheduling(query, GST_SCHEDULING_FLAG_SEEKABLE, 1, -1, 0);
-            gst_query_add_scheduling_mode(query, GST_PAD_MODE_PUSH);
-            gst_query_add_scheduling_mode(query, GST_PAD_MODE_PULL);
-            return TRUE;
-
-        default:
-            GST_WARNING("Unhandled query type %s.", GST_QUERY_TYPE_NAME(query));
-            return FALSE;
-    }
-}
-
-static void *push_data(void *arg)
-{
-    struct wg_parser *parser = arg;
-    GstBuffer *buffer;
-    guint max_size;
-
-    GST_DEBUG("Starting push thread.");
-
-    if (!(buffer = gst_buffer_new_allocate(NULL, 16384, NULL)))
-    {
-        GST_ERROR("Failed to allocate memory.");
-        return NULL;
-    }
-
-    max_size = parser->stop_offset ? parser->stop_offset : parser->file_size;
-
-    for (;;)
-    {
-        ULONG size;
-        int ret;
-
-        if (parser->next_offset >= max_size)
-            break;
-        size = min(16384, max_size - parser->next_offset);
-
-        if ((ret = src_getrange_cb(parser->my_src, NULL, parser->next_offset, size, &buffer)) < 0)
-        {
-            GST_ERROR("Failed to read data, ret %s.", gst_flow_get_name(ret));
-            break;
-        }
-
-        parser->next_offset += size;
-
-        buffer->duration = buffer->pts = -1;
-        if ((ret = gst_pad_push(parser->my_src, buffer)) < 0)
-        {
-            GST_ERROR("Failed to push data, ret %s.", gst_flow_get_name(ret));
-            break;
-        }
+        pthread_mutex_unlock(&parser->mutex);
+        return;
     }

-    gst_buffer_unref(buffer);
-
-    gst_pad_push_event(parser->my_src, gst_event_new_eos());
+    parser->read_request.pending_read = true;
+    parser->read_request.size = length;

-    GST_DEBUG("Stopping push thread.");
+    pthread_cond_signal(&parser->read_cond);

-    return NULL;
+    pthread_mutex_unlock(&parser->mutex);
 }

-static gboolean activate_push(GstPad *pad, gboolean activate)
+static gboolean src_seek_data(GstElement *appsrc, guint64 offset, gpointer user)
 {
-    struct wg_parser *parser = gst_pad_get_element_private(pad);
-
-    if (!activate)
-    {
-        if (parser->push_thread)
-        {
-            pthread_join(parser->push_thread, NULL);
-            parser->push_thread = 0;
-        }
-    }
-    else if (!parser->push_thread)
-    {
-        int ret;
+    struct wg_parser *parser = user;

-        if ((ret = pthread_create(&parser->push_thread, NULL, push_data, parser)))
-        {
-            GST_ERROR("Failed to create push thread: %s", strerror(errno));
-            parser->push_thread = 0;
-            return FALSE;
-        }
-    }
-    return TRUE;
-}
+    pthread_mutex_lock(&parser->mutex);

-static gboolean src_activate_mode_cb(GstPad *pad, GstObject *parent, GstPadMode mode, gboolean activate)
-{
-    struct wg_parser *parser = gst_pad_get_element_private(pad);
+    assert(!parser->read_request.pending_read);
+    parser->next_pull_offset = offset;

-    GST_DEBUG("%s source pad for parser %p in %s mode.",
-            activate ? "Activating" : "Deactivating", parser, gst_pad_mode_get_name(mode));
+    pthread_mutex_unlock(&parser->mutex);

-    switch (mode)
-    {
-        case GST_PAD_MODE_PULL:
-            return TRUE;
-        case GST_PAD_MODE_PUSH:
-            return activate_push(pad, activate);
-        case GST_PAD_MODE_NONE:
-            break;
-    }
-    return FALSE;
+    return true;
 }

static GstBusSyncReply bus_handler_cb(GstBus *bus, GstMessage *msg, gpointer user)
@@ -1414,85 +1305,8 @@ static GstBusSyncReply bus_handler_cb(GstBus *bus, GstMessage *msg, gpointer use
     return GST_BUS_DROP;
 }

-static gboolean src_perform_seek(struct wg_parser *parser, GstEvent *event)
-{
-    BOOL thread = !!parser->push_thread;
-    GstSeekType cur_type, stop_type;
-    GstFormat seek_format;
-    GstEvent *flush_event;
-    GstSeekFlags flags;
-    gint64 cur, stop;
-    guint32 seqnum;
-    gdouble rate;
-
-    gst_event_parse_seek(event, &rate, &seek_format, &flags,
-                         &cur_type, &cur, &stop_type, &stop);
-
-    if (seek_format != GST_FORMAT_BYTES)
-    {
-        GST_FIXME("Unhandled format "%s".", gst_format_get_name(seek_format));
-        return FALSE;
-    }
-
-    seqnum = gst_event_get_seqnum(event);
-
-    /* send flush start */
-    if (flags & GST_SEEK_FLAG_FLUSH)
-    {
-        flush_event = gst_event_new_flush_start();
-        gst_event_set_seqnum(flush_event, seqnum);
-        gst_pad_push_event(parser->my_src, flush_event);
-        if (thread)
-            gst_pad_set_active(parser->my_src, 1);
-    }
-
-    parser->next_offset = parser->start_offset = cur;
-
-    /* and prepare to continue streaming */
-    if (flags & GST_SEEK_FLAG_FLUSH)
-    {
-        flush_event = gst_event_new_flush_stop(TRUE);
-        gst_event_set_seqnum(flush_event, seqnum);
-        gst_pad_push_event(parser->my_src, flush_event);
-        if (thread)
-            gst_pad_set_active(parser->my_src, 1);
-    }
-
-    return TRUE;
-}
-
-static gboolean src_event_cb(GstPad *pad, GstObject *parent, GstEvent *event)
-{
-    struct wg_parser *parser = gst_pad_get_element_private(pad);
-    gboolean ret = TRUE;
-
-    GST_LOG("parser %p, type "%s".", parser, GST_EVENT_TYPE_NAME(event));
-
-    switch (event->type)
-    {
-        case GST_EVENT_SEEK:
-            ret = src_perform_seek(parser, event);
-            break;
-
-        case GST_EVENT_FLUSH_START:
-        case GST_EVENT_FLUSH_STOP:
-        case GST_EVENT_QOS:
-        case GST_EVENT_RECONFIGURE:
-            break;
-
-        default:
-            GST_WARNING("Ignoring "%s" event.", GST_EVENT_TYPE_NAME(event));
-            ret = FALSE;
-            break;
-    }
-    gst_event_unref(event);
-    return ret;
-}
-
 static HRESULT CDECL wg_parser_connect(struct wg_parser *parser, uint64_t file_size)
 {
-    GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE("quartz_src",
-            GST_PAD_SRC, GST_PAD_ALWAYS, GST_STATIC_CAPS_ANY);
     unsigned int i;

parser->file_size = file_size;
@@ -1507,14 +1321,15 @@ static HRESULT CDECL wg_parser_connect(struct wg_parser *parser, uint64_t file_s
     parser->container = gst_bin_new(NULL);
     gst_element_set_bus(parser->container, parser->bus);

-    parser->my_src = gst_pad_new_from_static_template(&src_template, "quartz-src");
-    gst_pad_set_getrange_function(parser->my_src, src_getrange_cb);
-    gst_pad_set_query_function(parser->my_src, src_query_cb);
-    gst_pad_set_activatemode_function(parser->my_src, src_activate_mode_cb);
-    gst_pad_set_event_function(parser->my_src, src_event_cb);
-    gst_pad_set_element_private(parser->my_src, parser);
+    if (!(parser->appsrc = create_element("appsrc", "base")))
+        return E_FAIL;
+    gst_bin_add(GST_BIN(parser->container), parser->appsrc);
+
+    g_object_set(parser->appsrc, "stream-type", GST_APP_STREAM_TYPE_RANDOM_ACCESS, NULL);
+    g_object_set(parser->appsrc, "size", parser->file_size, NULL);
+    g_signal_connect(parser->appsrc, "need-data", G_CALLBACK(src_need_data), parser);
+    g_signal_connect(parser->appsrc, "seek-data", G_CALLBACK(src_seek_data), parser);

-    parser->start_offset = parser->next_offset = parser->stop_offset = 0;
     parser->next_pull_offset = 0;

if (!parser->init_gst(parser))
@@ -1594,7 +1409,6 @@ static HRESULT CDECL wg_parser_connect(struct wg_parser *parser, uint64_t file_s

pthread_mutex_unlock(&parser->mutex);

-    parser->next_offset = 0;
     return S_OK;
 }

@@ -1636,10 +1450,6 @@ static void CDECL wg_parser_disconnect(struct wg_parser *parser)
     pthread_mutex_unlock(&parser->mutex);

gst_element_set_state(parser->container, GST_STATE_NULL);
-    gst_pad_unlink(parser->my_src, parser->their_sink);
-    gst_object_unref(parser->my_src);
-    gst_object_unref(parser->their_sink);
-    parser->my_src = parser->their_sink = NULL;

pthread_mutex_lock(&parser->mutex);
     parser->sink_connected = false;
@@ -1674,15 +1484,13 @@ static BOOL decodebin_parser_init_gst(struct wg_parser *parser)
     g_signal_connect(element, "autoplug-select", G_CALLBACK(autoplug_select_cb), parser);
     g_signal_connect(element, "no-more-pads", G_CALLBACK(no_more_pads_cb), parser);

-    parser->their_sink = gst_element_get_static_pad(element, "sink");
-
     pthread_mutex_lock(&parser->mutex);
     parser->no_more_pads = parser->error = false;
     pthread_mutex_unlock(&parser->mutex);

-    if ((ret = gst_pad_link(parser->my_src, parser->their_sink)) < 0)
+    if (!gst_element_link(parser->appsrc, parser->decodebin))
     {
-        GST_ERROR("Failed to link pads, error %d.\n", ret);
+        GST_ERROR("Failed to link app source.\n");
         return FALSE;
     }

@@ -1721,15 +1529,13 @@ static BOOL avi_parser_init_gst(struct wg_parser *parser)
     g_signal_connect(element, "pad-removed", G_CALLBACK(pad_removed_cb), parser);
     g_signal_connect(element, "no-more-pads", G_CALLBACK(no_more_pads_cb), parser);

-    parser->their_sink = gst_element_get_static_pad(element, "sink");
-
     pthread_mutex_lock(&parser->mutex);
     parser->no_more_pads = parser->error = false;
     pthread_mutex_unlock(&parser->mutex);

-    if ((ret = gst_pad_link(parser->my_src, parser->their_sink)) < 0)
+    if (!gst_element_link(parser->appsrc, element))
     {
-        GST_ERROR("Failed to link pads, error %d.\n", ret);
+        GST_ERROR("Failed to link app source.\n");
         return FALSE;
     }

@@ -1765,10 +1571,9 @@ static BOOL mpeg_audio_parser_init_gst(struct wg_parser *parser)

gst_bin_add(GST_BIN(parser->container), element);

-    parser->their_sink = gst_element_get_static_pad(element, "sink");
-    if ((ret = gst_pad_link(parser->my_src, parser->their_sink)) < 0)
+    if (!gst_element_link(parser->appsrc, element))
     {
-        GST_ERROR("Failed to link sink pads, error %d.\n", ret);
+        GST_ERROR("Failed to link app source.\n");
         return FALSE;
     }

@@ -1805,10 +1610,9 @@ static BOOL wave_parser_init_gst(struct wg_parser *parser)

gst_bin_add(GST_BIN(parser->container), element);

-    parser->their_sink = gst_element_get_static_pad(element, "sink");
-    if ((ret = gst_pad_link(parser->my_src, parser->their_sink)) < 0)
+    if (!gst_element_link(parser->appsrc, element))
     {
-        GST_ERROR("Failed to link sink pads, error %d.\n", ret);
+        GST_ERROR("Failed to link app source.\n");
         return FALSE;
     }

@@ -1845,8 +1649,8 @@ static struct wg_parser *wg_parser_create(void)
     pthread_mutex_init(&parser->mutex, NULL);
     pthread_cond_init(&parser->init_cond, NULL);
     pthread_cond_init(&parser->read_cond, NULL);
-    pthread_cond_init(&parser->read_done_cond, NULL);
     parser->flushing = true;
+    parser->read_request.pending_read = false;

GST_DEBUG("Created winegstreamer parser %p.\n", parser);
     return parser;
@@ -1899,7 +1703,6 @@ static void CDECL wg_parser_destroy(struct wg_parser *parser)
     pthread_mutex_destroy(&parser->mutex);
     pthread_cond_destroy(&parser->init_cond);
     pthread_cond_destroy(&parser->read_cond);
-    pthread_cond_destroy(&parser->read_done_cond);

free(parser);
 }

[PATCH v2 3/3] winegstreamer: Replace source pad interface with GstAppSrc.

[Zebediah Figura]

On 9/10/21 12:04 PM, Derek Lesho wrote:
> Signed-off-by: Derek Lesho
>
> v2:
> - Remodel read_request to only store a flag and the request size.
> - Fix race condition documented at https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/issues/937
> - Send error message to bus to indicate a read error, instead of using an EOS.  (gst_app_src_create will now unblock on shutdown).
>
> dlls/winegstreamer/wg_parser.c | 395 +++++++++------------------------
>   1 file changed, 99 insertions(+), 296 deletions(-)
>
> diff --git a/dlls/winegstreamer/wg_parser.c b/dlls/winegstreamer/wg_parser.c
> index fc3ea49d0a7..68b26d7093c 100644
> --- a/dlls/winegstreamer/wg_parser.c
> +++ b/dlls/winegstreamer/wg_parser.c
> @@ -33,6 +33,7 @@
>   #include <gst/gst.h>
>   #include <gst/video/video.h>
>   #include <gst/audio/audio.h>
> +#include <gst/app/gstappsrc.h>
>
> /* GStreamer callbacks may be called on threads not created by Wine, and
> - therefore cannot access the Wine TEB. This means that we must use GStreamer
>
> @@ -49,28 +50,22 @@ struct wg_parser
>       struct wg_parser_stream **streams;
>       unsigned int stream_count;
> - GstElement *container, *decodebin;
> - GstElement *container, *appsrc, *decodebin;
>  GstBus *bus;
> -
>
> GstPad *my_src, *their_sink;
> -
>
> guint64 file_size, start_offset, next_offset, stop_offset;
> - guint64 file_size;
>  guint64 next_pull_offset;
> -
>
> pthread_t push_thread;
> -
> pthread_mutex_t mutex;
>
> pthread_cond_t init_cond;
> bool no_more_pads, has_duration, error;
>
> -
>
> pthread_cond_t read_cond, read_done_cond;
> - pthread_cond_t read_cond;
>  struct
>  {
> -
>    void *data;
> -
>    uint64_t offset;
>     uint32_t size;
> -
>    bool done;
> -
>    bool ret;
> -
>    bool pending_read;
> } read_request;
>
> bool flushing, sink_connected;

I would personally be inclined to either keep "read_request.offset" 
rather than "next_pull_offset", or get rid of the read_request 
structure. Probably the former.
> @@ -522,7 +517,7 @@ static bool CDECL wg_parser_get_next_read_offset(struct wg_parser *parser,
>   {
>       pthread_mutex_lock(&parser->mutex);
> - while (parser->sink_connected && !parser->read_request.data)
> -
>
> while (parser->sink_connected && !parser->read_request.pending_read)
>      pthread_cond_wait(&parser->read_cond, &parser->mutex);
>
> if (!parser->sink_connected)
>
> @@ -531,7 +526,7 @@ static bool CDECL wg_parser_get_next_read_offset(struct wg_parser *parser,
>           return false;
>       }
> - *offset = parser->read_request.offset;
> -
>
> *offset = parser->next_pull_offset;
>  *size = parser->read_request.size;
>
> pthread_mutex_unlock(&parser->mutex);
>
> @@ -541,15 +536,63 @@ static bool CDECL wg_parser_get_next_read_offset(struct wg_parser *parser,
>   static void CDECL wg_parser_push_data(struct wg_parser *parser,
>           const void *data, uint32_t size)
>   {
> - GstBuffer *buffer;
> - GstFlowReturn ret;
> - GError *error;
> - GstMessage *message;
> -
> - if (!data)
> - {
> -
>    pthread_mutex_lock(&parser->mutex);
> -
> -
>    error = g_error_new(G_FILE_ERROR, G_FILE_ERROR_FAILED, "WG-Parser client failed to read data at offset %" G_GUINT64_FORMAT, parser->next_pull_offset);
> -
>    message = gst_message_new_error(NULL, error, "");
> -
>    if (!gst_bus_post(parser->bus, message))
> -
>    {
> -
>        GST_ERROR("Failed to post error message to bus!\n");
> -
>        gst_message_unref(message);
> -
>    }

Is there a point in posting a message to the bus?
> -
>    parser->read_request.pending_read = false;

This is fine temporarily. Ultimately I think we should be terminating 
the client-side thread on error, and if possible not calling 
wg_parser_push_data() at all.
> -
> -
>    pthread_mutex_unlock(&parser->mutex);
> -
>    return;
> - }
> -
> - if (!size)
> - {
> -
>    pthread_mutex_lock(&parser->mutex);
> -
> -
>    g_signal_emit_by_name(G_OBJECT(parser->appsrc), "end-of-stream", &ret);
> -
>    parser->read_request.pending_read = false;
> -
> -
>    pthread_mutex_unlock(&parser->mutex);
> -
>    return;
> - }
> -
> - /*  We could avoid this extra copy using gst_buffer_new_wrapped.
> -
>    However, PE wouldn't know when to release the buffer allocations as the buffer
> -
>    objects are queued, so we'd have to create a ring buffer the size of the gstappsrc
> -
>    queue on the PE side and validate that we don't overrun on the unix side.  I'm not
> -
>    yet convinced that trying to reduce copies of compressed data is worth the
> -
>    complexity.  */
> - buffer = gst_buffer_new_and_alloc(size);
> - gst_buffer_fill(buffer, 0, data, size);
> -
> pthread_mutex_lock(&parser->mutex);
> - parser->read_request.size = size;
> - parser->read_request.done = true;
> - parser->read_request.ret = !!data;
> - if (data)
> -
>    memcpy(parser->read_request.data, data, size);
> - parser->read_request.data = NULL;
> - assert(parser->read_request.pending_read);
> -
> - GST_BUFFER_OFFSET(buffer) = parser->next_pull_offset;
> - g_signal_emit_by_name(G_OBJECT(parser->appsrc), "push-buffer", buffer, &ret);
> -
> - /* In random-access mode, GST_FLOW_EOS shouldn't be returned */
> - assert(ret == GST_FLOW_OK || ret == GST_FLOW_FLUSHING);
> - if (ret == GST_FLOW_OK)
> -
>    parser->next_pull_offset += size;
> - else
> -
>    gst_buffer_unref(buffer);

According to my reading of the source code, GstAppSrc will take the 
reference even if it returns GST_FLOW_FLUSHING. That's a bit unintuitive 
and should probably be clearly documented on the GStreamer side, but 
anyway...
> -
> - parser->read_request.pending_read = false;
>  pthread_mutex_unlock(&parser->mutex);
> -
>
> pthread_cond_signal(&parser->read_done_cond);
> }
>
> static void CDECL wg_parser_set_unlimited_buffering(struct wg_parser *parser)

[PATCH v2 3/3] winegstreamer: Replace source pad interface with GstAppSrc.

[Derek Lesho]

On 9/13/21 1:57 PM, Zebediah Figura wrote:
> > @@ -541,15 +536,63 @@ static bool CDECL 
> > wg_parser_get_next_read_offset(struct wg_parser *parser,
> >   static void CDECL wg_parser_push_data(struct wg_parser *parser,
> >           const void *data, uint32_t size)
> >   {
> > +    GstBuffer *buffer;
> > +    GstFlowReturn ret;
> > +    GError *error;
> > +    GstMessage *message;
> > -
> >
> > +    if (!data)
> > +    {
> > +        pthread_mutex_lock(&parser->mutex);
> > -
> >
> > +        error = g_error_new(G_FILE_ERROR, G_FILE_ERROR_FAILED, 
> > "WG-Parser client failed to read data at offset %" G_GUINT64_FORMAT, 
> > parser->next_pull_offset);
> > +        message = gst_message_new_error(NULL, error, "");
> > +        if (!gst_bus_post(parser->bus, message))
> > +        {
> > +            GST_ERROR("Failed to post error message to bus!\n");
> > +            gst_message_unref(message);
> > +        }
>
> Is there a point in posting a message to the bus?

The point would be to have a unified error path when something goes 
wrong.  (whether the error be GStreamer internal or a read failure)
> > - parser->read_request.pending_read = false;
>
> This is fine temporarily. Ultimately I think we should be terminating 
> the client-side thread on error, and if possible not calling 
> wg_parser_push_data() at all.

It would be a little tricky to do so given that this thread exists 
throughout initialization, and not signaling the error to GStreamer 
would mean we have to add yet another shutdown path that doesn't 
interfere with the others.
> > -
> >
> > +        pthread_mutex_unlock(&parser->mutex);
> > +        return;
> > +    }
> > -
> >
> > +    if (!size)
> > +    {
> > +        pthread_mutex_lock(&parser->mutex);
> > -
> >
> > +        g_signal_emit_by_name(G_OBJECT(parser->appsrc), 
> > "end-of-stream", &ret);
> > +        parser->read_request.pending_read = false;
> > -
> >
> > +        pthread_mutex_unlock(&parser->mutex);
> > +        return;
> > +    }
> > -
> >
> > +    /*  We could avoid this extra copy using gst_buffer_new_wrapped.
> > +        However, PE wouldn't know when to release the buffer 
> > allocations as the buffer
> > +        objects are queued, so we'd have to create a ring buffer the 
> > size of the gstappsrc
> > +        queue on the PE side and validate that we don't overrun on 
> > the unix side.  I'm not
> > +        yet convinced that trying to reduce copies of compressed 
> > data is worth the
> > +        complexity.  */
> > +    buffer = gst_buffer_new_and_alloc(size);
> > +    gst_buffer_fill(buffer, 0, data, size);
> > -
> >
> > pthread_mutex_lock(&parser->mutex);
> > -    parser->read_request.size = size;
> > -    parser->read_request.done = true;
> > -    parser->read_request.ret = !!data;
> > -    if (data)
> > -        memcpy(parser->read_request.data, data, size);
> > -    parser->read_request.data = NULL;
> > +    assert(parser->read_request.pending_read);
> > -
> >
> > +    GST_BUFFER_OFFSET(buffer) = parser->next_pull_offset;
> > +    g_signal_emit_by_name(G_OBJECT(parser->appsrc), "push-buffer", 
> > buffer, &ret);
> > -
> >
> > +    /* In random-access mode, GST_FLOW_EOS shouldn't be returned */
> > +    assert(ret == GST_FLOW_OK || ret == GST_FLOW_FLUSHING);
> > +    if (ret == GST_FLOW_OK)
> > +        parser->next_pull_offset += size;
> > +    else
> > +        gst_buffer_unref(buffer);
>
> According to my reading of the source code, GstAppSrc will take the 
> reference even if it returns GST_FLOW_FLUSHING. That's a bit 
> unintuitive and should probably be clearly documented on the GStreamer 
> side, but anyway...

Good catch, thanks.
> > -
> >
> > +    parser->read_request.pending_read = false;
> >       pthread_mutex_unlock(&parser->mutex);
> > -    pthread_cond_signal(&parser->read_done_cond);
> >   }
> >     static void CDECL wg_parser_set_unlimited_buffering(struct 
> > wg_parser *parser)

[PATCH v2 3/3] winegstreamer: Replace source pad interface with GstAppSrc.

[Zebediah Figura]

On 9/13/21 1:08 PM, Derek Lesho wrote:
> On 9/13/21 1:57 PM, Zebediah Figura wrote:
> > > @@ -541,15 +536,63 @@ static bool CDECL
> > > wg_parser_get_next_read_offset(struct wg_parser *parser,
> > >    static void CDECL wg_parser_push_data(struct wg_parser *parser,
> > >            const void *data, uint32_t size)
> > >    {
> > > +    GstBuffer *buffer;
> > > +    GstFlowReturn ret;
> > > +    GError *error;
> > > +    GstMessage *message;
> > > -
> > >
> > > +    if (!data)
> > > +    {
> > > +        pthread_mutex_lock(&parser->mutex);
> > > -
> > >
> > > +        error = g_error_new(G_FILE_ERROR, G_FILE_ERROR_FAILED,
> > > "WG-Parser client failed to read data at offset %" G_GUINT64_FORMAT,
> > > parser->next_pull_offset);
> > > +        message = gst_message_new_error(NULL, error, "");
> > > +        if (!gst_bus_post(parser->bus, message))
> > > +        {
> > > +            GST_ERROR("Failed to post error message to bus!\n");
> > > +            gst_message_unref(message);
> > > +        }
> >
> > Is there a point in posting a message to the bus?
>
> The point would be to have a unified error path when something goes
> wrong.  (whether the error be GStreamer internal or a read failure)
> >

Oh, I guess we would probably need to unblock initialization here; I 
forgot that we did something other than just printing an error. Okay, 
that makes sense.

I don't think there's any point checking for failure from 
gst_bus_post(), though.
> > > - parser->read_request.pending_read = false;
> >
> > This is fine temporarily. Ultimately I think we should be terminating
> > the client-side thread on error, and if possible not calling
> > wg_parser_push_data() at all.
>
> It would be a little tricky to do so given that this thread exists
> throughout initialization, and not signaling the error to GStreamer
> would mean we have to add yet another shutdown path that doesn't
> interfere with the others.
> > > -
> > >
> > > +        pthread_mutex_unlock(&parser->mutex);
> > > +        return;
> > > +    }
> > > -
> > >
> > > +    if (!size)
> > > +    {
> > > +        pthread_mutex_lock(&parser->mutex);
> > > -
> > >
> > > +        g_signal_emit_by_name(G_OBJECT(parser->appsrc),
> > > "end-of-stream", &ret);
> > > +        parser->read_request.pending_read = false;
> > > -
> > >
> > > +        pthread_mutex_unlock(&parser->mutex);
> > > +        return;
> > > +    }
> > > -
> > >
> > > +    /*  We could avoid this extra copy using gst_buffer_new_wrapped.
> > > +        However, PE wouldn't know when to release the buffer
> > > allocations as the buffer
> > > +        objects are queued, so we'd have to create a ring buffer the
> > > size of the gstappsrc
> > > +        queue on the PE side and validate that we don't overrun on
> > > the unix side.  I'm not
> > > +        yet convinced that trying to reduce copies of compressed
> > > data is worth the
> > > +        complexity.  */
> > > +    buffer = gst_buffer_new_and_alloc(size);
> > > +    gst_buffer_fill(buffer, 0, data, size);
> > > -
> > >
> > > pthread_mutex_lock(&parser->mutex);
> > > -    parser->read_request.size = size;
> > > -    parser->read_request.done = true;
> > > -    parser->read_request.ret = !!data;
> > > -    if (data)
> > > -        memcpy(parser->read_request.data, data, size);
> > > -    parser->read_request.data = NULL;
> > > +    assert(parser->read_request.pending_read);
> > > -
> > >
> > > +    GST_BUFFER_OFFSET(buffer) = parser->next_pull_offset;
> > > +    g_signal_emit_by_name(G_OBJECT(parser->appsrc), "push-buffer",
> > > buffer, &ret);
> > > -
> > >
> > > +    /* In random-access mode, GST_FLOW_EOS shouldn't be returned */
> > > +    assert(ret == GST_FLOW_OK || ret == GST_FLOW_FLUSHING);
> > > +    if (ret == GST_FLOW_OK)
> > > +        parser->next_pull_offset += size;
> > > +    else
> > > +        gst_buffer_unref(buffer);
> >
> > According to my reading of the source code, GstAppSrc will take the
> > reference even if it returns GST_FLOW_FLUSHING. That's a bit
> > unintuitive and should probably be clearly documented on the GStreamer
> > side, but anyway...
>
> Good catch, thanks.
> > > -
> > >
> > > +    parser->read_request.pending_read = false;
> > >        pthread_mutex_unlock(&parser->mutex);
> > > -    pthread_cond_signal(&parser->read_done_cond);
> > >    }
> > >      static void CDECL wg_parser_set_unlimited_buffering(struct
> > > wg_parser *parser)

[Zebediah Figura]

On 9/10/21 12:04 PM, Derek Lesho wrote:
> @@ -857,6 +854,8 @@ done:
>           IMFMediaTypeHandler_Release(type_handler);
>       for (i = 0; i < type_count; i++)
>           IMFMediaType_Release(stream_types[i]);
> - if (FAILED(hr))
> -
>    IMFStreamDescriptor_Release(stream->descriptor);
> return hr;
> }

There are failure paths from media_stream_init_desc() with FAILED(hr) && 
!stream->descriptor.

...
> @@ -1392,8 +1385,40 @@ static HRESULT media_source_constructor(IMFByteStream *bytestream, struct media_
>       fail:
>       WARN("Failed to construct MFMediaSource, hr %#x.\n", hr);
> - free(descriptors);
> - IMFMediaSource_Release(&object->IMFMediaSource_iface);
> - if (descriptors)
> - {
> -
>    for(i = 0; i < object->stream_count; i++)
> -
>        IMFStreamDescriptor_Release(descriptors[i]);
> -
>    free(descriptors);
> - }
> - for (i = 0; i < object->stream_count; i++)
> - {
> -
>    struct media_stream *stream = object->streams[i];
> -
> -
>    IMFMediaEventQueue_Release(stream->event_queue);
> -
>    IMFStreamDescriptor_Release(stream->descriptor);
> -
>    IMFMediaSource_Release(&stream->parent_source->IMFMediaSource_iface);
> -
> -
>    free(stream);
> - }
> - if (object->streams)
> -
>    free(object->streams);

As long as we're here, this if() is unnecessary.
> - if (stream_count)
> -
>    unix_funcs->wg_parser_disconnect(object->wg_parser);

This doesn't work; the parser could have zero streams. Granted, that 
usually means an error condition on the GStreamer side, but it's still 
worth handling right.
> - if (object->read_thread)
> - {
> -
>    object->read_thread_shutdown = true;
> -
>    WaitForSingleObject(object->read_thread, INFINITE);
> -
>    CloseHandle(object->read_thread);
> - }
> - if (object->wg_parser)
> -
>    unix_funcs->wg_parser_destroy(object->wg_parser);
> - if (object->async_commands_queue)
> -
>    MFUnlockWorkQueue(object->async_commands_queue);
> - if (object->event_queue)
> -
>    IMFMediaEventQueue_Release(object->event_queue);
> - IMFByteStream_Release(object->byte_stream);
> - free(object);
>  return hr;
> }