bufferpool: Add method and virtuals to set flushing state

Currently there is no other way to unlock a buffer pool other then
stopping it. This may have the effect of freeing all the buffers,
which is too heavy for a seek. This patch add a method to enter and
leave flushing state. As a convenience, flush_start/flush_stop
virtual are added so pool implementation can also unblock their own
internal poll atomically with the rest of the pool.  This is fully
backward compatible with doing stop/start to actually flush the pool
(as being done in GstBaseSrc).

https://bugzilla.gnome.org/show_bug.cgi?id=727611

docs/gst/gstreamer-sections.txt

@@ -325,6 +325,7 @@ gst_buffer_pool_set_config
 
 gst_buffer_pool_set_active
 gst_buffer_pool_is_active
+gst_buffer_pool_set_flushing
 
 GstBufferPoolAcquireFlags
 GstBufferPoolAcquireParams

gst/gstbufferpool.c

@@ -357,7 +357,6 @@ do_start (GstBufferPool * pool)
   return TRUE;
 }
 
-
 static void
 default_free_buffer (GstBufferPool * pool, GstBuffer * buffer)
 {
@@ -417,6 +416,33 @@ do_stop (GstBufferPool * pool)
   return TRUE;
 }
 
+/* must be called with the lock */
+static void
+do_set_flushing (GstBufferPool * pool, gboolean flushing)
+{
+  GstBufferPoolPrivate *priv = pool->priv;
+  GstBufferPoolClass *pclass;
+
+  pclass = GST_BUFFER_POOL_GET_CLASS (pool);
+
+  if (GST_BUFFER_POOL_IS_FLUSHING (pool) == flushing)
+    return;
+
+  if (flushing) {
+    g_atomic_int_set (&pool->flushing, 1);
+    gst_poll_write_control (priv->poll);
+
+    if (pclass->flush_start)
+      pclass->flush_start (pool);
+  } else {
+    if (pclass->flush_stop)
+      pclass->flush_stop (pool);
+
+    gst_poll_read_control (priv->poll);
+    g_atomic_int_set (&pool->flushing, 0);
+  }
+}
+
 /**
  * gst_buffer_pool_set_active:
  * @pool: a #GstBufferPool
@@ -460,15 +486,17 @@ gst_buffer_pool_set_active (GstBufferPool * pool, gboolean active)
     if (!do_start (pool))
       goto start_failed;
 
+    /* flush_stop my release buffers, setting to active to avoid running
+     * do_stop while activating the pool */
+    priv->active = TRUE;
+
     /* unset the flushing state now */
-    gst_poll_read_control (priv->poll);
-    g_atomic_int_set (&pool->flushing, 0);
+    do_set_flushing (pool, FALSE);
   } else {
     gint outstanding;
 
     /* set to flushing first */
-    g_atomic_int_set (&pool->flushing, 1);
-    gst_poll_write_control (priv->poll);
+    do_set_flushing (pool, TRUE);
 
     /* when all buffers are in the pool, free them. Else they will be
      * freed when they are released */
@@ -478,8 +506,9 @@ gst_buffer_pool_set_active (GstBufferPool * pool, gboolean active)
       if (!do_stop (pool))
         goto stop_failed;
     }
+
+    priv->active = FALSE;
   }
-  priv->active = active;
   GST_BUFFER_POOL_UNLOCK (pool);
 
   return res;
@@ -1107,9 +1136,9 @@ dec_outstanding (GstBufferPool * pool)
     if (GST_BUFFER_POOL_IS_FLUSHING (pool)) {
       /* take the lock so that set_active is not run concurrently */
       GST_BUFFER_POOL_LOCK (pool);
-      /* recheck the flushing state in the lock, the pool could have been
-       * set to active again */
-      if (GST_BUFFER_POOL_IS_FLUSHING (pool))
+      /* now that we have the lock, check if we have been de-activated with
+       * outstanding buffers */
+      if (!pool->priv->active)
         do_stop (pool);
 
       GST_BUFFER_POOL_UNLOCK (pool);
@@ -1259,3 +1288,37 @@ gst_buffer_pool_release_buffer (GstBufferPool * pool, GstBuffer * buffer)
   /* decrease the refcount that the buffer had to us */
   gst_object_unref (pool);
 }
