Are Bus Messages Queued?

[Nastooh Avessta]

Hi

Got an application that would take an action, upon receiving messages from multiple plugins , through a watch on the app bus: (Note that these messages are asynchronous and may arrive at the ~ same time.)

…

   loop = g_main_loop_new (NULL, FALSE);

   bus = gst_element_get_bus (pipeline);

   bus_watch_id = gst_bus_add_watch (bus, bus_call, (gpointer *) &data);

…

Where, bus_call:

static gboolean bus_call (GstBus     *bus, GstMessage *msg, gpointer    gdata){

…

   switch (GST_MESSAGE_TYPE (msg)) {

…

    case GST_MESSAGE_ELEMENT:

    g_print("Got message from %s\n",gst_object_get_name (GST_MESSAGE_SRC(msg)));

…

   break;

…

}

 

What I am seeing is that on different runs not all messages make it to    “case GST_MESSAGE_ELEMENT”, and am wondering if I need to queue messages myself, start a new processing thread per message, etc.?

Cheers,

_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

[Tim Müller]

On Thu, 2016-11-03 at 17:42 -0700, Nastooh Avessta wrote:

Hi Nastooh,

> Got an application that would take an action, upon receiving messages
> from multiple plugins , through a watch on the app bus: (Note that
> these messages are asynchronous and may arrive at the ~ same time.)
> …
>  
> What I am seeing is that on different runs not all messages make it
> to    “case GST_MESSAGE_ELEMENT”, and am wondering if I need to queue
> messages myself, start a new processing thread per message, etc.?

No, GstBus is a message queue itself. All messages should make it
through to your handler. No messages should be lost unless you pop them
off the bus yourself.

Two exceptions:

If you set a pipeline to NULL state the bus is 'flushed' in the NULL-
>READY state change transition. You can disable that behaviour though.

If you pop messages off the bus e.g. with gst_bus_*_pop_filtered() in
another thread whist at the same time having a bus watch set up that
might interact in weird ways.

Check the GST_DEBUG=GST_BUS:6 log to see whether those messages make it
to the bus in the first place.

Cheers
 -Tim
--
Tim Müller, Centricular Ltd - http://www.centricular.com
_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

[Nastooh Avessta]

Hi Tim
Thank you for your reply. Here is my use case:
1- Application solicits plugins for messages, in a serialized ordered manner
2- Plugins reply in an asynchronous manner
3- Upon receiving messages from all targeted plugins, an action is taken
Part 3, is coded inside the bus message handler:
…
case GST_MESSAGE_ELEMENT:
            g_print("Got message from %s\n",gst_object_get_name (GST_MESSAGE_SRC(msg)));
     int i;
                for(i=0;i<messageCnt;i++){
                    if(!strcmp(messagePlug[i],gst_object_get_name (GST_MESSAGE_SRC(msg)))){
                        noMessage++;
                        if(noMessages==messageCnt){
                            g_print("Life is good\n");
                           
                        }
                    }
                }
            break;
…
(Here, messageCnt is the expected number of messages to be received, and messagePlug is a pointer array containing the names of the plugins.)
What I was wondering is that while processing the current message, i.e., somewhere in that for loop, could I miss an incoming message? And if I understand you correctly, that should not be the case, as I am not popping any messages or going to NULL state. Am I correct in my understanding?
I should mention that this gstreamer 1.4.5.
Cheers,

-----Original Message-----
From: gstreamer-devel [mailto:[hidden email]] On Behalf Of Tim Müller
Sent: Friday, November 04, 2016 1:47 AM
To: [hidden email]
Subject: Re: Are Bus Messages Queued?

On Thu, 2016-11-03 at 17:42 -0700, Nastooh Avessta wrote:

Hi Nastooh,

> Got an application that would take an action, upon receiving messages
> from multiple plugins , through a watch on the app bus: (Note that
> these messages are asynchronous and may arrive at the ~ same time.) …
>  
> What I am seeing is that on different runs not all messages make it to    
> “case GST_MESSAGE_ELEMENT”, and am wondering if I need to queue
> messages myself, start a new processing thread per message, etc.?

No, GstBus is a message queue itself. All messages should make it through to your handler. No messages should be lost unless you pop them off the bus yourself.

Two exceptions:

If you set a pipeline to NULL state the bus is 'flushed' in the NULL-
>READY state change transition. You can disable that behaviour though.

If you pop messages off the bus e.g. with gst_bus_*_pop_filtered() in another thread whist at the same time having a bus watch set up that might interact in weird ways.

Check the GST_DEBUG=GST_BUS:6 log to see whether those messages make it to the bus in the first place.

Cheers
 -Tim
--
Tim Müller, Centricular Ltd - http://www.centricular.com _______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

[Ian Davidson]

In the code that you show, I do not see you initialising 'noMessage'.  I
assume that you DO initialise it and just have not shown the code.

Ian


On 04/11/2016 17:07, Nastooh Avessta wrote:

> Hi Tim
> Thank you for your reply. Here is my use case:
> 1- Application solicits plugins for messages, in a serialized ordered manner
> 2- Plugins reply in an asynchronous manner
> 3- Upon receiving messages from all targeted plugins, an action is taken
> Part 3, is coded inside the bus message handler:
> …
> case GST_MESSAGE_ELEMENT:
>    g_print("Got message from %s\n",gst_object_get_name (GST_MESSAGE_SRC(msg)));
>       int i;
>        for(i=0;i<messageCnt;i++){
>            if(!strcmp(messagePlug[i],gst_object_get_name (GST_MESSAGE_SRC(msg)))){
>                noMessage++;
>                if(noMessages==messageCnt){
>                    g_print("Life is good\n");
>
>                }
>            }
>        }
>    break;

_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

[Nastooh Avessta]

Indeed, I do. (The same goes for array pointer messagePlug and messageCnt.)
Right now, the only way that I get all  of the messages, is to put artificial delay between consecutive calls, in Step 1, below. This; however, causes other problems.
Cheers,

-----Original Message-----
From: gstreamer-devel [mailto:[hidden email]] On Behalf Of Ian Davidson
Sent: Saturday, November 05, 2016 12:26 PM
To: Discussion of the development of and with GStreamer
Subject: Re: Are Bus Messages Queued?

In the code that you show, I do not see you initialising 'noMessage'.  I assume that you DO initialise it and just have not shown the code.

Ian


On 04/11/2016 17:07, Nastooh Avessta wrote:

> Hi Tim
> Thank you for your reply. Here is my use case:
> 1- Application solicits plugins for messages, in a serialized ordered manner
> 2- Plugins reply in an asynchronous manner
> 3- Upon receiving messages from all targeted plugins, an action is taken
> Part 3, is coded inside the bus message handler:
> …
> case GST_MESSAGE_ELEMENT:
>    g_print("Got message from %s\n",gst_object_get_name (GST_MESSAGE_SRC(msg)));
>       int i;
>        for(i=0;i<messageCnt;i++){
>            if(!strcmp(messagePlug[i],gst_object_get_name (GST_MESSAGE_SRC(msg)))){
>                noMessage++;
>                if(noMessages==messageCnt){
>                    g_print("Life is good\n");
>
>                }
>            }
>        }
>    break;

_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

[Ian Davidson]

Hi Nastooh,

I think I would work on the assumption that all the messages DO get
created and passed through one at a time.  I would suspect that there is
still an error in your code.  What I would do first would be to have a
g_print to report the 'failures' in your checking - for example, a 'Life
is Bad' if the counts do not match (and print out the actual and
expected counts), and some other message if/when the message source is
not in your list - it may be that you are ignoring a valid message.

