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