Improve latency by waiting for the next keyframe

Instead of caching all the data from the previous keyframe till right now and
sending it to the client in a burst, since latency is important, we instead wait
for the next keyframe before sending buffers to the clients. This will lead to
a delay of upto 128 frames before the stream starts.

The keyframe distance and hence the delay can be tweaked by setting the
"keyframe-max-dist" property on vp8enc in src/encode.c:create_webm_profile()

4 files changed, 14 insertions, 61 deletions

diff --git a/src/encode.c b/src/encode.c
index 4054c89..5e33980 100644
--- a/src/encode.c
+++ b/src/encode.c
@@ -95,27 +95,6 @@ out:
   return;
 }
 
-static void
-update_keyframe_buffer_cb (GstElement         *fakesink,
-                           GstBuffer          *buffer,
-                           GstPad             *pad,
-                           TranscodeServerCtx *ctx)
-{
-  if (GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DECODE_ONLY) ||
-      GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_CORRUPTED) ||
-      GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_GAP) ||
-      GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DROPPABLE))
-    return;
-  
-  if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT))
-    gst_buffer_list_foreach (ctx->keyframe,
-                             stp_unref_gst_buffer, NULL);
-
-  gst_buffer_ref (buffer);
-  gst_buffer_list_add (ctx->keyframe, buffer);
-  return;
-}
-
 static GstEncodingProfile *
 create_webm_profile (void)
 {
@@ -197,8 +176,7 @@ stp_encode_from_msg (TranscodeServerCtx *ctx)
 
   /* Ensure that the stream is always realtime */
   g_object_set (fakesink,
-                "sync", TRUE,
-                "signal-handoffs", TRUE, NULL);
+                "sync", TRUE, NULL);
 
   gst_bin_add_many (GST_BIN (ctx->pipeline), encodebin, tee, q1, fakesink, NULL);
   gst_element_link_many (encodebin, tee, q1, fakesink, NULL);
@@ -228,11 +206,6 @@ stp_encode_from_msg (TranscodeServerCtx *ctx)
   g_free (tmp);
 #endif
 
-  ctx->keyframe = gst_buffer_list_new ();
-
-  g_signal_connect (fakesink, "handoff",
-                    G_CALLBACK (update_keyframe_buffer_cb), ctx);
-
   /* The pads of decodebin and encodebin are dynamic,
    * so those will be linked when streams/pads are added */
 
diff --git a/src/lib.c b/src/lib.c
index bb076f6..bd5a8d6 100644
--- a/src/lib.c
+++ b/src/lib.c
@@ -59,16 +59,6 @@ stp_on_gst_bus_message (GstBus             *bus,
   return TRUE;
 }
 
-gboolean
-stp_unref_gst_buffer (GstBuffer **buffer,
-                      guint       idx,
-                      gpointer    user_data)
-{
-  gst_buffer_unref (*buffer);
-  *buffer = NULL;
-  return TRUE;
-}
-
 /* When the incoming stream reaches EOS, we call this
  * which initiates a shutdown for all clients and then
  * the server itself */
@@ -85,9 +75,6 @@ stp_cleanup_transcode_server_ctx (TranscodeServerCtx *ctx)
   gst_element_set_state (ctx->pipeline, GST_STATE_NULL);
   gst_object_unref (ctx->pipeline);
 
-  gst_buffer_list_foreach (ctx->keyframe, stp_unref_gst_buffer, NULL);
-  gst_buffer_list_unref (ctx->keyframe);
-
   g_free (ctx);
 }
 
diff --git a/src/lib.h b/src/lib.h
index db57354..6894c0b 100644
--- a/src/lib.h
+++ b/src/lib.h
@@ -34,8 +34,6 @@ struct _TranscodeServerCtx {
   SoupMessage *msg;
   GstElement *pipeline;
   GstElement *appsrc;
-  /* Contains buffers from the previous keyframe till right now */
-  GstBufferList *keyframe;
   /* If the encoding is not chunked, we'll get multiple requests
    * with separate Content-Length headers on the same path */
   SoupEncoding encoding;
@@ -59,6 +57,7 @@ struct _TranscodeClientCtx {
   GstElement *appsink; /* We hold an extra ref to this */
   GstPad *ghostsinkpad;
 
+  gboolean keyframe_found;
   guint timeout_handler_id;
   guint seconds_since_write;
   /* The transcode server context; we don't hold a ref to this */
diff --git a/src/main.c b/src/main.c
index 6e61c3e..fe9d44a 100644
--- a/src/main.c
+++ b/src/main.c
@@ -134,6 +134,15 @@ invoke_write_client_chunk (TranscodeClientCtx *ctx)
   }
 
   buffer = gst_sample_get_buffer (sample);
+
+  /* Skip all frames till the next keyframe */
+  if (!ctx->keyframe_found &&
+      GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT)) {
+    gst_sample_unref (sample);
+    return FALSE;
+  }
+  ctx->keyframe_found = TRUE;
+
   gst_buffer_map (buffer, &info, GST_MAP_READ);
 
   soup_message_body_append (ctx->msg->response_body, SOUP_MEMORY_COPY,
@@ -151,7 +160,7 @@ static GstFlowReturn
 write_client_chunk_cb (GstElement         *appsink,
                        TranscodeClientCtx *ctx)
 {
-  g_main_context_invoke_full (NULL, G_PRIORITY_HIGH,
+  g_main_context_invoke_full (NULL, G_PRIORITY_DEFAULT,
                               (GSourceFunc)invoke_write_client_chunk,
                               ctx, NULL);
   return GST_FLOW_OK;
@@ -227,7 +236,6 @@ GET:
     /* A GET request was received. We connect from the pipeline to the
      * client requesting the stream and start writing the response. */
     GstCaps *caps = NULL;
-    GstState state;
     GstBuffer *buffer;
     GstFlowReturn ret;
     GstStateChangeReturn state_change;
@@ -311,7 +319,7 @@ GET:
     buffer = stp_get_streamheader_from_caps (caps);
     if (!buffer) {
       g_critical ("Unable to get streamheader from caps");
-      goto push_keyframe;
+      goto nostreamheader;
     }
 
     ret = gst_pad_push (srcpad, buffer);
@@ -321,21 +329,7 @@ GET:
       goto err;
     }
 
-push_keyframe:
-    ret = gst_pad_push_list (srcpad,
-                             gst_buffer_list_copy (server_ctx->keyframe));
-    if (ret != GST_FLOW_OK) {
-      g_critical ("Unable to push keyframe data: %s",
-                  gst_flow_get_name (ret));
-      goto err;
-    }
-
-    state_change = gst_element_get_state (server_ctx->pipeline,
-                                          &state, NULL, 1);
-    g_debug ("Pushed buffer. State was %s:%s.",
-             gst_element_state_change_return_get_name (state_change),
-             gst_element_state_get_name (state));
-
+nostreamheader:
     client_ctx->appsink = appsink;
 
     g_signal_connect (appsink, "new-sample",