libobs: Add `obs_encoder_group_keyframe_aligned_encoders`

Ensures grouped encoders start on the same input frame

libobs/obs-encoder.c

@@ -324,6 +324,17 @@ static void add_connection(struct obs_encoder *encoder)
 		}
 	}
 
+	if (encoder->encoder_group) {
+		bool ready = false;
+		pthread_mutex_lock(&encoder->encoder_group->mutex);
+		encoder->encoder_group->encoders_started += 1;
+		ready = encoder->encoder_group->encoders_started ==
+			encoder->encoder_group->encoders_added;
+		pthread_mutex_unlock(&encoder->encoder_group->mutex);
+		if (ready)
+			add_ready_encoder_group(encoder);
+	}
+
 	set_encoder_active(encoder, true);
 }
 
@@ -340,6 +351,12 @@ static void remove_connection(struct obs_encoder *encoder, bool shutdown)
 		}
 	}
 
+	if (encoder->encoder_group) {
+		pthread_mutex_lock(&encoder->encoder_group->mutex);
+		encoder->encoder_group->encoders_started -= 1;
+		pthread_mutex_unlock(&encoder->encoder_group->mutex);
+	}
+
 	/* obs_encoder_shutdown locks init_mutex, so don't call it on encode
 	 * errors, otherwise you can get a deadlock with outputs when they end
 	 * data capture, which will lock init_mutex and the video callback
@@ -375,6 +392,23 @@ static void obs_encoder_actually_destroy(obs_encoder_t *encoder)
 		blog(LOG_DEBUG, "encoder '%s' destroyed",
 		     encoder->context.name);
 
+		if (encoder->encoder_group) {
+			struct encoder_group *group = encoder->encoder_group;
+			bool release = false;
+
+			encoder->encoder_group = NULL;
+
+			pthread_mutex_lock(&group->mutex);
+			group->encoders_added -= 1;
+			release = group->encoders_added == 0;
+			pthread_mutex_unlock(&group->mutex);
+
+			if (release) {
+				pthread_mutex_destroy(&group->mutex);
+				bfree(group);
+			}
+		}
+
 		free_audio_buffers(encoder);
 
 		if (encoder->context.data)
@@ -1408,6 +1442,16 @@ static void receive_video(void *param, struct video_data *frame)
 	struct obs_encoder **paired = encoder->paired_encoders.array;
 	struct encoder_frame enc_frame;
 
+	if (encoder->encoder_group && !encoder->start_ts) {
+		struct encoder_group *group = encoder->encoder_group;
+		bool ready = false;
+		pthread_mutex_lock(&group->mutex);
+		ready = group->start_timestamp == frame->timestamp;
+		pthread_mutex_unlock(&group->mutex);
+		if (!ready)
+			goto wait_for_audio;
+	}
+
 	if (!encoder->first_received && encoder->paired_encoders.num) {
 		for (size_t i = 0; i < encoder->paired_encoders.num; i++) {
 			if (!paired[i]->first_received ||
@@ -1995,3 +2039,133 @@ uint32_t obs_encoder_get_roi_increment(const obs_encoder_t *encoder)
 {
 	return encoder->roi_increment;
 }
+
+bool obs_encoder_group_keyframe_aligned_encoders(
+	obs_encoder_t *encoder, obs_encoder_t *encoder_to_be_grouped)
+{
+	if (!obs_encoder_valid(encoder,
+			       "obs_encoder_group_keyframe_aligned_encoders") ||
+	    !obs_encoder_valid(encoder_to_be_grouped,
+			       "obs_encoder_group_keyframe_aligned_encoders"))
+		return false;
+
+	if (obs_encoder_active(encoder) ||
+	    obs_encoder_active(encoder_to_be_grouped)) {
+		obs_encoder_t *active = obs_encoder_active(encoder)
+						? encoder
+						: encoder_to_be_grouped;
+		obs_encoder_t *other = active == encoder ? encoder_to_be_grouped
+							 : encoder;
+		blog(LOG_ERROR,
+		     "obs_encoder_group_keyframe_aligned_encoders: encoder '%s' "
+		     "is already active, could not group with '%s'",
+		     obs_encoder_get_name(active), obs_encoder_get_name(other));
+		return false;
+	}
+
+	if (encoder_to_be_grouped->encoder_group) {
+		blog(LOG_ERROR,
+		     "obs_encoder_group_keyframe_aligned_encoders: encoder '%s' "
+		     "is already part of a keyframe aligned group while trying "
+		     "to group with encoder '%s'",
+		     obs_encoder_get_name(encoder_to_be_grouped),
+		     obs_encoder_get_name(encoder));
+		return false;
+	}
+
+	bool unlock = false;
+	if (!encoder->encoder_group) {
+		encoder->encoder_group = bzalloc(sizeof(struct encoder_group));
+		if (pthread_mutex_init(&encoder->encoder_group->mutex, NULL) <
+		    0) {
+			bfree(encoder->encoder_group);
+			encoder->encoder_group = NULL;
+			return false;
+		}
+
+		encoder->encoder_group->encoders_added = 1;
+	} else {
+		pthread_mutex_lock(&encoder->encoder_group->mutex);
+		unlock = true;
+		if (encoder->encoder_group->encoders_started != 0) {
+			blog(LOG_ERROR,
+			     "obs_encoder_group_keyframe_aligned_encoders: "
+			     "Can't add encoder '%s' to active group "
+			     "from encoder '%s'",
+			     obs_encoder_get_name(encoder_to_be_grouped),
+			     obs_encoder_get_name(encoder));
+			pthread_mutex_unlock(&encoder->encoder_group->mutex);
+			return false;
+		}
+	}
+
+	encoder->encoder_group->encoders_added += 1;
+	encoder_to_be_grouped->encoder_group = encoder->encoder_group;
+
+	if (unlock)
+		pthread_mutex_unlock(&encoder->encoder_group->mutex);
+
+	return true;
+}
+
+bool obs_encoder_group_remove_keyframe_aligned_encoder(
+	obs_encoder_t *encoder, obs_encoder_t *encoder_to_be_ungrouped)
+{
+	if (!obs_encoder_valid(
+		    encoder,
+		    "obs_encoder_group_remove_keyframe_aligned_encoder") ||
+	    !obs_encoder_valid(
+		    encoder_to_be_ungrouped,
+		    "obs_encoder_group_remove_keyframe_aligned_encoder"))
+		return false;
+
+	if (obs_encoder_active(encoder) ||
+	    obs_encoder_active(encoder_to_be_ungrouped)) {
+		blog(LOG_ERROR,
+		     "obs_encoder_group_remove_keyframe_aligned_encoder: encoders are active, "
+		     "could not ungroup encoder '%s' from '%s'",
+		     obs_encoder_get_name(encoder_to_be_ungrouped),
+		     obs_encoder_get_name(encoder));
+		return false;
+	}
+
+	if (encoder->encoder_group != encoder_to_be_ungrouped->encoder_group) {
+		blog(LOG_ERROR,
+		     "obs_encoder_group_remove_keyframe_aligned_encoder: "
+		     "encoder '%s' does not belong to the same group as encoder '%s'",
+		     obs_encoder_get_name(encoder_to_be_ungrouped),
+		     obs_encoder_get_name(encoder));
+		return false;
+	}
+
+	struct encoder_group *current_group = encoder->encoder_group;
+	struct encoder_group *free_group = NULL;
+
+	pthread_mutex_lock(&current_group->mutex);
+
+	if (current_group->encoders_started != 0) {
+		blog(LOG_ERROR,
+		     "obs_encoder_group_remove_keyframe_aligned_encoder: "
+		     "could not ungroup encoder '%s' from '%s' while "
+		     "the group contains active encoders",
+		     obs_encoder_get_name(encoder_to_be_ungrouped),
+		     obs_encoder_get_name(encoder));
+		pthread_mutex_unlock(&current_group->mutex);
+		return false;
+	}
+
+	current_group->encoders_added -= 1;
+	encoder_to_be_ungrouped->encoder_group = NULL;
+	if (current_group->encoders_added == 1) {
+		free_group = current_group;
+		encoder->encoder_group = NULL;
+	}
+	pthread_mutex_unlock(&current_group->mutex);
+
+	if (free_group) {
+		pthread_mutex_destroy(&free_group->mutex);
+		bfree(free_group);
+	}
+
+	return true;
+}

libobs/obs-internal.h

@@ -361,11 +361,16 @@ struct obs_core_video {
 	pthread_mutex_t task_mutex;
 	struct deque tasks;
 
+	pthread_mutex_t encoder_group_mutex;
+	DARRAY(obs_weak_encoder_t *) ready_encoder_groups;
+
 	pthread_mutex_t mixes_mutex;
 	DARRAY(struct obs_core_video_mix *) mixes;
 	struct obs_core_video_mix *main_mix;
 };
 
+extern void add_ready_encoder_group(obs_encoder_t *encoder);
+
 struct audio_monitor;
 
 struct obs_core_audio {
@@ -1218,6 +1223,13 @@ struct encoder_callback {
 	void *param;
 };
 
+struct encoder_group {
+	pthread_mutex_t mutex;
+	uint32_t encoders_added;
+	uint32_t encoders_started;
+	uint64_t start_timestamp;
+};
+
 struct obs_encoder {
 	struct obs_context_data context;
 	struct obs_encoder_info info;
@@ -1282,6 +1294,9 @@ struct obs_encoder {
 	uint64_t first_raw_ts;
 	uint64_t start_ts;
 
+	/* track encoders that are part of a gop-aligned multi track group */
+	struct encoder_group *encoder_group;
+
 	pthread_mutex_t outputs_mutex;
 	DARRAY(obs_output_t *) outputs;
 

