gst/gstqueue.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *
   5  * gstqueue.c:
   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 DEBUG_ENABLED
  24 //#define STATUS_ENABLED
  25 #ifdef STATUS_ENABLED
  26 #define STATUS(A) GST_DEBUG(0,A, gst_element_get_name(GST_ELEMENT(queue)))
  27 #else
  28 #define STATUS(A)
  29 #endif
  30
  31 #include <pthread.h>
  32
  33 #include "config.h"
  34 #include "gst_private.h"
  35
  36 #include "gstqueue.h"
  37
  38 GstElementDetails gst_queue_details = {
  39   "Queue",
  40   "Connection",
  41   "Simple data queue",
  42   VERSION,
  43   "Erik Walthinsen <omega@cse.ogi.edu>",
  44   "(C) 1999",
  45 };
  46
  47
  48 /* Queue signals and args */
  49 enum {
  50   /* FILL ME */
  51   LAST_SIGNAL
  52 };
  53
  54 enum {
  55   ARG_0,
  56   ARG_LEVEL,
  57   ARG_MAX_LEVEL,
  58   ARG_BLOCK,
  59 };
  60
  61
  62 static void                     gst_queue_class_init    (GstQueueClass *klass);
  63 static void                     gst_queue_init          (GstQueue *queue);
  64
  65 static void                     gst_queue_set_arg       (GtkObject *object, GtkArg *arg, guint id);
  66 static void                     gst_queue_get_arg       (GtkObject *object, GtkArg *arg, guint id);
  67
  68 static gboolean                 gst_queue_handle_eos    (GstPad *pad);
  69 static GstPadNegotiateReturn    gst_queue_handle_negotiate_src (GstPad *pad, GstCaps **caps, gpointer *data);
  70 static GstPadNegotiateReturn    gst_queue_handle_negotiate_sink (GstPad *pad, GstCaps **caps, gpointer *data);
  71 static void                     gst_queue_chain         (GstPad *pad, GstBuffer *buf);
  72 static GstBuffer *              gst_queue_get           (GstPad *pad);
  73 static GstBufferPool*           gst_queue_get_bufferpool (GstPad *pad);
  74
  75 static void                     gst_queue_flush         (GstQueue *queue);
  76
  77 static GstElementStateReturn    gst_queue_change_state  (GstElement *element);
  78
  79
  80 static GstElementClass *parent_class = NULL;
  81 //static guint gst_queue_signals[LAST_SIGNAL] = { 0 };
  82
  83 GtkType
  84 gst_queue_get_type(void) {
  85   static GtkType queue_type = 0;
  86
  87   if (!queue_type) {
  88     static const GtkTypeInfo queue_info = {
  89       "GstQueue",
  90       sizeof(GstQueue),
  91       sizeof(GstQueueClass),
  92       (GtkClassInitFunc)gst_queue_class_init,
  93       (GtkObjectInitFunc)gst_queue_init,
  94       (GtkArgSetFunc)gst_queue_set_arg,
  95       (GtkArgGetFunc)gst_queue_get_arg,
  96       (GtkClassInitFunc)NULL,
  97     };
  98     queue_type = gtk_type_unique (GST_TYPE_ELEMENT, &queue_info);
  99   }
 100   return queue_type;
 101 }
 102
 103 static void
 104 gst_queue_class_init (GstQueueClass *klass)
 105 {
 106   GtkObjectClass *gtkobject_class;
 107   GstElementClass *gstelement_class;
 108
 109   gtkobject_class = (GtkObjectClass*)klass;
 110   gstelement_class = (GstElementClass*)klass;
 111
 112   parent_class = gtk_type_class (GST_TYPE_ELEMENT);
 113
 114   gtk_object_add_arg_type ("GstQueue::level", GTK_TYPE_INT,
 115                            GTK_ARG_READABLE, ARG_LEVEL);
 116   gtk_object_add_arg_type ("GstQueue::max_level", GTK_TYPE_INT,
 117                            GTK_ARG_READWRITE, ARG_MAX_LEVEL);
 118   gtk_object_add_arg_type ("GstQueue::block", GTK_TYPE_BOOL,
 119                            GTK_ARG_READWRITE, ARG_BLOCK);
 120
 121   gtkobject_class->set_arg = gst_queue_set_arg;
 122   gtkobject_class->get_arg = gst_queue_get_arg;
 123
 124   gstelement_class->change_state = gst_queue_change_state;
 125 }
 126
 127 static void
 128 gst_queue_init (GstQueue *queue)
 129 {
 130   // scheduling on this kind of element is, well, interesting
 131   GST_FLAG_SET (queue, GST_ELEMENT_DECOUPLED);
 132
 133   queue->sinkpad = gst_pad_new ("sink", GST_PAD_SINK);
 134   gst_pad_set_chain_function (queue->sinkpad, GST_DEBUG_FUNCPTR(gst_queue_chain));
 135   gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
 136   gst_pad_set_eos_function (queue->sinkpad, gst_queue_handle_eos);
 137   gst_pad_set_negotiate_function (queue->sinkpad, gst_queue_handle_negotiate_sink);
 138   gst_pad_set_bufferpool_function (queue->sinkpad, gst_queue_get_bufferpool);
 139
 140   queue->srcpad = gst_pad_new ("src", GST_PAD_SRC);
 141   gst_pad_set_get_function (queue->srcpad, GST_DEBUG_FUNCPTR(gst_queue_get));
 142   gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
 143   gst_pad_set_negotiate_function (queue->srcpad, gst_queue_handle_negotiate_src);
 144
 145   queue->queue = NULL;
 146   queue->level_buffers = 0;
 147   queue->max_buffers = 100;
 148   queue->block = TRUE;
 149   queue->level_bytes = 0;
 150   queue->size_buffers = 0;
 151   queue->size_bytes = 0;
 152
 153   queue->emptycond = g_cond_new ();
 154   queue->fullcond = g_cond_new ();
 155 }
 156
 157 static GstBufferPool*
 158 gst_queue_get_bufferpool (GstPad *pad)
 159 {
 160   GstQueue *queue;
 161
 162   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 163
 164   return gst_pad_get_bufferpool (queue->srcpad);
 165 }
 166
 167 static GstPadNegotiateReturn
 168 gst_queue_handle_negotiate_src (GstPad *pad, GstCaps **caps, gpointer *data)
 169 {
 170   GstQueue *queue;
 171
 172   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 173
 174   return gst_pad_negotiate_proxy (pad, queue->sinkpad, caps);
 175
 176
 177   //return GST_PAD_NEGOTIATE_FAIL;
 178 }
 179
 180 static GstPadNegotiateReturn
 181 gst_queue_handle_negotiate_sink (GstPad *pad, GstCaps **caps, gpointer *data)
 182 {
 183   GstQueue *queue;
 184
 185   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 186
 187   /*
 188   if (counter == 0) {
 189      *caps = NULL;
 190      return GST_PAD_NEGOTIATE_TRY;
 191   }
 192   if (*caps) {
 193   */
 194     return gst_pad_negotiate_proxy (pad, queue->srcpad, caps);
 195     /*
 196   }
 197
 198   return GST_PAD_NEGOTIATE_FAIL;
 199   */
 200 }
 201
 202 static gboolean
 203 gst_queue_handle_eos (GstPad *pad)
 204 {
 205   GstQueue *queue;
 206
 207   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 208
 209   GST_DEBUG (0,"queue: %s received eos\n", GST_ELEMENT_NAME (queue));
 210
 211   GST_LOCK (queue);
 212   GST_DEBUG (0,"queue: %s has %d buffers left\n", GST_ELEMENT_NAME (queue),
 213                   queue->level_buffers);
 214
 215   GST_FLAG_SET (pad, GST_PAD_EOS);
 216
 217   g_cond_signal (queue->emptycond);
 218
 219   GST_UNLOCK (queue);
 220
 221   return TRUE;
 222 }
 223
 224 static void
 225 gst_queue_cleanup_buffers (gpointer data, const gpointer user_data)
 226 {
 227   GST_DEBUG (0,"queue: %s cleaning buffer %p\n", (gchar *)user_data, data);
 228
 229   gst_buffer_unref (GST_BUFFER (data));
 230 }
 231
 232 static void
 233 gst_queue_flush (GstQueue *queue)
 234 {
 235   g_slist_foreach (queue->queue, gst_queue_cleanup_buffers,
 236                   (char *) GST_ELEMENT_NAME (queue));
 237   g_slist_free (queue->queue);
 238
 239   queue->queue = NULL;
 240   queue->level_buffers = 0;
 241   queue->timeval = NULL;
 242 }
 243
 244 static void
 245 gst_queue_chain (GstPad *pad, GstBuffer *buf)
 246 {
 247   GstQueue *queue;
 248   gboolean tosignal = FALSE;
 249   const guchar *name;
 250
 251   g_return_if_fail (pad != NULL);
 252   g_return_if_fail (GST_IS_PAD (pad));
 253   g_return_if_fail (buf != NULL);
 254
 255   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 256   name = GST_ELEMENT_NAME (queue);
 257
 258   /* we have to lock the queue since we span threads */
 259
 260   GST_DEBUG (0,"queue: try have queue lock\n");
 261   GST_LOCK (queue);
 262   GST_DEBUG (0,"queue: %s adding buffer %p %ld\n", name, buf, pthread_self ());
 263   GST_DEBUG (0,"queue: have queue lock\n");
 264
 265   if (GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLUSH)) {
 266     gst_queue_flush (queue);
 267   }
 268
 269   GST_DEBUG (0,"queue: %s: chain %d %p\n", name, queue->level_buffers, buf);
 270
 271   while (queue->level_buffers >= queue->max_buffers) {
 272     GST_DEBUG (0,"queue: %s waiting %d\n", name, queue->level_buffers);
 273     STATUS("%s: O\n");
 274     //g_cond_timed_wait (queue->fullcond, queue->fulllock, queue->timeval);
 275     //FIXME need to signal other thread in case signals got lost?
 276     g_cond_signal (queue->emptycond);
 277     g_cond_wait (queue->fullcond, GST_OBJECT(queue)->lock);
 278     STATUS("%s: O+\n");
 279     GST_DEBUG (0,"queue: %s waiting done %d\n", name, queue->level_buffers);
 280   }
 281
 282   /* put the buffer on the tail of the list */
 283   queue->queue = g_slist_append (queue->queue, buf);
 284 //  STATUS("%s: +\n");
 285   GST_DEBUG (0,"(%s:%s)+ ",GST_DEBUG_PAD_NAME(pad));
 286
 287   /* if we were empty, but aren't any more, signal a condition */
 288   tosignal = (queue->level_buffers >= 0);
 289   queue->level_buffers++;
 290
 291   /* we can unlock now */
 292   GST_DEBUG (0,"queue: %s chain %d end signal(%d,%p)\n", name, queue->level_buffers, tosignal, queue->emptycond);
 293
 294   if (tosignal) {
 295 //    STATUS("%s: >\n");
 296     g_cond_signal (queue->emptycond);
 297 //    STATUS("%s: >>\n");
 298   }
 299   GST_UNLOCK (queue);
 300 }
 301
 302 static GstBuffer *
 303 gst_queue_get (GstPad *pad)
 304 {
 305   GstQueue *queue;
 306   GstBuffer *buf = NULL;
 307   GSList *front;
 308   const guchar *name;
 309
 310   g_return_val_if_fail (pad != NULL, NULL);
 311   g_return_val_if_fail (GST_IS_PAD (pad), NULL);
 312
 313   queue = GST_QUEUE (GST_OBJECT_PARENT (pad));
 314   name = GST_ELEMENT_NAME (queue);
 315
 316   /* have to lock for thread-safety */
 317   GST_DEBUG (0,"queue: %s try have queue lock\n", name);
 318   GST_LOCK (queue);
 319   GST_DEBUG (0,"queue: %s push %d %ld %p\n", name, queue->level_buffers, pthread_self (), queue->emptycond);
 320   GST_DEBUG (0,"queue: %s have queue lock\n", name);
 321
 322   // we bail if there's nothing there
 323   if (!queue->level_buffers && !queue->block) {
 324     GST_UNLOCK(queue);
 325     return NULL;
 326   }
 327
 328   while (!queue->level_buffers) {
 329     STATUS("queue: %s U released lock\n");
 330     //g_cond_timed_wait (queue->emptycond, queue->emptylock, queue->timeval);
 331     if (GST_FLAG_IS_SET (queue->sinkpad, GST_PAD_EOS)) {
 332       gst_pad_set_eos (queue->srcpad);
 333       return NULL;
 334     }
 335     //FIXME need to signal other thread in case signals got lost?
 336     g_cond_signal (queue->fullcond);
 337     g_cond_wait (queue->emptycond, GST_OBJECT(queue)->lock);
 338 //    STATUS("queue: %s U- getting lock\n");
 339   }
 340
 341   front = queue->queue;
 342   buf = (GstBuffer *)(front->data);
 343   GST_DEBUG (0,"retrieved buffer %p from queue\n",buf);
 344   queue->queue = g_slist_remove_link (queue->queue, front);
 345   g_slist_free (front);
 346
 347   queue->level_buffers--;
 348 //  STATUS("%s: -\n");
 349   GST_DEBUG (0,"(%s:%s)- ",GST_DEBUG_PAD_NAME(pad));
 350
 351   if (queue->level_buffers < queue->max_buffers) {
 352 //    STATUS("%s: < \n");
 353     g_cond_signal (queue->fullcond);
 354 //    STATUS("%s: << \n");
 355   }
 356   GST_UNLOCK(queue);
 357
 358 //  GST_DEBUG (0,"queue: %s pushing %d %p \n", name, queue->level_buffers, buf);
 359 //  gst_pad_push (queue->srcpad, buf);
 360 //  GST_DEBUG (0,"queue: %s pushing %d done \n", name, queue->level_buffers);
 361
 362   return buf;
 363   /* unlock now */
 364 }
 365
 366 static GstElementStateReturn
 367 gst_queue_change_state (GstElement *element)
 368 {
 369   GstQueue *queue;
 370   g_return_val_if_fail (GST_IS_QUEUE (element), GST_STATE_FAILURE);
 371
 372   queue = GST_QUEUE (element);
 373   GST_DEBUG (0,"gstqueue: state pending %d\n", GST_STATE_PENDING (element));
 374
 375   /* if going down into NULL state, clear out buffers*/
 376   if (GST_STATE_PENDING (element) == GST_STATE_READY) {
 377     /* otherwise (READY or higher) we need to open the file */
 378     gst_queue_flush (queue);
 379   }
 380
 381   /* if we haven't failed already, give the parent class a chance to ;-) */
 382   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 383     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 384
 385   return GST_STATE_SUCCESS;
 386 }
 387
 388
 389 static void
 390 gst_queue_set_arg (GtkObject *object, GtkArg *arg, guint id)
 391 {
 392   GstQueue *queue;
 393
 394   /* it's not null if we got it, but it might not be ours */
 395   g_return_if_fail (GST_IS_QUEUE (object));
 396
 397   queue = GST_QUEUE (object);
 398
 399   switch(id) {
 400     case ARG_MAX_LEVEL:
 401       queue->max_buffers = GTK_VALUE_INT (*arg);
 402       break;
 403     case ARG_BLOCK:
 404       queue->block = GTK_VALUE_BOOL (*arg);
 405       break;
 406     default:
 407       break;
 408   }
 409 }
 410
 411 static void
 412 gst_queue_get_arg (GtkObject *object, GtkArg *arg, guint id)
 413 {
 414   GstQueue *queue;
 415
 416   /* it's not null if we got it, but it might not be ours */
 417   g_return_if_fail (GST_IS_QUEUE (object));
 418
 419   queue = GST_QUEUE (object);
 420
 421   switch (id) {
 422     case ARG_LEVEL:
 423       GTK_VALUE_INT (*arg) = queue->level_buffers;
 424       break;
 425     case ARG_MAX_LEVEL:
 426       GTK_VALUE_INT (*arg) = queue->max_buffers;
 427       break;
 428     case ARG_BLOCK:
 429       GTK_VALUE_BOOL (*arg) = queue->block;
 430       break;
 431     default:
 432       arg->type = GTK_TYPE_INVALID;
 433       break;
 434   }
 435 }