component.go

// The flow package is a framework for Flow-based programming in Go.
package flow

import (
	"reflect"
	"sync"
)

const (
	// ComponentModeUndefined stands for a fallback component mode (Async).
	ComponentModeUndefined = iota
	// ComponentModeAsync stands for asynchronous functioning mode.
	ComponentModeAsync
	// ComponentModeSync stands for synchronous functioning mode.
	ComponentModeSync
	// ComponentModePool stands for async functioning with a fixed pool.
	ComponentModePool
)

// DefaultComponentMode is the preselected functioning mode of all components being run.
var DefaultComponentMode = ComponentModeAsync

// Component is a generic flow component that has to be contained in concrete components.
// It stores network-specific information.
type Component struct {
	// Net is a pointer to network to inform it when the process is started and over
	// or to change its structure at run time.
	Net *Graph
	// Mode is component's functioning mode.
	Mode int8
	// PoolSize is used to define pool size when using ComponentModePool.
	PoolSize uint8
}

// Initalizable is the interface implemented by components/graphs with custom initialization code.
type Initializable interface {
	Init()
}

// Finalizable is the interface implemented by components/graphs with extra finalization code.
type Finalizable interface {
	Finish()
}

// Shutdowner is the interface implemented by components overriding default Shutdown() behavior.
type Shutdowner interface {
	Shutdown()
}

// postHandler is used to bind handlers to a port
type portHandler struct {
	onRecv  reflect.Value
	onClose reflect.Value
}

