[PATCH v8 0/1] MR7288: winegstreamer: Avoid large buffer pushes in wg_transform.

[Yuxuan Shui (＠yshui)]

## Context

This is a problem discovered when running "Max: The Curse of Brotherhood" under proton. This game writes 4MB a piece into `wg_transform`, which contains about 4 seconds of compressed video. `wg_transform` calls `gst_pad_push` in `transform_ProcessOutput`, which takes ~300ms (i.e. 20 frames @ 60fps) to decode this buffer of video. So the end result is the game hitches every 4 seconds while playing cut scene videos.

Proton currently has a special case for this particular game, for which it breaks up the buffer into small chunks to avoid long blocks. This MR adopts that and applies it generally.

One concern raised by @redmcg is:
> The only issue I can think of is if there are any decoders which don't have a parser; then they might not know how to deal with an arbitrary 4096 byte buffer (the parser is generally responsible for taking arbitrary buffers and producing something the decoder can work with, for example: a full frame).

So this MR only enables this strategy when there is a parser element.

--
  v8: winegstreamer: Avoid large buffer pushes in wg_transform.

https://gitlab.winehq.org/wine/wine/-/merge_requests/7288

[PATCH v8 1/1] winegstreamer: Avoid large buffer pushes in wg_transform.

[Yuxuan Shui]

From: Yuxuan Shui

This is done by inserting a special "stepper" element behind the parser (if
there is one). This element will keep the incoming buffers in a internal queue
and not push them downstream immediately. Instead, it gives us control over
when we push and how much is pushed.
---
 dlls/winegstreamer/unix_private.h |   2 +
 dlls/winegstreamer/unixlib.c      |   4 +
 dlls/winegstreamer/wg_transform.c | 173 +++++++++++++++++++++++++++++-
 3 files changed, 174 insertions(+), 5 deletions(-)

diff --git a/dlls/winegstreamer/unix_private.h b/dlls/winegstreamer/unix_private.h
index b2cc036c914..7ac82a72fe5 100644
--- a/dlls/winegstreamer/unix_private.h
+++ b/dlls/winegstreamer/unix_private.h
@@ -104,4 +104,6 @@ extern void wg_allocator_provide_sample(GstAllocator *allocator, struct wg_sampl
 extern void wg_allocator_release_sample(GstAllocator *allocator, struct wg_sample *sample,
         bool discard_data);

+GST_ELEMENT_REGISTER_DECLARE(winegstreamerstepper);
+
 #endif /* __WINE_WINEGSTREAMER_UNIX_PRIVATE_H */
diff --git a/dlls/winegstreamer/unixlib.c b/dlls/winegstreamer/unixlib.c
index 729008e1f46..7a07b2e0a2a 100644
--- a/dlls/winegstreamer/unixlib.c
+++ b/dlls/winegstreamer/unixlib.c
@@ -300,6 +300,10 @@ NTSTATUS wg_init_gstreamer(void *arg)

GST_INFO("GStreamer library version %s; wine built with %d.%d.%d.",
             gst_version_string(), GST_VERSION_MAJOR, GST_VERSION_MINOR, GST_VERSION_MICRO);
+
+    if (!GST_ELEMENT_REGISTER(winegstreamerstepper, NULL))
+        GST_ERROR("Failed to register the stepper element");
+
     return STATUS_SUCCESS;
 }

diff --git a/dlls/winegstreamer/wg_transform.c b/dlls/winegstreamer/wg_transform.c
index 33758401a06..75b49bdca61 100644
--- a/dlls/winegstreamer/wg_transform.c
+++ b/dlls/winegstreamer/wg_transform.c
@@ -42,11 +42,39 @@

#define GST_SAMPLE_FLAG_WG_CAPS_CHANGED (GST_MINI_OBJECT_FLAG_LAST << 0)

