Improve audio/video_buffer_push() logic

Signed-off-by: Andrew Opalach <andrew@akon.city>

10 files changed, 150 insertions, 123 deletions

diff --git a/src/buffer/audio.c b/src/buffer/audio.c
index 89f1b70..0d78382 100644
--- a/src/buffer/audio.c
+++ b/src/buffer/audio.c
@@ -14,7 +14,7 @@
 #define BUFFER_MARK_BUFFERED 4.0
 
 #ifdef CAMU_AUDIO_BUFFER_FADE
-#define FADE_STEP(fmt) (1.f / (fmt)->sample_rate)
+#define FADE_STEP(fmt) (1.75f / (fmt)->sample_rate)
 #endif
 
 enum {
@@ -60,8 +60,9 @@ bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_code
     buf->fmt.resampler_needed = !camu_resampler_format_matches(&buf->fmt);
     if (buf->fmt.resampler_needed) {
 #ifdef CAMU_HAVE_FFMPEG
-        buf->resamp = camu_ff_resampler_create();
-        if (!buf->resamp->init(buf->resamp, &buf->fmt)) {
+        buf->resampler = camu_ff_resampler_create();
+        if (!buf->resampler->init(buf->resampler, &buf->fmt)) {
+            // Resampler will be freed in audio_buffer_free().
             return false;
         }
 #else
@@ -106,8 +107,34 @@ void camu_audio_buffer_set_latency(struct camu_audio_buffer *buf, f64 latency)
 #define OCCUPIED_SECONDS_DEBUG(buf) 0
 #endif
 
-static bool push_internal(struct camu_audio_buffer *buf, u8 *data, s32 sample_count)
+static inline bool frame_is_late(struct camu_clock *clock, f64 base, f64 pts, f64 duration)
 {
+    if (base == -1.0) base = camu_clock_get_base_pts(clock);
+    return pts + duration < base;
+}
+
+// return value of false indicates we pushed to the peak buffer.
+static bool push_internal(struct camu_audio_buffer *buf, f64 pts, u8 **data, s32 sample_count, bool flush)
+{
+    if (!flush) {
+        f64 duration = camu_audio_format_samples_to_sec(&buf->fmt.in, sample_count);
+        f64 base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
+        if (frame_is_late(buf->clock, base, pts, duration)) return true;
+        if (buf->fmt.resampler_needed) {
+            sample_count = buf->resampler->convert(buf->resampler, (const u8 **)data, sample_count);
+            data = buf->resampler->get_data(buf->resampler);
+        }
+        if (sample_count == 0) return true;
+        if (base == -1.0) al_atomic_store(f64)(&buf->pts, pts, AL_ATOMIC_RELEASE);
+    } else {
+        al_assert(!data && sample_count == 0);
+        if (buf->fmt.resampler_needed) {
+            sample_count = buf->resampler->flush(buf->resampler);
+            data = buf->resampler->get_data(buf->resampler);
+        }
+        if (sample_count == 0) return true;
+    }
+
     size_t space = al_ring_buffer_space(&buf->rb);
     if (!buf->buffered && buf->size - space > buf->mark.buffered) {
         al_log_debug("audio_buffer", "Buffered (mark: %.2fs).", OCCUPIED_SECONDS_DEBUG(buf));
@@ -119,7 +146,7 @@ static bool push_internal(struct camu_audio_buffer *buf, u8 *data, s32 sample_co
     size_t peak = camu_peak_buffer_size(&buf->peak);
     if (space < have + peak) {
         al_assert(have < buf->mark.min);
-        camu_peak_buffer_push(&buf->peak, data, have);
+        camu_peak_buffer_push(&buf->peak, data[0], have);
         peak += have;
         if (peak >= buf->mark.min) {
             // If this happens the writer of this buffer is taking way too long to stop.
@@ -138,20 +165,11 @@ static bool push_internal(struct camu_audio_buffer *buf, u8 *data, s32 sample_co
         al_ring_buffer_write(&buf->rb, camu_peak_buffer_flush(&buf->peak), peak);
     }
 
-    al_ring_buffer_write(&buf->rb, data, have);
+    al_ring_buffer_write(&buf->rb, data[0], have);
 
     return true;
 }
 
-static bool frame_is_late(struct camu_audio_buffer *buf, f64 pts, f64 duration)
-{
-    f64 base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
-    if (base == -1.0) {
-        base = camu_clock_get_base_pts(buf->clock);
-        al_atomic_store(f64)(&buf->pts, base, AL_ATOMIC_RELEASE);
-    }
-    return pts + duration < base;
-}
 
 #ifdef CAMU_HAVE_FFMPEG
 static void push_av_frame_internal(struct camu_audio_buffer *buf, AVFrame *frame)
@@ -159,15 +177,7 @@ static void push_av_frame_internal(struct camu_audio_buffer *buf, AVFrame *frame
     s32 sample_count = frame->nb_samples;
     AVStream *stream = buf->stream->av.stream;
     f64 pts = frame->best_effort_timestamp * av_q2d(stream->time_base);
-    f64 duration = camu_audio_format_samples_to_sec(&buf->fmt.in, sample_count);
-    if (!frame_is_late(buf, pts, duration)) {
-        u8 **data = frame->data;
-        if (buf->fmt.resampler_needed) {
-            sample_count = buf->resamp->convert(buf->resamp, (const u8 **)data, sample_count);
-            data = buf->resamp->get_data(buf->resamp);
-        }
-        if (sample_count > 0) push_internal(buf, data[0], sample_count);
-    }
+    push_internal(buf, pts, frame->data, sample_count, false);
     av_frame_free(&frame);
 }
 #endif
@@ -180,17 +190,9 @@ void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_fra
     case CAMU_NORMAL: {
         s32 sample_count = frame->audio.sample_count;
         f64 pts = frame->pts;
-        f64 duration = camu_audio_format_samples_to_sec(&buf->fmt.in, sample_count);
-        if (!frame_is_late(buf, pts, duration)) {
-            u8 *store[CAMU_NUM_DATA_POINTERS] = { 0 };
-            store[0] = frame->data;
-            u8 **data = store;
-            if (buf->fmt.resampler_needed) {
-                sample_count = buf->resamp->convert(buf->resamp, (const u8 **)store, sample_count);
-                data = buf->resamp->get_data(buf->resamp);
-            }
-            push_internal(buf, data[0], sample_count);
-        }
+        u8 *planes[CAMU_PLANAR_DATA_POINTERS] = { 0 };
+        planes[0] = frame->data;
+        push_internal(buf, pts, planes, sample_count, false);
         break;
     }
 #ifdef CAMU_HAVE_FFMPEG
@@ -203,23 +205,12 @@ void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_fra
     al_free(frame);
 }
 
-void camu_audio_buffer_unpause(struct camu_audio_buffer *buf)
-{
-    al_atomic_add(s32)(&buf->unpause, 1, AL_ATOMIC_RELAXED);
-}
-
 // flush() always comes from the same thread as push().
 void camu_audio_buffer_flush(struct camu_audio_buffer *buf)
 {
     al_log_debug("audio_buffer", "Flush requested.");
-    if (buf->fmt.resampler_needed) {
-        s32 sample_count = buf->resamp->flush(buf->resamp);
-        if (sample_count > 0) {
-            u8 **data = buf->resamp->get_data(buf->resamp);
-            if (!push_internal(buf, data[0], sample_count)) {
-                al_log_debug("audio_buffer", "Buffer filled by resampler flush.");
-            }
-        }
+    if (!push_internal(buf, 0.0, NULL, 0, true)) {
+        al_log_debug("audio_buffer", "Buffer filled by flush.");
     }
     if (!buf->buffered) {
         al_log_debug("audio_buffer", "Buffered (mark: %.2fs).", OCCUPIED_SECONDS_DEBUG(buf));
@@ -236,10 +227,15 @@ void camu_audio_buffer_reset(struct camu_audio_buffer *buf)
     al_ring_buffer_reset(&buf->rb);
     camu_peak_buffer_flush(&buf->peak);
     if (buf->fmt.resampler_needed) {
-        buf->resamp->flush(buf->resamp);
+        buf->resampler->flush(buf->resampler);
     }
 }
 
+void camu_audio_buffer_unpause(struct camu_audio_buffer *buf)
+{
+    al_atomic_add(s32)(&buf->unpause, 1, AL_ATOMIC_RELAXED);
+}
+
 static void increment_pts(struct camu_audio_buffer *buf, f64 amount, f64 *base)
 {
     *base += amount;
@@ -426,7 +422,7 @@ size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t re
 void camu_audio_buffer_free(struct camu_audio_buffer *buf)
 {
     if (buf->fmt.resampler_needed) {
-        buf->resamp->free(&buf->resamp);
+        buf->resampler->free(&buf->resampler);
     }
     if (buf->data) al_free(buf->data);
     camu_peak_buffer_free(&buf->peak);
diff --git a/src/buffer/audio.h b/src/buffer/audio.h
index b4d7594..6fc3cca 100644
--- a/src/buffer/audio.h
+++ b/src/buffer/audio.h
@@ -22,7 +22,7 @@ struct camu_audio_buffer {
     f64 latency;
 
     struct camu_resampler_format fmt;
-    struct camu_resampler *resamp;
+    struct camu_resampler *resampler;
 
     u8 *data;
     size_t size;
@@ -62,8 +62,8 @@ bool camu_audio_buffer_configure(struct camu_audio_buffer *buf, struct camu_code
 void camu_audio_buffer_set_volume(struct camu_audio_buffer *buf, f32 volume);
 void camu_audio_buffer_set_latency(struct camu_audio_buffer *buf, f64 latency);
 void camu_audio_buffer_push(struct camu_audio_buffer *buf, struct camu_codec_frame *frame);
-void camu_audio_buffer_unpause(struct camu_audio_buffer *buf);
 void camu_audio_buffer_flush(struct camu_audio_buffer *buf);
 void camu_audio_buffer_reset(struct camu_audio_buffer *buf);
+void camu_audio_buffer_unpause(struct camu_audio_buffer *buf);
 size_t camu_audio_buffer_read(struct camu_audio_buffer *buf, u8 *data, size_t req);
 void camu_audio_buffer_free(struct camu_audio_buffer *buf);
diff --git a/src/buffer/video.c b/src/buffer/video.c
index 92bcb2d..ba43685 100644
--- a/src/buffer/video.c
+++ b/src/buffer/video.c
@@ -44,7 +44,9 @@ bool camu_video_buffer_configure(struct camu_video_buffer *buf, struct camu_code
     case CAMU_NORMAL: {
         camu_video_format_copy(&buf->fmt.in, fmt);
         const char *format_name = camu_pixel_format_name(fmt->format);
+        if (!format_name) format_name = "unknown";
         al_log_info("video_buffer", "Stream: %s (%dx%d) %s.", format_name, fmt->width, fmt->height, "IMAGE");
+
         break;
     }
 #ifdef CAMU_HAVE_FFMPEG
@@ -71,17 +73,19 @@ bool camu_video_buffer_configure(struct camu_video_buffer *buf, struct camu_code
     }
 #ifdef CAMU_VIDEO_BUFFER_FORCE_SCALER
 #ifdef CAMU_HAVE_FFMPEG
-    if (fmt->format != CAMU_PIXEL_FORMAT_RGBA && fmt->format != CAMU_PIXEL_FORMAT_RGB) {
+    if (fmt->width > 0 && fmt->height > 0 && fmt->format != CAMU_PIXEL_FORMAT_RGBA &&
+        fmt->format != CAMU_PIXEL_FORMAT_RGB) {
         buf->fmt.scaler_needed = true;
         buf->fmt.req.width = fmt->width;
         buf->fmt.req.height = fmt->height;
         buf->fmt.req.format = CAMU_PIXEL_FORMAT_RGBA;
-        buf->scale = camu_ff_scaler_create();
-        if (buf->fmt.scaler_needed && buf->scale->init(buf->scale, &buf->fmt)) {
+        buf->scaler = camu_ff_scaler_create();
+        if (buf->scaler->init(buf->scaler, &buf->fmt)) {
             const char *format_name = av_get_pix_fmt_name(buf->fmt.req.format);
             al_log_info("video_buffer", "Scaling to: %s (%dx%d).",
                 format_name, buf->fmt.req.width, buf->fmt.req.height);
         } else {
+            // Scaler will be freed in video_buffer_free().
             return false;
         }
     } else {
@@ -97,6 +101,14 @@ bool camu_video_buffer_configure(struct camu_video_buffer *buf, struct camu_code
     return true;
 }
 
+#ifdef CAMU_HAVE_FFMPEG
+bool camu_video_buffer_configure_subtitles(struct camu_video_buffer *buf, AVCodecParameters *codecpar)
+{
+    struct camu_video_format *fmt = &buf->stream->video.fmt;
+    return buf->queue->configure_subtitles(buf->queue, fmt->width, fmt->height, codecpar);
+}
+#endif
+
 void camu_video_buffer_set_latency(struct camu_video_buffer *buf, s32 frames)
 {
     buf->latency = frames * buf->avg_frame_duration;
@@ -122,35 +134,31 @@ static void after_push_internal(struct camu_video_buffer *buf)
     else if (have >= BUFFER_MARK_HIGH) buf->callback(buf->userdata, CAMU_BUFFER_CORK);
 }
 
+static inline bool frame_is_late(struct camu_clock *clock, f64 base, f64 pts, f64 duration)
+{
+    if (base == -1.0) base = camu_clock_get_base_pts(clock);
+    return pts + duration < base;
+}
+
 #ifdef CAMU_HAVE_FFMPEG
 static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame)
 {
-    f64 pts = 0.0;
+    AVStream *stream = buf->stream->av.stream;
+    f64 pts = frame->best_effort_timestamp * av_q2d(stream->time_base);
+    f64 duration = frame->duration * av_q2d(stream->time_base);
     f64 base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
-    if (!buf->single_frame) {
-        AVStream *stream = buf->stream->av.stream;
-        pts = frame->best_effort_timestamp * av_q2d(stream->time_base);
-        // If buf->pts = -1.0, this is the first frame pushed either on
-        // a new stream or after a reset(). Meaning, we can safely set buf->pts
-        // from this thread.
-        if (base == -1.0) {
-            base = camu_clock_get_base_pts(buf->clock);
-            al_atomic_store(f64)(&buf->pts, base, AL_ATOMIC_RELEASE);
-        }
-        if (pts + buf->avg_frame_duration < base) {
-            av_frame_free(&frame);
-            return false;
-        }
-    } else if (base == -1.0) {
-        al_atomic_store(f64)(&buf->pts, 0.0, AL_ATOMIC_RELEASE);
+    if (!buf->single_frame && frame_is_late(buf->clock, base, pts, duration)) {
+        av_frame_free(&frame);
+        return false;
     }
+    if (base == -1.0) al_atomic_store(f64)(&buf->pts, pts, AL_ATOMIC_RELEASE);
 #ifdef CAMU_VIDEO_BUFFER_FORCE_SCALER
     if (buf->fmt.scaler_needed) {
-        if (!buf->scale->scale(buf->scale, (const u8 **)frame->data, frame->linesize)) {
+        if (!buf->scaler->scale(buf->scaler, (const u8 **)frame->data, frame->linesize)) {
             return false;
         }
         av_frame_free(&frame);
-        frame = av_frame_clone(buf->scale->get_frame(buf->scale)->av.frame);
+        frame = av_frame_clone(buf->scaler->get_frame(buf->scaler)->av.frame);
     }
 #endif
     buf->queue->push_av_frame(buf->queue, frame, pts);
@@ -187,12 +195,6 @@ void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_fra
 }
 
 #ifdef CAMU_HAVE_FFMPEG
-bool camu_video_buffer_configure_subtitles(struct camu_video_buffer *buf, AVCodecParameters *codecpar)
-{
-    struct camu_video_format *fmt = &buf->stream->video.fmt;
-    return buf->queue->configure_subtitles(buf->queue, fmt->width, fmt->height, codecpar);
-}
-
 void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, AVPacket *pkt)
 {
     buf->queue->push_subtitle(buf->queue, pkt);
@@ -224,9 +226,8 @@ void camu_video_buffer_reset(struct camu_video_buffer *buf)
 
 bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out)
 {
-    f64 base = 0.0;
+    f64 base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
     if (!buf->single_frame) {
-        base = al_atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
         f64 pts = camu_clock_get_pts(buf->clock, buf->latency);
         if (pts > base) {
             base = pts;
@@ -251,7 +252,7 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out)
 void camu_video_buffer_free(struct camu_video_buffer *buf)
 {
 #ifdef CAMU_VIDEO_BUFFER_FORCE_SCALER
-    if (buf->fmt.scaler_needed) buf->scale->free(&buf->scale);
+    if (buf->fmt.scaler_needed) buf->scaler->free(&buf->scaler);
 #endif
     if (buf->queue) buf->queue->free(&buf->queue);
 }
diff --git a/src/buffer/video.h b/src/buffer/video.h
index 3953fc1..f6f4e28 100644
--- a/src/buffer/video.h
+++ b/src/buffer/video.h
@@ -23,7 +23,7 @@ struct camu_video_buffer {
 
     struct camu_scaler_format fmt;
 #ifdef CAMU_VIDEO_BUFFER_FORCE_SCALER
-    struct camu_scaler *scale;
+    struct camu_scaler *scaler;
 #endif
 
     struct camu_frame_queue *queue;
@@ -44,11 +44,13 @@ struct camu_video_buffer {
 bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *clock);
 bool camu_video_buffer_configure(struct camu_video_buffer *buf, struct camu_codec_stream *stream,
     struct camu_renderer *renderer);
+#ifdef CAMU_HAVE_FFMPEG
+bool camu_video_buffer_configure_subtitles(struct camu_video_buffer *buf, AVCodecParameters *codecpar);
+#endif
 void camu_video_buffer_set_latency(struct camu_video_buffer *buf, s32 frames);
 bool camu_video_buffer_is_single_frame(struct camu_video_buffer *buf);
 void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_frame *frame);
 #ifdef CAMU_HAVE_FFMPEG
-bool camu_video_buffer_configure_subtitles(struct camu_video_buffer *buf, AVCodecParameters *codecpar);
 void camu_video_buffer_push_subtitle(struct camu_video_buffer *buf, AVPacket *pkt);
 #endif
 void camu_video_buffer_flush(struct camu_video_buffer *buf);
diff --git a/src/codec/codec.h b/src/codec/codec.h
index 4b93673..a9b188a 100644
--- a/src/codec/codec.h
+++ b/src/codec/codec.h
@@ -53,7 +53,7 @@ enum {
     CAMU_PIXEL_FORMAT_GREYA = AV_PIX_FMT_GRAY8A,
     CAMU_PIXEL_FORMAT_GREY = AV_PIX_FMT_GRAY8
 };
-#define CAMU_NUM_DATA_POINTERS AV_NUM_DATA_POINTERS
+#define CAMU_PLANAR_DATA_POINTERS AV_NUM_DATA_POINTERS
 #else
 enum {
     CAMU_OK = 0,
@@ -92,7 +92,7 @@ enum {
     CAMU_PIXEL_FORMAT_GREYA,
     CAMU_PIXEL_FORMAT_GREY
 };
-#define CAMU_NUM_DATA_POINTERS 8
+#define CAMU_PLANAR_DATA_POINTERS 8
 #endif
 
 enum {
diff --git a/src/codec/ffmpeg/decoder.c b/src/codec/ffmpeg/decoder.c
index 0fb60ac..51eca40 100644
--- a/src/codec/ffmpeg/decoder.c
+++ b/src/codec/ffmpeg/decoder.c
@@ -37,10 +37,13 @@ static bool ff_decoder_init(struct camu_decoder *dec, struct camu_renderer *rend
     }
 
     s32 cpus = av_cpu_count();
-    if (cpus >= 8) cpus = 4;
-    av->codec_context->thread_count = cpus / 2;
+    if (av->codec_context->codec_type == AVMEDIA_TYPE_VIDEO) {
+        cpus = MIN(4, MAX(1, cpus / 2));
+    } else {
+        cpus = MIN(2, MAX(1, cpus / 4));
+    }
+    av->codec_context->thread_count = cpus;
     // FF_THREAD_FRAME or FF_THREAD_SLICE.
-    //av->codec_context->thread_type = FF_THREAD_SLICE;
     av->codec_context->thread_type = FF_THREAD_FRAME;
     al_log_debug("ff_decoder", "Using %i threads for decoder.", cpus);
 
diff --git a/src/codec/ffmpeg/scaler.c b/src/codec/ffmpeg/scaler.c
index 0089dd1..bc8d41b 100644
--- a/src/codec/ffmpeg/scaler.c
+++ b/src/codec/ffmpeg/scaler.c
@@ -49,12 +49,12 @@ static bool ff_scaler_init(struct camu_scaler *scale, struct camu_scaler_format
         return false;
     }
 
-    AVFrame *frame = alloc_picture(fmt->req.format, fmt->req.width, fmt->req.height);
-    if (!frame) {
+    struct camu_codec_frame *frame = &av->frame;
+    frame->av.frame = alloc_picture(fmt->req.format, fmt->req.width, fmt->req.height);
+    if (!frame->av.frame) {
         al_log_error("ff_scaler", "Failed to allocate frame.");
         return false;
     }
-    av->frame.av.frame = frame;
 
     return true;
 }
@@ -81,9 +81,15 @@ static struct camu_codec_frame *ff_scaler_get_frame(struct camu_scaler *scale)
 static void ff_scaler_free(struct camu_scaler **scale)
 {
     struct camu_ff_scaler *av = (struct camu_ff_scaler *)*scale;
-    av_freep(&av->frame.av.frame->data[0]);
-    av_frame_free(&av->frame.av.frame);
-    sws_freeContext(av->scaler_context);
+    struct camu_codec_frame *frame = &av->frame;
+    if (frame->av.frame) {
+        av_freep(&frame->av.frame->data[0]);
+        av_frame_free(&frame->av.frame);
+    }
+    if (av->scaler_context) {
+        sws_freeContext(av->scaler_context);
+        av->scaler_context = NULL;
+    }
     al_free(av);
     *scale = NULL;
 }
diff --git a/src/liana/handlers/codec_client.c b/src/liana/handlers/codec_client.c
index 7c57d03..f15b3f8 100644
--- a/src/liana/handlers/codec_client.c
+++ b/src/liana/handlers/codec_client.c
@@ -55,50 +55,68 @@ static bool push_packet(struct lia_codec_client *codec, struct aki_buffer *buffe
 static bool codec_client_handle_packet(struct lia_client_handler *handler, struct aki_packet *packet)
 {
     struct lia_codec_client *codec = (struct lia_codec_client *)handler;
+
+    // NULL packet = flush.
     if (!packet) {
-        if (!codec->dec) return true;
-        s32 ret = codec->dec->push(codec->dec, NULL);
-        // Flush returns success.
-        ret = codec->dec->process(codec->dec);
-        // process() could still error.
-        codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_EOF, codec->handler.stream, NULL);
-        return ret == CAMU_ERR_EOF;
+        if (codec->dec) {
+            s32 ret = codec->dec->push(codec->dec, NULL);
+            // Flush always returns success.
+            ret = codec->dec->process(codec->dec);
+            // process() could still error.
+            codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_EOF, codec->handler.stream, NULL);
+            return ret == CAMU_ERR_EOF;
+        } else {
+            return true;
+        }
     }
-    bool success = false;
+
+    // Push packet.
+    bool success;
     u8 type = aki_packet_read_u8(packet);
     switch (type) {
     case CAMU_NORMAL: {
-        if (!codec->dec) return true;
-        struct aki_buffer buffer;
-        aki_packet_read_buffer(packet, &buffer);
-        success = push_packet(codec, &buffer);
+        if (codec->dec) {
+            struct aki_buffer buffer;
+            aki_packet_read_buffer(packet, &buffer);
+            success = push_packet(codec, &buffer);
+        } else {
+            success = true;
+        }
         break;
     }
 #ifdef CAMU_HAVE_FFMPEG
     case CAMU_FFMPEG_COMPAT: {
         AVPacket *pkt = aki_packet_read_av_packet(packet);
-        if (!codec->dec) {
-            codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_SUBTITLE, codec->handler.stream, pkt);
-            return true;
-        } else {
+        if (codec->dec) {
             success = push_av_packet(codec, pkt);
+        } else {
+            codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_SUBTITLE, codec->handler.stream, pkt);
+            success = true;
         }
         av_packet_unref(pkt);
         av_packet_free(&pkt);
         break;
     }
+    default:
+        al_assert_and_return(false);
 #endif
     }
-    // Forcing in EOF on an error is not necessary but should be a better experience client-side.
+
     if (!success) {
+        // Forcing in EOF on an error is not necessary but should exhibit less erratic behavior sink-side.
         codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_EOF, codec->handler.stream, NULL);
         return false;
     }
-    s32 ret = codec->dec->process(codec->dec);
-    if (!(ret == CAMU_ERR_AGAIN || ret == CAMU_ERR_EOF)) {
-        codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_EOF, codec->handler.stream, NULL);
-        return false;
+
+    // Process, if needed.
+    if (codec->dec) {
+        s32 ret = codec->dec->process(codec->dec);
+        if (!(ret == CAMU_ERR_AGAIN || ret == CAMU_ERR_EOF)) {
+            codec->handler.callback(codec->handler.userdata, LIANA_CLIENT_EOF, codec->handler.stream, NULL);
+            return false;
+        }
     }
+
     return true;
 }
 
diff --git a/src/render/queue_libplacebo.c b/src/render/queue_libplacebo.c
index b830bb2..8ecf9ce 100644
--- a/src/render/queue_libplacebo.c
+++ b/src/render/queue_libplacebo.c
@@ -188,7 +188,7 @@ static bool map_av_frame(pl_gpu gpu, pl_tex *tex, const struct pl_source_frame *
 
     ((struct pl_source_frame *)src)->frame_data = NULL;
 
-    if (lq->has_subtitles) {
+    if (lq->have_subtitles) {
         s64 now = av_rescale_q(frame->best_effort_timestamp, stream->time_base, (AVRational){ 1, 1000 });
         s32 change;
         ASS_Image *ass_frame = ass_render_frame(lq->ass_renderer, lq->ass_track, now, &change);
@@ -257,10 +257,11 @@ static void discard_av_frame(const struct pl_source_frame *src)
 static void queue_lp_push_av_frame(struct camu_frame_queue *queue, AVFrame *frame, f64 pts)
 {
     struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue;
+    AVStream *stream = lq->q.buf->stream->av.stream;
     frame->opaque = lq;
     pl_queue_push(lq->queue, &(struct pl_source_frame){
         .pts = pts,
-        .duration = lq->q.buf->avg_frame_duration,
+        .duration = frame->duration * av_q2d(stream->time_base),
         .map = map_av_frame,
         .unmap = unmap_av_frame,
         .discard = discard_av_frame,
@@ -288,7 +289,7 @@ static bool queue_lp_configure_subtitiles(struct camu_frame_queue *queue, s32 wi
     ass_set_fonts(lq->ass_renderer, NULL, NULL, ASS_FONTPROVIDER_AUTODETECT, NULL, 0);
     ass_process_codec_private(lq->ass_track, (const char *)codecpar->extradata,
         codecpar->extradata_size);
-    lq->has_subtitles = true;
+    lq->have_subtitles = true;
     return true;
 }
 
@@ -341,7 +342,7 @@ static void queue_lp_reset(struct camu_frame_queue *queue)
 static void queue_lp_free(struct camu_frame_queue **queue)
 {
     struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)*queue;
-    if (lq->has_subtitles) {
+    if (lq->have_subtitles) {
         ass_free_track(lq->ass_track);
         ass_renderer_done(lq->ass_renderer);
     }
@@ -354,7 +355,7 @@ struct camu_frame_queue *camu_frame_queue_lp_create(void)
 {
     struct camu_frame_queue_lp *lq = al_alloc_object(struct camu_frame_queue_lp);
     al_array_init(lq->overlays);
-    lq->has_subtitles = false;
+    lq->have_subtitles = false;
     lq->q.push = queue_lp_push;
 #ifdef CAMU_HAVE_FFMPEG
     lq->q.push_av_frame = queue_lp_push_av_frame;
diff --git a/src/render/queue_libplacebo.h b/src/render/queue_libplacebo.h
index 4879cdd..4f5f644 100644
--- a/src/render/queue_libplacebo.h
+++ b/src/render/queue_libplacebo.h
@@ -22,7 +22,7 @@ struct camu_frame_queue_lp {
     ASS_Library *ass;
     ASS_Renderer *ass_renderer;
     ASS_Track *ass_track;
-    bool has_subtitles;
+    bool have_subtitles;
 };
 
 struct camu_frame_queue *camu_frame_queue_lp_create(void);