Home Explore Help
Sign In
stdio
/
Flow
1
0
Fork 0
Files Issues 0 Pull Requests 0 Wiki
Browse Source

Fixing submodule issue

luis 7 years ago
parent
19c66c4d4c
commit
9631547566
14 changed files with 687 additions and 19 deletions
Split View Show Diff Stats
  1. 1 0
      .gitignore
  2. BIN
      DIST/flowserver
  3. 0 10
      DIST/web/index.html
  4. 0 7
      DIST/web/index.js
  5. 0 1
      DIST/web/index.js.map
  6. 1 0
      go/.gitignore
  7. BIN
      go/DIST/flowserver
  8. 0 1
      go/src/flow
  9. 80 0
      go/src/flow/README.md
  10. 12 0
      go/src/flow/errors.go
  11. 210 0
      go/src/flow/flow.go
  12. 212 0
      go/src/flow/flow_test.go
  13. 116 0
      go/src/flow/operation.go
  14. 55 0
      go/src/flow/registry.go

+ 1 - 0
.gitignore
View File 

@@ -0,0 +1 @@
 
+DIST

BIN
DIST/flowserver
View File


+ 0 - 10
DIST/web/index.html
View File 

@@ -1,10 +0,0 @@
 
-<!DOCTYPE html>
 
-<html lang="en">
 
-  <head>
 
-    <meta charset="utf-8">
 
-    <title>Flow</title>
 
-  </head>
 
-  <body>
 
-    <div id="app"></div>
 
-  <script type="text/javascript" src="/index.js"></script></body>
 
-</html>

File diff suppressed because it is too large
+ 0 - 7
DIST/web/index.js


File diff suppressed because it is too large
+ 0 - 1
DIST/web/index.js.map


+ 1 - 0
go/.gitignore
View File 

@@ -1,4 +1,5 @@
 
 deps
 
 pkg
 
 bin
 
+DIST

BIN
go/DIST/flowserver
View File


+ 0 - 1
go/src/flow 

@@ -1 +0,0 @@
 
-Subproject commit 7d8bf49e51ca5e346c44a99bbb7c2efaf873516d

+ 80 - 0
go/src/flow/README.md
View File 

@@ -0,0 +1,80 @@
 
+# Extensible Flow Engine
 
+
 
+## Features
 
+
 
+* Develop new Operators
 
+* Serialize graph of operations
 
+* Distribute operations to serveral workers/servers
 
+* HTTP Graphical UI Editor
 
+* Serve http or grpc API
 
+* Chain flow operators -- No needed since we can build operations,
 
+  might be possible to build entires/methods from operators
 
+* System containing more than one flow
 
+* Define stages of the system
 
+
 
+## Special operators
 
+
 
+Idea: requesting the value to the flow instead of creating operators
 
+allows to create a caching system to allow recursion with previous values
 
+
 
+```go
 
+f := flow.New(f);
 
+v := f.Var([]float32{1,2,3}) // Init value
 
+```
 
+
 
+### New version
 
+
 
+Simplified flow even more maintaining a func
 
+
 
+### flowt2
 
+
 
+Flow will create an array of operators with references as inputs
 
+the builder will create the Graph
 
+
 
+We should not allow user to directly pass Operation to the `flow.Op` method,
 
+since the serialization will be hard then
 
+
 
+define future inputs for operation groups
 
+
 
+```go
 
+g.Run([]O{op1,op2},1,2,3)
 
+```
 
+
 
+## Serialize
 
+
 
+Grab all operators in a list
 
+create a reference lookup table
 
+
 
+## System
 
+
 
+system combine several flows
 
+
 
+Idea:
 
+
 
+Develop and combine operators to create a function,
 
+
 
+### Components
 
+
 
+using ECS (entity component system) to extend each Operator, so if we have
 
+UI we store UI information in UI component
 
+
 
+### Describe components via code or serialization
 
+
 
+Each node has components
 
+create multiple channels to each node
 
+
 
+CNN - combines convolution matrixes with images then sends to a regular
 
+neural network
 
+
 
+a Node could have input and properties?
 
+
 
+## Backfetch:
 
+
 
+perpare linkages as AddSource(node)
 
+every time we call, it will call the results
 
