src/zm_packetqueue.cpp
Rama master · réplica de lectura del repositorio original.
//ZoneMinder Packet Queue Implementation Class
//Copyright 2016 Steve Gilvarry
//
//This file is part of ZoneMinder.
//
//ZoneMinder is free software: you can redistribute it and/or modify
//it under the terms of the GNU General Public License as published by
//the Free Software Foundation, either version 3 of the License, or
//(at your option) any later version.
//
//ZoneMinder is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
//GNU General Public License for more details.
//
//You should have received a copy of the GNU General Public License
//along with ZoneMinder. If not, see <http://www.gnu.org/licenses/>.
// PacketQueue must know about all iterators and manage them
#include "zm_packetqueue.h"
#include "zm_ffmpeg.h"
#include "zm_packet.h"
#include "zm_signal.h"
#include <algorithm>
#include <climits>
#include <vector>
PacketQueue::PacketQueue():
video_stream_id(-1),
max_video_packet_count(-1),
pre_event_video_packet_count(-1),
max_stream_id(-1),
packet_counts(nullptr),
deleting(false),
keep_keyframes(false),
warned_count(0),
has_out_of_order_packets_(false),
max_keyframe_interval_(0),
frames_since_last_keyframe_(0),
clear_packets_pending_(false),
next_queue_index_(0),
first_queue_index_(0),
released_before_(0),
capped_before_(0),
keyframe_interval_over_limit_(false)
{
}
/* Assumes queue is empty when adding streams
* Assumes first stream added will be the video stream
*/
int PacketQueue::addStream() {
std::lock_guard<std::mutex> lck(mutex);
deleting = false;
if (max_stream_id == -1) {
video_stream_id = 0;
max_stream_id = 0;
} else {
max_stream_id++;
}
packet_counts.reset(new int[max_stream_id + 1]);
for (int i = 0; i <= max_stream_id; ++i)
packet_counts[i] = 0;
return max_stream_id;
}
PacketQueue::~PacketQueue() {
clear();
while (!iterators.empty()) {
packetqueue_iterator *it = iterators.front();
iterators.pop_front();
delete it;
}
Debug(4, "Done in destructor");
}
/* Enqueues the given packet. Will maintain the it pointer and image packet counts.
* If we have reached our max image packet count, it will pop off as many packets as are needed.
* Thus it will ensure that the same packet never gets queued twice.
*/
bool PacketQueue::queuePacket(std::shared_ptr<ZMPacket> add_packet) {
std::vector<std::shared_ptr<ZMPacket>> packets_to_destroy;
{
std::unique_lock<std::mutex> lck(mutex);
if (deleting or zm_terminate) return false;
if (iterators.empty()) {
Debug(4, "No iterators so no one needs us to queue packets.");
return false;
}
if (!packet_counts[video_stream_id] and !add_packet->keyframe) {
Debug(4, "No video keyframe so no one needs us to queue packets.");
return false;
}
AVPacket *add_avpacket = add_packet->packet.get(); // because std::shared_ptr accesses are way more expensive
if (!has_out_of_order_packets_ and (add_avpacket->dts != AV_NOPTS_VALUE)) {
auto rit = pktQueue.rbegin();
// Find the previous packet for the stream, and check dts
while (rit != pktQueue.rend()) {
std::shared_ptr<ZMPacket> prev_packet = *rit;
if (prev_packet->packet->stream_index == add_avpacket->stream_index) {
if (prev_packet->packet->dts > add_avpacket->dts) {
Debug(1, "Have out of order packets");
ZM_DUMP_PACKET(prev_packet->packet, "queued_packet");
ZM_DUMP_PACKET(add_avpacket, "add_packet");
has_out_of_order_packets_ = true;
}
break;
}
rit++;
} // end while
} // end if doing out_of_order checking
if (video_stream_id == add_avpacket->stream_index) {
if (!add_packet->keyframe) {
frames_since_last_keyframe_ ++;
if (frames_since_last_keyframe_ > max_keyframe_interval_) {
max_keyframe_interval_ = frames_since_last_keyframe_;
Debug(1, "Have new keyframe interval %d", max_keyframe_interval_.load());
}
// Passthrough keeps every packet back to a keyframe, so a GOP longer
// than MaxImageBufferCount can't be trimmed: the queue holds all of it,
// and releaseDecoded() can only bound the decoded copies.
if (keep_keyframes and (max_video_packet_count > 0) and !keyframe_interval_over_limit_
and (frames_since_last_keyframe_ + 1 > max_video_packet_count)) {
keyframe_interval_over_limit_ = true;
Error("The camera's keyframe interval is longer than MaxImageBufferCount (%d frames). "
"Passthrough recording has to keep a whole keyframe interval of packets, so the packet queue "
"can't stay within MaxImageBufferCount. Raise MaxImageBufferCount above the keyframe interval "
"or shorten the camera's keyframe interval (GOP).",
max_video_packet_count);
}
} else {
frames_since_last_keyframe_ = 0;
if (!max_keyframe_interval_) max_keyframe_interval_ = 1;
}
} else {
Debug(1, "Not video stream %d", add_avpacket->stream_index);
}
add_packet->queue_index = next_queue_index_++;
pktQueue.push_back(add_packet);
/* Any iterators that are pointing to the end will now point to the newly pushed packet */
for (
auto iterators_it = iterators.begin();
iterators_it != iterators.end();
++iterators_it
) {
packetqueue_iterator *iterator_it = *iterators_it;
if (*iterator_it == pktQueue.end()) {
--(*iterator_it);
}
} // end foreach iterator
packet_counts[add_avpacket->stream_index] += 1;
Debug(2, "packet counts for %d is %d", add_avpacket->stream_index, packet_counts[add_avpacket->stream_index]);
if (
(add_avpacket->stream_index == video_stream_id)
and
(max_video_packet_count > 0)
and
(packet_counts[video_stream_id] > max_video_packet_count)
) {
if (!warned_count) {
warned_count++;
Warning("You have set the max video packets in the queue to %u."
" The queue is full %u. Either Analysis is not keeping up or"
" your camera's keyframe interval %d is larger than this setting."
, max_video_packet_count, packet_counts[video_stream_id], max_keyframe_interval_.load());
}
// Going to delete a GOP of packets. So find another keyframe and delete everything before it.
auto it = pktQueue.begin();
// Start at second packet because the first is always a keyframe unless we don't care about keyframes
it ++;
while ((*it != add_packet) && !(deleting or zm_terminate)) {
std::shared_ptr <ZMPacket>zm_packet = *it;
if (zm_packet->packet->stream_index == video_stream_id and zm_packet->keyframe) {
for ( it = pktQueue.begin(); *it != zm_packet; ) {
it = this->deletePacket(it, packets_to_destroy);
}
break;
} // end if erasing a whole gop
++it;
} // end foreach
} else if (warned_count > 0) {
warned_count--;
} // end if not able catch up
} // end scope for unique_lock — mutex released here
// We signal on every packet because someday we may analyze sound
Debug(4, "packetqueue queuepacket, unlocked signalling");
condition.notify_all();
// packets_to_destroy goes out of scope here, destroying packets without holding the mutex
return true;
} // end bool PacketQueue::queuePacket(ZMPacket* zm_packet)
packetqueue_iterator PacketQueue::deletePacket(packetqueue_iterator it, std::vector<std::shared_ptr<ZMPacket>> &deferred) {
auto zm_packet = *it;
for (
auto iterators_it = iterators.begin();
iterators_it != iterators.end();
++iterators_it
) {
auto iterator_it = *iterators_it;
// Have to check each iterator and make sure it doesn't point to the packet we are about to delete
if (*(*iterator_it) == zm_packet) {
Debug(1, "Bumping IT because it is at the front that we are deleting");
++(*iterator_it);
} else {
Debug(4, "Not Bumping IT because it is pointing at %d and we are %d", (*(*iterator_it))->image_index, zm_packet->image_index);
}
} // end foreach iterator
zm_packet->decoded = true; // analysis waits on decoding status so won't progress until we set this. Although we loop so maybe not.
zm_packet->notify_all();
packet_counts[zm_packet->packet->stream_index] -= 1;
Debug(4,
"Deleting a packet with stream index:%d image_index:%d with keyframe:%d, video frames in queue:%d max: %d, queuesize:%zu",
zm_packet->packet->stream_index,
zm_packet->image_index,
zm_packet->keyframe,
packet_counts[video_stream_id],
max_video_packet_count,
pktQueue.size());
// Move the shared_ptr into deferred so the packet is destroyed after the mutex is released
deferred.push_back(std::move(zm_packet));
return pktQueue.erase(it);
}
bool PacketQueue::clearPackets(const std::shared_ptr<ZMPacket> &add_packet) {
// Packets removed from the queue are collected here and destroyed after the
// mutex is released, so that expensive ZMPacket destructors (freeing images,
// av packets) don't run while the queue is locked.
std::vector<std::shared_ptr<ZMPacket>> packets_to_destroy;
{
std::unique_lock<std::mutex> lck(mutex);
// Only do queueCleaning if we are adding a video keyframe, so that we guarantee that there is one.
// No good. Have to satisfy two conditions:
// 1. packetqueue starts with a video keyframe
// 2. Have minimum # of video packets
// 3. No iterator can point to one of the packets
//
// So start at the beginning, counting video packets until the next keyframe.
// Then if deleting those packets doesn't break 1 and 2, then go ahead and delete them.
//
// One assumption that we can make is that there will be packets in the queue. Because we call it while holding a locked packet
if (deleting) return false;
// When keep_keyframes, normally only attempt clearing on video keyframes.
// If a previous clear attempt failed (clear_packets_pending_), retry on
// any video packet regardless of keyframe status.
if (keep_keyframes and ! (
add_packet->packet->stream_index == video_stream_id
and
(add_packet->keyframe or clear_packets_pending_)
and
(packet_counts[video_stream_id] > pre_event_video_packet_count)
and
*(pktQueue.begin()) != add_packet
)
) {
Debug(3, "stream index %d ?= video_stream_id %d, keyframe %d, keep_keyframes %d, pending %d, counts %d > pre_event_count %d at begin %d",
add_packet->packet->stream_index, video_stream_id, add_packet->keyframe, keep_keyframes,
clear_packets_pending_.load(std::memory_order_relaxed),
packet_counts[video_stream_id], pre_event_video_packet_count,
( *(pktQueue.begin()) != add_packet )
);
return false;
}
// If analysis_it isn't at the end, we need to keep that many additional packets
int tail_count = 0;
for (auto it = pktQueue.rbegin(); it != pktQueue.rend(); ++it) {
if (*it == add_packet) break;
if ((*it)->packet->stream_index == video_stream_id)
++tail_count;
}
Debug(1, "Tail count is %d, queue size is %zu, video_packets %d", tail_count, pktQueue.size(), packet_counts[video_stream_id]);
// Find the earliest queue_index that any iterator points to. Since every
// packet now carries a monotonic queue_index assigned at enqueue time, this
// single integer comparison per packet replaces the previous per-packet
// is_there_an_iterator_pointing_to_packet() scan over all iterators.
uint64_t min_iterator_queue_index = UINT64_MAX;
for (auto iterators_it = iterators.begin(); iterators_it != iterators.end(); ++iterators_it) {
packetqueue_iterator *iterator_it = *iterators_it;
if (*iterator_it != pktQueue.end()) {
uint64_t qi = (*(*iterator_it))->queue_index;
if (qi < min_iterator_queue_index)
min_iterator_queue_index = qi;
}
}
if (!keep_keyframes) {
int packets_removed = 0;
Debug(3, "Not keeping keyframes");
// If not doing passthrough, we don't care about starting with a keyframe so logic is simpler
while ((*pktQueue.begin() != add_packet) and (packet_counts[video_stream_id] > pre_event_video_packet_count + tail_count)) {
std::shared_ptr<ZMPacket> zm_packet = *pktQueue.begin();
if (zm_packet->queue_index >= min_iterator_queue_index) {
Debug(1, "Found iterator at beginning of queue.");
break;
}
packets_removed ++;
pktQueue.pop_front();
int stream_index = zm_packet->packet ? zm_packet->packet->stream_index : 0;
packet_counts[stream_index] -= 1;
Debug(4,
"Deleting a packet with stream index:%d image_index:%d with keyframe:%d, video frames in queue:%d max: %d, queuesize:%zu",
stream_index,
zm_packet->image_index,
zm_packet->keyframe,
packet_counts[video_stream_id],
pre_event_video_packet_count,
pktQueue.size());
packets_to_destroy.push_back(std::move(zm_packet));
} // end while
Debug(3, "Done removing %d packets from queue. packet_counts %d >? pre_event %d + tail %d = %d",
packets_removed, packet_counts[video_stream_id], pre_event_video_packet_count, tail_count, pre_event_video_packet_count + tail_count);
clear_packets_pending_ = (packets_removed == 0);
// mutex released here at end of scope, packets destroyed after
return (packets_removed > 0);
}
auto it = pktQueue.begin();
auto next_front = pktQueue.begin();
// First packet is special because we know it is a video keyframe and only need to check for iterator
std::shared_ptr<ZMPacket> zm_packet = *it;
if (zm_packet == add_packet) {
Debug(1, "First packet in queue is the analysis packet.");
return false;
}
int keyframe_interval_count = 0;
int video_packets_to_delete = 0; // This is a count of how many packets we will delete so we know when to stop looking
if (zm_packet->queue_index >= min_iterator_queue_index) {
Debug(3, "Found iterator Counted %d video packets. Which would leave %d in packetqueue tail count is %d",
video_packets_to_delete, packet_counts[video_stream_id]-video_packets_to_delete, tail_count);
clear_packets_pending_ = true;
return false;
}
++it;
// Since we have many packets in the queue, we should NOT be pointing at end so don't need to test for that
while (*it != add_packet) {
zm_packet = *it;
// Use > (not >=) so we still consider the iterator's own packet as a
// next_front candidate. Setting next_front to a packet that an iterator
// points at is safe: we delete strictly before next_front, so the
// iterator's packet stays in the queue.
if (zm_packet->queue_index > min_iterator_queue_index) {
Debug(3, "Past iterator. Counted %d video packets. Which would leave %d in packetqueue tail count is %d",
video_packets_to_delete, packet_counts[video_stream_id]-video_packets_to_delete, tail_count);
break;
}
if (zm_packet->packet->stream_index == video_stream_id) {
keyframe_interval_count++;
if (zm_packet->keyframe) {
Debug(3, "Have a video keyframe so setting next front to it. Keyframe interval so far is %d", keyframe_interval_count);
if (keyframe_interval_count > max_keyframe_interval_)
max_keyframe_interval_ = keyframe_interval_count;
keyframe_interval_count = 1;
next_front = it;
}
++video_packets_to_delete;
if (packet_counts[video_stream_id] - video_packets_to_delete <= pre_event_video_packet_count + tail_count) {
Debug(3, "Counted %d video packets. Which would leave %d in packetqueue tail count is %d",
video_packets_to_delete, packet_counts[video_stream_id]-video_packets_to_delete, tail_count);
break;
}
}
++it;
} // end while
Debug(1, "Resulting it pointing at latest packet? %d, next front points to begin? %d, Keyframe interval %d",
( *it == add_packet ),
( next_front == pktQueue.begin() ),
keyframe_interval_count
);
if (next_front != pktQueue.begin()) {
while (pktQueue.begin() != next_front) {
zm_packet = *pktQueue.begin();
if (!zm_packet) {
Error("NULL zm_packet in queue");
continue;
}
Debug(4,
"Deleting a packet with stream index:%d image_index:%d with keyframe:%d, video frames in queue:%d max: %d, queuesize:%zu",
zm_packet->packet->stream_index,
zm_packet->image_index,
zm_packet->keyframe,
packet_counts[video_stream_id],
pre_event_video_packet_count,
pktQueue.size());
pktQueue.pop_front();
packet_counts[zm_packet->packet->stream_index] -= 1;
packets_to_destroy.push_back(std::move(zm_packet));
}
clear_packets_pending_ = false;
} else {
clear_packets_pending_ = true;
}
} // end scope for unique_lock — mutex released here
// packets_to_destroy goes out of scope here, destroying packets without holding the mutex
return !clear_packets_pending_;
} // end bool PacketQueue::clearPackets(ZMPacket* zm_packet)
// Packets stay queued after every consumer has read them: passthrough needs the
// compressed data back to a keyframe, and an event may still open with
// pre-event frames. Their decoded copies (images, decoder frames) are needed
// far less: only from where an event could still start. That is
// pre_event_video_packet_count video packets before the analysis iterator,
// back to a keyframe when passthrough needs one (see get_event_start_packet_it).
// The earliest iterator is never ahead of the analysis one, so starting from
// it is conservative. Everything before that point keeps its compressed packet
// and loses its decoded data, which at high resolutions is several MB a frame
// (refs #4860). Events only open on the analysis thread, which also calls
// this, so no event starts behind the boundary while it is being computed. The
// window itself can change from another thread: a RELOAD action (zmu --reload)
// runs Monitor::Reload() on zmc's main loop, which can widen
// pre_event_video_packet_count or set keep_keyframes without clearing the
// queue. get_event_start_packet_it() therefore never starts an event before
// released_before_: an event right after such a reload gets fewer pre-event
// frames rather than frames without their images. A restart, SIGHUP or capture
// reconnect clears the queue, so only RELOAD gets there.
void PacketQueue::releaseDecoded() {
std::vector<std::shared_ptr<ZMPacket>> to_release;
{
std::unique_lock<std::mutex> lck(mutex);
if (deleting or pktQueue.empty() or iterators.empty()) return;
uint64_t min_iterator_queue_index = UINT64_MAX;
for (const packetqueue_iterator *iterator_it : iterators) {
if (*iterator_it != pktQueue.end()) {
uint64_t qi = (*(*iterator_it))->queue_index;
if (qi < min_iterator_queue_index) min_iterator_queue_index = qi;
}
}
// Walk back from the earliest iterator's packet (or the newest packet when
// every iterator is at the end) as openEvent would.
auto it = pktQueue.end();
--it;
while (it != pktQueue.begin() and (*it)->queue_index > min_iterator_queue_index) --it;
int pre_event = pre_event_video_packet_count;
while (pre_event > 0 and it != pktQueue.begin()) {
if ((*it)->packet->stream_index == video_stream_id) pre_event--;
--it;
}
if (pre_event > 0) return; // the whole queue is still within reach of an event
const uint64_t pre_event_start = (*it)->queue_index;
if (keep_keyframes) {
while (it != pktQueue.begin() and !((*it)->packet->stream_index == video_stream_id and (*it)->keyframe)) --it;
}
const uint64_t boundary = (*it)->queue_index;
if (boundary > released_before_) {
for (auto p = pktQueue.begin(); p != it; ++p) {
if ((*p)->queue_index < released_before_ or (*p)->queue_index < capped_before_) continue;
to_release.push_back(*p);
}
released_before_ = boundary;
}
// Passthrough keeps every packet back to a keyframe. When the keyframe
// interval is longer than MaxImageBufferCount, queuePacket() finds no later
// keyframe to trim to and the queue holds the whole GOP, each packet with a
// decoded frame: several MB each at high resolutions, GBs per queue. Bound
// the decoded data instead: only the newest max_video_packet_count video
// packets keep it, and never less than the pre-event window. The compressed
// packets stay, so an event still starts on the keyframe with all its video;
// it writes no capture JPEGs for the frames that lost their image, as for
// frames before released_before_ (refs #4860).
if (keep_keyframes and max_video_packet_count > 0) {
auto oldest_kept = pktQueue.end();
int video_packets = 0;
while (oldest_kept != pktQueue.begin() and video_packets < max_video_packet_count) {
--oldest_kept;
if ((*oldest_kept)->packet->stream_index == video_stream_id) video_packets++;
}
if (video_packets == max_video_packet_count) {
const uint64_t cap = std::min((*oldest_kept)->queue_index, pre_event_start);
if (cap > capped_before_) {
for (auto p = pktQueue.begin(); p != pktQueue.end() and (*p)->queue_index < cap; ++p) {
if ((*p)->queue_index < released_before_ or (*p)->queue_index < capped_before_) continue;
to_release.push_back(*p);
}
capped_before_ = cap;
}
}
}
}
// Free outside the queue lock, like clearPackets(). Nothing should hold these
// packets, but never wait on one that is.
for (const auto &packet : to_release) {
std::unique_lock<std::mutex> plck(packet->mutex_, std::try_to_lock);
if (plck.owns_lock()) packet->release_decoded();
}
}
void PacketQueue::stop() {
std::lock_guard<std::mutex> lck(mutex);
deleting = true;
condition.notify_all();
for (const auto &p : pktQueue) {
p->notify_all();
}
}
void PacketQueue::clear() {
Debug(1, "Clearing packetqueue");
// Move packets out under the lock, then let their destructors run after
// the mutex is released. Mirrors queuePacket()/clearPackets() so we don't
// stall other threads on expensive ZMPacket teardown (Image, AVPacket).
std::vector<std::shared_ptr<ZMPacket>> packets_to_destroy;
{
std::lock_guard<std::mutex> lck(mutex);
deleting = true;
// Why are we notifying?
condition.notify_all();
if (!packet_counts) // special case, not initialised
return;
while (!pktQueue.empty()) {
std::shared_ptr<ZMPacket> packet = pktQueue.front();
Debug(1,
"Deleting a packet with stream index:%d image_index:%d with keyframe:%d, video frames in queue:%d max: %d, queuesize:%zu",
packet->packet->stream_index,
packet->image_index,
packet->keyframe,
packet_counts[video_stream_id],
pre_event_video_packet_count,
pktQueue.size());
packet_counts[packet->packet->stream_index] -= 1;
pktQueue.pop_front();
packets_to_destroy.push_back(std::move(packet));
}
Debug(1, "Packetqueue is clear, deleting iterators");
for (
std::list<packetqueue_iterator *>::iterator iterators_it = iterators.begin();
iterators_it != iterators.end();
++iterators_it
) {
packetqueue_iterator *iterator_it = *iterators_it;
*iterator_it = pktQueue.begin();
} // end foreach iterator
packet_counts.reset();
max_stream_id = -1;
max_keyframe_interval_ = 0;
keyframe_interval_over_limit_ = false;
first_queue_index_ = next_queue_index_;
Debug(1, "Packetqueue is clear, notifying");
condition.notify_all();
} // end scope for lock_guard — mutex released here
// packets_to_destroy goes out of scope here, destroying packets without holding the mutex
} // end void PacketQueue::clear()
unsigned int PacketQueue::size() {
std::unique_lock<std::mutex> lck(mutex);
return pktQueue.size();
}
int PacketQueue::packet_count(int stream_id) {
std::unique_lock<std::mutex> lck(mutex);
if (stream_id < 0 or stream_id > max_stream_id) {
Error("Invalid stream_id %d max is %d", stream_id, max_stream_id);
return -1;
}
if (!packet_counts) {
Error("packet_counts not allocated");
return -1;
}
return packet_counts[stream_id];
} // end int PacketQueue::packet_count(int stream_id)
ZMPacketLock PacketQueue::get_packet_no_wait(packetqueue_iterator *it) {
std::unique_lock<std::mutex> lck(mutex);
Debug(4, "get_packet_no_wait using it %p queue end? %d",
std::addressof(*it), (*it == pktQueue.end()));
if (deleting or zm_terminate)
return ZMPacketLock();
if (*it == pktQueue.end())
return ZMPacketLock();
std::shared_ptr<ZMPacket> p = *(*it);
ZMPacketLock packet_lock(p);
if (packet_lock.trylock()) {
Debug(2, "Locked packet %d, unlocking packetqueue mutex", p->image_index);
return packet_lock;
}
//Debug(1, "Failed to lock packet %d, returning... something?", p->image_index);
return ZMPacketLock();
}
// Returns a packet. Packet will be locked
ZMPacketLock PacketQueue::get_packet(packetqueue_iterator *it) {
if (deleting or zm_terminate)
return ZMPacketLock();
{
// scope for lock
std::unique_lock<std::mutex> lck(mutex);
Debug(4, "get_packet using it %p queue end? %d",
std::addressof(*it), (*it == pktQueue.end()));
while (!(deleting or zm_terminate)) {
while ((*it == pktQueue.end()) and !(deleting or zm_terminate)) {
Debug(4, "waiting. Queue size %zu it == end? %d", pktQueue.size(), (*it == pktQueue.end()));
condition.wait(lck);
}
if (deleting or zm_terminate) break;
std::shared_ptr<ZMPacket> p = *(*it);
if (!p) {
Error("Null p?!");
return ZMPacketLock();
}
Debug(3, "get_packet using it %p trylocking index %d", std::addressof(*it), p->image_index);
{
std::shared_ptr<ZMPacket> p = *(*it);
ZMPacketLock packet_lock(p);
if (packet_lock.trylock()) {
Debug(4, "Locked packet %d, unlocking packetqueue mutex", p->image_index);
return packet_lock;
}
}
Debug(4, "waiting on packet %d. Queue size %zu it == end? %d", p->image_index, pktQueue.size(), (*it == pktQueue.end()));
condition.wait(lck);
} // end while ! deleting or zm_terminate
condition.notify_all();
} // end scope for lock
Debug(1, "terminated, leaving");
return ZMPacketLock();
} // end ZMPacketLock *PacketQueue::get_packet(it)
// Returns a packet. Packet will be locked
ZMPacketLock PacketQueue::get_packet_and_increment_it(packetqueue_iterator *it) {
if (deleting or zm_terminate)
return ZMPacketLock();
{
// scope for lock
std::unique_lock<std::mutex> lck(mutex);
Debug(4, "get_packet_and_increment_it using it %p queue end? %d",
std::addressof(*it), (*it == pktQueue.end()));
while (!(deleting or zm_terminate)) {
while ((*it == pktQueue.end()) and !(deleting or zm_terminate)) {
Debug(2, "waiting. Queue size %zu it == end? %d", pktQueue.size(), (*it == pktQueue.end()));
condition.wait(lck);
}
if (deleting or zm_terminate) break;
std::shared_ptr<ZMPacket> p = *(*it);
if (!p) {
Error("Null p?!");
break;
}
Debug(3, "get_packet using it %p locking index %d", std::addressof(*it), p->image_index);
{
ZMPacketLock packet_lock(p);
if (packet_lock.trylock()) {
Debug(2, "Locked packet %d, unlocking packetqueue mutex, incrementing it", p->image_index);
++(*it);
return packet_lock;
}
}
Debug(2, "waiting. Queue size %zu it == end? %d", pktQueue.size(), (*it == pktQueue.end()));
condition.wait(lck);
} // end while !lp
} // end scope for lock
return ZMPacketLock();
} // end ZMPacketLock *PacketQueue::get_packet_and_increment_it(it)
bool PacketQueue::increment_it(packetqueue_iterator *it, bool wait) {
std::unique_lock<std::mutex> lck(mutex);
Debug(2, "Incrementing %p, queue size %zu, end? %d, deleting %d", it, pktQueue.size(), ((*it) == pktQueue.end()), deleting);
if (wait) {
while ((*it == pktQueue.end()) and !(deleting or zm_terminate)) {
Debug(2, "waiting. Queue size %zu it == end? %d", pktQueue.size(), (*it == pktQueue.end()));
condition.wait(lck);
Debug(2, "waiting. Queue size %zu it == end? %d", pktQueue.size(), (*it == pktQueue.end()));
}
if (deleting or zm_terminate) return false;
} else {
if (((*it) == pktQueue.end()) or deleting) {
if (!deleting) Debug(1, "increment_it at end!");
return false;
}
}
++(*it);
if (*it != pktQueue.end()) {
Debug(2, "Incrementing %p, %p still not at end, so returning true", it, std::addressof(*it));
return true;
}
Debug(2, "At end");
return false;
} // end bool PacketQueue::increment_it(packetqueue_iterator *it)
// Increment it only considering packets for a given stream
bool PacketQueue::increment_it(packetqueue_iterator *it, int stream_id) {
std::lock_guard<std::mutex> lck(mutex);
Debug(2, "Incrementing %p, queue size %zu, end? %d", it, pktQueue.size(), (*it == pktQueue.end()));
if (*it == pktQueue.end()) {
return false;
}
do {
++(*it);
} while ( (*it != pktQueue.end()) and ( (*(*it))->packet->stream_index != stream_id) );
if ( *it != pktQueue.end() ) {
Debug(2, "Incrementing %p, still not at end, so incrementing", it);
return true;
}
return false;
} // end bool PacketQueue::increment_it(packetqueue_iterator *it)
packetqueue_iterator *PacketQueue::get_event_start_packet_it(
packetqueue_iterator snapshot_it,
unsigned int pre_event_count
) {
std::lock_guard<std::mutex> lck(mutex);
packetqueue_iterator *it = new packetqueue_iterator;
iterators.push_back(it);
*it = snapshot_it;
if (*it == pktQueue.end()) {
// Should never be able to happen.
*it = pktQueue.begin();
}
std::shared_ptr<ZMPacket> packet = *(*it);
// Packets before released_before_ may have lost their decoded data to
// releaseDecoded(), so an event must not start on one: treat the first packet
// after them as the front of the queue. With the window unchanged the walk
// never gets that far; it only can after a RELOAD widened the window or set
// keep_keyframes (refs #4860).
auto at_front = [this, &it]() {
return (*it) == pktQueue.begin() or (*std::prev(*it))->queue_index < released_before_;
};
//ZM_DUMP_PACKET(packet->packet, "snapshot packet");
// Step one count back pre_event_count frames as the minimum
// Do not assume that snapshot_it is video
// snapshot it might already point to the beginning
while (pre_event_count and !at_front()) {
/*
Debug(1, "Previous packet pre_event_count %d stream_index %d keyframe %d score %d",
pre_event_count, packet->packet->stream_index, packet->keyframe, packet->score);
ZM_DUMP_PACKET(packet->packet, "");
*/
if (packet->packet->stream_index == video_stream_id)
pre_event_count --;
(*it)--;
packet = *(*it);
}
// it either points to beginning or we have seen pre_event_count video packets.
// The loop above doesn't count the begin packet, so check it now.
if (pre_event_count and (packet->packet->stream_index == video_stream_id)) {
pre_event_count--;
}
if (pre_event_count) {
if ((*it) != pktQueue.begin()) {
Warning("Hit packets released under a smaller pre-event window before satisfying pre_event_count. "
"Needed %d more video frames", pre_event_count);
return start_on_keyframe(it, snapshot_it);
} else if (packet->queue_index == first_queue_index_) {
// Nothing has been trimmed since startup or the last clear() (e.g. a
// capture reconnect), so the queue never held enough packets.
Debug(1, "Hit end of packetqueue before satisfying pre_event_count. Needed %d more video frames", pre_event_count);
} else {
Warning("Hit end of packetqueue before satisfying pre_event_count. Needed %d more video frames", pre_event_count);
}
//ZM_DUMP_PACKET(packet->packet, "");
return it;
} else if (!keep_keyframes) {
// Are encoding, so don't care about keyframes
// We could be pointing to a non-video frame though. Do we care?
return it;
}
while (!at_front()) {
//ZM_DUMP_PACKET(packet->packet, "No keyframe");
if ((packet->packet->stream_index == video_stream_id) and packet->keyframe)
return it; // Success
--(*it);
packet = *(*it);
}
if ((*it) != pktQueue.begin()) return start_on_keyframe(it, snapshot_it);
if (!packet->keyframe) {
Warning("Hit beginning of packetqueue and packet is not a keyframe. index is %d", packet->image_index);
}
return it;
} // end packetqueue_iterator *PacketQueue::get_event_start_packet_it
// The walk back in get_event_start_packet_it() stopped at the oldest packet
// that still has its decoded data. For passthrough, move forward to a video
// keyframe if that packet isn't one, so the recording can start on it. It
// normally is: releaseDecoded() stops on a keyframe when keep_keyframes is set,
// and only a reload from encode to passthrough leaves a boundary off one. If
// there is no keyframe before the snapshot, stay put; the event writes video
// from the next keyframe, as it does when the front of the queue isn't one.
packetqueue_iterator *PacketQueue::start_on_keyframe(packetqueue_iterator *it, packetqueue_iterator snapshot_it) {
if (!keep_keyframes) return it;
for (auto k = *it; ; ++k) {
if (((*k)->packet->stream_index == video_stream_id) and (*k)->keyframe) {
*it = k;
return it;
}
if (k == snapshot_it or std::next(k) == pktQueue.end()) break;
}
Warning("No keyframe between the oldest packet with decoded data and the event's snapshot");
return it;
}
void PacketQueue::dumpQueue() {
std::lock_guard<std::mutex> lck(mutex);
for (auto it = pktQueue.rbegin(); it != pktQueue.rend(); ++it) {
std::shared_ptr<ZMPacket> zm_packet = *it;
ZM_DUMP_PACKET(zm_packet->packet, is_there_an_iterator_pointing_to_packet(zm_packet) ? "*" : "");
}
}
/* Returns an iterator to the first video keyframe in the queue.
* nullptr if no keyframe video packet exists.
*/
packetqueue_iterator * PacketQueue::get_video_it(bool wait) {
std::unique_lock<std::mutex> lck(mutex);
packetqueue_iterator *it = new packetqueue_iterator;
iterators.push_back(it);
*it = pktQueue.begin();
if (wait) {
while ( ((! pktQueue.size()) or (*it == pktQueue.end())) and !zm_terminate and !deleting ) {
Debug(2, "waiting for packets in queue. Queue size %zu it == end? %d", pktQueue.size(), (*it == pktQueue.end()));
condition.wait(lck);
*it = pktQueue.begin();
}
if (deleting or zm_terminate) {
iterators.remove(it);
delete it;
return nullptr;
}
}
while (*it != pktQueue.end()) {
std::shared_ptr<ZMPacket> zm_packet = *(*it);
if (!zm_packet) {
Error("Null zmpacket in queue!?");
iterators.remove(it);
delete it;
return nullptr;
}
Debug(1, "Packet keyframe %d for stream %d, so returning the it to it",
zm_packet->keyframe, zm_packet->packet->stream_index);
if (zm_packet->keyframe and ( zm_packet->packet->stream_index == video_stream_id )) {
Debug(1, "Found a keyframe for stream %d, so returning the it to it", video_stream_id);
return it;
}
++(*it);
}
Debug(1, "Didn't find a keyframe for stream %d, so returning the it to it", video_stream_id);
return it;
} // get video_it
void PacketQueue::free_it(packetqueue_iterator *it) {
std::lock_guard<std::mutex> lck(mutex);
for (
std::list<packetqueue_iterator *>::iterator iterators_it = iterators.begin();
iterators_it != iterators.end();
++iterators_it
) {
if (*iterators_it == it) {
delete *iterators_it; // delete the iterator
iterators.erase(iterators_it); // this will delete the pointer to it.
break;
}
}
}
bool PacketQueue::is_there_an_iterator_pointing_to_packet(const std::shared_ptr<ZMPacket> zm_packet) {
for (
std::list<packetqueue_iterator *>::iterator iterators_it = iterators.begin();
iterators_it != iterators.end();
++iterators_it
) {
packetqueue_iterator *iterator_it = *iterators_it;
if (*iterator_it == pktQueue.end()) {
Debug(4, "Checking iterator %p == end", std::addressof(*iterator_it));
continue;
}
Debug(4, "Checking iterator %p == packet ? %d", std::addressof(*iterator_it), ( *(*iterator_it) == zm_packet ));
// Have to check each iterator and make sure it doesn't point to the packet we are about to delete
if (*(*iterator_it) == zm_packet) {
return true;
}
} // end foreach iterator
return false;
}
void PacketQueue::setMaxVideoPackets(int p) {
std::lock_guard<std::mutex> lck(mutex);
max_video_packet_count = p;
Debug(1, "Setting max_video_packet_count to %d", p);
if (max_video_packet_count < 0)
max_video_packet_count = 0;
keyframe_interval_over_limit_ = false;
}
void PacketQueue::setKeepKeyframes(bool k) {
std::lock_guard<std::mutex> lck(mutex);
keep_keyframes = k;
keyframe_interval_over_limit_ = false;
}
void PacketQueue::setPreEventVideoPackets(int p) {
std::lock_guard<std::mutex> lck(mutex);
pre_event_video_packet_count = p;
Debug(1, "Setting pre_event_video_packet_count to %d", p);
if (pre_event_video_packet_count < 1)
pre_event_video_packet_count = 1;
// We can simplify a lot of logic in queuePacket if we can assume at least 1 packet in queue
}
void PacketQueue::notify_all() {
condition.notify_all();
};
void PacketQueue::wait() {
std::unique_lock<std::mutex> lck(mutex);
condition.wait(lck);
}
void PacketQueue::wait_for(Microseconds duration) {
std::unique_lock<std::mutex> lck(mutex);
condition.wait_for(lck, duration);
}