gst/gstscheduler.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *
   5  * gstscheduler.c: Default scheduling code for most cases
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 #define GST_DEBUG_ENABLED
  24 #include "gst_private.h"
  25
  26 #include "gstscheduler.h"
  27
  28
  29 static int
  30 gst_bin_loopfunc_wrapper (int argc,char *argv[])
  31 {
  32   GstElement *element = GST_ELEMENT (argv);
  33   G_GNUC_UNUSED const gchar *name = gst_element_get_name (element);
  34
  35   DEBUG_ENTER("(%d,'%s')",argc,name);
  36
  37   do {
  38     DEBUG("calling loopfunc %s for element %s\n",
  39           GST_DEBUG_FUNCPTR_NAME (element->loopfunc),name);
  40     (element->loopfunc) (element);
  41     DEBUG("element %s ended loop function\n", name);
  42   } while (!GST_ELEMENT_IS_COTHREAD_STOPPING(element));
  43   GST_FLAG_UNSET(element,GST_ELEMENT_COTHREAD_STOPPING);
  44
  45   DEBUG_LEAVE("(%d,'%s')",argc,name);
  46   return 0;
  47 }
  48
  49 static int
  50 gst_bin_chain_wrapper (int argc,char *argv[])
  51 {
  52   GstElement *element = GST_ELEMENT (argv);
  53   G_GNUC_UNUSED const gchar *name = gst_element_get_name (element);
  54   GList *pads;
  55   GstPad *pad;
  56   GstBuffer *buf;
  57
  58   DEBUG_ENTER("(\"%s\")",name);
  59   DEBUG("stepping through pads\n");
  60   do {
  61     pads = element->pads;
  62     while (pads) {
  63       pad = GST_PAD (pads->data);
  64       pads = g_list_next (pads);
  65       if (pad->direction == GST_PAD_SINK) {
  66         DEBUG("pulling a buffer from %s:%s\n", name, gst_pad_get_name (pad));
  67         buf = gst_pad_pull (pad);
  68         DEBUG("calling chain function of %s:%s\n", name, gst_pad_get_name (pad));
  69         (pad->chainfunc) (pad,buf);
  70         DEBUG("calling chain function of %s:%s done\n", name, gst_pad_get_name (pad));
  71       }
  72     }
  73   } while (!GST_ELEMENT_IS_COTHREAD_STOPPING(element));
  74   GST_FLAG_UNSET(element,GST_ELEMENT_COTHREAD_STOPPING);
  75
  76   DEBUG_LEAVE("(%d,'%s')",argc,name);
  77   return 0;
  78 }
  79
  80 static int
  81 gst_bin_src_wrapper (int argc,char *argv[])
  82 {
  83   GstElement *element = GST_ELEMENT (argv);
  84   GList *pads;
  85   GstPad *pad;
  86   GstBuffer *buf;
  87   G_GNUC_UNUSED const gchar *name = gst_element_get_name (element);
  88
  89   DEBUG_ENTER("(%d,\"%s\")",argc,name);
  90
  91   do {
  92     pads = element->pads;
  93     while (pads) {
  94       pad = GST_PAD (pads->data);
  95       if (pad->direction == GST_PAD_SRC) {
  96 //        region_struct *region = cothread_get_data (element->threadstate, "region");
  97         DEBUG("calling _getfunc for %s:%s\n",GST_DEBUG_PAD_NAME(pad));
  98 //        if (region) {
  99           //gst_src_push_region (GST_SRC (element), region->offset, region->size);
 100 //          if (pad->getregionfunc == NULL)
 101 //            fprintf(stderr,"error, no getregionfunc in \"%s\"\n", name);
 102 //          buf = (pad->getregionfunc)(pad, region->offset, region->size);
 103 //        } else {
 104           if (pad->getfunc == NULL)
 105             fprintf(stderr,"error, no getfunc in \"%s\"\n", name);
 106           buf = (pad->getfunc)(pad);
 107 //        }
 108
 109         DEBUG("calling gst_pad_push on pad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 110         gst_pad_push (pad, buf);
 111       }
 112       pads = g_list_next(pads);
 113     }
 114   } while (!GST_ELEMENT_IS_COTHREAD_STOPPING(element));
 115   GST_FLAG_UNSET(element,GST_ELEMENT_COTHREAD_STOPPING);
 116
 117   DEBUG_LEAVE("");
 118   return 0;
 119 }
 120
 121 static void
 122 gst_bin_pushfunc_proxy (GstPad *pad, GstBuffer *buf)
 123 {
 124   cothread_state *threadstate = GST_ELEMENT(pad->parent)->threadstate;
 125   DEBUG_ENTER("(%s:%s)",GST_DEBUG_PAD_NAME(pad));
 126   DEBUG("putting buffer %p in peer's pen\n",buf);
 127   pad->peer->bufpen = buf;
 128   DEBUG("switching to %p (@%p)\n",threadstate,&(GST_ELEMENT(pad->parent)->threadstate));
 129   cothread_switch (threadstate);
 130   DEBUG("done switching\n");
 131 }
 132
 133 static GstBuffer*
 134 gst_bin_pullfunc_proxy (GstPad *pad)
 135 {
 136   GstBuffer *buf;
 137
 138   cothread_state *threadstate = GST_ELEMENT(pad->parent)->threadstate;
 139   DEBUG_ENTER("(%s:%s)",GST_DEBUG_PAD_NAME(pad));
 140   if (pad->bufpen == NULL) {
 141     DEBUG("switching to %p (@%p)\n",threadstate,&(GST_ELEMENT(pad->parent)->threadstate));
 142     cothread_switch (threadstate);
 143   }
 144   DEBUG("done switching\n");
 145   buf = pad->bufpen;
 146   pad->bufpen = NULL;
 147   return buf;
 148 }
 149
 150 static GstBuffer *
 151 gst_bin_chainfunc_proxy (GstPad *pad)
 152 {
 153 // FIXME!!
 154 //  GstBuffer *buf;
 155   return NULL;
 156 }
 157
 158 // FIXME!!!
 159 static void
 160 gst_bin_pullregionfunc_proxy (GstPad *pad,
 161                                 gulong offset,
 162                                 gulong size)
 163 {
 164 //  region_struct region;
 165   cothread_state *threadstate;
 166
 167   DEBUG_ENTER("%s:%s,%ld,%ld",GST_DEBUG_PAD_NAME(pad),offset,size);
 168
 169 //  region.offset = offset;
 170 //  region.size = size;
 171
 172 //  threadstate = GST_ELEMENT(pad->parent)->threadstate;
 173 //  cothread_set_data (threadstate, "region", &region);
 174   cothread_switch (threadstate);
 175 //  cothread_set_data (threadstate, "region", NULL);
 176 }
 177
 178
 179 static void
 180 gst_schedule_cothreaded_chain (GstBin *bin, _GstBinChain *chain) {
 181   GList *elements;
 182   GstElement *element;
 183   cothread_func wrapper_function;
 184   GList *pads;
 185   GstPad *pad;
 186
 187   DEBUG("chain is using cothreads\n");
 188
 189   // first create thread context
 190   if (bin->threadcontext == NULL) {
 191     DEBUG("initializing cothread context\n");
 192     bin->threadcontext = cothread_init ();
 193   }
 194
 195   // walk through all the chain's elements
 196   elements = chain->elements;
 197   while (elements) {
 198     element = GST_ELEMENT (elements->data);
 199     elements = g_list_next (elements);
 200
 201     // start out without a wrapper function, we select it later
 202     wrapper_function = NULL;
 203
 204     // if the element has a loopfunc...
 205     if (element->loopfunc != NULL) {
 206       wrapper_function = GST_DEBUG_FUNCPTR(gst_bin_loopfunc_wrapper);
 207       DEBUG("\nelement '%s' is a loop-based\n",gst_element_get_name(element));
 208     } else {
 209       // otherwise we need to decide what kind of cothread
 210       // if it's not DECOUPLED, we decide based on whether it's a source or not
 211       if (!GST_FLAG_IS_SET (element, GST_ELEMENT_DECOUPLED)) {
 212         // if it doesn't have any sinks, it must be a source (duh)
 213         if (element->numsinkpads == 0) {
 214           wrapper_function = GST_DEBUG_FUNCPTR(gst_bin_src_wrapper);
 215           DEBUG("\nelement '%s' is a source, using _src_wrapper\n",gst_element_get_name(element));
 216         } else {
 217           wrapper_function = GST_DEBUG_FUNCPTR(gst_bin_chain_wrapper);
 218           DEBUG("\nelement '%s' is a filter, using _chain_wrapper\n",gst_element_get_name(element));
 219         }
 220       }
 221     }
 222
 223     // now we have to walk through the pads to set up their state
 224     pads = gst_element_get_pad_list (element);
 225     while (pads) {
 226       pad = GST_PAD (pads->data);
 227       pads = g_list_next (pads);
 228
 229       // if the element is DECOUPLED or outside the manager, we have to chain
 230       if ((wrapper_function == NULL) ||
 231           (GST_ELEMENT(pad->peer->parent)->manager != GST_ELEMENT(bin))) {
 232         // set the chain proxies
 233         if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 234           DEBUG("copying chain function into push proxy for %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 235           pad->pushfunc = pad->chainfunc;
 236         } else {
 237           DEBUG("copying get function into pull proxy for %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 238           pad->pullfunc = pad->getfunc;
 239           pad->pullregionfunc = pad->getregionfunc;
 240         }
 241
 242       // otherwise we really are a cothread
 243       } else {
 244         if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 245           DEBUG("setting cothreaded push proxy for sinkpad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 246           pad->pushfunc = GST_DEBUG_FUNCPTR(gst_bin_pushfunc_proxy);
 247         } else {
 248           DEBUG("setting cothreaded pull proxy for srcpad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 249           pad->pullfunc = GST_DEBUG_FUNCPTR(gst_bin_pullfunc_proxy);
 250         }
 251       }
 252     }
 253
 254     // need to set up the cothread now
 255     if (wrapper_function != NULL) {
 256       if (element->threadstate == NULL) {
 257         element->threadstate = cothread_create (bin->threadcontext);
 258         DEBUG("created cothread %p for '%s'\n",element->threadstate,gst_element_get_name(element));
 259       }
 260       cothread_setfunc (element->threadstate, wrapper_function, 0, (char **)element);
 261       DEBUG("set wrapper function for '%s' to &%s\n",gst_element_get_name(element),
 262             GST_DEBUG_FUNCPTR_NAME(wrapper_function));
 263     }
 264   }
 265 }
 266
 267 static void
 268 gst_schedule_chained_chain (GstBin *bin, _GstBinChain *chain) {
 269   GList *elements;
 270   GstElement *element;
 271   GList *pads;
 272   GstPad *pad;
 273
 274   DEBUG("chain entered\n");
 275   // walk through all the elements
 276   elements = chain->elements;
 277   while (elements) {
 278     element = GST_ELEMENT (elements->data);
 279     elements = g_list_next (elements);
 280
 281     // walk through all the pads
 282     pads = gst_element_get_pad_list (element);
 283     while (pads) {
 284       pad = GST_PAD (pads->data);
 285       pads = g_list_next (pads);
 286
 287       if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 288         DEBUG("copying chain function into push proxy for %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 289         pad->pushfunc = pad->chainfunc;
 290       } else {
 291         DEBUG("copying get function into pull proxy for %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 292         pad->pullfunc = pad->getfunc;
 293         pad->pullregionfunc = pad->getregionfunc;
 294       }
 295     }
 296   }
 297 }
 298
 299 static void gst_bin_schedule_cleanup(GstBin *bin) {
 300   GList *chains;
 301   _GstBinChain *chain;
 302
 303   chains = bin->chains;
 304   while (chains) {
 305     chain = (_GstBinChain *)(chains->data);
 306     chains = g_list_next(chains);
 307
 308     g_list_free(chain->elements);
 309     g_list_free(chain->entries);
 310
 311     g_free(chain);
 312   }
 313   g_list_free(bin->chains);
 314
 315   bin->chains = NULL;
 316 }
 317
 318 void gst_bin_schedule_func(GstBin *bin) {
 319 //  GstElement *manager;
 320   GList *elements;
 321   GstElement *element;
 322 //  const gchar *elementname;
 323   GSList *pending = NULL;
 324 //  GstBin *pending_bin;
 325   GList *pads;
 326   GstPad *pad;
 327 //  GstElement *peer_manager;
 328   GList *chains;
 329   _GstBinChain *chain;
 330
 331   DEBUG_SET_STRING("(\"%s\")",gst_element_get_name (GST_ELEMENT (bin)));
 332   DEBUG_ENTER_STRING;
 333
 334   gst_bin_schedule_cleanup(bin);
 335
 336   // next we have to find all the separate scheduling chains
 337   DEBUG("\nattempting to find scheduling chains...\n");
 338   // first make a copy of the managed_elements we can mess with
 339   elements = g_list_copy (bin->managed_elements);
 340   // we have to repeat until the list is empty to get all chains
 341   while (elements) {
 342     element = GST_ELEMENT (elements->data);
 343
 344     // if this is a DECOUPLED element
 345     if (GST_FLAG_IS_SET (element, GST_ELEMENT_DECOUPLED)) {
 346       // skip this element entirely
 347       DEBUG("skipping '%s' because it's decoupled\n",gst_element_get_name(element));
 348       elements = g_list_next (elements);
 349       continue;
 350     }
 351
 352     DEBUG("starting with element '%s'\n",gst_element_get_name(element));
 353
 354     // prime the pending list with the first element off the top
 355     pending = g_slist_prepend (NULL, element);
 356     // and remove that one from the main list
 357     elements = g_list_remove (elements, element);
 358
 359     // create a chain structure
 360     chain = g_new0 (_GstBinChain, 1);
 361
 362     // for each pending element, walk the pipeline
 363     do {
 364       // retrieve the top of the stack and pop it
 365       element = GST_ELEMENT (pending->data);
 366       pending = g_slist_remove (pending, element);
 367
 368       // add ourselves to the chain's list of elements
 369       DEBUG("adding '%s' to chain\n",gst_element_get_name(element));
 370       chain->elements = g_list_prepend (chain->elements, element);
 371       chain->num_elements++;
 372       // set the cothreads flag as appropriate
 373       if (GST_FLAG_IS_SET (element, GST_ELEMENT_USE_COTHREAD))
 374         chain->need_cothreads = TRUE;
 375       if (bin->use_cothreads == TRUE)
 376         chain->need_cothreads = TRUE;
 377
 378       // if we're managed by the current bin, and we're not decoupled,
 379       // go find all the peers and add them to the list of elements to check
 380       if ((element->manager == GST_ELEMENT(bin)) &&
 381           !GST_FLAG_IS_SET (element, GST_ELEMENT_DECOUPLED)) {
 382         // remove ourselves from the outer list of all managed elements
 383 //        DEBUG("removing '%s' from list of possible elements\n",gst_element_get_name(element));
 384         elements = g_list_remove (elements, element);
 385
 386         // if this element is a source, add it as an entry
 387         if (element->numsinkpads == 0) {
 388           chain->entries = g_list_prepend (chain->entries, element);
 389           DEBUG("added '%s' as SRC entry into the chain\n",gst_element_get_name(element));
 390         }
 391
 392         // now we have to walk the pads to find peers
 393         pads = gst_element_get_pad_list (element);
 394         while (pads) {
 395           pad = GST_PAD (pads->data);
 396           pads = g_list_next (pads);
 397           DEBUG("have pad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 398
 399           g_assert(pad->peer != NULL);
 400           g_assert(pad->peer->parent != NULL);
 401           //g_assert(GST_ELEMENT(pad->peer->parent)->manager != NULL);
 402
 403           DEBUG("peer pad %p\n", pad->peer);
 404           // only bother with if the pad's peer's parent is this bin or it's DECOUPLED
 405           // only add it if it's in the list of un-visited elements still
 406           if ((g_list_find (elements, pad->peer->parent) != NULL) ||
 407               GST_FLAG_IS_SET (pad->peer->parent, GST_ELEMENT_DECOUPLED)) {
 408             // add the peer element to the pending list
 409             DEBUG("adding '%s' to list of pending elements\n",gst_element_get_name(GST_ELEMENT(pad->peer->parent)));
 410             pending = g_slist_prepend (pending, GST_ELEMENT(pad->peer->parent));
 411
 412             // if this is a sink pad, then the element on the other side is an entry
 413             if ((gst_pad_get_direction (pad) == GST_PAD_SINK) &&
 414                 (GST_FLAG_IS_SET (pad->peer->parent, GST_ELEMENT_DECOUPLED))) {
 415               chain->entries = g_list_prepend (chain->entries, pad->peer->parent);
 416               DEBUG("added '%s' as DECOUPLED entry into the chain\n",gst_element_get_name(GST_ELEMENT(pad->peer->parent)));
 417             }
 418           } else
 419             DEBUG("element '%s' has already been dealt with\n",gst_element_get_name(GST_ELEMENT(pad->peer->parent)));
 420         }
 421       }
 422     } while (pending);
 423
 424     // add the chain to the bin
 425     DEBUG("have chain with %d elements: ",chain->num_elements);
 426     { GList *elements = chain->elements;
 427       while (elements) {
 428         element = GST_ELEMENT (elements->data);
 429         elements = g_list_next(elements);
 430         DEBUG_NOPREFIX("%s, ",gst_element_get_name(element));
 431       }
 432     }
 433     DEBUG_NOPREFIX("\n");
 434     bin->chains = g_list_prepend (bin->chains, chain);
 435     bin->num_chains++;
 436   }
 437   // free up the list in case it's full of DECOUPLED elements
 438   g_list_free (elements);
 439
 440   DEBUG("\nwe have %d chains to schedule\n",bin->num_chains);
 441
 442   // now we have to go through all the chains and schedule them
 443   chains = bin->chains;
 444   while (chains) {
 445     chain = (_GstBinChain *)(chains->data);
 446     chains = g_list_next (chains);
 447
 448     // schedule as appropriate
 449     if (chain->need_cothreads) {
 450       gst_schedule_cothreaded_chain (bin,chain);
 451     } else {
 452       gst_schedule_chained_chain (bin,chain);
 453     }
 454   }
 455
 456   DEBUG_LEAVE("(\"%s\")",gst_element_get_name(GST_ELEMENT(bin)));
 457 }
 458
 459
 460 /*
 461         // ***** check for possible connections outside
 462         // get the pad's peer
 463         peer = gst_pad_get_peer (pad);
 464         // FIXME this should be an error condition, if not disabled
 465         if (!peer) break;
 466         // get the parent of the peer of the pad
 467         outside = GST_ELEMENT (gst_pad_get_parent (peer));
 468         // FIXME this should *really* be an error condition
 469         if (!outside) break;
 470         // if it's a source or connection and it's not ours...
 471         if ((GST_IS_SRC (outside) || GST_IS_CONNECTION (outside)) &&
 472             (gst_object_get_parent (GST_OBJECT (outside)) != GST_OBJECT (bin))) {
 473           if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 474             DEBUG("dealing with outside source element %s\n",gst_element_get_name(outside));
 475 //            DEBUG("PUNT: copying pullfunc ptr from %s:%s to %s:%s (@ %p)\n",
 476 //GST_DEBUG_PAD_NAME(pad->peer),GST_DEBUG_PAD_NAME(pad),&pad->pullfunc);
 477 //            pad->pullfunc = pad->peer->pullfunc;
 478 //            DEBUG("PUNT: setting pushfunc proxy to fake proxy on %s:%s\n",GST_DEBUG_PAD_NAME(pad->peer));
 479 //            pad->peer->pushfunc = GST_DEBUG_FUNCPTR(gst_bin_pushfunc_fake_proxy);
 480             pad->pullfunc = GST_DEBUG_FUNCPTR(gst_bin_pullfunc_proxy);
 481           }
 482         } else {
 483 */
 484
 485
 486
 487
 488
 489 /*
 490       } else if (GST_IS_SRC (element)) {
 491         DEBUG("adding '%s' as entry point, because it's a source\n",gst_element_get_name (element));
 492         bin->entries = g_list_prepend (bin->entries,element);
 493         bin->num_entries++;
 494         cothread_setfunc(element->threadstate,gst_bin_src_wrapper,0,(char **)element);
 495       }
 496
 497       pads = gst_element_get_pad_list (element);
 498       while (pads) {
 499         pad = GST_PAD(pads->data);
 500
 501         if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 502           DEBUG("setting push proxy for sinkpad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 503           // set the proxy functions
 504           pad->pushfunc = GST_DEBUG_FUNCPTR(gst_bin_pushfunc_proxy);
 505           DEBUG("pushfunc %p = gst_bin_pushfunc_proxy %p\n",&pad->pushfunc,gst_bin_pushfunc_proxy);
 506         } else if (gst_pad_get_direction (pad) == GST_PAD_SRC) {
 507           DEBUG("setting pull proxies for srcpad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 508           // set the proxy functions
 509           pad->pullfunc = GST_DEBUG_FUNCPTR(gst_bin_pullfunc_proxy);
 510           DEBUG("pad->pullfunc(@%p) = gst_bin_pullfunc_proxy(@%p)\n",
 511                 &pad->pullfunc,gst_bin_pullfunc_proxy);
 512           pad->pullregionfunc = GST_DEBUG_FUNCPTR(gst_bin_pullregionfunc_proxy);
 513         }
 514         pads = g_list_next (pads);
 515       }
 516       elements = g_list_next (elements);
 517
 518       // if there are no entries, we have to pick one at random
 519       if (bin->num_entries == 0)
 520         bin->entries = g_list_prepend (bin->entries, GST_ELEMENT(bin->children->data));
 521     }
 522   } else {
 523     DEBUG("don't need cothreads, looking for entry points\n");
 524     // we have to find which elements will drive an iteration
 525     elements = bin->children;
 526     while (elements) {
 527       element = GST_ELEMENT (elements->data);
 528       DEBUG("found element \"%s\"\n", gst_element_get_name (element));
 529       if (GST_IS_BIN (element)) {
 530         gst_bin_create_plan (GST_BIN (element));
 531       }
 532       if (GST_IS_SRC (element)) {
 533         DEBUG("adding '%s' as entry point, because it's a source\n",gst_element_get_name (element));
 534         bin->entries = g_list_prepend (bin->entries, element);
 535         bin->num_entries++;
 536       }
 537
 538       // go through all the pads, set pointers, and check for connections
 539       pads = gst_element_get_pad_list (element);
 540       while (pads) {
 541         pad = GST_PAD (pads->data);
 542
 543         if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 544           DEBUG("found SINK pad %s:%s\n", GST_DEBUG_PAD_NAME(pad));
 545
 546           // copy the peer's chain function, easy enough
 547           DEBUG("copying peer's chainfunc to %s:%s's pushfunc\n",GST_DEBUG_PAD_NAME(pad));
 548           pad->pushfunc = GST_DEBUG_FUNCPTR(pad->peer->chainfunc);
 549
 550           // need to walk through and check for outside connections
 551 //FIXME need to do this for all pads
 552           // get the pad's peer
 553           peer = gst_pad_get_peer (pad);
 554           if (!peer) {
 555             DEBUG("found SINK pad %s has no peer\n", gst_pad_get_name (pad));
 556             break;
 557           }
 558           // get the parent of the peer of the pad
 559           outside = GST_ELEMENT (gst_pad_get_parent (peer));
 560           if (!outside) break;
 561           // if it's a connection and it's not ours...
 562           if (GST_IS_CONNECTION (outside) &&
 563                (gst_object_get_parent (GST_OBJECT (outside)) != GST_OBJECT (bin))) {
 564             gst_info("gstbin: element \"%s\" is the external source Connection "
 565                                     "for internal element \"%s\"\n",
 566                           gst_element_get_name (GST_ELEMENT (outside)),
 567                           gst_element_get_name (GST_ELEMENT (element)));
 568             bin->entries = g_list_prepend (bin->entries, outside);
 569             bin->num_entries++;
 570           }
 571         }
 572         else {
 573           DEBUG("found pad %s\n", gst_pad_get_name (pad));
 574         }
 575         pads = g_list_next (pads);
 576
 577       }
 578       elements = g_list_next (elements);
 579     }
 580 */
 581
 582
 583
 584
 585 /*
 586   // If cothreads are needed, we need to not only find elements but
 587   // set up cothread states and various proxy functions.
 588   if (bin->need_cothreads) {
 589     DEBUG("bin is using cothreads\n");
 590
 591     // first create thread context
 592     if (bin->threadcontext == NULL) {
 593       DEBUG("initializing cothread context\n");
 594       bin->threadcontext = cothread_init ();
 595     }
 596
 597     // walk through all the children
 598     elements = bin->managed_elements;
 599     while (elements) {
 600       element = GST_ELEMENT (elements->data);
 601       elements = g_list_next (elements);
 602
 603       // start out with a NULL warpper function, we'll set it if we want a cothread
 604       wrapper_function = NULL;
 605
 606       // have to decide if we need to or can use a cothreads, and if so which wrapper
 607       // first of all, if there's a loopfunc, the decision's already made
 608       if (element->loopfunc != NULL) {
 609         wrapper_function = GST_DEBUG_FUNCPTR(gst_bin_loopfunc_wrapper);
 610         DEBUG("element %s is a loopfunc, must use a cothread\n",gst_element_get_name(element));
 611       } else {
 612         // otherwise we need to decide if it needs a cothread
 613         // if it's complex, or cothreads are preferred and it's *not* decoupled, cothread it
 614         if (GST_FLAG_IS_SET (element,GST_ELEMENT_COMPLEX) ||
 615             (GST_FLAG_IS_SET (bin,GST_BIN_FLAG_PREFER_COTHREADS) &&
 616              !GST_FLAG_IS_SET (element,GST_ELEMENT_DECOUPLED))) {
 617           // base it on whether we're going to loop through source or sink pads
 618           if (element->numsinkpads == 0)
 619             wrapper_function = GST_DEBUG_FUNCPTR(gst_bin_src_wrapper);
 620           else
 621             wrapper_function = GST_DEBUG_FUNCPTR(gst_bin_chain_wrapper);
 622         }
 623       }
 624
 625       // walk through the all the pads for this element, setting proxy functions
 626       // the selection of proxy functions depends on whether we're in a cothread or not
 627       pads = gst_element_get_pad_list (element);
 628       while (pads) {
 629         pad = GST_PAD (pads->data);
 630         pads = g_list_next (pads);
 631
 632         // check to see if someone else gets to set up the element
 633         peer_manager = GST_ELEMENT((pad)->peer->parent)->manager;
 634         if (peer_manager != GST_ELEMENT(bin)) {
 635           DEBUG("WARNING: pad %s:%s is connected outside of bin\n",GST_DEBUG_PAD_NAME(pad));
 636         }
 637
 638         // if the wrapper_function is set, we need to use the proxy functions
 639         if (wrapper_function != NULL) {
 640           // set up proxy functions
 641           if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 642             DEBUG("setting push proxy for sinkpad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 643             pad->pushfunc = GST_DEBUG_FUNCPTR(gst_bin_pushfunc_proxy);
 644           } else if (gst_pad_get_direction (pad) == GST_PAD_SRC) {
 645             DEBUG("setting pull proxy for srcpad %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 646             pad->pullfunc = GST_DEBUG_FUNCPTR(gst_bin_pullfunc_proxy);
 647           }
 648         } else {
 649           // otherwise we need to set up for 'traditional' chaining
 650           if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 651             // we can just copy the chain function, since it shares the prototype
 652             DEBUG("copying chain function into push proxy for %s:%s\n",
 653                   GST_DEBUG_PAD_NAME(pad));
 654             pad->pushfunc = pad->chainfunc;
 655           } else if (gst_pad_get_direction (pad) == GST_PAD_SRC) {
 656             // we can just copy the get function, since it shares the prototype
 657             DEBUG("copying get function into pull proxy for %s:%s\n",
 658                   GST_DEBUG_PAD_NAME(pad));
 659             pad->pullfunc = pad->getfunc;
 660           }
 661         }
 662       }
 663
 664       // if a loopfunc has been specified, create and set up a cothread
 665       if (wrapper_function != NULL) {
 666         if (element->threadstate == NULL) {
 667           element->threadstate = cothread_create (bin->threadcontext);
 668           DEBUG("created cothread %p (@%p) for \"%s\"\n",element->threadstate,
 669                 &element->threadstate,gst_element_get_name(element));
 670         }
 671         cothread_setfunc (element->threadstate, wrapper_function, 0, (char **)element);
 672         DEBUG("set wrapper function for \"%s\" to &%s\n",gst_element_get_name(element),
 673               GST_DEBUG_FUNCPTR_NAME(wrapper_function));
 674       }
 675
 676 //      // HACK: if the element isn't decoupled, it's an entry
 677 //      if (!GST_FLAG_IS_SET(element,GST_ELEMENT_DECOUPLED))
 678 //        bin->entries = g_list_append(bin->entries, element);
 679     }
 680
 681   // otherwise, cothreads are not needed
 682   } else {
 683     DEBUG("bin is chained, no cothreads needed\n");
 684
 685     elements = bin->managed_elements;
 686     while (elements) {
 687       element = GST_ELEMENT (elements->data);
 688       elements = g_list_next (elements);
 689
 690       pads = gst_element_get_pad_list (element);
 691       while (pads) {
 692         pad = GST_PAD (pads->data);
 693         pads = g_list_next (pads);
 694
 695         if (gst_pad_get_direction (pad) == GST_PAD_SINK) {
 696           DEBUG("copying chain function into push proxy for %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 697           pad->pushfunc = pad->chainfunc;
 698         } else {
 699           DEBUG("copying get function into pull proxy for %s:%s\n",GST_DEBUG_PAD_NAME(pad));
 700           pad->pullfunc = pad->getfunc;
 701         }
 702       }
 703     }
 704   }
 705 */
 706
 707