+
 
+```go
 
+// Prototype
 
+n := flow.AddNode(&node)
 
+n.AddSource(&otherNode)
 
+```

+ 12 - 0
go/src/flow/errors.go
View File 

@@ -0,0 +1,12 @@
 
+package flow
 
+
 
+import "errors"
 
+
 
+// flow Errors
 
+var (
 
+	ErrNotFound  = errors.New("Entry not found")
 
+	ErrNotAFunc  = errors.New("Is not a function")
 
+	ErrInput     = errors.New("Invalid input")
 
+	ErrOutput    = errors.New("Invalid output")
 
+	ErrOperation = errors.New("Invalid operation")
 
+)

+ 210 - 0
go/src/flow/flow.go
View File 

@@ -0,0 +1,210 @@
 
+package flow
 
+
 
+import (
 
+	"bytes"
 
+	"encoding/json"
 
+	"fmt"
 
+	"reflect"
 
+)
 
+
 
+// Data interface
 
+type Data = interface{}
 
+
 
+type opEntry struct {
 
+	name     string
 
+	inputs   []*operation // still figuring, might be operation
 
+	executor interface{}
 
+}
 
+
 
+// Flow structure
 
+// We could Create a single array of operations
 
+// refs would only mean id, types would be embed in operation
 
+type Flow struct {
 
+	registry *Registry
 
+
 
+	consts     []Data
 
+	data       map[string]Data // Should be named, to fetch later
 
+	operations []opEntry
 
+
 
+	err   error
 
+	runID int
 
+}
 
+
 
+// New create a new flow
 
+func New() *Flow {
 
+	return &Flow{
 
+		registry: globalRegistry,
 
+		// Data
 
+		consts:     []Data{},
 
+		data:       map[string]Data{},
 
+		operations: []opEntry{},
 
+	}
 
+}
 
+
 
+//SetRegistry use the registry specified
 
+func (f *Flow) SetRegistry(r *Registry) *Flow {
 
+	f.registry = r
 
+	// chain
 
+	return f
 
+}
 
+
 
+// Op return an function operator
 
+//  name - a previous registered function
 
+//  params - the function inputs
 
+func (f *Flow) Op(name string, params ...interface{}) Operation {
 
+	// Use this on Set?
 
+	inputs := make([]*operation, len(params))
 
+	for i, p := range params {
 
+		switch v := p.(type) {
 
+		case *operation:
 
+			inputs[i] = v
 
+		default:
 
+			inputs[i] = f.Const(v).(*operation)
 
+		}
 
+	}
 
+
 
+	// Grab executor here
 
+	executor, err := f.registry.Get(name)
 
+	if err != nil {
 
+		f.err = err
 
+		return nil
 
+	}
 
+	f.operations = append(f.operations, opEntry{name, inputs, executor})
 
+	refID := len(f.operations) - 1
 
+	// Initialize opfunc maybe
 
+	return opFunc(f, refID)
 
+
 
+}
 
+
 
+// Const returns a const operation
 
+func (f *Flow) Const(value Data) Operation {
 
+	f.consts = append(f.consts, value)
 
+	refID := len(f.consts) - 1
 
+	return opConst(f, refID)
 
+}
 
+
 
+// Var operation
 
+func (f *Flow) Var(name string, initial ...Data) Operation {
 
+	if _, ok := f.data[name]; !ok {
 
+		var v interface{}
 
+		if len(initial) > 0 {
 
+			v = initial[0]
 
+		}
 
+		f.data[name] = v
 
+	}
 
+	return opVar(f, name)
 
+}
 
+
 
+// In flow input operator
 
+//  paramID - index of the parameter
 
+func (f *Flow) In(paramID int) Operation {
 
+	return opIn(f, paramID)
 
+}
 
+
 
+// Run a batch of operation?
 
+func (f *Flow) Run(op Operation, params ...Data) (Data, error) {
 
+	if f.err != nil {
 
+		return nil, f.err
 
+	}
 
+	cache := map[*operation]Data{}
 
+	return f.run(cache, op, params...)
 
+}
 
