Implement buffering

To minimize latency (at the cost of jitter), scrcpy always displays a
frame as soon as it available, without waiting.

However, when recording (--record), it still writes the captured
timestamps to the output file, so that the recorded file can be played
correctly without jitter.

Some real-time use cases might benefit from adding a small latency to
compensate for jitter too. For example, few tens of seconds of latency
for live-streaming are not important, but jitter is noticeable.

Therefore, implement a buffering mechanism (disabled by default) to add
a configurable latency delay.

PR #2417 <https://github.com/Genymobile/scrcpy/issues/2417>

app/meson.build

@@ -2,6 +2,7 @@ src = [
     'src/main.c',
     'src/adb.c',
     'src/cli.c',
+    'src/clock.c',
     'src/compat.c',
     'src/control_msg.c',
     'src/controller.c',

app/src/clock.c

@@ -0,0 +1,90 @@
+#include "clock.h"
+
+void
+sc_clock_init(struct sc_clock *clock) {
+    clock->count = 0;
+    clock->head = 0;
+    clock->left_sum.system = 0;
+    clock->left_sum.stream = 0;
+    clock->right_sum.system = 0;
+    clock->right_sum.stream = 0;
+}
+
+// Estimate the affine function f(stream) = slope * stream + offset
+static void
+sc_clock_estimate(struct sc_clock *clock,
+                  double *out_slope, sc_tick *out_offset) {
+    assert(clock->count > 1); // two points are necessary
+    struct sc_clock_point left_avg = {
+        .system = clock->left_sum.system / (clock->count / 2),
+        .stream = clock->left_sum.stream / (clock->count / 2),
+    };
+    struct sc_clock_point right_avg = {
+        .system = clock->right_sum.system / ((clock->count + 1) / 2),
+        .stream = clock->right_sum.stream / ((clock->count + 1) / 2),
+    };
+    struct sc_clock_point global_avg = {
+        .system = (clock->left_sum.system + clock->right_sum.system)
+                 / clock->count,
+        .stream = (clock->left_sum.stream + clock->right_sum.stream)
+                 / clock->count,
+    };
+
+    double slope = (double) (right_avg.system - left_avg.system)
+                 / (right_avg.stream - left_avg.stream);
+    sc_tick offset = global_avg.system - (sc_tick) (global_avg.stream * slope);
+
+    *out_slope = slope;
+    *out_offset = offset;
+}
+
+void
+sc_clock_update(struct sc_clock *clock, sc_tick system, sc_tick stream) {
+    struct sc_clock_point *point = &clock->points[clock->head];
+
+    if (clock->count == SC_CLOCK_RANGE || clock->count & 1) {
+        // One point passes from the right sum to the left sum
+
+        unsigned mid;
+        if (clock->count == SC_CLOCK_RANGE) {
+            mid = (clock->head + SC_CLOCK_RANGE / 2) % SC_CLOCK_RANGE;
+        } else {
+            // Only for the first frames
+            mid = clock->count / 2;
+        }
+
+        struct sc_clock_point *mid_point = &clock->points[mid];
+        clock->left_sum.system += mid_point->system;
+        clock->left_sum.stream += mid_point->stream;
+        clock->right_sum.system -= mid_point->system;
+        clock->right_sum.stream -= mid_point->stream;
+    }
+
+    if (clock->count == SC_CLOCK_RANGE) {
+        // The current point overwrites the previous value in the circular
+        // array, update the left sum accordingly
+        clock->left_sum.system -= point->system;
+        clock->left_sum.stream -= point->stream;
+    } else {
+        ++clock->count;
+    }
+
+    point->system = system;
+    point->stream = stream;
+
+    clock->right_sum.system += system;
+    clock->right_sum.stream += stream;
+
+    clock->head = (clock->head + 1) % SC_CLOCK_RANGE;
+
+    if (clock->count > 1) {
+        // Update estimation
+        sc_clock_estimate(clock, &clock->slope, &clock->offset);
+    }
+}
+
+sc_tick
+sc_clock_to_system_time(struct sc_clock *clock, sc_tick stream) {
+    assert(clock->count > 1); // sc_clock_update() must have been called
+    return (sc_tick) (stream * clock->slope) + clock->offset;
+}

app/src/clock.h

@@ -0,0 +1,70 @@
+#ifndef SC_CLOCK_H
+#define SC_CLOCK_H
+
+#include "common.h"
+
+#include <assert.h>
+
+#include "util/tick.h"
+
+#define SC_CLOCK_RANGE 32
+static_assert(!(SC_CLOCK_RANGE & 1), "SC_CLOCK_RANGE must be even");
+
+struct sc_clock_point {
+    sc_tick system;
+    sc_tick stream;
+};
+
+/**
+ * The clock aims to estimate the affine relation between the stream (device)
+ * time and the system time:
+ *
+ *     f(stream) = slope * stream + offset
+ *
+ * To that end, it stores the SC_CLOCK_RANGE last clock points (the timestamps
+ * of a frame expressed both in stream time and system time) in a circular
+ * array.
+ *
+ * To estimate the slope, it splits the last SC_CLOCK_RANGE points into two
+ * sets of SC_CLOCK_RANGE/2 points, and compute their centroid ("average
+ * point"). The slope of the estimated affine function is that of the line
+ * passing through these two points.
+ *
+ * To estimate the offset, it computes the centroid of all the SC_CLOCK_RANGE
+ * points. The resulting affine function passes by this centroid.
+ *
+ * With a circular array, the rolling sums (and average) are quick to compute.
+ * In practice, the estimation is stable and the evolution is smooth.
+ */
+struct sc_clock {
+    // Circular array
+    struct sc_clock_point points[SC_CLOCK_RANGE];
+
+    // Number of points in the array (count <= SC_CLOCK_RANGE)
+    unsigned count;
+
+    // Index of the next point to write
+    unsigned head;
+
+    // Sum of the first count/2 points
+    struct sc_clock_point left_sum;
+
+    // Sum of the last (count+1)/2 points
+    struct sc_clock_point right_sum;
+
+    // Estimated slope and offset
+    // (computed on sc_clock_update(), used by sc_clock_to_system_time())
+    double slope;
+    sc_tick offset;
+};
+
+void
+sc_clock_init(struct sc_clock *clock);
+
+void
+sc_clock_update(struct sc_clock *clock, sc_tick system, sc_tick stream);
+
+sc_tick
+sc_clock_to_system_time(struct sc_clock *clock, sc_tick stream);
+
+#endif

app/src/screen.c

@@ -308,15 +308,21 @@ screen_init(struct screen *screen, const struct screen_params *params) {
         .on_new_frame = sc_video_buffer_on_new_frame,
     };
 
-    bool ok = sc_video_buffer_init(&screen->vb, &cbs, screen);
+    bool ok = sc_video_buffer_init(&screen->vb, 0, &cbs, screen);
     if (!ok) {
         LOGE("Could not initialize video buffer");
         return false;
     }
 
+    ok = sc_video_buffer_start(&screen->vb);
+    if (!ok) {
+        LOGE("Could not start video_buffer");
+        goto error_destroy_video_buffer;
+    }
+
     if (!fps_counter_init(&screen->fps_counter)) {
         LOGE("Could not initialize FPS counter");
-        goto error_destroy_video_buffer;
+        goto error_stop_and_join_video_buffer;
     }
 
     screen->frame_size = params->frame_size;
@@ -457,6 +463,9 @@ error_destroy_window:
     SDL_DestroyWindow(screen->window);
 error_destroy_fps_counter:
     fps_counter_destroy(&screen->fps_counter);
+error_stop_and_join_video_buffer:
+    sc_video_buffer_stop(&screen->vb);
+    sc_video_buffer_join(&screen->vb);
 error_destroy_video_buffer:
     sc_video_buffer_destroy(&screen->vb);
 
@@ -475,11 +484,13 @@ screen_hide_window(struct screen *screen) {
 
 void
 screen_interrupt(struct screen *screen) {
+    sc_video_buffer_stop(&screen->vb);
     fps_counter_interrupt(&screen->fps_counter);
 }
 
 void
 screen_join(struct screen *screen) {
+    sc_video_buffer_join(&screen->vb);
     fps_counter_join(&screen->fps_counter);
 }
 

app/src/v4l2_sink.c

@@ -159,16 +159,22 @@ sc_v4l2_sink_open(struct sc_v4l2_sink *vs) {
         .on_new_frame = sc_video_buffer_on_new_frame,
     };
 
-    bool ok = sc_video_buffer_init(&vs->vb, &cbs, vs);
+    bool ok = sc_video_buffer_init(&vs->vb, 0, &cbs, vs);
     if (!ok) {
         LOGE("Could not initialize video buffer");
         return false;
     }
 
+    ok = sc_video_buffer_start(&vs->vb);
+    if (!ok) {
+        LOGE("Could not start video buffer");
+        goto error_video_buffer_destroy;
+    }
+
     ok = sc_mutex_init(&vs->mutex);
     if (!ok) {
         LOGC("Could not create mutex");
-        goto error_video_buffer_destroy;
+        goto error_video_buffer_stop_and_join;
     }
 
     ok = sc_cond_init(&vs->cond);
@@ -293,6 +299,9 @@ error_cond_destroy:
     sc_cond_destroy(&vs->cond);
 error_mutex_destroy:
     sc_mutex_destroy(&vs->mutex);
+error_video_buffer_stop_and_join:
+    sc_video_buffer_stop(&vs->vb);
+    sc_video_buffer_join(&vs->vb);
 error_video_buffer_destroy:
     sc_video_buffer_destroy(&vs->vb);
 
@@ -306,7 +315,10 @@ sc_v4l2_sink_close(struct sc_v4l2_sink *vs) {
     sc_cond_signal(&vs->cond);
     sc_mutex_unlock(&vs->mutex);
 
+    sc_video_buffer_stop(&vs->vb);
+
     sc_thread_join(&vs->thread, NULL);
+    sc_video_buffer_join(&vs->vb);
 
     av_packet_free(&vs->packet);
     av_frame_free(&vs->frame);

app/src/video_buffer.c

@@ -1,35 +1,44 @@
 #include "video_buffer.h"
 
 #include <assert.h>
+#include <stdlib.h>
+
 #include <libavutil/avutil.h>
 #include <libavformat/avformat.h>
 
 #include "util/log.h"
 
-bool
+static struct sc_video_buffer_frame *
-sc_video_buffer_init(struct sc_video_buffer *vb,
+sc_video_buffer_frame_new(const AVFrame *frame) {
-                     const struct sc_video_buffer_callbacks *cbs,
+    struct sc_video_buffer_frame *vb_frame = malloc(sizeof(*vb_frame));
-                     void *cbs_userdata) {
+    if (!vb_frame) {
-    bool ok = sc_frame_buffer_init(&vb->fb);
+        return NULL;
-    if (!ok) {
-        return false;
     }
 
-    assert(cbs);
+    vb_frame->frame = av_frame_alloc();
-    assert(cbs->on_new_frame);
+    if (!vb_frame->frame) {
-
+        free(vb_frame);
-    vb->cbs = cbs;
+        return NULL;
-    vb->cbs_userdata = cbs_userdata;
-    return true;
     }
 
-void
+    if (av_frame_ref(vb_frame->frame, frame)) {
-sc_video_buffer_destroy(struct sc_video_buffer *vb) {
+        av_frame_free(&vb_frame->frame);
-    sc_frame_buffer_destroy(&vb->fb);
+        free(vb_frame);
+        return NULL;
     }
 
-bool
+    return vb_frame;
-sc_video_buffer_push(struct sc_video_buffer *vb, const AVFrame *frame) {
+}
+
+static void
+sc_video_buffer_frame_delete(struct sc_video_buffer_frame *vb_frame) {
+    av_frame_unref(vb_frame->frame);
+    av_frame_free(&vb_frame->frame);
+    free(vb_frame);
+}
+
+static bool
+sc_video_buffer_offer(struct sc_video_buffer *vb, const AVFrame *frame) {
     bool previous_skipped;
     bool ok = sc_frame_buffer_push(&vb->fb, frame, &previous_skipped);
     if (!ok) {
@@ -40,6 +49,196 @@ sc_video_buffer_push(struct sc_video_buffer *vb, const AVFrame *frame) {
     return true;
 }
 
+static int
+run_buffering(void *data) {
+    struct sc_video_buffer *vb = data;
+
+    assert(vb->buffering_time > 0);
+
+    for (;;) {
+        sc_mutex_lock(&vb->b.mutex);
+
+        while (!vb->b.stopped && sc_queue_is_empty(&vb->b.queue)) {
+            sc_cond_wait(&vb->b.queue_cond, &vb->b.mutex);
+        }
+
+        if (vb->b.stopped) {
+            sc_mutex_unlock(&vb->b.mutex);
+            goto stopped;
+        }
+
+        struct sc_video_buffer_frame *vb_frame;
+        sc_queue_take(&vb->b.queue, next, &vb_frame);
+
+        sc_tick max_deadline = sc_tick_now() + vb->buffering_time;
+        // PTS (written by the server) are expressed in microseconds
+        sc_tick pts = SC_TICK_TO_US(vb_frame->frame->pts);
+
+        bool timed_out = false;
+        while (!vb->b.stopped && !timed_out) {
+            sc_tick deadline = sc_clock_to_system_time(&vb->b.clock, pts)
+                             + vb->buffering_time;
+            if (deadline > max_deadline) {
+                deadline = max_deadline;
+            }
+
+            timed_out =
+                !sc_cond_timedwait(&vb->b.wait_cond, &vb->b.mutex, deadline);
+        }
+
+        if (vb->b.stopped) {
+            sc_video_buffer_frame_delete(vb_frame);
+            sc_mutex_unlock(&vb->b.mutex);
+            goto stopped;
+        }
+
+        sc_mutex_unlock(&vb->b.mutex);
+
+        sc_video_buffer_offer(vb, vb_frame->frame);
+
+        sc_video_buffer_frame_delete(vb_frame);
+    }
+
+stopped:
+    // Flush queue
+    while (!sc_queue_is_empty(&vb->b.queue)) {
+        struct sc_video_buffer_frame *vb_frame;
+        sc_queue_take(&vb->b.queue, next, &vb_frame);
+        sc_video_buffer_frame_delete(vb_frame);
+    }
+
+    LOGD("Buffering thread ended");
+
+    return 0;
+}
+
+bool
+sc_video_buffer_init(struct sc_video_buffer *vb, sc_tick buffering_time,
+                     const struct sc_video_buffer_callbacks *cbs,
+                     void *cbs_userdata) {
+    bool ok = sc_frame_buffer_init(&vb->fb);
+    if (!ok) {
+        return false;
+    }
+
+    assert(buffering_time >= 0);
+    if (buffering_time) {
+        ok = sc_mutex_init(&vb->b.mutex);
+        if (!ok) {
+            LOGC("Could not create mutex");
+            sc_frame_buffer_destroy(&vb->fb);
+            return false;
+        }
+
+        ok = sc_cond_init(&vb->b.queue_cond);
+        if (!ok) {
+            LOGC("Could not create cond");
+            sc_mutex_destroy(&vb->b.mutex);
+            sc_frame_buffer_destroy(&vb->fb);
+            return false;
+        }
+
+        ok = sc_cond_init(&vb->b.wait_cond);
+        if (!ok) {
+            LOGC("Could not create wait cond");
+            sc_cond_destroy(&vb->b.queue_cond);
+            sc_mutex_destroy(&vb->b.mutex);
+            sc_frame_buffer_destroy(&vb->fb);
+            return false;
+        }
+
+        sc_clock_init(&vb->b.clock);
+        sc_queue_init(&vb->b.queue);
+    }
+
+    assert(cbs);
+    assert(cbs->on_new_frame);
+
+    vb->buffering_time = buffering_time;
+    vb->cbs = cbs;
+    vb->cbs_userdata = cbs_userdata;
+    return true;
+}
+
+bool
+sc_video_buffer_start(struct sc_video_buffer *vb) {
+    if (vb->buffering_time) {
+        bool ok =
+            sc_thread_create(&vb->b.thread, run_buffering, "buffering", vb);
+        if (!ok) {
+            LOGE("Could not start buffering thread");
+            return false;
+        }
+    }
+
+    return true;
+}
+
+void
+sc_video_buffer_stop(struct sc_video_buffer *vb) {
+    if (vb->buffering_time) {
+        sc_mutex_lock(&vb->b.mutex);
+        vb->b.stopped = true;
+        sc_cond_signal(&vb->b.queue_cond);
+        sc_cond_signal(&vb->b.wait_cond);
+        sc_mutex_unlock(&vb->b.mutex);
+    }
+}
+
+void
+sc_video_buffer_join(struct sc_video_buffer *vb) {
+    if (vb->buffering_time) {
+        sc_thread_join(&vb->b.thread, NULL);
+    }
+}
+
+void
+sc_video_buffer_destroy(struct sc_video_buffer *vb) {
+    sc_frame_buffer_destroy(&vb->fb);
+    if (vb->buffering_time) {
+        sc_cond_destroy(&vb->b.wait_cond);
+        sc_cond_destroy(&vb->b.queue_cond);
+        sc_mutex_destroy(&vb->b.mutex);
+    }
+}
+
+bool
+sc_video_buffer_push(struct sc_video_buffer *vb, const AVFrame *frame) {
+    if (!vb->buffering_time) {
+        // No buffering
+        return sc_video_buffer_offer(vb, frame);
+    }
+
+    sc_mutex_lock(&vb->b.mutex);
+
+    sc_tick pts = SC_TICK_FROM_US(frame->pts);
+    sc_clock_update(&vb->b.clock, sc_tick_now(), pts);
+    sc_cond_signal(&vb->b.wait_cond);
+
+    if (vb->b.clock.count == 1) {
+        sc_mutex_unlock(&vb->b.mutex);
+        // First frame, offer it immediately, for two reasons:
+        //  - not to delay the opening of the scrcpy window
+        //  - the buffering estimation needs at least two clock points, so it
+        //  could not handle the first frame
+        return sc_video_buffer_offer(vb, frame);
+    }
+
+    struct sc_video_buffer_frame *vb_frame = sc_video_buffer_frame_new(frame);
+    if (!vb_frame) {
+        sc_mutex_unlock(&vb->b.mutex);
+        LOGE("Could not allocate frame");
+        return false;
+    }
+
+    sc_queue_push(&vb->b.queue, next, vb_frame);
+    sc_cond_signal(&vb->b.queue_cond);
+
+    sc_mutex_unlock(&vb->b.mutex);
+
+    return true;
+}
+
 void
 sc_video_buffer_consume(struct sc_video_buffer *vb, AVFrame *dst) {
     sc_frame_buffer_consume(&vb->fb, dst);

app/src/video_buffer.h

@@ -5,14 +5,39 @@
 
 #include <stdbool.h>
 
+#include "clock.h"
 #include "frame_buffer.h"
+#include "util/queue.h"
+#include "util/thread.h"
+#include "util/tick.h"
 
 // forward declarations
 typedef struct AVFrame AVFrame;
 
+struct sc_video_buffer_frame {
+    AVFrame *frame;
+    struct sc_video_buffer_frame *next;
+};
+
+struct sc_video_buffer_frame_queue SC_QUEUE(struct sc_video_buffer_frame);
+
 struct sc_video_buffer {
     struct sc_frame_buffer fb;
 
+    sc_tick buffering_time;
+
+    // only if buffering_time > 0
+    struct {
+        sc_thread thread;
+        sc_mutex mutex;
+        sc_cond queue_cond;
+        sc_cond wait_cond;
+
+        struct sc_clock clock;
+        struct sc_video_buffer_frame_queue queue;
+        bool stopped;
+    } b; // buffering
+
     const struct sc_video_buffer_callbacks *cbs;
     void *cbs_userdata;
 };
@@ -23,10 +48,19 @@ struct sc_video_buffer_callbacks {
 };
 
 bool
-sc_video_buffer_init(struct sc_video_buffer *vb,
+sc_video_buffer_init(struct sc_video_buffer *vb, sc_tick buffering_time,
                      const struct sc_video_buffer_callbacks *cbs,
                      void *cbs_userdata);
 
+bool
+sc_video_buffer_start(struct sc_video_buffer *vb);
+
+void
+sc_video_buffer_stop(struct sc_video_buffer *vb);
+
+void
+sc_video_buffer_join(struct sc_video_buffer *vb);
+
 void
 sc_video_buffer_destroy(struct sc_video_buffer *vb);