PacedVideoOutput.h

Go to the documentation of this file.

#pragma once
#include <atomic>
#include <string.h>
#include "AudioTools/CoreAudio/AudioBasic/Collections/Vector.h"
#include "AudioTools/CoreAudio/AudioTypes.h"
#include "AudioTools/Concurrency/LockFree/RingBufferSPSC.h"
#include "AudioTools/Video/Video.h"
 
#ifdef __linux__
#include "AudioTools/Concurrency/Desktop/Task.h"
#else
#include "AudioTools/Concurrency/RTOS/Task.h"
#endif
 
namespace audio_tools {
 
class PacedVideoOutput : public VideoOutput {
 public:
  PacedVideoOutput(VideoOutput& target, float fps = 0,
                      uint32_t schedulingDelayMs = 0)
      : p_target(&target), scheduling_delay_ms(schedulingDelayMs) {
    setFps(fps);
  }
 
  void setFps(float fps) { frame_period_ms = fps > 0 ? 1000.0f / fps : 0; }
 
  void setAudioClock(TimeSource& clock) { p_clock = &clock; }
 
  void setSchedulingDelayMs(uint32_t delayMs) { scheduling_delay_ms = delayMs; }
 
  void setCatchUpThresholdFrames(float frames) {
    catch_up_threshold_frames = frames;
  }
 
  void setIgnorePFrames(bool active) { ignore_p_frames = active; }
  bool ignorePFrames() const { return ignore_p_frames; }
 
  void setResyncThresholdMs(uint32_t ms) { resync_threshold_ms = ms; }
 
  void setResyncQueueFillFraction(float fraction) {
    resync_queue_fill_fraction = fraction;
  }
 
  void setMaxQueuedIFrames(int count) { max_queued_i_frames = count; }
 
  void setTaskParameters(uint32_t stackSizeWords, uint8_t priority,
                         int core = -1) {
    task_stack_size = stackSizeWords;
    task_priority = priority;
    task_core = core;
  }
 
  void setQueueBytes(size_t bytes) { queue_bytes = bytes > 0 ? bytes : 1; }
 
  void setQueueUsePSRAM(bool flag) { queue_use_psram = flag; }
 
  bool begin() {
    frame_index = 0;
    frame_count = 0;
    dropped_frame_count = 0;
    dropped_i_frame_count = 0;
    ignored_frame_count = 0;
    queued_i_frame_count = 0;
    logged_drop_burst = false;
    awaiting_keyframe = false;
    render_lateness_ms = 0;
    i_frame_count = 0;
    p_frame_count = 0;
    i_frame_total_ms = 0;
    p_frame_total_ms = 0;
    start_set = false;
    input_start_set = false;
    output_start_set = false;
    have_header = false;
    frame_buf.clear();
    queue.setUsePSRAM(queue_use_psram);
    if (queue_bytes_allocated != queue_bytes) {
      queue.resize(queue_bytes);
      queue_bytes_allocated = queue_bytes;
    } else {
      queue.reset();
    }
    task.create("PacedVideoOutput", task_stack_size, task_priority,
                task_core);
    bool ok = task.begin([this]() { taskLoop(); });
    task_started = task_started || ok;
    LOGI(
        "PacedVideoOutput: %s (fps=%.2f, clock=%s, scheduling delay=%u ms, "
        "queue=%u bytes, stack=%u words, priority=%u)",
        ok ? "render task started" : "render task failed to start",
        frame_period_ms > 0 ? 1000.0f / frame_period_ms : 0.0f,
        p_clock != nullptr ? "external audio" : "wall",
        (unsigned)scheduling_delay_ms, (unsigned)queue.size(),
        (unsigned)task_stack_size, (unsigned)task_priority);
    return ok;
  }
 
  void end() {
    task.end();
    task_started = false;
    LOGI(
        "PacedVideoOutput: render task stopped (%u frames rendered - %u "
        "I / %u P, %u P dropped / %u I dropped, input=%.2f fps, "
        "output=%.2f fps)",
        (unsigned)frame_count, (unsigned)i_frame_count.load(),
        (unsigned)p_frame_count.load(), (unsigned)dropped_frame_count,
        (unsigned)dropped_i_frame_count, inputFPS(), outputFPS());
  }
 
