(Part 5)Golang Framework Hands-on – KisFlow Stream Computing Framework-Function Scheduling

package kis

import “context”

// FaaS Function as a Service
type FaaS func(context.Context, Flow) error

// funcRouter
// key: Function Name
// value: Function callback for custom business logic
type funcRouter map[string]FaaS

// flowRouter
// key: Flow Name
// value: Flow
type flowRouter map[string]Flow

package kis

import (
“context”
“errors”
“fmt”
“kis-flow/log”
“sync”
)

var _poolOnce sync.Once

// kisPool is used to manage all Function and Flow configuration pools
type kisPool struct {
fnRouter funcRouter // All Function management routes
fnLock sync.RWMutex // Lock for fnRouter

flowRouter flowRouter // All flow objects
flowLock sync.RWMutex // Lock for flowRouter
}

// Singleton
var _pool *kisPool

// Pool Singleton constructor
func Pool() *kisPool {
_poolOnce.Do(func() {
// Create kisPool object
_pool = new(kisPool)

// Initialize fnRouter
_pool.fnRouter = make(funcRouter)

// Initialize flowRouter
_pool.flowRouter = make(flowRouter)
})

return _pool
}

func (pool *kisPool) AddFlow(name string, flow Flow) {
pool.flowLock.Lock()
defer pool.flowLock.Unlock()

if _, ok := pool.flowRouter[name]; !ok {
pool.flowRouter[name] = flow
} else {
errString := fmt.Sprintf(“Pool AddFlow Repeat FlowName=%sn“, name)
panic(errString)
}

log.Logger().InfoF(“Add FlowRouter FlowName=%sn“, name)
}

func (pool *kisPool) GetFlow(name string) Flow {
pool.flowLock.RLock()
defer pool.flowLock.RUnlock()

if flow, ok := pool.flowRouter[name]; ok {
return flow
} else {
return nil
}
}

// FaaS registers Function computing business logic, indexed and registered by Function Name
func (pool *kisPool) FaaS(fnName string, f FaaS) {
pool.fnLock.Lock()
defer pool.fnLock.Unlock()

if _, ok := pool.fnRouter[fnName]; !ok {
pool.fnRouter[fnName] = f
} else {
errString := fmt.Sprintf(“KisPoll FaaS Repeat FuncName=%s”, fnName)
panic(errString)
}

log.Logger().InfoF(“Add KisPool FuncName=%s”, fnName)
}

// CallFunction schedules a Function
func (pool *kisPool) CallFunction(ctx context.Context, fnName string, flow Flow) error {

if f, ok := pool.fnRouter[fnName]; ok {
return f(ctx, flow)
}

log.Logger().ErrorFX(ctx, “FuncName: %s Can not find in KisPool, Not Added.n“, fnName)

return errors.New(“FuncName: “ + fnName + ” Can not find in NsPool, Not Added.”)
}

type Flow interface {
// Run schedules the Flow, sequentially schedules Functions in the Flow and executes them
Run(ctx context.Context) error
// Link connects the Functions in the Flow according to the configuration in the configuration file
Link(fConf *config.KisFuncConfig, fParams config.FParam) error
// CommitRow submits Flow data to the upcoming Function layer
CommitRow(row interface{}) error
// Input gets the input source data for the currently executing Function in the Flow
Input() common.KisRowArr

// ++++++++++++++++++++++++++++++++++
// GetName gets the name of the Flow
GetName() string
// GetThisFunction gets the currently executing Function
GetThisFunction() Function
// GetThisFuncConf gets the configuration of the currently executing Function
GetThisFuncConf() *config.KisFuncConfig
}

func (flow *KisFlow) GetName() string {
return flow.Name
}

func (flow *KisFlow) GetThisFunction() kis.Function {
return flow.ThisFunction
}

func (flow *KisFlow) GetThisFuncConf() *config.KisFuncConfig {
return flow.ThisFunction.GetConfig()
}

package function

import (
“context”
“kis-flow/kis”
“kis-flow/log”
)

type KisFunctionC struct {
BaseFunction
}

func (f *KisFunctionC) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().InfoF(“KisFunctionC, flow = %+vn“, flow)

