DeepStream smart record memory leak (DS 8.0, RTX 5070 16GB)

[nam278z01]

DeepStream smart record memory leak (DS 8.0, RTX 5070 16GB)

Summary

When enabling smart recording (smart_record: 2) with nvmultiurisrcbin, RSS grows steadily over hours. Disabling smart record keeps RAM stable. Leak persists even after app-level guards (SmartRecordHandlerV2) that prevent overlaps and clean state. Suspect nvurisrcbin/smart-record internal buffers not freed reliably.

Environment

• DeepStream: 8.0
• GPU: RTX 5070 16GB
• Driver: 580.82.07
• CUDA: (fill)
• OS: (fill, e.g., Ubuntu 24.04)
• GStreamer: from DeepStream 8.x
• App: Custom C++/CMake pipeline (not using deepstream-app); we build our own elements/blocks and integrate SmartRecordHandler in code (see source below).

Pipeline/config (minimal repro)

sources:
  - id: multi_rtsp_source
    type: rtsp
    mode: multi
    uris:
      - rtsp://<cam1>
      - rtsp://<cam2>
    backend_options:
      deepstream:
        smart_record: 2 # 0=off, 1=cloud-only, 2=cloud+local
        smart_rec_cache: 5 # seconds
        smart_rec_mode: 0 # audio+video
        smart_rec_dir_path: "/opt/lantana/rec"
        smart_rec_file_prefix: "lsr"
        smart_rec_default_duration: 20
        disable_audio: false
        max_batch_size: 4
        width: 1920
        height: 1080
        live_source: true
        drop_pipeline_eos: true
        batched_push_timeout: 40000
        latency: 400
        rtsp_reconnect_interval: 10
        rtsp_reconnect_attempts: -1
        max_latency: 3000000000
        sync_inputs: false

event_handlers:
  - id: detect_handler_1
    type: on_detect
    enable: true
    source_from: multi_rtsp_source
    signal_from: output_appsink # appsink emits new-sample
    signal_name: new-sample
    label_filter: [bike, bus, car, person, truck]
    record: true # triggers smart record
    mbroker_host: 192.168.1.99
    mbroker_port: 6979
    mbroker_channel: worker_lsr

outputs:
  - id: output_appsink
    type: appsink
    input_from: raw
    destination:
      appsink:
        emit_signals: true
        max_buffers: 5
        drop_buffers: true

Repro steps

1. Run pipeline with 4 RTSP sources (1080p).
2. Enable smart record as above (pre=5s cache, post=20s).
3. Let it run for several hours with regular detections (start-sr/stop-sr via handler).
4. Observe in docker stats: over ~1 hour it climbs roughly +500 MB.
5. Disable smart_record (set to 0) and rerun: RAM stays stable.

What happens

• With smart_record: 2: RSS keeps increasing; eventually OOM/kill.
• With smart_record: 0: RSS stable.
• No overlapping sessions at app layer; handler enforces safety gap and timeouts.

Expected

Smart record should release internal buffers after sr-done; RSS should stay bounded.

App-side safeguards already in place (SmartRecordHandlerV2)

• Prevent overlap: state machine IDLE/RECORDING/STOPPING + 2s safety gap.
• Force stop on expiry; STOPPING timeout 15s.
• Evict idle sources after 2 minutes; disconnect sr-done; drop cached nvurisrcbin refs if bin is recreated.
• Periodic cleanup every 500 frames; memory stats every 1000 frames.
• Redis publish uses stack strings; no long-lived GstSample refs.

Observations

• Leak only when smart record is enabled → points to nvurisrcbin smart-record internals.
• Appsink and handler maps remain bounded (logged); state maps do not grow.
• Overlap prevented; still leaks.
• Lowering cache/post reduces growth rate but leak persists.

Requests

• Is there a known memory leak with smart record in DS 8.x / nvurisrcbin?
• Any patches/workarounds (force flush, smart_rec_mode=video, disable audio, specific cache/duration constraints)?
• Guidance on safe start/stop patterns to avoid internal buffer retention.

Source code

#pragma once

/**
 * @file smart_record_handler_v2.hpp
 * @brief Optimized Smart Record Handler with improved memory safety
 *
 * Key improvements over v1:
 * - RAII pattern for all resources (no manual new/delete)
 * - GstElement weak references to avoid dangling pointers
 * - Thread-safe signal connection/disconnection
 * - Simplified state machine
 * - Proper cleanup on destruction
 */

#include "lantana/core/eventing/ievent_handler.hpp"
#include "lantana/core/utils/logger.hpp"

#include <gst/gst.h>
#include <gst/gstclock.h>
#include <gst/gstelement.h>
#include <gst-nvdssr.h>

#include <any>
#include <atomic>
#include <chrono>
#include <memory>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>

// Forward declaration
namespace lantana::infrastructure::messaging {
class RedisStreamProducer;
}

namespace lantana::backends::deepstream::events {

/**
 * @brief RAII wrapper for GstElement with automatic ref/unref
 */
class GstElementRef {
   public:
    GstElementRef() = default;

    explicit GstElementRef(GstElement* elem, bool add_ref = true) : element_(elem) {
        if (element_ && add_ref) {
            gst_object_ref(element_);
        }
    }

    ~GstElementRef() {
        reset();
    }

    // Move only
    GstElementRef(GstElementRef&& other) noexcept : element_(other.element_) {
        other.element_ = nullptr;
    }

    GstElementRef& operator=(GstElementRef&& other) noexcept {
        if (this != &other) {
            reset();
            element_ = other.element_;
            other.element_ = nullptr;
        }
        return *this;
    }

    // No copy
    GstElementRef(const GstElementRef&) = delete;
    GstElementRef& operator=(const GstElementRef&) = delete;

    void reset() {
        if (element_) {
            gst_object_unref(element_);
            element_ = nullptr;
        }
    }

    GstElement* get() const {
        return element_;
    }
    GstElement* operator->() const {
        return element_;
    }
    explicit operator bool() const {
        return element_ != nullptr && GST_IS_ELEMENT(element_);
    }

   private:
    GstElement* element_ = nullptr;
};

/**
 * @brief Recording session info - immutable after creation
 */
struct RecordingSessionV2 {
    uint32_t session_id;
    uint32_t source_id;
    std::string source_name;
    std::string pipeline_id;
    GstClockTime start_time;
    uint32_t duration_sec;
    std::chrono::system_clock::time_point wall_clock_start;

    bool is_expired(GstClockTime current_time) const {
        if (!GST_CLOCK_TIME_IS_VALID(start_time) || !GST_CLOCK_TIME_IS_VALID(current_time)) {
            return true;
        }
        return current_time >= (start_time + static_cast<GstClockTime>(duration_sec) * GST_SECOND);
    }
};

/**
 * @brief Configuration for SmartRecordHandlerV2
 */
struct SmartRecordConfigV2 {
    std::string source_from;
    std::string pipeline_id = "default";

    uint32_t pre_event_sec = 2;    // Seconds before detection (from cache)
    uint32_t post_event_sec = 15;  // Seconds after detection

