[gegl/wip/pippin/pipeline: 96/96] gegl: add pipeline for point operations

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

commit d74592ef15ad282fda09e7586b2b932bfdf66486
Author: Øyvind Kolås <pippin gimp org>
Date:   Mon Oct 8 20:15:07 2018 +0200

    gegl: add pipeline for point operations

 gegl/graph/gegl-node-private.h                  |  11 +
 gegl/graph/gegl-node.c                          |  98 ++++-
 gegl/graph/gegl-node.h                          |   5 +
 gegl/operation/Makefile.am                      |   2 +
 gegl/operation/gegl-operation-context-private.h |   3 +-
 gegl/operation/gegl-operation-context.c         |  13 +
 gegl/operation/gegl-operation-context.h         |   7 +
 gegl/operation/gegl-operation-pipeline.c        | 563 ++++++++++++++++++++++++
 gegl/operation/gegl-operation-pipeline.h        |  46 ++
 gegl/operation/gegl-operation-point-composer.c  |  64 ++-
 gegl/operation/gegl-operation-point-composer3.c |  61 ++-
 gegl/operation/gegl-operation-point-filter.c    |  80 +++-
 gegl/operation/gegl-operation-point-render.c    |   4 +-
 gegl/operation/gegl-operation.c                 |  36 ++
 gegl/operation/gegl-operation.h                 |   3 +
 15 files changed, 948 insertions(+), 48 deletions(-)
---
diff --git a/gegl/graph/gegl-node-private.h b/gegl/graph/gegl-node-private.h
index 5b892940c..cf86d67a5 100644
--- a/gegl/graph/gegl-node-private.h
+++ b/gegl/graph/gegl-node-private.h
@@ -142,6 +142,17 @@ gegl_node_emit_computed (GeglNode *node,
                          const GeglRectangle *rect);
 
 
+struct _GeglNodePrivate
+{
+  GSList          *source_connections;
+  GSList          *sink_connections;
+  GSList          *children;  /*  used for children */
+  GeglNode        *parent;
+  gchar           *name;
+  gchar           *debug_name;
+  GeglEvalManager *eval_manager;
+};
+
 G_END_DECLS
 
 #endif /* __GEGL_NODE_PRIVATE_H__ */
diff --git a/gegl/graph/gegl-node.c b/gegl/graph/gegl-node.c
index 220ce7476..26f040588 100644
--- a/gegl/graph/gegl-node.c
+++ b/gegl/graph/gegl-node.c
@@ -67,16 +67,6 @@ enum
 };
 
 
-struct _GeglNodePrivate
-{
-  GSList          *source_connections;
-  GSList          *sink_connections;
-  GSList          *children;  /*  used for children */
-  GeglNode        *parent;
-  gchar           *name;
-  gchar           *debug_name;
-  GeglEvalManager *eval_manager;
-};
 
 
 static guint gegl_node_signals[LAST_SIGNAL] = {0};
@@ -2362,3 +2352,91 @@ const char *gegl_operation_get_op_version (const char *op_name)
   return ret;
 }
 
+gint
+gegl_node_get_consumers2 (GeglNode      *node,
+                         const gchar   *output_pad,
+                         GeglNode    ***nodes,
+                         const gchar ***pads)
+{
+  GSList  *connections;
+  gint     n_connections;
+  GeglPad *pad;
+  gchar  **pasp = NULL;
+
+  g_return_val_if_fail (output_pad != NULL, 0);
+
+  if(node->is_graph)
+    node = gegl_node_get_input_proxy(node, "input");
+
+  g_return_val_if_fail (GEGL_IS_NODE (node), 0);
+
+  pad = gegl_node_get_pad (node, output_pad);
+
+  if (!pad)
+    {
+      g_warning ("%s: no such pad %s for %s",
+                 G_STRFUNC, output_pad, gegl_node_get_debug_name (node));
+      return 0;
+    }
+
+  connections = gegl_pad_get_connections (pad);
+  {
+    GSList *iter;
+    gint    pasp_size = 0;
+    gint    i;
+    gint    pasp_pos = 0;
+
+    n_connections = g_slist_length (connections);
+    pasp_size    += (n_connections + 1) * sizeof (gchar *);
+
+    for (iter = connections; iter; iter = g_slist_next (iter))
+      {
+        GeglConnection *connection = iter->data;
+        GeglPad        *pad        = gegl_connection_get_sink_pad (connection);
+        pasp_size += strlen (gegl_pad_get_name (pad)) + 1;
+      }
+    if (nodes)
+      *nodes = g_malloc ((n_connections + 1) * sizeof (void *));
+    if (pads)
+      {
+        pasp  = g_malloc (pasp_size);
+        *pads = (void *) pasp;
+      }
+    i        = 0;
+    pasp_pos = (n_connections + 1) * sizeof (void *);
+    for (iter = connections; iter; iter = g_slist_next (iter))
+      {
+        GeglConnection  *connection = iter->data;
+        GeglPad                 *pad        = gegl_connection_get_sink_pad (connection);
+        GeglNode        *node       = gegl_connection_get_sink_node (connection);
+        const gchar     *pad_name   = gegl_pad_get_name (pad);
+        const gchar     *name       = gegl_node_get_name(node);
+
+        gchar* proxy_name = g_strconcat("proxynop-", pad_name, NULL);
+        if(!strcmp(name, proxy_name))
+          {
+            node = g_object_get_data(G_OBJECT(node), "graph");
+            name = gegl_node_get_name(node);
+          }
+        else
+          {
+          }
+        g_free (proxy_name);
+
+        if (nodes)
+          (*nodes)[i] = node;
+        if (pasp)
+          {
+            pasp[i] = ((gchar *) pasp) + pasp_pos;
+            strcpy (pasp[i], pad_name);
+          }
+        pasp_pos += strlen (pad_name) + 1;
+        i++;
+      }
+    if (nodes)
+      (*nodes)[i] = NULL;
+    if (pads)
+      pasp[i] = NULL;
+  }
+  return n_connections;
+}
diff --git a/gegl/graph/gegl-node.h b/gegl/graph/gegl-node.h
index fdd803379..0b23f5374 100644
--- a/gegl/graph/gegl-node.h
+++ b/gegl/graph/gegl-node.h
@@ -691,6 +691,11 @@ gboolean       gegl_node_get_passthrough (GeglNode      *node);
 void           gegl_node_set_passthrough (GeglNode      *node,
                                           gboolean       passthrough);
 
