From 237a3a6402db7f64f319d084b191895ddebbce8d Mon Sep 17 00:00:00 2001
From: Richard Fuchs <rfuchs@sipwise.com>
Date: Thu, 11 May 2023 11:53:59 -0400
Subject: [PATCH] MT#57414 fix shutdown race condition

Packet processing (decoding, encoding, file writing, etc) happens with
the main parent object (metafile) unlocked and only with the specific
child object (SSRC) locked.

During teardown we close all SSRC objects with the metafile locked,
without locking the SSRC objects. This opens up a race condition during
which SSRC closing can happen at the same time as packet processing.

Fix this by locking each SSRC object during teardown, and also clear out
the SSRC hash table while the metafile is locked, which serves as a
marker to a packet processing thread that processing should be skipped.

Closes #1663

Change-Id: I4e8a6a5abfd678a6cfd794a98470a49d60694207
---
 recording-daemon/metafile.c | 11 ++++++++---
 recording-daemon/packet.c   | 16 ++++++++++++----
 2 files changed, 20 insertions(+), 7 deletions(-)

diff --git a/recording-daemon/metafile.c b/recording-daemon/metafile.c
index 163f03b0f..48fe4007a 100644
--- a/recording-daemon/metafile.c
+++ b/recording-daemon/metafile.c
@@ -31,8 +31,7 @@ static void meta_free(void *ptr) {
 	db_close_call(mf);
 	g_string_chunk_free(mf->gsc);
 	// SSRCs first as they have linked outputs which need to be closed first
-	if (mf->ssrc_hash)
-		g_hash_table_destroy(mf->ssrc_hash);
+	g_clear_pointer(&mf->ssrc_hash, g_hash_table_destroy);
 	for (int i = 0; i < mf->streams->len; i++) {
 		stream_t *stream = g_ptr_array_index(mf->streams, i);
 		stream_close(stream); // should be closed already
@@ -51,7 +50,9 @@ static void meta_free(void *ptr) {
 
 static void meta_close_ssrcs(gpointer key, gpointer value, gpointer user_data) {
 	ssrc_t *s = value;
+	pthread_mutex_lock(&s->lock);
 	ssrc_close(s);
+	pthread_mutex_unlock(&s->lock);
 }
 
 // mf is locked
@@ -72,7 +73,11 @@ static void meta_destroy(metafile_t *mf) {
 		mf->forward_fd = -1;
 	}
 	// shut down SSRCs, which closes TLS connections
-	g_hash_table_foreach(mf->ssrc_hash, meta_close_ssrcs, NULL);
+	if (mf->ssrc_hash) {
+		g_hash_table_foreach(mf->ssrc_hash, meta_close_ssrcs, NULL);
+		g_hash_table_destroy(mf->ssrc_hash);
+		mf->ssrc_hash = NULL;
+	}
 	db_close_call(mf);
 }
 
diff --git a/recording-daemon/packet.c b/recording-daemon/packet.c
index 87ba97cc7..63528d52d 100644
--- a/recording-daemon/packet.c
+++ b/recording-daemon/packet.c
@@ -188,6 +188,7 @@ void ssrc_tls_fwd_silence_frames_upto(ssrc_t *ssrc, AVFrame *frame, int64_t upto
 }
 
 
+// appropriate lock must be held (ssrc or metafile)
 void ssrc_close(ssrc_t *s) {
 	output_close(s->metafile, s->output, tag_get(s->metafile, s->stream->tag), s->metafile->discard);
 	s->output = NULL;
@@ -210,6 +211,10 @@ void ssrc_free(void *p) {
 static ssrc_t *ssrc_get(stream_t *stream, unsigned long ssrc) {
 	metafile_t *mf = stream->metafile;
 	pthread_mutex_lock(&mf->lock);
+	if (!mf->ssrc_hash) {
+		pthread_mutex_unlock(&mf->lock);
+		return NULL;
+	}
 	ssrc_t *ret = g_hash_table_lookup(mf->ssrc_hash, GUINT_TO_POINTER(ssrc));
 	if (ret)
 		goto out;
@@ -413,16 +418,19 @@ void packet_process(stream_t *stream, unsigned char *buf, unsigned len) {
 
 	// insert into ssrc queue
 	ssrc_t *ssrc = ssrc_get(stream, ssrc_num);
-	if (packet_sequencer_insert(&ssrc->sequencer, &packet->p) < 0)
-		goto dupe;
+	if (!ssrc) // stream shutdown
+		goto skip;
+	if (packet_sequencer_insert(&ssrc->sequencer, &packet->p) < 0) {
+		dbg("skipping dupe packet (new seq %i prev seq %i)", packet->p.seq, ssrc->sequencer.seq);
+		goto skip;
+	}
 
 	// got a new packet, run the decoder
 	ssrc_run(ssrc);
 	log_info_ssrc = 0;
 	return;
 
-dupe:
-	dbg("skipping dupe packet (new seq %i prev seq %i)", packet->p.seq, ssrc->sequencer.seq);
+skip:
 	pthread_mutex_unlock(&ssrc->lock);
 	packet_free(packet);
 	log_info_ssrc = 0;