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

@@ -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

@@ -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;