mirror/zoneminder
1
0 Fork 0
mirror of https://github.com/ZoneMinder/zoneminder.git synced 2026-10-12 23:03:52 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
7f7c9cc81bd8e761924e1bfac29394b28ea6fdff
zoneminder/src/zm_monitorstream.cpp
T
nicandClaude Opus 5.5 596fb5ab70 fix: free a stream's monitor and playback buffer when nph-zms exits (#5186) 
* fix: free a stream's monitor and playback buffer when nph-zms exits

Every nph-zms exited with ~1 MB still allocated, reported by LeakSanitizer
on every stream:

- The monitor's zones each hold a shared_ptr back to it, so releasing the
  stream's reference never freed the Monitor, its zones (a full-frame zone
  image each), camera or links. zmc breaks the cycle with disconnect() on
  exit; zms never did. ~StreamBase now calls it.
- The buffered-playback SwapImage array (connkey + buffer) was never
  deleted. MonitorStream now frees it in its destructor.

Freed at process exit either way, so nothing grows, but it shows up in
every leak report and hides real ones.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: check a stream frees its monitor and zones on destruction refs #5186

A stream holds a monitor with one zone; once the stream is destroyed, a
weak_ptr to the monitor must have expired. Each Zone holds a shared_ptr
back to its Monitor, so without ~StreamBase breaking the cycle the monitor,
its zones and their images outlive the stream. Fails on 82901ac1f
(REQUIRE( watched.expired() )), passes with the fix. Needs no database or
sanitizer, so ci-cpp-tests runs it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor: hold the buffered-playback array in a unique_ptr refs #5186

temp_image_buffer is now a std::unique_ptr<SwapImage[]>, so it is freed
with the stream without a hand-written destructor, and the leak cannot
come back by dropping that line. The array is private and only allocated
in runStream(), so a unit test cannot see it freed without a sanitizer;
under LeakSanitizer, buffered nph-zms streams leaked 48,008 bytes from
runStream() on 82901ac1f and nothing with this branch.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 17:34:09 -04:00

1150 lines
42 KiB
C++
Raw Blame History