  size_t write(const uint8_t* data, size_t len) override {
    if (len == 0) return 0;
    if (!task_started) begin();
    if (!input_start_set) {
      input_start_ms = clockMs();
      input_start_set = true;
    }
    uint32_t this_frame = frame_index;
    uint32_t target_ms = (uint32_t)((double)frame_index * frame_period_ms);
    frame_index++;
    bool is_key = isKeyFrame(data, len);
 
    // See setIgnorePFrames() - unconditional, independent of queue/
    // lateness state, and checked before the header/queue-space bookkeeping
    // below since a dropped-here frame never touches any of that. Counts
    // into droppedFrameCount() too (same externally visible effect as any
    // other dropped P-frame - never rendered), with ignoredFrameCount()
    // as the more specific breakdown of how many were dropped for this
    // reason rather than backlog.
    if (ignore_p_frames && !is_key) {
      ignored_frame_count++;
      dropped_frame_count++;
      frame_count++;
      return len;
    }
 
    FrameHeader header;
    header.size = (uint32_t)len;
    header.target_ms = target_ms;
    header.is_key = is_key ? 1 : 0;
    size_t needed = sizeof(header) + len;
 
    // decode+render isn't keeping up right now. A non-keyframe is
    // dropped outright instead of enqueued (see droppedFrameCount()) to
    // catch back up without blocking this call, which would otherwise
    // stall the caller's own thread (typically a demuxer's dispatch
    // loop, delaying audio too). Two triggers: the queue is actually
    // full, or the render task's most recently rendered frame was
    // already more than setCatchUpThresholdFrames() frame periods late.
    // Keyframes are never dropped here (losing one would desync every
    // dependent P-frame) - if the queue can't fit one, this still blocks
    // below.
    bool queue_full = (size_t)queue.availableForWrite() < needed;
    bool falling_behind =
        frame_period_ms > 0 &&
        render_lateness_ms.load() >
            (int32_t)(catch_up_threshold_frames * frame_period_ms);
    if (!is_key && (queue_full || falling_behind)) {
      dropped_frame_count++;
      frame_count++;
      // Rate-limited to one line per unbroken run of drops - this can
      // trigger on every P-frame in a row for a long stretch, and
      // logging each one individually is expensive enough (Serial I/O on
      // this thread) to measurably worsen the backlog it's trying to
      // fix.
      if (!logged_drop_burst) {
        LOGW(
            "PacedVideoOutput: dropping P frames starting at #%u - %s "
            "(%d/%u bytes free, need %u, render %d ms behind)",
            (unsigned)this_frame,
            queue_full ? "render queue full" : "catching up",
            queue.availableForWrite(), (unsigned)queue.size(),
            (unsigned)needed, (int)render_lateness_ms.load());
        logged_drop_burst = true;
      }
      return len;
    }
    logged_drop_burst = false;
 
    bool logged_wait = false;
    while ((size_t)queue.availableForWrite() < needed) {
      if (!logged_wait) {
        LOGW(
            "PacedVideoOutput: write() blocked on frame #%u - render "
            "queue full (%d/%u bytes free, need %u)",
            (unsigned)this_frame, queue.availableForWrite(),
            (unsigned)queue.size(), (unsigned)needed);
        logged_wait = true;
      }
      delay(1);
    }
    queue.writeArray((const uint8_t*)&header, sizeof(header));
    queue.writeArray(data, len);
    if (is_key) queued_i_frame_count++;
    LOGD(
        "PacedVideoOutput: queued frame #%u (%s, %u bytes, target=%u ms, "
        "%d bytes free)",
        (unsigned)this_frame, is_key ? "I" : "P", (unsigned)len,
        (unsigned)target_ms, queue.availableForWrite());
    frame_count++;
    return len;
  }
 
  void flush() override {}
 
  void setSkipRender(bool skip) override { p_target->setSkipRender(skip); }
 
  uint32_t frameCount() const { return frame_count; }
 
  uint32_t droppedFrameCount() const { return dropped_frame_count; }
 
  uint32_t ignoredFrameCount() const { return ignored_frame_count; }
 
  uint32_t droppedIFrameCount() const { return dropped_i_frame_count; }
 
  int queuedIFrameCount() const { return queued_i_frame_count.load(); }
 
  size_t queuedBytes() { return (size_t)queue.available(); }
 
  size_t queueCapacityBytes() { return queue.size(); }
 
  uint32_t frameCountI() const { return i_frame_count; }
  uint32_t frameCountP() const { return p_frame_count; }
 
