Error while removing appsink during runtime during runtime

Accelerated Computing Intelligent Video Analytics DeepStream SDK
1

Please provide complete information as applicable to your setup.

• Hardware Platform (Jetson / GPU) GPU
• DeepStream Version 6.3
• Issue Type( questions, new requirements, bugs) question

following is my deepstream pipeline

nvrurisrcbin ! pgi ! tracker ! nvvideoconvertr ! nvjpegenc ! queue ! appsink

       self.pipeline.add(source_bin)
        padname = "sink_%u" % stream_obj.ID
        sinkpad = self.streammux.get_request_pad(padname)
        if not sinkpad:
            logger.error("Unable to create sink pad bin ")
            return
        srcpad = source_bin.get_static_pad("src")
        if not srcpad:
            logger.error("Unable to create src pad bin \n")
        srcpad.link(sinkpad)

        logger.info(f"Adding downstream elements  for uri {stream_obj.uri} ID {stream_obj.ID}")
        # creating queue
        queue = self.make_element("queue", f"queue-{stream_obj.ID}")
        queue.set_property("leaky", 2)
        queue.set_property("max-size-buffers", 1)
        self.pipeline.add(queue)

        sink = self.make_element("appsink", f"sink-{stream_obj.ID}")
        sink.set_property("emit-signals", True)
        sink.set_property("max-buffers", 2)
        sink.set_property("drop", True)
        sink.set_property("sync", False)
        self.pipeline.add(sink)

        # connect tee -> queue -> appsink
        padname = "src_%u" % stream_obj.ID
        teesrcpad = self.tee.get_request_pad(padname)
        if not teesrcpad:
            logger.error("Unable to create demux src pad ")

        queuesinkpad = queue.get_static_pad("sink")
        if not queuesinkpad:
            logger.error("Unable to create queue sink pad ")
        teesrcpad.link(queuesinkpad)
        queue.link(sink)
        sinkpad = sink.get_static_pad("sink")

        sink.connect("new-sample", self.outbound_func, stream_obj)

        state_return = source_bin.set_state(Gst.State.PLAYING)
        queue.set_state(Gst.State.PLAYING)
        sink.set_state(Gst.State.PLAYING)

I get an error when I remove the appsink during runtime

 to_remove = ['source-bin', 'queue', 'sink']

        for el in to_remove:
            element_name = f'{el}-{stream_obj.ID}'
            try:
                logger.info(f"removing element {element_name}")
                element = self.pipeline.get_by_name(element_name)
                state_return = element.set_state(Gst.State.PAUSED)
                state_return = element.set_state(Gst.State.READY)
                state_return = element.set_state(Gst.State.NULL)
                logger.info(f'{element.get_state(0)}')
                logger.info(f"{element_name} set to NULL, state return :{state_return}")
                if el == 'source-bin' and sinkpad is not None:
                    sinkpad.send_event(Gst.Event.new_flush_stop(False))
                    self.streammux.release_request_pad(sinkpad)
                elif el == 'queue':
                    srcpad = self.tee.get_static_pad("src_%u" % stream_obj.ID)
                    self.tee.release_request_pad(srcpad)
                self.pipeline.remove(element)
                del element

sometimes I get this error (not everytime)

17-Nov-23 14:31:57 - ERROR - Error: gst-stream-error-quark: Failed to link decodebin to pipeline (1):: gstdsnvurisrcbin.cpp(1166): populate_uri_bin_audio (): /GstPipeline:pipeline0/GstBin:source-bin-45/GstDsNvUriSrcBin:uri-decode-bin
 
17-Nov-23 14:31:57 - ERROR - Error: gst-stream-error-quark: Failed to populate and link video elements (1):: gstdsnvurisrcbin.cpp(872): cb_newpad (): /GstPipeline:pipeline0/GstBin:source-bin-45/GstDsNvUriSrcBin:uri-decode-bin
3

You need to make sure remove the appsink first, then remove the queue .

