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   ARG_TIMEOUT,
  60 };
  61
  62
  63 static void                     gst_queue_class_init    (GstQueueClass *klass);
  64 static void                     gst_queue_init          (GstQueue *queue);
  65
  66 static void                     gst_queue_set_arg       (GtkObject *object, GtkArg *arg, guint id);
  67 static void                     gst_queue_get_arg       (GtkObject *object, GtkArg *arg, guint id);
  68
  69 static void                     gst_queue_chain         (GstPad *pad, GstBuffer *buf);
  70 static GstBuffer *              gst_queue_get           (GstPad *pad);
  71
  72 static void                     gst_queue_flush         (GstQueue *queue);
  73
  74 static GstElementStateReturn    gst_queue_change_state  (GstElement *element);
  75
  76
  77 static GstElementClass *parent_class = NULL;
  78 //static guint gst_queue_signals[LAST_SIGNAL] = { 0 };
  79
  80 GtkType
  81 gst_queue_get_type(void) {
  82   static GtkType queue_type = 0;
  83
  84   if (!queue_type) {
  85     static const GtkTypeInfo queue_info = {
  86       "GstQueue",
  87       sizeof(GstQueue),
  88       sizeof(GstQueueClass),
  89       (GtkClassInitFunc)gst_queue_class_init,
  90       (GtkObjectInitFunc)gst_queue_init,
  91       (GtkArgSetFunc)gst_queue_set_arg,
  92       (GtkArgGetFunc)gst_queue_get_arg,
  93       (GtkClassInitFunc)NULL,
  94     };
  95     queue_type = gtk_type_unique (GST_TYPE_ELEMENT, &queue_info);
  96   }
  97   return queue_type;
  98 }
  99
 100 static void
 101 gst_queue_class_init (GstQueueClass *klass)
 102 {
 103   GtkObjectClass *gtkobject_class;
 104   GstElementClass *gstelement_class;
 105
 106   gtkobject_class = (GtkObjectClass*)klass;
 107   gstelement_class = (GstElementClass*)klass;
 108
 109   parent_class = gtk_type_class (GST_TYPE_ELEMENT);
 110
 111   gtk_object_add_arg_type ("GstQueue::level", GTK_TYPE_INT,
 112                            GTK_ARG_READABLE, ARG_LEVEL);
 113   gtk_object_add_arg_type ("GstQueue::max_level", GTK_TYPE_INT,
 114                            GTK_ARG_READWRITE, ARG_MAX_LEVEL);
 115   gtk_object_add_arg_type ("GstQueue::block", GTK_TYPE_BOOL,
 116                            GTK_ARG_READWRITE, ARG_BLOCK);
 117   gtk_object_add_arg_type ("GstQueue::timeout", GTK_TYPE_INT,
 118                            GTK_ARG_READWRITE, ARG_TIMEOUT);
 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
 136   queue->srcpad = gst_pad_new ("src", GST_PAD_SRC);
 137   gst_pad_set_get_function (queue->srcpad, GST_DEBUG_FUNCPTR(gst_queue_get));
 138   gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
 139
 140   queue->queue = NULL;
 141   queue->level_buffers = 0;
 142   queue->max_buffers = 100;
 143   queue->block = TRUE;
 144   queue->level_bytes = 0;
 145   queue->size_buffers = 0;
 146   queue->size_bytes = 0;
 147
 148   queue->emptycond = g_cond_new ();
 149   queue->fullcond = g_cond_new ();
 150 }
 151
 152 static void
 153 gst_queue_cleanup_buffers (gpointer data, const gpointer user_data)
 154 {
 155   GST_DEBUG (0,"queue: %s cleaning buffer %p\n", (gchar *)user_data, data);
 156
 157   gst_buffer_unref (GST_BUFFER (data));
 158 }
 159
 160 static void
 161 gst_queue_flush (GstQueue *queue)
 162 {
 163   g_slist_foreach (queue->queue, gst_queue_cleanup_buffers,
 164                   (char *)gst_element_get_name (GST_ELEMENT (queue)));
 165   g_slist_free (queue->queue);
 166
 167   queue->queue = NULL;
 168   queue->level_buffers = 0;
 169   queue->timeval = NULL;
 170 }
 171
 172 static void
 173 gst_queue_chain (GstPad *pad, GstBuffer *buf)
 174 {
 175   GstQueue *queue;
 176   gboolean tosignal = FALSE;
 177   const guchar *name;
 178
 179   g_return_if_fail (pad != NULL);
 180   g_return_if_fail (GST_IS_PAD (pad));
 181   g_return_if_fail (buf != NULL);
 182
 183   queue = GST_QUEUE (pad->parent);
 184   name = gst_element_get_name (GST_ELEMENT (queue));
 185
 186   /* we have to lock the queue since we span threads */
 187
 188   GST_DEBUG (0,"queue: try have queue lock\n");
 189   GST_LOCK (queue);
 190   GST_DEBUG (0,"queue: %s adding buffer %p %ld\n", name, buf, pthread_self ());
 191   GST_DEBUG (0,"queue: have queue lock\n");
 192
 193   if (GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLUSH)) {
 194     gst_queue_flush (queue);
 195   }
 196
 197   GST_DEBUG (0,"queue: %s: chain %d %p\n", name, queue->level_buffers, buf);
 198
 199   while (queue->level_buffers >= queue->max_buffers) {
 200     GST_DEBUG (0,"queue: %s waiting %d\n", name, queue->level_buffers);
 201     STATUS("%s: O\n");
 202     //g_cond_timed_wait (queue->fullcond, queue->fulllock, queue->timeval);
 203     //FIXME need to signal other thread in case signals got lost?
 204     g_cond_signal (queue->emptycond);
 205     g_cond_wait (queue->fullcond, GST_OBJECT(queue)->lock);
 206     STATUS("%s: O+\n");
 207     GST_DEBUG (0,"queue: %s waiting done %d\n", name, queue->level_buffers);
 208   }
 209
 210   /* put the buffer on the tail of the list */
 211   queue->queue = g_slist_append (queue->queue, buf);
 212 //  STATUS("%s: +\n");
 213   GST_DEBUG (0,"(%s:%s)+ ",GST_DEBUG_PAD_NAME(pad));
 214
 215   /* if we were empty, but aren't any more, signal a condition */
 216   tosignal = (queue->level_buffers >= 0);
 217   queue->level_buffers++;
 218
 219   /* we can unlock now */
 220   GST_DEBUG (0,"queue: %s chain %d end signal(%d,%p)\n", name, queue->level_buffers, tosignal, queue->emptycond);
 221
 222   if (tosignal) {
 223 //    STATUS("%s: >\n");
 224     g_cond_signal (queue->emptycond);
 225 //    STATUS("%s: >>\n");
 226   }
 227   GST_UNLOCK (queue);
 228 }
 229
 230 static GstBuffer *
 231 gst_queue_get (GstPad *pad)
 232 {
 233   GstQueue *queue = GST_QUEUE (gst_pad_get_parent(pad));
 234   GstBuffer *buf = NULL;
 235   GSList *front;
 236   const guchar *name;
 237
 238   name = gst_element_get_name (GST_ELEMENT (queue));
 239
 240   /* have to lock for thread-safety */
 241   GST_DEBUG (0,"queue: %s try have queue lock\n", name);
 242   GST_LOCK (queue);
 243   GST_DEBUG (0,"queue: %s push %d %ld %p\n", name, queue->level_buffers, pthread_self (), queue->emptycond);
 244   GST_DEBUG (0,"queue: %s have queue lock\n", name);
 245
 246   // we bail if there's nothing there
 247   if (!queue->level_buffers && !queue->block) {
 248     GST_UNLOCK(queue);
 249     return NULL;
 250   }
 251
 252   while (!queue->level_buffers) {
 253     STATUS("queue: %s U released lock\n");
 254     //g_cond_timed_wait (queue->emptycond, queue->emptylock, queue->timeval);
 255     //FIXME need to signal other thread in case signals got lost?
 256     g_cond_signal (queue->fullcond);
 257     g_cond_wait (queue->emptycond, GST_OBJECT(queue)->lock);
 258 //    STATUS("queue: %s U- getting lock\n");
 259   }
 260
 261   front = queue->queue;
 262   buf = (GstBuffer *)(front->data);
 263   GST_DEBUG (0,"retrieved buffer %p from queue\n",buf);
 264   queue->queue = g_slist_remove_link (queue->queue, front);
 265   g_slist_free (front);
 266
 267   queue->level_buffers--;
 268 //  STATUS("%s: -\n");
 269   GST_DEBUG (0,"(%s:%s)- ",GST_DEBUG_PAD_NAME(pad));
 270
 271   if (queue->level_buffers < queue->max_buffers) {
 272 //    STATUS("%s: < \n");
 273     g_cond_signal (queue->fullcond);
 274 //    STATUS("%s: << \n");
 275   }
 276   GST_UNLOCK(queue);
 277
 278 //  GST_DEBUG (0,"queue: %s pushing %d %p \n", name, queue->level_buffers, buf);
 279 //  gst_pad_push (queue->srcpad, buf);
 280 //  GST_DEBUG (0,"queue: %s pushing %d done \n", name, queue->level_buffers);
 281
 282   return buf;
 283   /* unlock now */
 284 }
 285
 286 static GstElementStateReturn
 287 gst_queue_change_state (GstElement *element)
 288 {
 289   GstQueue *queue;
 290   g_return_val_if_fail (GST_IS_QUEUE (element), GST_STATE_FAILURE);
 291
 292   queue = GST_QUEUE (element);
 293   GST_DEBUG (0,"gstqueue: state pending %d\n", GST_STATE_PENDING (element));
 294
 295   /* if going down into NULL state, clear out buffers*/
 296   if (GST_STATE_PENDING (element) == GST_STATE_READY) {
 297     /* otherwise (READY or higher) we need to open the file */
 298     gst_queue_flush (queue);
 299   }
 300
 301   /* if we haven't failed already, give the parent class a chance to ;-) */
 302   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 303     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 304
 305   return GST_STATE_SUCCESS;
 306 }
 307
 308
 309 static void
 310 gst_queue_set_arg (GtkObject *object, GtkArg *arg, guint id)
 311 {
 312   GstQueue *queue;
 313
 314   /* it's not null if we got it, but it might not be ours */
 315   g_return_if_fail (GST_IS_QUEUE (object));
 316
 317   queue = GST_QUEUE (object);
 318
 319   switch(id) {
 320     case ARG_MAX_LEVEL:
 321       queue->max_buffers = GTK_VALUE_INT (*arg);
 322       break;
 323     case ARG_BLOCK:
 324       queue->block = GTK_VALUE_BOOL (*arg);
 325       break;
 326     case ARG_TIMEOUT:
 327       break;
 328     default:
 329       break;
 330   }
 331 }
 332
 333 static void
 334 gst_queue_get_arg (GtkObject *object, GtkArg *arg, guint id)
 335 {
 336   GstQueue *queue;
 337
 338   /* it's not null if we got it, but it might not be ours */
 339   g_return_if_fail (GST_IS_QUEUE (object));
 340
 341   queue = GST_QUEUE (object);
 342
 343   switch (id) {
 344     case ARG_LEVEL:
 345       GTK_VALUE_INT (*arg) = queue->level_buffers;
 346       break;
 347     case ARG_MAX_LEVEL:
 348       GTK_VALUE_INT (*arg) = queue->max_buffers;
 349       break;
 350     case ARG_BLOCK:
 351       GTK_VALUE_BOOL (*arg) = queue->block;
 352       break;
 353     case ARG_TIMEOUT:
 354       break;
 355     default:
 356       arg->type = GTK_TYPE_INVALID;
 357       break;
 358   }
 359 }