[gegl] buffer: added configurable limit to file backend writer thread's queue size

[Ville Sokk <villesokk src gnome org>]

commit 67688583b5499cab10aa89c1fec2e7a2f0668a5a
Author: Ville Sokk <ville sokk gmail com>
Date:   Mon Aug 6 21:39:10 2012 +0300

    buffer: added configurable limit to file backend writer thread's queue size

 gegl/buffer/gegl-tile-backend-file-async.c  |   16 +++++++++++++---
 gegl/buffer/gegl-tile-backend-file-mapped.c |   18 +++++++++++++++---
 gegl/gegl-config.c                          |   18 +++++++++++++++++-
 gegl/gegl-config.h                          |    1 +
 gegl/gegl-init.c                            |   17 ++++++++++-------
 5 files changed, 56 insertions(+), 14 deletions(-)
---
diff --git a/gegl/buffer/gegl-tile-backend-file-async.c b/gegl/buffer/gegl-tile-backend-file-async.c
index 70d7388..beed049 100644
--- a/gegl/buffer/gegl-tile-backend-file-async.c
+++ b/gegl/buffer/gegl-tile-backend-file-async.c
@@ -48,6 +48,7 @@
 #include "gegl-buffer-index.h"
 #include "gegl-buffer-types.h"
 #include "gegl-debug.h"
+#include "gegl-config.h"
 
 
 #ifndef HAVE_FSYNC
@@ -153,6 +154,7 @@ static gint peak_file_size = 0;
 static GQueue  queue      = G_QUEUE_INIT;
 static GMutex *mutex      = NULL;
 static GCond  *queue_cond = NULL;
+static GCond  *max_cond   = NULL;
 
 
 static void
@@ -167,14 +169,18 @@ gegl_tile_backend_file_finish_writing (GeglTileBackendFile *self)
 static void
 gegl_tile_backend_file_push_queue (GeglFileBackendThreadParams *params)
 {
-  gboolean was_empty;
+  guint length;
 
   g_mutex_lock (mutex);
 
-  was_empty = g_queue_is_empty (&queue);
+  length = g_queue_get_length (&queue);
+
+  if (length > gegl_config ()->queue_limit)
+    g_cond_wait (max_cond, mutex);
+
   params->file->pending_ops += 1;
   g_queue_push_tail (&queue, params);
-  if (was_empty)
+  if (length == 0) /* wake up the writer thread */
     g_cond_signal (queue_cond);
 
   g_mutex_unlock (mutex);
@@ -253,6 +259,9 @@ gegl_tile_backend_file_writer_thread (gpointer ignored)
       if (params->file->pending_ops == 0)
         g_cond_signal (params->file->cond);
 
+      if (g_queue_get_length (&queue) < gegl_config ()->queue_limit)
+        g_cond_signal (max_cond);
+
       if (params->entry)
         params->entry->in_queue = FALSE;
 
@@ -1142,6 +1151,7 @@ gegl_tile_backend_file_class_init (GeglTileBackendFileClass *klass)
   gobject_class->finalize     = gegl_tile_backend_file_finalize;
 
   queue_cond = g_cond_new ();
+  max_cond   = g_cond_new ();
   mutex      = g_mutex_new ();
   g_thread_create_full (gegl_tile_backend_file_writer_thread,
                         NULL, 0, TRUE, TRUE, G_THREAD_PRIORITY_NORMAL, NULL);
diff --git a/gegl/buffer/gegl-tile-backend-file-mapped.c b/gegl/buffer/gegl-tile-backend-file-mapped.c
index da44f38..b401bf8 100644
--- a/gegl/buffer/gegl-tile-backend-file-mapped.c
+++ b/gegl/buffer/gegl-tile-backend-file-mapped.c
@@ -48,6 +48,7 @@
 #include "gegl-buffer-index.h"
 #include "gegl-buffer-types.h"
 #include "gegl-debug.h"
+#include "gegl-config.h"
 
 
 #ifndef HAVE_FSYNC
@@ -152,6 +153,7 @@ static gint peak_file_size = 0;
 static GQueue  queue      = G_QUEUE_INIT;
 static GMutex *mutex      = NULL;
 static GCond  *queue_cond = NULL;
+static GCond  *max_cond   = NULL;
 
 
 static void
@@ -166,14 +168,18 @@ gegl_tile_backend_file_finish_writing (GeglTileBackendFile *self)
 static void
 gegl_tile_backend_file_push_queue (GeglFileBackendThreadParams *params)
 {
-  gboolean was_empty;
+  guint length;
 
   g_mutex_lock (mutex);
 
-  was_empty = g_queue_is_empty (&queue);
+  length = g_queue_get_length (&queue);
+
+  if (length > gegl_config()->queue_limit)
+    g_cond_wait (max_cond, mutex);
+
   params->file->pending_ops += 1;
   g_queue_push_tail (&queue, params);
-  if (was_empty)
+  if (length == 0) /* wake up the writer thread */
     g_cond_signal (queue_cond);
 
   g_mutex_unlock (mutex);
@@ -281,6 +287,9 @@ gegl_tile_backend_file_writer_thread (gpointer ignored)
       if (params->file->pending_ops == 0)
         g_cond_signal (params->file->cond);
 
+      if (g_queue_get_length (&queue) < gegl_config ()->queue_limit)
+        g_cond_signal (max_cond);
+
       if (params->entry)
         params->entry->in_queue = FALSE;
 
@@ -288,6 +297,7 @@ gegl_tile_backend_file_writer_thread (gpointer ignored)
         g_free (params->source);
 
       g_free (params);
+
       g_mutex_unlock (mutex);
     }
 
@@ -917,6 +927,7 @@ gegl_tile_backend_file_load_index (GeglTileBackendFile *self,
   self->free_list      = NULL;
   self->next_pre_alloc = max; /* if bigger than own? */
   self->tiles          = NULL;
+  self->total          = max;
 }
 
 static void
@@ -1069,6 +1080,7 @@ gegl_tile_backend_file_class_init (GeglTileBackendFileClass *klass)
   gobject_class->finalize     = gegl_tile_backend_file_finalize;
 
   queue_cond = g_cond_new ();
+  max_cond   = g_cond_new ();
   mutex      = g_mutex_new ();
   g_thread_create_full (gegl_tile_backend_file_writer_thread,
                         NULL, 0, TRUE, TRUE, G_THREAD_PRIORITY_NORMAL, NULL);
diff --git a/gegl/gegl-config.c b/gegl/gegl-config.c
index 43a5c7b..92d4753 100644
--- a/gegl/gegl-config.c
+++ b/gegl/gegl-config.c
@@ -44,7 +44,8 @@ enum
   PROP_TILE_WIDTH,
   PROP_TILE_HEIGHT,
   PROP_THREADS,
-  PROP_USE_OPENCL
+  PROP_USE_OPENCL,
+  PROP_QUEUE_LIMIT
 };
 
 static void
