Re: [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)