    // RECORDING SAFETY & OVERLAP BEHAVIOR:
    // ====================================
    //
    // Safety Mechanism:
    //   - Handler enforces 2s minimum safety gap after recording completes
    //   - This prevents memory corruption from overlapping nvurisrcbin operations
    //   - Even with min_interval_sec=0, actual gap = max(min_interval_sec, 2s)
    //
    // Timeline Overlap (when min_interval_sec=0):
    //   Event A (t=0):  [-5s cache] [0s trigger] → [+10s post] = 15s video
    //   Event B (t=5):  Must wait until t=17s (15s duration + 2s safety gap)
    //                   [-5s cache] [17s trigger] → [+27s post] = 15s video
    //
    // Gap Behavior:
    //   min_interval_sec=0  → Actual gap: 2s (safety enforced)
    //   min_interval_sec=5  → Actual gap: 5s (user config)
    //   min_interval_sec=20 → Actual gap: 20s (user config)
    //
    // Trade-offs:
    //   ✅ Memory safe: Prevents corruption from concurrent recordings
    //   ✅ Reliable: No invalid free or double-free errors
    //   ⚠️  Minimum 2s gap: Cannot record continuously without gap
    //   ⚠️  May miss events: If object appears during safety gap
    uint32_t min_interval_sec = 3;

    // Maximum number of concurrent Smart Record sessions across all sources.
    // 0 = unlimited (default, preserves current behavior).
    // Setting to 1 can be used as a mitigation/workaround for platform instability
    // when multiple sources record concurrently.
    uint32_t max_concurrent_recordings = 0;

    std::unordered_set<std::string> target_labels;

    // Redis config
    std::string redis_host = "localhost";
    uint16_t redis_port = 6379;
    std::string redis_channel = "smart_record_events";

    // Source name mapping
    std::unordered_map<uint32_t, std::string> source_id_to_name;
};

/**
 * @brief Optimized Smart Record Handler V2
 *
 * Thread-safe, memory-safe implementation using RAII patterns.
 */
class SmartRecordHandlerV2 : public core::events::IEventHandler {
   public:
    SmartRecordHandlerV2();
    ~SmartRecordHandlerV2() override;

    // Non-copyable, non-movable (due to signal connections)
    SmartRecordHandlerV2(const SmartRecordHandlerV2&) = delete;
    SmartRecordHandlerV2& operator=(const SmartRecordHandlerV2&) = delete;
    SmartRecordHandlerV2(SmartRecordHandlerV2&&) = delete;
    SmartRecordHandlerV2& operator=(SmartRecordHandlerV2&&) = delete;

    // IEventHandler interface
    bool init_context(const std::string& config) override;
    bool init_context(const std::string& config, const std::any& extra_context) override;
    void destroy_context() override;
    void register_callback() override;
    std::string get_handler_name() const override;
    std::vector<std::string> get_supported_event_types() const override;
    void process_event(GstSample* sample, const std::string& event_type, std::any* batch_meta,
                       std::any* frame_meta, std::any* object_meta,
                       std::any* analytics_meta) override;

   private:
    // --- Internal types ---

    enum class RecordingState {
        IDLE,       // No recording, can start new
        RECORDING,  // Recording in progress, cannot start new
        STOPPING    // Recording ending, wait for cleanup
    };

    struct SourceState {
        GstClockTime last_record_start = GST_CLOCK_TIME_NONE;
        GstClockTime last_record_end = GST_CLOCK_TIME_NONE;  // When recording actually finished
        std::optional<RecordingSessionV2> active_session;
        RecordingState recording_state = RecordingState::IDLE;

        // When we have requested a stop (stop-sr emitted) but have not yet received sr-done.
        // Used to prevent starting a new recording in this transient state.
        std::optional<uint32_t> stopping_session_id;
        GstClockTime stop_request_time = GST_CLOCK_TIME_NONE;

        gulong signal_handler_id = 0;
    };

    // --- Helper methods ---
    void parse_config(const std::string& config_str);
    GstElementRef get_nvurisrcbin(uint32_t source_id);
    bool can_start_recording(uint32_t source_id, GstClockTime current_time);
    void start_recording(uint32_t source_id, uint64_t trigger_object_id, GstClockTime current_time);
    void cleanup_expired_sessions(GstClockTime current_time);
    void disconnect_all_signals();
    std::string get_source_name(uint32_t source_id) const;

    // Signal callback (static to match GObject signature)
    static void on_recording_done(GstElement* element, NvDsSRRecordingInfo* info,
                                  gpointer recording_user_data, gpointer user_data);

    // --- Members ---
    std::atomic<bool> initialized_{false};
    std::atomic<bool> shutting_down_{false};

    // Configuration (immutable after init)
    SmartRecordConfigV2 config_;

    // GStreamer elements (protected by mutex)
    mutable std::shared_mutex elements_mutex_;
    GstElementRef multiuribin_;
    std::unordered_map<uint32_t, GstElementRef> source_elements_;

    // State (protected by mutex)
    mutable std::mutex state_mutex_;
    std::unordered_map<uint32_t, SourceState> source_states_;

    // Redis producer (thread-safe, owned)
    std::unique_ptr<infrastructure::messaging::RedisStreamProducer> redis_producer_;

    // Lightweight counters for debugging long-running RSS growth.
    // These are intentionally atomics so we can increment from callbacks safely.
    std::atomic<uint64_t> sr_started_{0};
    std::atomic<uint64_t> sr_done_{0};
    std::atomic<uint64_t> sr_stop_emitted_{0};
    std::atomic<uint64_t> sr_stopping_timeouts_{0};
};

}  // namespace lantana::backends::deepstream::events

/**
 * @file smart_record_handler_v2.cpp
 * @brief Optimized Smart Record Handler implementation
 *
 * Memory-safe implementation with proper RAII, no manual new/delete,
 * and proper GStreamer object lifecycle management.
 */

#include "lantana/backends/deepstream/event_handlers/smart_record_handler_v2.hpp"
#include "lantana/infrastructure/messaging/redis_stream_producer.hpp"

#include <nvdsmeta.h>
#include <gstnvdsmeta.h>

#include <algorithm>
#include <chrono>
#include <cstring>
#include <sstream>