+func (f *Flow) run(cache map[*operation]Data, op Operation, params ...Data) (Data, error) {
 
+	o := op.(*operation)
 
+	if v, ok := cache[o]; ok {
 
+		return v, nil
 
+	}
 
+	// Manually fetch func data because of caching
 
+	var r Data
 
+	// This is wrong since the only source of func should be on operation
 
+	if o.kind == "func" {
 
+		op := f.operations[o.id.(int)]
 
+		callParam := make([]reflect.Value, len(op.inputs))
 
+		for i, in := range op.inputs {
 
+			fr, _ := f.run(cache, in, params...) // ignore error
 
+			callParam[i] = reflect.ValueOf(fr)
 
+		}
 
+		r = reflect.ValueOf(op.executor).Call(callParam)[0].Interface()
 
+	} else {
 
+		r = o.process(nil, params...)
 
+	}
 
+	cache[o] = r
 
+	return r, nil
 
+}
 
+
 
+// Analyse every operations
 
+func (f *Flow) Analyse(params ...Data) string {
 
+	ret := bytes.NewBuffer(nil)
 
+	fmt.Fprintf(ret, "Ops analysis:\n")
 
+
 
+	for i, op := range f.operations {
 
+		fmt.Fprintf(ret, "  [%d] %s(", i, op.name)
 
+		for j, in := range op.inputs {
 
+			//ref := in.(Op)
 
+			if j != 0 {
 
+				fmt.Fprintf(ret, ", ")
 
+			}
 
+			ires := in.Process(params...)
 
+			fmt.Fprintf(ret, "%s[%v](%v)", in.kind, in.id, ires)
 
+		}
 
+		fmt.Fprintf(ret, ") - ")
 
+		// Create OpProcessor and execute
 
+		//
 
+		opfn := opFunc(f, i)
 
+		res := opfn.Process(params...)
 
+		fmt.Fprintf(ret, "%v\n", res)
 
+	}
 
+	return ret.String()
 
+}
 
+
 
+func (f *Flow) String() string {
 
+	ret := bytes.NewBuffer(nil)
 
+	// consts
 
+	fmt.Fprintf(ret, "consts:\n")
 
+	for i, v := range f.consts {
 
+		fmt.Fprintf(ret, "  [%d] %v\n", i, v)
 
+	}
 
+	fmt.Fprintf(ret, "data:\n") // Or variable
 
+	for k, v := range f.data {
 
+		fmt.Fprintf(ret, "  [%v] %v\n", k, v)
 
+	}
 
+
 
+	fmt.Fprintf(ret, "operations:\n")
 
+	for i, v := range f.operations {
 
+		fmt.Fprintf(ret, "  [%d] %s(", i, v.name)
 
+		for j, in := range v.inputs {
 
+			if j != 0 {
 
+				fmt.Fprintf(ret, ", ")
 
+			}
 
+			fmt.Fprintf(ret, "%s[%v]", in.kind, in.id)
 
+		}
 
+		fmt.Fprintf(ret, ")\n")
 
+	}
 
+
 
+	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 := make([]opMarshal, len(f.operations))
 
+	for i, o := range f.operations {
 
+		refs := []map[string]interface{}{}
 
+		for _, in := range o.inputs { // Switch type?
 
+			refs = append(refs, map[string]interface{}{
 
+				"type": in.kind,
 
+				"id":   in.id,
 
+			})
 
+		}
 
+		operations[i] = opMarshal{o.name, refs}
 
+	}
 
+	data["operations"] = operations
 
+	data["data"] = f.data
 
+	data["consts"] = f.consts
 
+
 
+	return json.Marshal(data)
 
+}

+ 212 - 0
go/src/flow/flow_test.go
View File 

@@ -0,0 +1,212 @@
 
+package flow_test
 
+
 
+import (
 
+	"bytes"
 
+	"encoding/json"
 
+	"fmt"
 
+	"reflect"
 
+	"testing"
 
+
 
+	"flow"
 
+
 
+	vecasm "github.com/gohxs/vec-benchmark/asm"
 
+)
 
+
 
