[gegl/threaded-base-classes: 20/22] work towards lock-step threaded point ops...

[Øyvind Kolås <ok src gnome org>]

commit 891906d770c96f5e6ef233f9208c6588f1cc9c1f
Author: Øyvind Kolås <pippin gimp org>
Date:   Sat Jun 28 11:33:14 2014 +0200

    work towards lock-step threaded point ops...

 gegl/operation/gegl-operation-composer.c        |    2 +-
 gegl/operation/gegl-operation-composer3.c       |    2 +-
 gegl/operation/gegl-operation-point-composer3.c |  124 ++++++++++++++++++++---
 3 files changed, 113 insertions(+), 15 deletions(-)
---
diff --git a/gegl/operation/gegl-operation-composer.c b/gegl/operation/gegl-operation-composer.c
index 1a1579f..096b836 100644
--- a/gegl/operation/gegl-operation-composer.c
+++ b/gegl/operation/gegl-operation-composer.c
@@ -205,7 +205,7 @@ gegl_operation_composer_process (GeglOperation        *operation,
           g_thread_pool_push (pool, &thread_data[i], NULL);
         thread_process (&thread_data[0], NULL);
 
-        while (pending > 0) {g_usleep(3);};
+        while (pending > 0) {g_usleep(0);};
         success = thread_data[0].success;
       }
       else
diff --git a/gegl/operation/gegl-operation-composer3.c b/gegl/operation/gegl-operation-composer3.c
index 6afcf8b..d62903f 100644
--- a/gegl/operation/gegl-operation-composer3.c
+++ b/gegl/operation/gegl-operation-composer3.c
@@ -243,7 +243,7 @@ gegl_operation_composer3_process (GeglOperation        *operation,
           g_thread_pool_push (pool, &thread_data[i], NULL);
         thread_process (&thread_data[0], NULL);
 
-        while (pending > 0) {g_usleep(3);};
+        while (pending > 0) {g_usleep(0);};
         success = thread_data[0].success;
       }
       else
diff --git a/gegl/operation/gegl-operation-point-composer3.c b/gegl/operation/gegl-operation-point-composer3.c
index b8c9add..f9cd039 100644
--- a/gegl/operation/gegl-operation-point-composer3.c
+++ b/gegl/operation/gegl-operation-point-composer3.c
@@ -25,8 +25,12 @@
 #include "gegl-operation-point-composer3.h"
 #include "gegl-operation-context.h"
 #include "gegl-config.h"
+#include <sys/types.h>
+#include <unistd.h>
 #include <string.h>
 
+pthread_t main_thread = 0;
+
 typedef struct ThreadData
 {
   GeglOperationPointComposer3Class *klass;
@@ -36,11 +40,15 @@ typedef struct ThreadData
   guchar                      *aux2;
   guchar                      *output;
   gint                        *pending;
+  gint                        *started;
   gint                         level;
   gboolean                     success;
   GeglRectangle                roi;
 } ThreadData;
 
+static GMutex pool_mutex = {0,};
+static GCond  pool_cond  = {0,};
+
 static void thread_process (gpointer thread_data, gpointer unused)
 {
   ThreadData *data = thread_data;
@@ -49,9 +57,18 @@ static void thread_process (gpointer thread_data, gpointer unused)
                        data->output, data->roi.width * data->roi.height, 
                        &data->roi, data->level))
     data->success = FALSE;
+  
   g_atomic_int_add (data->pending, -1);
+  if (*data->pending == 0)
+  {
+    g_mutex_lock (&pool_mutex);
+    g_cond_signal (&pool_cond);
+    g_mutex_unlock (&pool_mutex);
+  }
 }
 
+
+
 static GThreadPool *thread_pool (void)
 {
   static GThreadPool *pool = NULL;
@@ -171,6 +188,8 @@ gegl_operation_point_composer3_init (GeglOperationPointComposer3 *self)
 
 }
 