// Route to the specific executing computation Function through KisPool
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, “Function Called Error err = %sn“, err)
return err
}

return nil
}

package function

import (
“context”
“kis-flow/kis”
“kis-flow/log”
)

type KisFunctionE struct {
BaseFunction
}

func (f *KisFunctionE) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().InfoF(“KisFunctionE, flow = %+vn“, flow)

// Route to the specific executing computation Function through KisPool
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, “Function Called Error err = %sn“, err)
return err
}

return nil
}

package function

import (
“context”
“kis-flow/kis”
“kis-flow/log”
)

type KisFunctionL struct {
BaseFunction
}

func (f *KisFunctionL) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().InfoF(“KisFunctionL, flow = %+vn“, flow)

// Route to the specific executing computation Function through KisPool
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, “Function Called Error err = %sn“, err)
return err
}

return nil
}

package function

import (
“context”
“kis-flow/kis”
“kis-flow/log”
)

type KisFunctionS struct {
BaseFunction
}

func (f *KisFunctionS) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().InfoF(“KisFunctionS, flow = %+vn“, flow)

// Route to the specific executing computation Function through KisPool
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, “Function Called Error err = %sn“, err)
return err
}

return nil
}

package function

import (
“context”
“kis-flow/kis”
“kis-flow/log”
)

type KisFunctionV struct {
BaseFunction
}

func (f *KisFunctionV) Call(ctx context.Context, flow kis.Flow) error {
log.Logger().InfoF(“KisFunctionV, flow = %+vn“, flow)

// Route to the specific executing computation Function through KisPool
if err := kis.Pool().CallFunction(ctx, f.Config.FName, flow); err != nil {
log.Logger().ErrorFX(ctx, “Function Called Error err = %sn“, err)
return err
}

return nil
}

package test

import (
“context”
“fmt”
“kis-flow/common”
“kis-flow/config”
“kis-flow/flow”
“kis-flow/kis”
“testing”
)

func funcName1Handler(ctx context.Context, flow kis.Flow) error {
fmt.Println(“—> Call funcName1Handler —-“)

for index, row := range flow.Input() {
// Print data
str := fmt.Sprintf(“In FuncName = %s, FuncId = %s, row = %s”, flow.GetThisFuncConf().FName, flow.GetThisFunction().GetId(), row)
fmt.Println(str)

// Calculate result data
resultStr := fmt.Sprintf(“data from funcName[%s], index = %d”, flow.GetThisFuncConf().FName, index)

// Commit result data
_ = flow.CommitRow(resultStr)
}

return nil
}

func funcName2Handler(ctx context.Context, flow kis.Flow) error {

for _, row := range flow.Input() {
str := fmt.Sprintf(“In FuncName = %s, FuncId = %s, row = %s”, flow.GetThisFuncConf().FName, flow.GetThisFunction().GetId(), row)
fmt.Println(str)
}

return nil
}

func TestNewKisPool(t *testing.T) {

ctx := context.Background()

// 0. Register Functions
kis.Pool().FaaS(“funcName1”, funcName1Handler)
kis.Pool().FaaS(“funcName2”, funcName2Handler)

// 1. Create 2 KisFunction configuration instances
source1 := config.KisSource{
Name: “Public account data from TikTok Mall”,
Must: []string{“order_id”, “user_id”},
}

source2 := config.KisSource{
Name: “User order error rate”,
Must: []string{“order_id”, “user_id”},
}

myFuncConfig1 := config.NewFuncConfig(“funcName1”, common.C, &source1, nil)
if myFuncConfig1 == nil {
panic(“myFuncConfig1 is nil”)
}

myFuncConfig2 := config.NewFuncConfig(“funcName2”, common.E, &source2, nil)
if myFuncConfig2 == nil {
panic(“myFuncConfig2 is nil”)
}

// 2. Create a KisFlow configuration instance
myFlowConfig1 := config.NewFlowConfig(“flowName1”, common.FlowEnable)

// 3. Create a KisFlow object
flow1 := flow.NewKisFlow(myFlowConfig1)

// 4. Link Functions to Flow
if err := flow1.Link(myFuncConfig1, nil); err != nil {
panic(err)
}
if err := flow1.Link(myFuncConfig2, nil); err != nil {
panic(err)
}

// 5. Commit original data
_ = flow1.CommitRow(“This is Data1 from Test”)
_ = flow1.CommitRow(“This is Data2 from Test”)
_ = flow1.CommitRow(“This is Data3 from Test”)

// 6. Run flow1
if err := flow1.Run(ctx); err != nil {
panic(err)
}
}