+func TestSerialize(t *testing.T) {
 
+	// Does not text yet
 
+	f := flow.New()
 
+	var1 := f.Var("var1", []float32{4, 4, 4})
 
+
 
+	c1 := f.Const([]float32{1, 2, 3})
 
+	c2 := f.Const([]float32{2, 2, 2})
 
+
 
+	op1 := f.Op("vecmul", // op:0 - expected: [12,16,20,24]
 
+		f.Var("vec1", []float32{4, 4, 4, 4}),
 
+		f.Op("vecadd", // op:1 - expected: [3,4,5,6]
 
+			f.Const([]float32{1, 2, 3, 4}),
 
+			f.Const([]float32{2, 2, 2, 2}),
 
+		),
 
+	)
 
+	mul1 := f.Op("vecmul", c1, op1)       // op:2 - expected 12, 32, 60, 0
 
+	mul2 := f.Op("vecmul", mul1, var1)    // op:3 - expected 48, 128, 240, 0
 
+	mul3 := f.Op("vecmul", c2, mul2)      // op:4 - expected 96, 256, 480, 0
 
+	mul4 := f.Op("vecmul", mul3, f.In(0)) // op:5 - expected 96, 512, 1440,0
 
+
 
+	t.Log(f.Analyse([]float32{1, 2, 3, 4}))
 
+	res := mul4.Process([]float32{1, 2, 3, 4})
 
+
 
+	t.Log("Res:", res)
 
+	t.Log("Flow:\n", f)
 
+
 
+	ret := bytes.NewBuffer(nil)
 
+	e := json.NewEncoder(ret)
 
+	e.Encode(f)
 
+
 
+	t.Log("Flow:", ret)
 
+
 
+}
 
+func TestConst(t *testing.T) {
 
+	f := flow.New()
 
+
 
+	c := f.Const(1)
 
+	res := c.Process()
 
+	checkeq(t, res, 1)
 
+}
 
+func TestOp(t *testing.T) {
 
+	f := flow.New()
 
+
 
+	add := f.Op("vecadd",
 
+		f.Op("vecmul",
 
+			[]float32{1, 2, 3},
 
+			[]float32{2, 2, 2},
 
+		),
 
+		[]float32{1, 2, 3},
 
+	)
 
+	res, err := f.Run(add)
 
+	checkeq(t, err, nil)
 
+
 
+	test := []float32{3, 6, 9}
 
+	checkeq(t, test, res)
 
+}
 
+
 
+func TestVariable(t *testing.T) {
 
+	f := flow.New()
 
+	v := f.Var("v1", 1)
 
+
 
+	res := v.Process()
 
+	checkeq(t, res, 1)
 
+
 
+	v.Set(2)
 
+	res = v.Process()
 
+	checkeq(t, res, 2)
 
+}
 
+
 
+func TestCache(t *testing.T) {
 
+	f := flow.New()
 
+	{
 
+		r := f.Op("inc")
 
+		var res interface{}
 
+		for i := 1; i < 5; i++ {
 
+			res = r.Process()
 
+			checkeq(t, res, i)
 
+		}
 
+	}
 
+	{
 
+		var res flow.Data
 
+		inc := f.Op("inc")
 
+		add := f.Op("add", inc, inc)
 
+		res = add.Process() // 1+1
 
+		checkeq(t, res, 2)
 
+		res = add.Process() // 2+2
 
+		checkeq(t, res, 4)
 
+
 
+	}
 
+
 
+}
 
+
 
+func BenchmarkComplex(b *testing.B) {
 
+	f, op := prepareComplex()
 
+
 
+	b.Run("OP", func(b *testing.B) {
 
+		for i := 0; i < b.N; i++ {
 
+			op.Process()
 
+		}
 
+	})
 
+	b.Run("RUN", func(b *testing.B) {
 
+		for i := 0; i < b.N; i++ {
 
+			f.Run(op)
 
+		}
 
+	})
 
+}
 
+
 
+func init() {
 
+	flow.Register("vecmul", vecmul)
 
+	flow.Register("vecadd", vecadd)
 
+	flow.Register("vecdiv", vecdiv)
 
+	flow.Register("inc", inc)
 
+	flow.Register("add", add)
 
+}
 
+
 
