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 <![CDATA[
 153
 154 typedef struct _GstMyFilterInputContext {
 155   gboolean   eos;
 156   GstBuffer *lastbuf;
 157 } GstMyFilterInputContext;
 158
 159 [..]
 160
 161 static void
 162 gst_my_filter_init (GstMyFilter *filter)
 163 {
 164   GstElementClass *klass = GST_ELEMENT_GET_CLASS (filter);
 165   GstMyFilterInputContext *context;
 166
 167   filter->sinkpad1 = gst_pad_new_from_template (
 168         gst_element_class_get_pad_template (klass, "sink"), "sink_1");
 169   context = g_new0 (GstMyFilterInputContext, 1);
 170   gst_pad_set_private_data (filter->sinkpad1, context);
 171 [..]
 172   filter->sinkpad2 = gst_pad_new_from_template (
 173         gst_element_class_get_pad_template (klass, "sink"), "sink_2");
 174   context = g_new0 (GstMyFilterInputContext, 1);
 175   gst_pad_set_private_data (filter->sinkpad2, context);
 176 [..]
 177   gst_element_set_loopfunc (GST_ELEMENT (filter),
 178                             gst_my_filter_loopfunc);
 179 }
 180
 181 [..]
 182
 183 static void
 184 gst_my_filter_loopfunc (GstElement *element)
 185 {
 186   GstMyFilter *filter = GST_MY_FILTER (element);
 187   GList *padlist;
 188   GstMyFilterInputContext *first_context = NULL;
 189
 190   /* Go over each sink pad, update the cache if needed, handle EOS
 191    * or non-responding streams and see which data we should handle
 192    * next. */
 193   for (padlist = gst_element_get_padlist (element);
 194        padlist != NULL; padlist = g_list_next (padlist)) {
 195     GstPad *pad = GST_PAD (padlist->data);
 196     GstMyFilterInputContext *context = gst_pad_get_private_data (pad);
 197
 198     if (GST_PAD_IS_SRC (pad))
 199       continue;
 200
 201     while (GST_PAD_IS_USABLE (pad) &&
 202            !context->eos && !context->lastbuf) {
 203       GstData *data = gst_pad_pull (pad);
 204
 205       if (GST_IS_EVENT (data)) {
 206         /* We handle events immediately */
 207         GstEvent *event = GST_EVENT (data);
 208
 209         switch (GST_EVENT_TYPE (event)) {
 210           case GST_EVENT_EOS:
 211             context->eos = TRUE;
 212             gst_event_unref (event);
 213             break;
 214           case GST_EVENT_DISCONTINUOUS:
 215             g_warning ("HELP! How do I handle this?");
 216             /* fall-through */
 217           default:
 218             gst_pad_event_default (pad, event);
 219             break;
 220         }
 221       } else {
 222         /* We store the buffer to handle synchronization below */
 223         context->lastbuf = GST_BUFFER (data);
 224       }
 225     }
 226
 227     /* synchronize streams by always using the earliest buffer */
 228     if (context->lastbuf) {
 229       if (!first_context) {
 230         first_context = context;
 231       } else {
 232         if (GST_BUFFER_TIMESTAMP (context->lastbuf) <
 233                 GST_BUFFER_TIMESTAMP (first_context->lastbuf))
 234           first_context = context;
 235       }
 236     }
 237   }
 238
 239   /* If we handle no data at all, we're at the end-of-stream, so
 240    * we should signal EOS. */
 241   if (!first_context) {
 242     gst_pad_push (filter->srcpad, GST_DATA (gst_event_new (GST_EVENT_EOS)));
 243     gst_element_set_eos (element);
 244     return;
 245   }
 246
 247   /* So we do have data! Let's forward that to our source pad. */
 248   gst_pad_push (filter->srcpad, GST_DATA (first_context->lastbuf));
 249   first_context->lastbuf = NULL;
 250 }
 251 ]]>
 252     </programlisting>
 253     <para>
 254       Note that a loop-function is allowed to return. Better yet, a loop
 255       function <emphasis>has to</emphasis> return so the scheduler can
 256       let other elements run (this is particularly true for the optimal
 257       scheduler). Whenever the scheduler feels right, it will call the
 258       loop-function of the element again.
 259     </para>
 260   </sect1>
 261
 262   <sect1 id="section-loopfn-bytestream" xreflabel="The Bytestream Object">
 263     <title>The Bytestream Object</title>
 264     <para>
 265       A second type of elements that wants to be loop-based, are the so-called
 266       bytestream-elements. Until now, we've only dealt with elements that
 267       receive of pull full buffers of a random size from other elements. Often,
 268       however, it is wanted to have control over the stream at a byte-level,
 269       such as in stream parsers or demuxers. It is possible to manually pull
 270       buffers and merge them until a certain size; it is easier, however, to
 271       use bytestream, which wraps this behaviour.
 272     </para>
 273     <para>
 274       Bytestream-using elements are ususally stream parsers or demuxers. For
 275       now, we will take a parser as an example. Demuxers require some more
 276       magic that will be dealt with later in this guide:
 277       <xref linkend="chapter-advanced-request"/>. The goal of this parser will be
 278       to parse a text-file and to push each line of text as a separate buffer
 279       over its source pad.
 280     </para>
 281     <programlisting>
 282 <![CDATA[
 283 static void
 284 gst_my_filter_loopfunc (GstElement *element)
 285 {
 286   GstMyFilter *filter = GST_MY_FILTER (element);
 287   gint n, num;
 288   guint8 *data;
 289
 290   for (n = 0; ; n++) {
 291     num = gst_bytestream_peek_bytes (filter->bs, &data, n + 1);
 292     if (num != n + 1) {
 293       GstEvent *event = NULL;
 294       guint remaining;
 295
 296       gst_bytestream_get_status (filter->bs, &remaining, &event);
 297       if (event) {
 298         if (GST_EVENT_TYPE (event) == GST_EVENT_EOS)) {
 299           /* end-of-file */
 300           gst_pad_push (filter->srcpad, GST_DATA (event));
 301           gst_element_set_eos (element);
 302
 303           return;
 304         }
 305         gst_event_unref (event);
 306       }
 307
 308       /* failed to read - throw error and bail out */
 309       gst_element_error (element, STREAM, READ, (NULL), (NULL));
 310
 311       return;
 312     }
 313
 314     /* check if the last character is a newline */
 315     if (data[n] == '\n') {
 316       GstBuffer *buf = gst_buffer_new_and_alloc (n + 1);
 317
 318       /* read the line of text without newline - then flush the newline */
 319       gst_bytestream_peek_data (filter->bs, &data, n);
 320       memcpy (GST_BUFFER_DATA (buf), data, n);
 321       GST_BUFFER_DATA (buf)[n] = '\0';
 322       gst_bytestream_flush_fast (filter->bs, n + 1);
 323       g_print ("Pushing '%s'\n", GST_BUFFER_DATA (buf));
 324       gst_pad_push (filter->srcpad, GST_DATA (buf));
 325
 326       return;
 327     }
 328   }
 329 }
 330
 331 static void
 332 gst_my_filter_change_state (GstElement *element)
 333 {
 334   GstMyFilter *filter = GST_MY_FILTER (element);
 335
 336   switch (GST_STATE_TRANSITION (element)) {
 337     case GST_STATE_READY_TO_PAUSED:
 338       filter->bs = gst_bytestream_new (filter->sinkpad);
 339       break;
 340     case GST_STATE_PAUSED_TO_READY:
 341       gst_bytestream_destroy (filter->bs);
 342       break;
 343     default:
 344       break;
 345   }
 346
 347   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 348     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 349
 350   return GST_STATE_SUCCESS;
 351 }
 352 ]]>
 353     </programlisting>
 354     <para>
 355       In the above example, you'll notice how bytestream handles buffering of
 356       data for you. The result is that you can handle the same data multiple
 357       times. Event handling in bytestream is currently sort of
 358       <emphasis>wacky</emphasis>, but it works quite well. The one big
 359       disadvantage of bytestream is that it <emphasis>requires</emphasis>
 360       the element to be loop-based. Long-term, we hope to have a chain-based
 361       usable version of bytestream, too.
 362     </para>
 363   </sect1>
 364
 365   <sect1 id="section-loopbased-secnd">
 366     <title>Adding a second output</title>
 367     <para>
 368       Identity is now a tee
 369     </para>
 370   </sect1>
 371
 372   <sect1 id="section-loopbased-modappl">
 373     <title>Modifying the test application</title>
 374     <para>
 375       WRITEME
 376     </para>
 377   </sect1>
 378 </chapter>
 379