gst/elements/gstqueue.c

   1 /* Gnome-Streamer
   2  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, USA.
  18  */
  19
  20 //#define DEBUG_ENABLED
  21 //#define STATUS_ENABLED
  22 #ifdef STATUS_ENABLED
  23 #define STATUS(A) DEBUG(A, gst_element_get_name(GST_ELEMENT(queue)))
  24 #else
  25 #define STATUS(A)
  26 #endif
  27
  28 #include <gstqueue.h>
  29
  30 #include <gst/gstarch.h>
  31
  32 GstElementDetails gst_queue_details = {
  33   "Queue",
  34   "Connection",
  35   "Simple data queue",
  36   VERSION,
  37   "Erik Walthinsen <omega@cse.ogi.edu>",
  38   "(C) 1999",
  39 };
  40
  41
  42 /* Queue signals and args */
  43 enum {
  44   /* FILL ME */
  45   LAST_SIGNAL
  46 };
  47
  48 enum {
  49   ARG_0,
  50   ARG_LEVEL,
  51   ARG_MAX_LEVEL,
  52 };
  53
  54
  55 static void                     gst_queue_class_init    (GstQueueClass *klass);
  56 static void                     gst_queue_init          (GstQueue *queue);
  57
  58 static void                     gst_queue_set_arg       (GtkObject *object, GtkArg *arg, guint id);
  59 static void                     gst_queue_get_arg       (GtkObject *object, GtkArg *arg, guint id);
  60
  61 static void                     gst_queue_chain         (GstPad *pad, GstBuffer *buf);
  62 static GstBuffer *              gst_queue_get           (GstPad *pad);
  63
  64 static void                     gst_queue_flush         (GstQueue *queue);
  65
  66 static GstElementStateReturn    gst_queue_change_state  (GstElement *element);
  67
  68
  69 static GstConnectionClass *parent_class = NULL;
  70 //static guint gst_queue_signals[LAST_SIGNAL] = { 0 };
  71
  72 GtkType
  73 gst_queue_get_type(void) {
  74   static GtkType queue_type = 0;
  75
  76   if (!queue_type) {
  77     static const GtkTypeInfo queue_info = {
  78       "GstQueue",
  79       sizeof(GstQueue),
  80       sizeof(GstQueueClass),
  81       (GtkClassInitFunc)gst_queue_class_init,
  82       (GtkObjectInitFunc)gst_queue_init,
  83       (GtkArgSetFunc)gst_queue_set_arg,
  84       (GtkArgGetFunc)gst_queue_get_arg,
  85       (GtkClassInitFunc)NULL,
  86     };
  87     queue_type = gtk_type_unique (GST_TYPE_CONNECTION, &queue_info);
  88   }
  89   return queue_type;
  90 }
  91
  92 static void
  93 gst_queue_class_init (GstQueueClass *klass)
  94 {
  95   GtkObjectClass *gtkobject_class;
  96   GstElementClass *gstelement_class;
  97   GstConnectionClass *gstconnection_class;
  98
  99   gtkobject_class = (GtkObjectClass*)klass;
 100   gstelement_class = (GstElementClass*)klass;
 101   gstconnection_class = (GstConnectionClass*)klass;
 102
 103   parent_class = gtk_type_class (GST_TYPE_CONNECTION);
 104
 105   gtk_object_add_arg_type ("GstQueue::level", GTK_TYPE_INT,
 106                            GTK_ARG_READABLE, ARG_LEVEL);
 107   gtk_object_add_arg_type ("GstQueue::max_level", GTK_TYPE_INT,
 108                            GTK_ARG_READWRITE, ARG_MAX_LEVEL);
 109
 110   gtkobject_class->set_arg = gst_queue_set_arg;
 111   gtkobject_class->get_arg = gst_queue_get_arg;
 112
 113   gstelement_class->change_state = gst_queue_change_state;
 114 }
 115
 116 static void
 117 gst_queue_init (GstQueue *queue)
 118 {
 119   GST_FLAG_SET (queue, GST_ELEMENT_SCHEDULE_PASSIVELY);
 120
 121   queue->sinkpad = gst_pad_new ("sink", GST_PAD_SINK);
 122   gst_pad_set_chain_function (queue->sinkpad, GST_DEBUG_FUNCPTR(gst_queue_chain));
 123   gst_element_add_pad (GST_ELEMENT (queue), queue->sinkpad);
 124
 125   queue->srcpad = gst_pad_new ("src", GST_PAD_SRC);
 126   gst_pad_set_get_function (queue->srcpad, GST_DEBUG_FUNCPTR(gst_queue_get));
 127   gst_element_add_pad (GST_ELEMENT (queue), queue->srcpad);
 128
 129   queue->queue = NULL;
 130   queue->level_buffers = 0;
 131   queue->max_buffers = 20;
 132   queue->level_bytes = 0;
 133   queue->size_buffers = 0;
 134   queue->size_bytes = 0;
 135
 136   queue->emptylock = g_mutex_new ();
 137   queue->emptycond = g_cond_new ();
 138
 139   queue->fulllock = g_mutex_new ();
 140   queue->fullcond = g_cond_new ();
 141 }
 142
 143 static void
 144 gst_queue_cleanup_buffers (gpointer data, const gpointer user_data)
 145 {
 146   DEBUG("queue: %s cleaning buffer %p\n", (gchar *)user_data, data);
 147
 148   gst_buffer_unref (GST_BUFFER (data));
 149 }
 150
 151 static void
 152 gst_queue_flush (GstQueue *queue)
 153 {
 154   g_slist_foreach (queue->queue, gst_queue_cleanup_buffers,
 155                   (char *)gst_element_get_name (GST_ELEMENT (queue)));
 156   g_slist_free (queue->queue);
 157
 158   queue->queue = NULL;
 159   queue->level_buffers = 0;
 160 }
 161
 162 static void
 163 gst_queue_chain (GstPad *pad, GstBuffer *buf)
 164 {
 165   GstQueue *queue;
 166   gboolean tosignal = FALSE;
 167   const guchar *name;
 168
 169   g_return_if_fail (pad != NULL);
 170   g_return_if_fail (GST_IS_PAD (pad));
 171   g_return_if_fail (buf != NULL);
 172
 173   queue = GST_QUEUE (pad->parent);
 174   name = gst_element_get_name (GST_ELEMENT (queue));
 175
 176   /* we have to lock the queue since we span threads */
 177
 178   DEBUG("queue: try have queue lock\n");
 179   GST_LOCK (queue);
 180   DEBUG("queue: %s adding buffer %p %ld\n", name, buf, pthread_self ());
 181   DEBUG("queue: have queue lock\n");
 182
 183   if (GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLUSH)) {
 184     gst_queue_flush (queue);
 185   }
 186
 187   DEBUG("queue: %s: chain %d %p\n", name, queue->level_buffers, buf);
 188
 189   while (queue->level_buffers >= queue->max_buffers) {
 190     DEBUG("queue: %s waiting %d\n", name, queue->level_buffers);
 191     STATUS("%s: O\n");
 192     GST_UNLOCK (queue);
 193     g_mutex_lock (queue->fulllock);
 194     g_cond_wait (queue->fullcond, queue->fulllock);
 195     g_mutex_unlock (queue->fulllock);
 196     GST_LOCK (queue);
 197     STATUS("%s: O+\n");
 198     DEBUG("queue: %s waiting done %d\n", name, queue->level_buffers);
 199   }
 200
 201   /* put the buffer on the tail of the list */
 202   queue->queue = g_slist_append (queue->queue, buf);
 203 //  STATUS("%s: +\n");
 204   g_print("(%s:%s)+ ",GST_DEBUG_PAD_NAME(pad));
 205
 206   /* if we were empty, but aren't any more, signal a condition */
 207   tosignal = (queue->level_buffers >= 0);
 208   queue->level_buffers++;
 209
 210   /* we can unlock now */
 211   DEBUG("queue: %s chain %d end signal(%d,%p)\n", name, queue->level_buffers, tosignal, queue->emptycond);
 212   GST_UNLOCK (queue);
 213
 214   if (tosignal) {
 215     g_mutex_lock (queue->emptylock);
 216 //    STATUS("%s: >\n");
 217     g_cond_signal (queue->emptycond);
 218 //    STATUS("%s: >>\n");
 219     g_mutex_unlock (queue->emptylock);
 220   }
 221 }
 222
 223 static GstBuffer *
 224 gst_queue_get (GstPad *pad)
 225 {
 226   GstQueue *queue = GST_QUEUE (gst_pad_get_parent(pad));
 227   GstBuffer *buf = NULL;
 228   GSList *front;
 229   gboolean tosignal = FALSE;
 230   const guchar *name;
 231
 232   name = gst_element_get_name (GST_ELEMENT (queue));
 233
 234   /* have to lock for thread-safety */
 235   DEBUG("queue: %s try have queue lock\n", name);
 236   GST_LOCK (queue);
 237   DEBUG("queue: %s push %d %ld %p\n", name, queue->level_buffers, pthread_self (), queue->emptycond);
 238   DEBUG("queue: %s have queue lock\n", name);
 239
 240   while (!queue->level_buffers) {
 241     STATUS("queue: %s U released lock\n");
 242     GST_UNLOCK (queue);
 243     g_mutex_lock (queue->emptylock);
 244     g_cond_wait (queue->emptycond, queue->emptylock);
 245     g_mutex_unlock (queue->emptylock);
 246     GST_LOCK (queue);
 247 //    STATUS("queue: %s U- getting lock\n");
 248   }
 249
 250   front = queue->queue;
 251   buf = (GstBuffer *)(front->data);
 252   DEBUG("retrieved buffer %p from queue\n",buf);
 253   queue->queue = g_slist_remove_link (queue->queue, front);
 254   g_slist_free (front);
 255
 256   queue->level_buffers--;
 257 //  STATUS("%s: -\n");
 258   g_print("(%s:%s)- ",GST_DEBUG_PAD_NAME(pad));
 259   tosignal = queue->level_buffers < queue->max_buffers;
 260   GST_UNLOCK(queue);
 261
 262   if (tosignal) {
 263     g_mutex_lock (queue->fulllock);
 264 //    STATUS("%s: < \n");
 265     g_cond_signal (queue->fullcond);
 266 //    STATUS("%s: << \n");
 267     g_mutex_unlock (queue->fulllock);
 268   }
 269
 270 //  DEBUG("queue: %s pushing %d %p \n", name, queue->level_buffers, buf);
 271 //  gst_pad_push (queue->srcpad, buf);
 272 //  DEBUG("queue: %s pushing %d done \n", name, queue->level_buffers);
 273
 274   return buf;
 275   /* unlock now */
 276 }
 277
 278 static GstElementStateReturn
 279 gst_queue_change_state (GstElement *element)
 280 {
 281   GstQueue *queue;
 282   g_return_val_if_fail (GST_IS_QUEUE (element), GST_STATE_FAILURE);
 283
 284   queue = GST_QUEUE (element);
 285   DEBUG("gstqueue: state pending %d\n", GST_STATE_PENDING (element));
 286
 287   /* if going down into NULL state, clear out buffers*/
 288   if (GST_STATE_PENDING (element) == GST_STATE_READY) {
 289     /* otherwise (READY or higher) we need to open the file */
 290     gst_queue_flush (queue);
 291   }
 292
 293   /* if we haven't failed already, give the parent class a chance to ;-) */
 294   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 295     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 296
 297   return GST_STATE_SUCCESS;
 298 }
 299
 300
 301 static void
 302 gst_queue_set_arg (GtkObject *object, GtkArg *arg, guint id)
 303 {
 304   GstQueue *queue;
 305
 306   /* it's not null if we got it, but it might not be ours */
 307   g_return_if_fail (GST_IS_QUEUE (object));
 308
 309   queue = GST_QUEUE (object);
 310
 311   switch(id) {
 312     case ARG_MAX_LEVEL:
 313       queue->max_buffers = GTK_VALUE_INT (*arg);
 314       break;
 315     default:
 316       break;
 317   }
 318 }
 319
 320 static void
 321 gst_queue_get_arg (GtkObject *object, GtkArg *arg, guint id)
 322 {
 323   GstQueue *queue;
 324
 325   /* it's not null if we got it, but it might not be ours */
 326   g_return_if_fail (GST_IS_QUEUE (object));
 327
 328   queue = GST_QUEUE (object);
 329
 330   switch (id) {
 331     case ARG_LEVEL:
 332       GTK_VALUE_INT (*arg) = queue->level_buffers;
 333       break;
 334     case ARG_MAX_LEVEL:
 335       GTK_VALUE_INT (*arg) = queue->max_buffers;
 336       break;
 337     default:
 338       arg->type = GTK_TYPE_INVALID;
 339       break;
 340   }
 341 }