  float avgIFrameMs() const {
    uint32_t n = i_frame_count;
    return n > 0 ? (float)i_frame_total_ms.load() / n : 0.0f;
  }
  float avgPFrameMs() const {
    uint32_t n = p_frame_count;
    return n > 0 ? (float)p_frame_total_ms.load() / n : 0.0f;
  }
  float avgFrameMs() const {
    uint32_t n = i_frame_count + p_frame_count;
    return n > 0 ? (float)(i_frame_total_ms + p_frame_total_ms) / n : 0.0f;
  }
 
  float inputFPS() {
    if (!input_start_set) return 0.0f;
    uint32_t elapsed = clockMs() - input_start_ms;
    return elapsed > 0 ? (1000.0f * frame_count) / elapsed : 0.0f;
  }
 
  float outputFPS() {
    if (!output_start_set) return 0.0f;
    uint32_t elapsed = millis() - output_start_ms.load();
    uint32_t count = i_frame_count + p_frame_count;
    return elapsed > 0 ? (1000.0f * count) / elapsed : 0.0f;
  }
 
 protected:
  VideoOutput* p_target;
  TimeSource* p_clock = nullptr;
  float frame_period_ms = 0;
  uint32_t scheduling_delay_ms = 0;  // see setSchedulingDelayMs()
  float catch_up_threshold_frames = 1.0f;  // see setCatchUpThresholdFrames()
  bool ignore_p_frames = false;  // see setIgnorePFrames()
  uint32_t resync_threshold_ms = 2000;  // see setResyncThresholdMs()
  float resync_queue_fill_fraction = 0.8f;  // see setResyncQueueFillFraction()
  int max_queued_i_frames = 4;  // see setMaxQueuedIFrames()
  uint64_t frame_index = 0;
  uint32_t frame_count = 0;
  uint32_t dropped_frame_count = 0;  // see droppedFrameCount()
  uint32_t dropped_i_frame_count = 0;  // consumer-thread-only; see droppedIFrameCount()
  uint32_t ignored_frame_count = 0;  // write()-thread-only; see ignoredFrameCount()
  // Keyframes currently queued, not yet consumed - incremented by
  // write(), decremented by taskLoop()/drainQueueKeepingLastKeyframe();
  // atomic since producer and consumer both touch it. See
  // queuedIFrameCount().
  std::atomic<int32_t> queued_i_frame_count{0};
  bool logged_drop_burst = false;  // write()-thread-only; see write()
  uint32_t start_ms = 0;
  bool start_set = false;
  bool task_started = false;
  // True from the moment a resync fires until a real keyframe renders -
  // consumer-thread-only. A resync can leave the decoder's
  // reference-frame state unreliable, so P-frames are discarded until a
  // self-contained keyframe resets it.
  bool awaiting_keyframe = false;
  // How many ms late (or early) the most recently rendered frame was -
  // written by taskLoop(), read by write() (its proactive drop check),
  // hence atomic. A heuristic: stale by one frame's processing time by
  // the time write() reads it, same as output_start_ms below.
  std::atomic<int32_t> render_lateness_ms{0};
 
  struct FrameHeader {
    uint32_t size = 0;
    uint32_t target_ms = 0;
    uint8_t is_key = 0;
  };
  RingBufferSPSC<uint8_t> queue;
  size_t queue_bytes = 32 * 1024;      // desired capacity - see setQueueBytes()
  size_t queue_bytes_allocated = 0;    // what 'queue' was last resize()d to
  bool queue_use_psram = false;        // see setQueueUsePSRAM()
 
  // Consumer-side-only framing state (see taskLoop()): a header already
  // pulled out of 'queue' whose payload isn't fully written yet. Never
  // touched by write()/the caller thread.
  bool have_header = false;
  FrameHeader current_header;
  Vector<uint8_t> frame_buf;
 
  Task task;
  uint32_t task_stack_size = 4096;
  uint8_t task_priority = 2;
  int task_core = -1;
 
  // Per-frame-type render-time stats - written only from taskLoop(),
  // read via the getters above. Plain atomic (single writer; not for
  // contended updates) - the only shared state left outside 'queue'
  // itself.
  std::atomic<uint32_t> i_frame_count{0};
  std::atomic<uint32_t> p_frame_count{0};
  std::atomic<uint64_t> i_frame_total_ms{0};
  std::atomic<uint64_t> p_frame_total_ms{0};
 
  // inputFPS()/outputFPS() anchors, set once and never moved - averaging
  // runs "since the very start" up to now. Deliberately different
  // clocks: input_start_ms is clockMs() (see inputFPS()),
  // output_start_ms is always wall-clock millis() (see outputFPS()).
  // input_* is caller-thread-only; output_* is written by taskLoop(),
  // hence atomic.
  uint32_t input_start_ms = 0;
  bool input_start_set = false;
  std::atomic<uint32_t> output_start_ms{0};
  std::atomic<bool> output_start_set{false};
 