4

I changed the order of removing plugins to to_remove = ['source-bin', 'sink', 'queue']
still I get the error

(python3:343): GStreamer-CRITICAL **: 08:51:33.055: 
Trying to dispose element sink-1, but it is in READY instead of the NULL state.
You need to explicitly set elements to the NULL state before
dropping the final reference, to allow them to clean up.
This problem may also be caused by a refcounting bug in the
application or some element.

the process is as follows

  1. set appsink null

  2. remove appsink from pipeline

  3. set queue null

  4. release request pad from tee

srcpad = self.tee.get_static_pad("src_%u" % stream_obj.ID)
self.tee.release_request_pad(srcpad)
  1. remove queue from pipeline
5

Just from the log attached, you need to wait until the state transition is completed before deleting the plugin.Could you just set the state to NULL and check the state_return value.

state_return = element.set_state(Gst.State.NULL)
6

state return value is <enum GST_STATE_CHANGE_SUCCESS of type Gst.StateChangeReturn>

23-Nov-23 06:14:03 - INFO - sink-1 set to NULL, state return :<enum GST_STATE_CHANGE_SUCCESS of type Gst.StateChangeReturn>

I also checked the state of the plugin, and it is (<enum GST_STATE_CHANGE_ASYNC of type Gst.StateChangeReturn>, state=<enum GST_STATE_READY of type Gst.State>, pending=<enum GST_STATE_PLAYING of type Gst.State>)

7

I added a wait time and tried

to_remove = ['source-bin', 'sink', 'queue']

        for el in to_remove:
            element_name = f'{el}-{stream_obj.ID}'
            try:
                logger.info(f"removing element {element_name}")
                element = self.pipeline.get_by_name(element_name)
                state_return = element.set_state(Gst.State.PAUSED)
                time.sleep(1)
                state_return = element.set_state(Gst.State.READY)
                time.sleep(1)
                state_return = element.set_state(Gst.State.NULL)
                st = time.time()
                while element.get_state(0.1)[1] != Gst.State.NULL:
                    time.sleep(0.5)
                    print("looping \n")
                    if  time.time()-st >120:
                        print("timeout \n")
                        break

                logger.info(f'{element_name}    {element.get_state(0)}')
                logger.info(f"{element_name} set to NULL, state return :{state_return}")

the output is always

sink-1    (<enum GST_STATE_CHANGE_SUCCESS of type Gst.StateChangeReturn>, state=<enum GST_STATE_NULL of type Gst.State>, pending=<enum GST_STATE_VOID_PENDING of type Gst.State>)
23-Nov-23 13:13:21 - INFO - sink-1 set to NULL, state return :<enum GST_STATE_CHANGE_SUCCESS of type Gst.StateChangeReturn>
8

Could you attach the whole code to us? Also, since your pipeline only has one sink, why do you want to delete that?

9

no, source can be added in the runtime and all the added sources will have respective appsink

I am attaching a test code here with
dummy_appsink_remove.txt (15.9 KB) can’t upload a .py file. please change the extension

10

Just from your code, you can only unlink the queue plugin, like the deepstream_rt_src_add_del.py.

1. state_return = g_source_bin_list[source_id].set_state(Gst.State.NULL)
2. #do the process based on the return value
        pad_name = "sink_%u" % source_id
        print(pad_name)
        #Retrieve sink pad to be released
        sinkpad = streammux.get_static_pad(pad_name)
        #Send flush stop event to the sink pad, then release from the streammux
        sinkpad.send_event(Gst.Event.new_flush_stop(False))
        streammux.release_request_pad(sinkpad)
        print("STATE CHANGE SUCCESS\n")
        #Remove the source bin from the pipeline
        pipeline.remove(g_source_bin_list[source_id])
11

so I should remove all the plugins in the same way?

12

Yes. But as you want to delete the plugins connected after tee, there may be some buffer processing issues. Because the various branches after tee actually use the same gstbuffer.