Allow count to be more correct in frame_queue API

queue_libplacebo's behavior is currently unchanged, though.

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

4 files changed, 40 insertions, 43 deletions

diff --git a/src/buffer/frame_queue.h b/src/buffer/frame_queue.h
index ca47457..6626b95 100644
--- a/src/buffer/frame_queue.h
+++ b/src/buffer/frame_queue.h
@@ -17,14 +17,13 @@ enum {
 struct camu_frame_queue {
     struct camu_video_buffer *buf;
     bool (*configure_subtitles)(struct camu_frame_queue *, u32, u32, struct camu_codec_stream *);
-    void (*push)(struct camu_frame_queue *, struct camu_codec_frame *, f64);
+    void (*push)(struct camu_frame_queue *, struct camu_codec_frame *, f64, u32 *);
 #ifdef CAMU_HAVE_FFMPEG
-    void (*push_av_frame)(struct camu_frame_queue *, AVFrame *, f64);
+    void (*push_av_frame)(struct camu_frame_queue *, AVFrame *, f64, u32 *);
 #endif
     void (*push_subtitle)(struct camu_frame_queue *, struct camu_codec_packet *);
-    void (*flush)(struct camu_frame_queue *);
-    s32 (*count)(struct camu_frame_queue *);
-    u8 (*read)(struct camu_frame_queue *, f64, void *);
+    void (*flush)(struct camu_frame_queue *, u32 *);
+    u8 (*read)(struct camu_frame_queue *, f64, void *, u32 *);
     void (*reset)(struct camu_frame_queue *);
     void (*free)(struct camu_frame_queue **);
 };
diff --git a/src/buffer/video.c b/src/buffer/video.c
index fac08b9..1c37403 100644
--- a/src/buffer/video.c
+++ b/src/buffer/video.c
@@ -22,9 +22,11 @@
 #define BUFFER_MARK_RESET (BUFFER_MARK_HIGH * 2)
 #else
 #define BUFFER_MARK_LOW 4
-#define BUFFER_MARK_BUFFERED 6 // Must be >1.
-#define BUFFER_MARK_HIGH 8
-#define BUFFER_MARK_RESET (BUFFER_MARK_HIGH * 2)
+#define BUFFER_MARK_BUFFERED 7 // Must be >1.
+#define BUFFER_MARK_HIGH 9
+// Keep MARK_RESET high because currently the decoder can produce an
+// unpredictable amount of frames at any point.
+#define BUFFER_MARK_RESET (BUFFER_MARK_HIGH * 4)
 #endif
 
 bool camu_video_buffer_init(struct camu_video_buffer *buf, struct camu_clock *clock)
@@ -125,7 +127,7 @@ void camu_video_buffer_set_latency(struct camu_video_buffer *buf, f64 latency)
 }
 
 #ifdef CAMU_HAVE_FFMPEG
-static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame)
+static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame, u32 *count)
 {
     AVStream *stream = buf->stream->av.stream;
     f64 pts = frame->best_effort_timestamp * av_q2d(stream->time_base);
@@ -143,7 +145,7 @@ static bool push_av_frame_internal(struct camu_video_buffer *buf, AVFrame *frame
         av_frame_free(&frame);
         frame = av_frame_clone(buf->scaler->get_frame(buf->scaler)->av.frame);
     }
-    buf->queue->push_av_frame(buf->queue, frame, pts);
+    buf->queue->push_av_frame(buf->queue, frame, pts, count);
     return true;
 }
 #endif
@@ -160,18 +162,19 @@ void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_fra
         return;
     }
 
