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 <noreply@anthropic.com>

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()
               );

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<Image>(*monitor->ReadShmFrame(index));
         stopped_timestamp = SystemTimePoint(zm::chrono::duration_cast<Microseconds>(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;

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) {

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<int> 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<TimePoint> 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<bool> 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) {

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<int64> 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);
+}