[libgda] Improved GdaConnection locking to avoid using an idle source

[Vivien Malerba <vivien src gnome org>]

commit d21a6054e5408755b3d71c74ae96ab5d0baeef35
Author: Vivien Malerba <malerba gnome-db org>
Date:   Wed Apr 2 18:35:08 2014 +0200

    Improved GdaConnection locking to avoid using an idle source

 libgda/gda-connection.c |   62 ++++++++++++++++++++++++++++++++++++++--------
 libgda/gda-connection.h |    4 ++-
 libgda/test-cnc-exec.c  |   57 ++++++++++++++++++++++---------------------
 3 files changed, 83 insertions(+), 40 deletions(-)
---
diff --git a/libgda/gda-connection.c b/libgda/gda-connection.c
index d36028a..fc54bd8 100644
--- a/libgda/gda-connection.c
+++ b/libgda/gda-connection.c
@@ -69,6 +69,7 @@
 #include <sqlite/virtual/gda-vconnection-data-model.h>
 #include <libgda/gda-debug-macros.h>
 #include <libgda/gda-data-handler.h>
+#include <thread-wrapper/itsignaler.h>
 
 #include <glib/gstdio.h>
 #include <fcntl.h>
@@ -5734,20 +5735,35 @@ gda_connection_get_meta_store (GdaConnection *cnc)
        return cnc->priv->meta_store;
 }
 
+/* Note about the GdaConnection's locking mechanism:
+ * When a thread wants to lock a #GdaConnection:
+ *   - if it's the GdaWorker's thread then nothing needs to be done, and returns immediately
+ *   - otherwise try to lock the GRecMutex, and if Ok, then we are done.
+ *   - if the try_lock() did not work:
+ *     - if there is no GMainContext associated to the current thread, then block on mutex lock
+ *     - otherwise, we create a new main loop for that GMainContext and an ITSignaler source
+ *          which will receive a notification from the thread which currently
+ *          locks the GRecMutex right after it has released it.
+ *
+ *  The ITSignaler is appended to the @lock_its_list list, and removed once the lock has been
+ *  acquired by the thread
+ */
+static GSList *lock_its_list = NULL;
+
 typedef struct {
        GdaLockable *lockable;
        GMainLoop   *loop;
 } TryLockData;
 
 static gboolean
-idle_trylock (TryLockData *data)
+itsignaler_trylock (ITSignaler *its, TryLockData *data)
 {
        if (gda_connection_trylock (data->lockable)) {
                g_main_loop_quit (data->loop);
-               return FALSE; /* remove idle source */
+               return G_SOURCE_REMOVE;
        }
        else
-               return TRUE; /* keep idle source */
+               return G_SOURCE_CONTINUE;
 }
 
 /*
@@ -5772,10 +5788,20 @@ gda_connection_lock (GdaLockable *lockable)
        GMainContext *context;
        context = gda_connection_get_main_context (cnc, NULL);
        if (context) {
-               if (gda_connection_trylock (lockable))
+               g_mutex_lock (&global_mutex);
+               if (gda_connection_trylock (lockable)) {
+                       g_mutex_unlock (&global_mutex);
                        return;
+               }
+
+               /* create ITSignaler and add it to the list */
+               ITSignaler *its;
+               its = itsignaler_new ();
+               lock_its_list = g_slist_append (lock_its_list, its);
+               /*g_print ("++ ITSigaler_len is %u\n", g_slist_length (lock_its_list));*/
+               g_mutex_unlock (&global_mutex);
 
-               /* we must process events from @context */
+               /* create main loop to must process events from @context */
                GMainLoop *loop;
                loop = g_main_loop_new (context, FALSE);
 
@@ -5783,15 +5809,21 @@ gda_connection_lock (GdaLockable *lockable)
                lockdata.lockable = lockable;
                lockdata.loop = loop;
 
-               GSource *idle;
-               idle = g_idle_source_new ();
-               g_source_set_priority (idle, G_PRIORITY_DEFAULT);
-               g_source_attach (idle, context);
-               g_source_set_callback (idle, (GSourceFunc) idle_trylock, &lockdata, NULL);
-               g_source_unref (idle);
+               GSource *itsource;
+               itsource = itsignaler_create_source (its);
+               g_source_attach (itsource, context);
+               g_source_set_callback (itsource, (GSourceFunc) itsignaler_trylock, &lockdata, NULL);
+               g_source_unref (itsource);
 
                g_main_loop_run (loop);
                g_main_loop_unref (loop);
+
+               /* get rid of @its */
+               g_mutex_lock (&global_mutex);
+               lock_its_list = g_slist_remove (lock_its_list, its);
+               /*g_print ("-- ITSigaler_len is %u\n", g_slist_length (lock_its_list));*/
+               g_mutex_unlock (&global_mutex);
+               itsignaler_unref (its);
        }
        else
                g_rec_mutex_lock (& cnc->priv->rmutex);
@@ -5830,7 +5862,15 @@ gda_connection_unlock  (GdaLockable *lockable)
                return;
        }
 