namespace lantana::backends::deepstream::events {

namespace {

// Minimum safety gap (seconds) after recording ends before allowing new recording
// This prevents memory corruption from overlapping nvurisrcbin operations
constexpr uint32_t RECORDING_SAFETY_GAP_SEC = 2;

// Key used to attach source_id to nvurisrcbin for reliable callback mapping
constexpr const char* kSourceIdDataKey = "lantana_source_id";

bool is_descendant_of(GstObject* obj, GstObject* ancestor) {
    if (!obj || !ancestor) {
        return false;
    }

    GstObject* current = GST_OBJECT(gst_object_ref(obj));
    while (current) {
        GstObject* parent = gst_object_get_parent(current);
        gst_object_unref(current);

        if (!parent) {
            return false;
        }

        if (parent == ancestor) {
            gst_object_unref(parent);
            return true;
        }

        current = parent;
    }

    return false;
}

/**
 * @brief Get current Unix timestamp in milliseconds as string
 */
std::string get_unix_millis() {
    auto now = std::chrono::system_clock::now();
    auto millis =
        std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
    return std::to_string(millis);
}

/**
 * @brief Get GStreamer system clock time safely
 */
GstClockTime get_gst_system_time() {
    GstClock* clock = gst_system_clock_obtain();
    if (!clock) {
        return GST_CLOCK_TIME_NONE;
    }
    GstClockTime time = gst_clock_get_time(clock);
    gst_object_unref(clock);
    return time;
}

/**
 * @brief Find nvurisrcbin element by source ID within a bin
 */
GstElement* find_nvurisrcbin_in_bin(GstElement* bin, uint32_t source_id,
                                    const std::string& source_from) {
    if (!bin || !GST_IS_BIN(bin)) {
        return nullptr;
    }

    // Try to find the creator bin
    const gchar* parent_name = GST_ELEMENT_NAME(bin);
    gchar* creator_name = g_strdup_printf("%s_creator", parent_name);
    GstElement* creator_bin = gst_bin_get_by_name(GST_BIN(bin), creator_name);
    g_free(creator_name);

    // Fallback to hardcoded name
    if (!creator_bin && !source_from.empty()) {
        std::string fallback_name = source_from + "_creator";
        creator_bin = gst_bin_get_by_name(GST_BIN(bin), fallback_name.c_str());
    }

    if (!creator_bin || !GST_IS_BIN(creator_bin)) {
        if (creator_bin) {
            gst_object_unref(creator_bin);
        }
        return nullptr;
    }

    // Find the specific nvurisrcbin
    gchar* target_name = g_strdup_printf("dsnvurisrcbin%u", source_id);
    GstElement* urisrcbin = gst_bin_get_by_name(GST_BIN(creator_bin), target_name);
    g_free(target_name);
    gst_object_unref(creator_bin);

    // Validate it's actually a nvurisrcbin
    if (urisrcbin) {
        GstElementFactory* factory = gst_element_get_factory(urisrcbin);
        if (factory) {
            const gchar* factory_name = gst_plugin_feature_get_name(GST_PLUGIN_FEATURE(factory));
            if (!g_str_has_prefix(factory_name, "nvurisrcbin")) {
                LOGW("Element found for source {} is not nvurisrcbin (type: {})", source_id,
                     factory_name);
                gst_object_unref(urisrcbin);
                return nullptr;
            }
        }
    }

    return urisrcbin;  // Caller owns the reference
}

}  // namespace

// ============================================================================
// SmartRecordHandlerV2 Implementation
// ============================================================================

SmartRecordHandlerV2::SmartRecordHandlerV2() {
    LOGD("SmartRecordHandlerV2: Constructor");
}

SmartRecordHandlerV2::~SmartRecordHandlerV2() {
    LOGD("SmartRecordHandlerV2: Destructor");
    destroy_context();
}

bool SmartRecordHandlerV2::init_context(const std::string& config) {
    LOGW("SmartRecordHandlerV2: init_context called without multiuribin - handler disabled");
    parse_config(config);
    return false;  // Cannot function without multiuribin
}

bool SmartRecordHandlerV2::init_context(const std::string& config, const std::any& extra_context) {
    if (initialized_.load()) {
        LOGW("SmartRecordHandlerV2: Already initialized");
        return true;
    }

    // Extract multiuribin from context
    GstElement* multiuribin_raw = nullptr;
    try {
        multiuribin_raw = std::any_cast<GstElement*>(extra_context);
    } catch (const std::bad_any_cast& e) {
        LOGE("SmartRecordHandlerV2: Failed to cast extra_context: {}", e.what());
        return false;
    }

    if (!multiuribin_raw || !GST_IS_BIN(multiuribin_raw)) {
        LOGE("SmartRecordHandlerV2: Invalid multiuribin element");
        return false;
    }

    // Parse configuration
    parse_config(config);

    // Store multiuribin reference
    {
        std::unique_lock lock(elements_mutex_);
        multiuribin_ = GstElementRef(multiuribin_raw, true);
    }

    // Get source_from from element if not set
    if (config_.source_from.empty()) {
        config_.source_from = GST_ELEMENT_NAME(multiuribin_raw);
    }

    // Initialize Redis producer
    try {
        redis_producer_ = std::make_unique<infrastructure::messaging::RedisStreamProducer>(
            config_.redis_host, config_.redis_port, config_.redis_channel);
        LOGI("SmartRecordHandlerV2: Redis producer initialized for {}:{}/{}", config_.redis_host,
             config_.redis_port, config_.redis_channel);
    } catch (const std::exception& e) {
        LOGE("SmartRecordHandlerV2: Failed to init Redis producer: {}", e.what());
        // Continue without Redis - not a fatal error
    }

    initialized_.store(true);
    LOGI("SmartRecordHandlerV2: Initialized successfully with multiuribin '{}'",
         config_.source_from);

    return true;
}

void SmartRecordHandlerV2::destroy_context() {
    if (!initialized_.exchange(false)) {
        return;  // Already destroyed or never initialized
    }

    LOGI("SmartRecordHandlerV2: Destroying context...");
    shutting_down_.store(true);

    // Disconnect all signal handlers first (must be done before releasing elements)
    disconnect_all_signals();

    // Clear state
    {
        std::lock_guard lock(state_mutex_);
        source_states_.clear();
    }

    // Release GStreamer elements
    {
        std::unique_lock lock(elements_mutex_);
        source_elements_.clear();
        multiuribin_.reset();
    }

    // Release Redis producer
    redis_producer_.reset();

    shutting_down_.store(false);
    LOGI("SmartRecordHandlerV2: Context destroyed");
}

void SmartRecordHandlerV2::register_callback() {
    // No-op: Signal connections are made lazily when starting recordings
}

std::string SmartRecordHandlerV2::get_handler_name() const {
    return "SmartRecordHandlerV2";
}

std::vector<std::string> SmartRecordHandlerV2::get_supported_event_types() const {
    return {"on_detect"};
}

void SmartRecordHandlerV2::process_event(GstSample* sample, const std::string& event_type,
                                         std::any* batch_meta_any, std::any* frame_meta_any,
                                         std::any* object_meta_any, std::any* analytics_meta_any) {
    // Early exit checks
    if (!initialized_.load() || shutting_down_.load()) {
        return;
    }

    if (event_type != "on_detect") {
        return;
    }

    // Check multiuribin is valid
    {
        std::shared_lock lock(elements_mutex_);
        if (!multiuribin_) {
            return;
        }
    }

    // Extract batch metadata
    NvDsBatchMeta* batch_meta = nullptr;
    try {
        if (batch_meta_any && batch_meta_any->has_value()) {
            batch_meta = std::any_cast<NvDsBatchMeta*>(*batch_meta_any);
        }
    } catch (const std::bad_any_cast&) {
        LOGE("SmartRecordHandlerV2: Failed to cast batch_meta");
        return;
    }

    if (!batch_meta) {
        return;
    }

    GstClockTime current_time = get_gst_system_time();
    if (!GST_CLOCK_TIME_IS_VALID(current_time)) {
        LOGW("SmartRecordHandlerV2: Invalid system time");
        return;
    }

    // EMERGENCY CLEANUP: Check if state maps are growing out of control
    static std::atomic<uint64_t> frame_count{0};
    ++frame_count;

    const size_t MAX_SOURCES = 100;  // Safety limit for source tracking
    {
        std::shared_lock elem_lock(elements_mutex_);
        std::lock_guard state_lock(state_mutex_);

        size_t max_size = std::max(source_elements_.size(), source_states_.size());
        if (max_size > MAX_SOURCES) {
            LOGW(
                "SmartRecordHandlerV2: EMERGENCY - source tracking exceeded limit ({}), forcing "
                "cleanup",
                max_size);
            LOGW("  cached_elements={}, tracked_sources={}", source_elements_.size(),
                 source_states_.size());
        }
    }

    // Cleanup expired sessions MORE FREQUENTLY (every 500 frames instead of on-demand)
    if (frame_count % 500 == 0) {
        cleanup_expired_sessions(current_time);
    }

    // Log comprehensive memory stats periodically (every ~1000 frames)
    if (frame_count % 1000 == 0) {
        std::shared_lock elem_lock(elements_mutex_);
        std::lock_guard state_lock(state_mutex_);
        LOGI("SmartRecordHandlerV2: Memory stats - cached_elements={}, tracked_sources={}",
             source_elements_.size(), source_states_.size());

        // Count active recordings
        size_t active_count = 0;
        size_t idle_count = 0;
        for (const auto& [sid, state] : source_states_) {
            if (state.recording_state == RecordingState::RECORDING) {
                active_count++;
            } else if (state.recording_state == RecordingState::IDLE) {
                idle_count++;
            }
        }
        LOGI("  Recording states - active={}, idle={}, other={}", active_count, idle_count,
             source_states_.size() - active_count - idle_count);

        // Estimate memory usage
        size_t approx_mem_kb = (source_elements_.size() * 64) +  // GstElement refs + overhead
                               (source_states_.size() * 128);    // RecordingState structs
        LOGI("  Estimated handler memory: ~{} KB", approx_mem_kb / 1024);
    }

    // Process frames
    for (NvDsFrameMetaList* l_frame = batch_meta->frame_meta_list; l_frame != nullptr;
         l_frame = l_frame->next) {
        NvDsFrameMeta* frame_meta = static_cast<NvDsFrameMeta*>(l_frame->data);
        if (!frame_meta) {
            continue;
        }

        uint32_t source_id = frame_meta->source_id;

        // Process objects in frame
        for (NvDsObjectMetaList* l_obj = frame_meta->obj_meta_list; l_obj != nullptr;
             l_obj = l_obj->next) {
            NvDsObjectMeta* obj_meta = static_cast<NvDsObjectMeta*>(l_obj->data);
            if (!obj_meta || !obj_meta->obj_label) {
                continue;
            }

            std::string label(obj_meta->obj_label);

            // Check if this label is in our target list
            if (!config_.target_labels.empty() &&
                config_.target_labels.find(label) == config_.target_labels.end()) {
                continue;
            }

            // Check if we can start recording for this source
            if (!can_start_recording(source_id, current_time)) {
                continue;
            }

            // Start recording
            LOGI("SmartRecordHandlerV2: Triggering record for source {} by object {} ({})",
                 source_id, obj_meta->object_id, label);

            start_recording(source_id, obj_meta->object_id, current_time);

            // Only one recording trigger per frame per source
            break;
        }
    }
}

// ============================================================================
// Private Methods
// ============================================================================

void SmartRecordHandlerV2::parse_config(const std::string& config_str) {
    LOGD("SmartRecordHandlerV2: Parsing config: {}", config_str);

    // Helper lambda to extract value
    auto get_value = [&config_str](const std::string& key) -> std::string {
        size_t key_pos = config_str.find(key);
        if (key_pos == std::string::npos) {
            return "";
        }
        size_t eq_pos = config_str.find('=', key_pos);
        if (eq_pos == std::string::npos) {
            return "";
        }
        size_t start = eq_pos + 1;
        size_t end = config_str.find(';', start);
        if (end == std::string::npos) {
            end = config_str.length();
        }
        return config_str.substr(start, end - start);
    };

    // Parse string values
    std::string val;

    val = get_value("source_from=");
    if (!val.empty())
        config_.source_from = val;

    val = get_value("pipeline_id=");
    if (!val.empty())
        config_.pipeline_id = val;

    val = get_value("mbroker_host=");
    if (!val.empty())
        config_.redis_host = val;

    val = get_value("mbroker_channel=");
    if (!val.empty())
        config_.redis_channel = val;

    // Parse numeric values
    val = get_value("mbroker_port=");
    if (!val.empty()) {
        try {
            config_.redis_port = static_cast<uint16_t>(std::stoul(val));
        } catch (...) {
        }
    }

    val = get_value("pre=");
    if (!val.empty()) {
        try {
            config_.pre_event_sec = std::stoul(val);
        } catch (...) {
        }
    }

    val = get_value("post=");
    if (!val.empty()) {
        try {
            config_.post_event_sec = std::stoul(val);
        } catch (...) {
        }
    }

    val = get_value("min_interval=");
    if (!val.empty()) {
        try {
            config_.min_interval_sec = std::stoul(val);
        } catch (...) {
        }
    }

    val = get_value("max_concurrent_recordings=");
    if (!val.empty()) {
        try {
            config_.max_concurrent_recordings = std::stoul(val);
        } catch (...) {
        }
    }

    // Parse class_ids/labels
    val = get_value("class_ids=");
    if (!val.empty()) {
        std::istringstream iss(val);
        std::string label;
        while (std::getline(iss, label, ',')) {
            // Trim whitespace
            label.erase(0, label.find_first_not_of(" \t"));
            label.erase(label.find_last_not_of(" \t") + 1);
            if (!label.empty()) {
                config_.target_labels.insert(label);
            }
        }
    }

    // Parse source_names
    val = get_value("source_names=");
    if (!val.empty()) {
        std::istringstream iss(val);
        std::string name;
        uint32_t idx = 0;
        while (std::getline(iss, name, ',')) {
            name.erase(0, name.find_first_not_of(" \t"));
            name.erase(name.find_last_not_of(" \t") + 1);
            if (!name.empty()) {
                config_.source_id_to_name[idx] = name;
            }
            ++idx;
        }
    }

    // Log parsed config
    std::string labels_str;
    for (const auto& l : config_.target_labels) {
        if (!labels_str.empty())
            labels_str += ",";
        labels_str += l;
    }

    LOGI(
        "SmartRecordHandlerV2: Config - source_from={}, pre={}s, post={}s, "
        "min_interval={}s, max_concurrent_recordings={}, labels=[{}], redis={}:{}/{}",
        config_.source_from, config_.pre_event_sec, config_.post_event_sec,
        config_.min_interval_sec, config_.max_concurrent_recordings, labels_str, config_.redis_host,
        config_.redis_port, config_.redis_channel);
}

GstElementRef SmartRecordHandlerV2::get_nvurisrcbin(uint32_t source_id) {
    // Check cache first
    {
        std::shared_lock lock(elements_mutex_);
        auto it = source_elements_.find(source_id);
        if (it != source_elements_.end() && it->second) {
            // Return cached element only if it still belongs to our multiuribin.
            // On RTSP reconnect, DeepStream may recreate nvurisrcbin and remove the old element
            // from the bin; holding a ref would keep the old element alive and leak memory.
            if (multiuribin_ &&
                is_descendant_of(GST_OBJECT(it->second.get()), GST_OBJECT(multiuribin_.get()))) {
                return GstElementRef(it->second.get(), true);
            }
        }
    }

    // If cached element exists but is no longer in multiuribin, drop it + disconnect its signal.
    {
        std::unique_lock elem_lock(elements_mutex_);
        auto it = source_elements_.find(source_id);
        if (it != source_elements_.end() && it->second) {
            bool still_attached = multiuribin_ && is_descendant_of(GST_OBJECT(it->second.get()),
                                                                   GST_OBJECT(multiuribin_.get()));

            if (!still_attached) {
                LOGW(
                    "SmartRecordHandlerV2: Cached nvurisrcbin for source {} is stale (likely "
                    "reconnect) - dropping cache",
                    source_id);

                {
                    std::lock_guard state_lock(state_mutex_);
                    auto st_it = source_states_.find(source_id);
                    if (st_it != source_states_.end() && st_it->second.signal_handler_id != 0) {
                        g_signal_handler_disconnect(it->second.get(),
                                                    st_it->second.signal_handler_id);
                        st_it->second.signal_handler_id = 0;
                    }
                }

                source_elements_.erase(it);
            }
        }
    }

    // Find and cache
    GstElement* urisrcbin = nullptr;
    {
        std::shared_lock lock(elements_mutex_);
        if (!multiuribin_) {
            return GstElementRef();
        }
        urisrcbin = find_nvurisrcbin_in_bin(multiuribin_.get(), source_id, config_.source_from);
    }

    if (!urisrcbin) {
        LOGW("SmartRecordHandlerV2: Could not find nvurisrcbin for source {}", source_id);
        return GstElementRef();
    }

    // Cache it (transfer ownership to cache, get a new ref for return)
    {
        std::unique_lock lock(elements_mutex_);
        // Double-check
        auto it = source_elements_.find(source_id);
        if (it != source_elements_.end() && it->second) {
            gst_object_unref(urisrcbin);  // Already cached
            return GstElementRef(it->second.get(), true);
        }
        source_elements_[source_id] = GstElementRef(urisrcbin, false);  // Takes ownership
    }

    // Attach source_id to element so callback can map reliably even if cache changes
    g_object_set_data(G_OBJECT(urisrcbin), kSourceIdDataKey, GUINT_TO_POINTER(source_id));

    // Connect signal for this source
    {
        std::lock_guard lock(state_mutex_);
        auto& state = source_states_[source_id];
        if (state.signal_handler_id == 0) {
            state.signal_handler_id = g_signal_connect(
                urisrcbin, "sr-done", G_CALLBACK(SmartRecordHandlerV2::on_recording_done), this);
            LOGD("SmartRecordHandlerV2: Connected sr-done signal for source {}", source_id);
        }
    }

    // Return a new reference to the cached element
    return GstElementRef(urisrcbin, true);
}

bool SmartRecordHandlerV2::can_start_recording(uint32_t source_id, GstClockTime current_time) {
    std::lock_guard lock(state_mutex_);

    if (config_.max_concurrent_recordings > 0) {
        uint32_t active_count = 0;
        for (const auto& [sid, st] : source_states_) {
            if (st.recording_state != RecordingState::IDLE) {
                ++active_count;
            }
        }
        if (active_count >= config_.max_concurrent_recordings) {
            LOGT("SmartRecordHandlerV2: Skipping - max concurrent recordings reached ({})",
                 config_.max_concurrent_recordings);
            return false;
        }
    }

    auto it = source_states_.find(source_id);
    if (it == source_states_.end()) {
        return true;  // No state yet = can start
    }

    const auto& state = it->second;

    // Check #0: Recording state (CRITICAL for memory safety)
    // Purpose: Prevent overlapping recordings that cause memory corruption
    // Must wait for previous recording to fully complete and cleanup
    if (state.recording_state != RecordingState::IDLE) {
        LOGT("SmartRecordHandlerV2: Skipping - recording state is not IDLE (source {})", source_id);
        return false;
    }

    // Check #0.5: Safety gap after recording ends (CRITICAL)
    // Purpose: GStreamer/nvurisrcbin needs time to cleanup internal state
    // Even with min_interval_sec=0, we enforce minimum 2s gap for stability
    if (GST_CLOCK_TIME_IS_VALID(state.last_record_end)) {
        GstClockTime elapsed = current_time - state.last_record_end;
        GstClockTime safety_gap = RECORDING_SAFETY_GAP_SEC * GST_SECOND;
        if (elapsed < safety_gap) {
            LOGT("SmartRecordHandlerV2: Skipping due to safety gap ({}s < {}s)",
                 elapsed / GST_SECOND, RECORDING_SAFETY_GAP_SEC);
            return false;
        }
    }

    // Check #1: Minimum interval between recordings (user-configured throttle)
    // Purpose: Additional user-defined gap between recordings
    // Note: Actual minimum gap = max(min_interval_sec, RECORDING_SAFETY_GAP_SEC)
    if (GST_CLOCK_TIME_IS_VALID(state.last_record_start)) {
        GstClockTime elapsed = current_time - state.last_record_start;
        GstClockTime min_interval = config_.min_interval_sec * GST_SECOND;
        if (elapsed < min_interval) {
            LOGT("SmartRecordHandlerV2: Skipping due to min_interval ({}s < {}s)",
                 elapsed / GST_SECOND, config_.min_interval_sec);
            return false;
        }
    }

    return true;
}

void SmartRecordHandlerV2::start_recording(uint32_t source_id, uint64_t trigger_object_id,
                                           GstClockTime current_time) {
    // Get nvurisrcbin element
    GstElementRef urisrcbin = get_nvurisrcbin(source_id);
    if (!urisrcbin) {
        LOGE("SmartRecordHandlerV2: Cannot get nvurisrcbin for source {}", source_id);
        return;
    }

    // DeepStream Smart Record semantics: startTime = seconds BEFORE now, duration = seconds AFTER
    // now. If we pass (pre + post) as duration, recordings become ~pre seconds longer than expected
    // and can enlarge internal buffering requirements.
    uint32_t start_time_sec = config_.pre_event_sec;
    uint32_t duration_sec = config_.post_event_sec;
    uint32_t total_duration_sec = start_time_sec + duration_sec;
    if (total_duration_sec == 0) {
        // Minimum 1 second after-event duration.
        duration_sec = 1;
        total_duration_sec = start_time_sec + duration_sec;
    }

    std::string source_name = get_source_name(source_id);

    // Create session info
    RecordingSessionV2 session;
    // DeepStream assigns the actual session id via the start-sr out-parameter.
    // We set placeholder here and update immediately after emitting start-sr.
    session.session_id = 0;
    session.source_id = source_id;
    session.source_name = source_name;
    session.pipeline_id = config_.pipeline_id;
    session.start_time = current_time;
    session.duration_sec = total_duration_sec;
    session.wall_clock_start = std::chrono::system_clock::now();

    // Update state
    {
        std::lock_guard lock(state_mutex_);
        auto& state = source_states_[source_id];
        state.last_record_start = current_time;
        state.active_session = session;
        state.recording_state = RecordingState::RECORDING;  // Mark as recording
    }

    // Emit start-sr signal
    // Note: We pass nullptr for user_data since we track sessions internally
    NvDsSRSessionId sr_session_id = 0;
    g_signal_emit_by_name(urisrcbin.get(), "start-sr", &sr_session_id, start_time_sec, duration_sec,
                          nullptr);  // No user_data - avoids memory management issues

    const uint32_t ds_session_id = static_cast<uint32_t>(sr_session_id);
    {
        std::lock_guard lock(state_mutex_);
        auto it = source_states_.find(source_id);
        if (it != source_states_.end() && it->second.active_session) {
            it->second.active_session->session_id = ds_session_id;
        }
    }

    LOGI(
        "SmartRecordHandlerV2: Started recording - session={}, source={}, startTime={}s, "
        "duration={}s (total={}s)",
        ds_session_id, source_id, start_time_sec, duration_sec, total_duration_sec);

    {
        const uint64_t started = ++sr_started_;
        if (started % 100 == 0) {
            LOGI(
                "SmartRecordHandlerV2: SR stats - started={}, done={}, stop_emitted={}, "
                "stopping_timeouts={}",
                started, sr_done_.load(), sr_stop_emitted_.load(), sr_stopping_timeouts_.load());
        }
    }

    // Publish to Redis (compatible with V1 field names)
    if (redis_producer_) {
        // CRITICAL: Store string conversions in variables to ensure lifetime
        // std::to_string() returns temporary - must persist until send_message() completes
        std::string sid_str = std::to_string(source_id);
        std::string session_id_str = std::to_string(ds_session_id);
        std::string start_time_str = std::to_string(config_.pre_event_sec);
        std::string duration_str =
            std::to_string(total_duration_sec * 1000);  // V1 sends milliseconds
        std::string trigger_obj_str = std::to_string(trigger_object_id);
        std::string event_ts_str = get_unix_millis();

        std::vector<std::pair<std::string, std::string>> fields = {
            {"event", "record_started"},
            {"pid", config_.pipeline_id},
            {"sid", sid_str},
            {"sname", source_name},
            {"session_id", session_id_str},
            {"start_time", start_time_str},
            {"duration", duration_str},
            {"trigger_obj", trigger_obj_str},
            {"event_ts", event_ts_str},
        };
        redis_producer_->send_message(fields);
    }
}

void SmartRecordHandlerV2::cleanup_expired_sessions(GstClockTime current_time) {
    struct RemoveInfo {
        uint32_t source_id;
        gulong signal_handler_id;
    };

    struct StopInfo {
        uint32_t source_id;
        uint32_t session_id;
    };

    std::vector<RemoveInfo> to_remove;
    std::vector<StopInfo> to_stop;

    constexpr GstClockTime STOPPING_TIMEOUT = 15 * GST_SECOND;

    // Phase 1: decide what to evict + capture signal handler ids while holding state_mutex_
    {
        std::lock_guard lock(state_mutex_);

        for (auto& [source_id, state] : source_states_) {
            // If we're in STOPPING for too long, force-reset to IDLE.
            // This is a last resort to avoid permanently blocking a source if sr-done is lost.
            if (state.recording_state == RecordingState::STOPPING &&
                GST_CLOCK_TIME_IS_VALID(state.stop_request_time)) {
                GstClockTime elapsed = current_time - state.stop_request_time;
                if (elapsed > STOPPING_TIMEOUT) {
                    LOGW(
                        "SmartRecordHandlerV2: STOPPING timeout - forcing state to IDLE (source "
                        "{})",
                        source_id);
                    ++sr_stopping_timeouts_;
                    state.stopping_session_id.reset();
                    state.stop_request_time = GST_CLOCK_TIME_NONE;
                    state.active_session.reset();
                    state.recording_state = RecordingState::IDLE;
                    state.last_record_end = current_time;
                }
            }

            // Clear expired active sessions
            if (state.active_session && state.active_session->is_expired(current_time)) {
                LOGD("SmartRecordHandlerV2: Clearing expired session for source {}", source_id);

                // Best-effort stop to avoid stuck SR sessions if sr-done never arrives.
                if (state.active_session->session_id != 0) {
                    const uint32_t session_id =
                        static_cast<uint32_t>(state.active_session->session_id);
                    to_stop.push_back(StopInfo{source_id, session_id});

                    // IMPORTANT: Do NOT mark IDLE before stop-sr completes.
                    // Starting a new recording in this window can overlap and destabilize
                    // DeepStream smart record.
                    state.stopping_session_id = session_id;
                    state.stop_request_time = current_time;
                    state.recording_state = RecordingState::STOPPING;
                } else {
                    // No session id yet; treat as finished.
                    state.recording_state = RecordingState::IDLE;
                    state.last_record_end = current_time;
                }
                state.active_session.reset();
            }

            // Mark very old inactive entries for removal (prevent unbounded growth)
            // CRITICAL: Reduced from 5min to 2min for aggressive cleanup
            constexpr GstClockTime MAX_IDLE_TIME = 2 * 60 * GST_SECOND;  // 2 minutes (was 5)
            if (state.recording_state == RecordingState::IDLE && !state.active_session &&
                GST_CLOCK_TIME_IS_VALID(state.last_record_end)) {
                GstClockTime idle_time = current_time - state.last_record_end;
                if (idle_time > MAX_IDLE_TIME) {
                    to_remove.push_back(RemoveInfo{source_id, state.signal_handler_id});
                }
            }
        }

        // Remove states now (we'll disconnect signals + drop cached elements in phase 2)
        for (const auto& info : to_remove) {
            LOGI("SmartRecordHandlerV2: Evicting inactive source {} from state cache",
                 info.source_id);
            source_states_.erase(info.source_id);
        }
    }

    if (to_remove.empty()) {
        // Still may need to stop expired sessions.
    }

    // Phase 2a: emit stop-sr for expired sessions (no state_mutex_ held)
    for (const auto& info : to_stop) {
        GstElementRef urisrcbin = get_nvurisrcbin(info.source_id);
        if (urisrcbin) {
            NvDsSRSessionId sr_session_id = static_cast<NvDsSRSessionId>(info.session_id);
            g_signal_emit_by_name(urisrcbin.get(), "stop-sr", sr_session_id);
            ++sr_stop_emitted_;
            LOGD("SmartRecordHandlerV2: Emitted stop-sr for expired session {} (source {})",
                 info.session_id, info.source_id);
        }
    }

    if (to_remove.empty()) {
        return;
    }

    // Phase 2: disconnect signals and drop cached elements while holding elements_mutex_
    // Lock order across this class is elements_mutex_ -> state_mutex_. We intentionally
    // do NOT acquire state_mutex_ here to avoid lock inversion.
    {
        std::unique_lock elem_lock(elements_mutex_);
        for (const auto& info : to_remove) {
            auto elem_it = source_elements_.find(info.source_id);
            if (elem_it != source_elements_.end() && elem_it->second) {
                if (info.signal_handler_id != 0) {
                    g_signal_handler_disconnect(elem_it->second.get(), info.signal_handler_id);
                    LOGD("SmartRecordHandlerV2: Disconnected sr-done for evicted source {}",
                         info.source_id);
                }
                source_elements_.erase(elem_it);
                LOGI("SmartRecordHandlerV2: Removed cached element for evicted source {}",
                     info.source_id);
            }
        }
    }
}

void SmartRecordHandlerV2::disconnect_all_signals() {
    std::unique_lock elem_lock(elements_mutex_);
    std::lock_guard state_lock(state_mutex_);

    for (auto& [source_id, element] : source_elements_) {
        if (element) {
            auto state_it = source_states_.find(source_id);
            if (state_it != source_states_.end() && state_it->second.signal_handler_id != 0) {
                g_signal_handler_disconnect(element.get(), state_it->second.signal_handler_id);
                state_it->second.signal_handler_id = 0;
                LOGD("SmartRecordHandlerV2: Disconnected signal for source {}", source_id);
            }
        }
    }
}

std::string SmartRecordHandlerV2::get_source_name(uint32_t source_id) const {
    auto it = config_.source_id_to_name.find(source_id);
    if (it != config_.source_id_to_name.end()) {
        return it->second;
    }
    return "source_" + std::to_string(source_id);
}

// Static callback
void SmartRecordHandlerV2::on_recording_done(GstElement* element, NvDsSRRecordingInfo* info,
                                             gpointer recording_user_data, gpointer user_data) {
    auto* handler = static_cast<SmartRecordHandlerV2*>(user_data);
    if (!handler || handler->shutting_down_.load()) {
        return;
    }

    if (!info) {
        LOGW("SmartRecordHandlerV2: on_recording_done called with null info");
        return;
    }

    LOGI("SmartRecordHandlerV2: Recording done - file={}, width={}, height={}, duration={}",
         info->filename ? info->filename : "null", info->width, info->height, info->duration);

    {
        const uint64_t done = ++handler->sr_done_;
        if (done % 100 == 0) {
            LOGI(
                "SmartRecordHandlerV2: SR stats - started={}, done={}, stop_emitted={}, "
                "stopping_timeouts={}",
                handler->sr_started_.load(), done, handler->sr_stop_emitted_.load(),
                handler->sr_stopping_timeouts_.load());
        }
    }

    // Find which source this is from and get session_id
    uint32_t source_id = 0;
    uint32_t session_id = static_cast<uint32_t>(info->sessionId);
    bool found = false;

    // Preferred: read source_id directly from element qdata (safe across reconnect/cache eviction)
    if (element) {
        gpointer sid_ptr = g_object_get_data(G_OBJECT(element), kSourceIdDataKey);
        if (sid_ptr) {
            source_id = GPOINTER_TO_UINT(sid_ptr);
            found = true;
        }
    }

    {
        if (!found) {
            std::shared_lock lock(handler->elements_mutex_);
            for (const auto& [sid, elem] : handler->source_elements_) {
                if (elem.get() == element) {
                    source_id = sid;
                    found = true;
                    break;
                }
            }
        }
    }

    // Get session_id from active session if available
    if (found) {
        std::lock_guard lock(handler->state_mutex_);
        auto it = handler->source_states_.find(source_id);
        if (it != handler->source_states_.end()) {
            if (it->second.active_session) {
                if (it->second.active_session->session_id != session_id) {
                    LOGW(
                        "SmartRecordHandlerV2: sr-done session mismatch (source {}): active={}, "
                        "info={}",
                        source_id, it->second.active_session->session_id, session_id);
                }
                // Clear active session - recording is done (free memory)
                it->second.active_session.reset();
            }

            // Clear STOPPING bookkeeping (if any)
            it->second.stopping_session_id.reset();
            it->second.stop_request_time = GST_CLOCK_TIME_NONE;

            // Update recording state to IDLE - recording is done
            it->second.recording_state = RecordingState::IDLE;

            // Get current time for safety gap tracking
            GstClockTime end_time = get_gst_system_time();
            if (GST_CLOCK_TIME_IS_VALID(end_time)) {
                it->second.last_record_end = end_time;
            } else {
                // Fallback: use last_record_start + duration
                if (GST_CLOCK_TIME_IS_VALID(it->second.last_record_start)) {
                    it->second.last_record_end =
                        it->second.last_record_start +
                        (handler->config_.pre_event_sec + handler->config_.post_event_sec) *
                            GST_SECOND;
                }
            }

            LOGD("SmartRecordHandlerV2: Recording complete, state set to IDLE (source {})",
                 source_id);
        }
    }

    // Publish to Redis (compatible with V1 field names)
    if (handler->redis_producer_ && found) {
        // CRITICAL: Store string conversions to ensure lifetime
        std::string source_name = handler->get_source_name(source_id);
        std::string sid_str = std::to_string(source_id);
        std::string session_id_str = std::to_string(session_id);
        std::string width_str = info->width ? std::to_string(info->width) : "";
        std::string height_str = info->height ? std::to_string(info->height) : "";
        std::string filename_str = info->filename ? info->filename : "";
        std::string duration_str = std::to_string(info->duration);
        std::string event_ts_str = get_unix_millis();

        std::vector<std::pair<std::string, std::string>> fields = {
            {"event", "record_done"},
            {"pid", handler->config_.pipeline_id},
            {"sid", sid_str},
            {"sname", source_name},
            {"session_id", session_id_str},
            {"width", width_str},
            {"height", height_str},
            {"filename", filename_str},
            {"duration", duration_str},
            {"event_ts", event_ts_str},
        };
        handler->redis_producer_->send_message(fields);
    }
}

}  // namespace lantana::backends::deepstream::events

