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 #include "gstquark.h"
  39
  40 #include "gstbufferpool.h"
  41
  42
  43 #define GST_BUFFER_POOL_GET_PRIVATE(obj)  \
  44    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_BUFFER_POOL, GstBufferPoolPrivate))
  45
  46 struct _GstBufferPoolPrivate
  47 {
  48   guint min_buffers;
  49   guint max_buffers;
  50   guint size;
  51   guint prefix;
  52   guint postfix;
  53   guint align;
  54 };
  55
  56 enum
  57 {
  58   /* add more above */
  59   LAST_SIGNAL
  60 };
  61
  62 static void gst_buffer_pool_finalize (GObject * object);
  63
  64 G_DEFINE_TYPE (GstBufferPool, gst_buffer_pool, GST_TYPE_OBJECT);
  65
  66 static void default_set_flushing (GstBufferPool * pool, gboolean flushing);
  67 static gboolean default_set_config (GstBufferPool * pool,
  68     GstStructure * config);
  69 static GstFlowReturn default_alloc_buffer (GstBufferPool * pool,
  70     GstBuffer ** buffer, GstBufferPoolParams * params);
  71 static GstFlowReturn default_acquire_buffer (GstBufferPool * pool,
  72     GstBuffer ** buffer, GstBufferPoolParams * params);
  73 static void default_free_buffer (GstBufferPool * pool, GstBuffer * buffer);
  74 static void default_release_buffer (GstBufferPool * pool, GstBuffer * buffer);
  75
  76 static void
  77 gst_buffer_pool_class_init (GstBufferPoolClass * klass)
  78 {
  79   GObjectClass *gobject_class = (GObjectClass *) klass;
  80
  81   gobject_class->finalize = gst_buffer_pool_finalize;
  82
  83   klass->set_flushing = default_set_flushing;
  84   klass->set_config = default_set_config;
  85   klass->acquire_buffer = default_acquire_buffer;
  86   klass->alloc_buffer = default_alloc_buffer;
  87   klass->release_buffer = default_release_buffer;
  88   klass->free_buffer = default_free_buffer;
  89 }
  90
  91 static void
  92 gst_buffer_pool_init (GstBufferPool * pool)
  93 {
  94   pool->priv = GST_BUFFER_POOL_GET_PRIVATE (pool);
  95
  96   pool->config = gst_structure_id_new (GST_QUARK (BUFFER_POOL_CONFIG),
  97       GST_QUARK (SIZE), G_TYPE_UINT, 0,
  98       GST_QUARK (MIN_BUFFERS), G_TYPE_UINT, 0,
  99       GST_QUARK (MAX_BUFFERS), G_TYPE_UINT, 0,
 100       GST_QUARK (PREFIX), G_TYPE_UINT, 0,
 101       GST_QUARK (POSTFIX), G_TYPE_UINT, 0,
 102       GST_QUARK (ALIGN), G_TYPE_UINT, 1, NULL);
 103   pool->poll = gst_poll_new_timer ();
 104   pool->queue = gst_atomic_queue_new (10);
 105   default_set_flushing (pool, TRUE);
 106
 107   GST_DEBUG_OBJECT (pool, "created");
 108 }
 109
 110 static void
 111 gst_buffer_pool_finalize (GObject * object)
 112 {
 113   G_OBJECT_CLASS (gst_buffer_pool_parent_class)->finalize (object);
 114 }
 115
 116 /**
 117  * gst_buffer_pool_new:
 118  *
 119  * Creates a new #GstBufferPool instance.
 120  *
 121  * Returns: a new #GstBufferPool instance
 122  */
 123 GstBufferPool *
 124 gst_buffer_pool_new (void)
 125 {
 126   GstBufferPool *result;
 127
 128   result = g_object_newv (GST_TYPE_BUFFER_POOL, 0, NULL);
 129   GST_DEBUG_OBJECT (result, "created new buffer pool");
 130
 131   return result;
 132 }
 133
 134 static void
 135 flush_buffers (GstBufferPool * pool)
 136 {
 137   GstBuffer *buffer;
 138   GstBufferPoolClass *pclass;
 139
 140   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 141
 142   while ((buffer = gst_atomic_queue_pop (pool->queue))) {
 143     gst_poll_read_control (pool->poll);
 144     if (G_LIKELY (pclass->free_buffer))
 145       pclass->free_buffer (pool, buffer);
 146   }
 147 }
 148
 149 static void
 150 default_set_flushing (GstBufferPool * pool, gboolean flushing)
 151 {
 152   g_atomic_int_set (&pool->flushing, flushing);
 153
 154   if (flushing) {
 155     /* write the control socket so that waiters get woken up and can check the
 156      * flushing flag we set above */
 157     gst_poll_write_control (pool->poll);
 158     flush_buffers (pool);
 159   } else {
 160     gst_poll_read_control (pool->poll);
 161   }
 162 }
 163
 164 /**
 165  * gst_buffer_pool_set_flushing:
 166  * @pool: a #GstBufferPool
 167  * @flushing: the new flushing state
 168  *
 169  * Control the flushing state of @pool. When the pool is flushing, new calls to
 170  * gst_buffer_pool_acquire_buffer() will return with GST_FLOW_WRONG_STATE.
 171  */
 172 void
 173 gst_buffer_pool_set_flushing (GstBufferPool * pool, gboolean flushing)
 174 {
 175   GstBufferPoolClass *pclass;
 176
 177   g_return_if_fail (GST_IS_BUFFER_POOL (pool));
 178
 179   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 180
 181   if (G_LIKELY (pclass->set_flushing))
 182     pclass->set_flushing (pool, flushing);
 183 }
 184
 185 static gboolean
 186 default_set_config (GstBufferPool * pool, GstStructure * config)
 187 {
 188   guint i;
 189   GstBufferPoolClass *pclass;
 190   GstBufferPoolPrivate *priv = pool->priv;
 191
 192   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 193
 194   /* parse the config and keep around */
 195   gst_buffer_pool_config_get (config, &priv->size, &priv->min_buffers,
 196       &priv->max_buffers, &priv->prefix, &priv->postfix, &priv->align);
 197
 198   /* we need to prealloc buffers */
 199   for (i = priv->min_buffers; i > 0; i--) {
 200     GstBuffer *buffer;
 201
 202     if (G_LIKELY (pclass->alloc_buffer)) {
 203       if (!pclass->alloc_buffer (pool, &buffer, NULL))
 204         return FALSE;
 205     } else
 206       return FALSE;
 207
 208     /* store in the queue */
 209     gst_atomic_queue_push (pool->queue, buffer);
 210     gst_poll_write_control (pool->poll);
 211   }
 212
 213   return TRUE;
 214 }
 215
 216 /**
 217  * gst_buffer_pool_set_config:
 218  * @pool: a #GstBufferPool
 219  * @config: a #GstStructure
 220  *
 221  * Set the configuration of the pool. The pool must be flushing or else this
 222  * function will do nothing and return FALSE.
 223  *
 224  * @condfig is a #GstStructure that contains the configuration parameters for
 225  * the pool. A default and mandatory set of parameters can be configured with
 226  * gst_buffer_pool_config_set().
 227  *
 228  * Returns: TRUE when the configuration could be set.
 229  */
 230 gboolean
 231 gst_buffer_pool_set_config (GstBufferPool * pool, GstStructure * config)
 232 {
 233   gboolean result;
 234   GstBufferPoolClass *pclass;
 235
 236   g_return_val_if_fail (GST_IS_BUFFER_POOL (pool), FALSE);
 237   g_return_val_if_fail (config != NULL, FALSE);
 238
 239   if (!g_atomic_int_get (&pool->flushing))
 240     return FALSE;
 241
 242   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 243
 244   /* free the buffer when we are flushing */
 245   if (G_LIKELY (pclass->set_config))
 246     result = pclass->set_config (pool, config);
 247   else
 248     result = FALSE;
 249
 250   if (result) {
 251     if (pool->config)
 252       gst_structure_free (pool->config);
 253     pool->config = config;
 254   }
 255
 256   return result;
 257 }
 258
 259 /**
 260  * gst_buffer_pool_get_config:
 261  * @pool: a #GstBufferPool
 262  *
 263  * Get the current configuration of the pool. This configuration is read-only,
 264  * use gst_structure_copy() to make a writable copy.
 265  */
 266 const GstStructure *
 267 gst_buffer_pool_get_config (GstBufferPool * pool)
 268 {
 269   g_return_val_if_fail (GST_IS_BUFFER_POOL (pool), NULL);
 270
 271   return pool->config;
 272 }
 273
 274
 275 /**
 276  * gst_buffer_pool_config_set:
 277  * @pool: a #GstBufferPool
 278  * @size: the size of each buffer, not including pre and post fix
 279  * @min_buffers: the minimum amount of buffers to allocate.
 280  * @max_buffers: the maximum amount of buffers to allocate or 0 for unlimited.
 281  * @prefix: prefix each buffer with this many bytes
 282  * @postfix: postfix each buffer with this many bytes
 283  * @align: alignment of the buffer data.
 284  *
 285  * Configure @config with the given parameters.
 286  */
 287 void
 288 gst_buffer_pool_config_set (GstStructure * config, guint size,
 289     guint min_buffers, guint max_buffers, guint prefix, guint postfix,
 290     guint align)
 291 {
 292   g_return_if_fail (config != NULL);
 293
 294   gst_structure_id_set (config,
 295       GST_QUARK (SIZE), G_TYPE_UINT, size,
 296       GST_QUARK (MIN_BUFFERS), G_TYPE_UINT, min_buffers,
 297       GST_QUARK (MAX_BUFFERS), G_TYPE_UINT, max_buffers,
 298       GST_QUARK (PREFIX), G_TYPE_UINT, prefix,
 299       GST_QUARK (POSTFIX), G_TYPE_UINT, postfix,
 300       GST_QUARK (ALIGN), G_TYPE_UINT, align, NULL);
 301 }
 302
 303 /**
 304  * gst_buffer_pool_config_get:
 305  * @pool: a #GstBufferPool
 306  * @size: the size of each buffer, not including pre and post fix
 307  * @min_buffers: the minimum amount of buffers to allocate.
 308  * @max_buffers: the maximum amount of buffers to allocate or 0 for unlimited.
 309  * @prefix: prefix each buffer with this many bytes
 310  * @postfix: postfix each buffer with this many bytes
 311  * @align: alignment of the buffer data.
 312  *
 313  * Get the configuration values from @config.
 314  */
 315 gboolean
 316 gst_buffer_pool_config_get (GstStructure * config, guint * size,
 317     guint * min_buffers, guint * max_buffers, guint * prefix, guint * postfix,
 318     guint * align)
 319 {
 320   g_return_val_if_fail (config != NULL, FALSE);
 321
 322   return gst_structure_id_get (config,
 323       GST_QUARK (SIZE), G_TYPE_UINT, size,
 324       GST_QUARK (MIN_BUFFERS), G_TYPE_UINT, min_buffers,
 325       GST_QUARK (MAX_BUFFERS), G_TYPE_UINT, max_buffers,
 326       GST_QUARK (PREFIX), G_TYPE_UINT, prefix,
 327       GST_QUARK (POSTFIX), G_TYPE_UINT, postfix,
 328       GST_QUARK (ALIGN), G_TYPE_UINT, align, NULL);
 329 }
 330
 331 static GstFlowReturn
 332 default_alloc_buffer (GstBufferPool * pool, GstBuffer ** buffer,
 333     GstBufferPoolParams * params)
 334 {
 335   guint size, align;
 336   GstBufferPoolPrivate *priv = pool->priv;
 337
 338   *buffer = gst_buffer_new ();
 339
 340   align = priv->align - 1;
 341   size = priv->prefix + priv->postfix + priv->size + align;
 342   if (size > 0) {
 343     guint8 *memptr;
 344
 345     memptr = g_malloc (size);
 346     GST_BUFFER_MALLOCDATA (*buffer) = memptr;
 347     memptr = (guint8 *) ((guintptr) (memptr + align) & ~align);
 348     GST_BUFFER_DATA (*buffer) = memptr + priv->prefix;
 349     GST_BUFFER_SIZE (*buffer) = priv->size;
 350   }
 351
 352   return GST_FLOW_OK;
 353 }
 354
 355 static GstFlowReturn
 356 default_acquire_buffer (GstBufferPool * pool, GstBuffer ** buffer,
 357     GstBufferPoolParams * params)
 358 {
 359   GstFlowReturn result;
 360   GstBufferPoolClass *pclass;
 361   GstBufferPoolPrivate *priv = pool->priv;
 362
 363   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 364
 365   while (TRUE) {
 366     if (g_atomic_int_get (&pool->flushing))
 367       return GST_FLOW_WRONG_STATE;
 368
 369     /* try to get a buffer from the queue */
 370     *buffer = gst_atomic_queue_pop (pool->queue);
 371     if (*buffer) {
 372       /* FIXME check the size of the buffer */
 373       gst_poll_read_control (pool->poll);
 374       result = GST_FLOW_OK;
 375       break;
 376     }
 377
 378     /* no buffer */
 379     if (priv->max_buffers == 0) {
 380       /* no max_buffers, we allocate some more */
 381       if (G_LIKELY (pclass->alloc_buffer))
 382         result = pclass->alloc_buffer (pool, buffer, params);
 383       else
 384         result = GST_FLOW_NOT_SUPPORTED;
 385       break;
 386     }
 387
 388     /* check if we need to wait */
 389     if (!(params->flags & GST_BUFFER_POOL_FLAG_WAIT)) {
 390       result = GST_FLOW_UNEXPECTED;
 391       break;
 392     }
 393
 394     /* now wait */
 395     gst_poll_wait (pool->poll, GST_CLOCK_TIME_NONE);
 396   }
 397   return result;
 398 }
 399
 400 /**
 401  * gst_buffer_pool_acquire_buffer:
 402  * @pool: a #GstBufferPool
 403  * @buffer: a location for a #GstBuffer
 404  * @params: parameters.
 405  *
 406  * Acquire a buffer from @pool. @buffer should point to a memory location that
 407  * can hold a pointer to the new buffer.
 408  *
 409  * @params can be NULL or contain optional parameters to influence the allocation.
 410  *
 411  * Returns: a #GstFlowReturn such as GST_FLOW_WRONG_STATE when the pool is
 412  * flushing.
 413  */
 414 GstFlowReturn
 415 gst_buffer_pool_acquire_buffer (GstBufferPool * pool, GstBuffer ** buffer,
 416     GstBufferPoolParams * params)
 417 {
 418   GstBufferPoolClass *pclass;
 419   GstFlowReturn result;
 420
 421   g_return_val_if_fail (GST_IS_BUFFER_POOL (pool), GST_FLOW_ERROR);
 422   g_return_val_if_fail (buffer != NULL, GST_FLOW_ERROR);
 423
 424   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 425
 426   if (G_LIKELY (pclass->acquire_buffer))
 427     result = pclass->acquire_buffer (pool, buffer, params);
 428   else
 429     result = GST_FLOW_NOT_SUPPORTED;
 430
 431   return result;
 432 }
 433
 434 static void
 435 default_free_buffer (GstBufferPool * pool, GstBuffer * buffer)
 436 {
 437   gst_buffer_unref (buffer);
 438 }
 439
 440 static void
 441 default_release_buffer (GstBufferPool * pool, GstBuffer * buffer)
 442 {
 443   /* keep it around in our queue, we might be flushing but that's ok because we
 444    * handle that unlikely case below. */
 445   gst_atomic_queue_push (pool->queue, buffer);
 446   gst_poll_write_control (pool->poll);
 447
 448   if (G_UNLIKELY (g_atomic_int_get (&pool->flushing))) {
 449     /* we are flushing, remove the buffers again */
 450     flush_buffers (pool);
 451   }
 452 }
 453
 454 /**
 455  * gst_buffer_pool_release_buffer:
 456  * @pool: a #GstBufferPool
 457  * @buffer: a #GstBuffer
 458  *
 459  * Release @buffer to @pool. @buffer should have previously been allocated from
 460  * @pool with gst_buffer_pool_acquire_buffer().
 461  *
 462  * This function is usually called automatically when the last ref on @buffer
 463  * disappears.
 464  */
 465 void
 466 gst_buffer_pool_release_buffer (GstBufferPool * pool, GstBuffer * buffer)
 467 {
 468   GstBufferPoolClass *pclass;
 469
 470   g_return_if_fail (GST_IS_BUFFER_POOL (pool));
 471   g_return_if_fail (buffer != NULL);
 472
 473   pclass = GST_BUFFER_POOL_GET_CLASS (pool);
 474
 475   if (G_LIKELY (pclass->release_buffer))
 476     pclass->release_buffer (pool, buffer);
 477 }