+/* This GstElement takes buffers and events from its sink pad, instead of pushing them
+ * out the src pad, it keeps them in a internal queue until the step function
+ * is called manually.
+ */
+typedef struct _WgStepper
+{
+    GstElement element;
+    GstPad *src, *sink;
+    GstAtomicQueue *fifo;
+} WgStepper;
+
+typedef struct _WgStepperClass
+{
+    GstElementClass parent_class;
+} WgStepperClass;
+
+#define GST_TYPE_WG_STEPPER (wg_stepper_get_type())
+#define WG_STEPPER(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_WG_STEPPER, WgStepper))
+#define GST_WG_STEPPER_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_WG_STEPPER,WgStepperClass))
+#define GST_IS_WG_STEPPER(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_WG_STEPPER))
+#define GST_IS_WG_STEPPER_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_WG_STEPPER))
+
+G_DEFINE_TYPE (WgStepper, wg_stepper, GST_TYPE_ELEMENT);
+GST_ELEMENT_REGISTER_DEFINE(winegstreamerstepper, "winegstreamerstepper", GST_RANK_NONE, GST_TYPE_WG_STEPPER);
+
+static bool wg_stepper_step(WgStepper *stepper);
+
 struct wg_transform
 {
     struct wg_transform_attrs attrs;

GstElement *container;
+    WgStepper *stepper;
     GstAllocator *allocator;
     GstPad *my_src, *my_sink;
     GstSegment segment;
@@ -493,7 +521,22 @@ static bool transform_create_decoder_elements(struct wg_transform *transform,
     if ((element = find_element(GST_ELEMENT_FACTORY_TYPE_PARSER, transform->input_caps, parsed_caps))
             && !append_element(transform->container, element, first, last))
         goto done;
-    else if (!element)
+
+    if (element)
+    {
+        /* We try to intercept buffers produced by the parser, so if we push a large buffer into the
+         * parser, it won't push everything into the decoder all in one go.
+         */
+        if ((element = create_element("winegstreamerstepper", NULL)))
+        {
+            if (!append_element(transform->container, element, first, last))
+                g_object_unref(element);
+            else
+                /* element is owned by the container */
+                transform->stepper = WG_STEPPER(element);
+        }
+    }
+    else
     {
         gst_caps_unref(parsed_caps);
         parsed_caps = gst_caps_ref(transform->input_caps);
@@ -1008,7 +1051,8 @@ static NTSTATUS read_transform_output(struct wg_sample *sample, GstBuffer *buffe

static NTSTATUS complete_drain(struct wg_transform *transform)
 {
-    if (transform->draining && gst_atomic_queue_length(transform->input_queue) == 0)
+    bool stepper_empty = transform->stepper == NULL || gst_atomic_queue_length(transform->stepper->fifo) == 0;
+    if (transform->draining && gst_atomic_queue_length(transform->input_queue) == 0 && stepper_empty)
     {
         GstEvent *event;
         transform->draining = false;
@@ -1035,14 +1079,23 @@ error:

static bool get_transform_output(struct wg_transform *transform, struct wg_sample *sample)
 {
-    GstBuffer *input_buffer;
     GstFlowReturn ret;

wg_allocator_provide_sample(transform->allocator, sample);

-    while (!(transform->output_sample = gst_atomic_queue_pop(transform->output_queue))
-            && (input_buffer = gst_atomic_queue_pop(transform->input_queue)))
+    while (!(transform->output_sample = gst_atomic_queue_pop(transform->output_queue)))
     {
+        GstBuffer *input_buffer;
+        if (transform->stepper && wg_stepper_step(transform->stepper))
+        {
+            /* If we pushed anything from the stepper, we don't need to dequeue more buffers. */
+            complete_drain(transform);
+            continue;
+        }
+
+        if (!(input_buffer = gst_atomic_queue_pop(transform->input_queue)))
+            break;
+
         if ((ret = gst_pad_push(transform->my_src, input_buffer)))
             GST_WARNING("Failed to push transform input, error %d", ret);

@@ -1221,3 +1274,113 @@ NTSTATUS wg_transform_notify_qos(void *args)

return S_OK;
 }
+
+/* Move events and at most one buffer from the internal fifo queue to the output src pad.
+ * Returns true if anything is moved, or false if the fifo is empty.
+ */
+static bool wg_stepper_step(WgStepper *stepper)
+{
+    bool pushed = false;
+    gpointer ptr;
+    while ((ptr = gst_atomic_queue_pop(stepper->fifo)))
+    {
+        if (GST_IS_BUFFER(ptr))
+        {
+            GST_TRACE("Forwarding buffer %"GST_PTR_FORMAT" from fifo", ptr);
+            gst_pad_push(stepper->src, GST_BUFFER(ptr));
+            pushed = true;
+            break;
+        }
+
+        if (GST_IS_EVENT(ptr))
+        {
+            GST_TRACE("Processing event %"GST_PTR_FORMAT" from fifo", ptr);
+            gst_pad_event_default(stepper->sink, GST_OBJECT(stepper), GST_EVENT(ptr));
+            pushed = true;
+        }
+    }
+    return pushed;
+}
+
+static GstStateChangeReturn wg_stepper_change_state(GstElement *element, GstStateChange transition)
+{
+    WgStepper *this = WG_STEPPER(element);
+    if (transition == GST_STATE_CHANGE_READY_TO_NULL)
+    {
+        gpointer ptr;
+        GST_TRACE("Discarding all objects in the fifo");
+        while ((ptr = gst_atomic_queue_pop(this->fifo)))
+        {
+            if (GST_IS_EVENT(ptr))
+                gst_event_unref(ptr);
+            else if (GST_IS_BUFFER(ptr))
+                gst_buffer_unref(ptr);
+        }
+    }
+    return GST_STATE_CHANGE_SUCCESS;
+}
+
+static void wg_stepper_class_init(WgStepperClass *klass)
+{
+    gst_element_class_set_metadata(GST_ELEMENT_CLASS(klass), "winegstreamer buffer stepper", "Connector",
+        "Hold incoming buffer for manual pushing", "Yuxuan Shui ");
+    klass->parent_class.change_state = wg_stepper_change_state;
+}
+
+static GstFlowReturn wg_stepper_chain_cb(GstPad *pad, GstObject *parent, GstBuffer *buf)
+{
+    WgStepper *this = WG_STEPPER(parent);
+    GST_TRACE("Pushing buffer %"GST_PTR_FORMAT" into fifo", buf);
+    gst_atomic_queue_push(this->fifo, buf);
+    return GST_FLOW_OK;
+}
+
+static gboolean wg_stepper_event_cb(GstPad *pad, GstObject *parent, GstEvent *event)
+{
+    WgStepper *this = WG_STEPPER(parent);
+    GST_TRACE("Pushing event %"GST_PTR_FORMAT" into fifo", event);
+    gst_atomic_queue_push(this->fifo, event);
+    return true;
+}
+
+static gboolean wg_stepper_src_query_cb(GstPad *pad, GstObject *parent, GstQuery *query)
+{
+    WgStepper *this = WG_STEPPER(parent);
+    GstPad *peer = gst_pad_get_peer(this->sink);
+    if (!peer) return gst_pad_query_default(pad, parent, query);
+    GST_TRACE("Forwarding query %"GST_PTR_FORMAT" to upstream", query);
+    return gst_pad_query(peer, query);
+}
+
+static void wg_stepper_init(WgStepper *stepper)
+{
+    static GstStaticPadTemplate sink_factory =
+        GST_STATIC_PAD_TEMPLATE (
+            "sink",
+            GST_PAD_SINK,
+            GST_PAD_ALWAYS,
+            GST_STATIC_CAPS ("ANY")
+        );
+    static GstStaticPadTemplate src_factory =
+        GST_STATIC_PAD_TEMPLATE(
+            "src",
+            GST_PAD_SRC,
+            GST_PAD_ALWAYS,
+            GST_STATIC_CAPS("ANY")
+        );
+    stepper->sink = gst_pad_new_from_static_template(&sink_factory, "sink");
+    gst_element_add_pad(GST_ELEMENT(stepper), stepper->sink);
+    gst_pad_set_chain_function(stepper->sink, wg_stepper_chain_cb);
+    gst_pad_set_event_function(stepper->sink, wg_stepper_event_cb);
+    GST_PAD_SET_PROXY_CAPS(stepper->sink);
+    gst_pad_set_active(stepper->sink, true);
+
+    stepper->src = gst_pad_new_from_static_template(&src_factory, "src");
+    gst_element_add_pad(GST_ELEMENT(stepper), stepper->src);
+    gst_pad_set_query_function(stepper->src, wg_stepper_src_query_cb);
+    gst_pad_set_active(stepper->src, true);
+
+    stepper->fifo = gst_atomic_queue_new(4);
+
+    GST_DEBUG("Created new stepper element %"GST_PTR_FORMAT, stepper);
+}

[Yuxuan Shui (＠yshui)]

update: 1) change calls to `complete_drain` to match old behavior. 2) account for the fact that sometimes a output sample is generated after pushing an event (i.e. sample generation is not always triggered by a buffer).