  uint32_t clockMs() { return p_clock != nullptr ? p_clock->playbackTime() : millis(); }
 
  bool isKeyFrame(const uint8_t* data, size_t len) {
    return p_target->isKeyFrame(data, len);
  }
 
  static constexpr int kDrainYieldEvery = 32;
 
  bool drainQueueKeepingLastKeyframe(uint32_t& out_target_ms) {
    bool found_key = false;
    int count = 0;
    while (true) {
      if (!have_header) {
        if ((size_t)queue.available() < sizeof(FrameHeader)) break;
        queue.readArray((uint8_t*)&current_header, sizeof(FrameHeader));
        have_header = true;
      }
      if ((size_t)queue.available() < current_header.size) break;
 
      if (current_header.is_key) {
        if (found_key) dropped_i_frame_count++;  // superseded by this one
        frame_buf.resize(current_header.size);
        queue.readArray(frame_buf.data(), current_header.size);
        out_target_ms = current_header.target_ms;
        found_key = true;
        queued_i_frame_count--;
      } else {
        uint8_t scratch[256];
        size_t remaining = current_header.size;
        while (remaining > 0) {
          size_t chunk = remaining < sizeof(scratch) ? remaining : sizeof(scratch);
          size_t got = (size_t)queue.readArray(scratch, chunk);
          if (got == 0) break;  // shouldn't happen; avoids ever spinning forever
          remaining -= got;
        }
        dropped_frame_count++;
      }
      have_header = false;
      if (++count % kDrainYieldEvery == 0) delay(1);
    }
    return found_key;
  }
 
