From: Zebediah Figura Subject: [PATCH v6 5/5] winegstreamer: Replace source pad interface with GstAppSrc. Message-Id: <20210915230010.1333902-5-zfigura@codeweavers.com> Date: Wed, 15 Sep 2021 18:00:10 -0500 In-Reply-To: <20210915230010.1333902-1-zfigura@codeweavers.com> References: <20210915230010.1333902-1-zfigura@codeweavers.com> From: Derek Lesho Signed-off-by: Derek Lesho Signed-off-by: Zebediah Figura --- dlls/winegstreamer/wg_parser.c | 434 +++++++++------------------------ 1 file changed, 122 insertions(+), 312 deletions(-) diff --git a/dlls/winegstreamer/wg_parser.c b/dlls/winegstreamer/wg_parser.c index 7b2a12fd7a9..7d4eb135217 100644 --- a/dlls/winegstreamer/wg_parser.c +++ b/dlls/winegstreamer/wg_parser.c @@ -33,6 +33,7 @@ #include #include #include +#include /* 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,20 @@ 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 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_request; bool flushing, sink_connected; @@ -522,7 +515,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) pthread_cond_wait(&parser->read_cond, &parser->mutex); if (!parser->sink_connected) @@ -541,15 +534,69 @@ 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) { + GstMessage *message; + GstFlowReturn ret; + GstBuffer *buffer; + GError *error; + + if (!data) + { + pthread_mutex_lock(&parser->mutex); + + if (parser->sink_connected) + { + error = g_error_new(G_FILE_ERROR, G_FILE_ERROR_FAILED, "WG-Parser client failed to read data at offset %" G_GUINT64_FORMAT, parser->read_request.offset); + message = gst_message_new_error(NULL, error, ""); + gst_bus_post(parser->bus, message); + parser->read_request.pending = false; + } + + pthread_mutex_unlock(&parser->mutex); + return; + } + + if (!size) + { + pthread_mutex_lock(&parser->mutex); + + if (parser->sink_connected) + g_signal_emit_by_name(G_OBJECT(parser->appsrc), "end-of-stream", &ret); + parser->read_request.pending = false; + + pthread_mutex_unlock(&parser->mutex); + return; + } + + /* We will always perform an extra blit here. We can avoid this in some + * cases by wrapping a client-allocated buffer using + * gst_buffer_new_wrapped(). However, releasing the memory is non-trivial, + * since GStreamer will hold onto a reference for an arbitrarily long + * period of time. Until there's evidence to suggest that the blit causes a + * performance problem, leave it alone. */ + 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; + + if (!parser->sink_connected) + { + pthread_mutex_unlock(&parser->mutex); + gst_buffer_unref(buffer); + return; + } + + assert(parser->read_request.pending); + + GST_BUFFER_OFFSET(buffer) = parser->read_request.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->read_request.offset += size; + + parser->read_request.pending = 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) @@ -1204,196 +1251,56 @@ 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; + /* As of GStreamer 1.18, appsrc suffers from a race condition. When in + * random access mode (and when underlyingly in pull mode), appsrc may + * spuriously send multiple requests for the same offset and length. If it + * receives the same buffer twice (or consecutive buffers), it will blindly + * queue them and satisfy subsequent getrange requests from downstream + * elements with the wrong buffers. + * + * Internally, this function is called inside of a loop, which also pops + * buffers from the internal queue. Accordingly we can safely treat this + * request as spurious by checking if we have already sent data; since the + * data is consumed by this thread it will not have been consumed yet. + * + * The bug is documented in greater detail here: + * + * https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/issues/937 + */ + g_object_get(G_OBJECT(appsrc), "current-level-bytes", &queued_bytes, NULL); + if (queued_bytes) + { + pthread_mutex_unlock(&parser->mutex); + return; + } + + parser->read_request.pending = true; + parser->read_request.size = length; + 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. */ + pthread_mutex_unlock(&parser->mutex); +} - while (!parser->read_request.done) - pthread_cond_wait(&parser->read_done_cond, &parser->mutex); +static gboolean src_seek_data(GstElement *appsrc, guint64 offset, gpointer user) +{ + struct wg_parser *parser = user; - ret = parser->read_request.ret; - gst_buffer_set_size(*buffer, parser->read_request.size); + pthread_mutex_lock(&parser->mutex); + + assert(!parser->read_request.pending); + parser->read_request.offset = offset; 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); - - if (ret) - return size ? GST_FLOW_OK : GST_FLOW_EOS; - return 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)) - { - 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; - } - } - - gst_buffer_unref(buffer); - - gst_pad_push_event(parser->my_src, gst_event_new_eos()); - - GST_DEBUG("Stopping push thread."); - - return NULL; -} - -static gboolean activate_push(GstPad *pad, gboolean activate) -{ - 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; - - 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; -} - -static gboolean src_activate_mode_cb(GstPad *pad, GstObject *parent, GstPadMode mode, gboolean activate) -{ - struct wg_parser *parser = gst_pad_get_element_private(pad); - - GST_DEBUG("%s source pad for parser %p in %s mode.", - activate ? "Activating" : "Deactivating", parser, gst_pad_mode_get_name(mode)); - - 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) @@ -1440,89 +1347,11 @@ 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; int ret; - parser->file_size = file_size; parser->sink_connected = true; if (!parser->bus) @@ -1534,15 +1363,16 @@ 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); - parser->start_offset = parser->next_offset = parser->stop_offset = 0; - parser->next_pull_offset = 0; + g_object_set(parser->appsrc, "stream-type", GST_APP_STREAM_TYPE_RANDOM_ACCESS, NULL); + g_object_set(parser->appsrc, "size", 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->read_request.offset = 0; parser->error = false; if (!parser->init_gst(parser)) @@ -1638,18 +1468,11 @@ 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; out: if (parser->container) gst_element_set_state(parser->container, GST_STATE_NULL); - if (parser->their_sink) - { - gst_pad_unlink(parser->my_src, parser->their_sink); - gst_object_unref(parser->their_sink); - parser->my_src = parser->their_sink = NULL; - } for (i = 0; i < parser->stream_count; ++i) free_stream(parser->streams[i]); @@ -1686,10 +1509,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; @@ -1711,7 +1530,6 @@ static void CDECL wg_parser_disconnect(struct wg_parser *parser) static BOOL decodebin_parser_init_gst(struct wg_parser *parser) { GstElement *element; - int ret; if (!(element = create_element("decodebin", "base"))) return FALSE; @@ -1724,15 +1542,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 = 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 appsrc.\n"); return FALSE; } @@ -1753,15 +1569,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 = 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 appsrc.\n"); return FALSE; } @@ -1779,10 +1593,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 appsrc.\n"); return FALSE; } @@ -1813,10 +1626,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 appsrc.\n"); return FALSE; } @@ -1846,7 +1658,6 @@ 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; GST_DEBUG("Created winegstreamer parser %p.\n", parser); @@ -1900,7 +1711,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); } -- 2.33.0