OAK-D PoE frame rate slows down over time when using queue callbacks

I'm writing a C++ module to interface with OAK-D PoE cameras. One of my requirements is to retrieve and display RGB video at 1080p resolution and 30 FPS. I had the idea of using VideoEncoder to encode frames for faster transfer, then decode them in the host for display.

This works fine so long as I am polling the data output queue using get(). If however I use a callback, the actual frame rate starts at 30 but then after some time it starts steadily slowing down until stabilizing around 9 FPS. What's weirder is that it seems the slow down is triggered by changes in camera input — I had several cases where the FPS rate remained at 30 until I made a sudden movement, and then quickly dropped to about 9 FPS.

Any ideas about this?

See below a simple program I wrote to demonstrate the issue.

#include <chrono>
#include <iostream>
#include <queue>
#include <set>

#include "depthai/depthai.hpp"

class CameraRGB {
  std::shared_ptr<dai::Device> device_;

  dai::Pipeline pipeline_;

  std::shared_ptr<dai::node::ColorCamera> camera_;

  std::shared_ptr<dai::node::VideoEncoder> encoder_;

  std::shared_ptr<dai::node::XLinkOut> output_;

  std::shared_ptr<dai::DataOutputQueue> queue_;

public:
  CameraRGB(bool encoded) {
    camera_ = pipeline_.create<dai::node::ColorCamera>();
    float fps = camera_->getFps();

    std::cerr << "[INFO] Camera FPS rate: " << fps << std::endl;

    output_ = pipeline_.create<dai::node::XLinkOut>();
    output_->setStreamName("rgb");

    if (encoded) {
      encoder_ = pipeline_.create<dai::node::VideoEncoder>();
      encoder_->setDefaultProfilePreset(fps, dai::VideoEncoderProperties::Profile::MJPEG);

      camera_->video.link(encoder_->input);
      encoder_->bitstream.link(output_->input);
    }
    else {
      camera_->video.link(output_->input);
    }

    device_ = std::make_shared<dai::Device>(pipeline_);
    queue_ = device_->getOutputQueue("rgb", 1, false);
  }

  cv::Mat poll() {
    auto frame = queue_->get<dai::ImgFrame>();
    return getCvFrame(*frame);
  }

  cv::Mat getCvFrame(dai::ImgFrame& frame) {
    if (encoder_) {
      return cv::imdecode(frame.getData(), cv::IMREAD_UNCHANGED);
    }
    else {
      return frame.getCvFrame();
    }
  }

  std::shared_ptr<dai::DataOutputQueue> getQueue() {
    return queue_;
  }
};

void poll(CameraRGB& camera) {
  auto last = std::chrono::steady_clock::now();
  int frame_count = 0;

  while (true) {
    auto image = camera.poll();

    frame_count += 1;

    auto now = std::chrono::steady_clock::now();
    std::chrono::duration<double> dt = now - last;
    double dt_seconds = dt.count();
    if (dt_seconds >= 1.0) {
      std::cerr << "[INFO] Frame rate: " << (frame_count / dt_seconds) << std::endl;
      frame_count = 0;
      last = now;
    }

    cv::imshow("Image", image);

    int key = cv::waitKey(1);
    if(key == 'q' || key == 'Q') {
      return;
    }
  }
}

void callback(CameraRGB& camera) {
  auto last = std::chrono::steady_clock::now();
  int frame_count = 0;

  auto queue = std::queue<cv::Mat>();
  std::mutex queueMtx;

  auto showFrame = [&](std::shared_ptr<dai::ADatatype> callback) {
    auto* frame = dynamic_cast<dai::ImgFrame*>(callback.get());
    if (frame == nullptr) {
      std::cerr << "[ERROR] Received message is not a message frame" << std::endl;
      return;
    }

    {
      std::unique_lock<std::mutex> lock(queueMtx);
      auto image = camera.getCvFrame(*frame);
      queue.push(image);
    }

    frame_count += 1;

    auto now = std::chrono::steady_clock::now();
    std::chrono::duration<double> dt = now - last;
    double dt_seconds = dt.count();
    if (dt_seconds >= 1.0) {
      std::cerr << "[INFO] Frame rate: " << (frame_count / dt_seconds) << std::endl;
      frame_count = 0;
      last = now;
    }
  };

  camera.getQueue()->addCallback(showFrame);

  while (true) {
    int key = cv::waitKey(1);
    if(key == 'q' || key == 'Q') {
      return;
    }

    {
      std::unique_lock<std::mutex> lock(queueMtx);
      if (queue.empty()) {
        continue;
      }
    }

    auto image = queue.front();
    queue.pop();

    cv::imshow("Image", image);
  }
}

int main(int argc, char** argv) {
  std::set<std::string> options;
  for (int i = 1; i < argc; ++i) {
    options.insert(argv[i]);
  }

  bool encoded = (options.count("encoded") > 0);
  bool polling = (options.count("polling") > 0);

  CameraRGB camera(encoded);

  if (polling) {
    poll(camera);
  }
  else {
    callback(camera);
  }

  return 0;
}