gst/gstthread.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *
   5  * gstthread.c: Threaded container object
   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 #include <unistd.h>
  24
  25 /* #define GST_DEBUG_ENABLED */
  26 #include "gst_private.h"
  27
  28 #include "gstthread.h"
  29 #include "gstscheduler.h"
  30 #include "gstqueue.h"
  31
  32 GstElementDetails gst_thread_details = {
  33   "Threaded container",
  34   "Bin",
  35   "Container that creates/manages a thread",
  36   VERSION,
  37   "Erik Walthinsen <omega@cse.ogi.edu>",
  38   "(C) 1999, 2000",
  39 };
  40
  41
  42 /* Thread signals and args */
  43 enum {
  44   /* FILL ME */
  45   LAST_SIGNAL
  46 };
  47
  48 enum {
  49   SPINUP=0,
  50   STATECHANGE,
  51   STARTUP
  52 };
  53
  54 enum {
  55   ARG_0,
  56 };
  57
  58
  59
  60 static void                     gst_thread_class_init           (GstThreadClass *klass);
  61 static void                     gst_thread_init                 (GstThread *thread);
  62
  63 static void                     gst_thread_dispose      (GObject *object);
  64
  65 static void                     gst_thread_set_property         (GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec);
  66 static void                     gst_thread_get_property         (GObject *object, guint prop_id, GValue *value, GParamSpec *pspec);
  67
  68 static GstElementStateReturn    gst_thread_change_state         (GstElement *element);
  69
  70 #ifndef GST_DISABLE_LOADSAVE
  71 static xmlNodePtr               gst_thread_save_thyself         (GstObject *object, xmlNodePtr parent);
  72 static void                     gst_thread_restore_thyself      (GstObject *object, xmlNodePtr self);
  73 #endif
  74
  75 static void*                    gst_thread_main_loop            (void *arg);
  76
  77 static GstBinClass *parent_class = NULL;
  78 /* static guint gst_thread_signals[LAST_SIGNAL] = { 0 }; */
  79
  80 GType
  81 gst_thread_get_type(void) {
  82   static GType thread_type = 0;
  83
  84   if (!thread_type) {
  85     static const GTypeInfo thread_info = {
  86       sizeof(GstThreadClass),
  87       NULL,
  88       NULL,
  89       (GClassInitFunc)gst_thread_class_init,
  90       NULL,
  91       NULL,
  92       sizeof(GstThread),
  93       4,
  94       (GInstanceInitFunc)gst_thread_init,
  95       NULL
  96     };
  97     thread_type = g_type_register_static(GST_TYPE_BIN, "GstThread", &thread_info, 0);
  98   }
  99   return thread_type;
 100 }
 101
 102 static void
 103 gst_thread_class_init (GstThreadClass *klass)
 104 {
 105   GObjectClass *gobject_class;
 106   GstObjectClass *gstobject_class;
 107   GstElementClass *gstelement_class;
 108   GstBinClass *gstbin_class;
 109
 110   gobject_class =       (GObjectClass*)klass;
 111   gstobject_class =     (GstObjectClass*)klass;
 112   gstelement_class =    (GstElementClass*)klass;
 113   gstbin_class =        (GstBinClass*)klass;
 114
 115   parent_class = g_type_class_ref (GST_TYPE_BIN);
 116
 117   gobject_class->dispose =              gst_thread_dispose;
 118
 119 #ifndef GST_DISABLE_LOADSAVE
 120   gstobject_class->save_thyself =       GST_DEBUG_FUNCPTR (gst_thread_save_thyself);
 121   gstobject_class->restore_thyself =    GST_DEBUG_FUNCPTR(gst_thread_restore_thyself);
 122 #endif
 123
 124   gstelement_class->change_state =      GST_DEBUG_FUNCPTR (gst_thread_change_state);
 125
 126 /*  gstbin_class->schedule = gst_thread_schedule_dummy; */
 127
 128   gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_thread_set_property);
 129   gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_thread_get_property);
 130
 131 }
 132
 133 static void
 134 gst_thread_init (GstThread *thread)
 135 {
 136
 137   GST_DEBUG (GST_CAT_THREAD,"initializing thread\n");
 138
 139   /* we're a manager by default */
 140   GST_FLAG_SET (thread, GST_BIN_FLAG_MANAGER);
 141
 142   thread->lock = g_mutex_new();
 143   thread->cond = g_cond_new();
 144
 145   GST_ELEMENT_SCHED(thread) = gst_schedulerfactory_make ("basic", GST_ELEMENT(thread));
 146   GST_DEBUG(GST_CAT_THREAD, "thread's scheduler is %p\n",GST_ELEMENT_SCHED(thread));
 147
 148   thread->ppid = getpid();
 149   thread->thread_id = -1;
 150
 151 /*  gst_element_set_manager(GST_ELEMENT(thread),GST_ELEMENT(thread)); */
 152 }
 153
 154 static void
 155 gst_thread_dispose (GObject *object)
 156 {
 157   GstThread *thread = GST_THREAD (object);
 158
 159   GST_DEBUG (GST_CAT_REFCOUNTING,"dispose\n");
 160
 161   g_mutex_free (thread->lock);
 162   g_cond_free (thread->cond);
 163
 164   G_OBJECT_CLASS (parent_class)->dispose (object);
 165
 166   gst_object_destroy (GST_OBJECT (GST_ELEMENT_SCHED (thread)));
 167   gst_object_unref (GST_OBJECT (GST_ELEMENT_SCHED (thread)));
 168
 169 }
 170
 171 static void
 172 gst_thread_set_property (GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec)
 173 {
 174   /* it's not null if we got it, but it might not be ours */
 175   g_return_if_fail (GST_IS_THREAD (object));
 176
 177   switch (prop_id) {
 178     default:
 179       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 180       break;
 181   }
 182 }
 183
 184 static void
 185 gst_thread_get_property (GObject *object, guint prop_id, GValue *value, GParamSpec *pspec)
 186 {
 187   /* it's not null if we got it, but it might not be ours */
 188   g_return_if_fail (GST_IS_THREAD (object));
 189
 190   switch (prop_id) {
 191     default:
 192       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 193       break;
 194   }
 195 }
 196
 197
 198 /**
 199  * gst_thread_new:
 200  * @name: the name of the thread
 201  *
 202  * Create a new thread with the given name.
 203  *
 204  * Returns: The new thread
 205  */
 206 GstElement*
 207 gst_thread_new (const gchar *name)
 208 {
 209   return gst_elementfactory_make ("thread", name);
 210 }
 211
 212
 213 #define THR_INFO(format,args...) \
 214   GST_INFO_ELEMENT(GST_CAT_THREAD, thread, "sync(" GST_DEBUG_THREAD_FORMAT "): " format , \
 215   GST_DEBUG_THREAD_ARGS(thread->pid) , ## args )
 216 #define THR_DEBUG(format,args...) \
 217   GST_DEBUG_ELEMENT(GST_CAT_THREAD, thread, "sync(" GST_DEBUG_THREAD_FORMAT "): " format , \
 218   GST_DEBUG_THREAD_ARGS(thread->pid) , ## args )
 219
 220 #define THR_INFO_MAIN(format,args...) \
 221   GST_INFO_ELEMENT(GST_CAT_THREAD, thread, "sync-main(" GST_DEBUG_THREAD_FORMAT "): " format , \
 222   GST_DEBUG_THREAD_ARGS(thread->ppid) , ## args )
 223 #define THR_DEBUG_MAIN(format,args...) \
 224   GST_DEBUG_ELEMENT(GST_CAT_THREAD, thread, "sync-main(" GST_DEBUG_THREAD_FORMAT "): " format , \
 225   GST_DEBUG_THREAD_ARGS(thread->ppid) , ## args )
 226
 227 static GstElementStateReturn
 228 gst_thread_update_state (GstThread *thread)
 229 {
 230   /* check for state change */
 231   if (GST_STATE_PENDING(thread) != GST_STATE_VOID_PENDING) {
 232     /* punt and change state on all the children */
 233     if (GST_ELEMENT_CLASS (parent_class)->change_state)
 234       return GST_ELEMENT_CLASS (parent_class)->change_state (GST_ELEMENT(thread));
 235   }
 236
 237   return GST_STATE_SUCCESS;
 238 }
 239
 240
 241 static GstElementStateReturn
 242 gst_thread_change_state (GstElement * element)
 243 {
 244   GstThread *thread;
 245   gboolean stateset = GST_STATE_SUCCESS;
 246   gint transition;
 247   pthread_t self = pthread_self ();
 248
 249   g_return_val_if_fail (GST_IS_THREAD (element), FALSE);
 250
 251   thread = GST_THREAD (element);
 252
 253   transition = GST_STATE_TRANSITION (element);
 254
 255   THR_INFO ("changing state from %s to %s",
 256             gst_element_statename (GST_STATE (element)),
 257             gst_element_statename (GST_STATE_PENDING (element)));
 258
 259   if (pthread_equal (self, thread->thread_id)) {
 260     GST_DEBUG (GST_CAT_THREAD,
 261                "no sync(" GST_DEBUG_THREAD_FORMAT "): setting own thread's state to spinning\n",
 262                GST_DEBUG_THREAD_ARGS (thread->pid));
 263     return gst_thread_update_state (thread);
 264   }
 265
 266   switch (transition) {
 267     case GST_STATE_NULL_TO_READY:
 268       /* set the state to idle */
 269       GST_FLAG_UNSET (thread, GST_THREAD_STATE_SPINNING);
 270
 271       THR_DEBUG ("creating thread \"%s\"\n", GST_ELEMENT_NAME (element));
 272
 273       g_mutex_lock (thread->lock);
 274
 275       /* create the thread */
 276       pthread_create (&thread->thread_id, NULL, gst_thread_main_loop, thread);
 277
 278       /* wait for it to 'spin up' */
 279       THR_DEBUG ("waiting for child thread spinup\n");
 280       g_cond_wait (thread->cond, thread->lock);
 281       THR_DEBUG ("thread claims to be up\n");
 282       g_mutex_unlock (thread->lock);
 283       break;
 284     case GST_STATE_READY_TO_PAUSED:
 285       THR_INFO ("readying thread");
 286       g_mutex_lock (thread->lock);
 287       THR_DEBUG ("signaling\n");
 288       g_cond_signal (thread->cond);
 289       THR_DEBUG ("waiting for ack\n");
 290       g_cond_wait (thread->cond, thread->lock);
 291       THR_DEBUG ("got ack\n");
 292       g_mutex_unlock (thread->lock);
 293       break;
 294     case GST_STATE_PAUSED_TO_PLAYING:
 295       THR_DEBUG ("telling thread to start spinning\n");
 296       g_mutex_lock (thread->lock);
 297       THR_DEBUG ("signaling\n");
 298       g_cond_signal (thread->cond);
 299       THR_DEBUG ("waiting for ack\n");
 300       g_cond_wait (thread->cond, thread->lock);
 301       THR_DEBUG ("got ack\n");
 302       g_mutex_unlock (thread->lock);
 303       break;
 304     case GST_STATE_PLAYING_TO_PAUSED:
 305     {
 306       GList *elements = (element->sched)->elements;
 307
 308       THR_INFO ("pausing thread");
 309
 310       /* the following code ensures that the bottom half of thread will run
 311        * to perform each elements' change_state() (by calling gstbin.c::
 312        * change_state()).
 313        * + the pending state was already set by gstelement.c::set_state()
 314        * + find every queue we manage, and signal its empty and full conditions
 315        */
 316          g_mutex_lock (thread->lock);
 317
 318       GST_FLAG_UNSET (thread, GST_THREAD_STATE_SPINNING);
 319
 320       while (elements) {
 321         GstElement *e = GST_ELEMENT (elements->data);
 322
 323         g_assert (e);
 324         THR_DEBUG ("  element \"%s\"\n", GST_ELEMENT_NAME (e));
 325         elements = g_list_next (elements);
 326         if (GST_IS_QUEUE (e)) {
 327   /* FIXME make this more efficient by only waking queues that are asleep
 328    *  FIXME and only waking the appropriate condition (depending on if it's
 329    *  FIXME on up- or down-stream side)
 330    * FIXME also make this more efficient by keeping list of managed queues
 331    */
 332           THR_DEBUG ("waking queue \"%s\"\n", GST_ELEMENT_NAME (e));
 333           gst_element_set_state (e, GST_STATE_PAUSED);
 334         }
 335         else {
 336           GList *pads = GST_ELEMENT_PADS (e);
 337
 338           while (pads) {
 339             GstRealPad *peer;
 340             GstElement *peerelement;
 341             GstPad *p = GST_PAD (pads->data);
 342
 343             pads = g_list_next (pads);
 344
 345             peer = GST_PAD_PEER (p);
 346             if (!peer)
 347               continue;
 348
 349             peerelement = GST_PAD_PARENT (peer);
 350             if (!peerelement)
 351               continue;         /* deal with case where there's no peer */
 352
 353             if (!GST_FLAG_IS_SET (peerelement, GST_ELEMENT_DECOUPLED)) {
 354               GST_DEBUG (GST_CAT_THREAD, "peer element isn't DECOUPLED\n");
 355               continue;
 356             }
 357
 358             /* FIXME this needs to go away eventually */
 359             if (!GST_IS_QUEUE (peerelement)) {
 360               GST_DEBUG (GST_CAT_THREAD, "peer element isn't a Queue\n");
 361               continue;
 362             }
 363
 364             if (GST_ELEMENT_SCHED (peerelement) != GST_ELEMENT_SCHED (thread)) {
 365               GstQueue *queue = GST_QUEUE (peerelement);
 366
 367               THR_DEBUG ("  element \"%s\" has pad cross sched boundary\n", GST_ELEMENT_NAME (e));
 368               /* FIXME!! */
 369               g_mutex_lock (queue->qlock);
 370               g_cond_signal (queue->not_full);
 371               g_cond_signal (queue->not_empty);
 372               g_mutex_unlock (queue->qlock);
 373             }
 374           }
 375         }
 376       }
 377       THR_DEBUG ("telling thread to pause, signaling\n");
 378       g_cond_signal (thread->cond);
 379       THR_DEBUG ("waiting for ack\n");
 380       g_cond_wait (thread->cond, thread->lock);
 381       THR_DEBUG ("got ack\n");
 382       g_mutex_unlock (thread->lock);
 383       break;
 384     }
 385     case GST_STATE_READY_TO_NULL:
 386       THR_DEBUG ("telling thread to pause (null) - and joining\n");
 387       /* MattH FIXME revisit */
 388       g_mutex_lock (thread->lock);
 389       THR_DEBUG ("signaling\n");
 390       g_cond_signal (thread->cond);
 391       THR_DEBUG ("waiting for ack\n");
 392       g_cond_wait (thread->cond, thread->lock);
 393       THR_DEBUG ("got ack\n");
 394       pthread_join (thread->thread_id, NULL);
 395       thread->thread_id = -1;
 396       g_mutex_unlock (thread->lock);
 397
 398       GST_FLAG_UNSET (thread, GST_THREAD_STATE_REAPING);
 399       GST_FLAG_UNSET (thread, GST_THREAD_STATE_STARTED);
 400       GST_FLAG_UNSET (thread, GST_THREAD_STATE_SPINNING);
 401
 402       break;
 403     case GST_STATE_PAUSED_TO_READY:
 404       THR_DEBUG ("telling thread to stop spinning\n");
 405       g_mutex_lock (thread->lock);
 406       THR_DEBUG ("signaling\n");
 407       g_cond_signal (thread->cond);
 408       THR_DEBUG ("waiting for ack\n");
 409       g_cond_wait (thread->cond, thread->lock);
 410       THR_DEBUG ("got ack\n");
 411       g_mutex_unlock (thread->lock);
 412
 413       break;
 414     default:
 415       GST_DEBUG_ELEMENT (GST_CAT_THREAD, element, "UNHANDLED STATE CHANGE! %x\n", transition);
 416       break;
 417   }
 418
 419   return stateset;
 420 }
 421
 422 /**
 423  * gst_thread_main_loop:
 424  * @arg: the thread to start
 425  *
 426  * The main loop of the thread. The thread will iterate
 427  * while the state is GST_THREAD_STATE_SPINNING
 428  */
 429 static void *
 430 gst_thread_main_loop (void *arg)
 431 {
 432   GstThread *thread = GST_THREAD (arg);
 433   gint stateset;
 434
 435   g_mutex_lock (thread->lock);
 436
 437   gst_scheduler_setup (GST_ELEMENT_SCHED (thread));
 438   GST_FLAG_UNSET (thread, GST_THREAD_STATE_REAPING);
 439
 440   thread->pid = getpid();
 441   THR_INFO_MAIN("thread is running");
 442
 443   /* first we need to change the state of all the children */
 444   if (GST_ELEMENT_CLASS (parent_class)->change_state) {
 445     stateset = GST_ELEMENT_CLASS (parent_class)->change_state (GST_ELEMENT(thread));
 446
 447     if (stateset != GST_STATE_SUCCESS) {
 448       THR_DEBUG_MAIN ("state change of children failed\n");
 449     }
 450   }
 451
 452
 453   THR_DEBUG_MAIN ("indicating spinup\n");
 454   g_cond_signal (thread->cond);
 455   /* don't unlock the mutex because we hold it into the top of the while loop */
 456   THR_DEBUG_MAIN ("thread has indicated spinup to parent process\n");
 457
 458   /***** THREAD IS NOW IN READY STATE *****/
 459
 460   while (!GST_FLAG_IS_SET (thread, GST_THREAD_STATE_REAPING)) {
 461     /* NOTE we hold the thread lock at this point */
 462     /* what we do depends on what state we're in */
 463     switch (GST_STATE (thread)) {
 464       /* NOTE: cannot be in NULL, we're not running in that state at all */
 465       case GST_STATE_READY:
 466         /* wait to be set to either the NULL or PAUSED states */
 467         THR_DEBUG_MAIN ("thread in %s state, waiting for either %s or %s\n",
 468                         gst_element_statename (GST_STATE_READY),
 469                         gst_element_statename (GST_STATE_NULL),
 470                         gst_element_statename (GST_STATE_PAUSED));
 471         g_cond_wait (thread->cond,thread->lock);
 472
 473         g_assert (GST_STATE_PENDING (thread) == GST_STATE_NULL ||
 474                   GST_STATE_PENDING (thread) == GST_STATE_PAUSED);
 475
 476         /* been signaled, we need to state transition now and signal back */
 477         gst_thread_update_state (thread);
 478         THR_DEBUG_MAIN ("done with state transition, signaling back to parent process\n");
 479         g_cond_signal (thread->cond);
 480         /* now we decide what to do next */
 481         if (GST_STATE (thread) == GST_STATE_NULL) {
 482           /* REAPING must be set, we can simply break this iteration */
 483           GST_FLAG_SET (thread, GST_THREAD_STATE_REAPING);
 484         }
 485         continue;
 486       case GST_STATE_PAUSED:
 487         /* wait to be set to either the READY or PLAYING states */
 488         THR_DEBUG_MAIN("thread in %s state, waiting for either %s or %s\n",
 489                        gst_element_statename (GST_STATE_PAUSED),
 490                        gst_element_statename (GST_STATE_READY),
 491                        gst_element_statename (GST_STATE_PLAYING));
 492         g_cond_wait (thread->cond,thread->lock);
 493
 494         g_assert (GST_STATE_PENDING (thread) == GST_STATE_READY ||
 495                   GST_STATE_PENDING (thread) == GST_STATE_PLAYING);
 496
 497         /* been signaled, we need to state transition now and signal back */
 498         gst_thread_update_state (thread);
 499         /* now we decide what to do next */
 500         if (GST_STATE (thread) != GST_STATE_PLAYING) {
 501           /* either READY or the state change failed for some reason */
 502           g_cond_signal (thread->cond);
 503           continue;
 504         }
 505         else {
 506           GST_FLAG_SET (thread, GST_THREAD_STATE_SPINNING);
 507           /* PLAYING is coming up, so we can now start spinning */
 508           while (GST_FLAG_IS_SET (thread, GST_THREAD_STATE_SPINNING)) {
 509             gboolean status;
 510
 511             g_cond_signal (thread->cond);
 512             g_mutex_unlock (thread->lock);
 513             status = gst_bin_iterate (GST_BIN (thread));
 514             g_mutex_lock (thread->lock);
 515             /* g_cond_signal(thread->cond); */
 516
 517             if (!status || GST_STATE_PENDING (thread) != GST_STATE_VOID_PENDING)
 518               GST_FLAG_UNSET (thread, GST_THREAD_STATE_SPINNING);
 519           }
 520           /* looks like we were stopped because of a statechange */
 521           if (GST_STATE_PENDING (thread)) {
 522             gst_thread_update_state (thread);
 523           }
 524           /* once we're here, SPINNING has stopped, we should signal that we're done */
 525           THR_DEBUG_MAIN ("SPINNING stopped, signaling back to parent process\n");
 526           g_cond_signal (thread->cond);
 527           /* now we can wait for PAUSED */
 528           continue;
 529         }
 530       case GST_STATE_PLAYING:
 531         /* wait to be set to PAUSED */
 532         THR_DEBUG_MAIN ("thread in %s state, waiting for %s\n",
 533                         gst_element_statename(GST_STATE_PLAYING),
 534                         gst_element_statename(GST_STATE_PAUSED));
 535         g_cond_wait (thread->cond,thread->lock);
 536
 537         /* been signaled, we need to state transition now and signal back */
 538         gst_thread_update_state (thread);
 539         g_cond_signal (thread->cond);
 540         /* now we decide what to do next */
 541         /* there's only PAUSED, we we just wait for it */
 542         continue;
 543       case GST_STATE_NULL:
 544         THR_DEBUG_MAIN ("thread in %s state, preparing to die\n",
 545                         gst_element_statename(GST_STATE_NULL));
 546         GST_FLAG_SET (thread, GST_THREAD_STATE_REAPING);
 547         break;
 548       default:
 549         g_assert_not_reached ();
 550         break;
 551     }
 552   }
 553   /* we need to destroy the scheduler here bacause it has mapped it's
 554    * stack into the threads stack space */
 555   gst_scheduler_reset (GST_ELEMENT_SCHED (thread));
 556
 557   /* since we don't unlock at the end of the while loop, do it here */
 558   g_mutex_unlock (thread->lock);
 559
 560   GST_INFO (GST_CAT_THREAD, "gstthread: thread \"%s\" is stopped",
 561                   GST_ELEMENT_NAME (thread));
 562   return NULL;
 563 }
 564
 565 #ifndef GST_DISABLE_LOADSAVE
 566 static xmlNodePtr
 567 gst_thread_save_thyself (GstObject *object,
 568                          xmlNodePtr self)
 569 {
 570   if (GST_OBJECT_CLASS (parent_class)->save_thyself)
 571     GST_OBJECT_CLASS (parent_class)->save_thyself (object, self);
 572   return NULL;
 573 }
 574
 575 static void
 576 gst_thread_restore_thyself (GstObject *object,
 577                             xmlNodePtr self)
 578 {
 579   GST_DEBUG (GST_CAT_THREAD,"gstthread: restore\n");
 580
 581   if (GST_OBJECT_CLASS (parent_class)->restore_thyself)
 582     GST_OBJECT_CLASS (parent_class)->restore_thyself (object, self);
 583 }
 584 #endif /* GST_DISABLE_LOADSAVE */