waylandsink: Handle wl_buffer::release and don't reuse buffers that are not released

This is achieved by adding an extra reference on the buffers, which does
not allow them to return to the pool. When they are released, this reference
is dropped.

The rest complexity of this patch (hash table, mutex, flag, explicit release calls)
merely exists to allow a safe, guaranteed and deadlock-free destruction sequence.
See the added comment on gstwaylandsink.c for details.

ext/wayland/gstwaylandsink.c

@@ -3,6 +3,7 @@
  * Copyright (C) 2011 Intel Corporation
  * Copyright (C) 2011 Sreerenj Balachandran <sreerenj.balachandran@intel.com>
  * Copyright (C) 2012 Wim Taymans <wim.taymans@gmail.com>
+ * Copyright (C) 2014 Collabora Ltd.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Library General Public
@@ -201,10 +202,14 @@ gst_wayland_sink_finalize (GObject * object)
 
   GST_DEBUG_OBJECT (sink, "Finalizing the sink..");
 
+  if (sink->display) {
+    /* see comment about this call in gst_wayland_sink_change_state() */
+    gst_wayland_compositor_release_all_buffers (GST_WAYLAND_BUFFER_POOL
+        (sink->pool));
+    g_object_unref (sink->display);
+  }
   if (sink->window)
     g_object_unref (sink->window);
-  if (sink->display)
-    g_object_unref (sink->display);
   if (sink->pool)
     gst_object_unref (sink->pool);
 