go test -test.v -test.paniconexit0 -test.run TestNewKisPool

=== RUN TestNewKisPool
Add KisPool FuncName=funcName1
Add KisPool FuncName=funcName2
context.Background
====> After CommitSrcData, flow_name = flowName1, flow_id = flow-1fdae2bfac684f1d8edf89d9000208c0
All Level Data =
map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]]

KisFunctionC, flow = &{Id:flow-1fdae2bfac684f1d8edf89d9000208c0 Name:flowName1 Conf:0xc0000e27c0 Funcs:map[func-51527b72a4ee447fb0bd494bda9a84ad:0xc0000c0190 func-9cd2ab870b384794b312d2be10bb06fa:0xc0000c01e0] FlowHead:0xc0000c0190 FlowTail:0xc0000c01e0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0000c0190 ThisFunctionId:func-51527b72a4ee447fb0bd494bda9a84ad PrevFunctionId:FunctionIdFirstVirtual funcParams:map[func-51527b72a4ee447fb0bd494bda9a84ad:map[] func-9cd2ab870b384794b312d2be10bb06fa:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]] inPut:[This is Data1 from Test This is Data2 from Test This is Data3 from Test]}

—> Call funcName1Handler —-
In FuncName = funcName1, FuncId = func-51527b72a4ee447fb0bd494bda9a84ad, row = This is Data1 from Test
In FuncName = funcName1, FuncId = func-51527b72a4ee447fb0bd494bda9a84ad, row = This is Data2 from Test
In FuncName = funcName1, FuncId = func-51527b72a4ee447fb0bd494bda9a84ad, row = This is Data3 from Test
context.Background
====> After commitCurData, flow_name = flowName1, flow_id = flow-1fdae2bfac684f1d8edf89d9000208c0
All Level Data =
map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-51527b72a4ee447fb0bd494bda9a84ad:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]]

KisFunctionE, flow = &{Id:flow-1fdae2bfac684f1d8edf89d9000208c0 Name:flowName1 Conf:0xc0000e27c0 Funcs:map[func-51527b72a4ee447fb0bd494bda9a84ad:0xc0000c0190 func-9cd2ab870b384794b312d2be10bb06fa:0xc0000c01e0] FlowHead:0xc0000c0190 FlowTail:0xc0000c01e0 flock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} ThisFunction:0xc0000c01e0 ThisFunctionId:func-9cd2ab870b384794b312d2be10bb06fa PrevFunctionId:func-51527b72a4ee447fb0bd494bda9a84ad funcParams:map[func-51527b72a4ee447fb0bd494bda9a84ad:map[] func-9cd2ab870b384794b312d2be10bb06fa:map[]] fplock:{w:{state:0 sema:0} writerSem:0 readerSem:0 readerCount:0 readerWait:0} buffer:[] data:map[FunctionIdFirstVirtual:[This is Data1 from Test This is Data2 from Test This is Data3 from Test] func-51527b72a4ee447fb0bd494bda9a84ad:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]] inPut:[data from funcName[funcName1], index = 0 data from funcName[funcName1], index = 1 data from funcName[funcName1], index = 2]}

In FuncName = funcName2, FuncId = func-9cd2ab870b384794b312d2be10bb06fa, row = data from funcName[funcName1], index = 0
In FuncName = funcName2, FuncId = func-9cd2ab870b384794b312d2be10bb06fa, row = data from funcName[funcName1], index = 1
In FuncName = funcName2, FuncId = func-9cd2ab870b384794b312d2be10bb06fa, row = data from funcName[funcName1], index = 2
— PASS: TestNewKisPool (0.00s)
PASS
ok kis-flow/test 0.520s