[PATCH 5/9] mf/session: Request more samples from upstream when necessary.

[Rémi Bernon]

From: Rémi Bernon

---
 dlls/mf/session.c | 76 ++++++++++++++++++++++++++++++++++-------------
 1 file changed, 56 insertions(+), 20 deletions(-)

diff --git a/dlls/mf/session.c b/dlls/mf/session.c
index c8198353cb5..89c05e42764 100644
--- a/dlls/mf/session.c
+++ b/dlls/mf/session.c
@@ -199,6 +199,7 @@ struct topo_node
             BOOL *inputs_eos;
             DWORD *input_map;
             unsigned int input_count;
+            unsigned int next_input;

struct transform_stream *outputs;
             DWORD *output_map;
@@ -3161,6 +3162,17 @@ static BOOL transform_node_is_end_of_stream(struct topo_node *topo_node)
     return TRUE;
 }

+static BOOL transform_node_has_requests(struct topo_node *topo_node)
+{
+    UINT i;
+
+    for (i = 0; i < topo_node->u.transform.output_count; i++)
+        if (topo_node->u.transform.outputs[i].requests)
+            return TRUE;
+
+    return FALSE;
+}
+
 static void transform_node_push_sample(const struct media_session *session, struct topo_node *topo_node,
         UINT input, IMFSample *sample)
 {
@@ -3194,7 +3206,7 @@ static void transform_node_deliver_samples(struct media_session *session, struct
     HRESULT hr;

/* Push down all available output. */
-    for (output = 0; output < topo_node->u.transform.output_count; ++output)
+    for (output = 0, hr = S_OK; SUCCEEDED(hr) && output < topo_node->u.transform.output_count; ++output)
     {
         struct transform_stream *stream = &topo_node->u.transform.outputs[output];

@@ -3204,8 +3216,16 @@ static void transform_node_deliver_samples(struct media_session *session, struct
             continue;
         }

-        while (stream->requests && SUCCEEDED(hr = transform_stream_pop_sample(stream, &sample)))
+        while (stream->requests)
         {
+            if (FAILED(hr = transform_stream_pop_sample(stream, &sample)))
+            {
+                if (FAILED(hr = transform_node_pull_samples(session, topo_node)))
+                    break;
+                if (FAILED(hr = transform_stream_pop_sample(stream, &sample)))
+                    break;
+            }
+
             session_deliver_sample_to_node(session, down_node, input, sample);
             stream->requests--;
             IMFSample_Release(sample);
@@ -3219,6 +3239,20 @@ static void transform_node_deliver_samples(struct media_session *session, struct

IMFTopologyNode_Release(down_node);
     }
+
+    if (transform_node_has_requests(topo_node))
+    {
+        input = topo_node->u.transform.next_input++ % topo_node->u.transform.input_count;
+
+        if (FAILED(hr = IMFTopologyNode_GetInput(topo_node->node, input, &up_node, &output)))
+            WARN("Failed to get node %p/%lu input, hr %#lx\n", topo_node->node, input, hr);
+        else
+        {
+            if (FAILED(hr = session_request_sample_from_node(session, up_node, output)))
+                WARN("Failed to request sample from upstream node %p/%lu, hr %#lx\n", up_node, output, hr);
+            IMFTopologyNode_Release(up_node);
+        }
+    }
 }

static void session_deliver_sample_to_node(struct media_session *session, IMFTopologyNode *node, unsigned int input,
@@ -3293,13 +3327,13 @@ static void session_deliver_pending_samples(struct media_session *session, IMFTo

static HRESULT session_request_sample_from_node(struct media_session *session, IMFTopologyNode *node, DWORD output)
 {
-    IMFTopologyNode *downstream_node, *upstream_node;
-    DWORD downstream_input, upstream_output;
+    IMFTopologyNode *down_node;
     struct topo_node *topo_node;
     MF_TOPOLOGY_TYPE node_type;
+    HRESULT hr = S_OK;
     IMFSample *sample;
     TOPOID node_id;
-    HRESULT hr;
+    DWORD input;

IMFTopologyNode_GetNodeType(node, &node_type);
     IMFTopologyNode_GetTopoNodeID(node, &node_id);
@@ -3316,27 +3350,29 @@ static HRESULT session_request_sample_from_node(struct media_session *session, I
         {
             struct transform_stream *stream = &topo_node->u.transform.outputs[output];

-            if (FAILED(hr = transform_stream_pop_sample(stream, &sample)))
+            if (FAILED(hr = IMFTopologyNode_GetOutput(node, output, &down_node, &input)))
             {
-                /* Forward request to upstream node. */
-                if (SUCCEEDED(hr = IMFTopologyNode_GetInput(node, 0 /* FIXME */, &upstream_node, &upstream_output)))
-                {
-                    if (SUCCEEDED(hr = session_request_sample_from_node(session, upstream_node, upstream_output)))
-                        stream->requests++;
-                    IMFTopologyNode_Release(upstream_node);
-                }
+                WARN("Failed to node %p/%lu output, hr %#lx.\n", node, output, hr);
+                break;
             }
-            else
-            {
-                if (SUCCEEDED(hr = IMFTopologyNode_GetOutput(node, output, &downstream_node, &downstream_input)))
-                {
-                    session_deliver_sample_to_node(session, downstream_node, downstream_input, sample);
-                    IMFTopologyNode_Release(downstream_node);
-                }

+            if (SUCCEEDED(transform_stream_pop_sample(stream, &sample)))
+            {
+                session_deliver_sample_to_node(session, down_node, input, sample);
                 IMFSample_Release(sample);
             }
+            else if (transform_node_has_requests(topo_node))
+            {
+                /* there's already requests pending, just increase the counter */
+                stream->requests++;
+            }
+            else
+            {
+                stream->requests++;
+                transform_node_deliver_samples(session, topo_node);
+            }

+            IMFTopologyNode_Release(down_node);
             break;
         }
         case MF_TOPOLOGY_TEE_NODE: