gst: handle blocking in the _render() function

When we do any other blocking in the render function, we should unblock
and call _wait_preroll() when we go to PAUSED.

We can have this situation when all the buffers are queued in the
pw_stream and we get a new _render() call. We can't get more buffers
from the pool and so we must block and wait.  When we go to PAUSED we
need to unlock and go to _wait_preroll(). Implement this by setting a
pool paused flag that is set when the sink goes to paused, we can then
return a special value that does the wait_preroll().

See !2248

src/gst/gstpipewirepool.c

@@ -171,6 +171,9 @@ acquire_buffer (GstBufferPool * pool, GstBuffer ** buffer,
     if (params && (params->flags & GST_BUFFER_POOL_ACQUIRE_FLAG_DONTWAIT))
       goto no_more_buffers;
 
+    if (p->paused)
+      goto paused;
+
     GST_WARNING_OBJECT (pool, "failed to dequeue buffer: %s", strerror(errno));
     g_cond_wait (&p->cond, GST_OBJECT_GET_LOCK (pool));
   }
@@ -190,6 +193,11 @@ flushing:
     GST_OBJECT_UNLOCK (pool);
     return GST_FLOW_FLUSHING;
   }
+paused:
+  {
+    GST_OBJECT_UNLOCK (pool);
+    return GST_FLOW_CUSTOM_ERROR_1;
+  }
 no_more_buffers:
   {
     GST_LOG_OBJECT (pool, "no more buffers");
@@ -244,6 +252,16 @@ set_config (GstBufferPool * pool, GstStructure * config)
   return GST_BUFFER_POOL_CLASS (gst_pipewire_pool_parent_class)->set_config (pool, config);
 }
 
+
+void gst_pipewire_pool_set_paused (GstPipeWirePool *pool, gboolean paused)
+{
+  GST_DEBUG ("flush start");
+  GST_OBJECT_LOCK (pool);
+  pool->paused = paused;
+  g_cond_signal (&pool->cond);
+  GST_OBJECT_UNLOCK (pool);
+}
+
 static void
 flush_start (GstBufferPool * pool)
 {

src/gst/gstpipewirepool.h

@@ -44,6 +44,7 @@ struct _GstPipeWirePool {
   GstAllocator *dmabuf_allocator;
 
   GCond cond;
+  gboolean paused;
 };
 
 GstPipeWirePool *  gst_pipewire_pool_new (GstPipeWireStream *stream);
@@ -59,6 +60,8 @@ gst_pipewire_pool_has_buffers (GstPipeWirePool *pool)
 
 GstPipeWirePoolData *gst_pipewire_pool_get_data (GstBuffer *buffer);
 
+void gst_pipewire_pool_set_paused (GstPipeWirePool *pool, gboolean paused);
+
 G_END_DECLS
 
 #endif /* __GST_PIPEWIRE_POOL_H__ */

src/gst/gstpipewiresink.c

@@ -873,8 +873,15 @@ gst_pipewire_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
 
       pw_thread_loop_unlock (pwsink->stream->core->loop);
 
-      if ((res = gst_buffer_pool_acquire_buffer (GST_BUFFER_POOL_CAST (pwsink->stream->pool),
-          &b, &params)) != GST_FLOW_OK)
+      res = gst_buffer_pool_acquire_buffer (GST_BUFFER_POOL_CAST (pwsink->stream->pool),
+          &b, &params);
+      if (res == GST_FLOW_CUSTOM_ERROR_1) {
+	res = gst_base_sink_wait_preroll (bsink);
+	if (res != GST_FLOW_OK)
+          goto done;
+        continue;
+      }
+      if (res != GST_FLOW_OK)
         goto done;
 
       gst_buffer_map (b, &info, GST_MAP_WRITE);
@@ -946,9 +953,11 @@ gst_pipewire_sink_change_state (GstElement * element, GstStateChange transition)
       pw_thread_loop_lock (this->stream->core->loop);
       pw_stream_set_active(this->stream->pwstream, false);
       pw_thread_loop_unlock (this->stream->core->loop);
+      gst_pipewire_pool_set_paused(this->stream->pool, TRUE);
       break;
     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
       /* stop play ASAP by corking */
+      gst_pipewire_pool_set_paused(this->stream->pool, TRUE);
       pw_thread_loop_lock (this->stream->core->loop);
       pw_stream_set_active(this->stream->pwstream, false);
       pw_thread_loop_unlock (this->stream->core->loop);
@@ -965,6 +974,7 @@ gst_pipewire_sink_change_state (GstElement * element, GstStateChange transition)
        * from paused state to playing state which will wait until buffer pool is ready.
        * Guarantee to finish preoll if needed to active buffer pool before uncorking and
        * starting play */
+      gst_pipewire_pool_set_paused(this->stream->pool, FALSE);
       pw_thread_loop_lock (this->stream->core->loop);
       pw_stream_set_active(this->stream->pwstream, true);
       pw_thread_loop_unlock (this->stream->core->loop);
@@ -1027,4 +1037,4 @@ static  gboolean gst_pipewire_sink_event (GstBaseSink *sink, GstEvent *event) {
   }
 
   return GST_BASE_SINK_CLASS (parent_class)->event (sink, event);
-}
+}