vulkan/queue: be sure to take a lock around command submission

This ensures that only one thread is submitting commands at a time as
required by the Vulkan specification.

ext/vulkan/vkdownload.c

@@ -311,9 +311,11 @@ _image_to_raw_perform (gpointer impl, GstBuffer * inbuf, GstBuffer ** outbuf)
     if (!fence)
       goto error;
 
+    gst_vulkan_queue_submit_lock (raw->download->queue);
     err =
         vkQueueSubmit (raw->download->queue->queue, 1, &submit_info,
         GST_VULKAN_FENCE_FENCE (fence));
+    gst_vulkan_queue_submit_unlock (raw->download->queue);
     if (gst_vulkan_error_to_g_error (err, &error, "vkQueueSubmit") < 0)
       goto error;
 

ext/vulkan/vkfullscreenrender.c

@@ -900,9 +900,11 @@ gst_vulkan_full_screen_render_submit (GstVulkanFullScreenRender * render,
     gst_vulkan_fence_unref (render->last_fence);
   render->last_fence = gst_vulkan_fence_ref (fence);
 
+  gst_vulkan_queue_submit_lock (render->queue);
   err =
       vkQueueSubmit (render->queue->queue, 1, &submit_info,
       GST_VULKAN_FENCE_FENCE (fence));
+  gst_vulkan_queue_submit_unlock (render->queue);
   if (gst_vulkan_error_to_g_error (err, &error, "vkQueueSubmit") < 0)
     goto error;
 

ext/vulkan/vkupload.c

@@ -987,9 +987,11 @@ _raw_to_image_perform (gpointer impl, GstBuffer * inbuf, GstBuffer ** outbuf)
     if (!fence)
       goto error;
 
+    gst_vulkan_queue_submit_lock (raw->upload->queue);
     err =
         vkQueueSubmit (raw->upload->queue->queue, 1, &submit_info,
         GST_VULKAN_FENCE_FENCE (fence));
+    gst_vulkan_queue_submit_unlock (raw->upload->queue);
     if (gst_vulkan_error_to_g_error (err, &error, "vkQueueSubmit") < 0)
       goto error;
 

gst-libs/gst/vulkan/gstvkqueue.c

@@ -49,21 +49,31 @@ _init_debug (void)
   }
 }
 
+struct _GstVulkanQueuePrivate
+{
+  GMutex submit_lock;
+};
+
 #define parent_class gst_vulkan_queue_parent_class
 G_DEFINE_TYPE_WITH_CODE (GstVulkanQueue, gst_vulkan_queue, GST_TYPE_OBJECT,
-    _init_debug ());
+    G_ADD_PRIVATE (GstVulkanQueue); _init_debug ());
+
+#define GET_PRIV(queue) gst_vulkan_queue_get_instance_private (queue)
 
 static void gst_vulkan_queue_dispose (GObject * object);
 
 static void
-gst_vulkan_queue_init (GstVulkanQueue * device)
+gst_vulkan_queue_init (GstVulkanQueue * queue)
 {
+  GstVulkanQueuePrivate *priv = GET_PRIV (queue);
+
+  g_mutex_init (&priv->submit_lock);
 }
 
 static void
-gst_vulkan_queue_class_init (GstVulkanQueueClass * device_class)
+gst_vulkan_queue_class_init (GstVulkanQueueClass * queue_class)
 {
-  GObjectClass *gobject_class = (GObjectClass *) device_class;
+  GObjectClass *gobject_class = (GObjectClass *) queue_class;
 
   gobject_class->dispose = gst_vulkan_queue_dispose;
 }
@@ -72,11 +82,14 @@ static void
 gst_vulkan_queue_dispose (GObject * object)
 {
   GstVulkanQueue *queue = GST_VULKAN_QUEUE (object);
+  GstVulkanQueuePrivate *priv = GET_PRIV (queue);
 
   if (queue->device)
     gst_object_unref (queue->device);
   queue->device = NULL;
 
+  g_mutex_clear (&priv->submit_lock);
+
   G_OBJECT_CLASS (parent_class)->dispose (object);
 }
 
@@ -291,3 +304,34 @@ gst_vulkan_queue_run_context_query (GstElement * element,
 
   return FALSE;
 }
+
+/**
+ * gst_vulkan_queue_submit_lock:
+ * @queue: a #GstVulkanQueue
+ *
+ * Locks the queue for command submission using `vkQueueSubmit()` to meet the
+ * Vulkan requirements for externally synchronised resources.
+ */
+void
+gst_vulkan_queue_submit_lock (GstVulkanQueue * queue)
+{
+  GstVulkanQueuePrivate *priv = GET_PRIV (queue);
+
+  g_mutex_lock (&priv->submit_lock);
+}
+
+/**
+ * gst_vulkan_queue_submit_unlock:
+ * @queue: a #GstVulkanQueue
+ *
+ * Unlocks the queue for command submission using `vkQueueSubmit()`.
+ *
+ * See gst_vulkan_queue_submit_lock() for details on when this call is needed.
+ */
+void
+gst_vulkan_queue_submit_unlock (GstVulkanQueue * queue)
+{
+  GstVulkanQueuePrivate *priv = GET_PRIV (queue);
+
+  g_mutex_unlock (&priv->submit_lock);
+}

gst-libs/gst/vulkan/gstvkqueue.h

@@ -58,6 +58,11 @@ GST_VULKAN_API
 GstVulkanCommandPool *  gst_vulkan_queue_create_command_pool    (GstVulkanQueue * queue,
                                                                  GError ** error);
 
+GST_VULKAN_API
+void                gst_vulkan_queue_submit_lock                (GstVulkanQueue * queue);
+GST_VULKAN_API
+void                gst_vulkan_queue_submit_unlock              (GstVulkanQueue * queue);
+
 GST_VULKAN_API
 void                gst_context_set_vulkan_queue                (GstContext * context,
                                                                  GstVulkanQueue * queue);

gst-libs/gst/vulkan/gstvkswapper.c

@@ -1268,9 +1268,11 @@ reacquire:
     if (!fence)
       goto error;
 
+    gst_vulkan_queue_submit_lock (swapper->queue);
     err =
         vkQueueSubmit (swapper->queue->queue, 1, &submit_info,
         GST_VULKAN_FENCE_FENCE (fence));
+    gst_vulkan_queue_submit_unlock (swapper->queue);
     if (gst_vulkan_error_to_g_error (err, error, "vkQueueSubmit") < 0)
       goto error;
 
@@ -1325,9 +1327,11 @@ reacquire:
     if (!fence)
       goto error;
 
+    gst_vulkan_queue_submit_lock (swapper->queue);
     err =
         vkQueueSubmit (swapper->queue->queue, 1, &submit_info,
         GST_VULKAN_FENCE_FENCE (fence));
+    gst_vulkan_queue_submit_unlock (swapper->queue);
     if (gst_vulkan_error_to_g_error (err, error, "vkQueueSubmit") < 0)
       goto error;
 

gst-libs/gst/vulkan/vulkan_fwd.h

@@ -39,6 +39,7 @@ typedef struct _GstVulkanPhysicalDevicePrivate GstVulkanPhysicalDevicePrivate;
 
 typedef struct _GstVulkanQueue GstVulkanQueue;
 typedef struct _GstVulkanQueueClass GstVulkanQueueClass;
+typedef struct _GstVulkanQueuePrivate GstVulkanQueuePrivate;
 
 typedef struct _GstVulkanCommandPool GstVulkanCommandPool;
 typedef struct _GstVulkanCommandPoolClass GstVulkanCommandPoolClass;