+
+/**
+ * gst_buffer_pool_set_flushing:
+ * @pool: a #GstBufferPool
+ * @flushing: whether to start or stop flushing
+ *
+ * Enabled or disable the flushing state of a @pool without freeing or
+ * allocating buffers.
+ *
+ * Since: 1.4
+ */
+void
+gst_buffer_pool_set_flushing (GstBufferPool * pool, gboolean flushing)
+{
+  GstBufferPoolPrivate *priv;
+
+  g_return_if_fail (GST_IS_BUFFER_POOL (pool));
+
+  GST_LOG_OBJECT (pool, "flushing %d", flushing);
+
+  priv = pool->priv;
+
+  GST_BUFFER_POOL_LOCK (pool);
+
+  if (!priv->active) {
+    GST_WARNING_OBJECT (pool, "can't change flushing state of inactive pool");
+    goto done;
+  }
+
+  do_set_flushing (pool, flushing);
+
+done:
+  GST_BUFFER_POOL_UNLOCK (pool);
+}

gst/gstbufferpool.h

@@ -146,6 +146,8 @@ struct _GstBufferPool {
  *        is not set on the buffer. If #GST_BUFFER_FLAG_TAG_MEMORY is set, the 
  *        buffer will be freed with @free_buffer.
  * @free_buffer: free a buffer. The default implementation unrefs the buffer.
+ * @flush_start: enter the flushing state. Since: 1.4
+ * @flush_stop: leave the flushign state. Since: 1.4
  *
  * The GstBufferPool class.
  */
@@ -166,9 +168,11 @@ struct _GstBufferPoolClass {
   void           (*reset_buffer)   (GstBufferPool *pool, GstBuffer *buffer);
   void           (*release_buffer) (GstBufferPool *pool, GstBuffer *buffer);
   void           (*free_buffer)    (GstBufferPool *pool, GstBuffer *buffer);
+  void           (*flush_start)    (GstBufferPool *pool);
+  void           (*flush_stop)     (GstBufferPool *pool);
 
   /*< private >*/
-  gpointer _gst_reserved[GST_PADDING];
+  gpointer _gst_reserved[GST_PADDING - 2];
 };
 
 GType       gst_buffer_pool_get_type (void);
@@ -186,6 +190,8 @@ GstStructure *   gst_buffer_pool_get_config      (GstBufferPool *pool);
 const gchar **   gst_buffer_pool_get_options     (GstBufferPool *pool);
 gboolean         gst_buffer_pool_has_option      (GstBufferPool *pool, const gchar *option);
 
+void             gst_buffer_pool_set_flushing    (GstBufferPool *poo, gboolean flushing);
+
 /* helpers for configuring the config structure */
 void             gst_buffer_pool_config_set_params    (GstStructure *config, GstCaps *caps,
                                                        guint size, guint min_buffers, guint max_buffers);

tests/check/gst/gstbufferpool.c

@@ -242,6 +242,28 @@ GST_START_TEST (test_pool_config_validate)
 
 GST_END_TEST;
 
+GST_START_TEST (test_flushing_pool_returns_flushing)
+{
+  GstBufferPool *pool = create_pool (10, 0, 0);
+  GstFlowReturn ret;
+  GstBuffer *buf = NULL;
+
+  gst_buffer_pool_set_active (pool, TRUE);
+  gst_buffer_pool_set_flushing (pool, TRUE);
+
+  ret = gst_buffer_pool_acquire_buffer (pool, &buf, NULL);
+  ck_assert_int_eq (ret, GST_FLOW_FLUSHING);
+
+  gst_buffer_pool_set_flushing (pool, FALSE);
+  ret = gst_buffer_pool_acquire_buffer (pool, &buf, NULL);
+  ck_assert_int_eq (ret, GST_FLOW_OK);
+
+  gst_buffer_unref (buf);
+  gst_object_unref (pool);
+}
+
+GST_END_TEST;
+
 static Suite *
 gst_buffer_pool_suite (void)
 {
@@ -259,6 +281,7 @@ gst_buffer_pool_suite (void)
   tcase_add_test (tc_chain, test_buffer_modify_discard);
   tcase_add_test (tc_chain, test_pool_activation_and_config);
   tcase_add_test (tc_chain, test_pool_config_validate);
+  tcase_add_test (tc_chain, test_flushing_pool_returns_flushing);
 
   return s;
 }

win32/common/libgstreamer.def

@@ -158,6 +158,7 @@ EXPORTS
 	gst_buffer_pool_release_buffer
 	gst_buffer_pool_set_active
 	gst_buffer_pool_set_config
+	gst_buffer_pool_set_flushing
 	gst_buffer_prepend_memory
 	gst_buffer_remove_all_memory
 	gst_buffer_remove_memory