feat(obs-irl-source): dedicated video output thread

move the gpu-to-cpu frame transfer, format conversion, and
obs_source_output_video off the receiver thread onto a dedicated
video thread fed by a small frame queue, so a gpu or delivery stall
can never starve audio decode. queue overflow drops the oldest
frame (counted as vq_drops in stats) and never blocks the receiver.

the receiver converts video pts to nanoseconds before queueing
because it can close fmt_ctx mid-reconnect while frames are still
in flight; sws_ctx ownership moves to the video thread and is no
longer freed on stream close. the decoder gets extra_hw_frames
matching the queue depth so held frames cannot exhaust the hw
surface pool.

Author: datagutt

CLAUDE.md

@@ -84,7 +84,8 @@ Buffer regulation happens through playback speed only, asymmetric like IRLToolki
 ### Threading model
 
 - **Main/OBS thread**: calls create, destroy, update, tick, get_properties
-- **Receiver thread**: owns all FFmpeg state. Writes to the audio buffer (mutex protected). Outputs video frames directly to OBS via `obs_source_output_video`.
+- **Receiver thread**: owns demux/decode FFmpeg state. Writes to the audio buffer (mutex protected) and pushes decoded video frames (PTS pre-converted to nanoseconds) onto the video queue. Never blocks on GPU or OBS video delivery.
+- **Video thread**: pops the video queue, does the HW frame transfer and format conversion (owns sws_ctx), and calls `obs_source_output_video`. Queue overflow drops the oldest frame (`video_queue_drops`).
 - **Audio thread**: drains the jitter buffer and submits audio to OBS via `obs_source_output_audio`, paced against the sample counter output clock. Shared timing state is protected by `audio_state_lock` (lock order: `audio_state_lock` before the buffer mutex).
 
 ### OBS API conventions

include/irl-source.h

