diff --git a/ChangeLog.md b/ChangeLog.md index b014188..18f5adc 100644 --- a/ChangeLog.md +++ b/ChangeLog.md @@ -37,6 +37,9 @@ The entries below cover work done in the IRL fork on top of the upstream `rstula - `api_keys` directive supports comma separated keys with constant time comparison. - Atomic stats counters and a read locked `put()` so `/stats` no longer stalls the data path. - Replaced the blocking HTTP client with an `AsyncHttpClient` (thread pool backed) for stats posting, player key validation, and `on_event_url` callbacks. +- New per stream `/stats` diagnostics for the jump/replay class of viewer issue: `maxReaderBacklogBytes` and `maxReaderBacklogMs` (furthest any viewer fell behind the ring write head, the size of a potential catch up burst) alongside the existing `ringOverruns`. +- Fixed `sendBackpressure`, which was read off the publisher role (which never runs the egress write path) and so was structurally always zero. It is now aggregated across a stream's viewers via the shared ring, so it reflects real viewer backpressure. +- Teardown log lines (`check_invalid_sock`, `get_state`) now carry `stream=` for grep based correlation across many concurrent streams, and a flight recorder line at publisher teardown records the session's peak backlog, overruns, and viewer backpressure before the ring is freed (survives publisher reconnect). **Logging redesign** @@ -66,6 +69,7 @@ The entries below cover work done in the IRL fork on top of the upstream `rstula - Player latency clamped via `SRTO_PEERLATENCY`; bumped UDP buffers; explicit `SRTO_TLPKTDROP`. - Disconnect viewers stuck in continuous backpressure; handle `EASYNCSND` backpressure without disconnecting healthy viewers. - Race free `m_nDataCount`, write locked `setSize` on the publisher ring; size publisher ring per bitrate with reader overrun detection. +- Egress tuned to stop live viewers replaying stale video. The publisher ring is sized at 1x the latency window instead of 2x, and `MAX_EGRESS_BATCHES` is reduced from 8 to 2, so a viewer that falls behind is skipped forward per packet by SRT `TLPKTDROP` at the socket rather than catching up in a large burst of old frames (which a viewer perceives as a jump back then skip forward). The ring is treated as a small hand off buffer, not a jitter buffer; the SRT socket holds each viewer's latency window. - Compile out per packet `SPDLOG_TRACE` / `SPDLOG_DEBUG` on the data path; drop per packet string allocations. - Race and leak fixes on the accept path (close socket and free `addrinfo` on `libsrt_setup` errors; stop leaking `stat_info_t` per accepted role; null check `strdup(sid)`; kick publisher via atomic flag in `disconnect_stream`). - Removed HLS recording. diff --git a/src/core/SLSGroup.cpp b/src/core/SLSGroup.cpp index d87002c..18a17b4 100644 --- a/src/core/SLSGroup.cpp +++ b/src/core/SLSGroup.cpp @@ -449,10 +449,10 @@ void CSLSGroup::check_invalid_sock() int state = role->get_state(cur_time_ms); if (SLS_RS_INVALID == state || SLS_RS_UNINIT == state) { - spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={}, invalid sock={:d}, " + spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={} stream={}, invalid sock={:d}, " "state={:d}, role_map.size={:d}.", - fmt::ptr(this), m_worker_number, role->get_role_name(), fmt::ptr(role.get()), role->get_fd(), - state, m_map_role.size()); + fmt::ptr(this), m_worker_number, role->get_role_name(), fmt::ptr(role.get()), + role->get_map_data_key(), role->get_fd(), state, m_map_role.size()); // check relay if (role->is_reconnect()) { @@ -495,8 +495,9 @@ void CSLSGroup::check_invalid_sock() } else { - spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={}, delete.", fmt::ptr(this), - m_worker_number, role->get_role_name(), fmt::ptr(role.get())); + spdlog::info("[{}] CSLSGroup::check_invalid_sock, worker_number={:d}, {}={} stream={}, delete.", + fmt::ptr(this), m_worker_number, role->get_role_name(), fmt::ptr(role.get()), + role->get_map_data_key()); } m_map_role.erase(it_erase); continue; diff --git a/src/core/SLSManager.cpp b/src/core/SLSManager.cpp index cbd8907..b1e7c24 100644 --- a/src/core/SLSManager.cpp +++ b/src/core/SLSManager.cpp @@ -445,15 +445,30 @@ json CSLSManager::create_json_stats_for_publisher(CSLSRole *role, int clear) // subscriber's read position was forcibly resynced to the write head // to avoid handing back corrupted wrapped-around data. ret["ringOverruns"] = role->get_ring_overrun_count(); - // Egress send-buffer backpressure events. Counts how often - // srt_sendmsg to this role returned EASYNCSND (SRT send buffer - // full). Each event means the viewer's link could not absorb a - // write burst and SLS deferred the remainder to the next epoll - // wake instead of disconnecting the viewer. Steady growth on a - // player role indicates that viewer's link is under-provisioned - // for the stream bitrate or that the player negotiated too small - // a latency window. - ret["sendBackpressure"] = role->get_send_backpressure_count(); + // Egress send-buffer backpressure events, aggregated across every viewer of + // this stream. Counts how often srt_sendmsg to a viewer returned EASYNCSND + // (SRT send buffer full) — the viewer's link could not absorb a write burst, + // so SLS deferred the remainder to the next epoll wake instead of dropping + // the viewer. Sourced from the shared publisher ring, NOT the publisher + // role's own per-role counter: /stats enumerates only publishers, and a + // publisher never runs the egress write path, so that counter was always 0. + // Steady growth means viewers are falling behind — under-provisioned links + // for the stream bitrate, or a viewer latency window that is too small. + ret["sendBackpressure"] = role->get_viewer_backpressure_events(clear); + // Furthest any viewer of this stream fell behind the publisher ring write + // head (bytes) since the last clear. This is the catch-up burst a viewer + // will drain when it recovers, which the viewer perceives as a time-skip / + // "replay". Divide by the stream bitrate for a millisecond figure + // (maxReaderBacklogMs below). 0 on a healthy stream where every viewer keeps + // up with the write head. + int64_t max_backlog_bytes = role->get_max_reader_backlog(clear); + ret["maxReaderBacklogBytes"] = max_backlog_bytes; + // Same figure expressed as playout time, so operators do not have to do the + // bitrate math. ms = bytes * 8 / kbps. Only meaningful once a bitrate is + // known; reported as 0 otherwise. + int bitrate_kbps = role->get_bitrate(); + ret["maxReaderBacklogMs"] = + (max_backlog_bytes > 0 && bitrate_kbps > 0) ? (int64_t)(max_backlog_bytes * 8 / bitrate_kbps) : 0; ret["audioGapFill"] = json::object(); ret["audioGapFill"]["enabled"] = audio_gap_stats.enabled; diff --git a/src/core/SLSMapData.cpp b/src/core/SLSMapData.cpp index bbb6213..8bbbca1 100644 --- a/src/core/SLSMapData.cpp +++ b/src/core/SLSMapData.cpp @@ -80,19 +80,25 @@ int CSLSMapData::add(char *key, int max_bitrate_kbps, int latency_ms) // Decide the ring size up front (without resizing yet) so the memory cap // can be enforced before we commit the (potentially large) allocation. // The CSLSRecycleArray default (get_data_size()) is the floor; a bitrate + - // latency hint grows it so a subscriber falling up to one full SRT latency - // window behind is still safe from overruns. We multiply by 2 to give a - // viewer's effective lag headroom against publisher-side jitter. Skip the - // growth if the caller passed no hint — the default is enough for typical - // bitrates. + // latency hint grows it so a subscriber falling up to one SRT latency window + // behind is still safe from overruns. Sized at 1x the latency window (was + // 2x): the ring is a hand-off buffer, NOT the jitter buffer — the viewer's + // own SRT socket holds its latency window and drops late packets via + // TLPKTDROP. Oversizing the ring only lets a slow viewer bank multiple + // seconds of stale data that then replays as a rewind. One latency window is + // enough to ride out publisher-side jitter without overrun; anything beyond + // it should be skipped forward at the socket, not stored. Skip the growth if + // the caller passed no hint — the default is enough for typical bitrates. + // Watch /stats ringOverruns after changing this: 0 means the ring is still + // large enough; sustained growth means it was cut too small for the bitrate. int64_t intended_size = (int64_t)data_array->get_data_size(); if (max_bitrate_kbps > 0 && latency_ms > 0) { // bytes_per_sec is bounded by ~max_bitrate_kbps (config caps in // SLSPublisher.hpp limit it to 1_000_000 kbps = 125 MB/s). int64_t bytes_per_sec = (int64_t)max_bitrate_kbps * 1000 / 8; - // 2x the latency window in seconds, clamped to a minimum of 1s. - int64_t window_secs = (int64_t)latency_ms * 2 / 1000; + // 1x the latency window in seconds, clamped to a minimum of 1s. + int64_t window_secs = (int64_t)latency_ms / 1000; if (window_secs < 1) window_secs = 1; int64_t target_bytes = bytes_per_sec * window_secs; @@ -177,6 +183,39 @@ int64_t CSLSMapData::get_overrun_count(const char *key) return it->second->get_overrun_count(); } +int64_t CSLSMapData::get_max_reader_backlog(const char *key, bool clear) +{ + if (!key) + return -1; + CSLSLock lock(&m_rwclock, false); + auto it = m_map_array.find(std::string_view{key}); + if (it == m_map_array.end() || it->second == NULL) + return -1; + return it->second->get_max_reader_backlog(clear); +} + +void CSLSMapData::report_viewer_backpressure(const char *key) +{ + if (!key) + return; + CSLSLock lock(&m_rwclock, false); + auto it = m_map_array.find(std::string_view{key}); + if (it == m_map_array.end() || it->second == NULL) + return; + it->second->report_viewer_backpressure(); +} + +int64_t CSLSMapData::get_viewer_backpressure_events(const char *key, bool clear) +{ + if (!key) + return -1; + CSLSLock lock(&m_rwclock, false); + auto it = m_map_array.find(std::string_view{key}); + if (it == m_map_array.end() || it->second == NULL) + return -1; + return it->second->get_viewer_backpressure_events(clear); +} + int CSLSMapData::remove(char *key) { int ret = SLS_ERROR; diff --git a/src/core/SLSMapData.hpp b/src/core/SLSMapData.hpp index 02608cc..5043d1d 100644 --- a/src/core/SLSMapData.hpp +++ b/src/core/SLSMapData.hpp @@ -99,6 +99,14 @@ class CSLSMapData // (writer lapped the reader). Returns -1 if `key` is unknown. int64_t get_overrun_count(const char *key); + // Diagnostic gauges for the stream `key`, forwarded to its ring. See the + // matching CSLSRecycleArray accessors. Return -1 (getters) / no-op + // (reporter) if `key` is unknown. `clear` resets the high-water so /stats + // can report a per-poll-interval peak. + int64_t get_max_reader_backlog(const char *key, bool clear = false); + void report_viewer_backpressure(const char *key); + int64_t get_viewer_backpressure_events(const char *key, bool clear = false); + int put(char *key, char *data, int len, int64_t *last_read_time = NULL); void set_audio_gap_fill(const char *key, bool enabled); bool get_audio_gap_stats(const char *key, AudioGapStreamStats &stats, int clear = 0); diff --git a/src/core/SLSPublisher.cpp b/src/core/SLSPublisher.cpp index 2622779..e260d0a 100644 --- a/src/core/SLSPublisher.cpp +++ b/src/core/SLSPublisher.cpp @@ -111,6 +111,16 @@ int CSLSPublisher::uninit() if (m_map_data) { + // Flight recorder: the diagnostic gauges live on the ring we are about + // to free, so publisher takeover / reconnect would otherwise erase the + // one session's data an operator most wants (the streamer got fed up and + // reconnected). Flush the ring's final peaks to a single greppable line + // (stream=) before remove() drops it. One line per session end, so + // it stays quiet even with many concurrent streams. + spdlog::info("[{}] CSLSPublisher::uninit, session end stream={}, peakReaderBacklogBytes={}, " + "ringOverruns={}, viewerBackpressure={}.", + fmt::ptr(this), m_map_data_key, get_max_reader_backlog(false), get_ring_overrun_count(), + get_viewer_backpressure_events(false)); ret = m_map_data->remove(m_map_data_key); spdlog::info("[{}] CSLSPublisher::uninit, removed publisher from m_map_data, ret={:d}.", fmt::ptr(this), ret); } diff --git a/src/core/SLSRecycleArray.cpp b/src/core/SLSRecycleArray.cpp index a4a40ef..36d2f44 100644 --- a/src/core/SLSRecycleArray.cpp +++ b/src/core/SLSRecycleArray.cpp @@ -39,12 +39,25 @@ // fit (see CSLSMapData::add overload). const int DEFAULT_MAX_DATA_SIZE = 8 * 1024 * 1024; +// Monotonic-max update of an atomic without a lock. Used off the read path to +// keep a high-water gauge; relaxed ordering is fine (the value is diagnostic and +// has no happens-before relationship with the data it describes). +static void atomic_store_max(std::atomic &slot, int64_t candidate) +{ + int64_t prev = slot.load(std::memory_order_relaxed); + while (candidate > prev && !slot.compare_exchange_weak(prev, candidate, std::memory_order_relaxed)) + { + } +} + CSLSRecycleArray::CSLSRecycleArray() { m_nDataSize = DEFAULT_MAX_DATA_SIZE; m_nWritePos = 0; m_nDataCount.store(0, std::memory_order_relaxed); m_overrun_count.store(0, std::memory_order_relaxed); + m_max_reader_backlog.store(0, std::memory_order_relaxed); + m_viewer_backpressure_events.store(0, std::memory_order_relaxed); m_last_read_time.store(sls_gettime_ms(), std::memory_order_relaxed); @@ -183,6 +196,13 @@ int CSLSRecycleArray::get(char *data, int size, SLSRecycleArrayID *read_id, int return SLS_OK; } + // Diagnostic high-water: how far behind the write head this reader was on + // entry (i.e. the backlog it is about to drain). A large value means the + // reader's next drains will burst out a catch-up, which a viewer perceives + // as a time-skip. Non-overrun path only — the overrun case is capped at the + // buffer size and already has its own counter. See get_max_reader_backlog. + atomic_store_max(m_max_reader_backlog, bytes_since_last_read); + SPDLOG_TRACE( "[{}] CSLSRecycleArray::get, read_id->nReadPos={:d}, m_nWritePos={:d}, m_nDataCount={:d}, m_nDataSize={:d}.", fmt::ptr(this), read_id->nReadPos, m_nWritePos, cur_data_count, m_nDataSize); @@ -260,3 +280,34 @@ int64_t CSLSRecycleArray::get_last_read_time() { return m_last_read_time.load(std::memory_order_acquire); } + +int64_t CSLSRecycleArray::get_reader_backlog(const SLSRecycleArrayID *read_id) const +{ + if (NULL == read_id || read_id->bFirst) + return 0; + int64_t behind = m_nDataCount.load(std::memory_order_relaxed) - read_id->nDataCount; + if (behind < 0) + behind = 0; + if (behind > m_nDataSize) + behind = m_nDataSize; + return behind; +} + +int64_t CSLSRecycleArray::get_max_reader_backlog(bool clear) +{ + if (clear) + return m_max_reader_backlog.exchange(0, std::memory_order_relaxed); + return m_max_reader_backlog.load(std::memory_order_relaxed); +} + +void CSLSRecycleArray::report_viewer_backpressure() +{ + m_viewer_backpressure_events.fetch_add(1, std::memory_order_relaxed); +} + +int64_t CSLSRecycleArray::get_viewer_backpressure_events(bool clear) +{ + if (clear) + return m_viewer_backpressure_events.exchange(0, std::memory_order_relaxed); + return m_viewer_backpressure_events.load(std::memory_order_relaxed); +} diff --git a/src/core/SLSRecycleArray.hpp b/src/core/SLSRecycleArray.hpp index d0de092..d475b5c 100644 --- a/src/core/SLSRecycleArray.hpp +++ b/src/core/SLSRecycleArray.hpp @@ -70,6 +70,30 @@ class CSLSRecycleArray return m_overrun_count.load(std::memory_order_relaxed); } + // How many bytes this specific reader is behind the write head right now, + // i.e. the backlog it would burst out on its next drain. 0 for a reader + // that has not yet anchored (bFirst) or has caught up. Clamped to the + // buffer size because a reader further behind than that has already been + // (or is about to be) overrun-resynced. Lock-free: m_nDataCount is atomic + // and read_id is owned by the calling worker thread. Diagnostic only — this + // is the metric that answers "is SLS holding a catch-up burst for a viewer". + int64_t get_reader_backlog(const SLSRecycleArrayID *read_id) const; + + // High-water of get_reader_backlog seen across ALL readers since the last + // clear. This is the per-stream signal an operator polls to see whether any + // viewer fell far enough behind that its eventual catch-up drain is bursty + // (visible to that viewer as a time-skip). clear=true resets it so /stats + // can report a per-interval peak instead of a lifetime max. + int64_t get_max_reader_backlog(bool clear = false); + + // Aggregate egress backpressure across the viewers reading this ring. The + // per-role counter lived on the player role, but /stats only enumerates + // publishers, so it was never surfaced. Players report here (a shared, + // publisher-visible object) on each EASYNCSND so publisher /stats can show + // real viewer backpressure instead of the publisher role's always-zero one. + void report_viewer_backpressure(); + int64_t get_viewer_backpressure_events(bool clear = false); + private: char *m_arrayData; int m_nDataSize; @@ -91,5 +115,11 @@ class CSLSRecycleArray std::atomic m_last_read_time{0}; std::atomic m_overrun_count; + // Diagnostic gauges (see the public accessors). Written off the read path in + // get() and by viewer roles reporting backpressure; read by the /stats HTTP + // thread. Atomic, relaxed — purely observational, no ordering requirements. + std::atomic m_max_reader_backlog{0}; + std::atomic m_viewer_backpressure_events{0}; + CSLSRWLock m_rwclock; }; diff --git a/src/core/SLSRole.cpp b/src/core/SLSRole.cpp index 5cb1a2d..76c8458 100644 --- a/src/core/SLSRole.cpp +++ b/src/core/SLSRole.cpp @@ -182,8 +182,8 @@ int CSLSRole::get_state(int64_t cur_time_ms) // Acquire pairs with the release store in request_kick(). if (m_kick_requested.load(std::memory_order_acquire)) { - spdlog::info("[{}] CSLSRole::get_state, kick requested for {}, fd={:d}, call invalid_srt.", fmt::ptr(this), - m_role_name, get_fd()); + spdlog::info("[{}] CSLSRole::get_state, kick requested for {} stream={}, fd={:d}, call invalid_srt.", + fmt::ptr(this), m_role_name, m_map_data_key, get_fd()); m_state = SLS_RS_INVALID; invalid_srt(); return m_state; @@ -204,7 +204,9 @@ int CSLSRole::get_state(int64_t cur_time_ms) int ret = get_sock_state(); if (SLS_ERROR == ret || SRTS_BROKEN == ret || SRTS_CLOSED == ret || SRTS_NONEXIST == ret) { - spdlog::info("[{}] CSLSRole::get_state, get_sock_state, ret={:d}, call invalid_srt.", fmt::ptr(this), ret); + spdlog::info("[{}] CSLSRole::get_state, {} stream={}, get_sock_state ret={:d} " + "(2=broken/peer-closed, 3=closed, 4=nonexist), call invalid_srt.", + fmt::ptr(this), m_role_name, m_map_data_key, ret); if (SRTS_BROKEN == ret || SRTS_CLOSED == ret || SRTS_NONEXIST == ret) { CSLSSrt::libsrt_neterrno(); @@ -670,6 +672,12 @@ int CSLSRole::handler_write_data() if (err_no == SRT_EASYNCSND) { m_send_backpressure_count.fetch_add(1, std::memory_order_relaxed); + // Also aggregate onto the shared publisher ring so + // /stats (which enumerates only publishers) surfaces + // real viewer backpressure instead of the publisher + // role's structurally-zero per-role counter. + if (m_map_data != NULL && strlen(m_map_data_key) != 0) + m_map_data->report_viewer_backpressure(m_map_data_key); // Stuck-viewer detection. The per-packet success // path above has already reset m_backpressure_stuck_since_ms @@ -682,6 +690,16 @@ int CSLSRole::handler_write_data() // means the viewer's link cannot sustain the stream // and they would otherwise hold a publisher-ring // read position open indefinitely. + // Sustained zero-progress backpressure means the viewer's + // link cannot carry the stream. We do NOT drop or re-anchor + // here: late packets are dropped at the SRT socket by + // TLPKTDROP (a clean, per-packet skip-forward), and the + // small hand-off ring keeps the reader near the live head + // so there is no multi-second backlog to replay. This + // timer only guards the pathological case — a viewer that + // makes no progress at all for 3x its latency window — by + // disconnecting it so it cannot pin a ring read position + // open forever. int64_t stuck_timeout_ms = backpressure_stuck_timeout_ms(); if (m_backpressure_stuck_since_ms == 0) { @@ -968,6 +986,20 @@ int64_t CSLSRole::get_ring_overrun_count() const return m_map_data->get_overrun_count(m_map_data_key); } +int64_t CSLSRole::get_max_reader_backlog(bool clear) const +{ + if (m_map_data == NULL || strlen(m_map_data_key) == 0) + return -1; + return m_map_data->get_max_reader_backlog(m_map_data_key, clear); +} + +int64_t CSLSRole::get_viewer_backpressure_events(bool clear) const +{ + if (m_map_data == NULL || strlen(m_map_data_key) == 0) + return -1; + return m_map_data->get_viewer_backpressure_events(m_map_data_key, clear); +} + int CSLSRole::init_bitrate_limiter(int max_bitrate_kbps, int violation_timeout_seconds, float spike_tolerance) { cleanup_bitrate_limiter(); diff --git a/src/core/SLSRole.hpp b/src/core/SLSRole.hpp index 22f607f..443bb1f 100644 --- a/src/core/SLSRole.hpp +++ b/src/core/SLSRole.hpp @@ -66,12 +66,17 @@ const int UNLIMITED_TIMEOUT = -1; // Max publisher-ring batches drained per handler_write_data() call. // Egress is driven by the worker's periodic pass (players are no longer -// permanently armed for SRT_EPOLL_OUT), so without an inner drain loop -// throughput would be capped at one DATA_BUFF_SIZE per worker wakeup. -// 8 * DATA_BUFF_SIZE (~168 KB) per call lets a catching-up viewer pull a -// large backlog quickly while still bounding how long one role holds the -// worker before it yields to the publisher read and the other roles. -const int MAX_EGRESS_BATCHES = 8; +// permanently armed for SRT_EPOLL_OUT), so a small inner drain loop keeps +// throughput from being capped at one DATA_BUFF_SIZE per worker wakeup at +// high bitrate. Deliberately small (2 * ~21 KB = ~42 KB per call, ~2x +// realtime headroom at 20 Mbps): a LIVE viewer must not "catch up" by +// bursting a large backlog out — that is exactly what a viewer perceives as +// a replay/rewind. When a viewer falls behind, we want it to stay behind and +// let the SRT socket's TLPKTDROP skip it forward per-packet, not to fire a +// ~20x burst of stale frames (the old value of 8 = ~168 KB/call did this). +// If high-bitrate viewers under-drain (ring backlog grows in /stats without +// a slow-link cause), nudge this up; do not raise it to "help" a slow viewer. +const int MAX_EGRESS_BATCHES = 2; /** * CSLSRole , the base of player, publisher and listener */ @@ -206,6 +211,18 @@ class CSLSRole // a map_data key. int64_t get_ring_overrun_count() const; + // Diagnostic gauges forwarded to this role's publisher ring. Meaningful on + // the publisher role (which owns the ring); every viewer of the stream + // updates the same underlying counters, so publisher /stats reflects the + // whole stream. Return -1 if not bound to a ring. get_max_reader_backlog is + // the bytes the furthest-behind viewer fell behind the write head (convert + // to ms with the stream bitrate); get_viewer_backpressure_events is the + // aggregate EASYNCSND count across viewers (the per-role counter below is + // never surfaced because /stats enumerates only publishers). `clear` resets + // the gauge so /stats can report a per-interval peak/delta. + int64_t get_max_reader_backlog(bool clear = false) const; + int64_t get_viewer_backpressure_events(bool clear = false) const; + // Count of times handler_write_data() hit SRT send-buffer // backpressure (errno EASYNCSND) on this role. Each event means a // viewer egress write was deferred to the next epoll cycle rather @@ -366,6 +383,7 @@ class CSLSRole int64_t scaled = (int64_t)m_latency * kBackpressureStuckLatencyMultiple; return scaled > kBackpressureStuckFloorMs ? scaled : kBackpressureStuckFloorMs; } + stat_info_t m_stat_info_base; std::shared_ptr> m_http_future; diff --git a/tests/test_recycle_array.cpp b/tests/test_recycle_array.cpp index f14b7fd..e51ebfb 100644 --- a/tests/test_recycle_array.cpp +++ b/tests/test_recycle_array.cpp @@ -103,6 +103,52 @@ TEST_CASE("CSLSRecycleArray: reader that fell behind by > ring size is resynced CHECK(ring.get_overrun_count() >= 1); } +TEST_CASE("CSLSRecycleArray: reader backlog and high-water track how far a reader is behind") +{ + CSLSRecycleArray ring; + ring.setSize(4096); + + SLSRecycleArrayID id = fresh_reader(); + // A reader that has not anchored reports no backlog (avoids a bogus huge + // value from the uninitialised nDataCount before the first snapshot). + CHECK(ring.get_reader_backlog(&id) == 0); + anchor(ring, id); + CHECK(ring.get_reader_backlog(&id) == 0); + + char chunk[100]; + std::memset(chunk, 'Q', sizeof(chunk)); + CHECK(ring.put(chunk, sizeof(chunk)) == (int)sizeof(chunk)); + + // Writer is now 100 bytes ahead of this reader: that is its backlog. + CHECK(ring.get_reader_backlog(&id) == 100); + + char out[256]; + CHECK(ring.get(out, sizeof(out), &id, 0) == 100); + + // Drained: caught up to the write head, backlog back to 0, but the + // high-water remembers the peak for /stats. + CHECK(ring.get_reader_backlog(&id) == 0); + CHECK(ring.get_max_reader_backlog(false) >= 100); + + // clear=true returns the peak and resets, so the next interval starts fresh. + CHECK(ring.get_max_reader_backlog(true) >= 100); + CHECK(ring.get_max_reader_backlog(false) == 0); +} + +TEST_CASE("CSLSRecycleArray: viewer backpressure events accumulate and clear") +{ + CSLSRecycleArray ring; + ring.setSize(1024); + + CHECK(ring.get_viewer_backpressure_events(false) == 0); + ring.report_viewer_backpressure(); + ring.report_viewer_backpressure(); + CHECK(ring.get_viewer_backpressure_events(false) == 2); + // clear=true drains the counter so /stats can report a per-interval delta. + CHECK(ring.get_viewer_backpressure_events(true) == 2); + CHECK(ring.get_viewer_backpressure_events(false) == 0); +} + TEST_CASE("CSLSRecycleArray: zero-length put is rejected") { CSLSRecycleArray ring; diff --git a/tools/sls-diag-scrape.sh b/tools/sls-diag-scrape.sh new file mode 100644 index 0000000..b359f50 --- /dev/null +++ b/tools/sls-diag-scrape.sh @@ -0,0 +1,71 @@ +#!/usr/bin/env bash +# Unattended /stats scraper for irl-srt-server jump/replay diagnosis. +# +# Polls the stats API on an interval and appends one CSV row per stream per +# tick, so you never have to watch live. When a streamer reports a jump at +# 19:11, open the CSV and read that stream's rows around 19:11. +# +# It also prints a stderr WARN line the moment any stream crosses a backlog or +# backpressure threshold, so `docker logs`/journal shows the interesting moments +# without you tailing anything. +# +# Survives publisher reconnects by design: the ring-side gauges reset when a +# streamer's encoder reconnects, but this scraper has already captured the +# pre-reset rows with timestamps, so the history is on disk regardless. +# +# Usage: +# ./sls-diag-scrape.sh # defaults: localhost:8181, 5s, ./sls-diag.csv +# STATS_URL=http://127.0.0.1:8181/stats INTERVAL=5 OUT=/var/log/sls-diag.csv ./sls-diag-scrape.sh +# +# Requires: curl, jq. + +set -euo pipefail + +STATS_URL="${STATS_URL:-http://127.0.0.1:8181/stats}" +INTERVAL="${INTERVAL:-5}" +OUT="${OUT:-./sls-diag.csv}" +# Alert thresholds. backlog in ms of playout; backpressure/overrun are deltas +# per tick (counter differences between two polls). +BACKLOG_MS_WARN="${BACKLOG_MS_WARN:-750}" +BACKPRESSURE_WARN="${BACKPRESSURE_WARN:-1}" + +if ! command -v jq >/dev/null; then echo "need jq" >&2; exit 1; fi + +# clear=1 so each poll returns per-interval peaks/deltas rather than lifetime +# totals. Do NOT run two clearing scrapers at once or they steal each other's +# deltas. +sep='?'; case "$STATS_URL" in *\?*) sep='&';; esac +POLL_URL="${STATS_URL}${sep}reset=1" + +if [ ! -f "$OUT" ]; then + echo "ts,stream,bitrateKbps,maxReaderBacklogMs,maxReaderBacklogBytes,sendBackpressure,ringOverruns,pktRcvDrop,msRcvBuf,rtt,uptime" > "$OUT" +fi + +echo "scraping $POLL_URL every ${INTERVAL}s -> $OUT (backlog warn >${BACKLOG_MS_WARN}ms)" >&2 + +while true; do + ts="$(date -u +%Y-%m-%dT%H:%M:%SZ)" + body="$(curl -fsS --max-time 4 "$POLL_URL" 2>/dev/null || true)" + if [ -n "$body" ]; then + # Flatten publishers{} into CSV rows. Field names match SLSManager.cpp. + echo "$body" | jq -r --arg ts "$ts" ' + (.publishers // {}) | to_entries[] | + [ $ts, .key, + (.value.bitrate // 0), + (.value.maxReaderBacklogMs // 0), + (.value.maxReaderBacklogBytes // 0), + (.value.sendBackpressure // 0), + (.value.ringOverruns // 0), + (.value.pktRcvDrop // 0), + (.value.msRcvBuf // 0), + (.value.rtt // 0), + (.value.uptime // 0) ] | @csv' >> "$OUT" 2>/dev/null || true + + # Stderr alerts on threshold crossings (shows up in docker/journal logs). + echo "$body" | jq -r --arg ts "$ts" --argjson bl "$BACKLOG_MS_WARN" --argjson bp "$BACKPRESSURE_WARN" ' + (.publishers // {}) | to_entries[] | + select((.value.maxReaderBacklogMs // 0) >= $bl or (.value.sendBackpressure // 0) >= $bp) | + "[\($ts)] SLS-DIAG WARN stream=\(.key) backlogMs=\(.value.maxReaderBacklogMs // 0) backpressure=\(.value.sendBackpressure // 0) overruns=\(.value.ringOverruns // 0) pktRcvDrop=\(.value.pktRcvDrop // 0)"' >&2 2>/dev/null || true + fi + sleep "$INTERVAL" +done