Runs capture processing on a separate thread

app/src/capture.c

@@ -40,6 +40,7 @@
 #include <SDL2/SDL_events.h>
 
 #include "events.h"
+#include "util/lock.h"
 #include "util/log.h"
 
 static const char *string_error(int err) {
@@ -200,6 +201,25 @@ bool capture_init(struct capture *capture, const char *filename) {
     return false;
   }
 
+  capture->mutex = SDL_CreateMutex();
+  if (!capture->mutex) {
+    LOGC("Could not create mutex");
+    SDL_free(capture->filename);
+    return false;
+  }
+
+  capture->queue_cond = SDL_CreateCond();
+  if (!capture->queue_cond) {
+    LOGC("Could not create capture cond");
+    SDL_DestroyMutex(capture->mutex);
+    SDL_free(capture->filename);
+    return false;
+  }
+
+  queue_init(&capture->queue);
+  capture->stopped = false;
+  capture->finished = false;
+
   int ret = avio_open(&capture->file_context, capture->filename,
                         AVIO_FLAG_WRITE);
   if (ret < 0) {
@@ -245,10 +265,123 @@ void capture_destroy(struct capture *capture) {
   avio_close(capture->file_context);
 }
 
-bool capture_push(struct capture *capture, const AVPacket *packet) {
+static bool capture_process(struct capture *capture, const AVPacket *packet) {
-  if (decode_packet_to_png(capture->file_context, capture->context, packet, capture->working_frame)) {
+  if (capture->finished) {
+    LOGV("Skipping redundant call to capture_push");
+  } else {
+    bool found_png = decode_packet_to_png(capture->file_context, capture->context, packet, capture->working_frame);
+    if (found_png) {
       LOGI("Parsed PNG from incoming packets.");
-    return true;
+      capture->finished = found_png;
     }
-  return false;
+  }
+  return capture->finished;
+}
+
+static struct capture_packet *
+capture_packet_new(const AVPacket *packet) {
+    struct capture_packet *rec = SDL_malloc(sizeof(*rec));
+    if (!rec) {
+        return NULL;
+    }
+
+    // av_packet_ref() does not initialize all fields in old FFmpeg versions
+    // See <https://github.com/Genymobile/scrcpy/issues/707>
+    av_init_packet(&rec->packet);
+
+    if (av_packet_ref(&rec->packet, packet)) {
+        SDL_free(rec);
+        return NULL;
+    }
+    return rec;
+}
+
+static void
+capture_packet_delete(struct capture_packet *rec) {
+    av_packet_unref(&rec->packet);
+    SDL_free(rec);
+}
+
+static int
+run_capture(void *data) {
+    struct capture *capture = data;
+
+    for (;;) {
+        mutex_lock(capture->mutex);
+
+        while (!capture->stopped && queue_is_empty(&capture->queue)) {
+            cond_wait(capture->queue_cond, capture->mutex);
+        }
+
+        // if stopped is set, continue to process the remaining events (to
+        // finish the capture) before actually stopping
+
+        if (capture->stopped && queue_is_empty(&capture->queue)) {
+            mutex_unlock(capture->mutex);
+            break;
+        }
+
+        struct capture_packet *rec;
+        queue_take(&capture->queue, next, &rec);
+
+        mutex_unlock(capture->mutex);
+
+        bool ok = capture_process(capture, &rec->packet);
+        capture_packet_delete(rec);
+        if (ok) {
+            break;
+        }
+    }
+
+    LOGD("capture thread ended");
+
+    return 0;
+}
+
+bool
+capture_start(struct capture *capture) {
+
+    capture->thread = SDL_CreateThread(run_capture, "capture", capture);
+    if (!capture->thread) {
+        LOGC("Could not start capture thread");
+        return false;
+    }
+
+    return true;
+}
+
+void
+capture_stop(struct capture *capture) {
+    mutex_lock(capture->mutex);
+    capture->stopped = true;
+    cond_signal(capture->queue_cond);
+    mutex_unlock(capture->mutex);
+}
+
+void
+capture_join(struct capture *capture) {
+    SDL_WaitThread(capture->thread, NULL);
+}
+
+bool
+capture_push(struct capture *capture, const AVPacket *packet) {
+    mutex_lock(capture->mutex);
+    assert(!capture->stopped);
+
+    if (capture->finished) {
+        // reject any new packet (this will stop the stream)
+        return false;
+    }
+
+    struct capture_packet *rec = capture_packet_new(packet);
+    if (!rec) {
+        LOGC("Could not allocate capture packet");
+        return false;
+    }
+
+    queue_push(&capture->queue, next, rec);
+    cond_signal(capture->queue_cond);
+
+    mutex_unlock(capture->mutex);
+    return true;
 }

app/src/capture.h

@@ -4,18 +4,47 @@
 #include <stdbool.h>
 #include <libavformat/avio.h>
 #include <libavcodec/avcodec.h>
+#include <SDL2/SDL_mutex.h>
+#include <SDL2/SDL_thread.h>
+
+#include "config.h"
+#include "common.h"
+#include "util/queue.h"
+
+struct capture_packet {
+    AVPacket packet;
+    struct capture_packet *next;
+};
+
+struct capture_queue QUEUE(struct capture_packet);
 
 struct capture {
   char *filename;
   AVIOContext *file_context;
   AVCodecContext *context;
   AVFrame *working_frame;
+
+  SDL_Thread *thread;
+  SDL_mutex *mutex;
+  SDL_cond *queue_cond;
+  bool finished; // PNG has been captured
+  bool stopped; // set on recorder_stop() by the stream reader
+  struct capture_queue queue;
 };
 
 bool capture_init(struct capture *capture, const char *filename);
 
 void capture_destroy(struct capture *capture);
 
+bool
+capture_start(struct capture *capture);
+
+void
+capture_stop(struct capture *capture);
+
+void
+capture_join(struct capture *capture);
+
 bool capture_push(struct capture *capture, const AVPacket *packet);
 
 #endif  // CAPTURE_H

app/src/command.c

@@ -96,6 +96,7 @@ adb_execute(const char *serial, const char *const adb_cmd[], size_t len) {
     cmd[len + i] = NULL;
     enum process_result r = cmd_execute(cmd, &process);
     if (r != PROCESS_SUCCESS) {
+        LOGE("adb process execution returned an error.");
         show_adb_err_msg(r, cmd);
         return PROCESS_NONE;
     }

app/src/stream.c

@@ -228,10 +228,17 @@ run_stream(void *data) {
         }
     }
 
+    if (stream->capture) {
+        if (!capture_start(stream->capture)) {
+            LOGE("Could not start capture");
+            goto finally_stop_and_join_recorder;
+        }
+    }
+
     stream->parser = av_parser_init(AV_CODEC_ID_H264);
     if (!stream->parser) {
         LOGE("Could not initialize parser");
-        goto finally_stop_and_join_recorder;
+        goto finally_stop_and_join_capture;
     }
 
     // We must only pass complete frames to av_parser_parse2()!
@@ -261,6 +268,12 @@ run_stream(void *data) {
     }
 
     av_parser_close(stream->parser);
+finally_stop_and_join_capture:
+    if (stream->capture) {
+        capture_stop(stream->capture);
+        LOGI("Finishing capture...");
+        capture_join(stream->capture);
+    }
 finally_stop_and_join_recorder:
     if (stream->recorder) {
         recorder_stop(stream->recorder);