@@ -116,10 +116,27 @@ struct irl_source {
 	/* Receiver / demux thread */
 	pthread_t receiver_thread;
 	pthread_t audio_thread;
+	pthread_t video_thread;
 	pthread_mutex_t audio_state_lock;
 	volatile bool thread_active;
 	volatile bool reconnecting;
 
+	/* Video output queue (receiver thread → video thread).
+	 * Decouples the GPU→CPU frame transfer and format conversion
+	 * from the receiver thread so a GPU stall cannot starve audio
+	 * decode. Depth stays small because queued HW frames pin
+	 * decoder surface-pool entries (matched by extra_hw_frames at
+	 * decoder open). Queued frame->pts is in nanoseconds; the
+	 * receiver converts before queueing because it may close
+	 * fmt_ctx while frames are still in flight. */
+#define IRL_VIDEO_QUEUE_SIZE 4
+	pthread_mutex_t video_queue_lock;
+	pthread_cond_t video_queue_cond;
+	AVFrame *video_queue[IRL_VIDEO_QUEUE_SIZE];
+	int video_queue_head;
+	int video_queue_count;
+	uint64_t video_queue_drops;
+
 	/* FFmpeg state (owned by receiver thread) */
 	AVFormatContext *fmt_ctx;
 	/* Armed before each blocking FFmpeg I/O call; interrupt_cb

src/irl-source.c

@@ -131,11 +131,23 @@ static void start_receiver(struct irl_source *ctx)
 		os_atomic_store_bool(&ctx->thread_active, false);
 		return;
 	}
+	if (pthread_create(&ctx->video_thread, NULL, irl_video_thread, ctx) !=
+	    0) {
+		blog(LOG_ERROR,
+		     "[irl-source] Failed to create video thread");
+		os_atomic_store_bool(&ctx->thread_active, false);
+		pthread_join(ctx->audio_thread, NULL);
+		return;
+	}
 	if (pthread_create(&ctx->receiver_thread, NULL, irl_receiver_thread,
 			   ctx) != 0) {
 		blog(LOG_ERROR,
 		     "[irl-source] Failed to create receiver thread");
 		os_atomic_store_bool(&ctx->thread_active, false);
+		pthread_mutex_lock(&ctx->video_queue_lock);
+		pthread_cond_broadcast(&ctx->video_queue_cond);
+		pthread_mutex_unlock(&ctx->video_queue_lock);
+		pthread_join(ctx->video_thread, NULL);
 		pthread_join(ctx->audio_thread, NULL);
 	}
 }
@@ -257,6 +269,8 @@ void *irl_source_create(obs_data_t *settings, obs_source_t *source)
 	ctx->source = source;
 	ctx->current_speed = 1.0f;
 	pthread_mutex_init(&ctx->audio_state_lock, NULL);
+	pthread_mutex_init(&ctx->video_queue_lock, NULL);
+	pthread_cond_init(&ctx->video_queue_cond, NULL);
 
 	config_load(&ctx->config, settings);
 	apply_async_audio_mode(ctx);
@@ -304,6 +318,8 @@ void irl_source_destroy(void *data)
 	stop_receiver(ctx, false);
 	audio_buffer_free(&ctx->audio_buf);
 	pthread_mutex_destroy(&ctx->audio_state_lock);
+	pthread_cond_destroy(&ctx->video_queue_cond);
+	pthread_mutex_destroy(&ctx->video_queue_lock);
 
 	free(ctx->audio_pump_scratch);
 	free(ctx->audio_resample_scratch);

src/receiver-internal.h

@@ -20,4 +20,7 @@ void irl_handle_video_packet(struct irl_source *ctx, AVPacket *pkt,
 			     AVFrame *frame);
 void irl_handle_audio_frame(struct irl_source *ctx, AVFrame *frame);
 void irl_handle_video_frame(struct irl_source *ctx, AVFrame *frame);
+void irl_video_queue_push(struct irl_source *ctx, AVFrame *frame,
+			  int64_t pts_ns);
+void *irl_video_thread(void *data);
 void irl_log_receiver_stats(struct irl_source *ctx);

src/receiver-stream.c

@@ -106,6 +106,11 @@ static AVCodecContext *open_decoder(struct irl_source *src, AVStream *stream,
 		 * decode ignores both settings. */
 		ctx->flags |= AV_CODEC_FLAG_LOW_DELAY;
 		ctx->thread_count = 4;
+		/* The video output queue holds decoded HW frames, each
+		 * pinning a decoder surface; give the pool matching
+		 * headroom or the decoder can stall waiting for a
+		 * surface the queue is sitting on. */
+		ctx->extra_hw_frames = IRL_VIDEO_QUEUE_SIZE;
 	}
 
 	if (try_hw && stream->codecpar->codec_type == AVMEDIA_TYPE_VIDEO) {
@@ -164,10 +169,9 @@ void irl_close_ffmpeg(struct irl_source *ctx)
 	ctx->swr_in_rate = 0;
 	ctx->swr_in_channels = 0;
 	ctx->swr_in_format = AV_SAMPLE_FMT_NONE;
-	if (ctx->sws_ctx) {
-		sws_freeContext(ctx->sws_ctx);
-		ctx->sws_ctx = NULL;
-	}
+	/* sws_ctx is owned by the video thread (it converts queued
+	 * frames that may outlive this connection); it is recreated on
+	 * parameter change and freed at source destroy. */
 
 	if (ctx->audio_dec_ctx) {
 		avcodec_free_context(&ctx->audio_dec_ctx);
@@ -442,7 +446,7 @@ void irl_log_receiver_stats(struct irl_source *ctx)
 	     "norm=%llu interp=%llu silence=%llu resets=%llu "
 	     "last_gap=%dms max_gap=%dms underruns=%llu resync_skips=%llu "
 	     "hidden_trims=%llu quality_events=%llu "
-	     "audio_flushes=%llu video_flushes=%llu "
+	     "audio_flushes=%llu video_flushes=%llu vq_drops=%llu "
 	     "obs_lead=%lldms chunk=%u@%u "
 	     "stream_chunk=%llums obs_chunk=%llums "
 	     "restarts=%llu res=%dx%d",
@@ -464,6 +468,7 @@ void irl_log_receiver_stats(struct irl_source *ctx)
 	     (unsigned long long)ctx->audio_quality_events,
 	     (unsigned long long)ctx->audio_decoder_flushes,
 	     (unsigned long long)ctx->video_decoder_flushes,
+	     (unsigned long long)ctx->video_queue_drops,
 	     (long long)(ctx->audio_last_obs_lead_ns / 1000000LL),
 	     ctx->audio_last_frames_out, ctx->audio_last_samples_per_sec,
 	     (unsigned long long)(ctx->audio_last_chunk_stream_duration_ns /

src/receiver-video.c

@@ -8,6 +8,82 @@
 
 #include "receiver-internal.h"
 
+/* ── Video output queue ───────────────────────────────────── */
+
+static void video_queue_drain_locked(struct irl_source *ctx)
+{
+	while (ctx->video_queue_count > 0) {
+		AVFrame *f = ctx->video_queue[ctx->video_queue_head];
+		ctx->video_queue[ctx->video_queue_head] = NULL;
+		ctx->video_queue_head =
+			(ctx->video_queue_head + 1) % IRL_VIDEO_QUEUE_SIZE;
+		ctx->video_queue_count--;
+		av_frame_free(&f);
+	}
+}
+
+void irl_video_queue_push(struct irl_source *ctx, AVFrame *frame,
+			  int64_t pts_ns)
+{
+	AVFrame *clone = av_frame_alloc();
+	if (!clone)
+		return;
+	if (av_frame_ref(clone, frame) < 0) {
+		av_frame_free(&clone);
+		return;
+	}
+	clone->pts = pts_ns;
+
+	pthread_mutex_lock(&ctx->video_queue_lock);
+	if (ctx->video_queue_count >= IRL_VIDEO_QUEUE_SIZE) {
+		/* Video thread is stalled; keep the freshest frames and
+		 * never make the receiver (and therefore audio) wait. */
+		AVFrame *oldest = ctx->video_queue[ctx->video_queue_head];
+		ctx->video_queue[ctx->video_queue_head] = NULL;
+		ctx->video_queue_head =
+			(ctx->video_queue_head + 1) % IRL_VIDEO_QUEUE_SIZE;
+		ctx->video_queue_count--;
+		ctx->video_queue_drops++;
+		av_frame_free(&oldest);
+	}
+	int tail = (ctx->video_queue_head + ctx->video_queue_count) %
+		   IRL_VIDEO_QUEUE_SIZE;
+	ctx->video_queue[tail] = clone;
+	ctx->video_queue_count++;
+	pthread_cond_signal(&ctx->video_queue_cond);
+	pthread_mutex_unlock(&ctx->video_queue_lock);
+}
+
+void *irl_video_thread(void *data)
+{
+	struct irl_source *ctx = data;
+
+	pthread_mutex_lock(&ctx->video_queue_lock);
+	while (os_atomic_load_bool(&ctx->thread_active)) {
+		if (ctx->video_queue_count == 0) {
+			pthread_cond_wait(&ctx->video_queue_cond,
+					  &ctx->video_queue_lock);
+			continue;
+		}
+		AVFrame *f = ctx->video_queue[ctx->video_queue_head];
+		ctx->video_queue[ctx->video_queue_head] = NULL;
+		ctx->video_queue_head =
+			(ctx->video_queue_head + 1) % IRL_VIDEO_QUEUE_SIZE;
+		ctx->video_queue_count--;
+		pthread_mutex_unlock(&ctx->video_queue_lock);
+
+		irl_video_output_frame(ctx, f);
+		av_frame_free(&f);
+
+		pthread_mutex_lock(&ctx->video_queue_lock);
+	}
+	video_queue_drain_locked(ctx);
+	pthread_mutex_unlock(&ctx->video_queue_lock);
+	return NULL;
+}
+
+/* ── Decoded frame handling (receiver thread) ─────────────── */
+
 static int64_t video_frame_pts(const AVFrame *frame)
 {
 	if (frame->best_effort_timestamp != AV_NOPTS_VALUE)
@@ -76,18 +152,22 @@ void irl_handle_video_frame(struct irl_source *ctx, AVFrame *frame)
 	ctx->last_video_width = frame->width;
 	ctx->last_video_height = frame->height;
 
+	/* Convert PTS to nanoseconds here: the video thread must not
+	 * touch fmt_ctx, which this thread frees on reconnect while
+	 * queued frames may still be in flight. */
+	int64_t pts_ns = 0;
 	if (ctx->fmt_ctx && ctx->video_stream_idx >= 0) {
 		AVStream *vs =
 			ctx->fmt_ctx->streams[ctx->video_stream_idx];
-		int64_t pts_ns = av_rescale_q(frame->pts, vs->time_base,
-					      (AVRational){1, 1000000000});
+		pts_ns = av_rescale_q(frame->pts, vs->time_base,
+				      (AVRational){1, 1000000000});
 		pthread_mutex_lock(&ctx->audio_state_lock);
 		ctx->latest_video_stream_pts_ns = pts_ns;
 		pthread_mutex_unlock(&ctx->audio_state_lock);
 	}
 
-	irl_video_output_frame(ctx, frame);
+	irl_video_queue_push(ctx, frame, pts_ns);
 	ctx->total_video_frames++;
 	if (ctx->total_video_frames == 1)
-		blog(LOG_INFO, "[irl-source] First video frame output");
+		blog(LOG_INFO, "[irl-source] First video frame queued");
 }

src/receiver.c

@@ -121,6 +121,10 @@ void irl_receiver_stop(struct irl_source *ctx)
 		return;
 
 	os_atomic_store_bool(&ctx->thread_active, false);
+	pthread_mutex_lock(&ctx->video_queue_lock);
+	pthread_cond_broadcast(&ctx->video_queue_cond);
+	pthread_mutex_unlock(&ctx->video_queue_lock);
+	pthread_join(ctx->video_thread, NULL);
 	pthread_join(ctx->audio_thread, NULL);
 	pthread_join(ctx->receiver_thread, NULL);
 }

src/video-handler.c

@@ -128,9 +128,10 @@ static void setup_color_params(struct obs_source_frame *obs_frame,
  * video-only wall-clock anchor. */
 static uint64_t frame_timestamp(struct irl_source *ctx, const AVFrame *frame)
 {
-	AVStream *vs = ctx->fmt_ctx->streams[ctx->video_stream_idx];
-	int64_t pts_ns = av_rescale_q(frame->pts, vs->time_base,
-				      (AVRational){1, 1000000000});
+	/* frame->pts is pre-converted to nanoseconds by the receiver
+	 * thread (see irl_video_queue_push); fmt_ctx must not be
+	 * touched here, it can be freed mid-reconnect. */
+	int64_t pts_ns = frame->pts;
 	uint64_t now = os_gettime_ns();
 
 	/* Snapshot audio-thread-owned fields under the lock. */