[glib] _g_dbus_worker_flush_sync: always flush if we need to

[Simon McVittie <smcv src gnome org>]

commit bae9928246d5104e1d73a60dd66bfbc45740d640
Author: Simon McVittie <simon mcvittie collabora co uk>
Date:   Mon Nov 21 17:20:52 2011 +0000

    _g_dbus_worker_flush_sync: always flush if we need to
    
    We didn't previously flush in a couple of cases where we should have
    done:
    
    * a write is running when flush is called: we should flush after it
      finishes
    
    * writes have been made since the last flush, but none are pending or
      running right now: we should flush the underlying transport straight
      away
    
    Bug: https://bugzilla.gnome.org/show_bug.cgi?id=662395
    Signed-off-by: Simon McVittie <simon mcvittie collabora co uk>
    Reviewed-by: Cosimo Alfarano <cosimo alfarano collabora co uk>

 gio/gdbusprivate.c |   43 +++++++++++++++++++++++++++++++++++++------
 1 files changed, 37 insertions(+), 6 deletions(-)
---
diff --git a/gio/gdbusprivate.c b/gio/gdbusprivate.c
index 48606c4..178214e 100644
--- a/gio/gdbusprivate.c
+++ b/gio/gdbusprivate.c
@@ -387,6 +387,10 @@ struct GDBusWorker
   GQueue                             *write_queue;
   /* protected by write_lock */
   guint64                             write_num_messages_written;
+  /* number of messages we'd written out last time we flushed;
+   * protected by write_lock
+   */
+  guint64                             write_num_messages_flushed;
   /* list of FlushData, protected by write_lock */
   GList                              *write_pending_flushes;
   /* list of CloseData, protected by write_lock */
@@ -1208,6 +1212,7 @@ ostream_flush_cb (GObject      *source_object,
   /* Make sure we tell folks that we don't have additional
      flushes pending */
   g_mutex_lock (&data->worker->write_lock);
+  data->worker->write_num_messages_flushed = data->worker->write_num_messages_written;
   g_assert (data->worker->output_pending == PENDING_FLUSH);
   data->worker->output_pending = PENDING_NONE;
   g_mutex_unlock (&data->worker->write_lock);
@@ -1556,6 +1561,7 @@ continue_writing_in_idle_cb (gpointer user_data)
 
 /*
  * @write_data: (transfer full) (allow-none):
+ * @flush_data: (transfer full) (allow-none):
  * @close_data: (transfer full) (allow-none):
  *
  * Can be called from any thread
@@ -1566,15 +1572,26 @@ continue_writing_in_idle_cb (gpointer user_data)
 static void
 schedule_writing_unlocked (GDBusWorker        *worker,
                            MessageToWriteData *write_data,
+                           FlushData          *flush_data,
                            CloseData          *close_data)
 {
   if (write_data != NULL)
     g_queue_push_tail (worker->write_queue, write_data);
 
+  if (flush_data != NULL)
+    worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, flush_data);
+
   if (close_data != NULL)
     worker->pending_close_attempts = g_list_prepend (worker->pending_close_attempts,
                                                      close_data);
 
+  /* If we had output pending, the next bit of output will happen
+   * automatically when it finishes, so we only need to do this
+   * if nothing was pending.
+   *
+   * The idle callback will re-check that output_pending is still
+   * PENDING_NONE, to guard against output starting before the idle.
+   */
   if (worker->output_pending == PENDING_NONE)
     {
       GSource *idle_source;
@@ -1615,7 +1632,7 @@ _g_dbus_worker_send_message (GDBusWorker    *worker,
   data->blob_size = blob_len;
 
   g_mutex_lock (&worker->write_lock);
-  schedule_writing_unlocked (worker, data, NULL);
+  schedule_writing_unlocked (worker, data, NULL, NULL);
   g_mutex_unlock (&worker->write_lock);
 }
 
@@ -1700,7 +1717,7 @@ _g_dbus_worker_close (GDBusWorker         *worker,
    */
   g_cancellable_cancel (worker->cancellable);
   g_mutex_lock (&worker->write_lock);
-  schedule_writing_unlocked (worker, NULL, close_data);
+  schedule_writing_unlocked (worker, NULL, NULL, close_data);
   g_mutex_unlock (&worker->write_lock);
 }
 
@@ -1744,20 +1761,34 @@ _g_dbus_worker_flush_sync (GDBusWorker    *worker,
 {
   gboolean ret;
   FlushData *data;
+  guint64 pending_writes;
 
   data = NULL;
   ret = TRUE;
 
-  /* if the queue is empty, there's nothing to wait for */
   g_mutex_lock (&worker->write_lock);
-  if (g_queue_get_length (worker->write_queue) > 0)
+
+  /* if the queue is empty, no write is in-flight and we haven't written
+   * anything since the last flush, then there's nothing to wait for
+   */
+  pending_writes = g_queue_get_length (worker->write_queue);
+
+  /* if a write is in-flight, we shouldn't be satisfied until the first
+   * flush operation that follows it
+   */
+  if (worker->output_pending == PENDING_WRITE)
+    pending_writes += 1;
+
+  if (pending_writes > 0 ||
+      worker->write_num_messages_written != worker->write_num_messages_flushed)
     {
       data = g_new0 (FlushData, 1);
       g_mutex_init (&data->mutex);
       g_cond_init (&data->cond);
-      data->number_to_wait_for = worker->write_num_messages_written + g_queue_get_length (worker->write_queue);
+      data->number_to_wait_for = worker->write_num_messages_written + pending_writes;
       g_mutex_lock (&data->mutex);
-      worker->write_pending_flushes = g_list_prepend (worker->write_pending_flushes, data);
+
+      schedule_writing_unlocked (worker, NULL, data, NULL);
     }
   g_mutex_unlock (&worker->write_lock);