//
// ZoneMinder Monitor Class Implementation, $Date$, $Revision$
// Copyright (C) 2001-2008 Philip Coombes
//
// This program 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 2
// of the License, or (at your option) any later version.
//
// This program 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 this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
//
#include "zm_monitorstream.h"
#include "zm_monitor.h"
#include "zm_signal.h"
#include "zm_time.h"
#include <libavutil/pixdesc.h>
#include <arpa/inet.h>
#include <glob.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <unistd.h>
#ifdef __FreeBSD__
#include <netinet/in.h>
#endif
bool MonitorStream::checkSwapPath(const char *path, bool create_path) {
struct stat stat_buf;
if (stat(path, &stat_buf) < 0) {
if (create_path and (errno == ENOENT) ) {
Debug(3, "Swap path '%s' missing, creating", path);
if (mkdir(path, 0755)) {
Error("Can't mkdir %s: %s", path, strerror(errno));
return false;
}
if (stat(path, &stat_buf) < 0) {
Error("Can't stat '%s': %s", path, strerror(errno));
return false;
}
} else {
Error("Can't stat '%s': %s", path, strerror(errno));
return false;
}
}
if (!S_ISDIR(stat_buf.st_mode)) {
Error("Swap image path '%s' is not a directory", path);
return false;
}
uid_t uid = getuid();
gid_t gid = getgid();
mode_t mask = 0;
if (uid == stat_buf.st_uid) {
// If we are the owner
mask = 00700;
} else if (gid == stat_buf.st_gid) {
// If we are in the owner group
mask = 00070;
} else {
// We are neither the owner nor in the group
mask = 00007;
}
if ((stat_buf.st_mode & mask) != mask) {
Error("Insufficient permissions on swap image path '%s'", path);
return false;
}
return true;
} // end bool MonitorStream::checkSwapPath(const char *path, bool create_path)
void MonitorStream::processCommand(const CmdMsg *msg) {
Debug(2, "Got message, type %d, msg %d", msg->msg_type, msg->msg_data[0]);
if (!monitor) {
Warning("Cannot process command, monitor is not loaded");
return;
}
// Check for incoming command
switch ((MsgCommand)msg->msg_data[0]) {
case CMD_PAUSE :
Debug(1, "Got PAUSE command");
stopped = false;
paused = true;
delayed = true;
last_frame_sent = now;
break;
case CMD_PLAY :
Debug(1, "Got PLAY command");
stopped = false;
if (paused) {
paused = false;
delayed = true;
}
replay_rate = ZM_RATE_BASE;
break;
case CMD_VARPLAY :
Debug(1, "Got VARPLAY command");
stopped = false;
if (paused) {
paused = false;
delayed = true;
}
replay_rate = (((unsigned char)msg->msg_data[1]<<8)|(unsigned char)msg->msg_data[2])-VARPLAY_RATE_OFFSET;
break;
case CMD_STOP :
Debug(1, "Got STOP command");
stopped = true;
paused = false;
delayed = false;
step = 0;
send_twice = false;
break;
case CMD_FASTFWD :
Debug(1, "Got FAST FWD command");
stopped = false;
if (paused) {
paused = false;
delayed = true;
}
// Set play rate
switch (replay_rate) {
case 2 * ZM_RATE_BASE :
replay_rate = 5 * ZM_RATE_BASE;
break;
case 5 * ZM_RATE_BASE :
replay_rate = 10 * ZM_RATE_BASE;
break;
case 10 * ZM_RATE_BASE :
replay_rate = 25 * ZM_RATE_BASE;
break;
case 25 * ZM_RATE_BASE :
case 50 * ZM_RATE_BASE :
replay_rate = 50 * ZM_RATE_BASE;
break;
default :
replay_rate = 2 * ZM_RATE_BASE;
break;
}
break;
case CMD_MAXFPS : {
double int_part = ((unsigned char) msg->msg_data[1] << 24) | ((unsigned char) msg->msg_data[2] << 16)
| ((unsigned char) msg->msg_data[3] << 8) | (unsigned char) msg->msg_data[4];
double dec_part = ((unsigned char) msg->msg_data[5] << 24) | ((unsigned char) msg->msg_data[6] << 16)
| ((unsigned char) msg->msg_data[7] << 8) | (unsigned char) msg->msg_data[8];
maxfps = (int_part + dec_part / 1000000.0);
Debug(1, "Got MAXFPS %f", maxfps.load());
break;
}
case CMD_SLOWFWD :
Debug(1, "Got SLOW FWD command");
stopped = false;
paused = true;
delayed = true;
replay_rate = ZM_RATE_BASE;
step = 1;
break;
case CMD_SLOWREV :
Debug(1, "Got SLOW REV command");
stopped = false;
paused = true;
delayed = true;
replay_rate = ZM_RATE_BASE;
step = -1;
break;
case CMD_FASTREV :
Debug(1, "Got FAST REV command");
stopped = false;
if (paused) {
paused = false;
delayed = true;
}
// Set play rate
switch (replay_rate) {
case -2 * ZM_RATE_BASE :
replay_rate = -5 * ZM_RATE_BASE;
break;
case -5 * ZM_RATE_BASE :
replay_rate = -10 * ZM_RATE_BASE;
break;
case -10 * ZM_RATE_BASE :
replay_rate = -25 * ZM_RATE_BASE;
break;
case -25 * ZM_RATE_BASE :
case -50 * ZM_RATE_BASE :
replay_rate = -50 * ZM_RATE_BASE;
break;
default :
replay_rate = -2 * ZM_RATE_BASE;
break;
}
break;
case CMD_ZOOMIN :
x = ((unsigned char)msg->msg_data[1]<<8)|(unsigned char)msg->msg_data[2];
y = ((unsigned char)msg->msg_data[3]<<8)|(unsigned char)msg->msg_data[4];
zoom += 10;
Debug(1, "Got ZOOM IN command, to %d,%d zoom value %d%%", x.load(), y.load(), zoom.load());
break;
case CMD_ZOOMOUT :
zoom -= 10;
if (zoom.load() < 100) zoom = 100;
Debug(1, "Got ZOOM OUT command resulting zoom %d%%", zoom.load());
break;
case CMD_ZOOMSTOP :
zoom = 100;
Debug(1, "Got ZOOM OUT FULL command resulting zoom %d%%", zoom.load());
break;
case CMD_PAN :
x = ((unsigned char)msg->msg_data[1]<<8)|(unsigned char)msg->msg_data[2];
y = ((unsigned char)msg->msg_data[3]<<8)|(unsigned char)msg->msg_data[4];
Debug(1, "Got PAN command, to %d,%d", x.load(), y.load());
break;
case CMD_SCALE :
scale = ((unsigned char)msg->msg_data[1]<<8)|(unsigned char)msg->msg_data[2];
Debug(1, "Got SCALE command, to %d", scale.load());
break;
case CMD_QUIT :
Info("User initiated exit - CMD_QUIT");
zm_terminate = true;
break;
case CMD_ANALYZE_ON :
frame_type = FRAME_ANALYSIS;
Debug(1, "ANALYSIS on");
break;
case CMD_ANALYZE_OFF :
frame_type = FRAME_NORMAL;
Debug(1, "ANALYSIS off");
break;
case CMD_QUERY :
Debug(1, "Got QUERY command, sending STATUS");
break;
default :
Error("Got unexpected command %d", msg->msg_data[0]);
break;
} // end switch command
struct {
int id;
int state;
double fps;
double capture_fps;
double analysis_fps;
int buffer_level;
int rate;
double delay;
int zoom;
int scale;
bool delayed;
bool paused;
bool enabled;
bool forced;
int score;
int analysing;
bool analysis_image;
bool stopped;
} status_data;
status_data.id = monitor->Id();
{
// Hold monitor_mutex while accessing shared memory to prevent
// loadMonitor() from disconnecting/remapping underneath us.
std::lock_guard<std::mutex> lck(monitor_mutex);
if (!monitor->ShmValid()) {
status_data.fps = 0.0;
status_data.capture_fps = 0.0;
status_data.analysis_fps = 0.0;
status_data.state = Monitor::UNKNOWN;
status_data.enabled = false;
status_data.forced = false;
status_data.buffer_level = 0;
status_data.analysing = 0;
status_data.score = 0;
} else {
FPSeconds elapsed = now - last_fps_update;
if (elapsed.count()) {
actual_fps = (actual_fps + (frame_count - last_frame_count) / elapsed.count())/2;
last_frame_count = frame_count;
last_fps_update = now;
}
status_data.fps = actual_fps;
status_data.capture_fps = monitor->get_capture_fps();
status_data.analysis_fps = monitor->get_analysis_fps();
status_data.state = monitor->shared_data->state;
status_data.enabled = monitor->trigger_data->trigger_state != Monitor::TriggerState::TRIGGER_OFF;
status_data.forced = monitor->trigger_data->trigger_state == Monitor::TriggerState::TRIGGER_ON;
status_data.analysing = monitor->shared_data->analysing;
status_data.score = monitor->shared_data->last_frame_score;
status_data.buffer_level =
MonitorStreamBufferLevel(temp_write_index, temp_read_index, temp_image_buffer_count);
}
} // end monitor_mutex scope
status_data.delayed = delayed;
status_data.paused = paused;
status_data.stopped = stopped;
status_data.rate = replay_rate;
status_data.delay = FPSeconds(now - last_frame_sent).count();
status_data.zoom = zoom;
status_data.scale = scale;
status_data.analysis_image = (frame_type == FRAME_ANALYSIS) &&
monitor->ShmValid() &&
(monitor->Analysing() != Monitor::ANALYSING_NONE);
Debug(2, "viewing fps: %.2f capture_fps: %.2f analysis_fps: %.2f Buffer Level:%d, Delayed:%d, Paused:%d, Stopped:%d, Rate:%d, delay:%.3f, Zoom:%d, Enabled:%d Forced:%d score: %d analysis_image: %d",
status_data.fps,
status_data.capture_fps,
status_data.analysis_fps,
status_data.buffer_level,
status_data.delayed,
status_data.paused,
status_data.stopped,
status_data.rate,
status_data.delay,
status_data.zoom,
status_data.enabled,
status_data.forced,
status_data.score,
status_data.analysis_image
);
DataMsg status_msg;
status_msg.msg_type = MSG_DATA_WATCH;
memcpy(&status_msg.msg_data, &status_data, sizeof(status_data));
int nbytes = 0;
if ((nbytes = sendto(sd, &status_msg, sizeof(status_msg), MSG_DONTWAIT, (sockaddr *)&rem_addr, sizeof(rem_addr))) < 0) {
Error("Can't sendto on sd %d: %s", sd, strerror(errno));
}
Debug(2, "Number of bytes sent to (%s): (%d)", rem_addr.sun_path, nbytes);
} // end void MonitorStream::processCommand(const CmdMsg *msg)
bool MonitorStream::sendFrame(const std::string &filepath, SystemTimePoint timestamp) {
bool send_raw = ((scale>=ZM_SCALE_BASE)&&(zoom==ZM_SCALE_BASE));
if (
(type != STREAM_JPEG)
||
(!config.timestamp_on_capture)
)
send_raw = false;
if (!send_raw) {
Image temp_image(filepath.c_str());
return sendFrame(&temp_image, timestamp);
} else {
int img_buffer_size = 0;
static unsigned char img_buffer[ZM_MAX_IMAGE_SIZE];
if (FILE *fdj = fopen(filepath.c_str(), "r")) {
img_buffer_size = fread(img_buffer, 1, sizeof(img_buffer), fdj);
fclose(fdj);
} else {
Error("Can't open %s: %s", filepath.c_str(), strerror(errno));
return false;
}
// Calculate how long it takes to actually send the frame
TimePoint send_start_time = std::chrono::steady_clock::now();
if (
(0 > fprintf(stdout, "Content-Length: %d\r\nX-Timestamp: %.6f\r\n\r\n",
img_buffer_size, std::chrono::duration_cast<FPSeconds>(timestamp.time_since_epoch()).count()))
||
(fwrite(img_buffer, img_buffer_size, 1, stdout) != 1)
) {
if (!zm_terminate)
Warning("Unable to send stream frame: %s", strerror(errno));
return false;
}
fputs("\r\n", stdout);
fflush(stdout);
if (maxfps > 0.0) {
TimePoint send_end_time = std::chrono::steady_clock::now();
TimePoint::duration frame_send_time = send_end_time - send_start_time;
if (frame_send_time > Milliseconds(lround(Milliseconds::period::den / maxfps))) {
Debug(1, "Frame send time %" PRIi64 " ms too slow, throttling maxfps to %.2f",
static_cast<int64>(std::chrono::duration_cast<Milliseconds>(frame_send_time).count()),
maxfps.load());
}
}
last_frame_sent = now;
return true;
}
return false;
}
bool MonitorStream::sendFrame(Image *image, SystemTimePoint timestamp) {
if (!config.timestamp_on_capture) {
monitor->TimestampImage(image, timestamp);
}
Image *send_image = prepareImage(image);
fputs("--" BOUNDARY "\r\n", stdout);
// Calculate how long it takes to actually send the frame
TimePoint send_start_time = std::chrono::steady_clock::now();
if (type == STREAM_MPEG) {
if (!vid_stream) {
vid_stream = new VideoStream("pipe:", format, bitrate, effective_fps, send_image->Colours(), send_image->SubpixelOrder(), send_image->Width(), send_image->Height());
fprintf(stdout, "Content-Type: %s\r\n\r\n", vid_stream->MimeType());
if (!vid_stream->OpenStream()) {
Error("Failed to open video stream");
delete vid_stream;
vid_stream = nullptr;
return false;
}
}
static SystemTimePoint base_time;
if (!frame_count) {
base_time = timestamp;
}
SystemTimePoint::duration delta_time = timestamp - base_time;
/* double pts = */ vid_stream->EncodeFrame(send_image->Buffer(), send_image->Size(), config.mpeg_timed_frames, delta_time.count());
} else {
reserveTempImgBuffer(send_image->Size());
size_t img_buffer_size = 0;
unsigned char *img_buffer = temp_img_buffer;
switch (type) {
case STREAM_JPEG :
send_image->EncodeJpeg(img_buffer, &img_buffer_size);
fputs("Content-Type: image/jpeg\r\n", stdout);
break;
case STREAM_RAW :
fputs("Content-Type: image/x-rgb\r\n", stdout);
img_buffer = send_image->Buffer();
img_buffer_size = send_image->Size();
break;
case STREAM_ZIP :
#if HAVE_ZLIB_H
fputs("Content-Type: image/x-rgbz\r\n", stdout);
unsigned long zip_buffer_size;
send_image->Zip(img_buffer, &zip_buffer_size);
img_buffer_size = zip_buffer_size;
#else
Error("zlib is required for zipped images. Falling back to raw image");
type = STREAM_RAW;
#endif // HAVE_ZLIB_H
break;
default :
Error("Unexpected frame type %d", type);
return false;
}
if (
(0 > fprintf(stdout, "Content-Length: %zu\r\nX-Timestamp: %.6f\r\n\r\n",
img_buffer_size, std::chrono::duration_cast<FPSeconds>(timestamp.time_since_epoch()).count()))
||
(fwrite(img_buffer, img_buffer_size, 1, stdout) != 1)
) {
// If the browser disconnected, SIGPIPE (handled by zm_pipe_handler in
// zms) sets zm_terminate; the fwrite above also returned an error with
// errno=EPIPE, which we log below before the loop exits cleanly.
Debug(1, "Unable to send stream frame: %s, zm_terminate: %d", strerror(errno), zm_terminate.load());
return false;
}
fputs("\r\n", stdout);
fflush(stdout);
} // Not mpeg
last_frame_sent = std::chrono::steady_clock::now();
if (maxfps > 0.0) {
TimePoint::duration frame_send_time = last_frame_sent - send_start_time;
TimePoint::duration maxfps_milliseconds = Milliseconds(lround(Milliseconds::period::den / maxfps));
if (frame_send_time > maxfps_milliseconds) {
//maxfps /= 1.5;
Debug(1, "Frame send time %" PRIi64 " msec too slow (> %" PRIi64 ", %.3f",
static_cast<int64>(std::chrono::duration_cast<Milliseconds>(frame_send_time).count()),
static_cast<int64>(std::chrono::duration_cast<Milliseconds>(maxfps_milliseconds).count()),
maxfps.load());
}
}
return true;
} // end bool MonitorStream::sendFrame(Image *image, SystemTimePoint timestamp)
void MonitorStream::runStream() {
if (type == STREAM_SINGLE) {
Debug(1, "Single");
if (!checkInitialised()) {
if (!loadMonitor(monitor_id)) {
sendTextFrame("Not connected");
} else if (monitor->Deleted()) {
sendTextFrame("Monitor has been deleted");
} else if (monitor->Capturing() == Monitor::CAPTURING_ONDEMAND) {
// Notify capture that we might want to view
monitor->setLastViewed();
sendTextFrame("Waiting for capture");
} else if (monitor->Decoding() == Monitor::DECODING_NONE) {
sendTextFrame("Monitor has Decoding==None. We will not be able to provide live stream.");
} else {
sendTextFrame("Unable to stream");
}
} else {
// Not yet migrated over to stream class
SingleImage(scale);
}
zm_terminate = true;
return;
}
if (!monitor) {
Error("Cannot stream monitor %d: not loaded", monitor_id);
if (type == STREAM_JPEG)
fputs("Content-Type: multipart/x-mixed-replace; boundary=" BOUNDARY "\r\n\r\n", stdout);
sendTextFrame("Not connected");
zm_terminate = true;
return;
}
openComms();
// Declared here so it stays in scope for the join() at the end of the
// stream loop, but not started until the temporary buffer state below has
// been initialised (see the launch after the playback_buffer setup).
std::thread command_processor;
if (type == STREAM_JPEG)
fputs("Content-Type: multipart/x-mixed-replace; boundary=" BOUNDARY "\r\n\r\n", stdout);
updateFrameRate(monitor->GetFPS());
// point to end which is theoretically not a valid value because all indexes are % image_buffer_count
int32_t last_read_index = monitor->image_buffer_count;
int32_t last_image_count = 0;
TimePoint stream_start_time = std::chrono::steady_clock::now();
when_to_send_next_frame = stream_start_time; // initialize it to now so that we spit out a frame immediately
// Periodic RSS trace so a runaway nph-zms can be caught in the act. refs #5006
TimePoint last_mem_report = stream_start_time;
const Seconds mem_report_interval = Seconds(30);
temp_image_buffer_count = playback_buffer;
temp_read_index = temp_image_buffer_count.load();
temp_write_index = temp_image_buffer_count.load();
std::string swap_path;
bool buffered_playback = false;
// Last image and timestamp when paused, will be resent occasionally to prevent timeout.
// unique_ptr because the loop below has break paths that leave the function
// while one of these is still held.
std::unique_ptr<Image> paused_image;
SystemTimePoint paused_timestamp;
// Same, for the stopped state. See the stopped branch in the loop below.
std::unique_ptr<Image> stopped_image;
SystemTimePoint stopped_timestamp;
if (connkey && (playback_buffer > 0)) {
// 15 is the max length for the swap path suffix, /zmswap-whatever, assuming max 6 digits for monitor id
const int max_swap_len_suffix = 15;
int swap_path_length = staticConfig.PATH_SWAP.length() + 1; // +1 for NULL terminator
int subfolder1_length = snprintf(nullptr, 0, "/zmswap-m%u", monitor->Id()) + 1;
int subfolder2_length = snprintf(nullptr, 0, "/zmswap-q%06d", connkey) + 1;
int total_swap_path_length = swap_path_length + subfolder1_length + subfolder2_length;
if (total_swap_path_length + max_swap_len_suffix > PATH_MAX) {
Error("Swap Path is too long. %d > %d ", total_swap_path_length+max_swap_len_suffix, PATH_MAX);
} else {
swap_path = staticConfig.PATH_SWAP;
Debug(3, "Checking swap path folder: %s", swap_path.c_str());
if (checkSwapPath(swap_path.c_str(), true)) {
swap_path += stringtf("/zmswap-m%d", monitor->Id());
Debug(4, "Checking swap path subfolder: %s", swap_path.c_str());
if (checkSwapPath(swap_path.c_str(), true)) {
swap_path += stringtf("/zmswap-q%06d", connkey);
Debug(4, "Checking swap path subfolder: %s", swap_path.c_str());
if (checkSwapPath(swap_path.c_str(), true)) {
buffered_playback = true;
}
}
}
if (!buffered_playback) {
Error("Unable to validate swap image path, disabling buffered playback");
} else {
Debug(2, "Assigning temporary buffer");
temp_image_buffer = std::make_unique<SwapImage[]>(temp_image_buffer_count);
Debug(2, "Assigned temporary buffer");
}
}
} else {
Debug(2, "Not using playback_buffer");
} // end if connkey && playback_buffer
// Start the command processor only now that temp_image_buffer_count, the
// read/write indices and temp_image_buffer are initialised, so a command
// arriving early cannot observe a half-initialised stream (the original
// cause of the divide-by-zero in issue #4936).
if (connkey) {
command_processor = std::thread(&MonitorStream::checkCommandQueue, this);
}
while (!zm_terminate) {
if (feof(stdout)) {
Debug(2, "feof stdout");
zm_terminate = true;
break;
} else if (ferror(stdout)) {
Debug(2, "ferror stdout");
zm_terminate = true;
break;
}
now = std::chrono::steady_clock::now();
if (now - last_mem_report >= mem_report_interval) {
last_mem_report = now;
Debug(1, "mem trace m%d: rss=%zuKB frame_count=%d buffer_level=%d temp_count=%d paused=%d delayed=%d",
monitor_id, zm_get_rss_kb(), frame_count,
MonitorStreamBufferLevel(temp_write_index.load(), temp_read_index.load(), temp_image_buffer_count.load()),
temp_image_buffer_count.load(), paused.load(), delayed.load());
}
bool was_paused = paused;
if (!checkInitialised()) {
int rc = -1;
if (!loadMonitor(monitor_id)) {
rc = sendTextFrame("Not connected");
} else if (monitor->Deleted()) {
rc = sendTextFrame("Monitor has been deleted");
zm_terminate = true;
} else if (monitor->Capturing() == Monitor::CAPTURING_ONDEMAND) {
monitor->setLastViewed();
rc = sendTextFrame("Waiting for capture");
} else if (monitor->Decoding() == Monitor::DECODING_NONE) {
rc = sendTextFrame("Monitor has Decoding==None. We will not be able to provide a live image");
} else {
rc = sendTextFrame("Unable to stream");
}
if (rc <= 0) {
Debug(1, "Failed Send unable to stream. rc=%d", rc);
zm_terminate = true;
continue;
}
// Timeout if we've never received a frame and have been waiting too long.
// Use ttl if set, otherwise default to 60 seconds.
Seconds wait_timeout = (ttl > Seconds(0)) ? std::chrono::duration_cast<Seconds>(ttl) : Seconds(60);
if (now - stream_start_time > wait_timeout) {
logPrintf(Logger::WARNING + monitor->Importance(), "Timed out waiting for capture daemon after %" PRIi64 " seconds. ttl is %" PRIi64,
static_cast<int64>(std::chrono::duration_cast<Seconds>(now - stream_start_time).count()),
static_cast<int64>(std::chrono::duration_cast<Seconds>(wait_timeout).count())
);
zm_terminate = true;
continue;
}
std::this_thread::sleep_for(MAX_SLEEP);
continue;
}
if (stopped) {
// Stopped halts playback but keeps the connection, so that a later
// command can resume on the same process. Hold the last frame and
// re-send it every few seconds, as the paused state does.
//
// The write is the point: with nothing written, this branch never
// touches the socket, so a stream whose client has gone away is never
// told about it and never sees EPIPE. Since it also skipped the ttl
// check at the bottom of the loop, such a process had no way to exit at
// all and stayed until the machine was restarted. refs #4706
//
// setLastViewed is deliberately still not called: capture and decoding
// should not be held active for a stream that is not playing.
if (!stopped_image
&& monitor->shared_data->valid
&& (monitor->shared_data->last_write_index != monitor->image_buffer_count)) {
int index = monitor->shared_data->last_write_index % monitor->image_buffer_count;
Debug(1, "Saving stopped image from index %d", index);
stopped_image = zm::make_unique<Image>(*monitor->ReadShmFrame(index));
stopped_timestamp = SystemTimePoint(zm::chrono::duration_cast<Microseconds>(monitor->shared_timestamps[index]));
}
if (now - last_frame_sent > Seconds(5)) {
if (stopped_image) {
Debug(2, "Sending keepalive frame while stopped");
if (!sendFrame(stopped_image.get(), stopped_timestamp)) zm_terminate = true;
} else if (sendTextFrame("Stopped") <= 0) {
// Nothing captured yet to hold, but we still have to write something
// to notice a client that is no longer there.
Debug(2, "Failed to send stopped keepalive text frame");
zm_terminate = true;
}
}
// The ttl check lives at the bottom of the loop, which this branch never
// reaches, so a stopped stream would otherwise outlive its own deadline.
if (ttl > Seconds(0) && (now - stream_start_time) > ttl) {
Debug(2, "Stopped and now - start > ttl. break");
// checkCommandQueue() only returns when zm_terminate is set, and
// runStream() joins it below, so leaving the loop without setting it
// hangs in join() forever. See the two breaks at the bottom of the loop.
zm_terminate = true;
break;
}
std::this_thread::sleep_for(MAX_SLEEP);
continue;
}
stopped_image.reset();
monitor->setLastViewed();
if (frame_type == FRAME_ANALYSIS)
monitor->setLastAnalysisViewed();
if (paused) {
if (!was_paused) {
int index = monitor->shared_data->last_write_index % monitor->image_buffer_count;
Debug(1, "Saving paused image from index %d",index);
paused_image = zm::make_unique<Image>(*monitor->ReadShmFrame(index));
paused_timestamp = SystemTimePoint(zm::chrono::duration_cast<Microseconds>(monitor->shared_timestamps[index]));
}
} else {
paused_image.reset();
}
if (buffered_playback && delayed) {
if (temp_read_index == temp_write_index) {
// Go back to live viewing
Debug(1, "Exceeded temporary streaming buffer");
paused = false;
delayed = false;
replay_rate = ZM_RATE_BASE;
} else {
if (!paused) {
int temp_index = MOD_ADD(temp_read_index, 0, temp_image_buffer_count);
SwapImage *swap_image = &temp_image_buffer[temp_index];
if (!swap_image->valid) {
paused = true;
delayed = true;
temp_read_index = MOD_ADD(temp_read_index, (replay_rate>=0?-1:1), temp_image_buffer_count);
} else {
FPSeconds expected_delta_time = ((FPSeconds(swap_image->timestamp - last_frame_timestamp)) * ZM_RATE_BASE) / replay_rate.load();
TimePoint::duration actual_delta_time = now - last_frame_sent;
// If the next frame is due
if (actual_delta_time > expected_delta_time) {
// Debug( 2, "eDT: %.3lf, aDT: %.3f", expected_delta_time, actual_delta_time );
if ((temp_index % frame_mod) == 0) {
Debug(2, "Sending delayed frame %d", temp_index);
// Send the next frame
if (!sendFrame(temp_image_buffer[temp_index].file_name, temp_image_buffer[temp_index].timestamp)) {
zm_terminate = true;
}
frame_count++;
last_frame_timestamp = swap_image->timestamp;
// frame_sent = true;
}
temp_read_index = MOD_ADD(temp_read_index, (replay_rate > 0 ? 1 : -1), temp_image_buffer_count);
}
}
} else if (step != 0) {
temp_read_index = MOD_ADD(temp_read_index, (step>0?1:-1), temp_image_buffer_count);
SwapImage *swap_image = &temp_image_buffer[temp_read_index];
// Send the next frame
if (!sendFrame(
temp_image_buffer[temp_read_index].file_name,
temp_image_buffer[temp_read_index].timestamp)
) {
zm_terminate = true;
}
frame_count++;
last_frame_timestamp = swap_image->timestamp;
// frame_sent = true;
step = 0;
} else {
//paused?
int temp_index = MOD_ADD(temp_read_index, 0, temp_image_buffer_count);
if (got_command || (now - last_frame_sent > Seconds(5))) {
// Send keepalive
Debug(2, "Sending keepalive frame %d", temp_index);
// Send the next frame
if (!sendFrame(temp_image_buffer[temp_index].file_name, temp_image_buffer[temp_index].timestamp)) {
zm_terminate = true;
}
frame_count++;
// frame_sent = true;
}
} // end if (!paused) or step or paused
} // end if have exceeded buffer or not
if (temp_read_index == temp_write_index) {
// Go back to live viewing
Warning("Rewound over write index, resuming live play");
// Clear paused flag
paused = false;
// Clear delayed_play flag
delayed = false;
replay_rate = ZM_RATE_BASE;
}
} // end if (buffered_playback && delayed)
if (last_read_index != monitor->shared_data->last_write_index || last_image_count < monitor->shared_data->image_count) {
// have a new image to send
int last_write_index = monitor->shared_data->last_write_index;
int index = last_write_index % monitor->image_buffer_count;
//if ((frame_mod == 1) || ((frame_count%frame_mod) == 0)) {
if ( now >= when_to_send_next_frame ) {
if (!paused && !delayed) {
Debug(2, "Sending frame index: %d(%d%%%d): frame_mod: %d frame count: %d last image count %d image count %d paused %d delayed %d",
index, last_write_index, monitor->image_buffer_count, frame_mod, frame_count, last_image_count, monitor->shared_data->image_count, paused.load(), delayed.load());
last_read_index = last_write_index;
last_image_count = monitor->shared_data->image_count;
// Send the next frame
//
// Perhaps we should use NOW instead.
last_frame_timestamp =
SystemTimePoint(zm::chrono::duration_cast<Microseconds>(monitor->shared_timestamps[index]));
Image *send_image = nullptr;
if ((frame_type == FRAME_ANALYSIS) &&
(monitor->Analysing() != Monitor::ANALYSING_NONE)) {
Debug(1, "Sending analysis image");
send_image = monitor->GetAlarmImage();
if (!send_image) {
Debug(1, "Falling back");
send_image = monitor->ReadShmFrame(index);
}
} else {
//AVPixelFormat pixformat = monitor->image_pixelformats[index];
//Debug(1, "Sending regular image index %d, pix format is %d %s", index, pixformat, av_get_pix_fmt_name(pixformat));
send_image = monitor->ReadShmFrame(index);
}
if (!sendFrame(send_image, last_frame_timestamp)) {
Debug(2, "sendFrame failed, quitting.");
zm_terminate = true;
break;
}
frame_count++;
if (frame_count == 0) {
// Chrome will not display the first frame until it receives another.
// Firefox is fine. So just send the first frame twice.
if (!sendFrame(send_image, last_frame_timestamp)) {
Debug(2, "sendFrame failed, quitting.");
zm_terminate = true;
break;
}
}
temp_read_index = temp_write_index.load();
} else {
if (delayed.load() && !buffered_playback) {
Debug(2, "Can't delay when not buffering.");
delayed = false;
}
if (last_zoom != zoom) {
Debug(2, "Sending 2 frames because change in zoom %d ?= %d", last_zoom, zoom.load());
if (!sendFrame(paused_image.get(), paused_timestamp))
zm_terminate = true;
if (!sendFrame(paused_image.get(), paused_timestamp))
zm_terminate = true;
frame_count++;
frame_count++;
} else {
TimePoint::duration actual_delta_time = now - last_frame_sent;
if (actual_delta_time > Seconds(5)) {
if (paused_image) {
// Send keepalive
Debug(2, "Sending keepalive frame because delta time %.2f s > 5 s",
FPSeconds(actual_delta_time).count());
// Send the next frame
if (!sendFrame(paused_image.get(), paused_timestamp))
zm_terminate = true;
frame_count++;
} else {
Debug(2, "Would have sent keepalive frame, but had no paused_image");
}
} // end if actual_delta_time > 5
} // end if change in zoom
} // end if paused or not
//} else {
//frame_count++;
} else {
Debug(2, "Not time to send next frame.");
} // end if should send frame now > when_to_send_next_frame
if (buffered_playback && !paused) {
if (monitor->shared_data->valid) {
if (monitor->shared_timestamps[index].tv_sec) {
int temp_index = temp_write_index%temp_image_buffer_count;
Debug(2, "Storing frame %d", temp_index);
if ( !temp_image_buffer[temp_index].valid ) {
temp_image_buffer[temp_index].file_name = stringtf("%s/zmswap-i%05d.jpg", swap_path.c_str(), temp_index);
temp_image_buffer[temp_index].valid = true;
}
temp_image_buffer[temp_index].timestamp =
SystemTimePoint(zm::chrono::duration_cast<Microseconds>(monitor->shared_timestamps[index]));
monitor->ReadShmFrame(index)->WriteJpeg(temp_image_buffer[temp_index].file_name, config.jpeg_file_quality);
temp_write_index = MOD_ADD(temp_write_index, 1, temp_image_buffer_count);
if (temp_write_index == temp_read_index) {
// Go back to live viewing
Warning("Exceeded temporary buffer, resuming live play");
paused = false;
delayed = false;
replay_rate = ZM_RATE_BASE;
}
} else {
Warning("Unable to store frame as timestamp invalid");
}
} else {
Warning("Unable to store frame as shared memory invalid");
}
} // end if buffered playback
} else {
Debug(3, "Waiting for capture last_write_index=%u == last_read_index=%u",
monitor->shared_data->last_write_index,
last_read_index);
if (now - last_frame_sent > Seconds(5)) {
if (last_read_index == monitor->GetImageBufferCount()) {
sendTextFrame("Waiting for initial capture");
// Timeout if we've never received a frame from capture
Seconds wait_timeout = (ttl > Seconds(0)) ? std::chrono::duration_cast<Seconds>(ttl) : Seconds(60);
if (now - stream_start_time > wait_timeout) {
logPrintf(Logger::WARNING + monitor->Importance(), "Timed out waiting for initial capture after %" PRIi64 " seconds",
static_cast<int64>(std::chrono::duration_cast<Seconds>(now - stream_start_time).count()));
zm_terminate = true;
continue;
}
} else {
sendTextFrame("Waiting for capture");
}
}
} // end if ( (unsigned int)last_read_index != monitor->shared_data->last_write_index )
FPSeconds sleep_time;
if (now >= when_to_send_next_frame) {
// sent a frame, so update
// Snapshot both once: the command thread can change them at any point,
// and re-reading mid-expression could mix an old value with a new one.
const double max_fps = maxfps.load();
const int rate = replay_rate.load();
double capture_fps = monitor->GetFPS();
double fps = ((max_fps > 0.0) && (capture_fps > max_fps)) ? max_fps : capture_fps;
double sleep_time_seconds = (1 / ((fps ? fps : 1))) // 1 second / fps
* (rate ? abs(rate)/ZM_RATE_BASE : 1); // replay_rate is 100 for 1x
Debug(3, "Using %f for maxfps. capture_fps: %f maxfps %f * replay_rate: %d = %f", fps, capture_fps, max_fps, rate, sleep_time_seconds);
sleep_time = FPSeconds(sleep_time_seconds);
if (when_to_send_next_frame > now) {
sleep_time -= (when_to_send_next_frame - now);
Debug(2, "Adjusting sleep time for when_to_send_next_frame - now = %f", FPSeconds(when_to_send_next_frame - now).count());
}
if (last_frame_sent > now) {
FPSeconds elapsed = last_frame_sent - now;
if (sleep_time > elapsed) {
Debug(2, "Adjusting sleep time by %f elapsed", elapsed.count());
sleep_time -= elapsed;
}
}
when_to_send_next_frame = now + std::chrono::duration_cast<Microseconds>(sleep_time);
} else {
sleep_time = when_to_send_next_frame - now;
}
if (sleep_time > MonitorStream::MAX_SLEEP) {
Debug(3, "Sleeping for MAX_SLEEP_USEC instead of %" PRIi64 " us",
static_cast<int64>(std::chrono::duration_cast<Microseconds>(sleep_time).count()));
// Shouldn't sleep for long because we need to check command queue, etc.
sleep_time = MonitorStream::MAX_SLEEP;
} else {
Debug(3, "Sleeping for %" PRIi64 " us",
static_cast<int64>(std::chrono::duration_cast<Microseconds>(sleep_time).count()));
}
if (!zm_terminate)
std::this_thread::sleep_for(sleep_time);
// Both of these have to set zm_terminate, not just break: checkCommandQueue()
// loops until that flag is set and runStream() joins it below, so a stream
// that ended on its ttl or its frame count would sit in join() forever
// rather than exiting. That is a way for a zms to outlive its client. refs #4706
if (ttl > Seconds(0) && (now - stream_start_time) > ttl) {
Debug(2, "now - start > ttl (%" PRIi64 " us). break",
static_cast<int64>(std::chrono::duration_cast<Microseconds>(ttl).count()));
zm_terminate = true;
break;
}
if (frames_to_send > 0 && frame_count >= frames_to_send) {
zm_terminate = true;
break;
}
} // end while ! zm_terminate
if (buffered_playback) {
Debug(1, "Cleaning swap files from %s", swap_path.c_str());
struct stat stat_buf = {};
if (stat(swap_path.c_str(), &stat_buf) < 0) {
if (errno != ENOENT) {
Error("Can't stat '%s': %s", swap_path.c_str(), strerror(errno));
}
} else if (!S_ISDIR(stat_buf.st_mode)) {
Error("Swap image path '%s' is not a directory", swap_path.c_str());
} else {
std::string glob_pattern = stringtf("%s/*.*", swap_path.c_str());
glob_t pglob;
int glob_status = glob(glob_pattern.c_str(), 0, 0, &pglob);
if (glob_status != 0) {
if (glob_status < 0) {
Error("Can't glob '%s': %s", glob_pattern.c_str(), strerror(errno));
} else {
Debug(1, "Can't glob '%s': %d", glob_pattern.c_str(), glob_status);
}
} else {
for (unsigned int i = 0; i < pglob.gl_pathc; i++) {
if (unlink(pglob.gl_pathv[i]) < 0) {
Error("Can't unlink '%s': %s", pglob.gl_pathv[i], strerror(errno));
}
}
}
globfree(&pglob);
if (rmdir(swap_path.c_str()) < 0) {
Error("Can't rmdir '%s': %s", swap_path.c_str(), strerror(errno));
}
} // end if checking for swap_path
} // end if buffered_playback
if (zm_terminate)
Debug(1, "zm_terminate");
if (connkey) {
if (command_processor.joinable()) {
Debug(1, "command_processor is joinable");
command_processor.join();
} else {
Debug(1, "command_processor is not joinable");
}
}
Debug(1, "command_processor has joined");
} // end MonitorStream::runStream
void MonitorStream::SingleImage(int scale) {
size_t img_buffer_size = 0;
static JOCTET img_buffer[ZM_MAX_IMAGE_SIZE];
Image scaled_image;
int count = 10; // Give it 1 second to connect or else send text frame.
while (count and (monitor->shared_data->last_write_index >= monitor->image_buffer_count) and !zm_terminate) {
Debug(1, "Waiting for capture to begin. last write index %d >=? %d",
monitor->shared_data->last_write_index, monitor->image_buffer_count);
std::this_thread::sleep_for(Milliseconds(100));
count--;
}
if (!count) {
sendTextFrame("No image available.");
return;
}
int index = monitor->shared_data->last_write_index % monitor->image_buffer_count;
AVPixelFormat pixformat = monitor->image_pixelformats[index];
Debug(1, "Sending regular image index %d, pix format is %d %s", index, pixformat, zm_get_pix_fmt_name(pixformat));
Image *snap_image = monitor->ReadShmFrame(index);
if (!config.timestamp_on_capture) {
monitor->TimestampImage(snap_image,
SystemTimePoint(zm::chrono::duration_cast<Microseconds>(monitor->shared_timestamps[index])));
}
if ( scale != ZM_SCALE_BASE ) {
scaled_image.Assign(*snap_image);
scaled_image.Scale(scale);
snap_image = &scaled_image;
}
snap_image->EncodeJpeg(img_buffer, &img_buffer_size);
fprintf(stdout,
"Content-Length: %zu\r\n"
"Content-Type: image/jpeg\r\n\r\n",
img_buffer_size);
fwrite(img_buffer, img_buffer_size, 1, stdout);
} // end void MonitorStream::SingleImage(int scale)
void MonitorStream::SingleImageRaw(int scale) {
Image scaled_image;
std::shared_ptr<ZMPacket> snap = monitor->getSnapshot();
Image *snap_image = snap->image;
if ( scale != ZM_SCALE_BASE ) {
scaled_image.Assign(*snap_image);
scaled_image.Scale(scale);
snap_image = &scaled_image;
}
if ( !config.timestamp_on_capture ) {
monitor->TimestampImage(snap_image, snap->timestamp);
}
fprintf(stdout,
"Content-Length: %u\r\n"
"Content-Type: image/x-rgb\r\n\r\n",
snap_image->Size());
fwrite(snap_image->Buffer(), snap_image->Size(), 1, stdout);
} // end void MonitorStream::SingleImageRaw(int scale)
#ifdef HAVE_ZLIB_H
void MonitorStream::SingleImageZip(int scale) {
unsigned long img_buffer_size = 0;
static Bytef img_buffer[ZM_MAX_IMAGE_SIZE];
Image scaled_image;
std::shared_ptr<ZMPacket> snap = monitor->getSnapshot();
Image *snap_image = snap->image;
if ( scale != ZM_SCALE_BASE ) {
scaled_image.Assign(*snap_image);
scaled_image.Scale(scale);
snap_image = &scaled_image;
}
if ( !config.timestamp_on_capture ) {
monitor->TimestampImage(snap_image, snap->timestamp);
}
snap_image->Zip(img_buffer, &img_buffer_size);
fprintf(stdout,
"Content-Length: %lu\r\n"
"Content-Type: image/x-rgbz\r\n\r\n",
img_buffer_size);
fwrite(img_buffer, img_buffer_size, 1, stdout);
} // end void MonitorStream::SingleImageZip(int scale)
#endif // HAVE_ZLIB_H
Reference in new issue View Git Blame Copy Permalink