+       /* help other threads locking the connection */
        g_rec_mutex_unlock (& cnc->priv->rmutex);
+       g_mutex_lock (&global_mutex);
+       if (lock_its_list) {
+               ITSignaler *its;
+               its = (ITSignaler*) lock_its_list->data;
+               itsignaler_push_notification (its, (gpointer) 0x01, NULL); /* "wake up" the waiting thread */
+       }
+       g_mutex_unlock (&global_mutex);
 }
 
 /*
diff --git a/libgda/gda-connection.h b/libgda/gda-connection.h
index 149faee..34fcab9 100644
--- a/libgda/gda-connection.h
+++ b/libgda/gda-connection.h
@@ -103,7 +103,9 @@ typedef enum {
  * the execution of a statement</para></listitem>
  * </itemizedlist>
  *
- * The #GdaConnection object implements its own locking mechanism so it is thread-safe.
+ * The #GdaConnection object implements its own locking mechanism so it is thread-safe. If a #GMainContext 
has been
+ * specified using gda_connection_set_main_context(), then the events will continue to be processed while the
+ * lock on the connection is being acquired.
  */
 
 struct _GdaConnection {
diff --git a/libgda/test-cnc-exec.c b/libgda/test-cnc-exec.c
index 795e3e9..22d6db4 100644
--- a/libgda/test-cnc-exec.c
+++ b/libgda/test-cnc-exec.c
@@ -157,25 +157,29 @@ test2_th (GdaConnection *cnc)
 {
        /* setup main context */
        guint counter = 0;
-       setup_main_context (cnc, 100, &counter);
-
+       setup_main_context (cnc, 10, &counter);
+#define LOCK_DELAY 50000
        gda_lockable_lock (GDA_LOCKABLE (cnc));
-       g_usleep (200000);
+       /*g_print ("Locked by %p\n", g_thread_self());*/
+       g_usleep (LOCK_DELAY);
        gda_lockable_unlock (GDA_LOCKABLE (cnc));
        g_thread_yield ();
 
        gda_lockable_lock (GDA_LOCKABLE (cnc));
-       g_usleep (200000);
+       /*g_print ("Locked by %p\n", g_thread_self());*/
+       g_usleep (LOCK_DELAY);
        gda_lockable_unlock (GDA_LOCKABLE (cnc));
        g_thread_yield ();
 
        gda_lockable_lock (GDA_LOCKABLE (cnc));
-       g_usleep (200000);
+       /*g_print ("Locked by %p\n", g_thread_self());*/
+       g_usleep (LOCK_DELAY);
        gda_lockable_unlock (GDA_LOCKABLE (cnc));
        g_thread_yield ();
 
        gda_lockable_lock (GDA_LOCKABLE (cnc));
-       g_usleep (200000);
+       /*g_print ("Locked by %p\n", g_thread_self());*/
+       g_usleep (LOCK_DELAY);
        gda_lockable_unlock (GDA_LOCKABLE (cnc));
 
        guint *ret;
@@ -213,31 +217,28 @@ test2 (void)
        }
        g_free (cnc_string);
 
-       GThread *tha, *thb;
-
-       tha = g_thread_new ("thA", (GThreadFunc) test2_th, cnc);
-       thb = g_thread_new ("thB", (GThreadFunc) test2_th, cnc);
-       g_print ("Thread A is %p\n", tha);
-       g_print ("Thread B is %p\n", thb);
-
-       guint *counter;
-       counter = g_thread_join (tha);
-       if (*counter == 0) {
-               g_print ("Thread A: gda_connection_lock() failed: did not make GMainContext 'run'\n");
-               return 1;
+#define NB_THREADS 3
+       GThread *ths[NB_THREADS];
+       guint i;
+       for (i = 0; i < NB_THREADS; i++) {
+               gchar *tmp;
+               tmp = g_strdup_printf ("th%u", i);
+               ths[i] = g_thread_new (tmp, (GThreadFunc) test2_th, cnc);
+               g_free (tmp);
+               g_print ("Thread %u is %p\n", i, ths[i]);
        }
-       else
-               g_print ("Thread A: Counter incremented to %u\n", *counter);
-       g_free (counter);
 
-       counter = g_thread_join (thb);
-       if (*counter == 0) {
-               g_print ("Thread B: gda_connection_lock() failed: did not make GMainContext 'run'\n");
-               return 1;
+       for (i = 0; i < NB_THREADS; i++) {
+               guint *counter;
+               counter = g_thread_join (ths[i]);
+               if (*counter == 0) {
+                       g_print ("Thread %u: gda_connection_lock() failed: did not make GMainContext 
'run'\n", i);
+                       return 1;
+               }
+               else
+                       g_print ("Thread %u: Counter incremented to %u\n", i, *counter);
+               g_free (counter);
        }
-       else
-               g_print ("Thread B: Counter incremented to %u\n", *counter);
-       g_free (counter);
 
        g_object_unref (cnc);