From 22d02ce0661aeec8a5cb2b71f257ebedc32776ca Mon Sep 17 00:00:00 2001 From: Nic Boet Date: Mon, 12 Oct 2026 10:05:40 -0500 Subject: [PATCH] fix: stop the nph-zms command thread racing runStream on frame bookkeeping checkCommandQueue runs processCommand on its own thread, which read `now` and frame_count and wrote last_frame_sent (CMD_PAUSE) and got_command while the streaming loop writes or reads them. ThreadSanitizer reports these as data races (zm_monitorstream.cpp processCommand and runStream's got_command read against the command thread, and checkCommandQueue's `now` read against EventStream::runStream). Make frame_count, last_frame_sent and got_command atomic, as #4939 did for the playback state, and have the command thread read the clock itself instead of the loop's `now`. The status reply's fps and delay are then measured against the current time rather than the start of the loop's last pass. Test: [races] case for the frame bookkeeping, for TSan builds. fixes #5230 Co-Authored-By: Claude Opus 5.5 --- src/zm_eventstream.cpp | 18 ++++++++------- src/zm_monitorstream.cpp | 33 +++++++++++++++------------ src/zm_stream.cpp | 7 ++++-- src/zm_stream.h | 11 ++++++--- tests/zm_stream_races.cpp | 47 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 89 insertions(+), 27 deletions(-) diff --git a/src/zm_eventstream.cpp b/src/zm_eventstream.cpp index da7aee960..b3240a2fc 100644 --- a/src/zm_eventstream.cpp +++ b/src/zm_eventstream.cpp @@ -736,13 +736,15 @@ void EventStream::processCommand(const CmdMsg *msg) { status_data.paused = paused; status_data.stopped = stopped; - FPSeconds elapsed = now - last_fps_update; + // The command thread: read the clock here rather than the streaming loop's `now`. + const TimePoint cmd_now = std::chrono::steady_clock::now(); + FPSeconds elapsed = cmd_now - last_fps_update; if (elapsed.count() > 0) { actual_fps = (actual_fps + (frame_count - last_frame_count) / elapsed.count())/2; - Debug(1, "actual_fps %f = old + frame_count %d - last %d / elapsed %.2f from %.2f - %.2f scale %d", actual_fps, frame_count, last_frame_count, - elapsed.count(), FPSeconds(now.time_since_epoch()).count(), FPSeconds(last_fps_update.time_since_epoch()).count(), scale.load()); + Debug(1, "actual_fps %f = old + frame_count %d - last %d / elapsed %.2f from %.2f - %.2f scale %d", actual_fps, frame_count.load(), last_frame_count, + elapsed.count(), FPSeconds(cmd_now.time_since_epoch()).count(), FPSeconds(last_fps_update.time_since_epoch()).count(), scale.load()); last_frame_count = frame_count; - last_fps_update = now; + last_fps_update = cmd_now; } status_data.fps = actual_fps; @@ -848,7 +850,7 @@ bool EventStream::checkEventLoaded() { // Reset last_frame_sent so the next loop iteration sends a keepalive // immediately, rather than leaving a 5-second gap (MAX_STREAM_DELAY) // during which the HTTP connection can time out and break the pipe. - last_frame_sent = {}; + last_frame_sent = TimePoint(); } return false; } // void EventStream::checkEventLoaded() @@ -1101,14 +1103,14 @@ void EventStream::runStream() { event_data ? event_data->event_id : 0, event_data ? event_data->frames.size() : 0, event_data ? event_data->frame_count : 0, - curr_frame_id, frame_count); + curr_frame_id, frame_count.load()); } { std::scoped_lock lck{mutex}; send_frame = false; - TimePoint::duration time_since_last_send = now - last_frame_sent; + TimePoint::duration time_since_last_send = now - last_frame_sent.load(); if (stopped) { // In stopped state, skip all frame processing until a new command is received. @@ -1138,7 +1140,7 @@ void EventStream::runStream() { } else { Debug(4, "Not Sending keepalive frame now %.2f - %.2f last = %.2f > Max %.2f", FPSeconds(now.time_since_epoch()).count(), - FPSeconds(last_frame_sent.time_since_epoch()).count(), + FPSeconds(last_frame_sent.load().time_since_epoch()).count(), FPSeconds(time_since_last_send).count(), FPSeconds(MAX_STREAM_DELAY).count() ); diff --git a/src/zm_monitorstream.cpp b/src/zm_monitorstream.cpp index 3ed4c1de2..ac137ee1c 100644 --- a/src/zm_monitorstream.cpp +++ b/src/zm_monitorstream.cpp @@ -88,6 +88,10 @@ void MonitorStream::processCommand(const CmdMsg *msg) { return; } + // This runs on the command thread: read the clock here rather than the + // streaming loop's `now`. + const TimePoint cmd_now = std::chrono::steady_clock::now(); + // Check for incoming command switch ((MsgCommand)msg->msg_data[0]) { case CMD_PAUSE : @@ -95,7 +99,7 @@ void MonitorStream::processCommand(const CmdMsg *msg) { stopped = false; paused = true; delayed = true; - last_frame_sent = now; + last_frame_sent = cmd_now; break; case CMD_PLAY : Debug(1, "Got PLAY command"); @@ -286,11 +290,11 @@ void MonitorStream::processCommand(const CmdMsg *msg) { status_data.analysing = 0; status_data.score = 0; } else { - FPSeconds elapsed = now - last_fps_update; + FPSeconds elapsed = cmd_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; + last_fps_update = cmd_now; } status_data.fps = actual_fps; @@ -310,7 +314,7 @@ void MonitorStream::processCommand(const CmdMsg *msg) { status_data.paused = paused; status_data.stopped = stopped; status_data.rate = replay_rate; - status_data.delay = FPSeconds(now - last_frame_sent).count(); + status_data.delay = FPSeconds(cmd_now - last_frame_sent.load()).count(); status_data.zoom = zoom; status_data.scale = scale; status_data.analysis_image = (frame_type == FRAME_ANALYSIS) && @@ -481,7 +485,7 @@ bool MonitorStream::sendFrame(Image *image, SystemTimePoint timestamp) { 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 frame_send_time = last_frame_sent.load() - send_start_time; TimePoint::duration maxfps_milliseconds = Milliseconds(lround(Milliseconds::period::den / maxfps)); if (frame_send_time > maxfps_milliseconds) { @@ -633,7 +637,7 @@ void MonitorStream::runStream() { 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, + monitor_id, zm_get_rss_kb(), frame_count.load(), MonitorStreamBufferLevel(temp_write_index.load(), temp_read_index.load(), temp_image_buffer_count.load()), temp_image_buffer_count.load(), paused.load(), delayed.load()); } @@ -694,7 +698,7 @@ void MonitorStream::runStream() { stopped_image = zm::make_unique(*monitor->ReadShmFrame(index)); stopped_timestamp = SystemTimePoint(zm::chrono::duration_cast(monitor->shared_timestamps[index])); } - if (now - last_frame_sent > Seconds(5)) { + if (now - last_frame_sent.load() > Seconds(5)) { if (stopped_image) { Debug(2, "Sending keepalive frame while stopped"); if (!sendFrame(stopped_image.get(), stopped_timestamp)) zm_terminate = true; @@ -752,7 +756,7 @@ void MonitorStream::runStream() { 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; + TimePoint::duration actual_delta_time = now - last_frame_sent.load(); // If the next frame is due if (actual_delta_time > expected_delta_time) { @@ -791,7 +795,7 @@ void MonitorStream::runStream() { //paused? int temp_index = MOD_ADD(temp_read_index, 0, temp_image_buffer_count); - if (got_command || (now - last_frame_sent > Seconds(5))) { + if (got_command || (now - last_frame_sent.load() > Seconds(5))) { // Send keepalive Debug(2, "Sending keepalive frame %d", temp_index); // Send the next frame @@ -823,7 +827,7 @@ void MonitorStream::runStream() { 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()); + index, last_write_index, monitor->image_buffer_count, frame_mod, frame_count.load(), 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 @@ -878,7 +882,7 @@ void MonitorStream::runStream() { frame_count++; frame_count++; } else { - TimePoint::duration actual_delta_time = now - last_frame_sent; + TimePoint::duration actual_delta_time = now - last_frame_sent.load(); if (actual_delta_time > Seconds(5)) { if (paused_image) { // Send keepalive @@ -933,7 +937,7 @@ void MonitorStream::runStream() { monitor->shared_data->last_write_index, last_read_index); - if (now - last_frame_sent > Seconds(5)) { + if (now - last_frame_sent.load() > Seconds(5)) { if (last_read_index == monitor->GetImageBufferCount()) { sendTextFrame("Waiting for initial capture"); // Timeout if we've never received a frame from capture @@ -972,8 +976,9 @@ void MonitorStream::runStream() { } - if (last_frame_sent > now) { - FPSeconds elapsed = last_frame_sent - now; + const TimePoint sent = last_frame_sent.load(); + if (sent > now) { + FPSeconds elapsed = sent - now; if (sleep_time > elapsed) { Debug(2, "Adjusting sleep time by %f elapsed", elapsed.count()); sleep_time -= elapsed; diff --git a/src/zm_stream.cpp b/src/zm_stream.cpp index 6627bb92d..a27928bcf 100644 --- a/src/zm_stream.cpp +++ b/src/zm_stream.cpp @@ -126,9 +126,12 @@ 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. - if (now - last_comm_update > Hours(1)) { + // 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 = now; + last_comm_update = cmd_now; } if (sd >= 0) { diff --git a/src/zm_stream.h b/src/zm_stream.h index 5d4284e63..26c5a1c61 100644 --- a/src/zm_stream.h +++ b/src/zm_stream.h @@ -155,20 +155,24 @@ class StreamBase { double effective_fps; // Target fps after taking max_fps into account double actual_fps; // sliding calculated actual streaming fps achieved TimePoint last_fps_update; - int frame_count; // Count of frames sent + // frame_count and last_frame_sent are written by the streaming loop and + // read (last_frame_sent also written) by processCommand on the command + // thread, so they are atomic like the playback state above. `now` stays + // the loop's own: the command thread reads the clock itself. + std::atomic frame_count; // Count of frames sent int last_frame_count; // Used in calculating actual_fps from frame_count - last_frame_count int frame_mod; int frames_to_send; - TimePoint last_frame_sent; + std::atomic last_frame_sent; SystemTimePoint last_frame_timestamp; TimePoint when_to_send_next_frame; // When to send next frame so if now < send_next_frame, skip VideoStream *vid_stream; CmdMsg msg; - bool got_command = false; // commands like zoom should output a frame even if paused + std::atomic got_command; // commands like zoom should output a frame even if paused; set on the command thread uint8_t *temp_img_buffer; // Used when encoding or sending file data size_t temp_img_buffer_size; @@ -227,6 +231,7 @@ class StreamBase { last_frame_count(0), frame_mod(1), frames_to_send(-1), + last_frame_sent(TimePoint()), got_command(false), temp_img_buffer(nullptr), temp_img_buffer_size(0) { diff --git a/tests/zm_stream_races.cpp b/tests/zm_stream_races.cpp index 837be5b57..db14d5dd2 100644 --- a/tests/zm_stream_races.cpp +++ b/tests/zm_stream_races.cpp @@ -59,6 +59,35 @@ class PlaybackStateProbe : public StreamBase { } } + // What processCommand does with the frame bookkeeping on the command thread: + // CMD_PAUSE stores last_frame_sent, every status reply reads frame_count and + // last_frame_sent, and checkCommandQueue sets got_command. + int64 commandFrameState(int iterations) { + int64 sink = 0; + for (int i = 0; i < iterations; i++) { + if (i % 4 == 0) last_frame_sent = std::chrono::steady_clock::now(); + got_command = true; // checkCommandQueue, after each command + sink += frame_count; + const TimePoint sent = last_frame_sent; + sink += sent.time_since_epoch().count() & 1; + } + return sink; + } + + // And runStream()'s side: count each frame sent and stamp it, and check + // got_command for the paused keepalive. + int64 streamFrameState(int iterations) { + int64 sink = 0; + for (int i = 0; i < iterations; i++) { + frame_count++; + last_frame_sent = std::chrono::steady_clock::now(); + sink += got_command ? 1 : 0; + const TimePoint sent = last_frame_sent; + sink += (std::chrono::steady_clock::now() - sent).count() & 1; + } + return sink; + } + // The reads runStream() performs on the streaming thread. The accumulator // exists so none of this is optimised away. int64 readPlaybackState(int iterations) { @@ -106,3 +135,21 @@ TEST_CASE("StreamBase playback state is shared safely between the command and st // produced no race report. REQUIRE(observed.load() >= 0); } + +TEST_CASE("StreamBase frame bookkeeping is shared safely between the command and stream threads", "[races]") { + // processCommand reads frame_count and last_frame_sent for its status reply + // and CMD_PAUSE stores last_frame_sent, while runStream() updates both for + // every frame. Like the case above, this only shows something under TSan. + PlaybackStateProbe probe; + + constexpr int kIterations = 20000; + std::atomic observed{0}; + + std::thread command_thread([&probe, &observed]() { observed += probe.commandFrameState(kIterations); }); + std::thread stream_thread([&probe, &observed]() { observed += probe.streamFrameState(kIterations); }); + + command_thread.join(); + stream_thread.join(); + + REQUIRE(observed.load() >= 0); +}