+func prepareComplex() (*flow.Flow, flow.Operation) {
 
+	vecsize := 5
 
+	v1 := make([]float32, vecsize)
 
+	v2 := make([]float32, vecsize)
 
+	for i := range v1 {
 
+		v1[i], v2[i] = float32(i+1), 2
 
+	}
 
+
 
+	f := flow.New()
 
+	f1 := f.Var("f1", v1)
 
+	f2 := f.Var("f2", v2)
 
+
 
+	mul := f.Op("vecmul", f1, f2)      // Doubles 2,4,6,8...
 
+	add := f.Op("vecadd", mul, f2)     // Sum 4,8,10,12...
 
+	mul2 := f.Op("vecmul", mul, add)   // mul again
 
+	mul3 := f.Op("vecmul", mul2, f1)   // mul with f1
 
+	div1 := f.Op("vecdiv", mul3, mul2) // div
 
+
 
+	return f, div1
 
+}
 
+
 
+// Some funcs
 
+func vecmul(a, b []float32) []float32 {
 
+
 
+	sz := min(len(a), len(b))
 
+
 
+	out := make([]float32, sz)
 
+	vecasm.VecMulf32x8(a, b, out)
 
+	return out
 
+}
 
+func vecadd(a, b []float32) []float32 {
 
+	sz := min(len(a), len(b))
 
+	out := make([]float32, sz)
 
+	for i := 0; i < sz; i++ {
 
+		out[i] = a[i] + b[i]
 
+	}
 
+	return out
 
+}
 
+func vecdiv(a, b []float32) []float32 {
 
+	sz := min(len(a), len(b))
 
+	out := make([]float32, sz)
 
+	for i := 0; i < sz; i++ {
 
+		out[i] = a[i] / b[i]
 
+	}
 
+	return out
 
+}
 
+
 
+// ScalarInt
 
+// Every time this operator is called we increase 1
 
+func inc() func() int {
 
+	i := 0
 
+	return func() int {
 
+		i++
 
+		return i
 
+	}
 
+}
 
+func add(a, b int) int {
 
+	return a + b
 
+}
 
+
 
+// Utils
 
+func min(p ...int) int {
 
+	min := p[0]
 
+	for _, v := range p[1:] {
 
+		if min < v {
 
+			min = v
 
+		}
 
+	}
 
+	return min
 
+}
 
+func checkerr(t *testing.T, err error) {
 
+	if err != nil {
 
+		t.Fatal(err)
 
+	}
 
+}
 
+
 
+func checkeq(t *testing.T, a, b interface{}) {
 
+	msg := fmt.Sprintf("Expect %v got %v", a, b)
 
+	if !reflect.DeepEqual(a, b) {
 
+		t.Fatalf("\033[31m[FAIL] %s\033[0m", msg)
 
+	}
 
+	t.Logf("\033[32m[PASS]\033[m %s", msg)
 
+}

+ 116 - 0
go/src/flow/operation.go
View File 

@@ -0,0 +1,116 @@
 
+package flow
 
+
 
+//
 
+// Find a way to improve this mess, maybe it can be merged in one func
 
+//
 
+//
 
+
 
+import (
 
+	"errors"
 
+	"reflect"
 
+)
 
+
 
+// OpCtx operation Context
 
+type OpCtx map[Operation]Data
 
+
 
+// NewOpCtx creates a running context
 
+func NewOpCtx() OpCtx {
 
+	return OpCtx{}
 
+}
 
+
 
+// dumbSet
 
+func dumbSet(params ...Data) {}
 
+
 
+// Operation interface
 
+type Operation interface {
 
+	Set(inputs ...Data) // Special var method
 
+	Process(params ...Data) Data
 
+	ProcessWithCtx(ctx OpCtx, params ...Data) Data
 
+}
 
+
 
+// Run Context actually not OpCTX
 
+
 
+//local operation information
 
+type operation struct {
 
+	flow    *Flow
 
+	id      interface{} // Interface key
 
+	kind    string
 
+	set     func(params ...Data)
 
+	process func(ctx OpCtx, params ...Data) Data
 
+}
 
+
 
+// Process operation process wrapper
 
+func (o *operation) Process(params ...Data) Data {
 
+	return o.ProcessWithCtx(NewOpCtx(), params...)
 
+}
 
+
 
+// Every single one is run with this internally
 