[fanzh]

@nam278z01

Suspect nvurisrcbin/smart-record internal buffers not freed reliably.

If you suspect there is memory leak in nvurisrcbin, you can you use the following simple project to test nvurisrcbin separately.
test.zip (1.9 KB)
On DS8.0, there is no known memory leak with smart record.

[fanzh]

Sorry for the late reply, Is this still an DeepStream issue to support? Thanks!
If you still observe memory leaks,Please provide more logs by nvmemstat, which can save the memory usage at intervals. Please provide more valgrind logs by this method vagrid, it can give memory details.

[nam278z01]

I rewrote the sample source (from the zip file you provided) and ran it inside the DeepStream 8.0 container. I can clearly observe that the container’s RAM usage keeps increasing over time. Specifically, over about 5 minutes the container’s memory increased by ~50 MB and is still continuing to increase

// basic-tutorial-1.c
#include <gst/gst.h>
#include <gst/app/gstappsrc.h>
#include <gst/app/gstappsink.h>
#include "nvdsmeta.h"
#include "gstnvdsmeta.h"
#include <stdio.h>
#include <glib.h>


GstElement *pipeline = NULL;
GstElement *muti = NULL;
GstElement *urisrc[4] = {NULL, NULL, NULL, NULL};  // Array to store 4 sources
int urisrc_count = 0;

