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
(cherry picked from commit 237a3a6402)

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);
 }
 

recording-daemon/packet.c

@@ -144,6 +144,7 @@ void ssrc_tls_state(ssrc_t *ssrc) {
 }
 
 
+// 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->output = NULL;
@@ -165,6 +166,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;
@@ -368,16 +373,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;