+func (o *operation) ProcessWithCtx(ctx OpCtx, params ...Data) Data {
 
+	if ctx == nil { // No cache/Context
 
+		return o.process(ctx, params...)
 
+	}
 
+	if v, ok := ctx[o]; ok {
 
+		return v
 
+	}
 
+	res := o.process(ctx, params...)
 
+	ctx[o] = res
 
+
 
+	return res
 
+}
 
+
 
+// Set setter for certain operations (Var)
 
+func (o *operation) Set(params ...Data) {
 
+	o.set(params...)
 
+}
 
+
 
+func opIn(f *Flow, id int) *operation {
 
+	return &operation{
 
+		flow: f,
 
+		id:   id,
 
+		kind: "in",
 
+		set:  dumbSet,
 
+		process: func(ctx OpCtx, params ...Data) Data {
 
+			if id >= len(params) || id < 0 {
 
+				f.err = errors.New("Invalid input")
 
+				return nil
 
+			}
 
+			return params[id]
 
+		},
 
+	}
 
+}
 
+func opConst(f *Flow, id int) *operation {
 
+	return &operation{
 
+		flow:    f,
 
+		id:      id,
 
+		kind:    "const",
 
+		set:     dumbSet,
 
+		process: func(ctx OpCtx, params ...Data) Data { return f.consts[id] },
 
+	}
 
+}
 
+func opFunc(f *Flow, id int) *operation {
 
+	return &operation{
 
+		flow: f,
 
+		id:   id,
 
+		kind: "func",
 
+		set:  dumbSet,
 
+		process: func(ctx OpCtx, params ...Data) Data {
 
+			op := f.operations[id]
 
+			callParam := make([]reflect.Value, len(op.inputs))
 
+			for i, in := range op.inputs {
 
+				fr := in.ProcessWithCtx(ctx, params...)
 
+				callParam[i] = reflect.ValueOf(fr)
 
+			}
 
+			return reflect.ValueOf(op.executor).Call(callParam)[0].Interface()
 
+		},
 
+	}
 
+}
 
+
 
+func opVar(f *Flow, id string) *operation {
 
+	return &operation{
 
+		flow:    f,
 
+		id:      id,
 
+		kind:    "var",
 
+		set:     func(params ...Data) { f.data[id] = params[0] },
 
+		process: func(ctx OpCtx, params ...Data) Data { return f.data[id] },
 
+	}
 
+}

+ 55 - 0
go/src/flow/registry.go
View File 

@@ -0,0 +1,55 @@
 
+package flow
 
+
 
+import "reflect"
 
+
 
+// Global
 
+var (
 
+	globalRegistry = NewRegistry()
 
+	Register       = globalRegistry.Register
 
+)
 
+
 
+// Registry function registry
 
+type Registry struct {
 
+	data map[string]interface{}
 
+}
 
+
 
+// NewRegistry creates a new registry
 
+func NewRegistry() *Registry {
 
+	return &Registry{map[string]interface{}{}}
 
+}
 
+
 
+//Register should be a func only register
 
+func (r *Registry) Register(name string, v interface{}) error {
 
+	fntyp := reflect.TypeOf(v)
 
+	if fntyp.Kind() != reflect.Func {
 
+		return ErrNotAFunc
 
+	}
 
+	if fntyp.NumOut() == 0 {
 
+		return ErrOutput
 
+	}
 
+
 
+	r.data[name] = v
 
+	return nil
 
+}
 
+
 
+// Get an entry
 
+func (r *Registry) Get(name string, params ...interface{}) (interface{}, error) {
 
+	v, ok := r.data[name]
 
+	if !ok {
 
+		return nil, ErrNotFound
 
+	}
 
+
 
+	// We already know this is a function
 
+	// and that returns 1 or more values
 
+	vtyp := reflect.TypeOf(v)
 
+	if vtyp.Out(0).Kind() == reflect.Func {
 
+		fparam := make([]reflect.Value, len(params))
 
+		for i := range params {
 
+			fparam[i] = reflect.ValueOf(params[i])
 
+		}
 
+		// Call the func and return the thing
 
+		v = reflect.ValueOf(v).Call(fparam)[0].Interface()
 
+	}
 
+
 
+	return v, nil
 
+}