@@ -238,6 +243,21 @@ gst_wayland_sink_change_state (GstElement * element, GstStateChange transition)
   switch (transition) {
     case GST_STATE_CHANGE_PAUSED_TO_READY:
       if (gst_wl_window_is_toplevel (sink->window)) {
+        /* Force all buffers to return to the pool, regardless of
+         * whether the compositor has released them or not. We are
+         * going to kill the display, so we need to return all buffers
+         * to be destroyed before this happens.
+         * Note that this is done here instead of the pool destructor
+         * because the buffers hold a reference to the pool. Also,
+         * the buffers can only be unref'ed from the display's event loop
+         * and the pool holds a reference to the display. If we drop
+         * our references here, when the compositor releases the buffers,
+         * they will be unref'ed from the event loop thread, which will
+         * unref the pool and therefore the display, which will try to
+         * stop the thread from within itself and cause a deadlock.
+         */
+        gst_wayland_compositor_release_all_buffers (GST_WAYLAND_BUFFER_POOL
+            (sink->pool));
         g_clear_object (&sink->window);
         g_clear_object (&sink->display);
         g_clear_object (&sink->pool);
@@ -508,7 +528,7 @@ gst_wayland_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
 
   meta = gst_buffer_get_wl_meta (buffer);
 
-  if (meta && meta->display == sink->display) {
+  if (meta && meta->pool->display == sink->display) {
     GST_LOG_OBJECT (sink, "buffer %p from our pool, writing directly", buffer);
     to_render = buffer;
   } else {
@@ -541,6 +561,12 @@ gst_wayland_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
 
   surface = gst_wl_window_get_wl_surface (sink->window);
 
+  /* Here we essentially add a reference to the buffer. This represents
+   * the fact that the compositor is using the buffer and it should
+   * not return back to the pool and be reused until the compositor
+   * releases it. The release is handled internally in the pool */
+  gst_wayland_compositor_acquire_buffer (meta->pool, to_render);
+
   wl_surface_attach (surface, meta->wbuffer, 0, 0);
   wl_surface_damage (surface, 0, 0, res.w, res.h);
   callback = wl_surface_frame (surface);

ext/wayland/waylandpool.c

@@ -1,7 +1,8 @@
 /* GStreamer
  * Copyright (C) 2012 Intel Corporation
  * Copyright (C) 2012 Sreerenj Balachandran <sreerenj.balachandran@intel.com>
-
+ * Copyright (C) 2014 Collabora Ltd.
+ *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Library General Public
  * License as published by the Free Software Foundation; either
@@ -55,8 +56,6 @@ gst_wl_meta_api_get_type (void)
 static void
 gst_wl_meta_free (GstWlMeta * meta, GstBuffer * buffer)
 {
-  g_object_unref (meta->display);
-
   GST_DEBUG ("destroying wl_buffer %p", meta->wbuffer);
   wl_buffer_destroy (meta->wbuffer);
 }
@@ -108,6 +107,8 @@ static void
 gst_wayland_buffer_pool_init (GstWaylandBufferPool * self)
 {
   gst_video_info_init (&self->info);
+  g_mutex_init (&self->buffers_map_mutex);
+  self->buffers_map = g_hash_table_new (g_direct_hash, g_direct_equal);
 }
 
 static void
@@ -118,11 +119,86 @@ gst_wayland_buffer_pool_finalize (GObject * object)
   if (pool->wl_pool)
     gst_wayland_buffer_pool_stop (GST_BUFFER_POOL (pool));
 
+  g_mutex_clear (&pool->buffers_map_mutex);
+  g_hash_table_unref (pool->buffers_map);
+
   g_object_unref (pool->display);
 
   G_OBJECT_CLASS (gst_wayland_buffer_pool_parent_class)->finalize (object);
 }
 
+static void
+buffer_release (void *data, struct wl_buffer *wl_buffer)
+{
+  GstWaylandBufferPool *self = data;
+  GstBuffer *buffer;
+  GstWlMeta *meta;
+
+  g_mutex_lock (&self->buffers_map_mutex);
+  buffer = g_hash_table_lookup (self->buffers_map, wl_buffer);
+
+  GST_LOG_OBJECT (self, "wl_buffer::release (GstBuffer: %p)", buffer);
+
+  if (buffer) {
+    meta = gst_buffer_get_wl_meta (buffer);
+    if (meta->used_by_compositor) {
+      meta->used_by_compositor = FALSE;
+      /* unlock before unref because stop() may be called from here */
+      g_mutex_unlock (&self->buffers_map_mutex);
+      gst_buffer_unref (buffer);
+      return;
+    }
+  }
+  g_mutex_unlock (&self->buffers_map_mutex);
+}
+
+static const struct wl_buffer_listener buffer_listener = {
+  buffer_release
+};
+
+void
+gst_wayland_compositor_acquire_buffer (GstWaylandBufferPool * self,
+    GstBuffer * buffer)
+{
+  GstWlMeta *meta;
+
+  meta = gst_buffer_get_wl_meta (buffer);
+  g_return_if_fail (meta != NULL);
+  g_return_if_fail (meta->pool == self);
+  g_return_if_fail (meta->used_by_compositor == FALSE);
+
+  meta->used_by_compositor = TRUE;
+  gst_buffer_ref (buffer);
+}
+
+static void
+unref_used_buffers (gpointer key, gpointer value, gpointer data)
+{
+  GstBuffer *buffer = value;
+  GstWlMeta *meta = gst_buffer_get_wl_meta (buffer);
+  GList **to_unref = data;
+
+  if (meta->used_by_compositor) {
+    meta->used_by_compositor = FALSE;
+    *to_unref = g_list_prepend (*to_unref, buffer);
+  }
+}
+
+void
+gst_wayland_compositor_release_all_buffers (GstWaylandBufferPool * self)
+{
+  GList *to_unref = NULL;
+
+  g_mutex_lock (&self->buffers_map_mutex);
+  g_hash_table_foreach (self->buffers_map, unref_used_buffers, &to_unref);
+  g_mutex_unlock (&self->buffers_map_mutex);
+
+  /* unref without the lock because stop() may be called from here */
+  if (to_unref) {
+    g_list_free_full (to_unref, (GDestroyNotify) gst_buffer_unref);
+  }
+}
+
 static gboolean
 gst_wayland_buffer_pool_set_config (GstBufferPool * pool, GstStructure * config)
 {
@@ -226,6 +302,12 @@ gst_wayland_buffer_pool_stop (GstBufferPool * pool)
   self->size = 0;
   self->used = 0;
 
+  /* all buffers are about to be destroyed;
+   * we should no longer do anything with them */
+  g_mutex_lock (&self->buffers_map_mutex);
+  g_hash_table_remove_all (self->buffers_map);
+  g_mutex_unlock (&self->buffers_map_mutex);
+
   return GST_BUFFER_POOL_CLASS (parent_class)->stop (pool);
 }
 
@@ -263,9 +345,17 @@ gst_wayland_buffer_pool_alloc (GstBufferPool * pool, GstBuffer ** buffer,
   /* create buffer and its metadata object */
   *buffer = gst_buffer_new ();
   meta = (GstWlMeta *) gst_buffer_add_meta (*buffer, GST_WL_META_INFO, NULL);
-  meta->display = g_object_ref (self->display);
+  meta->pool = self;
   meta->wbuffer = wl_shm_pool_create_buffer (self->wl_pool, offset,
       width, height, stride, format);
+  meta->used_by_compositor = FALSE;
+
+  /* configure listening to wl_buffer.release */
+  g_mutex_lock (&self->buffers_map_mutex);
+  g_hash_table_insert (self->buffers_map, meta->wbuffer, *buffer);
+  g_mutex_unlock (&self->buffers_map_mutex);
+
+  wl_buffer_add_listener (meta->wbuffer, &buffer_listener, self);
 
   /* add the allocated memory on the GstBuffer */
   gst_buffer_append_memory (*buffer,

ext/wayland/waylandpool.h

@@ -1,6 +1,7 @@
 /* GStreamer Wayland buffer pool
  * Copyright (C) 2012 Intel Corporation
  * Copyright (C) 2012 Sreerenj Balachandran <sreerenj.balachandran@intel.com>
+ * Copyright (C) 2014 Collabora Ltd.
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Library General Public
@@ -28,6 +29,14 @@
 
 G_BEGIN_DECLS
 
+#define GST_TYPE_WAYLAND_BUFFER_POOL      (gst_wayland_buffer_pool_get_type())
+#define GST_IS_WAYLAND_BUFFER_POOL(obj)   (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_WAYLAND_BUFFER_POOL))
+#define GST_WAYLAND_BUFFER_POOL(obj)      (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_WAYLAND_BUFFER_POOL, GstWaylandBufferPool))
+#define GST_WAYLAND_BUFFER_POOL_CAST(obj) ((GstWaylandBufferPool*)(obj))
+
+typedef struct _GstWaylandBufferPool GstWaylandBufferPool;
+typedef struct _GstWaylandBufferPoolClass GstWaylandBufferPoolClass;
+
 /* buffer meta */
 typedef struct _GstWlMeta GstWlMeta;
 
@@ -42,19 +51,12 @@ const GstMetaInfo * gst_wl_meta_get_info (void);
 struct _GstWlMeta {
   GstMeta meta;
 
-  GstWlDisplay *display;
+  GstWaylandBufferPool *pool;
   struct wl_buffer *wbuffer;
+  gboolean used_by_compositor;
 };
 
 /* buffer pool */
-#define GST_TYPE_WAYLAND_BUFFER_POOL      (gst_wayland_buffer_pool_get_type())
-#define GST_IS_WAYLAND_BUFFER_POOL(obj)   (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_WAYLAND_BUFFER_POOL))
-#define GST_WAYLAND_BUFFER_POOL(obj)      (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_WAYLAND_BUFFER_POOL, GstWaylandBufferPool))
-#define GST_WAYLAND_BUFFER_POOL_CAST(obj) ((GstWaylandBufferPool*)(obj))
-
-typedef struct _GstWaylandBufferPool GstWaylandBufferPool;
-typedef struct _GstWaylandBufferPoolClass GstWaylandBufferPoolClass;
-
 struct _GstWaylandBufferPool
 {
   GstBufferPool bufferpool;
@@ -68,6 +70,9 @@ struct _GstWaylandBufferPool
   size_t size;
   size_t used;
   void *data;
+
+  GMutex buffers_map_mutex;
+  GHashTable *buffers_map;
 };
 
 struct _GstWaylandBufferPoolClass
@@ -79,6 +84,11 @@ GType gst_wayland_buffer_pool_get_type (void);
 
 GstBufferPool *gst_wayland_buffer_pool_new (GstWlDisplay * display);
 
+
+void gst_wayland_compositor_acquire_buffer (GstWaylandBufferPool * self,
+    GstBuffer * buffer);
+void gst_wayland_compositor_release_all_buffers (GstWaylandBufferPool * self);
+
 G_END_DECLS
 
 #endif /*__GST_WAYLAND_BUFFER_POOL_H__*/