From 402b2180f800a4bf0dc90157938eca2890dff8ac Mon Sep 17 00:00:00 2001
From: "Steinar H. Gunderson" <sgunderson@bigfoot.com>
Date: Sat, 2 Apr 2016 15:22:11 +0200
Subject: [PATCH] Add a system for queue length policies, so that we have as
 little as possible (but no less) delay between inputs.

---
 mixer.cpp | 35 +++++++++++++++++++++++++++++++++++
 mixer.h   | 43 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 78 insertions(+)

diff --git a/mixer.cpp b/mixer.cpp
index 104a6c0..52d285c 100644
--- a/mixer.cpp
+++ b/mixer.cpp
@@ -111,6 +111,29 @@ string generate_local_dump_filename(int frame)
 
 }  // namespace
 
+void QueueLengthPolicy::update_policy(int queue_length)
+{
+	if (queue_length < 0) {  // Starvation.
+		if (safe_queue_length < 5) {
+			++safe_queue_length;
+			fprintf(stderr, "Card %u: Starvation, increasing safe limit to %u frames\n",
+				card_index, safe_queue_length);
+		}
+		frames_with_at_least_one = 0;
+		return;
+	}
+	if (queue_length > 0) {
+		if (++frames_with_at_least_one >= 50) {
+			--safe_queue_length;
+			fprintf(stderr, "Card %u: Spare frames for more than 50 frames, reducing safe limit to %u frames\n",
+				card_index, safe_queue_length);
+			frames_with_at_least_one = 0;
+		}
+	} else {
+		frames_with_at_least_one = 0;
+	}
+}
+
 Mixer::Mixer(const QSurfaceFormat &format, unsigned num_cards)
 	: httpd(WIDTH, HEIGHT),
 	  num_cards(num_cards),
@@ -182,6 +205,7 @@ Mixer::Mixer(const QSurfaceFormat &format, unsigned num_cards)
 	}
 
 	for (card_index = 0; card_index < num_cards; ++card_index) {
+		cards[card_index].queue_length_policy.reset(card_index);
 		cards[card_index].capture->start_bm_capture();
 	}
 
@@ -593,6 +617,8 @@ void Mixer::thread_func()
 			for (unsigned card_index = 0; card_index < num_cards; ++card_index) {
 				CaptureCard *card = &cards[card_index];
 				if (card->new_frames.empty()) {
+					assert(card_index != master_card_index);
+					card->queue_length_policy.update_policy(-1);
 					continue;
 				}
 				new_frames[card_index] = move(card->new_frames.front());
@@ -604,6 +630,15 @@ void Mixer::thread_func()
 				num_samples[card_index] = num_samples_times_timebase / TIMEBASE;
 				card->fractional_samples = num_samples_times_timebase % TIMEBASE;
 				assert(num_samples[card_index] >= 0);
+
+				if (card_index != master_card_index) {
+					// If we have excess frames compared to the policy for this card,
+					// drop frames from the head.
+					card->queue_length_policy.update_policy(card->new_frames.size());
+					while (card->new_frames.size() > card->queue_length_policy.get_safe_queue_length()) {
+						card->new_frames.pop();
+					}
+				}
 			}
 		}
 
diff --git a/mixer.h b/mixer.h
index 90a8869..f1e5fa2 100644
--- a/mixer.h
+++ b/mixer.h
@@ -52,6 +52,47 @@ class YCbCrInput;
 class QOpenGLContext;
 class QSurfaceFormat;
 
+// For any card that's not the master (where we pick out the frames as they
+// come, as fast as we can process), there's going to be a queue. The question
+// is when we should drop frames from that queue (apart from the obvious
+// dropping if the 16-frame queue should become full), especially given that
+// the frame rate could be lower or higher than the master (either subtly or
+// dramatically). We have two (conflicting) demands:
+//
+//   1. We want to avoid starving the queue.
+//   2. We don't want to add more delay than is needed.
+//
+// Our general strategy is to drop as many frames as we can (helping for #2)
+// that we think is safe for #1 given jitter. To this end, we set a lower floor N,
+// where we assume that if we have N frames in the queue, we're always safe from
+// starvation. (Typically, N will be 0 or 1. It starts off at 0.) If we have
+// more than N frames in the queue after reading out the one we need, we head-drop
+// them to reduce the queue.
+//
+// N is reduced as follows: If the queue has had at least one spare frame for
+// at least 50 (master) frames (ie., it's been too conservative for a second),
+// we reduce N by 1 and reset the timers.
+//
+// Whenever the queue is starved (we needed a frame but there was none), N was
+// obviously too low, so we increment N. We will never set N above 5, though.
+class QueueLengthPolicy {
+public:
+	QueueLengthPolicy() {}
+	void reset(unsigned card_index) {
+		this->card_index = card_index;
+		safe_queue_length = 0;
+		frames_with_at_least_one = 0;
+	}
+
+	void update_policy(int queue_length);  // Give in -1 for starvation.
+	unsigned get_safe_queue_length() const { return safe_queue_length; }
+
+private:
+	unsigned card_index;  // For debugging only.
+	unsigned safe_queue_length = 0;  // Called N in the comments.
+	unsigned frames_with_at_least_one = 0;
+};
+
 class Mixer {
 public:
 	// The surface format is used for offscreen destinations for OpenGL contexts we need.
@@ -340,6 +381,8 @@ private:
 		bool should_quit = false;
 		std::condition_variable new_frames_changed;  // Set whenever new_frames (or should_quit) is changed.
 
+		QueueLengthPolicy queue_length_policy;  // Refers to the "new_frames" queue.
+
 		// Accumulated errors in number of 1/TIMEBASE samples. If OUTPUT_FREQUENCY divided by
 		// frame rate is integer, will always stay zero.
 		unsigned fractional_samples = 0;
-- 
2.39.2