static gboolean
bus_call (GstBus * bus, GstMessage * msg, gpointer data)
{
  GMainLoop *loop = (GMainLoop *) data;
  switch (GST_MESSAGE_TYPE (msg)) {
    case GST_MESSAGE_EOS:
      g_print ("End of stream\n");
      g_main_loop_quit (loop);
      break;
    case GST_MESSAGE_ERROR:{
      gchar *debug = NULL;
      GError *error = NULL;
      gst_message_parse_error (msg, &error, &debug);
      g_printerr ("ERROR from element %s: %s\n",
          GST_OBJECT_NAME (msg->src), error->message);
      if (debug)
        g_printerr ("Error details: %s\n", debug);
      g_free (debug);
      g_error_free (error);
      g_main_loop_quit (loop);
      break;
    }
    case GST_MESSAGE_STATE_CHANGED: {
      GstState old_state, new_state, pending_state;
      gst_message_parse_state_changed(msg, &old_state, &new_state, &pending_state);

      if (new_state == GST_STATE_PLAYING) {
        if (strncmp(GST_OBJECT_NAME (msg->src) ,"dsnvurisrcbin", strlen("dsnvurisrcbin")) == 0) {
            printf("dsnvurisrcbin play - storing source %d: %s\n", urisrc_count, GST_OBJECT_NAME (msg->src));
            if (urisrc_count < 4) {
                urisrc[urisrc_count] = GST_ELEMENT(msg->src);
                urisrc_count++;
            }
        }
      }
      break;
    }
    default:
      break;
  }
  return TRUE;
}