libobs/obs-video-gpu-encode.c

@@ -91,6 +91,17 @@ static void *gpu_encode_thread(void *data)
 			pkt.timebase_den = encoder->timebase_den;
 			pkt.encoder = encoder;
 
+			if (encoder->encoder_group && !encoder->start_ts) {
+				struct encoder_group *group =
+					encoder->encoder_group;
+				bool ready = false;
+				pthread_mutex_lock(&group->mutex);
+				ready = group->start_timestamp == timestamp;
+				pthread_mutex_unlock(&group->mutex);
+				if (!ready)
+					continue;
+			}
+
 			if (!encoder->first_received && num_paired) {
 				bool wait_for_audio = false;
 

libobs/obs-video.c

@@ -887,6 +887,14 @@ static inline void output_video_data(struct obs_core_video_mix *video,
 	}
 }
 
+void add_ready_encoder_group(obs_encoder_t *encoder)
+{
+	obs_weak_encoder_t *weak = obs_encoder_get_weak_encoder(encoder);
+	pthread_mutex_lock(&obs->video.encoder_group_mutex);
+	da_push_back(obs->video.ready_encoder_groups, &weak);
+	pthread_mutex_unlock(&obs->video.encoder_group_mutex);
+}
+
 static inline void video_sleep(struct obs_core_video *video, uint64_t *p_time,
 			       uint64_t interval_ns)
 {
@@ -915,6 +923,28 @@ static inline void video_sleep(struct obs_core_video *video, uint64_t *p_time,
 	vframe_info.timestamp = cur_time;
 	vframe_info.count = count;
 
+	pthread_mutex_lock(&video->encoder_group_mutex);
+	for (size_t i = 0; i < video->ready_encoder_groups.num; i++) {
+		obs_encoder_t *encoder = obs_weak_encoder_get_encoder(
+			video->ready_encoder_groups.array[i]);
+		obs_weak_encoder_release(video->ready_encoder_groups.array[i]);
+		if (!encoder)
+			continue;
+
+		if (encoder->encoder_group) {
+			struct encoder_group *group = encoder->encoder_group;
+			pthread_mutex_lock(&group->mutex);
+			if (group->encoders_added == group->encoders_started &&
+			    !group->start_timestamp) {
+				group->start_timestamp = *p_time;
+			}
+			pthread_mutex_unlock(&group->mutex);
+		}
+		obs_encoder_release(encoder);
+	}
+	da_clear(video->ready_encoder_groups);
+	pthread_mutex_unlock(&video->encoder_group_mutex);
+
 	pthread_mutex_lock(&obs->video.mixes_mutex);
 	for (size_t i = 0, num = obs->video.mixes.num; i < num; i++) {
 		struct obs_core_video_mix *video = obs->video.mixes.array[i];

libobs/obs.c

@@ -746,6 +746,8 @@ static int obs_init_video(struct obs_video_info *ovi)
 
 	if (pthread_mutex_init(&video->task_mutex, NULL) < 0)
 		return OBS_VIDEO_FAIL;
+	if (pthread_mutex_init(&video->encoder_group_mutex, NULL) < 0)
+		return OBS_VIDEO_FAIL;
 	if (pthread_mutex_init(&video->mixes_mutex, NULL) < 0)
 		return OBS_VIDEO_FAIL;
 
@@ -884,6 +886,15 @@ static void obs_free_video(void)
 	pthread_mutex_init_value(&obs->video.mixes_mutex);
 	da_free(obs->video.mixes);
 
+	for (size_t i = 0; i < obs->video.ready_encoder_groups.num; i++) {
+		obs_weak_encoder_release(
+			obs->video.ready_encoder_groups.array[i]);
+	}
+	da_free(obs->video.ready_encoder_groups);
+
+	pthread_mutex_destroy(&obs->video.encoder_group_mutex);
+	pthread_mutex_init_value(&obs->video.encoder_group_mutex);
+
 	pthread_mutex_destroy(&obs->video.task_mutex);
 	pthread_mutex_init_value(&obs->video.task_mutex);
 	deque_free(&obs->video.tasks);
@@ -1236,6 +1247,7 @@ static bool obs_init(const char *locale, const char *module_config_path,
 	pthread_mutex_init_value(&obs->audio.monitoring_mutex);
 	pthread_mutex_init_value(&obs->audio.task_mutex);
 	pthread_mutex_init_value(&obs->video.task_mutex);
+	pthread_mutex_init_value(&obs->video.encoder_group_mutex);
 	pthread_mutex_init_value(&obs->video.mixes_mutex);
 
 	obs->name_store_owned = !store;

libobs/obs.h

@@ -2600,6 +2600,11 @@ EXPORT void obs_encoder_set_last_error(obs_encoder_t *encoder,
 
 EXPORT uint64_t obs_encoder_get_pause_offset(const obs_encoder_t *encoder);
 
+EXPORT bool obs_encoder_group_keyframe_aligned_encoders(
+	obs_encoder_t *encoder, obs_encoder_t *encoder_to_be_grouped);
+EXPORT bool obs_encoder_group_remove_keyframe_aligned_encoder(
+	obs_encoder_t *encoder, obs_encoder_t *encoder_to_be_ungrouped);
+
 /* ------------------------------------------------------------------------- */
 /* Stream Services */