Merge pull request #24 from themactep/feature/rtsp-stability-improvements

fix: harden RTSP timestamp handling

Author: themactep

src/AudioReframer.cpp

@@ -5,7 +5,7 @@
 AudioReframer::AudioReframer(unsigned int inputSampleRate, unsigned int inputSamplesPerFrame,
                              unsigned int outputSamplesPerFrame)
     : inputSampleRate(inputSampleRate), inputSamplesPerFrame(inputSamplesPerFrame),
-      outputSamplesPerFrame(outputSamplesPerFrame), currentTimestamp(0), samplesAccumulated(0),
+      outputSamplesPerFrame(outputSamplesPerFrame), currentTimestamp(0), timestampRemainder(0), samplesAccumulated(0),
       buffer(2 * std::max(inputSamplesPerFrame, outputSamplesPerFrame) * sizeof(uint16_t)) {
   if (inputSamplesPerFrame == 0 || outputSamplesPerFrame == 0) {
     throw std::invalid_argument("Number of samples per frame must be greater than zero.");
@@ -22,6 +22,7 @@ void AudioReframer::addFrame(const uint8_t *frameData, int64_t timestamp) {
 
   if (samplesAccumulated == 0) {
     currentTimestamp = timestamp; // Initialize timestamp with the first frame
+    timestampRemainder = 0;
   }
 
   samplesAccumulated += inputSamplesPerFrame;
@@ -42,7 +43,14 @@ void AudioReframer::getReframedFrame(uint8_t *frameData, int64_t &timestamp) {
 
   timestamp = currentTimestamp;
   if (inputSampleRate > 0) {
-    currentTimestamp += (outputSamplesPerFrame * 1000) / inputSampleRate;
+    // Timestamp is in microseconds; preserve fractional precision between frames.
+    int64_t numer = static_cast<int64_t>(outputSamplesPerFrame) * 1000000LL + timestampRemainder;
+    int64_t step_us = numer / static_cast<int64_t>(inputSampleRate);
+    timestampRemainder = numer % static_cast<int64_t>(inputSampleRate);
+    if (step_us < 1) {
+      step_us = 1;
+    }
+    currentTimestamp += step_us;
   } // else: keep currentTimestamp stable if misconfigured
 }
 

src/AudioReframer.hpp

@@ -20,6 +20,7 @@ class AudioReframer {
   unsigned int inputSamplesPerFrame;
   unsigned int outputSamplesPerFrame;
   int64_t currentTimestamp;
+  int64_t timestampRemainder;
   size_t samplesAccumulated;
 
   RingBuffer buffer;

src/Config.cpp

@@ -470,8 +470,8 @@ std::vector<ConfigItem<int>> CFG::getIntItems() {
       {"audio.spk_gain", audio.output_gain, 20, [](const int &v) { return v >= 0 && v <= 31; }},
       {"audio.spk_vol", audio.output_vol, 60, [](const int &v) { return v >= -30 && v <= 120; }},
 #endif
-      {"audio.buffer_warn_frames", audio.buffer_warn_frames, 80, [](const int &v) { return v >= 10 && v <= 1000; }},
-      {"audio.buffer_cap_frames", audio.buffer_cap_frames, 100, [](const int &v) { return v >= 10 && v <= 1000; }},
+      {"audio.buffer_warn_frames", audio.buffer_warn_frames, 240, [](const int &v) { return v >= 10 && v <= 1000; }},
+      {"audio.buffer_cap_frames", audio.buffer_cap_frames, 400, [](const int &v) { return v >= 10 && v <= 1000; }},
       {"daynight.switch_below_percent", daynight.switch_below_percent, 15, [](const int &v) { return v >= 0 && v <= 100; }},
       {"daynight.switch_above_percent", daynight.switch_above_percent, 80, [](const int &v) { return v >= 0 && v <= 100; }},
       {"daynight.tolerance_percent", daynight.tolerance_percent, 50, [](const int &v) { return v >= 0 && v <= 100; }},

src/IMPAudioServerMediaSubsession.cpp

@@ -29,6 +29,9 @@ FramedSource *IMPAudioServerMediaSubsession::createNewStreamSource(unsigned clie
     return nullptr;
   }
 
+  if (global_audio[audioChn]->msgChannel) {
+    global_audio[audioChn]->msgChannel->clear();
+  }
   FramedSource *audioSourceReplica = replicator->createStreamReplica();
   if (audioSourceReplica) {
     global_audio[audioChn]->rtsp_client_count.fetch_add(1, std::memory_order_relaxed);

src/IMPDeviceSource.cpp

@@ -1,8 +1,32 @@
 #include "IMPDeviceSource.hpp"
 #include "GroupsockHelper.hh"
+#include <cstring>
 #include <iostream>
 #include <type_traits>
 
+static inline int64_t tv_to_us(const struct timeval &tv) {
+  return static_cast<int64_t>(tv.tv_sec) * 1000000LL + static_cast<int64_t>(tv.tv_usec);
+}
+
+static inline struct timeval us_to_tv(int64_t us) {
+  struct timeval tv;
+  tv.tv_sec = static_cast<time_t>(us / 1000000LL);
+  tv.tv_usec = static_cast<suseconds_t>(us % 1000000LL);
+  if (tv.tv_usec < 0) {
+    tv.tv_sec -= 1;
+    tv.tv_usec += 1000000;
+  }
+  return tv;
+}
+
+static inline bool is_plausible_wallclock_tv(const struct timeval &tv) {
+  if (tv.tv_usec < 0 || tv.tv_usec >= 1000000) {
+    return false;
+  }
+  constexpr time_t kMinUnixTime = 946684800; // 2000-01-01
+  return tv.tv_sec >= kMinUnixTime;
+}
+
 // explicit instantiation
 template class IMPDeviceSource<H264NALUnit, video_stream>;
 template class IMPDeviceSource<AudioFrame, audio_stream>;
@@ -56,7 +80,7 @@ template <typename FrameType, typename Stream> void IMPDeviceSource<FrameType, S
   }
 
   FrameType nal;
-  if (stream->msgChannel->read(&nal)) {
+  while (stream->msgChannel->read(&nal)) {
     if (nal.data.size() > fMaxSize) {
       fFrameSize = fMaxSize;
       fNumTruncatedBytes = nal.data.size() - fMaxSize;
@@ -68,15 +92,40 @@ template <typename FrameType, typename Stream> void IMPDeviceSource<FrameType, S
        steady +1024 RTP PTS increments and avoid gettimeofday jitter. */
     if constexpr (std::is_same_v<FrameType, AudioFrame>) {
       auto *imp_audio = global_audio[encChn]->imp_audio;
-      if (imp_audio && imp_audio->format == IMPAudioFormat::AAC) {
+      bool use_aac_clock = false;
+      int sampleRate = 16000;
+      if (cfg && cfg->audio.input_sample_rate > 0) {
+        sampleRate = cfg->audio.input_sample_rate;
+      }
+      if (imp_audio) {
+        use_aac_clock = (imp_audio->format == IMPAudioFormat::AAC);
+        if (imp_audio->sample_rate > 0) {
+          sampleRate = imp_audio->sample_rate;
+        }
+      } else if (cfg && cfg->audio.input_format && std::strcmp(cfg->audio.input_format, "AAC") == 0) {
+        // During transient audio restarts, keep AAC timing behavior stable.
+        use_aac_clock = true;
+      }
+
+      bool has_audio_time = is_plausible_wallclock_tv(nal.time);
+      if (has_audio_time) {
+        fPresentationTime = nal.time;
+      } else if (use_aac_clock) {
         if (audioFirstFrame) {
           gettimeofday(&audioStartTime, NULL);
           audioFrameCount = 0;
+          audioClockSampleRate = sampleRate;
           audioFirstFrame = false;
+        } else if (audioClockSampleRate <= 0 && sampleRate > 0) {
+          audioClockSampleRate = sampleRate;
+        }
+        int effectiveSampleRate = (audioClockSampleRate > 0) ? audioClockSampleRate : sampleRate;
+        if (effectiveSampleRate <= 0) {
+          effectiveSampleRate = 16000;
         }
-        int sampleRate = (imp_audio->sample_rate > 0) ? imp_audio->sample_rate : 16000;
         constexpr uint64_t kSamplesPerFrame = 1024;
-        uint64_t usec_offset = (audioFrameCount * kSamplesPerFrame * 1000000ULL) / static_cast<uint64_t>(sampleRate);
+        uint64_t usec_offset =
+            (audioFrameCount * kSamplesPerFrame * 1000000ULL) / static_cast<uint64_t>(effectiveSampleRate);
         fPresentationTime.tv_sec  = audioStartTime.tv_sec  + static_cast<time_t>(usec_offset / 1000000ULL);
         fPresentationTime.tv_usec = audioStartTime.tv_usec + static_cast<suseconds_t>(usec_offset % 1000000ULL);
         if (fPresentationTime.tv_usec >= 1000000) {
@@ -87,53 +136,96 @@ template <typename FrameType, typename Stream> void IMPDeviceSource<FrameType, S
       } else {
         gettimeofday(&fPresentationTime, NULL);
       }
-    } else {
-      // Video: use encoder timestamps for stable, jitter-free PTS.
-      // The IMP encoder provides a monotonic microsecond timestamp per NAL.
-      // We anchor it to wall-clock at the first frame, then derive all
-      // subsequent presentation times from the encoder's own clock.
-      if (videoFirstFrame) {
-        gettimeofday(&videoBaseTime, NULL);
-        videoFirstImpTs = nal.imp_ts;
-        videoFirstFrame = false;
+
+      int64_t pts_us = tv_to_us(fPresentationTime);
+      int rate_for_tick = sampleRate > 0 ? sampleRate : 16000;
+      int64_t min_audio_step_us = (1000000LL + rate_for_tick - 1) / rate_for_tick;
+      if (use_aac_clock) {
+        min_audio_step_us = (1024LL * 1000000LL + rate_for_tick - 1) / rate_for_tick;
       }
-      int64_t delta_us = nal.imp_ts - videoFirstImpTs;
-      // Re-anchor on negative delta (wrap/reset) or any step > 2s relative to
-      // the previous frame — forward (encoder uptime jump) or backward (encoder
-      // counter reset that doesn't go below the anchor).
-      bool needs_reanchor = delta_us < 0;
-      if (!needs_reanchor && videoLastDelta >= 0) {
-        int64_t step = delta_us - videoLastDelta;
-        needs_reanchor = (step > 2000000LL) || (step < -500000LL);
+      if (min_audio_step_us < 1) {
+        min_audio_step_us = 1;
       }
-      if (needs_reanchor) {
-        gettimeofday(&videoBaseTime, NULL);
-        videoFirstImpTs = nal.imp_ts;
-        delta_us = 0;
-      } else if (videoLastDelta >= 0 && delta_us <= videoLastDelta) {
-        // Clamp small backward jitter (<500ms) to keep PTS strictly
-        // monotonically increasing.  The <= also handles the SPS/PPS/IDR
-        // triplet where all three NALs share the same imp_ts (delta_us ==
-        // videoLastDelta): each gets nudged forward by one 90 kHz tick
-        // (≈11 µs) so the RTP sender never emits two packets with the
-        // same timestamp.
-        delta_us = videoLastDelta + 12;
+      if (audioLastPtsUs >= 0) {
+        int64_t delta_pts_us = pts_us - audioLastPtsUs;
+        int64_t max_forward_jump_us = 1000000LL;
+        if (use_aac_clock && min_audio_step_us > 0) {
+          int64_t aac_jump_limit = min_audio_step_us * 120;
+          if (aac_jump_limit > max_forward_jump_us) {
+            max_forward_jump_us = aac_jump_limit;
+          }
+        }
+        if (delta_pts_us <= 0 || delta_pts_us > max_forward_jump_us) {
+          pts_us = audioLastPtsUs + min_audio_step_us;
+          fPresentationTime = us_to_tv(pts_us);
+        }
       }
-      videoLastDelta = delta_us;
-      fPresentationTime.tv_sec  = videoBaseTime.tv_sec  + static_cast<time_t>(delta_us / 1000000LL);
-      fPresentationTime.tv_usec = videoBaseTime.tv_usec + static_cast<suseconds_t>(delta_us % 1000000LL);
-      if (fPresentationTime.tv_usec >= 1000000) {
-        fPresentationTime.tv_sec++;
-        fPresentationTime.tv_usec -= 1000000;
+      audioLastPtsUs = pts_us;
+    } else {
+      if (is_plausible_wallclock_tv(nal.time)) {
+        fPresentationTime = nal.time;
+      } else {
+        // Video fallback path for sources that don't provide explicit timeval.
+        int64_t nominal_step_us = 33333;
+        if (stream && stream->stream && stream->stream->fps > 0) {
+          nominal_step_us = 1000000LL / stream->stream->fps;
+        }
+        if (nominal_step_us < 12) {
+          nominal_step_us = 12;
+        }
+        if (videoFirstFrame) {
+          gettimeofday(&videoBaseTime, NULL);
+          videoFirstImpTs = nal.imp_ts;
+          videoFirstFrame = false;
+        }
+
+        int64_t delta_us = nal.imp_ts - videoFirstImpTs;
+
+        bool needs_reanchor = delta_us < 0;
+        if (!needs_reanchor && videoLastDelta >= 0) {
+          int64_t step = delta_us - videoLastDelta;
+          needs_reanchor = (step > 2000000LL) || (step < -500000LL);
+        }
+        if (needs_reanchor) {
+          int64_t target_delta = (videoLastDelta >= 0) ? (videoLastDelta + nominal_step_us) : 0;
+          videoFirstImpTs = nal.imp_ts - target_delta;
+          delta_us = target_delta;
+        } else if (videoLastDelta >= 0 && delta_us <= videoLastDelta) {
+          delta_us = videoLastDelta + 12;
+        }
+        videoLastDelta = delta_us;
+        fPresentationTime.tv_sec  = videoBaseTime.tv_sec  + static_cast<time_t>(delta_us / 1000000LL);
+        fPresentationTime.tv_usec = videoBaseTime.tv_usec + static_cast<suseconds_t>(delta_us % 1000000LL);
+        if (fPresentationTime.tv_usec >= 1000000) {
+          fPresentationTime.tv_sec++;
+          fPresentationTime.tv_usec -= 1000000;
+        }
+      }
+
+      int64_t pts_us = tv_to_us(fPresentationTime);
+      int64_t min_video_step_us = 12;
+      if (stream && stream->stream && stream->stream->fps > 0) {
+        min_video_step_us = 1000000LL / stream->stream->fps;
+        if (min_video_step_us < 12) {
+          min_video_step_us = 12;
+        }
+      }
+      if (videoLastPtsUs >= 0) {
+        int64_t delta_pts_us = pts_us - videoLastPtsUs;
+        if (delta_pts_us <= 0 || delta_pts_us > 2000000LL) {
+          pts_us = videoLastPtsUs + min_video_step_us;
+          fPresentationTime = us_to_tv(pts_us);
+        }
       }
+      videoLastPtsUs = pts_us;
     }
 
     memcpy(fTo, &nal.data[0], fFrameSize);
 
     if (fFrameSize > 0) {
       FramedSource::afterGetting(this);
+      return;
     }
-  } else {
-    fFrameSize = 0;
   }
+  fFrameSize = 0;
 }

src/IMPDeviceSource.hpp

@@ -34,11 +34,14 @@ template <typename FrameType, typename Stream> class IMPDeviceSource : public Fr
   bool audioFirstFrame{true};
   struct timeval audioStartTime{};
   uint64_t audioFrameCount{0};
+  int audioClockSampleRate{0};
+  int64_t audioLastPtsUs{-1};
 
   // Video encoder timestamp tracking (avoids gettimeofday jitter)
   bool videoFirstFrame{true};
   int64_t videoFirstImpTs{0};
   int64_t videoLastDelta{-1};
+  int64_t videoLastPtsUs{-1};
   struct timeval videoBaseTime{};
 };
 

src/IMPEncoder.cpp

@@ -39,7 +39,6 @@ IMPEncoder *IMPEncoder::createNew(_stream *stream, int encChn, int encGrp, const
 void IMPEncoder::flush(int encChn) {
   LOG_DDEBUG("flush(" << encChn << ")");
   IMP_Encoder_RequestIDR(encChn);
-  IMP_Encoder_FlushStream(encChn);
 }
 
 void MakeTables(int q, uint8_t *lqt, uint8_t *cqt) {

src/IMPServerMediaSubsession.hpp

@@ -33,14 +33,19 @@ class IMPServerMediaSubsession : public OnDemandServerMediaSubsession {
                            void *rtcpRRHandlerClientData, unsigned short &rtpSeqNum, unsigned &rtpTimestamp,
                            ServerRequestAlternativeByteHandler *serverRequestAlternativeByteHandler,
                            void *serverRequestAlternativeByteHandlerClientData) override {
+    // Drop any stale queued NAL units so a new client starts from a fresh IDR
+    // timeline and does not see pre-flush timestamp history.
+    if (encChn >= 0 && encChn < NUM_VIDEO_CHANNELS && global_video[encChn] && global_video[encChn]->msgChannel) {
+      global_video[encChn]->msgChannel->clear();
+    }
+    // request idr frame every second for the next x seconds
+    global_video[encChn]->idr_fix = 5;
+    IMPEncoder::flush(encChn);
+
     OnDemandServerMediaSubsession::startStream(clientSessionId, streamToken, rtcpRRHandler, rtcpRRHandlerClientData,
                                                rtpSeqNum, rtpTimestamp, serverRequestAlternativeByteHandler,
                                                serverRequestAlternativeByteHandlerClientData);
-
     global_rtsp_clients.fetch_add(1, std::memory_order_relaxed);
-    // request idr frame every second for the next x seconds
-    global_video[encChn]->idr_fix = 5;
-    IMPEncoder::flush(encChn);
   }
 
   virtual void deleteStream(unsigned clientSessionId, void *&streamToken) override {