static gboolean smart_record_event_generator (gpointer data)
{
  printf("in smart_record_event_generator\n");

  // CRITICAL: Stop previous sessions first to prevent memory accumulation
  for (int source_id = 0; source_id < 4; source_id++) {
    if (urisrc[source_id] != NULL) {
      printf("Stopping previous smart record for source_id=%d\n", source_id);
      g_signal_emit_by_name(G_OBJECT (urisrc[source_id]), "stop-sr", NULL);
    }
  }

  // Wait for cleanup - MUST be at least 3s for 4 sources (like 1.c reference)
  printf("Waiting 3s for buffer cleanup...\n");
  g_usleep(3 * 1000000); // 3 seconds - same as 1.c

  // Start new recording sessions
  for (int source_id = 0; source_id < 4; source_id++) {
    if (urisrc[source_id] != NULL) {
      int sessId = 100 + source_id;
      printf("Starting smart record for source_id=%d, sessId=%d\n", source_id, sessId);
      // Call start-sr on individual dsnvurisrcbin element (not on muti)
      g_signal_emit_by_name(G_OBJECT (urisrc[source_id]), "start-sr", &sessId, 5, 20, NULL);
    } else {
      printf("WARNING: urisrc[%d] is NULL, skipping\n", source_id);
    }
  }

  return TRUE;
}