// RunProc runs event handling loop on component ports.
// It returns true on success or panics with error message and returns false on error.
func RunProc(c interface{}) bool {
	// Check if passed interface is a valid pointer to struct
	v := reflect.ValueOf(c)
	if v.Kind() != reflect.Ptr || v.IsNil() {
		panic("Argument of flow.Run() is not a valid pointer")
		return false
	}
	vp := v
	v = v.Elem()
	if v.Kind() != reflect.Struct {
		panic("Argument of flow.Run() is not a valid pointer to structure. Got type: " + vp.Type().Name())
		return false
	}
	t := v.Type()

	// Get internal state lock if available
	hasLock := false
	var locker sync.Locker
	if lockField := v.FieldByName("StateLock"); lockField.IsValid() && lockField.Elem().IsValid() {
		locker, hasLock = lockField.Interface().(sync.Locker)
	}

	// Call user init function if exists
	if initable, ok := c.(Initializable); ok {
		initable.Init()
	}

	// A group to wait for all inputs to be closed
	inputsClose := new(sync.WaitGroup)
	// A group to wait for all recv handlers to finish
	handlersDone := new(sync.WaitGroup)

	// Get the embedded flow.Component
	vCom := v.FieldByName("Component")
	isComponent := vCom.IsValid() && vCom.Type().Name() == "Component"

	if !isComponent {
		panic("Argument of flow.Run() is not a flow.Component")
	}

	// Get the component mode
	componentMode := DefaultComponentMode
	var poolSize uint8 = 0
	if isComponent {
		if vComMode := vCom.FieldByName("Mode"); vComMode.IsValid() {
			componentMode = int(vComMode.Int())
		}
		if vComPoolSize := vCom.FieldByName("PoolSize"); vComPoolSize.IsValid() {
			poolSize = uint8(vComPoolSize.Uint())
		}
	}

	// Create a slice of select cases and port handlers
	cases := make([]reflect.SelectCase, 0, t.NumField())
	handlers := make([]portHandler, 0, t.NumField())

	// Iterate over struct fields and bind handlers
	for i := 0; i < t.NumField(); i++ {
		fv := v.Field(i)
		ff := t.Field(i)
		ft := fv.Type()
		// Detect control channels
		if fv.IsValid() && fv.Kind() == reflect.Chan && !fv.IsNil() && (ft.ChanDir()&reflect.RecvDir) != 0 {
			// Bind handlers for an input channel
			cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: fv})
			h := portHandler{onRecv: vp.MethodByName("On" + ff.Name), onClose: vp.MethodByName("On" + ff.Name + "Close")}
			handlers = append(handlers, h)
			if h.onClose.IsValid() || h.onRecv.IsValid() {
				// Add the input to the wait group
				inputsClose.Add(1)
			}
		}
	}

	// Prepare handler closures
	recvHandler := func(onRecv, value reflect.Value) {
		if hasLock {
			locker.Lock()
		}
		valArr := [1]reflect.Value{value}
		onRecv.Call(valArr[:])
		if hasLock {
			locker.Unlock()
		}
		handlersDone.Done()
	}
	closeHandler := func(onClose reflect.Value) {
		if onClose.IsValid() {
			// Lock the state and call OnClose handler
			if hasLock {
				locker.Lock()
			}
			onClose.Call([]reflect.Value{})
			if hasLock {
				locker.Unlock()
			}
		}
		inputsClose.Done()
	}

	// Run the port handlers depending on component mode
	if componentMode == ComponentModePool && poolSize > 0 {
		// Pool mode, prefork limited goroutine pool for all inputs
		var poolIndex uint8
		poolWait := new(sync.WaitGroup)
		once := new(sync.Once)
		for poolIndex = 0; poolIndex < poolSize; poolIndex++ {
			poolWait.Add(1)
			go func() {
				for {
					chosen, recv, recvOK := reflect.Select(cases)
					if !recvOK {
						// Port has been closed
						poolWait.Done()
						once.Do(func() {
							// Wait for other workers
							poolWait.Wait()
							// Close output down
							closeHandler(handlers[chosen].onClose)
						})
						return
					}
					if handlers[chosen].onRecv.IsValid() {
						handlersDone.Add(1)
						recvHandler(handlers[chosen].onRecv, recv)
					}
				}
			}()
		}
	} else {
		go func() {
			for {
				chosen, recv, recvOK := reflect.Select(cases)
				if !recvOK {
					// Port has been closed
					closeHandler(handlers[chosen].onClose)
					return
				}
				if handlers[chosen].onRecv.IsValid() {
					handlersDone.Add(1)
					if componentMode == ComponentModeAsync || componentMode == ComponentModeUndefined && DefaultComponentMode == ComponentModeAsync {
						// Async mode
						go recvHandler(handlers[chosen].onRecv, recv)
					} else {
						// Sync mode
						recvHandler(handlers[chosen].onRecv, recv)
					}
				}
			}
		}()
	}

	go func() {
		// Wait for all inputs to be closed
		inputsClose.Wait()
		// Wait all inport handlers to finish their job
		handlersDone.Wait()

		// Call shutdown handler (user or default)
		shutdownProc(c)

		// Get the embedded flow.Component and check if it belongs to a network
		if isComponent {
			if vNet := vCom.FieldByName("Net"); vNet.IsValid() && !vNet.IsNil() {
				if vNetCtr, hasNet := vNet.Interface().(netController); hasNet {
					// Remove the instance from the network's WaitGroup
					vNetCtr.getWait().Done()
				}
			}
		}
	}()
	return true
}

// closePorts closes all output channels of a process.
func closePorts(c interface{}) {
	v := reflect.ValueOf(c).Elem()
	t := v.Type()
	// Iterate over struct fields
	for i := 0; i < t.NumField(); i++ {
		fv := v.Field(i)
		ft := fv.Type()
		// Detect and close send-only channels
		if fv.IsValid() {
			if fv.Kind() == reflect.Chan && (ft.ChanDir()&reflect.SendDir) != 0 && (ft.ChanDir()&reflect.RecvDir) == 0 {
				fv.Close()
			} else if fv.Kind() == reflect.Slice && ft.Elem().Kind() == reflect.Chan {
				ll := fv.Len()
				for i := 0; i < ll; i += 1 {
					fv.Index(i).Close()
				}
			}
		}
	}
}

// shutdownProc represents a standard process shutdown procedure.
func shutdownProc(c interface{}) {
	if s, ok := c.(Shutdowner); ok {
		// Custom shutdown behavior
		s.Shutdown()
	} else {
		// Call user finish function if exists
		if finable, ok := c.(Finalizable); ok {
			finable.Finish()
		}
		// Close all output ports
		closePorts(c)
	}
}