Ian


On 05/11/2016 21:07, Nastooh Avessta wrote:

> Indeed, I do. (The same goes for array pointer messagePlug and messageCnt.)
> Right now, the only way that I get all  of the messages, is to put artificial delay between consecutive calls, in Step 1, below. This; however, causes other problems.
> Cheers,
>
> -----Original Message-----
> From: gstreamer-devel [mailto:[hidden email]] On Behalf Of Ian Davidson
> Sent: Saturday, November 05, 2016 12:26 PM
> To: Discussion of the development of and with GStreamer
> Subject: Re: Are Bus Messages Queued?
>
> In the code that you show, I do not see you initialising 'noMessage'.  I assume that you DO initialise it and just have not shown the code.
>
> Ian
>
>
> On 04/11/2016 17:07, Nastooh Avessta wrote:
>> Hi Tim
>> Thank you for your reply. Here is my use case:
>> 1- Application solicits plugins for messages, in a serialized ordered manner
>> 2- Plugins reply in an asynchronous manner
>> 3- Upon receiving messages from all targeted plugins, an action is taken
>> Part 3, is coded inside the bus message handler:
>> …
>> case GST_MESSAGE_ELEMENT:
>>    g_print("Got message from %s\n",gst_object_get_name (GST_MESSAGE_SRC(msg)));
>>        int i;
>>        for(i=0;i<messageCnt;i++){
>>            if(!strcmp(messagePlug[i],gst_object_get_name (GST_MESSAGE_SRC(msg)))){
>>                noMessage++;
>>                if(noMessages==messageCnt){
>>                    g_print("Life is good\n");
>>
>>                }
>>            }
>>        }
>>    break;
> _______________________________________________
> gstreamer-devel mailing list
> [hidden email]
> https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel
>
> _______________________________________________
> gstreamer-devel mailing list
> [hidden email]

_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel

[Sebastian Dröge-3]

On Sun, 2016-11-06 at 15:39 +0000, Ian Davidson wrote:
> Hi Nastooh,
>
> I think I would work on the assumption that all the messages DO get 
> created and passed through one at a time.  I would suspect that there is 
> still an error in your code.  What I would do first would be to have a 
> g_print to report the 'failures' in your checking - for example, a 'Life 
> is Bad' if the counts do not match (and print out the actual and 
> expected counts), and some other message if/when the message source is 
> not in your list - it may be that you are ignoring a valid message.

Maybe try providing a standalone testcase that reproduces the problem
so someone can take a closer look without having to guess what mistakes
you make or don't make in the parts of the code that you didn't show :)

--
Sebastian Dröge, Centricular Ltd · http://www.centricular.com
_______________________________________________
gstreamer-devel mailing list
[hidden email]
https://lists.freedesktop.org/mailman/listinfo/gstreamer-devel