few front end fixes

* merged severap ops into machine learning
* optimization of flow package

browser/vue-flow/src/components/flow/editor.js

@@ -467,6 +467,7 @@ export default {
         y: y,
         defaultInputs: {},
         label: src,
+        color: this.registry[src].style && this.registry[src].style.color, /* NEW 12/02/2018 */
         src: src
       }
       // Setup Props

browser/vue-flow/src/components/flow/node-activity.vue

@@ -72,7 +72,7 @@ export default {
     }
   },
   watch: {
-    node (val, oldVal) {
+    nodeActivity (val, oldVal) {
       this.recalcStartTime()
     }
   },
@@ -86,6 +86,8 @@ export default {
   },
   methods: {
     recalcStartTime () {
+      this.startTime = null
+      this.finishTime = null
       const serverStartTime = new Date(Date.parse(this.nodeActivity.startTime))
       if (!utils.dateIsValid(serverStartTime)) { return null }
 

browser/vue-flow/src/components/flow/node.vue

@@ -191,6 +191,10 @@ export default {
     nodeActivity () {
       return this.activity && this.activity.nodes && this.activity.nodes[this.id]
     },
+    color () {
+      return this.node.color ||
+        (this.registry[this.node.src].style && this.registry[this.node.src].style.color)
+    },
     style () {
       return this.registry[this.node.src].style || {}
     },
@@ -224,8 +228,8 @@ export default {
         y: -height / 2,
         width: width,
         height: height,
-        stroke: this.style.color || '#777',
-        fill: this.style.color || '#777'
+        stroke: this.color || '#777',
+        fill: this.color || '#777'
       }
       if (this.style.shape === 'circle') {
         rect.r = Math.max(width / 2, height / 2)
@@ -315,7 +319,7 @@ export default {
   },
   watch: {
     // the only thing now that affects geometry
-    'label' () {
+    'node.label' () {
       this.$nextTick(() => {
         this.labelRect = this.$refs.label.getBBox()
       })

browser/vue-flow/src/components/flow/panel-inspector.vue

@@ -26,10 +26,9 @@
 
         <!-- DESCRIPTIONS -->
         <div class="flow-inspector__area flow-inspector--properties ">
+          <h2 class="property">{{ nodeInspect.src }}</h2>
           <label>ID</label>
           <div class="property">[{{ nodeInspect.id }}]</div>
-          <label>src</label>
-          <div class="property">{{ nodeInspect.src }}</div>
           <!--
           <label>Description</label>
           <div class="property">Bogus description</div>
@@ -37,8 +36,22 @@
           <div class="property">Connect to input a thing and goes to output another thing</div>
           -->
         </div>
+        <!-- STATIC PARAM -->
+        <div class="flow-inspector__area  flow-inspector--static">
+          <label>label</label>
+          <input ref="label" type="text" v-model="nodeInspect.label" @input="localChange">
+          <label>color  <small>*experimental*</small> </label>
+          <input type="text" v-model="nodeInspect.color" @input="localChange">
+          <div class="flow-inspector__param" v-for="(v,k) in nodeInspect.prop">
+            <label>{{ k }}</label>
+            <input
+              ref="props"
+              type="text"
+              v-model="nodeInspect.prop[k]" @input="localChange">
+          </div>
+        </div>
         <div class="flow-inspector__area flow-inspector--inputs">
-          <h3>Inputs defaults:</h3>
+          <h3>Inputs constants</h3>
           <div
             class="flow-inspector__param"
             v-for="(n,i) in registry[nodeInspect.src].inputs"
@@ -67,23 +80,6 @@
             <div class="property">{{ nodeActivity && nodeActivity.error }}</div>
           </div>
         </div>
-        <!-- PARAMETERS -->
-        <div class="flow-inspector--params flow-inspector__area" v-if="nodeInspect.prop" >
-          <div class="flow-inspector__param" v-for="(v,k) in nodeInspect.prop">
-            <label>{{ k }}</label>
-            <input
-              ref="props"
-              type="text"
-              v-model="nodeInspect.prop[k]" @input="localChange">
-          </div>
-
-        </div>
-
-        <!-- STATIC PARAM -->
-        <div class="flow-inspector__area  flow-inspector--static">
-          <label>label</label>
-          <input ref="label" type="text" v-model="nodeInspect.label" @input="localChange">
-        </div>
       </div><!-- /container -->
 
       <div class="flow-inspector__area flow-inspector--control">
@@ -162,11 +158,16 @@ export default {
 }
 
 .flow-inspector__area{
-  flex:1;
+  flex:0;
   padding: 10px 0;
   border-bottom: solid 1px rgba(150,150,150,0.3);
 }
 
+.flow-inspector__area h2{
+  font-size:16px;
+  padding-bottom:15px;
+}
+
 .flow-inspector__area h3{
   font-size:14px;
   color: var(--primary-lighter);
@@ -181,6 +182,11 @@ export default {
   color: var(--primary-darker);
 }
 
+.flow-inspector__area small {
+  font-size:9px;
+  color: var(--normal);
+}
+
 .flow-inspector--view {
   flex-basis:150px;
   flex-shrink: 0;
@@ -189,7 +195,10 @@ export default {
 .flow-inspector__area .property {
   white-space: normal;
   word-wrap: break-word;
+}
 
+.flow-inspector--activity {
+  flex:1;
 }
 
 .flow-inspector--properties-error {

browser/vue-flow/src/store/flow/default-registry.js

@@ -5,17 +5,26 @@ export default{
     style: { color: '#686', shape: 'circle' },
     props: {'input': ''} // should be sent in the node
   },
-  'Variable': {
+  'Output': {
+    categories: ['core'],
+    inputs: [{type: 'interface {}'}],
+    style: { color: '#111', shape: 'circle' }
+  },
+  'Var': {
     categories: ['core'],
     inputs: [{type: 'interface {}', name: 'initial'}],
     output: {type: 'interface {}'},
-    style: { color: '#88a', shape: 'circle' }
+    style: { color: '#88a', shape: 'circle' },
+    props: {'variable name': ''}
   },
-  'Output': {
+  'SetVar': {
     categories: ['core'],
-    inputs: [{type: 'interface {}'}],
-    style: { color: '#111', shape: 'circle' }
+    inputs: [{type: 'interface {}', name: 'value'}],
+    output: {type: 'interface{}'},
+    style: { color: '#88a', shape: 'circle' },
+    props: {'variable name': ''}
   },
+
   'Notify': {
     categories: ['flow-web'],
     inputs: [{type: 'interface {}'}, {type: 'string', name: 'msg'}],

go/src/demos/cmd/demo1/main.go

@@ -4,7 +4,10 @@ import (
 	"demos/cmd/demo1/assets"
 	"demos/ops/defaultops"
 	"demos/ops/devops"
+	"demos/ops/genericops"
 	"demos/ops/ml"
+	"demos/ops/stringops"
+	"demos/ops/webops"
 	"flow/flowserver"
 	"log"
 	"mime"
@@ -32,11 +35,21 @@ func main() {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", assetFunc)
 
+	defops := defaultops.New()
+	defops.Merge(genericops.New())
+	defops.Merge(stringops.New())
+
 	mux.Handle("/default/", c.Build(
-		http.StripPrefix("/default", flowserver.New(defaultops.New(), "default")),
+		http.StripPrefix("/default", flowserver.New(defops, "default")),
 	))
+
+	mlReg := ml.New()
+	mlReg.Merge(genericops.New())
+	mlReg.Merge(stringops.New())
+	mlReg.Merge(webops.New())
+
 	mux.Handle("/machinelearning/", c.Build(
-		http.StripPrefix("/machinelearning", flowserver.New(ml.New(), "ml")),
+		http.StripPrefix("/machinelearning", flowserver.New(mlReg, "ml")),
 	))
 
 	mux.Handle("/devops/", c.Build(

go/src/demos/cmd/noui/flow.go

@@ -1,127 +0,0 @@
-package main
-
-import (
-	"demos/ops/ml"
-	"flow"
-	"log"
-)
-
-func main() {
-
-	r := ml.New()
-
-	f := flow.New()
-	f.UseRegistry(r)
-
-	samples := [][]float64{
-		{0, 0},
-		{0, 1},
-		{1, 0},
-		{1, 1},
-	}
-	labels := []float64{1, 0, 0, 1}
-	_ = samples
-	_ = labels
-
-	// Try binary operations first
-	// 2 inputs 1 output
-	// Load minst somehow 28x28 784
-	nSample := 2
-
-	learningRate := float64(1.0)
-	_ = learningRate
-	// Load minst into floats array
-
-	// Make a matrix from the weights variable
-	// 5 outputs
-
-	// Layer 1
-	//
-	sample := samples[nSample]
-	label := labels[nSample] // Label output
-
-	process := func(op flow.Operation) flow.Data { // just an helper
-		res, err := op.Process(sample, []float64{label})
-		if err != nil {
-			panic(err)
-		}
-		return res
-	}
-
-	// Define input
-	// Make a matrix out of the input
-	input := f.Op("matNew", 2, 1, f.In(0))
-	log.Println("inMat:", process(input))
-
-	output := f.Op("matNew", 1, 1, f.In(1))
-	log.Println("inRes:", process(output))
-
-	// Layer1 weights 2 in 3 out
-	w1 := f.Op("matNew", 3, 2, f.Var("w1", f.Op("normFloat", 3*2)))
-	log.Println("w1 weights:", process(w1))
-	// Layour 1 result
-	l1res := f.Op("matSigmoid", f.Op("matMul", w1, input))
-
-	// weights 2 // 3 in 1 out
-	w2 := f.Op("matNew", 1, 3, f.Var("w2", f.Op("normFloat", 1*3)))
-	log.Println("w2 weights:", process(w2))
-
-	// Layour 2 result
-	l2res := f.Op("matSigmoid", f.Op("matMul", w2, l1res))
-
-	log.Println("Layer2 result:", process(l2res))
-	// Network output
-
-	//////////////////////
-	// Create trainer ops
-	/////////
-	// Backpropagation
-
-	log.Println("Grab error from output layer")
-	errOp := f.Op("matSub", output, l2res)
-	log.Println("ErrorOp:", process(errOp))
-
-	log.Println("Calculate deltas")
-	deltaOp := f.Op("matMulElem", f.Op("matSigmoidD", l2res), errOp)
-	log.Println("Delta:", process(deltaOp))
-
-	outChangesOp := f.Op("matMul", l2res, deltaOp)
-	log.Println("Changes to be made", process(outChangesOp))
-
-	/*
-		outChangesOp = f.Op("matScale", outChangesOp, learningRate)
-
-		train1 := f.SetVar("w2", f.Op("matAdd", outChangesOp, w1))
-		// Set Var w1
-		//
-		log.Println("Training 1")
-		log.Println(train1.Process(sample))
-
-		log.Println("deltaOutputLayer")
-		log.Println(deltaOp.Process(sample))
-
-		// Do we need this?
-		//netOp := f.Op("toFloatArr", l2op)*/
-
-	/*netResI, _ := netOp.Process(sample)
-	res := netResI.([]float64)
-	log.Println("Network result:", res)
-	log.Println("Error", sample[0]-res[0], sample[1]-res[1])*/
-	// Back propagation
-
-	//log.Println(layerRes.Process(samples[0]))
-
-	// Operation
-	/*mulRes := f.Op("matrixMul", w1, inp)
-	res, _ := mulRes.Process([]float64{0, 0})
-
-	sig := f.Op("matrixSigmoid", mulRes)
-
-	res, _ = sig.Process(samples[0])
-
-	log.Println("Result:", res)*/
-	//f.Analyse(os.Stdout, []float64{0, 1})
-
-	// weights
-
-}

go/src/demos/cmd/xor/main.go

@@ -0,0 +1,131 @@
+package main
+
+import (
+	"demos/ops/ml"
+	"flag"
+	"flow"
+	"fmt"
+	"log"
+	"os"
+	"runtime/pprof"
+
+	"gonum.org/v1/gonum/mat"
+)
+
+var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")
+var memprofile = flag.String("memprofile", "", "write mem profile to file")
+
+func main() {
+	flag.Parse()
+
+	if *cpuprofile != "" {
+		f, err := os.Create(*cpuprofile)
+		if err != nil {
+			log.Fatal(err)
+		}
+		pprof.StartCPUProfile(f)
+		defer pprof.StopCPUProfile()
+	}
+	if *memprofile != "" {
+		f, err := os.Create(*memprofile)
+		if err != nil {
+			log.Fatal(err)
+		}
+		defer pprof.WriteHeapProfile(f)
+	}
+	// Registry for machine learning
+	r := ml.New()
+
+	f := flow.New()
+	f.UseRegistry(r)
+
+	samples := []float64{
+		0, 0,
+		0, 1,
+		1, 0,
+		1, 1,
+	}
+	labels := []float64{
+		0,
+		1,
+		1,
+		0,
+	}
+	learningRate := float64(0.3)
+
+	nInputs := 2
+	nHidden := 5
+	nOutput := 1
+	nSamples := 4
+
+	matSamples := mat.NewDense(nSamples, 2, samples)
+	matLabels := mat.NewDense(nSamples, 1, labels)
+
+	// Define input
+	// Make a matrix out of the input and output
+	x := f.In(0)
+	y := f.In(1)
+
+	wHidden := f.Var("wHidden", f.Op("matNewRand", nInputs, nHidden))
+	wOut := f.Var("wOut", f.Op("matNewRand", nHidden, nOutput))
+
+	// Forward process
+	hiddenLayerInput := f.Op("matMul", x, wHidden)
+	hiddenLayerActivations := f.Op("matSigmoid", hiddenLayerInput)
+	outputLayerInput := f.Op("matMul", hiddenLayerActivations, wOut)
+	output := f.Op("matSigmoid", outputLayerInput)
+
+	// Back propagation
+	// output weights
+	networkError := f.Op("matSub", y, output)
+	slopeOutputLayer := f.Op("matSigmoidPrime", output)
+	dOutput := f.Op("matMulElem", networkError, slopeOutputLayer)
+	wOutAdj := f.Op("matMul", f.Op("matTranspose", hiddenLayerActivations), dOutput)
+	wOutAdj = f.Op("matScale", learningRate, wOutAdj)
+	// hidden weights
+	errorAtHiddenLayer := f.Op("matMul", dOutput, f.Op("matTranspose", wOut))
+	slopeHiddenLayer := f.Op("matSigmoidPrime", hiddenLayerActivations)
+	dHiddenLayer := f.Op("matMulElem", errorAtHiddenLayer, slopeHiddenLayer)
+	wHiddenAdj := f.Op("matMul", f.Op("matTranspose", x), dHiddenLayer)
+	wHiddenAdj = f.Op("matScale", learningRate, wHiddenAdj)
+
+	// Adjust the parameters
+	setwOut := f.SetVar("wOut", f.Op("matAdd", wOut, wOutAdj))
+	setwHidden := f.SetVar("wHidden", f.Op("matAdd", wHidden, wHiddenAdj))
+
+	// Training
+	for i := 0; i < 5000; i++ {
+		sess := f.NewSession()
+		sess.Inputs(matSamples, matLabels)
+		_, err := sess.Run(setwOut, setwHidden)
+		if err != nil {
+			log.Fatal(err)
+		}
+	}
+
+	// Same as above because its simple
+	testSamples := matSamples
+	testLabels := matLabels
+
+	res, err := output.Process(testSamples)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	predictions := res.(mat.Matrix)
+	log.Println("Predictions", predictions)
+
+	var rights int
+	numPreds, _ := predictions.Dims()
+	log.Println("Number of predictions:", numPreds)
+	for i := 0; i < numPreds; i++ {
+		if predictions.At(i, 0) > 0.5 && testLabels.At(i, 0) == 1.0 ||
+			predictions.At(i, 0) < 0.5 && testLabels.At(i, 0) == 0 {
+			rights++
+		}
+	}
+
+	accuracy := float64(rights) / float64(numPreds)
+	fmt.Printf("\nAccuracy = %0.2f\n\n", accuracy)
+
+}

go/src/demos/ops/defaultops/defaultops.go

@@ -1,28 +1,15 @@
 package defaultops
 
 import (
-	"errors"
-	"flow"
 	"flow/registry"
-	"fmt"
 	"math"
 	"math/rand"
-	"strings"
-	"time"
 )
 
 // New create a registry
 func New() *registry.R {
 	r := registry.New()
 	// String functions
-	registry.Describer(
-		r.Add(strings.Split).Inputs("string", "separator"),
-		r.Add(strings.Join).Inputs("", "sep"),
-		r.Add(strings.Compare, strings.Contains),
-		r.Add("Cat", func(a, b string) string { return a + " " + b }),
-		r.Add("ToString", func(a interface{}) string { return fmt.Sprint(a) }),
-	).Tags("string").Extra("style", registry.M{"color": "#839"})
-
 	// Math functions
 	r.Add(
 		math.Abs, math.Cos, math.Sin, math.Exp, math.Exp2, math.Tanh, math.Max, math.Min,
@@ -38,48 +25,5 @@ func New() *registry.R {
 		}),
 	).Tags("rand").Extra("style", registry.M{"color": "#486"})
 
-	// Test functions
-	r.Add(testErrorPanic, testErrorDelayed, testRandomError).
-		Tags("testing-errors")
-
-	registry.Describer(
-		r.Add("wait", wait),
-		r.Add("waitRandom", waitRandom),
-	).Tags("testing-time").Extra("style", map[string]string{"color": "#8a5"})
 	return r
 }
-
-func wait(data flow.Data, n int) flow.Data {
-	time.Sleep(time.Duration(n) * time.Second) // Simulate
-	return data
-}
-func waitRandom(data flow.Data) flow.Data {
-	time.Sleep(time.Duration(rand.Intn(10)) * time.Second)
-	return data
-}
-
-////////////////////
-// Testing custom struct operations
-/////
-
-////////////////////////
-// testOps
-////////////////
-func testErrorPanic(n int) flow.Data {
-	dur := time.Duration(n)
-	time.Sleep(dur * time.Second)
-	panic("I panicked")
-}
-
-func testErrorDelayed(n int) (flow.Data, error) {
-	dur := time.Duration(n)
-	time.Sleep(dur * time.Second)
-	return nil, fmt.Errorf("I got an error %v", dur)
-}
-func testRandomError(d flow.Data) (flow.Data, error) {
-	r := rand.Intn(10)
-	if r > 5 {
-		return nil, errors.New("I failed on purpose")
-	}
-	return d, nil
-}

go/src/demos/ops/genericops/registry.go

@@ -0,0 +1,59 @@
+package genericops
+
+import (
+	"errors"
+	"flow"
+	"flow/registry"
+	"fmt"
+	"math/rand"
+	"strings"
+	"time"
+)
+
+// New create new registry with generic operations
+func New() *registry.R {
+
+	r := registry.New()
+	// Test functions
+	r.Add(testErrorPanic, testErrorDelayed, testRandomError).
+		Tags("testing")
+
+	registry.Describer(
+		r.Add("wait", wait),
+		r.Add("waitRandom", waitRandom),
+	).Tags("testing").Extra("style", map[string]string{"color": "#8a5"})
+
+	r.Add(strings.Split, strings.Join).Tags("strings")
+
+	return r
+}
+func wait(data flow.Data, n int) flow.Data {
+	time.Sleep(time.Duration(n) * time.Second) // Simulate
+	return data
+}
+func waitRandom(data flow.Data) flow.Data {
+	time.Sleep(time.Duration(rand.Intn(10)) * time.Second)
+	return data
+}
+
+////////////////////////
+// testOps
+////////////////
+func testErrorPanic(n int) flow.Data {
+	dur := time.Duration(n)
+	time.Sleep(dur * time.Second)
+	panic("I panicked")
+}
+
+func testErrorDelayed(n int) (flow.Data, error) {
+	dur := time.Duration(n)
+	time.Sleep(dur * time.Second)
+	return nil, fmt.Errorf("I got an error %v", dur)
+}
+func testRandomError(d flow.Data) (flow.Data, error) {
+	r := rand.Intn(10)
+	if r > 5 {
+		return nil, errors.New("I failed on purpose")
+	}
+	return d, nil
+}

go/src/demos/ops/ml/gonumops.go

@@ -29,11 +29,11 @@ func New() *registry.R {
 			matScale,
 			matTranspose,
 			matSigmoid,
-			matSigmoidD,
+			matSigmoidPrime,
 			toFloatArr,
 		),
 	).Description("gonum functions").
-		Tags("gonum").
+		Tags("machine learning").
 		Extra("style", registry.M{"color": "#953"})
 
 	return r
@@ -80,7 +80,7 @@ func matMulElem(a, b Matrix) Matrix {
 }
 
 // Scalar per element multiplication
-func matScale(a Matrix, f float64) Matrix {
+func matScale(f float64, a Matrix) Matrix {
 	var r mat.Dense
 	r.Scale(f, a)
 	return &r
@@ -92,25 +92,14 @@ func matTranspose(a Matrix) Matrix {
 
 // sigmoid Activator
 func matSigmoid(a Matrix) Matrix {
-	// Should be a vector perhaps
-	r, c := a.Dims()
-	ret := mat.NewDense(r, c, nil)
-	for c--; c >= 0; c-- {
-		for r--; r >= 0; r-- {
-			ret.Set(r, c, activator(a.At(r, c)))
-		}
-	}
+	ret := &mat.Dense{}
+	ret.Apply(func(_, _ int, v float64) float64 { return sigmoid(v) }, a)
 	return ret
 }
 
-func matSigmoidD(a Matrix) Matrix {
-	r, c := a.Dims()
-	ret := mat.NewDense(r, c, nil)
-	for c--; c >= 0; c-- {
-		for r--; r >= 0; r-- {
-			ret.Set(r, c, derivative(a.At(r, c)))
-		}
-	}
+func matSigmoidPrime(a Matrix) Matrix {
+	ret := &mat.Dense{}
+	ret.Apply(func(_, _ int, v float64) float64 { return sigmoidPrime(v) }, a)
 	return ret
 }
 
@@ -125,9 +114,9 @@ func toFloatArr(a Matrix) []float64 {
 	return ret
 }
 
-func activator(v float64) float64 {
+func sigmoid(v float64) float64 {
 	return 1 / (1 + math.Exp(-v))
 }
-func derivative(v float64) float64 {
+func sigmoidPrime(v float64) float64 {
 	return v * (1 - v)
 }

go/src/demos/ops/stringops/stringops.go

@@ -0,0 +1,21 @@
+package stringops
+
+import (
+	"flow/registry"
+	"fmt"
+	"strings"
+)
+
+// New create string operations
+func New() *registry.R {
+	r := registry.New()
+	registry.Describer(
+		r.Add(strings.Split).Inputs("string", "separator"),
+		r.Add(strings.Join).Inputs("", "sep"),
+		r.Add(strings.Compare, strings.Contains),
+		r.Add("Cat", func(a, b string) string { return a + " " + b }),
+		r.Add("ToString", func(a interface{}) string { return fmt.Sprint(a) }),
+	).Tags("string").Extra("style", registry.M{"color": "#839"})
+
+	return r
+}

go/src/demos/ops/webops/webops.go

@@ -0,0 +1,34 @@
+package webops
+
+import (
+	"flow/registry"
+	"io/ioutil"
+	"net/http"
+)
+
+// New creates web operations for flow
+func New() *registry.R {
+	r := registry.New()
+
+	r.Add(httpGet).Tags("http").
+		Extra("style", registry.M{"color": "#828"})
+
+	return r
+}
+
+func httpGet(url string) (string, error) {
+
+	res, err := http.Get(url)
+	if err != nil {
+		return "", err
+	}
+	defer res.Body.Close()
+
+	out, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		return "", err
+	}
+
+	return string(out), nil
+
+}

go/src/flow/flow.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"sync"
 )
 
 // Data interface
@@ -17,7 +18,7 @@ type Data = interface{}
 type Flow struct {
 	//sync.Mutex // Needed?
 	registry   *registry.R
-	Data       map[string]Data // Should be named, to fetch later
+	Data       sync.Map // Should be named, to fetch later
 	consts     []Data
 	operations []*operation
 
@@ -29,7 +30,7 @@ type Flow struct {
 func New() *Flow {
 	return &Flow{
 		registry:   registry.Global,
-		Data:       map[string]Data{},
+		Data:       sync.Map{},
 		operations: []*operation{},
 		consts:     []Data{},
 	}
@@ -91,9 +92,11 @@ func (f *Flow) String() string {
 		fmt.Fprintf(ret, "  [%v] %v\n", i, v)
 	}
 	fmt.Fprintf(ret, "data:\n") // Or variable
-	for k, v := range f.Data {
+
+	f.Data.Range(func(k, v interface{}) bool {
 		fmt.Fprintf(ret, "  [%v] %v\n", k, v)
-	}
+		return true
+	})
 
 	fmt.Fprintf(ret, "operations:\n")
 	for k, op := range f.operations {
@@ -121,32 +124,6 @@ func (f *Flow) String() string {
 	return ret.String()
 }
 
-// MarshalJSON implementation
-/*func (f *Flow) MarshalJSON() ([]byte, error) {
-	data := map[string]interface{}{}
-	type opMarshal struct {
-		Name  string
-		Input []map[string]interface{}
-	}
-	operations := map[int]opMarshal{}
-	for k, op := range f.operations {
-		refs := []map[string]interface{}{}
-		for _, in := range op.inputs { // Switch type?
-			refs = append(refs, map[string]interface{}{
-				"type": in.kind,
-				"id":   in.id,
-			})
-		}
-		operations[k] = opMarshal{o.name, refs}
-	}
-
-	data["operations"] = operations
-	data["data"] = f.Data
-	data["consts"] = f.consts
-
-	return json.Marshal(data)
-}*/
-
 //////////////////////////////////////////////
 // Experimental event hooks
 ////////////////

go/src/flow/flowserver/flowbuilder/builder.go

@@ -106,12 +106,14 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 		op := f.In(inputID) // By id perhaps
 		fb.OperationMap[node.ID] = op
 		return op
-	case "Variable":
+	case "Var":
 		log.Println("Source is a variable")
 		var t interface{}
-		inputs = []reflect.Type{
-			reflect.TypeOf(t),
-		}
+		inputs = []reflect.Type{reflect.TypeOf(t)}
+	case "SetVar":
+		log.Println("Source is a setvariable")
+		var t interface{}
+		inputs = []reflect.Type{reflect.TypeOf(t)}
 	default:
 		log.Println("Loading entry:", node.Src)
 		entry, err := r.Entry(node.Src)
@@ -141,21 +143,28 @@ func (fb *FlowBuilder) Build(ID string) flow.Operation {
 		param[i] = fb.Build(l.From)
 	}
 
-	if node.Src == "Variable" {
-		op = f.Var(node.Label, param[0])
-	} else {
+	//Switch again
+	switch node.Src {
+	case "Var":
+		op = f.Var(node.Prop["variable name"], param[0])
+	case "SetVar":
+		op = f.SetVar(node.Prop["variable name"], param[0])
+	default:
 		op = f.Op(node.Src, param...)
 	}
+
 	fb.OperationMap[node.ID] = op
-	fb.addTriggersTo(node, op)
+	fb.buildTriggersFor(node, op)
 
 	return op
 }
 
-func (fb *FlowBuilder) addTriggersTo(node *Node, op flow.Operation) error {
+func (fb *FlowBuilder) buildTriggersFor(node *Node, targetOp flow.Operation) error {
 	// Process triggers for this node
 	triggers := fb.doc.fetchTriggerFrom(node.ID)
+	log.Println("Operation has this triggers:", triggers)
 	for _, t := range triggers {
+		log.Println("will build for")
 		op := fb.Build(t.To)
 		// Register the thing here
 		fb.flow.Hook(flow.Hook{
@@ -163,7 +172,7 @@ func (fb *FlowBuilder) addTriggersTo(node *Node, op flow.Operation) error {
 				if name != "Error" && name != "Finish" {
 					return
 				}
-				if triggerOp != op {
+				if triggerOp != targetOp {
 					log.Printf("ID triggered [%v], I'm t.From: %v", name, t.From)
 					return
 				}

go/src/flow/operation.go

@@ -8,8 +8,9 @@ package flow
 import (
 	"errors"
 	"fmt"
+	"path"
 	"reflect"
-	"runtime/debug"
+	"runtime"
 	"sync"
 )
 
@@ -17,100 +18,72 @@ type executorFunc func(*Session, ...Data) (Data, error)
 
 // Operation interface
 type Operation interface { // Id perhaps?
-	//ID() string
-	//Set(input Data) // Special var method
 	Process(params ...Data) (Data, error)
 }
 
 type operation struct {
 	sync.Mutex
-	flow *Flow
-	name string
-	kind string
-
-	// Could be a simple ID for operation, but it will depend on flow
-	inputs []*operation // still figuring, might be Operation
-
-	fn       interface{}  // the registry func
+	flow     *Flow
+	name     string
+	kind     string
+	inputs   []*operation // still figuring, might be Operation
 	executor executorFunc // the executor?
-	//processor func(OpCtx, params ...Data) (Data, error)
+
+	// Debug information for each operation
+	file string
+	line int
 }
 
 // NewOperation creates an operation
 func (f *Flow) newOperation(kind string, inputs []*operation) *operation {
+	_, file, line, _ := runtime.Caller(2) // outside of operation.go?
 	return &operation{
 		Mutex:  sync.Mutex{},
 		flow:   f,
 		kind:   kind,
 		inputs: inputs,
+
+		file: file,
+		line: line,
 		//name:   fmt.Sprintf("(var)<%s>", name),
 	}
 
 }
+func (o *operation) String() string {
+	_, file := path.Split(o.file)
+	return fmt.Sprintf("[%s:%d]:{%s,%s}", file, o.line, o.kind, o.name)
+}
 
-// Process params are the global inputs
+// Process the operation with a new session
+// ginputs are the global inputs
 func (o *operation) Process(ginputs ...Data) (Data, error) {
-	// Create CTX
 	s := o.flow.NewSession()
-	return s.Run(o, ginputs...)
+	return s.run(o, ginputs...)
 }
 
-// make Executor for func
-// safe run a func
-func (f *Flow) asTrigger(op *operation) executorFunc {
-	return func(sess *Session, params ...Data) (Data, error) {
-		f.hooks.start(op)
-
-		//panic recoverer, since nodes are not our functions
-		var err error
-		var res Data
-		// Safe thing
-		func() {
-			defer func() {
-				if r := recover(); r != nil {
-					debug.PrintStack()
-					err = fmt.Errorf("%v", r)
-				}
-			}()
-			res, err = op.executor(sess, params...)
-		}()
-
-		if err != nil {
-			f.hooks.error(op, err)
-		} else {
-			f.hooks.finish(op, res)
-		}
-		return res, err
-
-	}
-}
-
-// processInputs will run a list of operations and return reflect values
-// to be processed next
-// NEW PARALLEL PROCESSING
-
 // Var create a operation
 func (f *Flow) Var(name string, initial Data) Operation {
 	inputs := f.makeInputs(initial)
 
 	op := f.newOperation("var", inputs)
-	op.name = fmt.Sprintf("(var)<%s>", name)
-
 	op.executor = func(sess *Session, ginputs ...Data) (Data, error) {
-		val, ok := f.Data[name]
+		if name == "" {
+			return nil, errors.New("Invalid name")
+		}
+		val, ok := f.Data.Load(name)
 		if !ok {
 			var initial Data
-			f.hooks.wait(op)
-			res, err := sess.RunList(op.inputs, ginputs...)
+			res, err := sess.processInputs(op, ginputs...)
 			if err != nil {
 				return nil, err
 			}
+
 			if len(res) > 0 {
 				initial = res[0]
 			}
 
 			val = initial
-			f.Data[name] = val
+			f.Data.Store(name, val)
 		}
 		return val, nil
 	}
@@ -121,14 +94,15 @@ func (f *Flow) Var(name string, initial Data) Operation {
 func (f *Flow) SetVar(name string, data Data) Operation {
 	inputs := f.makeInputs(data)
 	op := f.newOperation("setvar", inputs)
-	op.name = fmt.Sprintf("(setvar)<%s>", name)
 	op.executor = func(sess *Session, ginputs ...Data) (Data, error) {
-		f.hooks.wait(op)
-		res, err := sess.RunList(op.inputs, ginputs...)
+		if name == "" {
+			return nil, errors.New("Invalid name")
+		}
+		res, err := sess.processInputs(op, ginputs...)
 		if err != nil {
 			return nil, err
 		}
-		f.Data[name] = res[0]
+		f.Data.Store(name, res[0])
 		return res[0], nil
 	}
 
@@ -176,25 +150,19 @@ func (f *Flow) Const(value Data) Operation {
 	}
 
 	op := f.newOperation("const", nil)
-	op.name = fmt.Sprintf("(const)<%v:%v>", constID, value)
 	op.executor = func(*Session, ...Data) (Data, error) { return f.consts[constID], nil }
-	//f.operations = append(f.operations, op)
-
 	return op
 }
 
 // In define input operation
 func (f *Flow) In(paramID int) Operation {
-
 	op := f.newOperation("in", nil)
-	op.name = fmt.Sprintf("(in)<%d>", paramID)
 	op.executor = func(sess *Session, ginputs ...Data) (Data, error) {
 		if paramID < 0 || paramID >= len(ginputs) {
 			return nil, ErrInput
 		}
 		return ginputs[paramID], nil
 	}
-	//f.operations = append(f.operations, op)
 	return op
 }
 
@@ -213,20 +181,22 @@ func (f *Flow) makeInputs(params ...Data) []*operation {
 }
 
 // make any go func as an executor
-// Trigger
 func makeExecutor(op *operation, fn interface{}) executorFunc {
+	fnval := reflect.ValueOf(fn)
+	callParam := make([]reflect.Value, fnval.Type().NumIn())
+
 	// ExecutorFunc
-	return func(sess *Session, ginput ...Data) (Data, error) {
+	return func(sess *Session, ginputs ...Data) (Data, error) {
 		// Change to wait to wait for the inputs
-
-		op.flow.hooks.wait(op)
-		inRes, err := sess.RunList(op.inputs, ginput...)
+		inRes, err := sess.processInputs(op, ginputs...)
 		if err != nil {
 			return nil, err
 		}
+		// If the fn is a special we execute directly
+		if gFn, ok := fn.(func(...Data) (Data, error)); ok {
+			return gFn(inRes...)
+		}
 
-		fnval := reflect.ValueOf(fn)
-		callParam := make([]reflect.Value, len(inRes))
 		for i, r := range inRes {
 			if r == nil {
 				callParam[i] = reflect.Zero(fnval.Type().In(i))
@@ -236,7 +206,6 @@ func makeExecutor(op *operation, fn interface{}) executorFunc {
 		}
 
 		// Start again and execute function
-		op.flow.hooks.start(op)
 		fnret := fnval.Call(callParam)
 		if len(fnret) == 0 {
 			return nil, nil

go/src/flow/registry/registry.go

@@ -1,6 +1,7 @@
 package registry
 
 import (
+	"fmt"
 	"path"
 	"reflect"
 	"runtime"
@@ -107,7 +108,7 @@ func (r *R) register(name string, v interface{}) (*Entry, error) {
 func (r *R) Get(name string, params ...interface{}) (interface{}, error) {
 	e, ok := r.entries[name]
 	if !ok {
-		return nil, ErrNotFound
+		return nil, fmt.Errorf("Entry not found '%s'", name)
 	}
 	v := e.fn
 	// We already know this is a function

go/src/flow/registry/registry_test.go

@@ -76,7 +76,7 @@ func TestRegistryGetEmpty(t *testing.T) {
 	a := assert.A(t)
 	r := registry.New()
 	fn, err := r.Get("notfoundfunc")
-	a.Eq(err, registry.ErrNotFound, "should fail fetching a unregistered func")
+	a.NotEq(err, nil, "should fail fetching a unregistered func")
 	a.Eq(fn, nil, "should return a <nil> func")
 }
 func TestRegistryGetConstructor(t *testing.T) {

go/src/flow/session.go

@@ -2,13 +2,17 @@ package flow
 
 import (
 	"errors"
+	"fmt"
+	"os"
+	"runtime/debug"
 	"sync"
 )
 
 // Session operation session
 type Session struct {
 	*sync.Map
-	flow *Flow
+	flow    *Flow
+	ginputs []Data
 }
 
 // NewSession creates a running context
@@ -19,8 +23,24 @@ func (f *Flow) NewSession() *Session {
 	}
 }
 
-// Run run operation
-func (s *Session) Run(op *operation, ginputs ...Data) (Data, error) {
+// Inputs sets the global graph inputs
+func (s *Session) Inputs(ginputs ...Data) {
+	s.ginputs = ginputs
+}
+
+// Run session run
+func (s *Session) Run(ops ...Operation) ([]Data, error) {
+	oplist := make([]*operation, len(ops))
+	for i, op := range ops {
+		oplist[i] = op.(*operation)
+	}
+
+	return s.runList(oplist, s.ginputs...)
+}
+
+// The main run function?
+// run runs the operation
+func (s *Session) run(op *operation, ginputs ...Data) (Data, error) {
 	op.Lock()
 	defer op.Unlock()
 	// Load from cache if any
@@ -28,7 +48,7 @@ func (s *Session) Run(op *operation, ginputs ...Data) (Data, error) {
 		return v, nil
 	}
 
-	res, err := s.flow.asTrigger(op)(s, ginputs...)
+	res, err := s.triggerRun(op, ginputs...)
 	if err != nil {
 		return nil, err
 	}
@@ -39,7 +59,23 @@ func (s *Session) Run(op *operation, ginputs ...Data) (Data, error) {
 }
 
 // RunList more than one operations in this session
-func (s *Session) RunList(ops []*operation, ginputs ...Data) ([]Data, error) {
+func (s *Session) runList(ops []*operation, ginputs ...Data) ([]Data, error) {
+	nOps := len(ops)
+	// Total inputs
+	callParam := make([]Data, nOps)
+	// Parallel processing if inputs
+	for i, op := range ops {
+		res, err := s.run(op, ginputs...)
+		if err != nil {
+			return nil, err
+		}
+		callParam[i] = res
+	}
+	return callParam, nil
+}
+
+// Parallel
+func (s *Session) goRunList(ops []*operation, ginputs ...Data) ([]Data, error) {
 	nOps := len(ops)
 
 	// Total inputs
@@ -54,7 +90,7 @@ func (s *Session) RunList(ops []*operation, ginputs ...Data) ([]Data, error) {
 		go func(i int, op *operation) {
 			defer wg.Done()
 
-			res, err := s.Run(op, ginputs...)
+			res, err := s.run(op, ginputs...)
 			paramMutex.Lock()
 			defer paramMutex.Unlock()
 			if err != nil {
@@ -72,3 +108,40 @@ func (s *Session) RunList(ops []*operation, ginputs ...Data) ([]Data, error) {
 
 	return callParam, nil
 }
+
+// make Executor for func
+// safe run a func
+//
+func (s *Session) triggerRun(op *operation, ginputs ...Data) (Data, error) {
+	s.flow.hooks.start(op)
+	var err error
+	var res Data
+
+	func() {
+		defer func() {
+			if r := recover(); r != nil {
+				fmt.Fprintf(os.Stderr, "%v, %v\n", op, r)
+				for i, in := range op.inputs {
+					res, _ := s.run(in, ginputs...)
+					fmt.Fprintf(os.Stderr, "Input: %d = %v\n", i, res)
+				}
+				debug.PrintStack()
+				err = fmt.Errorf("%v %v", op, r)
+			}
+		}()
+		res, err = op.executor(s, ginputs...)
+	}()
+	if err != nil {
+		s.flow.hooks.error(op, err)
+	} else {
+		s.flow.hooks.finish(op, res)
+	}
+	return res, err
+
+}
+func (s *Session) processInputs(op *operation, ginputs ...Data) ([]Data, error) {
+	s.flow.hooks.wait(op)
+	res, err := s.goRunList(op.inputs, ginputs...)
+	s.flow.hooks.start(op) // Back to start
+	return res, err
+}