src/zm_stream.cpp
Rama master · réplica de lectura del repositorio original.
//
// ZoneMinder Stream 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_stream.h"
#include "zm_box.h"
#include "zm_monitor.h"
#include "zm_signal.h"
#include <cmath>
#include <sys/file.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <unistd.h>
constexpr Seconds StreamBase::MAX_STREAM_DELAY;
constexpr Milliseconds StreamBase::MAX_SLEEP;
StreamBase::~StreamBase() {
delete vid_stream;
delete[] temp_img_buffer;
closeComms();
// The monitor's zones each hold a shared_ptr back to it, so releasing ours
// alone never frees it. disconnect() clears the zones, as zmc does on exit.
if (monitor) monitor->disconnect();
}
bool StreamBase::loadMonitor(int p_monitor_id) {
monitor_id = p_monitor_id;
if ( !(monitor or (monitor = Monitor::Load(monitor_id, false, Monitor::QUERY))) ) {
Error("Unable to load monitor id %d for streaming", monitor_id);
return false;
}
if (monitor->Capturing() == Monitor::CAPTURING_NONE) {
Info("Monitor %d has capturing == NONE. Will not be able to connect to it.", monitor_id);
return false;
}
// Hold monitor_mutex during disconnect/connect to prevent the command
// processor thread from accessing shared memory while it is remapped.
std::lock_guard<std::mutex> lck(monitor_mutex);
if (monitor->isConnected()) {
monitor->disconnect();
}
if (!monitor->connect()) {
Info("Unable to connect to monitor id %d for streaming", monitor_id);
monitor->disconnect();
// If we couldn't connect, it might be due to size mismatch in shm. Need to reload
if ( !(monitor = Monitor::Load(monitor_id, false, Monitor::QUERY)))
Error("Unable to reload monitor id %d for streaming", monitor_id);
return false;
}
return true;
}
bool StreamBase::checkInitialised() {
if (!monitor) {
Error("Cannot stream, not initialised");
return false;
}
if (monitor->Capturing() == Monitor::CAPTURING_NONE) {
Info("Monitor %d has capturing == NONE. Will not be able to connect to it.", monitor_id);
return false;
}
if (!monitor->ShmValid()) {
Debug(1, "Monitor shm is not connected");
return false;
}
if ((monitor->GetType() == Monitor::FFMPEG) and (monitor->Decoding() == Monitor::DECODING_NONE) ) {
Debug(1, "Monitor is not decoding.");
return false;
}
return true;
}
void StreamBase::updateFrameRate(double fps) {
if ( (fps < 0) || !fps || std::isinf(fps) ) {
Debug(1, "Zero or negative fps %f in updateFrameRate. Setting frame_mod=1 and effective_fps=0.0", fps);
effective_fps = 0.0;
base_fps = 0.0;
return;
}
// Snapshot once so the loop below cannot see maxfps change underneath it.
const double max_fps = maxfps.load();
const int rate = replay_rate.load();
base_fps = fps;
effective_fps = (base_fps*abs(rate))/ZM_RATE_BASE;
frame_mod = 1;
Debug(3, "FPS:%.2f, MaxFPS:%.2f, BaseFPS:%.2f, EffectiveFPS:%.2f, FrameMod:%d, replay_rate(%d)",
fps, max_fps, base_fps, effective_fps, frame_mod, rate);
if (max_fps > 0.0) {
// Min frame repeat?
// We want to keep the frame skip easy... problem is ... if effective = 31 and max = 30 then we end up with 15.5 fps.
while ( (int)effective_fps > (int)max_fps ) {
effective_fps /= 2.0;
frame_mod *= 2;
Debug(3, "Changing fps to be < max %.2f EffectiveFPS:%.2f, FrameMod:%d",
max_fps, effective_fps, frame_mod);
}
}
} // void StreamBase::updateFrameRate(double fps)
void StreamBase::checkCommandQueue() {
while (!zm_terminate) {
// Update modified time of the socket .lock file so that we can tell which ones are stale.
// This is the command thread: read the clock here rather than the
// streaming loop's `now`.
TimePoint cmd_now = std::chrono::steady_clock::now();
if (cmd_now - last_comm_update > Hours(1)) {
touch(sock_path_lock);
last_comm_update = cmd_now;
}
if (sd >= 0) {
CmdMsg msg;
memset(&msg, 0, sizeof(msg));
int nbytes = recvfrom(sd, &msg, sizeof(msg), 0, /*MSG_DONTWAIT*/ 0, 0);
if (nbytes < 0) {
if (errno != EAGAIN) {
Error("recvfrom(), errno = %d, error = %s", errno, strerror(errno));
}
} else {
Debug(2, "Message length is (%d)", nbytes);
processCommand(&msg);
// A status query doesn't change the picture, so it doesn't need a frame.
if (msg.msg_data[0] != CMD_QUERY) got_command = true;
}
} else if (connkey) {
Warning("No sd in checkCommandQueue, comms not open for connkey %06d?", connkey);
} else {
// Perfectly valid if only getting a snapshot
Debug(1, "No sd in checkCommandQueue, comms not open.");
}
} // end while !zm_terminate
// Nothing reads the socket from here on, but the stream thread can take a
// while to wind down (blocked writing to a client that has gone away, say),
// and the destructor that would otherwise close it runs only after that.
// Until then a command sent to us is accepted and never answered, so
// web/ajax/stream.php sits out its whole MSG_TIMEOUT before reporting a
// timeout. Closing now makes it see a missing socket and report no_socket
// straight away. sd is only used from this thread, and closeComms() leaves
// nothing for the destructor's call to do.
closeComms();
} // end void StreamBase::checkCommandQueue()
int StreamBase::preScaleDimensions(int base_width, int base_height, int &width, int &height) const {
// Any zoom crops before scaling and needs the full resolution to crop out of.
if (zoom.load() != ZM_SCALE_BASE) return 0;
const int cur_scale = scale.load();
if (cur_scale == ZM_SCALE_BASE) return 0;
const int scaled_width = (base_width * cur_scale) / ZM_SCALE_BASE;
const int scaled_height = (base_height * cur_scale) / ZM_SCALE_BASE;
// CMD_SCALE stores whatever the client sent, so 0 gets here. Image::Scale
// refuses a zero factor, but converting straight to 0x0 would instead put a
// negative av_image_get_buffer_size result into an unsigned size and allocate
// on it, so leave anything degenerate to the path that already rejects it.
if ((scaled_width <= 0) or (scaled_height <= 0)) return 0;
width = scaled_width;
height = scaled_height;
return cur_scale;
}
Image *StreamBase::prepareImage(Image *image, int pre_scaled_by) {
if (pre_scaled_by) {
// The caller produced the image at exactly the size to send, folding the
// scale into a colour conversion it had to do anyway. Send it untouched.
//
// The command thread can change scale or zoom between the caller reading
// them and this call, in which case this one frame goes out at the scale it
// was built for and without any new zoom, and the next frame is built from
// the new values. Scaling here instead would apply the scale a second time
// (a 50% frame of a 640px source would go out at 160px, not 320px), and
// cropping here would do it in the reduced image's coordinates and leave
// last_crop in the wrong units for every later frame. refs #3681
Debug(3, "Sending pre-scaled %dx%d, scale %d", image->Width(), image->Height(), pre_scaled_by);
last_scale = pre_scaled_by;
last_zoom = ZM_SCALE_BASE;
return image;
}
/* zooming should happen before scaling to preserve quality
* scale is relative to base dimensions, and represents the rough ratio between desired view size and base dimensions
*/
bool image_copied = false;
// Snapshot the pan/zoom/scale state once. The command thread can change any
// of these at any point, and the crop below is derived from all three
// together, so re-reading them mid-calculation could mix a new zoom with an
// old click position and produce a crop that matches neither.
const int cur_zoom = zoom.load();
const int cur_scale = scale.load();
const unsigned short cur_x = x.load();
const unsigned short cur_y = y.load();
if (cur_zoom != 100) {
int base_image_width = image->Width(),
base_image_height = image->Height(),
disp_image_width = image->Width() * cur_scale/ZM_SCALE_BASE,
disp_image_height = image->Height() * cur_scale / ZM_SCALE_BASE;
/* x and y are scaled by web UI to base dimensions units.
* When zooming, we blow up the image by the amount 150 for first zoom, right? 150%, then cut out a base sized chunk
* However if we have zoomed before, then we are zooming into the previous cutout
* The box stored in last_crop should be in base_image units, So we need to turn x,y into percentages, then apply to last_crop
*/
if (!last_crop.Hi().x_ && !last_crop.Hi().y_) {
last_crop = Box({0, 0}, {base_image_width, base_image_height});
} else if (last_crop_width && (last_crop_width != base_image_width || last_crop_height != base_image_height)) {
// Keep the same part of the picture in the new frame's coordinates.
Debug(2, "Rescaling crop from %dx%d to %dx%d", last_crop_width, last_crop_height,
base_image_width, base_image_height);
last_crop = Box({last_crop.Lo().x_ * base_image_width / last_crop_width,
last_crop.Lo().y_ * base_image_height / last_crop_height},
{last_crop.Hi().x_ * base_image_width / last_crop_width,
last_crop.Hi().y_ * base_image_height / last_crop_height});
}
double x_percent = static_cast<double>(cur_x * ZM_SCALE_BASE) / base_image_width;
double y_percent = static_cast<double>(cur_y * ZM_SCALE_BASE) / base_image_height;
Debug(2, "click percent %dx%d => %.2fx%.2f", cur_x, cur_y, x_percent, y_percent);
// If we were previously zoomed in, then the coordinate percentages are into the crop, so calculate the click coordinates in base image
int crop_x = last_crop.Lo().x_ + (x_percent * last_crop.Width() / ZM_SCALE_BASE);
int crop_y = last_crop.Lo().y_ + (y_percent * last_crop.Height() / ZM_SCALE_BASE);
Debug(2, "crop click %dx%d => %dx%d out of %dx%d", cur_x, cur_y, crop_x, crop_y, last_crop.Width(), last_crop.Height());
int zoom_image_width = base_image_width * cur_zoom / ZM_SCALE_BASE,
zoom_image_height = base_image_height * cur_zoom / ZM_SCALE_BASE,
click_x = crop_x * cur_zoom / ZM_SCALE_BASE,
click_y = crop_y * cur_zoom / ZM_SCALE_BASE;
Debug(2, "adjusted click %dx%d * %d zoom => %dx%d out of %dx%d", cur_x, cur_y, cur_zoom, click_x, click_y, zoom_image_width, zoom_image_height);
// These can go out of image. Resulting size will be less than base image. That's ok.
// We don't want to center it, we want to keep the relative offset from center where the click is.
int left_dist = base_image_width * x_percent/ZM_SCALE_BASE;
int top_dist = base_image_height * y_percent/ZM_SCALE_BASE;
Debug(2, "Dest at %d,%d", left_dist, top_dist);
int lo_x = click_x - left_dist;
int hi_x = lo_x + base_image_width;
Debug(2, "hi_x = lo_x %d + base_image_width %d - left_dist %d = %d", lo_x, base_image_width, left_dist, hi_x);
int lo_y = click_y - top_dist;
int hi_y = lo_y + base_image_height;
Debug(2, "hi_y = lo_y %d + base_image_h %d - top_dist %d = %d", lo_y, base_image_height, top_dist, hi_y);
int amount_to_shrink_y = 0;
if (lo_x < 0) {
amount_to_shrink_y = (-1 * lo_x) * base_image_height / base_image_width;
lo_x = 0;
} else if (hi_x >= zoom_image_width) {
amount_to_shrink_y = (hi_x - zoom_image_width) * base_image_height / base_image_width;
hi_x = zoom_image_width - 1;
}
Debug(1, "Shrinking y by %d from %d,%d to %d,%d", amount_to_shrink_y, lo_y, hi_y, lo_y+(amount_to_shrink_y/2), hi_y-(amount_to_shrink_y/2));
if (amount_to_shrink_y) {
lo_y += amount_to_shrink_y/2;
hi_y -= amount_to_shrink_y/2;
}
int amount_to_shrink_x = 0;
if (lo_y < 0) {
amount_to_shrink_x = (-1 * lo_y) * base_image_width / base_image_height;
Debug(1, "%d to %d = %d", lo_y, -1*lo_y, amount_to_shrink_x);
lo_y = 0;
} else if (hi_y >= zoom_image_height) {
amount_to_shrink_x = (hi_y - zoom_image_height) * base_image_width / base_image_height;
hi_y = zoom_image_height - 1;
}
Debug(1, "Shrinking x by %d from %d,%d to %d,%d", amount_to_shrink_x, lo_x, hi_x, lo_x+(amount_to_shrink_x/2), hi_x-(amount_to_shrink_x/2));
if (amount_to_shrink_x) {
lo_x += amount_to_shrink_x/2;
hi_x -= amount_to_shrink_x/2;
}
Debug(3, "Cropping to %d,%d -> %d,%d %dx%din blown up image", lo_x, lo_y, hi_x, hi_y, hi_x-lo_x, hi_y-lo_y);
// Scaled back to base_image dimensions
last_crop = Box({lo_x*ZM_SCALE_BASE/cur_zoom, lo_y*ZM_SCALE_BASE/cur_zoom}, {hi_x*ZM_SCALE_BASE/cur_zoom, hi_y*ZM_SCALE_BASE/cur_zoom});
last_crop_width = base_image_width;
last_crop_height = base_image_height;
Debug(3, "Cropping to %d,%d -> %d,%d", last_crop.Lo().x_, last_crop.Lo().y_, last_crop.Hi().x_, last_crop.Hi().y_);
if ( !image_copied ) {
static Image copy_image;
copy_image.Assign(*image);
image = ©_image;
image_copied = true;
}
image->Crop(last_crop);
image->Scale(disp_image_width, disp_image_height);
} else if (cur_scale != ZM_SCALE_BASE) {
Debug(3, "scaling by %d from %dx%d", cur_scale, image->Width(), image->Height());
static Image copy_image;
copy_image.Assign(*image);
image = ©_image;
image_copied = true;
image->Scale(cur_scale);
}
Debug(3, "Sending %dx%d", image->Width(), image->Height());
// Record what was actually rendered, not whatever the command thread may
// have set since.
last_scale = cur_scale;
last_zoom = cur_zoom;
last_x = cur_x;
last_y = cur_y;
return image;
} // end Image *StreamBase::prepareImage(Image *image, int pre_scaled_by)
bool StreamBase::sendTextFrame(const char *frame_text) {
int width = 640;
int height = 480;
int colours = ZM_COLOUR_RGB32;
int subpixelorder = ZM_SUBPIX_ORDER_RGBA;
int labelsize = 2;
if (monitor) {
width = monitor->Width();
height = monitor->Height();
colours = monitor->Colours();
subpixelorder = monitor->SubpixelOrder();
labelsize = monitor->LabelSize();
}
Debug(2, "Sending %dx%dx%dx%d * %d scale text frame '%s'",
width, height, colours, subpixelorder, scale.load(), frame_text);
Image image(width, height, colours, subpixelorder);
image.Clear();
image.Annotate(frame_text, image.centreCoord(frame_text, labelsize), labelsize);
if (scale != 100) {
image.Scale(scale);
Debug(2, "Scaled to %dx%d", image.Width(), image.Height());
}
if (type == STREAM_MPEG) {
if (!vid_stream) {
vid_stream = new VideoStream("pipe:", format, bitrate, effective_fps, image.Colours(), image.SubpixelOrder(), image.Width(), 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;
}
}
// The text is drawn at the current monitor's size and colours, which need
// not be those the stream was opened with: event frames are RGB24, and
// gapless playback can move to another monitor (refs #4186).
/* double pts = */ vid_stream->EncodeImage(image);
} else {
static unsigned char buffer[ZM_MAX_IMAGE_SIZE];
size_t n_bytes = 0;
image.EncodeJpeg(buffer, &n_bytes);
if (type == STREAM_JPEG) {
if (0 > fputs("--" BOUNDARY "\r\n", stdout)) {
Debug(1, "Error sending --" BOUNDARY "\r\n");
return false;
}
}
if (0 > fprintf(stdout, "Content-Type: image/jpeg\r\nContent-Length: %zu\r\n\r\n", n_bytes)) {
Debug(1, "Error sending Content-Type: image/jpeg\r\nContent-Length: %zu\r\n\r\n", n_bytes);
return false;
}
int rc = fwrite(buffer, n_bytes, 1, stdout);
if (rc != 1) {
if (!zm_terminate)
Error("Unable to send stream text frame: %d %s", rc, strerror(errno));
return false;
}
fputs("\r\n\r\n", stdout);
fflush(stdout);
}
last_frame_sent = now;
return true;
}
void StreamBase::openComms() {
if ( connkey > 0 ) {
// Have to mkdir because systemd is now chrooting and the dir may not exist
if ( mkdir(staticConfig.PATH_SOCKS.c_str(), 0755) ) {
if ( errno != EEXIST ) {
Error("Can't mkdir ZM_PATH_SOCKS %s: %s.", staticConfig.PATH_SOCKS.c_str(), strerror(errno));
}
}
unsigned int length = snprintf(
sock_path_lock,
sizeof(sock_path_lock),
"%s/zms-%06d-%d.lock",
staticConfig.PATH_SOCKS.c_str(),
connkey,
owner
);
if ( length >= sizeof(sock_path_lock) ) {
Warning("Socket lock path was truncated.");
}
Debug(1, "Trying to open the lock on %s", sock_path_lock);
// Under systemd, we get chrooted to something like /tmp/systemd-apache-blh/ so the dir may not exist.
if ( mkdir(staticConfig.PATH_SOCKS.c_str(), 0755) ) {
if ( errno != EEXIST ) {
Error("Can't mkdir %s: %s", staticConfig.PATH_SOCKS.c_str(), strerror(errno));
return;
} else {
Debug(3, "SOCKS dir %s already exists", staticConfig.PATH_SOCKS.c_str() );
}
} else {
Debug(3, "Success making SOCKS dir %s", staticConfig.PATH_SOCKS.c_str() );
}
lock_fd = open(sock_path_lock, O_CREAT|O_WRONLY, S_IRUSR | S_IWUSR);
if ( lock_fd < 0 ) {
Error("Unable to open sock lock file %s: %s", sock_path_lock, strerror(errno));
lock_fd = -1;
} else if ( flock(lock_fd, LOCK_EX) != 0 ) {
Error("Unable to lock sock lock file %s: %s", sock_path_lock, strerror(errno));
close(lock_fd);
lock_fd = -1;
} else {
Debug(1, "We have obtained a lock on %s fd: %d", sock_path_lock, lock_fd);
}
sd = socket(AF_UNIX, SOCK_DGRAM, 0);
if ( sd < 0 ) {
Fatal("Can't create socket: %s", strerror(errno));
} else {
Debug(3, "Have socket %d", sd);
}
length = snprintf(
loc_sock_path,
sizeof(loc_sock_path),
"%s/zms-%06d-%ds.sock",
staticConfig.PATH_SOCKS.c_str(),
connkey,
owner
);
if ( length >= sizeof(loc_sock_path) ) {
Warning("Socket path was truncated.");
length = sizeof(loc_sock_path)-1;
}
// Unlink before bind, in case it already exists
unlink(loc_sock_path);
if ( sizeof(loc_addr.sun_path) < length ) {
Error("Not enough space %zu in loc_addr.sun_path for socket file %s", sizeof(loc_addr.sun_path), loc_sock_path);
}
strncpy(loc_addr.sun_path, loc_sock_path, sizeof(loc_addr.sun_path));
loc_addr.sun_family = AF_UNIX;
Debug(3, "Binding to %s", loc_sock_path);
if ( ::bind(sd, (struct sockaddr *)&loc_addr, strlen(loc_addr.sun_path)+sizeof(loc_addr.sun_family)+1) < 0 ) {
Fatal("Can't bind: %s", strerror(errno));
}
snprintf(rem_sock_path, sizeof(rem_sock_path), "%s/zms-%06d-%dw.sock", staticConfig.PATH_SOCKS.c_str(), connkey, owner);
strncpy(rem_addr.sun_path, rem_sock_path, sizeof(rem_addr.sun_path));
rem_addr.sun_family = AF_UNIX;
struct timeval tv {1,0}; /* 1 Secs Timeout */
setsockopt(sd, SOL_SOCKET, SO_RCVTIMEO,(struct timeval *)&tv, sizeof(struct timeval));
last_comm_update = std::chrono::steady_clock::now();
Debug(3, "comms open at %s", loc_sock_path);
} // end if connKey > 0
} // end void StreamBase::openComms()
void StreamBase::closeComms() {
if ( connkey > 0 ) {
if ( sd >= 0 ) {
close(sd);
sd = -1;
}
// Remove our command socket, but only while we still hold the lock, and only
// if we actually got the lock.
//
// Note this is zms-<connkey>-<owner>s.sock, not the .lock file. A second zms sharing
// this connkey and owner is blocked in flock(LOCK_EX) on the .lock file, and the first
// thing it does after winning that lock is unlink this very path itself
// ("Unlink before bind" in openComms()). It never reads our socket file, so
// removing it here only does early what our successor would do anyway.
//
// Ordering matters: if we unlinked after releasing the lock we could delete
// a socket the successor had already bound, leaving it unreachable.
//
// The lock_fd guard matters because openComms() carries on and binds even
// when it fails to take the lock, so without it a lockless zms could delete
// a socket belonging to the zms that does hold the lock. Leaking the file in
// that case is no worse than the behaviour before this check existed.
//
// This is worth doing because web/ajax/stream.php uses file_exists() on this
// path to decide whether zms is listening. A socket left behind by an exited
// zms makes that check succeed, so the sendto() fails with ECONNREFUSED
// instead of waiting for the new process to bind, and the command is lost.
if ( (lock_fd >= 0) && loc_sock_path[0]
&& (unlink(loc_sock_path) < 0) && (errno != ENOENT) ) {
Warning("Failed to unlink '%s': %s", loc_sock_path, strerror(errno));
}
// Don't unlink sock_path_lock: another zms may already be waiting on it.
if ( lock_fd >= 0 ) {
close(lock_fd); //close it rather than unlock it in case it got deleted.
lock_fd = -1;
}
}
} // end void StreamBase::closeComms
void StreamBase::reserveTempImgBuffer(size_t size) {
if (temp_img_buffer_size < size) {
Debug(1, "Resizing image buffer from %zu to %zu", temp_img_buffer_size, size);
delete[] temp_img_buffer;
temp_img_buffer = new uint8_t[size];
temp_img_buffer_size = size;
}
}