  void taskLoop() {
    if (!have_header) {
      if ((size_t)queue.available() < sizeof(FrameHeader)) {
        delay(1);
        return;
      }
      queue.readArray((uint8_t*)&current_header, sizeof(FrameHeader));
      have_header = true;
    }
    // The header may already be visible before its payload fully is -
    // write() writes them as two separate writeArray() calls - so just
    // wait for the rest; have_header stays true across calls in the
    // meantime.
    if ((size_t)queue.available() < current_header.size) {
      delay(1);
      return;
    }
    frame_buf.resize(current_header.size);
    queue.readArray(frame_buf.data(), current_header.size);
    uint32_t target_ms = current_header.target_ms;
    bool is_key = current_header.is_key != 0;
    have_header = false;
    // Leaving the queue for rendering (not a discard - see
    // drainQueueKeepingLastKeyframe() for that side of this same counter).
    if (is_key) queued_i_frame_count--;
 
    // Still waiting for a keyframe after a previous resync - skip
    // straight past any P-frame without considering its schedule,
    // reaching the next keyframe as fast as the queue allows.
    if (awaiting_keyframe && !is_key) {
      dropped_frame_count++;
      frame_count++;
      return;
    }
 
    if (!start_set) {
      start_ms = clockMs();
      start_set = true;
      LOGI("PacedVideoOutput: playback anchored at %u ms (%s clock)",
           (unsigned)start_ms, p_clock != nullptr ? "external audio" : "wall");
    }
    // scheduling_delay_ms (see setSchedulingDelayMs()) only ever pushes
    // this later, correcting for the audio output's own buffering
    // latency between "accepted by write()" and "actually audible".
    uint32_t scheduled_ms = start_ms + target_ms + scheduling_delay_ms;
    while ((int32_t)(scheduled_ms - clockMs()) > 0) {
      delay(1);
    }
    // Captured right as the wait loop exits, before write()/flush() - a
    // measure of how far behind schedule this frame already was when we
    // started rendering it, separate from processMs (how long rendering
    // itself then took).
    int32_t lateness_ms = (int32_t)(clockMs() - scheduled_ms);
    // Dropping non-keyframes alone can only prevent the backlog from
    // growing, not shrink it - the producer can't deliver frames faster
    // than real time. Past resync_threshold_ms, give up on the skipped
    // content and jump the schedule anchor forward, making *this* frame
    // "on time" - a visible jump instead of an ever-growing lag.
    //
    // Three independent triggers, since they can diverge: lateness_ms
    // only reflects the frame we're about to render - dropped P-frames
    // can keep that low even while newer bytes keep piling up in the
    // queue, since overall render throughput can still trail the arrival
    // rate. queue fill catches that directly, but only once
    // substantially full; queued keyframe count (setMaxQueuedIFrames())
    // catches it earlier still.
    bool lateness_resync =
        resync_threshold_ms > 0 && lateness_ms > (int32_t)resync_threshold_ms;
    bool queue_resync =
        resync_queue_fill_fraction > 0 && queue.size() > 0 &&
        (float)queue.available() >=
            resync_queue_fill_fraction * (float)queue.size();
    bool iframe_resync = max_queued_i_frames > 0 &&
                          queued_i_frame_count.load() >= max_queued_i_frames;
    if (lateness_resync || queue_resync || iframe_resync) {
      // Keep the freshest keyframe found in the backlog instead of
      // discarding it too (see drainQueueKeepingLastKeyframe()).
      uint32_t new_target_ms = target_ms;
      bool found_fresher_key = drainQueueKeepingLastKeyframe(new_target_ms);
      LOGW(
          "PacedVideoOutput: resyncing (%s) - %s instead of trying to "
          "catch up",
          lateness_resync
              ? "lateness"
              : (queue_resync ? "queue fill" : "I-frame backlog"),
          found_fresher_key ? "jumping to the newest available keyframe"
                             : "jumping the schedule forward");
      if (found_fresher_key) {
        // Supersede whatever was already dequeued.
        if (is_key) {
          dropped_i_frame_count++;
        } else {
          dropped_frame_count++;
        }
        target_ms = new_target_ms;
        is_key = true;
        awaiting_keyframe = false;
      } else {
        // A resync can leave the decoder's reference state unreliable,
        // so require a real (self-contained) keyframe before resuming -
        // discard any P-frame in between (see the check below) - unless
        // the frame we already have is one.
        awaiting_keyframe = true;
      }
      // Recomputed for whichever frame we're actually about to render.
      start_ms = clockMs() - target_ms - scheduling_delay_ms;
      scheduled_ms = start_ms + target_ms + scheduling_delay_ms;  // == clockMs() now
      lateness_ms = 0;
    }
    if (awaiting_keyframe) {
      if (!is_key) {
        dropped_frame_count++;
        frame_count++;
        render_lateness_ms = 0;
        return;
      }
      awaiting_keyframe = false;
    }
    // Published for write()'s proactive drop check (see
    // setCatchUpThresholdFrames()) before doing the possibly-slow render
    // call, so a caller blocked in write() sees it as soon as possible
    // rather than only after this frame finishes too.
    render_lateness_ms = lateness_ms;
    uint32_t processStart = millis();
    p_target->write(frame_buf.data(), frame_buf.size());
    p_target->flush();
    uint32_t processMs = millis() - processStart;
    // See VideoOutput::hadOutput() - true for every synchronous decoder
    // (the common case), but a decoder like MPGDecoder can legitimately
    // do real decode work here without a picture actually reaching the
    // screen yet (held back for B-picture display-order reordering), or
    // emit an earlier held picture instead. Only count/time this call as
    // a rendered frame when a picture was actually produced - otherwise
    // outputFPS()/frameCountI()/frameCountP()/avgFrameMs() would
    // overcount calls that did no real rendering work.
    if (p_target->hadOutput()) {
      if (!output_start_set) {
        output_start_ms = millis();
        output_start_set = true;
      }
      if (is_key) {
        i_frame_count++;
        i_frame_total_ms += processMs;
      } else {
        p_frame_count++;
        p_frame_total_ms += processMs;
      }
      LOGD(
          "PacedVideoOutput: rendered %s frame - scheduled=%u actual=%u "
          "(%d ms late), process=%u ms",
          is_key ? "I" : "P", (unsigned)scheduled_ms, (unsigned)clockMs(),
          (int)lateness_ms, (unsigned)processMs);
    } else {
      LOGD(
          "PacedVideoOutput: decoded %s frame, no picture emitted yet "
          "(scheduled=%u actual=%u, %d ms late), process=%u ms",
          is_key ? "I" : "P", (unsigned)scheduled_ms, (unsigned)clockMs(),
          (int)lateness_ms, (unsigned)processMs);
    }
    if (frame_period_ms > 0 && processMs > (uint32_t)frame_period_ms) {
      LOGW(
          "PacedVideoOutput: %s frame took %u ms to render - longer than "
          "the %.1f ms frame period, falling behind",
          is_key ? "I" : "P", (unsigned)processMs, frame_period_ms);
    }
  }
};
 
}  // namespace audio_tools