@@ -97,6 +98,10 @@ gegl_config_get_property (GObject    *gobject,
         g_value_set_boolean (value, config->use_opencl);
         break;
 
+      case PROP_QUEUE_LIMIT:
+        g_value_set_int (value, config->queue_limit);
+        break;
+
       default:
         G_OBJECT_WARN_INVALID_PROPERTY_ID (gobject, property_id, pspec);
         break;
@@ -175,6 +180,9 @@ gegl_config_set_property (GObject      *gobject,
           gegl_cl_init (NULL);
 
         break;
+      case PROP_QUEUE_LIMIT:
+        config->queue_limit = g_value_get_int (value);
+        break;
       default:
         G_OBJECT_WARN_INVALID_PROPERTY_ID (gobject, property_id, pspec);
         break;
@@ -283,6 +291,14 @@ gegl_config_class_init (GeglConfigClass *klass)
                                                          TRUE,
                                                          G_PARAM_READWRITE |
                                                          G_PARAM_CONSTRUCT));
+
+  g_object_class_install_property (gobject_class, PROP_QUEUE_LIMIT,
+                                   g_param_spec_int ("queue-limit",
+                                                     "Queue limit",
+                                                     "Maximum number of entries in the file tile backend's writer queue",
+                                                     1, G_MAXINT, 1000,
+                                                     G_PARAM_READWRITE |
+                                                     G_PARAM_CONSTRUCT));
 }
 
 static void
diff --git a/gegl/gegl-config.h b/gegl/gegl-config.h
index e65c63f..4d4c3b4 100644
--- a/gegl/gegl-config.h
+++ b/gegl/gegl-config.h
@@ -44,6 +44,7 @@ struct _GeglConfig
   gint     tile_height;
   gint     threads;
   gboolean use_opencl;
+  gint     queue_limit;
 };
 
 struct _GeglConfigClass
diff --git a/gegl/gegl-init.c b/gegl/gegl-init.c
index d67f970..74dbe3d 100644
--- a/gegl/gegl-init.c
+++ b/gegl/gegl-init.c
@@ -187,13 +187,13 @@ gegl_init (gint    *argc,
   g_option_context_free (context);
 }
 
-static gchar   *cmd_gegl_swap=NULL;
-static gchar   *cmd_gegl_cache_size=NULL;
-static gchar   *cmd_gegl_chunk_size=NULL;
-static gchar   *cmd_gegl_quality=NULL;
-static gchar   *cmd_gegl_tile_size=NULL;
-static gchar   *cmd_babl_tolerance =NULL;
-static gchar   *cmd_gegl_threads=NULL;
+static gchar *cmd_gegl_swap        = NULL;
+static gchar *cmd_gegl_cache_size  = NULL;
+static gchar *cmd_gegl_chunk_size  = NULL;
+static gchar *cmd_gegl_quality     = NULL;
+static gchar *cmd_gegl_tile_size   = NULL;
+static gchar *cmd_babl_tolerance   = NULL;
+static gchar *cmd_gegl_threads     = NULL;
 
 static const GOptionEntry cmd_entries[]=
 {
@@ -284,6 +284,9 @@ GeglConfig *gegl_config (void)
       else
         config->use_opencl = TRUE;
 
+      if (g_getenv ("GEGL_QUEUE_LIMIT"))
+        config->queue_limit = atoi(g_getenv ("GEGL_QUEUE_LIMIT"));
+
       if (gegl_swap_dir())
         config->swap = g_strdup(gegl_swap_dir ());
     }