docs/pwg/advanced-scheduling.xml

   1 <chapter id="chapter-loopbased-sched">
   2   <title>How scheduling works</title>
   3   <para>
   4     Scheduling is, in short, a method for making sure that every element gets
   5     called once in a while to process data and prepare data for the next
   6     element. Likewise, a kernel has a scheduler to for processes, and your
   7     brain is a very complex scheduler too in a way.
   8     Randomly calling elements' chain functions won't bring us far, however, so
   9     you'll understand that the schedulers in &GStreamer; are a bit more complex
  10     than this. However, as a start, it's a nice picture.
  11     &GStreamer; currently provides two schedulers: a <emphasis>basic</emphasis>
  12     scheduler and an <emphasis>optimal</emphasis> scheduler. As the name says,
  13     the basic scheduler (<quote>basic</quote>) is an unoptimized, but very
  14     complete and simple scheduler. The optimal scheduler (<quote>opt</quote>),
  15     on the other hand, is optimized for media processing, but therefore also
  16     more complex.
  17   </para>
  18   <para>
  19     Note that schedulers only operate on one thread. If your pipeline contains
  20     multiple threads, each thread will run with a separate scheduler. That is
  21     the reason why two elements running in different threads need a queue-like
  22     element (a <classname>DECOUPLED</classname> element) in between them.
  23   </para>
  24
  25   <sect1 id="section-sched-basic" xreflabel="The Basic Scheduler">
  26   <title>The Basic Scheduler</title>
  27   <para>
  28     The <emphasis>basic</emphasis> scheduler assumes that each element is its
  29     own process. We don't use UNIX processes or POSIX threads for this,
  30     however; instead, we use so-called <emphasis>co-threads</emphasis>.
  31     Co-threads are threads that run besides each other, but only one is active
  32     at a time. The advantage of co-threads over normal threads is that they're
  33     lightweight. The disadvantage is that UNIX or POSIX do not provide such a
  34     thing, so we need to include our own co-threads stack for this to run.
  35   </para>
  36   <para>
  37     The task of the scheduler here is to control which co-thread runs at what
  38     time. A well-written scheduler based on co-threads will let an element run
  39     until it outputs one piece of data. Upon pushing one piece of data to the
  40     next element, it will let the next element run, and so on. Whenever a
  41     running element requires data from the previous element, the scheduler will
  42     switch to that previous element and run that element until it has provided
  43     data for use in the next element.
  44   </para>
  45   <para>
  46     This method of running elements as needed has the disadvantage that a lot
  47     of data will often be queued in between two elements, as the one element
  48     has provided data but the other element hasn't actually used it yet. These
  49     storages of in-between-data are called <emphasis>bufpens</emphasis>, and
  50     they can be visualized as a light <quote>queue</quote>.
  51   </para>
  52   <para>
  53     Note that since every element runs in its own (co-)thread, this scheduler
  54     is rather heavy on your system for larger pipelines.
  55   </para>
  56   </sect1>
  57
  58   <sect1 id="section-sched-opt" xreflabel="The Optimal Scheduler">
  59   <title>The Optimal Scheduler</title>
  60   <para>
  61     The <emphasis>optimal</emphasis> scheduler takes advantage of the fact that
  62     several elements can be linked together in one thread, with one element
  63     controlling the other. This works as follows: in a series of chain-based
  64     elements, each element has a function that accepts one piece of data, and
  65     it calls a function that provides one piece of data to the next element.
  66     The optimal scheduler will make sure that the <function>gst_pad_push ()</function>
  67     function of the first element <emphasis>directly</emphasis> calls the
  68     chain-function of the second element. This significantly decreases the
  69     latency in a pipeline. It takes similar advantage of other possibilities
  70     of short-cutting the data path from one element to the next.
  71   </para>
  72   <para>
  73     The disadvantage of the optimal scheduler is that it is not fully
  74     implemented. Also it is badly documented; for most developers, the opt
  75     scheduler is one big black box. Features that are not implemented
  76     include pad-unlinking within a group while running, pad-selecting
  77     (i.e. waiting for data to arrive on a list of pads), and it can't really
  78     cope with multi-input/-output elements (with the elements linked to each
  79     of these in-/outputs running in the same thread) right now.
  80   </para>
  81   <para>
  82     Some of our developers are intending to write a new scheduler, similar to
  83     the optimal scheduler (but better documented and more completely
  84     implemented).
  85   </para>
  86   </sect1>
  87 </chapter>
  88
  89 <chapter id="chapter-loopbased-loopfn">
  90   <title>How a loopfunc works</title>
  91   <para>
  92     A <function>_loop ()</function> function is a function that is called by
  93     the scheduler, but without providing data to the element. Instead, the
  94     element will become responsible for acquiring its own data, and it will
  95     still be responsible of sending data over to its source pads. This method
  96     noticeably complicates scheduling; you should only write loop-based
  97     elements when you need to. Normally, chain-based elements are preferred.
  98     Examples of elements that <emphasis>have</emphasis> to be loop-based are
  99     elements with multiple sink pads. Since the scheduler will push data into
 100     the pads as it comes (and this might not be synchronous), you will easily
 101     get ascynronous data on both pads, which means that the data that arrives
 102     on the first pad has a different display timestamp then the data arriving
 103     on the second pad at the same time. To get over these issues, you should
 104     write such elements in a loop-based form. Other elements that are
 105     <emphasis>easier</emphasis> to write in a loop-based form than in a
 106     chain-based form are demuxers and parsers. It is not required to write such
 107     elements in a loop-based form, though.
 108   </para>
 109   <para>
 110     Below is an example of the easiest loop-function that one can write:
 111   </para>
 112   <programlisting>
 113 static void     gst_my_filter_loopfunc  (GstElement *element);
 114
 115 static void
 116 gst_my_filter_init (GstMyFilter *filter)
 117 {
 118 [..]
 119   gst_element_set_loopfunc (GST_ELEMENT (filter), gst_my_filter_loopfunc);
 120 [..]
 121 }
 122
 123 static void
 124 gst_my_filter_loopfunc (GstElement *element)
 125 {
 126   GstMyFilter *filter = GST_MY_FILTER (element);
 127   GstData *data;
 128
 129   /* acquire data */
 130   data = gst_pad_pull (filter->sinkpad);
 131
 132   /* send data */
 133   gst_pad_push (filter->srcpad, data);
 134 }
 135   </programlisting>
 136   <para>
 137     Obviously, this specific example has no single advantage over a chain-based
 138     element, so you should never write such elements. However, it's a good
 139     introduction to the concept.
 140   </para>
 141
 142   <sect1 id="section-loopfn-multiinput" xreflabel="Multi-Input Elements">
 143     <title>Multi-Input Elements</title>
 144     <para>
 145       Elements with multiple sink pads need to take manual control over their
 146       input to assure that the input is synchronized. The following example
 147       code could (should) be used in an aggregator, i.e. an element that takes
 148       input from multiple streams and sends it out intermangled. Not really
 149       useful in practice, but a good example, again.
 150     </para>
 151     <programlisting>
 152 typedef struct _GstMyFilterInputContext {
 153   gboolean   eos;
 154   GstBuffer *lastbuf;
 155 } GstMyFilterInputContext;
 156
 157 [..]
 158
 159 static void
 160 gst_my_filter_init (GstMyFilter *filter)
 161 {
 162   GstElementClass *klass = GST_ELEMENT_GET_CLASS (filter);
 163   GstMyFilterInputContext *context;
 164
 165   filter->sinkpad1 = gst_pad_new_from_template (
 166         gst_element_class_get_pad_template (klass, "sink"), "sink_1");
 167   context = g_new0 (GstMyFilterInputContext, 1);
 168   gst_pad_set_private_data (filter->sinkpad1, context);
 169 [..]
 170   filter->sinkpad2 = gst_pad_new_from_template (
 171         gst_element_class_get_pad_template (klass, "sink"), "sink_2");
 172   context = g_new0 (GstMyFilterInputContext, 1);
 173   gst_pad_set_private_data (filter->sinkpad2, context);
 174 [..]
 175   gst_element_set_loopfunc (GST_ELEMENT (filter),
 176                             gst_my_filter_loopfunc);
 177 }
 178
 179 [..]
 180
 181 static void
 182 gst_my_filter_loopfunc (GstElement *element)
 183 {
 184   GstMyFilter *filter = GST_MY_FILTER (element);
 185   GList *padlist;
 186   GstMyFilterInputContext *first_context = NULL;
 187
 188   /* Go over each sink pad, update the cache if needed, handle EOS
 189    * or non-responding streams and see which data we should handle
 190    * next. */
 191   for (padlist = gst_element_get_padlist (element);
 192        padlist != NULL; padlist = g_list_next (padlist)) {
 193     GstPad *pad = GST_PAD (padlist->data);
 194     GstMyFilterInputContext *context = gst_pad_get_private_data (pad);
 195
 196     if (GST_PAD_IS_SRC (pad))
 197       continue;
 198
 199     while (GST_PAD_IS_USABLE (pad) &amp;&amp;
 200            !context->eos &amp;&amp; !context->lastbuf) {
 201       GstData *data = gst_pad_pull (pad);
 202
 203       if (GST_IS_EVENT (data)) {
 204         /* We handle events immediately */
 205         GstEvent *event = GST_EVENT (data);
 206
 207         switch (GST_EVENT_TYPE (event)) {
 208           case GST_EVENT_EOS:
 209             context->eos = TRUE;
 210             gst_event_unref (event);
 211             break;
 212           case GST_EVENT_DISCONTINUOUS:
 213             g_warning ("HELP! How do I handle this?");
 214             /* fall-through */
 215           default:
 216             gst_pad_event_default (pad, event);
 217             break;
 218         }
 219       } else {
 220         /* We store the buffer to handle synchronization below */
 221         context->lastbuf = GST_BUFFER (data);
 222       }
 223     }
 224
 225     /* synchronize streams by always using the earliest buffer */
 226     if (context->lastbuf) {
 227       if (!first_context) {
 228         first_context = context;
 229       } else {
 230         if (GST_BUFFER_TIMESTAMP (context->lastbuf) <
 231                 GST_BUFFER_TIMESTAMP (first_context->lastbuf))
 232           first_context = context;
 233       }
 234     }
 235   }
 236
 237   /* If we handle no data at all, we're at the end-of-stream, so
 238    * we should signal EOS. */
 239   if (!first_context) {
 240     gst_pad_push (filter->srcpad, GST_DATA (gst_event_new (GST_EVENT_EOS)));
 241     gst_element_set_eos (element);
 242     return;
 243   }
 244
 245   /* So we do have data! Let's forward that to our source pad. */
 246   gst_pad_push (filter->srcpad, GST_DATA (first_context->lastbuf));
 247   first_context->lastbuf = NULL;
 248 }
 249     </programlisting>
 250     <para>
 251       Note that a loop-function is allowed to return. Better yet, a loop
 252       function <emphasis>has to</emphasis> return so the scheduler can
 253       let other elements run (this is particularly true for the optimal
 254       scheduler). Whenever the scheduler feels right, it will call the
 255       loop-function of the element again.
 256     </para>
 257   </sect1>
 258
 259   <sect1 id="section-loopfn-bytestream" xreflabel="The Bytestream Object">
 260     <title>The Bytestream Object</title>
 261     <para>
 262       A second type of elements that wants to be loop-based, are the so-called
 263       bytestream-elements. Until now, we've only dealt with elements that
 264       receive of pull full buffers of a random size from other elements. Often,
 265       however, it is wanted to have control over the stream at a byte-level,
 266       such as in stream parsers or demuxers. It is possible to manually pull
 267       buffers and merge them until a certain size; it is easier, however, to
 268       use bytestream, which wraps this behaviour.
 269     </para>
 270     <para>
 271       Bytestream-using elements are ususally stream parsers or demuxers. For
 272       now, we will take a parser as an example. Demuxers require some more
 273       magic that will be dealt with later in this guide:
 274       <xref linkend="chapter-advanced-request"/>. The goal of this parser will be
 275       to parse a text-file and to push each line of text as a separate buffer
 276       over its source pad.
 277     </para>
 278     <programlisting>
 279 static void
 280 gst_my_filter_loopfunc (GstElement *element)
 281 {
 282   GstMyFilter *filter = GST_MY_FILTER (element);
 283   gint n, num;
 284   guint8 *data;
 285
 286   for (n = 0; ; n++) {
 287     num = gst_bytestream_peek_bytes (filter->bs, &amp;data, n + 1);
 288     if (num != n + 1) {
 289       GstEvent *event = NULL;
 290       guint remaining;
 291
 292       gst_bytestream_get_status (filter->bs, &amp;remaining, &amp;event);
 293       if (event) {
 294         if (GST_EVENT_TYPE (event) == GST_EVENT_EOS)) {
 295           /* end-of-file */
 296           gst_pad_push (filter->srcpad, GST_DATA (event));
 297           gst_element_set_eos (element);
 298
 299           return;
 300         }
 301         gst_event_unref (event);
 302       }
 303
 304       /* failed to read - throw error and bail out */
 305       gst_element_error (element, STREAM, READ, (NULL), (NULL));
 306
 307       return;
 308     }
 309
 310     /* check if the last character is a newline */
 311     if (data[n] == '\n') {
 312       GstBuffer *buf = gst_buffer_new_and_alloc (n + 1);
 313
 314       /* read the line of text without newline - then flush the newline */
 315       gst_bytestream_peek_data (filter->bs, &amp;data, n);
 316       memcpy (GST_BUFFER_DATA (buf), data, n);
 317       GST_BUFFER_DATA (buf)[n] = '\0';
 318       gst_bytestream_flush_fast (filter->bs, n + 1);
 319       g_print ("Pushing '%s'\n", GST_BUFFER_DATA (buf));
 320       gst_pad_push (filter->srcpad, GST_DATA (buf));
 321
 322       return;
 323     }
 324   }
 325 }
 326
 327 static void
 328 gst_my_filter_change_state (GstElement *element)
 329 {
 330   GstMyFilter *filter = GST_MY_FILTER (element);
 331
 332   switch (GST_STATE_TRANSITION (element)) {
 333     case GST_STATE_READY_TO_PAUSED:
 334       filter->bs = gst_bytestream_new (filter->sinkpad);
 335       break;
 336     case GST_STATE_PAUSED_TO_READY:
 337       gst_bytestream_destroy (filter->bs);
 338       break;
 339     default:
 340       break;
 341   }
 342
 343   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 344     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 345
 346   return GST_STATE_SUCCESS;
 347 }
 348     </programlisting>
 349     <para>
 350       In the above example, you'll notice how bytestream handles buffering of
 351       data for you. The result is that you can handle the same data multiple
 352       times. Event handling in bytestream is currently sort of
 353       <emphasis>wacky</emphasis>, but it works quite well. The one big
 354       disadvantage of bytestream is that it <emphasis>requires</emphasis>
 355       the element to be loop-based. Long-term, we hope to have a chain-based
 356       usable version of bytestream, too.
 357     </para>
 358   </sect1>
 359
 360   <sect1 id="section-loopbased-secnd">
 361     <title>Adding a second output</title>
 362     <para>
 363       Identity is now a tee
 364     </para>
 365   </sect1>
 366
 367   <sect1 id="section-loopbased-modappl">
 368     <title>Modifying the test application</title>
 369     <para>
 370       WRITEME
 371     </para>
 372   </sect1>
 373 </chapter>
 374