+    u32 count;
     switch (frame->mode) {
     case CAMU_NORMAL: {
         f64 base_pts = atomic_load(f64)(&buf->pts, AL_ATOMIC_ACQUIRE);
         if (base_pts == -1.0) {
             atomic_store(f64)(&buf->pts, 0.0, AL_ATOMIC_RELEASE);
         }
-        buf->queue->push(buf->queue, frame, 0.0);
+        buf->queue->push(buf->queue, frame, 0.0, &count);
         break;
     }
 #ifdef CAMU_HAVE_FFMPEG
     case CAMU_FFMPEG_COMPAT: {
-        if (!push_av_frame_internal(buf, frame->av.frame)) {
+        if (!push_av_frame_internal(buf, frame->av.frame, &count)) {
             camu_codec_frame_discard(frame);
             return;
         }
@@ -181,12 +184,11 @@ void camu_video_buffer_push(struct camu_video_buffer *buf, struct camu_codec_fra
 #endif
     }
 
-    s32 count = buf->queue->count(buf->queue);
     if (!buf->buffered && (buf->is_static || count >= BUFFER_MARK_BUFFERED)) {
         // Preserve order of: set flow -> flush -> callback, for static buffers.
         if (buf->is_static) {
             atomic_store(u32)(&buf->flow, FLUSHED, AL_ATOMIC_RELEASE);
-            buf->queue->flush(buf->queue);
+            buf->queue->flush(buf->queue, &count);
         }
         buf->buffered = true;
         buf->buffered_with_one_frame = count == 1;
@@ -212,9 +214,9 @@ void camu_video_buffer_flush(struct camu_video_buffer *buf, bool error)
     log_debug("Flush requested.");
     u8 flow = error ? FLUSHED_ERROR : FLUSHED;
     atomic_store(u32)(&buf->flow, flow, AL_ATOMIC_RELAXED);
-    buf->queue->flush(buf->queue);
+    u32 count;
+    buf->queue->flush(buf->queue, &count);
     if (!buf->buffered) {
-        s32 count = buf->queue->count(buf->queue);
         buf->buffered = true;
         buf->buffered_with_one_frame = count == 1;
         log_debug("Buffered (mark: %.2fs).", count * buf->avg_frame_duration);
@@ -251,7 +253,9 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weig
 
     u8 flow = atomic_load(u32)(&buf->flow, AL_ATOMIC_ACQUIRE);
     if (flow == ERRORED) return false;
-    u8 ret = buf->queue->read(buf->queue, base_pts, out);
+
+    u32 count;
+    u8 ret = buf->queue->read(buf->queue, base_pts, out, &count);
     if (ret == CAMU_QUEUE_ERR || flow == FLUSHED_ERROR) {
         buf->callback(buf->userdata, CAMU_BUFFER_ERRORED);
         atomic_store(u32)(&buf->flow, ERRORED, AL_ATOMIC_RELEASE);
@@ -274,7 +278,6 @@ bool camu_video_buffer_read(struct camu_video_buffer *buf, void *out, bool *weig
         atomic_store(u32)(&buf->flow, SIGNALED, AL_ATOMIC_RELEASE);
     } else if (flow == FLOWING) {
         // Never attempt to uncork if flow = FLUSHED.
-        s32 count = buf->queue->count(buf->queue);
         if (count <= BUFFER_MARK_LOW) {
             buf->callback(buf->userdata, CAMU_BUFFER_UNCORK);
         } else if (count >= BUFFER_MARK_RESET) {
diff --git a/src/render/queue_libplacebo.c b/src/render/queue_libplacebo.c
index 408af51..f2a2898 100644
--- a/src/render/queue_libplacebo.c
+++ b/src/render/queue_libplacebo.c
@@ -157,7 +157,7 @@ static void discard_frame(const struct pl_source_frame *src)
     camu_codec_frame_discard(frame);
 }
 
-static void queue_lp_push(struct camu_frame_queue *queue, struct camu_codec_frame *frame, f64 pts)
+static void queue_lp_push(struct camu_frame_queue *queue, struct camu_codec_frame *frame, f64 pts, u32 *count)
 {
     struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue;
     pl_queue_push(lq->queue, &(struct pl_source_frame){
@@ -168,6 +168,7 @@ static void queue_lp_push(struct camu_frame_queue *queue, struct camu_codec_fram
         .discard = discard_frame,
         .frame_data = frame
     });
+    *count = (u32)pl_queue_num_frames(lq->queue);
 }
 
 #ifdef CAMU_HAVE_FFMPEG
@@ -391,7 +392,7 @@ static void discard_av_frame(const struct pl_source_frame *src)
     log_warn("Dropped frame with PTS %.3f.", src->pts);
 }
 
-static void queue_lp_push_av_frame(struct camu_frame_queue *queue, AVFrame *frame, f64 pts)
+static void queue_lp_push_av_frame(struct camu_frame_queue *queue, AVFrame *frame, f64 pts, u32 *count)
 {
     struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue;
     AVStream *stream = lq->q.buf->stream->av.stream;
@@ -405,6 +406,7 @@ static void queue_lp_push_av_frame(struct camu_frame_queue *queue, AVFrame *fram
         .frame_data = frame,
         .first_field = pl_field_from_avframe(frame)
     });
+    *count = (u32)pl_queue_num_frames(lq->queue);
 }
 #endif
 
@@ -448,9 +450,10 @@ static void queue_lp_push_subtitle(struct camu_frame_queue *queue, struct camu_c
 #endif
 }
 
-static void queue_lp_flush(struct camu_frame_queue *queue)
+static void queue_lp_flush(struct camu_frame_queue *queue, u32 *count)
 {
     struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue;
+    *count = (u32)pl_queue_num_frames(lq->queue);
     pl_queue_push(lq->queue, NULL);
 #ifdef CAMU_HAVE_SUBTITLES
     if (lq->subs.are_present) {
@@ -464,18 +467,14 @@ static void queue_lp_flush(struct camu_frame_queue *queue)
 #endif
 }
 
-static s32 queue_lp_count(struct camu_frame_queue *queue)
-{
-    struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue;
-    return pl_queue_num_frames(lq->queue);
-}
-
-static u8 queue_lp_read(struct camu_frame_queue *queue, f64 pts, void *out)
+static u8 queue_lp_read(struct camu_frame_queue *queue, f64 pts, void *out, u32 *count)
 {
     struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue;
     struct pl_frame_mix *mix = (struct pl_frame_mix *)out;
     lq->params.pts = pts;
-    switch (pl_queue_update(lq->queue, mix, &lq->params)) {
+    enum pl_queue_status ret = pl_queue_update(lq->queue, mix, &lq->params);
+    *count = (u32)pl_queue_num_frames(lq->queue);
+    switch (ret) {
     case PL_QUEUE_OK:
         break;
     case PL_QUEUE_MORE:
@@ -492,7 +491,8 @@ static u8 queue_lp_read(struct camu_frame_queue *queue, f64 pts, void *out)
 static void queue_lp_reset(struct camu_frame_queue *queue)
 {
     struct camu_frame_queue_lp *lq = (struct camu_frame_queue_lp *)queue;
-    queue_lp_flush(&lq->q);
+    u32 count;
+    queue_lp_flush(&lq->q, &count);
     pl_queue_reset(lq->queue);
 }
 
@@ -543,7 +543,6 @@ struct camu_frame_queue *camu_frame_queue_lp_create(void)
 #endif
     lq->q.push_subtitle = queue_lp_push_subtitle;
     lq->q.flush = queue_lp_flush;
-    lq->q.count = queue_lp_count;
     lq->q.read = queue_lp_read;
     lq->q.reset = queue_lp_reset;
     lq->q.free = queue_lp_free;
diff --git a/src/render/queue_momo.c b/src/render/queue_momo.c
index bc7caf8..2f7d535 100644
--- a/src/render/queue_momo.c
+++ b/src/render/queue_momo.c
@@ -11,20 +11,22 @@ static bool queue_momo_configure_subtitles(struct camu_frame_queue *queue, u32 w
     return false;
 }
 
-static void queue_momo_push(struct camu_frame_queue *queue, struct camu_codec_frame *frame, f64 pts)
+static void queue_momo_push(struct camu_frame_queue *queue, struct camu_codec_frame *frame, f64 pts, u32 *count)
 {
     struct camu_frame_queue_momo *mq = (struct camu_frame_queue_momo *)queue;
     (void)mq;
     (void)pts;
+    (void)count;
     camu_codec_frame_discard(frame);
 }
 
 #ifdef CAMU_HAVE_FFMPEG
-static void queue_momo_push_av_frame(struct camu_frame_queue *queue, AVFrame *frame, f64 pts)
+static void queue_momo_push_av_frame(struct camu_frame_queue *queue, AVFrame *frame, f64 pts, u32 *count)
 {
     struct camu_frame_queue_momo *mq = (struct camu_frame_queue_momo *)queue;
     (void)mq;
     (void)pts;
+    (void)count;
     av_frame_free(&frame);
 }
 #endif
@@ -36,24 +38,19 @@ static void queue_momo_push_subtitle(struct camu_frame_queue *queue, struct camu
     (void)packet;
 }
 
-static void queue_momo_flush(struct camu_frame_queue *queue)
+static void queue_momo_flush(struct camu_frame_queue *queue, u32 *count)
 {
     (void)queue;
+    (void)count;
 }
 
-static s32 queue_momo_count(struct camu_frame_queue *queue)
-{
-    struct camu_frame_queue_momo *mq = (struct camu_frame_queue_momo *)queue;
-    (void)mq;
-    return 1;
-}
-
-static u8 queue_momo_read(struct camu_frame_queue *queue, f64 pts, void *out)
+static u8 queue_momo_read(struct camu_frame_queue *queue, f64 pts, void *out, u32 *count)
 {
     struct camu_frame_queue_momo *mq = (struct camu_frame_queue_momo *)queue;
     (void)mq;
     (void)pts;
     (void)out;
+    (void)count;
     return CAMU_QUEUE_OK;
 }
 
@@ -80,7 +77,6 @@ struct camu_frame_queue *camu_frame_queue_momo_create(void)
 #endif
     mq->q.push_subtitle = queue_momo_push_subtitle;
     mq->q.flush = queue_momo_flush;
-    mq->q.count = queue_momo_count;
     mq->q.read = queue_momo_read;
     mq->q.reset = queue_momo_reset;
     mq->q.free = queue_momo_free;