]> git.sesse.net Git - nageru/blobdiff - futatabi/main.cpp
Log a warning when we kill a client that is not keeping up.
[nageru] / futatabi / main.cpp
index 9585dd4e4b3180201cd8d210187d4cf697114724..19cd5fb3e4162439e22e7cac2cd8208b3a84686e 100644 (file)
@@ -75,16 +75,16 @@ Summary metric_received_frame_size_bytes;
 
 namespace {
 
-FrameOnDisk write_frame(int stream_idx, int64_t pts, const uint8_t *data, size_t size, DB *db)
+FrameOnDisk write_frame(int stream_idx, int64_t pts, const uint8_t *data, size_t size, vector<uint32_t> audio, DB *db)
 {
        if (open_frame_files.count(stream_idx) == 0) {
                char filename[256];
-               snprintf(filename, sizeof(filename), "%s/frames/cam%d-pts%09ld.frames",
+               snprintf(filename, sizeof(filename), "%s/frames/cam%d-pts%09" PRId64 ".frames",
                         global_flags.working_directory.c_str(), stream_idx, pts);
                FILE *fp = fopen(filename, "wb");
                if (fp == nullptr) {
                        perror(filename);
-                       exit(1);
+                       abort();
                }
 
                lock_guard<mutex> lock(frame_mu);
@@ -105,30 +105,37 @@ FrameOnDisk write_frame(int stream_idx, int64_t pts, const uint8_t *data, size_t
        hdr.set_stream_idx(stream_idx);
        hdr.set_pts(pts);
        hdr.set_file_size(size);
+       hdr.set_audio_size(audio.size() * sizeof(audio[0]));
 
        string serialized;
        if (!hdr.SerializeToString(&serialized)) {
                fprintf(stderr, "Frame header serialization failed.\n");
-               exit(1);
+               abort();
        }
        uint32_t len = htonl(serialized.size());
 
        if (fwrite(frame_magic, frame_magic_len, 1, file.fp) != 1) {
                perror("fwrite");
-               exit(1);
+               abort();
        }
        if (fwrite(&len, sizeof(len), 1, file.fp) != 1) {
                perror("fwrite");
-               exit(1);
+               abort();
        }
        if (fwrite(serialized.data(), serialized.size(), 1, file.fp) != 1) {
                perror("fwrite");
-               exit(1);
+               abort();
        }
        off_t offset = ftell(file.fp);
        if (fwrite(data, size, 1, file.fp) != 1) {
                perror("fwrite");
-               exit(1);
+               abort();
+       }
+       if (audio.size() > 0) {
+               if (fwrite(audio.data(), hdr.audio_size(), 1, file.fp) != 1) {
+                       perror("fwrite");
+                       exit(1);
+               }
        }
        fflush(file.fp);  // No fsync(), though. We can accept losing a few frames.
        global_disk_space_estimator->report_write(filename, 8 + sizeof(len) + serialized.size() + size, pts);
@@ -138,6 +145,7 @@ FrameOnDisk write_frame(int stream_idx, int64_t pts, const uint8_t *data, size_t
        frame.filename_idx = filename_idx;
        frame.offset = offset;
        frame.size = size;
+       frame.audio_size = audio.size() * sizeof(audio[0]);
 
        {
                lock_guard<mutex> lock(frame_mu);
@@ -151,7 +159,7 @@ FrameOnDisk write_frame(int stream_idx, int64_t pts, const uint8_t *data, size_t
                // Start a new file next time.
                if (fclose(file.fp) != 0) {
                        perror("fclose");
-                       exit(1);
+                       abort();
                }
                open_frame_files.erase(stream_idx);
 
@@ -199,7 +207,7 @@ int main(int argc, char **argv)
                global_flags.stream_source = argv[optind];
        } else {
                usage();
-               exit(1);
+               abort();
        }
 
        string frame_dir = global_flags.working_directory + "/frames";
@@ -208,7 +216,7 @@ int main(int argc, char **argv)
                fprintf(stderr, "%s does not exist, creating it.\n", frame_dir.c_str());
        } else if (errno != EEXIST) {
                perror(global_flags.working_directory.c_str());
-               exit(1);
+               abort();
        }
 
        avformat_network_init();
@@ -237,7 +245,7 @@ int main(int argc, char **argv)
        global_share_widget = new QGLWidget();
        if (!global_share_widget->isValid()) {
                fprintf(stderr, "Failed to initialize OpenGL. Futatabi needs at least OpenGL 4.5 to function properly.\n");
-               exit(1);
+               abort();
        }
 
        // Initialize Movit.
@@ -246,7 +254,7 @@ int main(int argc, char **argv)
                QOpenGLContext *context = create_context(surface);
                if (!make_current(context, surface)) {
                        printf("oops\n");
-                       exit(1);
+                       abort();
                }
                CHECK(movit::init_movit(MOVIT_SHADER_DIR, movit::MOVIT_DEBUG_OFF));
                delete_context(context);
@@ -275,7 +283,6 @@ int main(int argc, char **argv)
 
        should_quit = true;
        record_thread.join();
-       JPEGFrameView::shutdown();
 
        return ret;
 }
@@ -285,7 +292,7 @@ void load_frame_file(const char *filename, const string &basename, unsigned file
        struct stat st;
        if (stat(filename, &st) == -1) {
                perror(filename);
-               exit(1);
+               abort();
        }
 
        vector<DB::FrameOnDiskAndStreamIdx> all_frames = db->load_frame_file(basename, st.st_size, filename_idx);
@@ -303,7 +310,19 @@ void load_frame_file(const char *filename, const string &basename, unsigned file
        FILE *fp = fopen(filename, "rb");
        if (fp == nullptr) {
                perror(filename);
-               exit(1);
+               abort();
+       }
+
+       // Find the actual length of the file, since fseek() past the end of the file
+       // will succeed without an error.
+       if (fseek(fp, 0, SEEK_END) == -1) {
+               perror("fseek(SEEK_END)");
+               abort();
+       }
+       off_t file_len = ftell(fp);
+       if (fseek(fp, 0, SEEK_SET) == -1) {
+               perror("fseek(SEEK_SET)");
+               abort();
        }
 
        size_t magic_offset = 0;
@@ -360,10 +379,12 @@ void load_frame_file(const char *filename, const string &basename, unsigned file
                }
                frame.filename_idx = filename_idx;
                frame.size = hdr.file_size();
+               frame.audio_size = hdr.audio_size();
 
-               if (fseek(fp, frame.offset + frame.size, SEEK_SET) == -1) {
+               if (frame.offset + frame.size + frame.audio_size > file_len ||
+                   fseek(fp, frame.offset + frame.size + frame.audio_size, SEEK_SET) == -1) {
                        fprintf(stderr, "WARNING: %s: Could not seek past frame (probably truncated).\n", filename);
-                       continue;
+                       break;
                }
 
                if (hdr.stream_idx() >= 0 && hdr.stream_idx() < MAX_STREAMS) {
@@ -413,7 +434,7 @@ void load_existing_frames()
                if (de == nullptr) {
                        if (errno != 0) {
                                perror("readdir");
-                               exit(1);
+                               abort();
                        }
                        break;
                }
@@ -425,7 +446,7 @@ void load_existing_frames()
                }
 
                if (progress.wasCanceled()) {
-                       exit(1);
+                       abort();
                }
        }
        closedir(dir);
@@ -443,7 +464,7 @@ void load_existing_frames()
                load_frame_file(frame_filenames[i].c_str(), frame_basenames[i], i, &db);
                progress.setValue(i + 3);
                if (progress.wasCanceled()) {
-                       exit(1);
+                       abort();
                }
        }
 
@@ -488,8 +509,22 @@ void record_thread_func()
                        continue;
                }
 
-               int64_t last_pts = -1;
+               // Match any audio streams to video streams, sequentially.
+               vector<int> video_stream_idx, audio_stream_idx;
+               for (unsigned i = 0; i < format_ctx->nb_streams; ++i) {
+                       if (format_ctx->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
+                               video_stream_idx.push_back(i);
+                       } else if (format_ctx->streams[i]->codecpar->codec_type == AVMEDIA_TYPE_AUDIO) {
+                               audio_stream_idx.push_back(i);
+                       }
+               }
+               unordered_map<int, int> audio_stream_to_video_stream_idx;
+               for (size_t i = 0; i < min(video_stream_idx.size(), audio_stream_idx.size()); ++i) {
+                       audio_stream_to_video_stream_idx[audio_stream_idx[i]] = video_stream_idx[i];
+               }
 
+               vector<uint32_t> pending_audio[MAX_STREAMS];
+               int64_t last_pts = -1;
                while (!should_quit.load()) {
                        AVPacket pkt;
                        unique_ptr<AVPacket, decltype(av_packet_unref) *> pkt_cleanup(
@@ -503,7 +538,23 @@ void record_thread_func()
                        if (av_read_frame(format_ctx.get(), &pkt) != 0) {
                                break;
                        }
-                       if (pkt.stream_index >= MAX_STREAMS) {
+
+                       AVStream *stream = format_ctx->streams[pkt.stream_index];
+                       if (stream->codecpar->codec_type == AVMEDIA_TYPE_AUDIO &&
+                           audio_stream_to_video_stream_idx.count(pkt.stream_index)) {
+                               if ((pkt.size % (sizeof(uint32_t) * 2)) != 0) {
+                                       fprintf(stderr, "Audio stream %u had a packet of strange length %d, ignoring.\n",
+                                               pkt.stream_index, pkt.size);
+                               } else {
+                                       // TODO: Endianness?
+                                       const uint32_t *begin = (const uint32_t *)pkt.data;
+                                       const uint32_t *end = (const uint32_t *)(pkt.data + pkt.size);
+                                       pending_audio[audio_stream_to_video_stream_idx[pkt.stream_index]].assign(begin, end);
+                               }
+                       }
+
+                       if (pkt.stream_index >= MAX_STREAMS ||
+                           stream->codecpar->codec_type != AVMEDIA_TYPE_VIDEO) {
                                continue;
                        }
 
@@ -511,7 +562,7 @@ void record_thread_func()
                        metric_received_frame_size_bytes.count_event(pkt.size);
 
                        // Convert pts to our own timebase.
-                       AVRational stream_timebase = format_ctx->streams[pkt.stream_index]->time_base;
+                       AVRational stream_timebase = stream->time_base;
                        int64_t pts = av_rescale_q(pkt.pts, stream_timebase, AVRational{ 1, TIMEBASE });
 
                        // Translate offset into our stream.
@@ -522,7 +573,7 @@ void record_thread_func()
 
                        //fprintf(stderr, "Got a frame from camera %d, pts = %ld, size = %d\n",
                        //      pkt.stream_index, pts, pkt.size);
-                       FrameOnDisk frame = write_frame(pkt.stream_index, pts, pkt.data, pkt.size, &db);
+                       FrameOnDisk frame = write_frame(pkt.stream_index, pts, pkt.data, pkt.size, move(pending_audio[pkt.stream_index]), &db);
 
                        post_to_main_thread([pkt, frame] {
                                global_mainwindow->display_frame(pkt.stream_index, frame);
@@ -535,8 +586,10 @@ void record_thread_func()
                        current_pts = pts;
                }
 
-               fprintf(stderr, "%s: Hit EOF. Waiting one second and trying again...\n", global_flags.stream_source.c_str());
-               sleep(1);
+               if (!should_quit.load()) {
+                       fprintf(stderr, "%s: Hit EOF. Waiting one second and trying again...\n", global_flags.stream_source.c_str());
+                       sleep(1);
+               }
 
                start_pts = last_pts + TIMEBASE;
        }