Extract packet merging

Config packets must be prepended to the next media packet. Extract the
logic to a new sc_packet_merger helper to simplify the demuxer code.

app/meson.build

@@ -21,6 +21,7 @@ src = [
     'src/mouse_inject.c',
     'src/opengl.c',
     'src/options.c',
+    'src/packet_merger.c',
     'src/receiver.c',
     'src/recorder.c',
     'src/scrcpy.c',

app/src/demuxer.c

@@ -6,6 +6,7 @@
 
 #include "decoder.h"
 #include "events.h"
+#include "packet_merger.h"
 #include "recorder.h"
 #include "util/binary.h"
 #include "util/log.h"
@@ -18,17 +19,10 @@
 #define SC_PACKET_PTS_MASK (SC_PACKET_FLAG_KEY_FRAME - 1)
 
 static enum AVCodecID
-sc_demuxer_recv_codec_id(struct sc_demuxer *demuxer) {
+sc_demuxer_to_avcodec_id(uint32_t codec_id) {
-    uint8_t data[4];
-    ssize_t r = net_recv_all(demuxer->socket, data, 4);
-    if (r < 4) {
-        return false;
-    }
-
 #define SC_CODEC_ID_H264 UINT32_C(0x68323634) // "h264" in ASCII
 #define SC_CODEC_ID_H265 UINT32_C(0x68323635) // "h265" in ASCII
 #define SC_CODEC_ID_AV1 UINT32_C(0x00617631) // "av1" in ASCII
-    uint32_t codec_id = sc_read32be(data);
     switch (codec_id) {
         case SC_CODEC_ID_H264:
             return AV_CODEC_ID_H264;
@@ -42,6 +36,18 @@ sc_demuxer_recv_codec_id(struct sc_demuxer *demuxer) {
     }
 }
 
+static bool
+sc_demuxer_recv_codec_id(struct sc_demuxer *demuxer, uint32_t *codec_id) {
+    uint8_t data[4];
+    ssize_t r = net_recv_all(demuxer->socket, data, 4);
+    if (r < 4) {
+        return false;
+    }
+
+    *codec_id = sc_read32be(data);
+    return true;
+}
+
 static bool
 sc_demuxer_recv_packet(struct sc_demuxer *demuxer, AVPacket *packet) {
     // The video stream contains raw packets, without time information. When we
@@ -105,7 +111,7 @@ push_packet_to_sinks(struct sc_demuxer *demuxer, const AVPacket *packet) {
     for (unsigned i = 0; i < demuxer->sink_count; ++i) {
         struct sc_packet_sink *sink = demuxer->sinks[i];
         if (!sink->ops->push(sink, packet)) {
-            LOGE("Could not send config packet to sink %d", i);
+            LOGE("Could not send packet to sink %d", i);
             return false;
         }
     }
@@ -115,48 +121,7 @@ push_packet_to_sinks(struct sc_demuxer *demuxer, const AVPacket *packet) {
 
 static bool
 sc_demuxer_push_packet(struct sc_demuxer *demuxer, AVPacket *packet) {
-    bool is_config = packet->pts == AV_NOPTS_VALUE;
-
-    // A config packet must not be decoded immediately (it contains no
-    // frame); instead, it must be concatenated with the future data packet.
-    if (demuxer->pending || is_config) {
-        if (demuxer->pending) {
-            size_t offset = demuxer->pending->size;
-            if (av_grow_packet(demuxer->pending, packet->size)) {
-                LOG_OOM();
-                return false;
-            }
-
-            memcpy(demuxer->pending->data + offset, packet->data, packet->size);
-        } else {
-            demuxer->pending = av_packet_alloc();
-            if (!demuxer->pending) {
-                LOG_OOM();
-                return false;
-            }
-            if (av_packet_ref(demuxer->pending, packet)) {
-                LOG_OOM();
-                av_packet_free(&demuxer->pending);
-                return false;
-            }
-        }
-
-        if (!is_config) {
-            // prepare the concat packet to send to the decoder
-            demuxer->pending->pts = packet->pts;
-            demuxer->pending->dts = packet->dts;
-            demuxer->pending->flags = packet->flags;
-            packet = demuxer->pending;
-        }
-    }
-
     bool ok = push_packet_to_sinks(demuxer, packet);
-
-    if (!is_config && demuxer->pending) {
-        // the pending packet must be discarded (consumed or error)
-        av_packet_free(&demuxer->pending);
-    }
-
     if (!ok) {
         LOGE("Could not process packet");
         return false;
@@ -196,7 +161,17 @@ static int
 run_demuxer(void *data) {
     struct sc_demuxer *demuxer = data;
 
-    enum AVCodecID codec_id = sc_demuxer_recv_codec_id(demuxer);
+    // Flag to report end-of-stream (i.e. device disconnected)
+    bool eos = false;
+
+    uint32_t raw_codec_id;
+    bool ok = sc_demuxer_recv_codec_id(demuxer, &raw_codec_id);
+    if (!ok) {
+        eos = true;
+        goto end;
+    }
+
+    enum AVCodecID codec_id = sc_demuxer_to_avcodec_id(raw_codec_id);
     if (codec_id == AV_CODEC_ID_NONE) {
         // Error already logged
         goto end;
@@ -208,37 +183,31 @@ run_demuxer(void *data) {
         goto end;
     }
 
-    demuxer->codec_ctx = avcodec_alloc_context3(codec);
+    if (!sc_demuxer_open_sinks(demuxer, codec)) {
-    if (!demuxer->codec_ctx) {
+        LOGE("Could not open demuxer sinks");
-        LOG_OOM();
         goto end;
     }
 
-    if (!sc_demuxer_open_sinks(demuxer, codec)) {
+    struct sc_packet_merger merger;
-        LOGE("Could not open demuxer sinks");
+    sc_packet_merger_init(&merger);
-        goto finally_free_codec_ctx;
-    }
-
-    demuxer->parser = av_parser_init(codec_id);
-    if (!demuxer->parser) {
-        LOGE("Could not initialize parser");
-        goto finally_close_sinks;
-    }
-
-    // We must only pass complete frames to av_parser_parse2()!
-    // It's more complicated, but this allows to reduce the latency by 1 frame!
-    demuxer->parser->flags |= PARSER_FLAG_COMPLETE_FRAMES;
 
     AVPacket *packet = av_packet_alloc();
     if (!packet) {
         LOG_OOM();
-        goto finally_close_parser;
+        goto finally_close_sinks;
     }
 
     for (;;) {
         bool ok = sc_demuxer_recv_packet(demuxer, packet);
         if (!ok) {
             // end of stream
+            eos = true;
+            break;
+        }
+
+        // Prepend any config packet to the next media packet
+        ok = sc_packet_merger_merge(&merger, packet);
+        if (!ok) {
             break;
         }
 
@@ -252,19 +221,13 @@ run_demuxer(void *data) {
 
     LOGD("End of frames");
 
-    if (demuxer->pending) {
+    sc_packet_merger_destroy(&merger);
-        av_packet_free(&demuxer->pending);
-    }
 
     av_packet_free(&packet);
-finally_close_parser:
-    av_parser_close(demuxer->parser);
 finally_close_sinks:
     sc_demuxer_close_sinks(demuxer);
-finally_free_codec_ctx:
-    avcodec_free_context(&demuxer->codec_ctx);
 end:
-    demuxer->cbs->on_eos(demuxer, demuxer->cbs_userdata);
+    demuxer->cbs->on_ended(demuxer, eos, demuxer->cbs_userdata);
 
     return 0;
 }
@@ -273,10 +236,9 @@ void
 sc_demuxer_init(struct sc_demuxer *demuxer, sc_socket socket,
                 const struct sc_demuxer_callbacks *cbs, void *cbs_userdata) {
     demuxer->socket = socket;
-    demuxer->pending = NULL;
     demuxer->sink_count = 0;
 
-    assert(cbs && cbs->on_eos);
+    assert(cbs && cbs->on_ended);
 
     demuxer->cbs = cbs;
     demuxer->cbs_userdata = cbs_userdata;

app/src/demuxer.h

@@ -21,18 +21,12 @@ struct sc_demuxer {
     struct sc_packet_sink *sinks[SC_DEMUXER_MAX_SINKS];
     unsigned sink_count;
 
-    AVCodecContext *codec_ctx;
-    AVCodecParserContext *parser;
-    // successive packets may need to be concatenated, until a non-config
-    // packet is available
-    AVPacket *pending;
-
     const struct sc_demuxer_callbacks *cbs;
     void *cbs_userdata;
 };
 
 struct sc_demuxer_callbacks {
-    void (*on_eos)(struct sc_demuxer *demuxer, void *userdata);
+    void (*on_ended)(struct sc_demuxer *demuxer, bool eos, void *userdata);
 };
 
 void

app/src/events.h

@@ -1,5 +1,6 @@
-#define EVENT_NEW_FRAME                 SDL_USEREVENT
+#define SC_EVENT_NEW_FRAME                 SDL_USEREVENT
-#define EVENT_STREAM_STOPPED           (SDL_USEREVENT + 1)
+#define SC_EVENT_DEVICE_DISCONNECTED      (SDL_USEREVENT + 1)
-#define EVENT_SERVER_CONNECTION_FAILED (SDL_USEREVENT + 2)
+#define SC_EVENT_SERVER_CONNECTION_FAILED (SDL_USEREVENT + 2)
-#define EVENT_SERVER_CONNECTED         (SDL_USEREVENT + 3)
+#define SC_EVENT_SERVER_CONNECTED         (SDL_USEREVENT + 3)
-#define EVENT_USB_DEVICE_DISCONNECTED  (SDL_USEREVENT + 4)
+#define SC_EVENT_USB_DEVICE_DISCONNECTED  (SDL_USEREVENT + 4)
+#define SC_EVENT_DEMUXER_ERROR            (SDL_USEREVENT + 5)

app/src/packet_merger.c

@@ -0,0 +1,52 @@
+#include "packet_merger.h"
+
+#include "util/log.h"
+
+void
+sc_packet_merger_init(struct sc_packet_merger *merger) {
+    merger->pending_config = NULL;
+}
+
+void
+sc_packet_merger_destroy(struct sc_packet_merger *merger) {
+    av_packet_free(&merger->pending_config);
+}
+
+bool
+sc_packet_merger_merge(struct sc_packet_merger *merger, AVPacket *packet) {
+    bool is_config = packet->pts == AV_NOPTS_VALUE;
+
+    if (is_config) {
+        av_packet_free(&merger->pending_config);
+
+        merger->pending_config = av_packet_alloc();
+        if (!merger->pending_config) {
+            LOG_OOM();
+            goto error;
+        }
+
+        if (av_packet_ref(merger->pending_config, packet)) {
+            LOG_OOM();
+            av_packet_free(&merger->pending_config);
+            goto error;
+        }
+    } else if (merger->pending_config) {
+        size_t config_size = merger->pending_config->size;
+        size_t media_size = packet->size;
+        if (av_grow_packet(packet, config_size)) {
+            LOG_OOM();
+            goto error;
+        }
+
+        memmove(packet->data + config_size, packet->data, media_size);
+        memcpy(packet->data, merger->pending_config->data, config_size);
+
+        av_packet_free(&merger->pending_config);
+    }
+
+    return true;
+
+error:
+    av_packet_unref(packet);
+    return false;
+}

app/src/packet_merger.h

@@ -0,0 +1,41 @@
+#ifndef SC_PACKET_MERGER_H
+#define SC_PACKET_MERGER_H
+
+#include "common.h"
+
+#include <stdbool.h>
+#include <libavcodec/avcodec.h>
+
+/**
+ * Config packets (containing the SPS/PPS) are sent in-band. A new config
+ * packet is sent whenever a new encoding session is started (on start and on
+ * device orientation change).
+ *
+ * Every time a config packet is received, it must be sent alone (for recorder
+ * extradata), then concatenated to the next media packet (for correct decoding
+ * and recording).
+ *
+ * This helper reads every input packet and modifies each media packet which
+ * immediately follows a config packet to prepend the config packet payload.
+ */
+
+struct sc_packet_merger {
+    AVPacket *pending_config;
+};
+
+void
+sc_packet_merger_init(struct sc_packet_merger *merger);
+
+void
+sc_packet_merger_destroy(struct sc_packet_merger *merger);
+
+/**
+ * If the packet is a config packet, then reference it for later.
+ * Otherwise (if the packet is a media packet), then if a config packet is
+ * pending, prepend the config packet to this packet (so the packet is
+ * modified!).
+ */
+bool
+sc_packet_merger_merge(struct sc_packet_merger *merger, AVPacket *packet);
+
+#endif

app/src/scrcpy.c

@@ -155,9 +155,12 @@ event_loop(struct scrcpy *s) {
     SDL_Event event;
     while (SDL_WaitEvent(&event)) {
         switch (event.type) {
-            case EVENT_STREAM_STOPPED:
+            case SC_EVENT_DEVICE_DISCONNECTED:
                 LOGW("Device disconnected");
                 return SCRCPY_EXIT_DISCONNECTED;
+            case SC_EVENT_DEMUXER_ERROR:
+                LOGE("Demuxer error");
+                return SCRCPY_EXIT_FAILURE;
             case SDL_QUIT:
                 LOGD("User requested to quit");
                 return SCRCPY_EXIT_SUCCESS;
@@ -179,10 +182,10 @@ await_for_server(bool *connected) {
                 LOGD("User requested to quit");
                 *connected = false;
                 return true;
-            case EVENT_SERVER_CONNECTION_FAILED:
+            case SC_EVENT_SERVER_CONNECTION_FAILED:
                 LOGE("Server connection failed");
                 return false;
-            case EVENT_SERVER_CONNECTED:
+            case SC_EVENT_SERVER_CONNECTED:
                 LOGD("Server connected");
                 *connected = true;
                 return true;
@@ -233,11 +236,15 @@ av_log_callback(void *avcl, int level, const char *fmt, va_list vl) {
 }
 
 static void
-sc_demuxer_on_eos(struct sc_demuxer *demuxer, void *userdata) {
+sc_demuxer_on_ended(struct sc_demuxer *demuxer, bool eos, void *userdata) {
     (void) demuxer;
     (void) userdata;
 
-    PUSH_EVENT(EVENT_STREAM_STOPPED);
+    if (eos) {
+        PUSH_EVENT(SC_EVENT_DEVICE_DISCONNECTED);
+    } else {
+        PUSH_EVENT(SC_EVENT_DEMUXER_ERROR);
+    }
 }
 
 static void
@@ -245,7 +252,7 @@ sc_server_on_connection_failed(struct sc_server *server, void *userdata) {
     (void) server;
     (void) userdata;
 
-    PUSH_EVENT(EVENT_SERVER_CONNECTION_FAILED);
+    PUSH_EVENT(SC_EVENT_SERVER_CONNECTION_FAILED);
 }
 
 static void
@@ -253,7 +260,7 @@ sc_server_on_connected(struct sc_server *server, void *userdata) {
     (void) server;
     (void) userdata;
 
-    PUSH_EVENT(EVENT_SERVER_CONNECTED);
+    PUSH_EVENT(SC_EVENT_SERVER_CONNECTED);
 }
 
 static void
@@ -421,7 +428,7 @@ scrcpy(struct scrcpy_options *options) {
     av_log_set_callback(av_log_callback);
 
     static const struct sc_demuxer_callbacks demuxer_cbs = {
-        .on_eos = sc_demuxer_on_eos,
+        .on_ended = sc_demuxer_on_ended,
     };
     sc_demuxer_init(&s->demuxer, s->server.video_socket, &demuxer_cbs, NULL);
 

app/src/screen.c

@@ -371,7 +371,7 @@ sc_video_buffer_on_new_frame(struct sc_video_buffer *vb, bool previous_skipped,
     bool need_new_event;
     if (previous_skipped) {
         sc_fps_counter_add_skipped_frame(&screen->fps_counter);
-        // The EVENT_NEW_FRAME triggered for the previous frame will consume
+        // The SC_EVENT_NEW_FRAME triggered for the previous frame will consume
         // this new frame instead, unless the previous event failed
         need_new_event = screen->event_failed;
     } else {
@@ -380,7 +380,7 @@ sc_video_buffer_on_new_frame(struct sc_video_buffer *vb, bool previous_skipped,
 
     if (need_new_event) {
         static SDL_Event new_frame_event = {
-            .type = EVENT_NEW_FRAME,
+            .type = SC_EVENT_NEW_FRAME,
         };
 
         // Post the event on the UI thread
@@ -820,7 +820,7 @@ sc_screen_handle_event(struct sc_screen *screen, SDL_Event *event) {
     bool relative_mode = sc_screen_is_relative_mode(screen);
 
     switch (event->type) {
-        case EVENT_NEW_FRAME: {
+        case SC_EVENT_NEW_FRAME: {
             bool ok = sc_screen_update_frame(screen);
             if (!ok) {
                 LOGW("Frame update failed\n");

app/src/usb/scrcpy_otg.c

@@ -22,7 +22,7 @@ sc_usb_on_disconnected(struct sc_usb *usb, void *userdata) {
     (void) userdata;
 
     SDL_Event event;
-    event.type = EVENT_USB_DEVICE_DISCONNECTED;
+    event.type = SC_EVENT_USB_DEVICE_DISCONNECTED;
     int ret = SDL_PushEvent(&event);
     if (ret < 0) {
         LOGE("Could not post USB disconnection event: %s", SDL_GetError());
@@ -34,7 +34,7 @@ event_loop(struct scrcpy_otg *s) {
     SDL_Event event;
     while (SDL_WaitEvent(&event)) {
         switch (event.type) {
-            case EVENT_USB_DEVICE_DISCONNECTED:
+            case SC_EVENT_USB_DEVICE_DISCONNECTED:
                 LOGW("Device disconnected");
                 return SCRCPY_EXIT_DISCONNECTED;
             case SDL_QUIT:

server/src/main/java/com/genymobile/scrcpy/Controller.java

@@ -90,6 +90,7 @@ public class Controller {
                 control();
             } catch (IOException e) {
                 // this is expected on close
+            } finally {
                 Ln.d("Controller stopped");
             }
         });
@@ -100,11 +101,17 @@ public class Controller {
     public void stop() {
         if (thread != null) {
             thread.interrupt();
-            thread = null;
         }
         sender.stop();
     }
 
+    public void join() throws InterruptedException {
+        if (thread != null) {
+            thread.join();
+        }
+        sender.join();
+    }
+
     public DeviceMessageSender getSender() {
         return sender;
     }

server/src/main/java/com/genymobile/scrcpy/DeviceMessageSender.java

@@ -57,6 +57,7 @@ public final class DeviceMessageSender {
                 loop();
             } catch (IOException | InterruptedException e) {
                 // this is expected on close
+            } finally {
                 Ln.d("Device message sender stopped");
             }
         });
@@ -66,7 +67,12 @@ public final class DeviceMessageSender {
     public void stop() {
         if (thread != null) {
             thread.interrupt();
-            thread = null;
+        }
+    }
+
+    public void join() throws InterruptedException {
+        if (thread != null) {
+            thread.join();
         }
     }
 }

server/src/main/java/com/genymobile/scrcpy/Server.java

@@ -111,12 +111,21 @@ public final class Server {
                 screenEncoder.streamScreen(device, videoStreamer);
             } catch (IOException e) {
                 // this is expected on close
-                Ln.d("Screen streaming stopped");
             } finally {
+                Ln.d("Screen streaming stopped");
                 initThread.interrupt();
                 if (controller != null) {
                     controller.stop();
                 }
+
+                try {
+                    initThread.join();
+                    if (controller != null) {
+                        controller.join();
+                    }
+                } catch (InterruptedException e) {
+                    // ignore
+                }
             }
         }
     }