gst/gstbufferpool.c

   1 /* GStreamer
   2  * Copyright (C) 2010 Wim Taymans <wim.taymans@gmail.com>
   3  *
   4  * gstbufferpool.c: GstBufferPool baseclass
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 /**
  23  * SECTION:gstbufferpool
  24  * @short_description: Pool for buffers
  25  * @see_also: #GstBuffer
  26  *
  27  */
  28
  29 #include "gst_private.h"
  30
  31 #include <errno.h>
  32 #ifdef HAVE_UNISTD_H
  33 #  include <unistd.h>
  34 #endif
  35 #include <sys/types.h>
  36
  37 #include "gstinfo.h"
  38
  39 #include "gstbufferpool.h"
  40
  41 enum
  42 {
  43   /* add more above */
  44   LAST_SIGNAL
  45 };
  46
  47 static void gst_buffer_pool_finalize (GObject * object);
  48
  49 G_DEFINE_TYPE (GstBufferPool, gst_buffer_pool, GST_TYPE_OBJECT);
  50
  51 static void default_set_flushing (GstBufferPool * pool, gboolean flushing);
  52 static gboolean default_set_config (GstBufferPool * pool,
  53     GstBufferPoolConfig * config);
  54 static GstFlowReturn default_alloc_buffer (GstBufferPool * pool,
  55     GstBuffer ** buffer, GstBufferPoolConfig * config,
  56     GstBufferPoolParams * params);
  57 static GstFlowReturn default_acquire_buffer (GstBufferPool * pool,
  58     GstBuffer ** buffer, GstBufferPoolParams * params);
  59 static void default_free_buffer (GstBufferPool * pool, GstBuffer * buffer);
  60 static void default_release_buffer (GstBufferPool * pool, GstBuffer * buffer);
  61
  62 static void
  63 gst_buffer_pool_class_init (GstBufferPoolClass * klass)
  64 {
  65   GObjectClass *gobject_class = (GObjectClass *) klass;
  66
  67   gobject_class->finalize = gst_buffer_pool_finalize;
  68
  69   klass->set_flushing = default_set_flushing;
  70   klass->set_config = default_set_config;
  71   klass->acquire_buffer = default_acquire_buffer;
  72   klass->alloc_buffer = default_alloc_buffer;
  73   klass->release_buffer = default_release_buffer;
  74   klass->free_buffer = default_free_buffer;
  75 }
  76
  77 static void
  78 gst_buffer_pool_init (GstBufferPool * pool)
  79 {
  80   pool->config.align = 1;
  81   pool->poll = gst_poll_new_timer ();
  82   pool->queue = gst_atomic_queue_new (10);
  83   default_set_flushing (pool, TRUE);
  84
  85   GST_DEBUG_OBJECT (pool, "created");
  86 }
  87
  88 static void
  89 gst_buffer_pool_finalize (GObject * object)
  90 {
  91   G_OBJECT_CLASS (gst_buffer_pool_parent_class)->finalize (object);
  92 }
  93
  94 /**
  95  * gst_buffer_pool_new:
  96  *
  97  * Creates a new #GstBufferPool instance.
  98  *
  99  * Returns: a new #GstBufferPool instance
 100  */
 101 GstBufferPool *
 102 gst_buffer_pool_new (void)
 103 {
 104   GstBufferPool *result;
 105
 106   result = g_object_newv (GST_TYPE_BUFFER_POOL, 0, NULL);
 107   GST_DEBUG_OBJECT (result, "created new buffer pool");
 108
 109   return result;
 110 }
 111
 112 static void
 113 flush_buffers (GstBufferPool * pool)
 114 {
 115   GstBuffer *buffer;
 116   GstBufferPoolClass *pclass;
 117
 118   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 119
 120   while ((buffer = gst_atomic_queue_pop (pool->queue))) {
 121     gst_poll_read_control (pool->poll);
 122     if (G_LIKELY (pclass->free_buffer))
 123       pclass->free_buffer (pool, buffer);
 124   }
 125 }
 126
 127 static void
 128 default_set_flushing (GstBufferPool * pool, gboolean flushing)
 129 {
 130   g_atomic_int_set (&pool->flushing, flushing);
 131
 132   if (flushing) {
 133     /* write the control socket so that waiters get woken up and can check the
 134      * flushing flag we set above */
 135     gst_poll_write_control (pool->poll);
 136     flush_buffers (pool);
 137   } else {
 138     gst_poll_read_control (pool->poll);
 139   }
 140 }
 141
 142 /**
 143  * gst_buffer_pool_set_flushing:
 144  * @pool: a #GstBufferPool
 145  * @flushing: the new flushing state
 146  *
 147  * Control the flushing state of @pool. When the pool is flushing, new calls to
 148  * gst_buffer_pool_acquire_buffer() will return with GST_FLOW_WRONG_STATE.
 149  */
 150 void
 151 gst_buffer_pool_set_flushing (GstBufferPool * pool, gboolean flushing)
 152 {
 153   GstBufferPoolClass *pclass;
 154
 155   g_return_if_fail (GST_IS_BUFFER_POOL (pool));
 156
 157   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 158
 159   if (G_LIKELY (pclass->set_flushing))
 160     pclass->set_flushing (pool, flushing);
 161 }
 162
 163 static gboolean
 164 default_set_config (GstBufferPool * pool, GstBufferPoolConfig * config)
 165 {
 166   guint i;
 167   GstBufferPoolClass *pclass;
 168
 169   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 170
 171   /* we need to prealloc buffers */
 172   for (i = config->min_buffers; i > 0; i--) {
 173     GstBuffer *buffer;
 174
 175     if (G_LIKELY (pclass->alloc_buffer)) {
 176       if (!pclass->alloc_buffer (pool, &buffer, config, NULL))
 177         return FALSE;
 178     } else
 179       return FALSE;
 180
 181     /* store in the queue */
 182     gst_atomic_queue_push (pool->queue, buffer);
 183     gst_poll_write_control (pool->poll);
 184   }
 185
 186   return TRUE;
 187 }
 188
 189 /**
 190  * gst_buffer_pool_set_config:
 191  * @pool: a #GstBufferPool
 192  * @config: a #GstBufferPoolConfig
 193  *
 194  * Set the configuration of the pool. The pool must be flushing or else this
 195  * function will do nothing and return FALSE.
 196  *
 197  * Returns: TRUE when the configuration could be set.
 198  */
 199 gboolean
 200 gst_buffer_pool_set_config (GstBufferPool * pool, GstBufferPoolConfig * config)
 201 {
 202   gboolean result;
 203   GstBufferPoolClass *pclass;
 204
 205   g_return_val_if_fail (GST_IS_BUFFER_POOL (pool), FALSE);
 206   g_return_val_if_fail (config != NULL, FALSE);
 207
 208   if (!g_atomic_int_get (&pool->flushing))
 209     return FALSE;
 210
 211   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 212
 213   /* free the buffer when we are flushing */
 214   if (G_LIKELY (pclass->set_config))
 215     result = pclass->set_config (pool, config);
 216   else
 217     result = FALSE;
 218
 219   if (result)
 220     pool->config = *config;
 221
 222   return result;
 223 }
 224
 225 /**
 226  * gst_buffer_pool_get_config:
 227  * @pool: a #GstBufferPool
 228  * @config: a #GstBufferPoolConfig
 229  *
 230  * Get the current configuration of the pool.
 231  */
 232 void
 233 gst_buffer_pool_get_config (GstBufferPool * pool, GstBufferPoolConfig * config)
 234 {
 235   *config = pool->config;
 236 }
 237
 238 static GstFlowReturn
 239 default_alloc_buffer (GstBufferPool * pool, GstBuffer ** buffer,
 240     GstBufferPoolConfig * config, GstBufferPoolParams * params)
 241 {
 242   guint size, align;
 243
 244   *buffer = gst_buffer_new ();
 245
 246   align = config->align - 1;
 247   size = config->prefix + config->postfix + config->size + align;
 248   if (size > 0) {
 249     guint8 *memptr;
 250
 251     memptr = g_malloc (size);
 252     GST_BUFFER_MALLOCDATA (*buffer) = memptr;
 253     memptr = (guint8 *) ((guintptr) (memptr + align) & ~align);
 254     GST_BUFFER_DATA (*buffer) = memptr + config->prefix;
 255     GST_BUFFER_SIZE (*buffer) = config->size;
 256   }
 257
 258   return GST_FLOW_OK;
 259 }
 260
 261 static GstFlowReturn
 262 default_acquire_buffer (GstBufferPool * pool, GstBuffer ** buffer,
 263     GstBufferPoolParams * params)
 264 {
 265   GstFlowReturn result;
 266   GstBufferPoolClass *pclass;
 267
 268   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 269
 270   while (TRUE) {
 271     if (g_atomic_int_get (&pool->flushing))
 272       return GST_FLOW_WRONG_STATE;
 273
 274     /* try to get a buffer from the queue */
 275     *buffer = gst_atomic_queue_pop (pool->queue);
 276     if (*buffer) {
 277       /* FIXME check the size of the buffer */
 278       gst_poll_read_control (pool->poll);
 279       result = GST_FLOW_OK;
 280       break;
 281     }
 282
 283     /* no buffer */
 284     if (pool->config.max_buffers == 0) {
 285       /* no max_buffers, we allocate some more */
 286       if (G_LIKELY (pclass->alloc_buffer))
 287         result = pclass->alloc_buffer (pool, buffer, &pool->config, params);
 288       else
 289         result = GST_FLOW_NOT_SUPPORTED;
 290       break;
 291     }
 292
 293     /* check if we need to wait */
 294     if (!(params->flags & GST_BUFFER_POOL_FLAG_WAIT)) {
 295       result = GST_FLOW_UNEXPECTED;
 296       break;
 297     }
 298
 299     /* now wait */
 300     gst_poll_wait (pool->poll, GST_CLOCK_TIME_NONE);
 301   }
 302   return result;
 303 }
 304
 305 /**
 306  * gst_buffer_pool_acquire_buffer:
 307  * @pool: a #GstBufferPool
 308  * @buffer: a location for a #GstBuffer
 309  * @params: parameters.
 310  *
 311  * Acquire a buffer from @pool. @buffer should point to a memory location that
 312  * can hold a pointer to the new buffer.
 313  *
 314  * @params can be NULL or contain optional parameters to influence the allocation.
 315  *
 316  * Returns: a #GstFlowReturn such as GST_FLOW_WRONG_STATE when the pool is
 317  * flushing.
 318  */
 319 GstFlowReturn
 320 gst_buffer_pool_acquire_buffer (GstBufferPool * pool, GstBuffer ** buffer,
 321     GstBufferPoolParams * params)
 322 {
 323   GstBufferPoolClass *pclass;
 324   GstFlowReturn result;
 325
 326   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 327
 328   if (G_LIKELY (pclass->acquire_buffer))
 329     result = pclass->acquire_buffer (pool, buffer, params);
 330   else
 331     result = GST_FLOW_NOT_SUPPORTED;
 332
 333   return result;
 334 }
 335
 336 static void
 337 default_free_buffer (GstBufferPool * pool, GstBuffer * buffer)
 338 {
 339   gst_buffer_unref (buffer);
 340 }
 341
 342 static void
 343 default_release_buffer (GstBufferPool * pool, GstBuffer * buffer)
 344 {
 345   /* keep it around in our queue, we might be flushing but that's ok because we
 346    * handle that unlikely case below. */
 347   gst_atomic_queue_push (pool->queue, buffer);
 348   gst_poll_write_control (pool->poll);
 349
 350   if (G_UNLIKELY (g_atomic_int_get (&pool->flushing))) {
 351     /* we are flushing, remove the buffers again */
 352     flush_buffers (pool);
 353   }
 354 }
 355
 356 /**
 357  * gst_buffer_pool_release_buffer:
 358  * @pool: a #GstBufferPool
 359  * @buffer: a #GstBuffer
 360  *
 361  * Release @buffer to @pool. @buffer should have previously been allocated from
 362  * @pool with gst_buffer_pool_acquire_buffer().
 363  *
 364  * This function is usually called automatically when the last ref on @buffer
 365  * disappears.
 366  */
 367 void
 368 gst_buffer_pool_release_buffer (GstBufferPool * pool, GstBuffer * buffer)
 369 {
 370   GstBufferPoolClass *pclass;
 371
 372   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 373
 374   if (G_LIKELY (pclass->release_buffer))
 375     pclass->release_buffer (pool, buffer);
 376 }