From 9e772be0d5e19dc4dcef05ee2b4ce47c42788f8e Mon Sep 17 00:00:00 2001
From: Sylvain BERTRAND <sylvain.bertrand@legeek.net>
Date: Sun, 16 May 2021 18:06:23 +0000
Subject: [PATCH] npv: inverted resync logic and rate limiter

---
 npv/config.h                    |  5 +++++
 npv/pipeline/local/code.frag.c  | 29 ++++++++++++++++++++++++++++-
 npv/pipeline/local/state.frag.c |  1 +
 npv/pipeline/namespace/main.c   |  4 ++++
 npv/pipeline/public/code.frag.c |  1 +
 npv/video/local/code.frag.c     |  2 +-
 6 files changed, 40 insertions(+), 2 deletions(-)

diff --git a/npv/config.h b/npv/config.h
index 090af26..fea4c64 100644
--- a/npv/config.h
+++ b/npv/config.h
@@ -85,4 +85,9 @@ struct npv_x11_bind_t npv_x11_binds[] = {
  * a time window based on the following value.
  */
 #define DISCONT_BACKWARD_RESYNC_MS 500
+/*
+ * the rate limiter while resyncing, since it can flood memory. We target 16 ms
+ * or 60 fps
+ */
+#define RESYNC_RATE_LIMITER_MS 16
 #endif
diff --git a/npv/pipeline/local/code.frag.c b/npv/pipeline/local/code.frag.c
index aac6b3d..4ab9aec 100644
--- a/npv/pipeline/local/code.frag.c
+++ b/npv/pipeline/local/code.frag.c
@@ -141,6 +141,31 @@ STATIC void *audio_thd_entry(void *arg)
 	audio();
 	/* unreachable */
 }
+/* will return true to block, false to let it pass */
+STATIC bool rate_limiter_do_block(void)
+{
+	int r;
+	struct timespec current_tp;
+	s64 current;
+	s64 previous;
+
+	memset(&current_tp, 0, sizeof(current_tp));
+	r = clock_gettime(CLOCK_MONOTONIC, &current_tp);
+	if (r == -1)
+		return false; /* error, let it pass */
+	previous = (s64)rate_limiter_previous_tp.tv_sec * 1000 +
+				(s64)rate_limiter_previous_tp.tv_nsec / 1000000;
+	current = (s64)current_tp.tv_sec * 1000
+					+ (s64)current_tp.tv_nsec / 1000000;
+#ifdef NPV_DEBUG
+	if (npv_debug_p)
+		npv_perr("DEBUG:%s:delta=%ld\n", __func__, current - previous);
+#endif
+	if ((current - previous) <= RESYNC_RATE_LIMITER_MS)
+		return true;
+	memcpy(&rate_limiter_previous_tp, &current_tp, sizeof(current_tp));
+	return false;
+}
 /*
  * predecoded video frs are expensive from the perspectives of the cpu and the
  * mem hierarchy.
@@ -158,8 +183,10 @@ STATIC bool have_enough_predecoded_video_frs(void)
 		 * We are looking for a "resyncing" fr, and to secure we 
 		 * receive one, we must decode more frs or the dec frs a
 		 * may endup full and dead locked.
+		 * In pratice, this can burst mem with frs, let's rate
+		 * limit it 16 ms (~60fps).
 		 */ 
-		r = false;
+		r = rate_limiter_do_block();
 	else if (npv_video_dec_frs_p.n < DEC_FRS_ARRAY_N_MAX) {
 		r = false;
 	} else /* >= DEC_FRS_ARRAY_N_MAX && ! RESYNCING */
diff --git a/npv/pipeline/local/state.frag.c b/npv/pipeline/local/state.frag.c
index c537fb4..a8c0388 100644
--- a/npv/pipeline/local/state.frag.c
+++ b/npv/pipeline/local/state.frag.c
@@ -1 +1,2 @@
 STATIC avcodec_pkt_ref_t *eof_pkt_l;
+STATIC struct timespec rate_limiter_previous_tp;
diff --git a/npv/pipeline/namespace/main.c b/npv/pipeline/namespace/main.c
index df1bddf..d303c84 100644
--- a/npv/pipeline/namespace/main.c
+++ b/npv/pipeline/namespace/main.c
@@ -12,6 +12,8 @@
 #define have_enough_predecoded_audio_frs	npv_pipeline_have_enough_predecoded_audio_frs
 #define have_enough_predecoded_video_frs	npv_pipeline_have_enough_predecoded_video_frs
 #define pout					npv_pipeline_pout
+#define rate_limiter_do_block			npv_pipeline_rate_limiter_do_block
+#define rate_limiter_previous_tp		npv_pipeline_rate_limiter_previous_tp
 #define read					npv_pipeline_read
 #define read_thd_entry				npv_pipeline_read_thd_entry
 #define timer_ack				npv_pipeline_timer_ack
@@ -28,6 +30,8 @@
 #undef have_enough_predecoded_audio_frs
 #undef have_enough_predecoded_video_frs
 #undef pout
+#undef rate_limiter_do_block
+#undef rate_limiter_previous_tp
 #undef read
 #undef read_thd_entry
 #undef timer_ack
diff --git a/npv/pipeline/public/code.frag.c b/npv/pipeline/public/code.frag.c
index f522590..bc341e7 100644
--- a/npv/pipeline/public/code.frag.c
+++ b/npv/pipeline/public/code.frag.c
@@ -38,6 +38,7 @@ STATIC void init_once(void)
 	r = pthread_mutex_init(&limits_p.mutex, 0);
 	if (r != 0)
 		fatal("unable to initialize the mutex to guard the accounting of limits\n");
+	memset(&rate_limiter_previous_tp, 0, sizeof(rate_limiter_previous_tp));
 }
 STATIC void limits_lock(void)
 {
diff --git a/npv/video/local/code.frag.c b/npv/video/local/code.frag.c
index 20a1a18..292835d 100644
--- a/npv/video/local/code.frag.c
+++ b/npv/video/local/code.frag.c
@@ -748,7 +748,7 @@ STATIC void discont_backward(s64 now)
 	if (dec_frs_p.discont_backward.prev_now != AV_NOPTS_VALUE
 				&& now < dec_frs_p.discont_backward.prev_now)
 		dec_frs_p.discont_backward.resyncing = true;
-	if (dec_frs_p.discont_backward.resyncing)
+	if (!dec_frs_p.discont_backward.resyncing)
 		goto exit;
 #ifdef NPV_DEBUG
 	if (npv_debug_p)
-- 
2.11.4.GIT