void *func(void *ptr) {
  g_usleep (5 * 1000000);  // Wait 5s for all sources to reach PLAYING state
  printf("in func - all sources ready, starting smart recording\n");

  // Trigger smart recording immediately for testing
  smart_record_event_generator(NULL);

  // Then trigger every 60 seconds (match 1.c interval)
  guint timer_id = g_timeout_add_seconds (60, smart_record_event_generator, NULL);

  return NULL;
}

int
main (int argc, char *argv[])
{
  GstBus *bus;
  GstMessage *msg;
  GstPad *tiler_src_pad = NULL;
  guint bus_watch_id = 0;
  GstElement *creator = NULL;
  /* Initialize GStreamer */
  gst_init (&argc, &argv);

  /* Build the pipeline */
  pipeline =
      gst_parse_launch
      ("nvmultiurisrcbin name=muti "
       "uri-list=rtsp://192.168.1.99:8554/view_cam_camera-ah,"
                "rtsp://192.168.1.99:8554/view_cam_camera-hanh-trinh,"
                "rtsp://192.168.1.99:8554/view_cam_djuong-5,"
                "rtsp://192.168.1.99:8554/view_cam_san "
       "sensor-id-list=0,1,2,3,4 "
       "num-extra-surfaces=3 "
       "dec-skip-frames=0 "
       "max-batch-size=4 "
       "live-source=1 "
       "sync-inputs=0 "
       "drop-frame-interval=0 "
       "max-latency=120000000 "
       "latency=400 "
       "batched-push-timeout=40000 "
       "width=1920 height=1080 "
       "drop-pipeline-eos=1 "
       "! fakesink",
      NULL);

  muti  = gst_bin_get_by_name(GST_BIN(pipeline), "muti");
  if (!muti) {
    g_printerr("ERROR: Could not get 'muti' element from pipeline\n");
    return -1;
  }
  printf("muti:%s\n", GST_OBJECT_NAME (muti));

  g_object_set (G_OBJECT (muti),
    "smart-record", 2,
    "smart-rec-dir-path", "/opt/lantana/lantana_data/rec",
    "smart-rec-file-prefix", "lsr",
    "smart-rec-cache", 2,
    "smart-rec-container", 0,  // 0=mp4, 1=mkv
    "smart-rec-mode", 0,       // 0=audio+video
    "smart-rec-default-duration", 20,
    NULL);

  printf("Smart recording configured: dir=/opt/lantana/lantana_data/rec, prefix=lsr\n");

  /* Start playing */
  gst_element_set_state (pipeline, GST_STATE_PLAYING);


  pthread_t tid;
  pthread_create(&tid, NULL, &func, NULL);
  GMainLoop *loop = g_main_loop_new(NULL, FALSE);

  /* we add a message handler */
  bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
  bus_watch_id = gst_bus_add_watch (bus, bus_call, loop);
  gst_object_unref (bus);


  g_main_loop_run(loop);



  /* Out of the main loop, clean up nicely */
  g_print ("Returned, stopping playback\n");
  gst_element_set_state (pipeline, GST_STATE_NULL);
  g_print ("Deleting pipeline\n");
  gst_object_unref (GST_OBJECT (pipeline));
  g_source_remove (bus_watch_id);
  g_main_loop_unref (loop);
  return 0;
}