+#include <math.h>
+
 static gboolean
 gegl_operation_point_composer3_process (GeglOperation       *operation,
                                         GeglBuffer          *input,
@@ -188,22 +207,101 @@ gegl_operation_point_composer3_process (GeglOperation       *operation,
 
   if ((result->width > 0) && (result->height > 0))
     {
-      GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, out_format, 
GEGL_BUFFER_WRITE, GEGL_ABYSS_NONE);
-      gint foo = 0, bar = 0, read = 0;
+      if (gegl_operation_use_threading (operation, result) && result->height > 1)
+      {
+        gint threads = gegl_config ()->threads;
+        GThreadPool *pool = thread_pool ();
+        GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, out_format, 
GEGL_BUFFER_WRITE, GEGL_ABYSS_NONE);
+        gint foo = 0, bar = 0, read = 0;
 
-      if (input)
-        read = gegl_buffer_iterator_add (i, input, result, level, in_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
-      if (aux)
-        foo = gegl_buffer_iterator_add (i, aux, result, level, aux_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
-      if (aux2)
-        bar = gegl_buffer_iterator_add (i, aux2, result, level, aux2_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
+        gint in_bpp = babl_format_get_bytes_per_pixel (in_format);
+        gint aux_bpp = babl_format_get_bytes_per_pixel (aux_format);
+        gint aux2_bpp = babl_format_get_bytes_per_pixel (aux2_format);
+        gint out_bpp = babl_format_get_bytes_per_pixel (out_format);
+
+        main_thread = pthread_self();
+
+        if (input)
+          read = gegl_buffer_iterator_add (i, input, result, level, in_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
+        if (aux)
+          foo = gegl_buffer_iterator_add (i, aux, result, level, aux_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
+        if (aux2)
+          bar = gegl_buffer_iterator_add (i, aux2, result, level, aux2_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
+
+        while (gegl_buffer_iterator_next (i))
+          {
+            if (i->roi[0].height >= threads)
+            {
+            ThreadData thread_data[GEGL_MAX_THREADS];
+            gint pending = threads;
+            gint bit = i->roi[0].height / threads;
+            for (gint j = 0; j < threads; j++)
+            {
+              thread_data[j].roi.x = (i->roi[0]).x;
+              thread_data[j].roi.width = (i->roi[0]).width;
+              thread_data[j].roi.y = (i->roi[0]).y + bit * j;
+              thread_data[j].roi.height = bit;
+            }
+            thread_data[threads-1].roi.height = i->roi[0].height - (bit * (threads-1));
+            
+            for (gint j = 0; j < threads; j++)
+            {
+              thread_data[j].klass = point_composer3_class;
+              thread_data[j].operation = operation;
+              thread_data[j].input = input?((guchar*)i->data[read]) + (bit * j * i->roi[0].width * 
in_bpp):NULL;
+              thread_data[j].aux = aux?((guchar*)i->data[foo]) + (bit * j * i->roi[0].width * aux_bpp):NULL;
+              thread_data[j].aux2 = aux2?((guchar*)i->data[bar]) + (bit * j * i->roi[0].width * 
aux2_bpp):NULL;
+              thread_data[j].output = ((guchar*)i->data[0]) + (bit * j * i->roi[0].width * out_bpp);
+              thread_data[j].pending = &pending;
+              thread_data[j].level = level;
+              thread_data[j].success = TRUE;
+            }
+            pending = threads;
+
+            g_mutex_lock (&pool_mutex);
+
+            for (gint j = 0; j < threads; j++)
+              g_thread_pool_push (pool, &thread_data[j], NULL);
+
+            while (pending != 0)
+              g_cond_wait (&pool_cond, &pool_mutex);
+
+            g_mutex_unlock (&pool_mutex);
+
+            }
+            else
+            {
+              point_composer3_class->process (operation, input?i->data[read]:NULL,
+                                                         aux?i->data[foo]:NULL,
+                                                         aux2?i->data[bar]:NULL,
+                                                         i->data[0], i->length, &(i->roi[0]), level);
+            }
+          }
+
+        return TRUE;
+      }
+      else
+      {
+        GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, out_format, 
GEGL_BUFFER_WRITE, GEGL_ABYSS_NONE);
+        gint foo = 0, bar = 0, read = 0;
+
+        if (input)
+          read = gegl_buffer_iterator_add (i, input, result, level, in_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
+        if (aux)
+          foo = gegl_buffer_iterator_add (i, aux, result, level, aux_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
+        if (aux2)
+          bar = gegl_buffer_iterator_add (i, aux2, result, level, aux2_format, GEGL_BUFFER_READ, 
GEGL_ABYSS_NONE);
 
-      while (gegl_buffer_iterator_next (i))
-        {
-          point_composer3_class->process (operation, i->data[read], i->data[foo], i->data[bar], i->data[0], 
i->length, &(i->roi[0]), level);
-        }
+        while (gegl_buffer_iterator_next (i))
+          {
+            point_composer3_class->process (operation, input?i->data[read]:NULL,
+                                                       aux?i->data[foo]:NULL,
+                                                       aux2?i->data[bar]:NULL,
+                                                       i->data[0], i->length, &(i->roi[0]), level);
+          }
 
-      return TRUE;
+        return TRUE;
+      }
     }
   return TRUE;
 }