+gint
+gegl_node_get_consumers2 (GeglNode      *node,
+                         const gchar   *output_pad,
+                         GeglNode    ***nodes,
+                         const gchar ***pads);
 
 void       gegl_node_progress (GeglNode *node, gdouble progress, gchar *message);
 
diff --git a/gegl/operation/Makefile.am b/gegl/operation/Makefile.am
index c9aaaf543..01ce0969e 100644
--- a/gegl/operation/Makefile.am
+++ b/gegl/operation/Makefile.am
@@ -61,6 +61,8 @@ liboperation_sources = \
        gegl-operation-temporal.c               \
        gegl-operation-context.c                \
        gegl-operation-context-private.h \
+       gegl-operation-pipeline.c        \
+       gegl-operation-pipeline.h        \
        gegl-operations.c                       \
        gegl-operations.h
 
diff --git a/gegl/operation/gegl-operation-context-private.h b/gegl/operation/gegl-operation-context-private.h
index b624c4756..bf7149233 100644
--- a/gegl/operation/gegl-operation-context-private.h
+++ b/gegl/operation/gegl-operation-context-private.h
@@ -23,7 +23,6 @@
 
 G_BEGIN_DECLS
 
-#include "gegl-operation.h"
 
 /**
  * When a node in a GEGL graph does processing, it needs context such
@@ -61,6 +60,8 @@ struct _GeglOperationContext
                                                                    2 = 1:4,
                                                                    4 = 1:8,
                                                                    6 = 1:16 .. */
+  GeglOperationPipeLine  *pipeline;
+
   GHashTable    *contexts;      /* to be able to look up the context of
                                    other nodes/ops in the graph we store the
                                    hashtable we will be stored in */
diff --git a/gegl/operation/gegl-operation-context.c b/gegl/operation/gegl-operation-context.c
index 59ba46992..4fd595469 100644
--- a/gegl/operation/gegl-operation-context.c
+++ b/gegl/operation/gegl-operation-context.c
@@ -413,6 +413,19 @@ gegl_operation_context_node_get_context (GeglOperationContext *context,
 }
 
 