#!/usr/bin/env bash
set -e

# thư mục chứa build.sh (lab/test)
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"

SRC="$SCRIPT_DIR/sr.c"
OUT="$SCRIPT_DIR/sr"

gcc "$SRC" -o "$OUT" \
  $(pkg-config --cflags --libs gstreamer-1.0) \
  -I/opt/nvidia/deepstream/deepstream/sources/includes \
  -L/opt/nvidia/deepstream/deepstream/lib \
  -lnvdsgst_meta \
  -lnvds_meta \
  -lgstapp-1.0

[nam278z01]

valgrind.log (83.4 KB)
I’ve attached the log file — please take a look

[nam278z01]

I see Valgrind reports only small leaks, but the container’s RAM usage still keeps increasing steadily over time

[fanzh]

@nam278z01

but the container’s RAM usage still keeps increasing steadily over time

Thanks for the sharing! rtsp receiving will take some memory due to ‘latency=400ms’, and smart record will take some memroy because of video cache. Hence, theoritically the memory utilization will increase at the beginning, then keep stable.
For simplification, you can only test one souce first. Please provide memory utilization log mentioned in my last commnet. Thanks! Please refer to this topic.

[yingliu]

There is no update from you for a period, assuming this is not an issue anymore. Hence we are closing this topic. If need further support, please open a new one. Thanks.