[gegl/wip/pippin/pipeline: 89/95] pipeline segfault fix

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

commit c45224dbfeba4d9ed1c0c868c44afdd15245708a
Author: Øyvind Kolås <pippin gimp org>
Date:   Wed Jul 25 21:23:30 2018 +0200

    pipeline segfault fix

 gegl/operation/gegl-operation-pipeline.c | 120 ++++++++++++++++++++++++++-----
 1 file changed, 104 insertions(+), 16 deletions(-)
---
diff --git a/gegl/operation/gegl-operation-pipeline.c b/gegl/operation/gegl-operation-pipeline.c
index 74b3beedc..521a028e6 100644
--- a/gegl/operation/gegl-operation-pipeline.c
+++ b/gegl/operation/gegl-operation-pipeline.c
@@ -20,7 +20,7 @@
 #include <string.h>
 #include <stdio.h>
 
-#define PIPELINE_MAX  32
+#define PIPELINE_MAX  64
 
 typedef struct {
   GeglOperation *operation;
@@ -77,7 +77,7 @@ static GeglNode *gegl_node_get_non_nop_producer (GeglNode *n)
 {
   // XXX: look out for forks of the data/
   GeglNode *node = gegl_node_get_producer (n, "input", NULL);
-  if (!strcmp (gegl_node_get_operation (node), "gegl:nop"))
+  if (node && !strcmp (gegl_node_get_operation (node), "gegl:nop"))
     return NULL;
   return node;
   while (node &&
@@ -178,11 +178,11 @@ gegl_operation_pipeline_is_intermediate_node (GeglOperation *op,
   return is_intermediate;
 }
 
-gboolean
-gegl_operation_pipeline_process (GeglOperationPipeLine            *pipeline,
-                                 GeglBuffer          *output,
-                                 const GeglRectangle *result,
-                                 gint                 level)
+static gboolean
+_gegl_operation_pipeline_process (GeglOperationPipeLine            *pipeline,
+                                  GeglBuffer          *output,
+                                  const GeglRectangle *result,
+                                  gint                 level)
 {
   GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, 
pipeline->entry[pipeline->entries-1].out_format,
                                                     GEGL_ACCESS_WRITE, GEGL_ABYSS_NONE, 
pipeline->buffers_used+1);
@@ -342,8 +342,18 @@ gegl_operation_pipeline_process (GeglOperationPipeLine            *pipeline,
         }
       }
     }
-  g_clear_object (&pipeline->input);
 
+
+  if (temp[0])
+    g_free (temp[0]);
+
+  return TRUE;
+}
+
+void
+gegl_operation_pipeline_destroy (GeglOperationPipeLine *pipeline)
+{
+  g_clear_object (&pipeline->input);
   {
     gint e;
     for (e = 0; e < pipeline->entries; e++)
@@ -365,20 +375,98 @@ gegl_operation_pipeline_process (GeglOperationPipeLine            *pipeline,
     }
   }
 
-  if (temp[0])
-    g_free (temp[0]);
+  g_free (pipeline);
+}
 
-  return TRUE;
+typedef struct ThreadData
+{
+  GeglOperationPipeLine *pipeline;
+  GeglBuffer            *output;
+  gint                  *pending;
+  gint                   level;
+  GeglRectangle          result;
+} ThreadData;
+
+static void thread_process (gpointer thread_data, gpointer unused)
+{
+  ThreadData *data = thread_data;
+  _gegl_operation_pipeline_process (data->pipeline, data->output, &data->result, data->level);
+  g_atomic_int_add (data->pending, -1);
 }
 
+static GThreadPool *thread_pool (void)
+{
+  static GThreadPool *pool = NULL;
+  if (!pool)
+    {
+      pool =  g_thread_pool_new (thread_process, NULL, gegl_config_threads (),
+                                 FALSE, NULL);
+    }
+  return pool;
+}
 
 gboolean
-gegl_operation_pipeline_process_threaded  (GeglOperationPipeLine            *pipeline,
-                                           GeglBuffer          *output,
-                                           const GeglRectangle *result,
-                                           gint                 level)
+gegl_operation_pipeline_process (GeglOperationPipeLine *pipeline,
+                                 GeglBuffer            *output,
+                                 const GeglRectangle   *result,
+                                 gint                   level)
 {
-  return gegl_operation_pipeline_process_threaded (pipeline, output, result, level);
+  gint threads = gegl_config_threads();
+  if (threads == 1 || result->width * result->height < 64*64 || result->height < threads)
+  {
+    return _gegl_operation_pipeline_process (pipeline, output, result, level);
+  }
+  else
+  {
+    ThreadData thread_data[threads];
+    GThreadPool *pool = thread_pool ();
+    gint pending;
+    gint j;
+    if (result->width > result->height)
+          for (j = 0; j < threads; j++)
+          {
+            GeglRectangle rect = *result;
+
+            rect.width /= threads;
+            rect.x += rect.width * j;
+
+            if (j == threads-1)
+              rect.width = (result->width + result->x) - rect.x;
+
+            thread_data[j].result = rect;
+          }
+        else
+          for (j = 0; j < threads; j++)
+          {
+            GeglRectangle rect = *result;
+
+            rect = *result;
+            rect.height /= threads;
+            rect.y += rect.height * j;
+
+            if (j == threads-1)
+              rect.height = (result->height + result->y) - rect.y;
+
+            thread_data[j].result = rect;
+          }
+    for (j = 0; j < threads; j++)
+     {
+       thread_data[j].output = output;
+       thread_data[j].pipeline = pipeline;
+       thread_data[j].pending = &pending;
+       thread_data[j].level = level;
+     }
+    pending = threads;
+
+    // XXX cl flushes?
+
+    for (gint j = 1; j < threads; j++)
+      g_thread_pool_push (pool, &thread_data[j], NULL);
+    thread_process (&thread_data[0], NULL);
+    while (g_atomic_int_get (&pending)) {};
+
+    return TRUE;
+  }
 }
 
 void