+GeglOperationPipeLine *gegl_operation_context_get_pipeline (GeglOperationContext *context)
+{
+  if (!context) return NULL;
+  return context->pipeline;
+}
+void gegl_operation_context_set_pipeline (GeglOperationContext  *context,
+                                          GeglOperationPipeLine *pipeline)
+{
+  if (!context) return;
+  context->pipeline = pipeline;
+}
+
+
 GeglBuffer *
 gegl_operation_context_dup_input_maybe_copy (GeglOperationContext *context,
                                              const gchar          *padname,
diff --git a/gegl/operation/gegl-operation-context.h b/gegl/operation/gegl-operation-context.h
index 53af4736e..658c2d038 100644
--- a/gegl/operation/gegl-operation-context.h
+++ b/gegl/operation/gegl-operation-context.h
@@ -23,6 +23,8 @@
 
 G_BEGIN_DECLS
 
+typedef struct _GeglOperationPipeLine GeglOperationPipeLine;
+
 GeglBuffer     *gegl_operation_context_get_target      (GeglOperationContext *self,
                                                         const gchar          *padname);
 GeglBuffer     *gegl_operation_context_get_source      (GeglOperationContext *self,
@@ -57,6 +59,11 @@ GeglBuffer *    gegl_operation_context_dup_input_maybe_copy      (GeglOperationC
 GeglOperationContext *gegl_operation_context_node_get_context (GeglOperationContext *context,
                                                                GeglNode             *node);
 
+GeglOperationPipeLine *gegl_operation_context_get_pipeline (GeglOperationContext *context);
+
+void      gegl_operation_context_set_pipeline (GeglOperationContext  *context,
+                                               GeglOperationPipeLine *pipeline);
+
 
 G_END_DECLS
 
diff --git a/gegl/operation/gegl-operation-pipeline.c b/gegl/operation/gegl-operation-pipeline.c
new file mode 100644
index 000000000..86e3d967d
--- /dev/null
+++ b/gegl/operation/gegl-operation-pipeline.c
@@ -0,0 +1,563 @@
+#include "config.h"
+
+#include <glib-object.h>
+#define GEGL_ITERATOR2_API // opt in to new buffer iterator API
+
+#include "gegl.h"
+#include "gegl-debug.h"
+#include "gegl-operation-point-filter.h"
+#include "gegl-operation-point-composer.h"
+#include "gegl-operation-point-composer3.h"
+#include "gegl-operation-context.h"
+#include "gegl-operation-pipeline.h"
+#include "gegl-config.h"
+#include "gegl-types-internal.h"
+#include "gegl-buffer-private.h"
+#include "gegl-tile-storage.h"
+#include <gegl-node-private.h>
+#include <sys/types.h>
+#include <unistd.h>
+#include <string.h>
+#include <stdio.h>
+
+#define PIPELINE_MAX  64
+
+typedef struct {
+  GeglOperation *operation;
+  int            in_pads;
+  void         (*process)(void *);
+  GeglBuffer    *aux;
+  GeglBuffer    *aux2;
+  int            aux_handle;  /* these values are deterministic */
+  int            aux2_handle; /* and can scarily be shared among processes */
+  const Babl    *input_fish;
+  const Babl    *in_format;
+  const Babl    *aux_format;
+  const Babl    *aux2_format;
+  const Babl    *out_format;
+} PipeEntry;
+
+
+/* first make it work for a filter op masking its own output as
+ * a single entry pipeline
+ *
+ * then generalize/expand
+ */
+
+
+struct _GeglOperationPipeLine {
+  int entries;
+  GeglBuffer *input;
+  int buffers_used;
+  PipeEntry entry[PIPELINE_MAX];
+};
+
+/* returns true if the passed op could be part of a pipeline
+ */
+gboolean gegl_operation_is_pipelinable (GeglOperation *op)
+{
+  gboolean ret = GEGL_IS_OPERATION_POINT_FILTER (op) ||
+                 GEGL_IS_OPERATION_POINT_COMPOSER (op) ||
+                 GEGL_IS_OPERATION_POINT_COMPOSER3 (op);
+  if (ret)
+  {
+    GeglOperationClass *op_klass = GEGL_OPERATION_GET_CLASS (op);
+    if (op_klass->want_in_place == FALSE)
+      return FALSE;
+  }
+  if (0 && ret)
+  {
+    const char *name = gegl_operation_get_name (op);
+    if (!strcmp (name, "gimp:mask-components"))
+      return FALSE;
+  }
+  return ret;
+}
+
+static GeglNode *gegl_node_get_non_nop_producer (GeglNode *n)
+{
+  GeglNode *node = gegl_operation_get_source_node (n->operation, "input");
+  gint n_consumers;
+  fprintf (stderr, "%s.\n", gegl_node_get_operation (n));
+  n_consumers = gegl_node_get_consumers (node, "output", NULL, NULL);
+  fprintf (stderr, ".%s %i.\n", node?gegl_node_get_operation (node):"-", n_consumers);
+  while (node &&
+   (!strcmp (gegl_node_get_operation (node), "gegl:nop") ||
+   !strcmp (gegl_node_get_operation (node), "GraphNode"))
+   && (n_consumers == 1 /*|| n_consumers ==2*/) && !
+   node->priv->eval_manager)
+  {
+    node = gegl_operation_get_source_node (node->operation, "input");
+    n_consumers = gegl_node_get_consumers (node, "output", NULL, NULL);
+  }
+  if (n_consumers != 1 /*||n_consumers == 2*/)
+    node = NULL;
+  fprintf (stderr, "..%s %i\n", node?gegl_node_get_operation (node):"-", n_consumers);
+  return node;
+}
+
+GeglOperationPipeLine *
+gegl_operation_pipeline_ensure (GeglOperation        *operation,
+                                GeglOperationContext *context,
+                                GeglBuffer           *input)
+{
+  GeglOperationPipeLine *pipeline = NULL;
+  GeglNode *source_node;
+  GeglOperationContext *source_context;
+  source_node = gegl_node_get_non_nop_producer (operation->node);
+  if (source_node && !source_node->priv->eval_manager)
+  {
+    source_context = gegl_operation_context_node_get_context (context, source_node);
+    pipeline = gegl_operation_context_get_pipeline (source_context);
+    gegl_operation_context_set_pipeline (source_context, NULL);
+  }
+
+  if (!pipeline)
+  {
+    pipeline = g_malloc0 (sizeof (GeglOperationPipeLine));
+    pipeline->buffers_used = 2; // input and output
+  }
+  gegl_operation_context_set_pipeline (context, pipeline);
+
+  if (gegl_operation_pipeline_get_entries (pipeline) == 0)
+    gegl_operation_pipeline_set_input (pipeline, input);
+
+  return pipeline;
+}
+
+
+/* we should not be intermediate if adding more bits to the pipeline
+   would not work
+ */
+gboolean
+gegl_operation_pipeline_is_intermediate_node (GeglOperation *op,
+                                              GeglOperationPipeLine *pipeline)
+{
+  gboolean is_intermediate = TRUE;
+  gint n_consumers;
+  GeglNode *it = op->node;
+  GeglNode **consumers = NULL;
+
+  fprintf (stderr, "[%s\n", gegl_operation_get_name (op));
+  if (op->node->priv->eval_manager)
+  {
+    fprintf (stderr, "chaughta a fraggel!\n");
+    return FALSE;
+  }
+
+  n_consumers = gegl_node_get_consumers2 (it, "output", &consumers, NULL);
+
+  if (n_consumers == 0)
+    {
+      fprintf (stderr, "[[--\n");
+      is_intermediate = FALSE;
+    }
+
+  it = consumers[0];
+#if 1
+  while ((n_consumers == 1 /* || n_consumers==2*/)&& 
+       (!strcmp (gegl_node_get_operation (consumers[0]), "gegl:nop")||
+        !strcmp (gegl_node_get_operation (consumers[0]), "GraphNode")))
+  {
+    it = consumers[0];
+    g_free (consumers);
+    n_consumers = gegl_node_get_consumers2 (it, "output", &consumers, NULL);
+  }
+#endif
+
+  if (n_consumers == 0 && it != op->node)
+    {
+      return TRUE;
+    }
+  else if (n_consumers == 1)
+    {
+      GeglOperation *sink = gegl_node_get_gegl_operation (consumers[0]);
+
+      if (! gegl_operation_is_pipelinable (sink))
+          is_intermediate = FALSE;
+      else if (pipeline->entries + 1 >= PIPELINE_MAX)
+          is_intermediate = FALSE;
+      else
+          is_intermediate = TRUE;
+      fprintf (stderr, "[[%s %s\n", gegl_operation_get_name (sink), is_intermediate?"pipelineable":"not 
pipelineable");
+    }
+  else
+   {
+    is_intermediate = FALSE;
+      fprintf (stderr, "[%s=--%i-\n",
+         gegl_operation_get_name (consumers[0]->operation), n_consumers);
+   }
+
+  g_free (consumers);
+
+  return is_intermediate;
+}
+
+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);
+  gint input_handle = 0;
+  void *temp[2]={NULL, NULL};
+  glong high_tide = 0;
+  void *cur_input = NULL;
+  void *cur_output = NULL;
+  int   buf_mod = 0;
+
+  if (pipeline->input)
+    input_handle = gegl_buffer_iterator_add (i, pipeline->input, result, level,
+                                            pipeline->entry[0].in_format,
+                                            GEGL_ACCESS_READ, GEGL_ABYSS_NONE);
+
+  {
+    gint e;
+    for (e = 0; e < pipeline->entries; e++)
+    {
+      PipeEntry *entry = &pipeline->entry[e];
+      switch (pipeline->entry[e].in_pads)
+      {
+        case 3:
+          if (entry->aux2)
+            entry->aux2_handle =
+               gegl_buffer_iterator_add (i, entry->aux2, result, level,
+                                         entry->aux2_format,
+                                         GEGL_ACCESS_READ, GEGL_ABYSS_NONE);
+        case 2:
+          if (entry->aux)
+            entry->aux_handle =
+               gegl_buffer_iterator_add (i, entry->aux, result, level,
+                                         entry->aux_format,
+                                         GEGL_ACCESS_READ, GEGL_ABYSS_NONE);
+        case 1:
+        default:
+          break;
+      }
+    }
+  }
+
+  while (gegl_buffer_iterator_next (i))
+    {
+      gint e;
+      if (i->length > high_tide)
+      {
+        if (temp[0])
+          g_free (temp[0]);
+        high_tide = i->length;
+        temp[0] = g_malloc (4 * 8 * high_tide * 2);
+        temp[1] = (char*)(temp[0]) + 4 * 8 * high_tide;
+        /* the two allocations are merged into one, to reduce overhead,
+           if sufficiently small for the architecture - this could use
+           stack allocations
+         */
+      }
+
+      for (e = 0; e < pipeline->entries; e++)
+      {
+        PipeEntry *entry = &pipeline->entry[e];
+        if (e == 0)
+          {
+            if (pipeline->input)
+              cur_input = i->items[input_handle].data;
+          }
+        else
+          {
+            if (!entry->input_fish)
+            {
+              cur_input = cur_output;
+            }
+            else
+            {
+              cur_input = temp[(buf_mod++)&1];
+              babl_process (entry->input_fish,
+                            cur_output, cur_input, i->length);
+
+            }
+          }
+        if (pipeline->entries == e+1)
+        {
+          cur_output = i->items[0].data;
+        }
+        else
+        {
+          if (entry->in_format == entry->out_format)
+            cur_output = cur_input;
+          else
+            cur_output = temp[(buf_mod++)&1];
+        }
+
+        switch (entry->in_pads)
+        {
+          case 0:
+          {
+            gboolean (*process)(GeglOperation *,
+                                void *,
+                                glong,
+                                const GeglRectangle *,
+                                gint) = (void*)entry->process;
+
+            process (entry->operation,
+                     cur_output,
+                     i->length,
+                     &(i->items[0].roi),
+                     level);
+          }break;
+          case 1:
+          {
+            gboolean (*process)(GeglOperation *,
+                                void *,
+                                void *,
+                                glong,
+                                const GeglRectangle *,
+                                gint) = (void*)entry->process;
+
+            process (entry->operation,
+                     cur_input,
+                     cur_output,
+                     i->length,
+                     &(i->items[0].roi),
+                     level);
+          }break;
+
+          case 2:
+          {
+            gboolean (*process)(GeglOperation *,
+                                void *,
+                                void *,
+                                void *,
+                                glong,
+                                const GeglRectangle *,
+                                gint) = (void*)entry->process;
+
+            process (entry->operation,
+                     cur_input,
+                     entry->aux?
+                       i->items[entry->aux_handle].data:
+                       NULL,
+                     cur_output,
+                     i->length,
+                     &(i->items[0].roi),
+                     level);
+          }break;
+
+          case 3:
+          {
+            gboolean (*process)(GeglOperation *,
+                                void *,
+                                void *,
+                                void *,
+                                void *,
+                                glong,
+                                const GeglRectangle *,
+                                gint) = (void*)entry->process;
+            process (entry->operation,
+                     cur_input,
+                     entry->aux?
+                       i->items[entry->aux_handle].data:
+                       NULL,
+                     entry->aux2?
+                       i->items[entry->aux2_handle].data:
+                       NULL,
+                     cur_output,
+                     i->length,
+                     &(i->items[0].roi),
+                     level);
+          }break;
+        }
+      }
+    }
+
+
+  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++)
+    {
+      PipeEntry *entry = &pipeline->entry[e];
+
+      switch (entry->in_pads)
+      {
+        case 3:
+          if (entry->aux2)
+            g_clear_object (&entry->aux2);
+        case 2:
+          if (entry->aux)
+            g_clear_object (&entry->aux);
+        case 1:
+        default:
+          break;
+      }
+    }
+  }
+
+  g_free (pipeline);
+}
+
+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 (GeglOperationPipeLine *pipeline,
+                                 GeglBuffer            *output,
+                                 const GeglRectangle   *result,
+                                 gint                   level)
+{
+  gint threads = gegl_config_threads();
+
+  if (1) {
+    gint e;
+    fprintf (stderr, "{%i}", pipeline->entries);
+    for (e = 0; e < pipeline->entries; e++)
+    {
+      PipeEntry *entry = &pipeline->entry[e];
+      fprintf (stderr, "%s ", gegl_operation_get_name (entry->operation));
+  }
+  fprintf (stderr, "\n");
+  }
+
+  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
+gegl_operation_pipeline_add (GeglOperationPipeLine      *pipeline,
+                             GeglOperation *operation,
+                             gint           in_pads,
+                             const Babl    *in_format,
+                             const Babl    *out_format,
+                             const Babl    *aux_format,
+                             const Babl    *aux2_format,
+                             GeglBuffer    *aux,
+                             GeglBuffer    *aux2,
+                             void *process)
+{
+  PipeEntry *entry;
+  PipeEntry *prev_entry;
+
+  g_assert (pipeline);
+  g_assert (pipeline->entries + 1 <= PIPELINE_MAX);
+
+  entry = &pipeline->entry[pipeline->entries];
+  prev_entry = pipeline->entries?&pipeline->entry[pipeline->entries-1]:NULL;
+
+  entry->operation  = operation;
+  entry->in_pads    = in_pads;
+  entry->in_format  = in_format;
+  entry->aux_format = aux_format;
+  entry->aux        = aux;
+  entry->aux2_format= aux2_format;
+  entry->aux2       = aux2;
+  entry->out_format = out_format;
+  entry->process    = process;
+
+  if (prev_entry && entry->in_format != prev_entry->out_format)
+  {
+    entry->input_fish = babl_fish (prev_entry->out_format, entry->in_format);
+  }
+
+  pipeline->entries++;
+
+  pipeline->buffers_used += ( (aux?1:0) + (aux2?1:0));
+}
+
+int gegl_operation_pipeline_get_entries (GeglOperationPipeLine *pipeline)
+{
+  return pipeline->entries;
+}
+
+
+
+void gegl_operation_pipeline_set_input  (GeglOperationPipeLine   *pipeline,
+                                         GeglBuffer *buffer)
+{
+  pipeline->input = buffer;
+}
diff --git a/gegl/operation/gegl-operation-pipeline.h b/gegl/operation/gegl-operation-pipeline.h
new file mode 100644
index 000000000..9f0835c94
--- /dev/null
+++ b/gegl/operation/gegl-operation-pipeline.h
@@ -0,0 +1,46 @@
+#ifndef _GEGL_OPERATION_PIPELINE_H
+#define _GEGL_OPERATION_PIPELINE_H
+
+
+int gegl_operation_pipeline_get_entries (GeglOperationPipeLine   *pipeline);
+void gegl_operation_pipeline_set_input  (GeglOperationPipeLine   *pipeline,
+                                         GeglBuffer *buffer);
+
+gboolean
+gegl_operation_pipeline_is_intermediate_node (GeglOperation *op,
+                                              GeglOperationPipeLine *pipeline);
+gboolean gegl_operation_pipeline_is_composite_node (GeglOperation *op);
+
+gboolean gegl_operation_pipeline_process (GeglOperationPipeLine            *pipeline,
+                                          GeglBuffer          *output,
+                                          const GeglRectangle *result,
+                                          gint                 level);
+GeglOperationPipeLine *
+gegl_operation_pipeline_ensure (GeglOperation *operation,
+                                GeglOperationContext *context,
+                                GeglBuffer           *input);
+
+
+void
+gegl_operation_pipeline_add (GeglOperationPipeLine *pipeline,
+                             GeglOperation *operation,
+                             gint           in_pads,
+                             const Babl    *in_format,
+                             const Babl    *out_format,
+                             const Babl    *aux_format,
+                             const Babl    *aux2_format,
+                             GeglBuffer    *aux,
+                             GeglBuffer    *aux2,
+                             void *process);
+
+gboolean
+gegl_operation_pipeline_process_threaded  (GeglOperationPipeLine            *pipeline,
+                                           GeglBuffer          *output,
+                                           const GeglRectangle *result,
+                                           gint                 level);
+
+gboolean gegl_operation_is_pipelinable (GeglOperation *op);
+void
+gegl_operation_pipeline_destroy (GeglOperationPipeLine *pipeline);
+
+#endif
diff --git a/gegl/operation/gegl-operation-point-composer.c b/gegl/operation/gegl-operation-point-composer.c
index bacc14688..06ff4ecb9 100644
--- a/gegl/operation/gegl-operation-point-composer.c
+++ b/gegl/operation/gegl-operation-point-composer.c
@@ -17,16 +17,16 @@
  */
 
 
+#define GEGL_ITERATOR2_API
 #include "config.h"
 
 #include <glib-object.h>
 
-#define GEGL_ITERATOR2_API
-
 #include "gegl.h"
 #include "gegl-debug.h"
 #include "gegl-operation-point-composer.h"
 #include "gegl-operation-context.h"
+#include "gegl-operation-pipeline.h"
 #include "gegl-config.h"
 #include "gegl-types-internal.h"
 #include "gegl-buffer-private.h"
@@ -66,8 +66,7 @@ static void thread_process (gpointer thread_data, gpointer unused)
                                                     data->level,
                                                     data->output_format,
                                                     GEGL_ACCESS_WRITE,
-                                                    GEGL_ABYSS_NONE,
-                                                    4);
+                                                    GEGL_ABYSS_NONE, 3);
 
   if (data->input)
     read = gegl_buffer_iterator_add (i, data->input, &data->result, data->level,
@@ -108,10 +107,12 @@ gegl_operation_composer_process (GeglOperation        *operation,
                                  gint                  level)
 {
   GeglOperationComposerClass *klass   = GEGL_OPERATION_COMPOSER_GET_CLASS (operation);
+  GeglOperationPointComposerClass *point_composer_class = GEGL_OPERATION_POINT_COMPOSER_GET_CLASS 
(operation);
   GeglBuffer                 *input;
   GeglBuffer                 *aux;
   GeglBuffer                 *output;
   gboolean                    success = FALSE;
+  GeglOperationPipeLine      *pipeline;
 
   GeglRectangle scaled_result = *result;
   if (level)
@@ -136,13 +137,56 @@ gegl_operation_composer_process (GeglOperation        *operation,
   }
 
   input  = (GeglBuffer*) gegl_operation_context_dup_object (context, "input");
-  output = gegl_operation_context_get_output_maybe_in_place (operation,
-                                                             context,
-                                                             input,
-                                                             result);
-
   aux   = (GeglBuffer*) gegl_operation_context_dup_object (context, "aux");
 
+  if (!input && !aux)
+    return FALSE;
+
+  if (gegl_operation_is_pipelinable (operation))
+  {
+    pipeline = gegl_operation_pipeline_ensure (operation, context, input);
+
+    gegl_operation_pipeline_add (pipeline, operation, 2,
+         gegl_operation_get_format (operation, "input"),
+         gegl_operation_get_format (operation, "output"),
+         gegl_operation_get_format (operation, "aux"),
+         NULL,
+         aux, NULL,
+         point_composer_class->process);
+
+    if (gegl_operation_pipeline_is_intermediate_node (operation, pipeline))
+    {
+      gegl_operation_context_take_object (context, "output", G_OBJECT (input));
+
+      return TRUE;
+    }
+
+    output = gegl_operation_context_get_output_maybe_in_place (operation,
+                                                               context,
+                                                               input,
+                                                               result);
+
+    gegl_operation_context_set_pipeline (context, NULL);
+    if (gegl_operation_pipeline_get_entries (pipeline) > 1)
+    {
+      gegl_operation_pipeline_process (pipeline, output, result, level);
+      gegl_operation_pipeline_destroy (pipeline);
+      return TRUE;
+    }
+    g_object_ref (input);
+    if (aux)
+      g_object_ref (aux);
+    gegl_operation_pipeline_destroy (pipeline);
+
+  }
+  else
+  {
+    output = gegl_operation_context_get_output_maybe_in_place (operation,
+                                                               context,
+                                                               input,
+                                                               result);
+  }
+
   /* A composer with a NULL aux, can still be valid, the
    * subclass has to handle it.
    */
@@ -388,7 +432,7 @@ gegl_operation_point_composer_process (GeglOperation       *operation,
       }
       else
       {
-        GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, out_format, 
GEGL_ACCESS_WRITE, GEGL_ABYSS_NONE, 4);
+        GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, out_format, 
GEGL_ACCESS_WRITE, GEGL_ABYSS_NONE, 3);
         gint foo = 0, read = 0;
 
         if (input)
diff --git a/gegl/operation/gegl-operation-point-composer3.c b/gegl/operation/gegl-operation-point-composer3.c
index ebfcd22a5..492ea3f11 100644
--- a/gegl/operation/gegl-operation-point-composer3.c
+++ b/gegl/operation/gegl-operation-point-composer3.c
@@ -20,11 +20,12 @@
 #include "config.h"
 
 #include <glib-object.h>
-#define GEGL_ITERATOR2_API
+#define GEGL_ITERATOR2_API // opt in to new buffer iterator API
 
 #include "gegl.h"
 #include "gegl-operation-point-composer3.h"
 #include "gegl-operation-context.h"
+#include "gegl-operation-pipeline.h"
 #include "gegl-types-internal.h"
 #include "gegl-config.h"
 #include "gegl-buffer-private.h"
@@ -111,11 +112,13 @@ gegl_operation_composer3_process (GeglOperation        *operation,
                                   gint                  level)
 {
   GeglOperationComposer3Class *klass   = GEGL_OPERATION_COMPOSER3_GET_CLASS (operation);
+  GeglOperationPointComposer3Class *point_composer3_class = GEGL_OPERATION_POINT_COMPOSER3_GET_CLASS 
(operation);
   GeglBuffer                  *input;
   GeglBuffer                  *aux;
   GeglBuffer                  *aux2;
   GeglBuffer                  *output;
   gboolean                     success = FALSE;
+  GeglOperationPipeLine       *pipeline;
 
   if (strcmp (output_prop, "output"))
     {
@@ -130,13 +133,57 @@ gegl_operation_composer3_process (GeglOperation        *operation,
   }
 
   input  = (GeglBuffer*) gegl_operation_context_dup_object (context, "input");
-  output = gegl_operation_context_get_output_maybe_in_place (operation,
-                                                             context,
-                                                             input,
-                                                             result);
-
   aux   = (GeglBuffer*) gegl_operation_context_dup_object (context, "aux");
-  aux2  = (GeglBuffer*) gegl_operation_context_dup_object (context, "aux2");
+  aux2   = (GeglBuffer*) gegl_operation_context_dup_object (context, "aux2");
+
+  if (!input && !aux && !aux2)
+    return FALSE;
+
+  if (gegl_operation_is_pipelinable (operation))
+  {
+    pipeline = gegl_operation_pipeline_ensure (operation, context, input);
+
+    gegl_operation_pipeline_add (pipeline, operation, 3,
+       gegl_operation_get_format (operation, "input"),
+       gegl_operation_get_format (operation, "output"),
+       gegl_operation_get_format (operation, "aux"),
+       gegl_operation_get_format (operation, "aux2"),
+       aux, aux2,
+       point_composer3_class->process);
+
+    if (gegl_operation_pipeline_is_intermediate_node (operation, pipeline))
+    {
+      gegl_operation_context_take_object (context, "output", G_OBJECT (input));
+
+      return TRUE;
+    }
+
+    output = gegl_operation_context_get_output_maybe_in_place (operation,
+                                                                context,
+                                                                input,
+                                                                result);
+    gegl_operation_context_set_pipeline (context, NULL);
+
+    if (gegl_operation_pipeline_get_entries (pipeline) > 1)
+    {
+      gegl_operation_pipeline_process (pipeline, output, result, level);
+      gegl_operation_pipeline_destroy (pipeline);
+      return TRUE;
+    }
+    g_object_ref (input);
+    if (aux)
+      g_object_ref (aux);
+    if (aux2)
+      g_object_ref (aux2);
+    gegl_operation_pipeline_destroy (pipeline);
+  }
+  else
+  {
+    output = gegl_operation_context_get_output_maybe_in_place (operation,
+                                                               context,
+                                                               input,
+                                                               result);
+  }
 
   /* A composer with a NULL aux, can still be valid, the
    * subclass has to handle it.
diff --git a/gegl/operation/gegl-operation-point-filter.c b/gegl/operation/gegl-operation-point-filter.c
index 6fffa74a1..1ce1f7133 100644
--- a/gegl/operation/gegl-operation-point-filter.c
+++ b/gegl/operation/gegl-operation-point-filter.c
@@ -19,13 +19,14 @@
 
 #include "config.h"
 
+#define GEGL_ITERATOR2_API
 #include <glib-object.h>
 
-#define GEGL_ITERATOR2_API
 #include "gegl.h"
 #include "gegl-debug.h"
 #include "gegl-operation-point-filter.h"
 #include "gegl-operation-context.h"
+#include "gegl-operation-pipeline.h"
 #include "gegl-config.h"
 #include "gegl-types-internal.h"
 #include "gegl-buffer-private.h"
@@ -60,7 +61,7 @@ static void thread_process (gpointer thread_data, gpointer unused)
                                                     data->level,
                                                     data->output_format,
                                                     GEGL_ACCESS_WRITE,
-                                                    GEGL_ABYSS_NONE, 4);
+                                                    GEGL_ABYSS_NONE, 2);
   gint read = 0;
   if (data->input)
     read = gegl_buffer_iterator_add (i, data->input, &data->result, data->level,
@@ -88,17 +89,20 @@ static GThreadPool *thread_pool (void)
   return pool;
 }
 
+
 static gboolean
 gegl_operation_filter_process (GeglOperation        *operation,
-                                 GeglOperationContext *context,
-                                 const gchar          *output_prop,
-                                 const GeglRectangle  *result,
-                                 gint                  level)
+                               GeglOperationContext *context,
+                               const gchar          *output_prop,
+                               const GeglRectangle  *result,
+                               gint                  level)
 {
   GeglOperationFilterClass *klass    = GEGL_OPERATION_FILTER_GET_CLASS (operation);
+  GeglOperationPointFilterClass *point_filter_class = GEGL_OPERATION_POINT_FILTER_GET_CLASS (operation);
   GeglBuffer                 *input;
   GeglBuffer                 *output;
   gboolean                    success = FALSE;
+  GeglOperationPipeLine      *pipeline;
 
   GeglRectangle scaled_result = *result;
   if (level)
@@ -123,10 +127,51 @@ gegl_operation_filter_process (GeglOperation        *operation,
   }
 
   input  = (GeglBuffer*)gegl_operation_context_dup_object (context, "input");
-  output = gegl_operation_context_get_output_maybe_in_place (operation,
-                                                             context,
-                                                             input,
-                                                             result);
+  if (!input)
+  {
+    return FALSE;
+  }
+
+
+  if (gegl_operation_is_pipelinable (operation))
+  {
+    pipeline = gegl_operation_pipeline_ensure (operation, context, input);
+
+    gegl_operation_pipeline_add (pipeline, operation, 1,
+         gegl_operation_get_format (operation, "input"),
+         gegl_operation_get_format (operation, "output"),
+         NULL, NULL, NULL, NULL, // auxes are not set
+         point_filter_class->process);
+
+    if (gegl_operation_pipeline_is_intermediate_node (operation, pipeline))
+    {
+      gegl_operation_context_take_object (context, "output", G_OBJECT (input));
+
+      return TRUE;
+    }
+
+    output = gegl_operation_context_get_output_maybe_in_place (operation,
+                                                               context,
+                                                               input,
+                                                               result);
+
+    gegl_operation_context_set_pipeline (context, NULL);
+    if (gegl_operation_pipeline_get_entries (pipeline) > 1)
+    {
+      gegl_operation_pipeline_process (pipeline, output, result, level);
+      gegl_operation_pipeline_destroy (pipeline);
+      return TRUE;
+    }
+    g_object_ref (input);
+    gegl_operation_pipeline_destroy (pipeline);
+  }
+  else
+  {
+    output = gegl_operation_context_get_output_maybe_in_place (operation,
+                                                               context,
+                                                               input,
+                                                               result);
+  }
 
   if (input != NULL)
     {
@@ -248,13 +293,13 @@ error:
 static void
 gegl_operation_point_filter_class_init (GeglOperationPointFilterClass *klass)
 {
-  GeglOperationClass          *operation_class = GEGL_OPERATION_CLASS (klass);
-  GeglOperationFilterClass *filter_class  = GEGL_OPERATION_FILTER_CLASS (klass);
+  GeglOperationClass       *operation_class = GEGL_OPERATION_CLASS (klass);
+  GeglOperationFilterClass *filter_class = GEGL_OPERATION_FILTER_CLASS (klass);
 
-  filter_class->process = gegl_operation_point_filter_process;
-  operation_class->process = gegl_operation_filter_process;
-  operation_class->prepare = prepare;
-  operation_class->no_cache =TRUE;
+  filter_class->process          = gegl_operation_point_filter_process;
+  operation_class->process       = gegl_operation_filter_process;
+  operation_class->prepare       = prepare;
+  operation_class->no_cache      = TRUE;
   operation_class->want_in_place = TRUE;
   operation_class->threaded = TRUE;
 }
@@ -277,7 +322,6 @@ gegl_operation_point_filter_process (GeglOperation       *operation,
   const Babl *in_format   = gegl_operation_get_format (operation, "input");
   const Babl *out_format  = gegl_operation_get_format (operation, "output");
 
-
   if ((result->width > 0) && (result->height > 0))
     {
       if (gegl_operation_use_opencl (operation) && (operation_class->cl_data || 
point_filter_class->cl_process))
@@ -345,7 +389,7 @@ gegl_operation_point_filter_process (GeglOperation       *operation,
       else
       {
         GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, out_format,
-                                                          GEGL_ACCESS_WRITE, GEGL_ABYSS_NONE, 4);
+                                                          GEGL_ACCESS_WRITE, GEGL_ABYSS_NONE, 2);
         gint read = 0;
 
         if (input)
diff --git a/gegl/operation/gegl-operation-point-render.c b/gegl/operation/gegl-operation-point-render.c
index 7c40508ac..2514a8ace 100644
--- a/gegl/operation/gegl-operation-point-render.c
+++ b/gegl/operation/gegl-operation-point-render.c
@@ -20,8 +20,8 @@
 #include "config.h"
 
 #include <glib-object.h>
-
 #define GEGL_ITERATOR2_API
+
 #include "gegl.h"
 #include "gegl-operation-point-render.h"
 #include "gegl-operation-context.h"
@@ -101,7 +101,7 @@ gegl_operation_point_render_process (GeglOperation       *operation,
   if ((result->width > 0) && (result->height > 0))
     {
       GeglBufferIterator *i = gegl_buffer_iterator_new (output, result, level, out_format,
-                                                        GEGL_ACCESS_WRITE, GEGL_ABYSS_NONE, 2);
+                                                        GEGL_ACCESS_WRITE, GEGL_ABYSS_NONE, 1);
 
       while (gegl_buffer_iterator_next (i))
           point_render_class->process (operation, i->items[0].data, i->length, &i->items[0].roi, level);
diff --git a/gegl/operation/gegl-operation.c b/gegl/operation/gegl-operation.c
index de7bc9ccd..956977f88 100644
--- a/gegl/operation/gegl-operation.c
+++ b/gegl/operation/gegl-operation.c
@@ -381,6 +381,42 @@ gegl_operation_get_source_node (GeglOperation *operation,
   return gegl_pad_get_node (pad);
 }
 
+GeglNode *
+gegl_operation_get_target_node (GeglOperation *operation,
+                                const gchar   *output_pad_name)
+{
+  GeglNode *node;
+  GeglPad *pad;
+
+  g_return_val_if_fail (GEGL_IS_OPERATION (operation), NULL);
+  g_return_val_if_fail (GEGL_IS_NODE (operation->node), NULL);
+  g_return_val_if_fail (output_pad_name != NULL, NULL);
+
+  node = operation->node;
+#if 0
+  if (node->is_graph)
+    {
+      node = gegl_node_get_output_proxy (node, output_pad_name);
+      output_pad_name = "input";
+    }
+#endif
+
+  pad = gegl_node_get_pad (node, output_pad_name);
+
+  if (!pad)
+    return NULL;
+
+  pad = gegl_pad_get_connected_to (pad);
+
+  if (!pad)
+    return NULL;
+
+  g_assert (gegl_pad_get_node (pad));
+
+  return gegl_pad_get_node (pad);
+}
+
+
 GeglRectangle *
 gegl_operation_source_get_bounding_box (GeglOperation *operation,
                                         const gchar   *input_pad_name)
diff --git a/gegl/operation/gegl-operation.h b/gegl/operation/gegl-operation.h
index a48e2adc9..c5a929d91 100644
--- a/gegl/operation/gegl-operation.h
+++ b/gegl/operation/gegl-operation.h
@@ -300,6 +300,9 @@ void       gegl_operation_progress (GeglOperation *operation, gdouble progress,
 
 const Babl *gegl_operation_get_source_space (GeglOperation *operation, const char *in_pad);
 
+GeglNode *
+gegl_operation_get_target_node (GeglOperation *operation,
+                                const gchar   *output_pad_name);
 
 G_END_DECLS