Log a warning when we kill a client that is not keeping up.

[nageru] / futatabi / frame_on_disk.cpp

diff --git a/futatabi/frame_on_disk.cpp b/futatabi/frame_on_disk.cpp
index 34fcd38ef768394edb59d4e08ea9796fd2451de1..f9a5639259be7a153777feb4ca76bded039d6185 100644 (file)
--- a/futatabi/frame_on_disk.cpp
+++ b/futatabi/frame_on_disk.cpp
@@ -1,12 +1,13 @@
-#include <fcntl.h>
-#include <unistd.h>
+#include "frame_on_disk.h"
+
+#include "shared/metrics.h"
 
 #include <atomic>
 #include <chrono>
+#include <assert.h>
+#include <fcntl.h>
 #include <mutex>
-
-#include "frame_on_disk.h"
-#include "shared/metrics.h"
+#include <unistd.h>
 
 using namespace std;
 using namespace std::chrono;
@@ -16,10 +17,10 @@ namespace {
 // There can be multiple FrameReader classes, so make all the metrics static.
 once_flag frame_metrics_inited;
 
-atomic<int64_t> metric_frame_opened_files{0};
-atomic<int64_t> metric_frame_closed_files{0};
-atomic<int64_t> metric_frame_read_bytes{0};
-atomic<int64_t> metric_frame_read_frames{0};
+atomic<int64_t> metric_frame_opened_files{ 0 };
+atomic<int64_t> metric_frame_closed_files{ 0 };
+atomic<int64_t> metric_frame_read_bytes{ 0 };
+atomic<int64_t> metric_frame_read_frames{ 0 };
 
 Summary metric_frame_read_time_seconds;
 
@@ -27,13 +28,13 @@ Summary metric_frame_read_time_seconds;
 
 FrameReader::FrameReader()
 {
-       call_once(frame_metrics_inited, []{
+       call_once(frame_metrics_inited, [] {
                global_metrics.add("frame_opened_files", &metric_frame_opened_files);
                global_metrics.add("frame_closed_files", &metric_frame_closed_files);
                global_metrics.add("frame_read_bytes", &metric_frame_read_bytes);
                global_metrics.add("frame_read_frames", &metric_frame_read_frames);
 
-               vector<double> quantiles{0.01, 0.1, 0.25, 0.5, 0.75, 0.9, 0.99};
+               vector<double> quantiles{ 0.01, 0.1, 0.25, 0.5, 0.75, 0.9, 0.99 };
                metric_frame_read_time_seconds.init(quantiles, 60.0);
                global_metrics.add("frame_read_time_seconds", &metric_frame_read_time_seconds);
        });
@@ -47,8 +48,30 @@ FrameReader::~FrameReader()
        }
 }
 
-string FrameReader::read_frame(FrameOnDisk frame)
+namespace {
+
+string read_string(int fd, size_t size, off_t offset)
 {
+       string str;
+       str.resize(size);
+       size_t str_offset = 0;
+       while (str_offset < size) {
+               int ret = pread(fd, &str[str_offset], size - str_offset, offset + str_offset);
+               if (ret <= 0) {
+                       perror("pread");
+                       abort();
+               }
+
+               str_offset += ret;
+       }
+       return str;
+}
+
+}  // namespace
+
+FrameReader::Frame FrameReader::read_frame(FrameOnDisk frame, bool read_video, bool read_audio)
+{
+       assert(read_video || read_audio);
        steady_clock::time_point start = steady_clock::now();
 
        if (int(frame.filename_idx) != last_filename_idx) {
@@ -66,7 +89,7 @@ string FrameReader::read_frame(FrameOnDisk frame)
                fd = open(filename.c_str(), O_RDONLY);
                if (fd == -1) {
                        perror(filename.c_str());
-                       exit(1);
+                       abort();
                }
 
                // We want readahead. (Ignore errors.)
@@ -76,17 +99,12 @@ string FrameReader::read_frame(FrameOnDisk frame)
                ++metric_frame_opened_files;
        }
 
-       string str;
-       str.resize(frame.size);
-       off_t offset = 0;
-       while (offset < frame.size) {
-               int ret = pread(fd, &str[offset], frame.size - offset, frame.offset + offset);
-               if (ret <= 0) {
-                       perror("pread");
-                       exit(1);
-               }
-
-               offset += ret;
+       Frame ret;
+       if (read_video) {
+               ret.video = read_string(fd, frame.size, frame.offset);
+       }
+       if (read_audio) {
+               ret.audio = read_string(fd, frame.audio_size, frame.offset + frame.size);
        }
 
        steady_clock::time_point stop = steady_clock::now();
@@ -95,5 +113,5 @@ string FrameReader::read_frame(FrameOnDisk frame)
        metric_frame_read_bytes += frame.size;
        ++metric_frame_read_frames;
 
-       return str;
+       return ret;
 }