@ -5,13 +5,12 @@ import (
"time"
"time"
"sync"
"sync"
"errors"
"errors"
"reflect"
"encoding/json"
"encoding/json"
) )
// Assumed that thread is already locked for signal
// Assumed that thread is already locked for signal
func ( thread * Simple Thread) Process ( context * StateContext , princ Node , signal GraphSignal ) error { func ( thread * ) Process ( context * StateContext , signal GraphSignal ) error {
err := thread . Simple Lockable. Process ( context , princ , signal )
err := thread . . Process ( context , signal )
if err != nil {
if err != nil {
return err
return err
}
}
@ -19,16 +18,16 @@ func (thread *SimpleThread) Process(context *StateContext, princ Node, signal Gr
switch signal . Direction ( ) {
switch signal . Direction ( ) {
case Up :
case Up :
err = UseStates ( context , thread , NewLockInfo ( thread , [ ] string { "parent" } ) , func ( context * StateContext ) error {
err = UseStates ( context , thread , NewLockInfo ( thread , [ ] string { "parent" } ) , func ( context * StateContext ) error {
if thread . p arent != nil {
if thread . P arent != nil {
return Signal ( context , thread . p arent, thread , signal )
return Signal ( context , thread . P arent, thread , signal )
} else {
} else {
return nil
return nil
}
}
} )
} )
case Down :
case Down :
err = UseStates ( context , thread , NewLockInfo ( thread , [ ] string { "children" } ) , func ( context * StateContext ) error {
err = UseStates ( context , thread , NewLockInfo ( thread , [ ] string { "children" } ) , func ( context * StateContext ) error {
for _ , child := range ( thread . c hildren) {
for _ , info := range ( thread . C hildren) {
err := Signal ( context , c hild, thread , signal )
err := Signal ( context , info. C hild, thread , signal )
if err != nil {
if err != nil {
return err
return err
}
}
@ -44,136 +43,35 @@ func (thread *SimpleThread) Process(context *StateContext, princ Node, signal Gr
return err
return err
}
}
thread . signal <- signal
thread . Chan <- signal
return nil
return nil
} }
// Interface to represent any type of thread information
type ThreadInfo interface {
}
func ( thread * SimpleThread ) SetTimeout ( timeout time . Time , action string ) {
thread . timeout = timeout
thread . timeout_action = action
thread . timeout_chan = time . After ( time . Until ( timeout ) )
}
func ( thread * SimpleThread ) TimeoutAction ( ) string {
return thread . timeout_action
}
func ( thread * SimpleThread ) State ( ) string {
return thread . state_name
}
func ( thread * SimpleThread ) SetState ( new_state string ) error {
if new_state == "" {
return fmt . Errorf ( "Cannot set state to '' with SetState" )
}
thread . state_name = new_state
return nil
}
func ( thread * SimpleThread ) Parent ( ) Thread {
return thread . parent
}
func ( thread * SimpleThread ) SetParent ( parent Thread ) {
thread . parent = parent
}
func ( thread * SimpleThread ) Children ( ) [ ] Thread {
return thread . children
}
func ( thread * SimpleThread ) Child ( id NodeID ) Thread {
for _ , child := range ( thread . children ) {
if child . ID ( ) == id {
return child
}
}
return nil
}
func ( thread * SimpleThread ) ChildInfo ( child NodeID ) ThreadInfo {
return thread . child_info [ child ]
}
// Requires thread and childs thread to be locked for write
// Requires thread and childs thread to be locked for write
func UnlinkThreads ( ctx * Context , thread Thread , child Thread ) error { func UnlinkThreads ( ctx * Context , node ThreadNode , child_node ThreadNode ) error {
var found Node = nil
thread := node . ThreadHandle ( )
for _ , c := range ( thread . Children ( ) ) {
child := child_node . ThreadHandle ( )
if child . ID ( ) == c . ID ( ) {
_ , is_child := thread . Children [ child_node . ID ( ) ]
found = c
if is_child == false {
break
}
}
if found == nil {
return fmt . Errorf ( "UNLINK_THREADS_ERR: %s is not a child of %s" , child . ID ( ) , thread . ID ( ) )
return fmt . Errorf ( "UNLINK_THREADS_ERR: %s is not a child of %s" , child . ID ( ) , thread . ID ( ) )
}
}
child . SetParent ( nil )
child . Parent = nil
thread . RemoveChild ( child )
delete ( thread . Children , child . ID ( ) )
return nil
return nil
} }
func ( thread * SimpleThread ) RemoveChild ( child Thread ) { func checkIfChild ( context * StateContext , target ThreadNode , cur ThreadNode ) bool {
idx := - 1
for _ , info := range ( cur . ThreadHandle ( ) . Children ) {
for i , c := range ( thread . children ) {
if info . Child . ID ( ) == target . ID ( ) {
if c . ID ( ) == child . ID ( ) {
idx = i
break
}
}
if idx == - 1 {
panic ( fmt . Sprintf ( "%s is not a child of %s" , child . ID ( ) , thread . Name ( ) ) )
}
child_len := len ( thread . children )
thread . children [ idx ] = thread . children [ child_len - 1 ]
thread . children = thread . children [ 0 : child_len - 1 ]
}
func ( thread * SimpleThread ) AddChild ( child Thread , info ThreadInfo ) error {
if child == nil {
return fmt . Errorf ( "Will not connect nil to the thread tree" )
}
_ , exists := thread . child_info [ child . ID ( ) ]
if exists == true {
return fmt . Errorf ( "Will not connect the same child twice" )
}
if info == nil && thread . InfoType != nil {
return fmt . Errorf ( "nil info passed when expecting info" )
} else if info != nil {
if reflect . TypeOf ( info ) != thread . InfoType {
return fmt . Errorf ( "info type mismatch, expecting %+v - %+v" , thread . InfoType , reflect . TypeOf ( info ) )
}
}
thread . children = append ( thread . children , child )
thread . child_info [ child . ID ( ) ] = info
return nil
}
func checkIfChild ( context * StateContext , target Thread , cur Thread ) bool {
for _ , child := range ( cur . Children ( ) ) {
if child . ID ( ) == target . ID ( ) {
return true
return true
}
}
is_child := false
is_child := false
UpdateStates ( context , cur , NewLockMap (
UpdateStates ( context , cur , NewLockMap (
NewLockInfo ( c hild, [ ] string { "children" } ) ,
NewLockInfo ( info . Child , [ ] string { "children" } ) ,
) , func ( context * StateContext ) error {
) , func ( context * StateContext ) error {
is_child = checkIfChild ( context , target , c hild)
is_child = checkIfChild ( context , target , info . Child )
return nil
return nil
} )
} )
if is_child {
if is_child {
@ -186,10 +84,12 @@ func checkIfChild(context *StateContext, target Thread, cur Thread) bool {
// Links child to parent with info as the associated info
// Links child to parent with info as the associated info
// Continues the write context with princ, getting children for thread and parent for child
// Continues the write context with princ, getting children for thread and parent for child
func LinkThreads ( context * StateContext , princ Node , thread Thread , child Thread , info Threa dInfo) error { func LinkThreads ( context * StateContext , princ Node , thread _node ThreadNode , info Chil dInfo) error {
if context == nil || thread == nil || c hild == nil {
if context == nil || thread _node == nil || info . C hild == nil {
return fmt . Errorf ( "invalid input" )
return fmt . Errorf ( "invalid input" )
}
}
thread := thread_node . ThreadHandle ( )
child := info . Child . ThreadHandle ( )
if thread . ID ( ) == child . ID ( ) {
if thread . ID ( ) == child . ID ( ) {
return fmt . Errorf ( "Will not link %s as a child of itself" , thread . ID ( ) )
return fmt . Errorf ( "Will not link %s as a child of itself" , thread . ID ( ) )
@ -199,7 +99,7 @@ func LinkThreads(context *StateContext, princ Node, thread Thread, child Thread,
child . ID ( ) : LockInfo { Node : child , Resources : [ ] string { "parent" } } ,
child . ID ( ) : LockInfo { Node : child , Resources : [ ] string { "parent" } } ,
thread . ID ( ) : LockInfo { Node : thread , Resources : [ ] string { "children" } } ,
thread . ID ( ) : LockInfo { Node : thread , Resources : [ ] string { "children" } } ,
} , func ( context * StateContext ) error {
} , func ( context * StateContext ) error {
if child . Parent ( ) != nil {
if child . Parent != nil {
return fmt . Errorf ( "EVENT_LINK_ERR: %s already has a parent, cannot link as child" , child . ID ( ) )
return fmt . Errorf ( "EVENT_LINK_ERR: %s already has a parent, cannot link as child" , child . ID ( ) )
}
}
@ -211,60 +111,23 @@ func LinkThreads(context *StateContext, princ Node, thread Thread, child Thread,
return fmt . Errorf ( "EVENT_LINK_ERR: %s is already a parent of %s so will not add again" , thread . ID ( ) , child . ID ( ) )
return fmt . Errorf ( "EVENT_LINK_ERR: %s is already a parent of %s so will not add again" , thread . ID ( ) , child . ID ( ) )
}
}
err := thread . AddChild ( child , info )
// TODO check for info types
if err != nil {
return fmt . Errorf ( "EVENT_LINK_ERR: error adding %s as child to %s: %e" , child . ID ( ) , thread . ID ( ) , err )
}
child . SetParent ( thread )
if err != nil {
thread . Children [ child . ID ( ) ] = info
return err
child . Parent = thread_node
}
return nil
return nil
} )
} )
} }
type ThreadAction func ( * Context , Thread ) ( string , error ) type ThreadAction func ( * Context , Thread Node ) ( string , error )
type ThreadActions map [ string ] ThreadAction type ThreadActions map [ string ] ThreadAction
type ThreadHandler func ( * Context , Thread , GraphSignal ) ( string , error ) type ThreadHandler func ( * Context , Thread Node , GraphSignal ) ( string , error )
type ThreadHandlers map [ string ] ThreadHandler type ThreadHandlers map [ string ] ThreadHandler
type Thread interface { type InfoType string
// All Threads are Lockables
func ( t InfoType ) String ( ) string {
Lockable
return string ( t )
/// State Modification Functions
SetParent ( parent Thread )
AddChild ( child Thread , info ThreadInfo ) error
RemoveChild ( child Thread )
SetState ( new_thread string ) error
SetTimeout ( end_time time . Time , action string )
/// State Reading Functions
Parent ( ) Thread
Children ( ) [ ] Thread
Child ( id NodeID ) Thread
ChildInfo ( child NodeID ) ThreadInfo
State ( ) string
TimeoutAction ( ) string
/// Functions that dont read/write thread
// Deserialize the attribute map from json.Unmarshal
DeserializeInfo ( ctx * Context , data [ ] byte ) ( ThreadInfo , error )
SetActive ( active bool ) error
Action ( action string ) ( ThreadAction , bool )
Handler ( signal_type string ) ( ThreadHandler , bool )
// Internal timeout channel for thread
Timeout ( ) <- chan time . Time
// Internal signal channel for thread
SignalChannel ( ) <- chan GraphSignal
ClearTimeout ( )
ChildWaits ( ) * sync . WaitGroup
}
type ParentInfo interface {
Parent ( ) * ParentThreadInfo
} }
// Data required by a parent thread to restore it's children
// Data required by a parent thread to restore it's children
@ -274,10 +137,6 @@ type ParentThreadInfo struct {
RestoreAction string ` json:"restore_action" `
RestoreAction string ` json:"restore_action" `
} }
func ( info * ParentThreadInfo ) Parent ( ) * ParentThreadInfo {
return info
}
func NewParentThreadInfo ( start bool , start_action string , restore_action string ) ParentThreadInfo { func NewParentThreadInfo ( start bool , start_action string , restore_action string ) ParentThreadInfo {
return ParentThreadInfo {
return ParentThreadInfo {
Start : start ,
Start : start ,
@ -286,83 +145,123 @@ func NewParentThreadInfo(start bool, start_action string, restore_action string)
}
}
} }
type SimpleThread struct { type ChildInfo struct {
SimpleLockable
Child ThreadNode
Infos map [ InfoType ] interface { }
}
actions ThreadActions
func NewChildInfo ( child ThreadNode , infos map [ InfoType ] interface { } ) ChildInfo {
handlers ThreadHandlers
if infos == nil {
infos = map [ InfoType ] interface { } { }
}
timeout_chan <- chan time . Time
return ChildInfo {
signal chan GraphSignal
Child : child ,
child_waits * sync . WaitGroup
Infos : infos ,
active bool
}
active_lock * sync . Mutex }
state_name string
type QueuedAction struct {
parent Thread
Timeout time . Time
children [ ] Thread
Action string
child_info map [ NodeID ] ThreadInfo
InfoType reflect . Type
timeout time . Time
timeout_action string
} }
func ( thread * SimpleThread ) Type ( ) NodeType { type ThreadNode interface {
LockableNode
ThreadHandle ( ) * Thread
}
type Thread struct {
Lockable
Actions ThreadActions
Handlers ThreadHandlers
TimeoutChan <- chan time . Time
Chan chan GraphSignal
ChildWaits sync . WaitGroup
Active bool
ActiveLock sync . Mutex
StateName string
Parent ThreadNode
Children map [ NodeID ] ChildInfo
InfoTypes [ ] InfoType
TimeoutAction string
Timeout time . Time
}
func ( thread * Thread ) ThreadHandle ( ) * Thread {
return thread
}
func ( thread * Thread ) Type ( ) NodeType {
return NodeType ( "simple_thread" )
return NodeType ( "simple_thread" )
} }
func ( thread * SimpleThread ) Serialize ( ) ( [ ] byte , error ) { func ( thread * ) Serialize ( ) ( [ ] byte , error ) {
thread_json := NewSimpleThreadJSON ( thread )
thread_json := New ThreadJSON( thread )
return json . MarshalIndent ( & thread_json , "" , " " )
return json . MarshalIndent ( & thread_json , "" , " " )
} }
func ( thread * SimpleThread ) SignalChannel ( ) <- chan GraphSignal { func ( thread * Thread ) ChildList ( ) [ ] ThreadNode {
return thread . signal
ret := make ( [ ] ThreadNode , len ( thread . Children ) )
i := 0
for _ , info := range ( thread . Children ) {
ret [ i ] = info . Child
i += 1
}
return ret
} }
type SimpleThreadJSON struct { type struct {
Parent string ` json:"parent" `
Parent string ` json:"parent" `
Children map [ string ] interface { } ` json:"children" `
Children map [ string ] map[ string ] interface{ } ` json:"children" `
Timeout time . Time ` json:"timeout" `
Timeout time . Time ` json:"timeout" `
TimeoutAction string ` json:"timeout_action" `
TimeoutAction string ` json:"timeout_action" `
StateName string ` json:"state_name" `
StateName string ` json:"state_name" `
SimpleLockableJSON
} }
func NewSimpleThreadJSON ( thread * SimpleThread ) SimpleThreadJSON { func NewThreadJSON ( thread * Thread ) ThreadJSON {
children := map [ string ] interface { } { }
children := map [ string ] map [ string ] interface { } { }
for _ , child := range ( thread . children ) {
for id , info := range ( thread . Children ) {
children [ child . ID ( ) . String ( ) ] = thread . child_info [ child . ID ( ) ]
tmp := map [ string ] interface { } { }
for name , i := range ( info . Infos ) {
tmp [ name . String ( ) ] = i
}
children [ id . String ( ) ] = tmp
}
}
parent_id := ""
parent_id := ""
if thread . parent != nil {
if thread . P arent != nil {
parent_id = thread . parent . ID ( ) . String ( )
parent_id = thread . P arent. ID ( ) . String ( )
}
}
lockable_json := New Simple LockableJSON( & thread . Simple Lockable)
lockable_json := New LockableJSON( & thread . )
return Simple ThreadJSON{
return {
Parent : parent_id ,
Parent : parent_id ,
Children : children ,
Children : children ,
Timeout : thread . t imeout,
Timeout : thread . T imeout,
TimeoutAction : thread . timeout_a ction,
TimeoutAction : thread . TimeoutA ction,
StateName : thread . state_n ame,
StateName : thread . StateN ame,
Simple LockableJSON: lockable_json ,
: lockable_json ,
}
}
} }
func Load Simple Thread( ctx * Context , id NodeID , data [ ] byte , nodes NodeMap ) ( Node , error ) { func Load Thread( ctx * Context , id NodeID , data [ ] byte , nodes NodeMap ) ( Node , error ) {
var j Simple ThreadJSON
var j
err := json . Unmarshal ( data , & j )
err := json . Unmarshal ( data , & j )
if err != nil {
if err != nil {
return nil , err
return nil , err
}
}
thread := New Simple Thread( id , j . Name , j . StateName , nil , BaseThreadActions , BaseThreadHandlers )
thread := New Thread( id , j . Name , j . StateName , nil , BaseThreadActions , BaseThreadHandlers )
nodes [ id ] = & thread
nodes [ id ] = & thread
err = Restore Simple Thread( ctx , & thread , j , nodes )
err = Restore Thread( ctx , & thread , j , nodes )
if err != nil {
if err != nil {
return nil , err
return nil , err
}
}
@ -370,17 +269,10 @@ func LoadSimpleThread(ctx *Context, id NodeID, data []byte, nodes NodeMap) (Node
return & thread , nil
return & thread , nil
} }
// SimpleThread has no associated info with children
func RestoreThread ( ctx * Context , thread * Thread , j ThreadJSON , nodes NodeMap ) error {
func ( thread * SimpleThread ) DeserializeInfo ( ctx * Context , data [ ] byte ) ( ThreadInfo , error ) {
if len ( data ) > 0 {
return nil , fmt . Errorf ( "SimpleThread expected to deserialize no info but got %d length data: %s" , len ( data ) , string ( data ) )
}
return nil , nil
}
func RestoreSimpleThread ( ctx * Context , thread Thread , j SimpleThreadJSON , nodes NodeMap ) error {
if j . TimeoutAction != "" {
if j . TimeoutAction != "" {
thread . SetTimeout ( j . Timeout , j . TimeoutAction )
thread . Timeout = j . Timeout
thread . TimeoutAction = j . TimeoutAction
}
}
if j . Parent != "" {
if j . Parent != "" {
@ -392,11 +284,11 @@ func RestoreSimpleThread(ctx *Context, thread Thread, j SimpleThreadJSON, nodes
if err != nil {
if err != nil {
return err
return err
}
}
p_t , ok := p . ( Thread )
p_t , ok := p . ( Thread Node )
if ok == false {
if ok == false {
return err
return err
}
}
thread . SetParent ( p_t )
thread . Parent = p_t
}
}
for id_str , info_raw := range ( j . Children ) {
for id_str , info_raw := range ( j . Children ) {
@ -404,63 +296,94 @@ func RestoreSimpleThread(ctx *Context, thread Thread, j SimpleThreadJSON, nodes
if err != nil {
if err != nil {
return err
return err
}
}
child_node , err := LoadNodeRecurse ( ctx , id , nodes )
child_node , err := LoadNodeRecurse ( ctx , id , nodes )
if err != nil {
if err != nil {
return err
return err
}
}
child_t , ok := child_node . ( Thread )
child_t , ok := child_node . ( ThreadNode )
if ok == false {
if ok == false {
return fmt . Errorf ( "%+v is not a Thread as expected" , child_node )
return fmt . Errorf ( "%+v is not a Thread as expected" , child_node )
}
}
var info_ser [ ] byte
parsed_info , err := DeserializeChildInfo ( ctx , info_raw )
if info_raw != nil {
info_ser , err = json . Marshal ( info_raw )
if err != nil {
if err != nil {
return err
return err
}
}
thread . Children [ id ] = ChildInfo { child_t , parsed_info }
}
}
parsed_info , err := thread . DeserializeInfo ( ctx , info_ser )
return RestoreLockable ( ctx , & thread . Lockable , j . LockableJSON , nodes )
if err != nil {
return err
} }
thread . AddChild ( child_t , parsed_info )
var deserializers = map [ InfoType ] func ( interface { } ) ( interface { } , error ) {
}
func DeserializeChildInfo ( ctx * Context , infos_raw map [ string ] interface { } ) ( map [ InfoType ] interface { } , error ) {
ret := map [ InfoType ] interface { } { }
for type_str , info_raw := range ( infos_raw ) {
info_type := InfoType ( type_str )
deserializer , exists := deserializers [ info_type ]
if exists == false {
return nil , fmt . Errorf ( "No deserializer for %s" , info_type )
}
var err error
ret [ info_type ] , err = deserializer ( info_raw )
if err != nil {
return nil , err
}
}
}
return RestoreSimpleLockable ( ctx , thread , j . SimpleLockableJSON , nodes )
return ret , nil
} }
const THREAD_SIGNAL_BUFFER_SIZE = 128 const THREAD_SIGNAL_BUFFER_SIZE = 128
func NewSimpleThread ( id NodeID , name string , state_name string , info_type reflect . Type , actions ThreadActions , handlers ThreadHandlers ) SimpleThread { func NewThread ( id NodeID , name string , state_name string , info_types [ ] InfoType , actions ThreadActions , handlers ThreadHandlers ) Thread {
return SimpleThread {
return Thread {
SimpleLockable : NewSimpleLockable ( id , name ) ,
Lockable : NewLockable ( id , name ) ,
InfoType : info_type ,
InfoTypes : info_types ,
state_name : state_name ,
StateName : state_name ,
signal : make ( chan GraphSignal , THREAD_SIGNAL_BUFFER_SIZE ) ,
Chan : make ( chan GraphSignal , THREAD_SIGNAL_BUFFER_SIZE ) ,
children : [ ] Thread { } ,
Children : map [ NodeID ] ChildInfo { } ,
child_info : map [ NodeID ] ThreadInfo { } ,
Actions : actions ,
actions : actions ,
Handlers : handlers ,
handlers : handlers ,
}
child_waits : & sync . WaitGroup { } ,
}
active_lock : & sync . Mutex { } ,
func ( thread * Thread ) SetActive ( active bool ) error {
thread . ActiveLock . Lock ( )
defer thread . ActiveLock . Unlock ( )
if thread . Active == true && active == true {
return fmt . Errorf ( "%s is active, cannot set active" , thread . ID ( ) )
} else if thread . Active == false && active == false {
return fmt . Errorf ( "%s is already inactive, canot set inactive" , thread . ID ( ) )
}
}
thread . Active = active
return nil
}
func ( thread * Thread ) SetState ( state string ) error {
thread . StateName = state
return nil
} }
// Requires the read permission of threads children
// Requires the read permission of threads children
func FindChild ( context * StateContext , princ Node , thread Thread , id NodeID ) Thread { func FindChild ( context * StateContext , princ Node , node ThreadNode , id NodeID ) Thread Node {
if thread == nil {
if node == nil {
panic ( "cannot recurse through nil" )
panic ( "cannot recurse through nil" )
}
}
thread := node . ThreadHandle ( )
if id == thread . ID ( ) {
if id == thread . ID ( ) {
return thread
return thread
}
}
for _ , child := range thread . Children ( ) {
for _ , info := range thread . Children {
var result Thread = nil
var result Thread Node
UseStates ( context , princ , NewLockInfo ( c hild, [ ] string { "children" } ) , func ( context * StateContext ) error {
UseStates ( context , princ , NewLockInfo ( info. C hild, [ ] string { "children" } ) , func ( context * StateContext ) error {
result = FindChild ( context , princ , c hild, id )
result = FindChild ( context , princ , info. C hild, id )
return nil
return nil
} )
} )
if result != nil {
if result != nil {
@ -471,11 +394,11 @@ func FindChild(context *StateContext, princ Node, thread Thread, id NodeID) Thre
return nil
return nil
} }
func ChildGo ( ctx * Context , thread Thread , child Thread , first_action string ) { func ChildGo ( ctx * Context , thread * Thread , child Thread Node , first_action string ) {
thread . ChildWaits ( ) . Add ( 1 )
thread . ChildWaits . Add ( 1 )
go func ( child Thread ) {
go func ( child Thread Node ) {
ctx . Log . Logf ( "thread" , "THREAD_START_CHILD: %s from %s" , thread . ID ( ) , child . ID ( ) )
ctx . Log . Logf ( "thread" , "THREAD_START_CHILD: %s from %s" , thread . ID ( ) , child . ID ( ) )
defer thread . ChildWaits ( ) . Done ( )
defer thread . ChildWaits . Done ( )
err := ThreadLoop ( ctx , child , first_action )
err := ThreadLoop ( ctx , child , first_action )
if err != nil {
if err != nil {
ctx . Log . Logf ( "thread" , "THREAD_CHILD_RUN_ERR: %s %e" , child . ID ( ) , err )
ctx . Log . Logf ( "thread" , "THREAD_CHILD_RUN_ERR: %s %e" , child . ID ( ) , err )
@ -486,8 +409,9 @@ func ChildGo(ctx * Context, thread Thread, child Thread, first_action string) {
} }
// Main Loop for Threads, starts a write context, so cannot be called from a write or read context
// Main Loop for Threads, starts a write context, so cannot be called from a write or read context
func ThreadLoop ( ctx * Context , thread Thread , first_action string ) error { func ThreadLoop ( ctx * Context , node ThreadNode , first_action string ) error {
// Start the thread, error if double-started
// Start the thread, error if double-started
thread := node . ThreadHandle ( )
ctx . Log . Logf ( "thread" , "THREAD_LOOP_START: %s - %s" , thread . ID ( ) , first_action )
ctx . Log . Logf ( "thread" , "THREAD_LOOP_START: %s - %s" , thread . ID ( ) , first_action )
err := thread . SetActive ( true )
err := thread . SetActive ( true )
if err != nil {
if err != nil {
@ -496,14 +420,14 @@ func ThreadLoop(ctx * Context, thread Thread, first_action string) error {
}
}
next_action := first_action
next_action := first_action
for next_action != "" {
for next_action != "" {
action , exists := thread . Action ( next_action )
action , exists := thread . Actions [ next_action ]
if exists == false {
if exists == false {
error_str := fmt . Sprintf ( "%s is not a valid action" , next_action )
error_str := fmt . Sprintf ( "%s is not a valid action" , next_action )
return errors . New ( error_str )
return errors . New ( error_str )
}
}
ctx . Log . Logf ( "thread" , "THREAD_ACTION: %s - %s" , thread . ID ( ) , next_action )
ctx . Log . Logf ( "thread" , "THREAD_ACTION: %s - %s" , thread . ID ( ) , next_action )
next_action , err = action ( ctx , thread )
next_action , err = action ( ctx , node )
if err != nil {
if err != nil {
return err
return err
}
}
@ -523,52 +447,8 @@ func ThreadLoop(ctx * Context, thread Thread, first_action string) error {
return nil
return nil
} }
func ThreadChildLinked ( ctx * Context , node ThreadNode , signal GraphSignal ) ( string , error ) {
func ( thread * SimpleThread ) ChildWaits ( ) * sync . WaitGroup { thread := node . ThreadHandle ( )
return thread . child_waits
}
func ( thread * SimpleThread ) SetActive ( active bool ) error {
thread . active_lock . Lock ( )
defer thread . active_lock . Unlock ( )
if thread . active == true && active == true {
return fmt . Errorf ( "%s is active, cannot set active" , thread . ID ( ) )
} else if thread . active == false && active == false {
return fmt . Errorf ( "%s is already inactive, canot set inactive" , thread . ID ( ) )
}
thread . active = active
return nil
}
func ( thread * SimpleThread ) Action ( action string ) ( ThreadAction , bool ) {
action_fn , exists := thread . actions [ action ]
return action_fn , exists
}
func ( thread * SimpleThread ) Handler ( signal_type string ) ( ThreadHandler , bool ) {
handler , exists := thread . handlers [ signal_type ]
return handler , exists
}
func ( thread * SimpleThread ) Timeout ( ) <- chan time . Time {
return thread . timeout_chan
}
func ( thread * SimpleThread ) ClearTimeout ( ) {
thread . timeout_chan = nil
thread . timeout_action = ""
}
func ( thread * SimpleThread ) AllowedToTakeLock ( new_owner Lockable , lockable Lockable ) bool {
for _ , child := range ( thread . children ) {
if new_owner . ID ( ) == child . ID ( ) {
return true
}
}
return false
}
func ThreadParentChildLinked ( ctx * Context , thread Thread , signal GraphSignal ) ( string , error ) {
ctx . Log . Logf ( "thread" , "THREAD_CHILD_LINKED: %+v" , signal )
ctx . Log . Logf ( "thread" , "THREAD_CHILD_LINKED: %+v" , signal )
context := NewWriteContext ( ctx )
context := NewWriteContext ( ctx )
err := UpdateStates ( context , thread , NewLockMap (
err := UpdateStates ( context , thread , NewLockMap (
@ -579,18 +459,18 @@ func ThreadParentChildLinked(ctx *Context, thread Thread, signal GraphSignal) (s
ctx . Log . Logf ( "thread" , "THREAD_NODE_LINKED_BAD_CAST" )
ctx . Log . Logf ( "thread" , "THREAD_NODE_LINKED_BAD_CAST" )
return nil
return nil
}
}
info _if := thread . ChildInfo ( sig . ID )
info , exists := thread . Children [ sig . ID ]
if info_if == nil {
if exists == false {
ctx . Log . Logf ( "thread" , "THREAD_NODE_LINKED: %s is not a child of %s" , sig . ID )
ctx . Log . Logf ( "thread" , "THREAD_NODE_LINKED: %s is not a child of %s" , sig . ID )
return nil
return nil
}
}
info_r, correct := info_if . ( Parent Info)
parent_info, exists := info . Infos [ "parent" ] . ( * ParentThread Info)
if correct == false {
if exists == false {
ctx . Log . Logf ( "thread" , "THREAD_NODE_LINKED_BAD_INFO_CAST )
panic ( "ran ThreadChildLinked from a thread that doesn't require 'parent' child info. library party foul )
}
}
info := info_r . Parent ( )
if . Start == true {
if parent_ info. Start == true {
ChildGo ( ctx , thread , thread. Child ( sig . ID ) , info. StartAction )
ChildGo ( ctx , thread , info. Child , parent_ info. StartAction )
}
}
return nil
return nil
} )
} )
@ -603,38 +483,30 @@ func ThreadParentChildLinked(ctx *Context, thread Thread, signal GraphSignal) (s
return "wait" , nil
return "wait" , nil
} }
func ThreadParentStartChild ( ctx * Context , thread Thread , signal GraphSignal ) ( string , error ) { // Helper function to start a child from a thread during a signal handler
ctx . Log . Logf ( "thread" , "THREAD_START_CHILD" )
// Starts a write context, so cannot be called from either a write or read context
func ThreadStartChild ( ctx * Context , node ThreadNode , signal GraphSignal ) ( string , error ) {
sig , ok := signal . ( StartChildSignal )
sig , ok := signal . ( StartChildSignal )
if ok == false {
if ok == false {
ctx . Log . Logf ( "thread" , "THREAD_START_CHILD_BAD_SIGNAL: %+v" , signal )
return "wait" , nil
return "wait" , nil
}
}
err := ThreadStartChild ( ctx , thread , sig )
thread := node . ThreadHandle ( )
if err != nil {
ctx . Log . Logf ( "thread" , "THREAD_START_CHILD_ERR: %s" , err )
} else {
ctx . Log . Logf ( "thread" , "THREAD_START_CHILD: %s" , sig . ID . String ( ) )
}
return "wait" , nil
}
// Helper function to start a child from a thread during a signal handler
// Starts a write context, so cannot be called from either a write or read context
func ThreadStartChild ( ctx * Context , thread Thread , signal StartChildSignal ) error {
context := NewWriteContext ( ctx )
context := NewWriteContext ( ctx )
return UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "children" } ) , func ( context * StateContext ) error {
return "wait" , UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "children" } ) , func ( context * StateContext ) error {
child := thread . Child ( signal . ID )
info , exists := thread . Children [ sig . ID ]
if child == nil {
if exists == false {
return fmt . Errorf ( "%s is not a child of %s" , sig nal . ID , thread . ID ( ) )
return fmt . Errorf ( "%s is not a child of %s" , sig . ID , thread . ID ( ) )
}
}
return UpdateStates ( context , thread , NewLockInfo ( c hild, [ ] string { "start" } ) , func ( context * StateContext ) error {
return UpdateStates ( context , thread , NewLockInfo ( info . Child , [ ] string { "start" } ) , func ( context * StateContext ) error {
info := thread . ChildInfo ( signal . ID ) . ( * ParentThreadInfo )
parent_info , exists := info . Infos [ "parent" ] . ( * ParentThreadInfo )
info . Start = true
if exists == false {
ChildGo ( ctx , thread , child , signal . Action )
return fmt . Errorf ( "Called ThreadStartChild from a thread that doesn't require parent child info" )
}
parent_info . Start = true
ChildGo ( ctx , thread , info . Child , sig . Action )
return nil
return nil
} )
} )
@ -643,18 +515,19 @@ func ThreadStartChild(ctx *Context, thread Thread, signal StartChildSignal) erro
// Helper function to restore threads that should be running from a parents restore action
// Helper function to restore threads that should be running from a parents restore action
// Starts a write context, so cannot be called from either a write or read context
// Starts a write context, so cannot be called from either a write or read context
func ThreadRestore ( ctx * Context , thread Thread , start bool ) error { func ThreadRestore ( ctx * Context , node ThreadNode , start bool ) error {
thread := node . ThreadHandle ( )
context := NewWriteContext ( ctx )
context := NewWriteContext ( ctx )
return UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "children" } ) , func ( context * StateContext ) error {
return UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "children" } ) , func ( context * StateContext ) error {
return UpdateStates ( context , thread , LockList ( thread . Child ren ( ) , [ ] string { "start" } ) , func ( context * StateContext ) error {
return UpdateStates ( context , thread , LockList ( thread . Child List ( ) , [ ] string { "start" } ) , func ( context * StateContext ) error {
for _ , child := range ( thread . Children ( ) ) {
for _ , info := range ( thread . Children ) {
info := ( thread . ChildInfo ( child . ID ( ) ) ) . ( ParentInfo ) . Parent ( )
parent_info := info . Infos [ "parent" ] . ( * ParentThreadInfo )
if . Start == true && child. State ( ) != "finished" {
if parent_ info. Start == true && info. Child . ThreadHandle ( ) . StateName != "finished" {
ctx . Log . Logf ( "thread" , "THREAD_RESTORED: %s -> %s" , thread . ID ( ) , c hild. ID ( ) )
ctx . Log . Logf ( "thread" , "THREAD_RESTORED: %s -> %s" , thread . ID ( ) , info. C hild. ID ( ) )
if start == true {
if start == true {
ChildGo ( ctx , thread , child, info. StartAction )
ChildGo ( ctx , thread , info. Child , parent_ info. StartAction )
} else {
} else {
ChildGo ( ctx , thread , child, info. RestoreAction )
ChildGo ( ctx , thread , info. Child , parent_ info. RestoreAction )
}
}
}
}
}
}
@ -665,10 +538,12 @@ func ThreadRestore(ctx * Context, thread Thread, start bool) error {
// Helper function to be called during a threads start action, sets the thread state to started
// Helper function to be called during a threads start action, sets the thread state to started
// Starts a write context, so cannot be called from either a write or read context
// Starts a write context, so cannot be called from either a write or read context
func ThreadStart ( ctx * Context , thread Thread ) error { // Returns "wait", nil on success, so the first return value can be ignored safely
func ThreadStart ( ctx * Context , node ThreadNode ) ( string , error ) {
thread := node . ThreadHandle ( )
context := NewWriteContext ( ctx )
context := NewWriteContext ( ctx )
return UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "state" } ) , func ( context * StateContext ) error {
return "wait" , UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "state" } ) , func ( context * StateContext ) error {
err := LockLockables ( context , [ ] Lockable { thread } , thread )
err := LockLockables ( context , map [ NodeID ] Lockable Node { thread . ID ( ) : thread } , thread )
if err != nil {
if err != nil {
return err
return err
}
}
@ -676,39 +551,28 @@ func ThreadStart(ctx * Context, thread Thread) error {
} )
} )
} }
func ThreadDefaultStart ( ctx * Context , thread Thread ) ( string , error ) { func ThreadWait ( ctx * Context , node ThreadNode ) ( string , error ) {
ctx . Log . Logf ( "thread" , "THREAD_DEFAULT_START: %s" , thread . ID ( ) )
thread := node . ThreadHandle ( )
err := ThreadStart ( ctx , thread )
ctx . Log . Logf ( "thread" , "THREAD_WAIT: %s TIMEOUT: %+v" , thread . ID ( ) , thread . Timeout )
if err != nil {
return "" , err
}
return "wait" , nil
}
func ThreadDefaultRestore ( ctx * Context , thread Thread ) ( string , error ) {
ctx . Log . Logf ( "thread" , "THREAD_DEFAULT_RESTORE: %s" , thread . ID ( ) )
return "wait" , nil
}
func ThreadWait ( ctx * Context , thread Thread ) ( string , error ) {
ctx . Log . Logf ( "thread" , "THREAD_WAIT: %s TIMEOUT: %+v" , thread . ID ( ) , thread . Timeout ( ) )
for {
for {
select {
select {
case signal := <- thread . SignalChannel ( ) :
case signal := <- thread . Chan :
ctx . Log . Logf ( "thread" , "THREAD_SIGNAL: %s %+v" , thread . ID ( ) , signal )
ctx . Log . Logf ( "thread" , "THREAD_SIGNAL: %s %+v" , thread . ID ( ) , signal )
signal_fn , exists := thread . Handler ( signal . Type ( ) )
signal_fn , exists := thread . Handlers [ signal . Type ( ) ]
if exists == true {
if exists == true {
ctx . Log . Logf ( "thread" , "THREAD_HANDLER: %s - %s" , thread . ID ( ) , signal . Type ( ) )
ctx . Log . Logf ( "thread" , "THREAD_HANDLER: %s - %s" , thread . ID ( ) , signal . Type ( ) )
return signal_fn ( ctx , thread , signal )
return signal_fn ( ctx , thread , signal )
} else {
} else {
ctx . Log . Logf ( "thread" , "THREAD_NOHANDLER: %s - %s" , thread . ID ( ) , signal . Type ( ) )
ctx . Log . Logf ( "thread" , "THREAD_NOHANDLER: %s - %s" , thread . ID ( ) , signal . Type ( ) )
}
}
case <- thread . Timeout ( ) :
case <- thread . TimeoutChan :
timeout_action := ""
timeout_action := ""
context := NewWriteContext ( ctx )
context := NewWriteContext ( ctx )
err := UpdateStates ( context , thread , NewLockMap ( NewLockInfo ( thread , [ ] string { "timeout" } ) ) , func ( context * StateContext ) error {
err := UpdateStates ( context , thread , NewLockMap ( NewLockInfo ( thread , [ ] string { "timeout" } ) ) , func ( context * StateContext ) error {
timeout_action = thread . TimeoutAction ( )
timeout_action = thread . TimeoutAction
thread . ClearTimeout ( )
thread . TimeoutChan = nil
thread . TimeoutAction = ""
thread . Timeout = time . Time { }
return nil
return nil
} )
} )
if err != nil {
if err != nil {
@ -720,26 +584,23 @@ func ThreadWait(ctx * Context, thread Thread) (string, error) {
}
}
} }
func ThreadDefaultFinish ( ctx * Context , thread Thread ) ( string , error ) { func ThreadFinish ( ctx * Context , node ThreadNode ) ( string , error ) {
ctx . Log . Logf ( "thread" , "THREAD_DEFAULT_FINISH: %s" , thread . ID ( ) . String ( ) )
thread := node . ThreadHandle ( )
return "" , ThreadFinish ( ctx , thread )
}
func ThreadFinish ( ctx * Context , thread Thread ) error {
context := NewWriteContext ( ctx )
context := NewWriteContext ( ctx )
return UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "state" } ) , func ( context * StateContext ) error {
return "" , UpdateStates ( context , thread , NewLockInfo ( thread , [ ] string { "state" } ) , func ( context * StateContext ) error {
err := thread . SetState ( "finished" )
err := thread . SetState ( "finished" )
if err != nil {
if err != nil {
return err
return err
}
}
return UnlockLockables ( context , [ ] Lockable { thread } , thread )
return UnlockLockables ( context , map [ NodeID ] Lockable Node { thread . ID ( ) : thread } , thread )
} )
} )
} }
var ThreadAbortedError = errors . New ( "Thread aborted by signal" ) var ThreadAbortedError = errors . New ( "Thread aborted by signal" )
// Default thread action function for "abort", sends a signal and returns a ThreadAbortedError
// Default thread action function for "abort", sends a signal and returns a ThreadAbortedError
func ThreadAbort ( ctx * Context , thread Thread , signal GraphSignal ) ( string , error ) { func ThreadAbort ( ctx * Context , node ThreadNode , signal GraphSignal ) ( string , error ) {
thread := node . ThreadHandle ( )
context := NewReadContext ( ctx )
context := NewReadContext ( ctx )
err := Signal ( context , thread , thread , NewStatusSignal ( "aborted" , thread . ID ( ) ) )
err := Signal ( context , thread , thread , NewStatusSignal ( "aborted" , thread . ID ( ) ) )
if err != nil {
if err != nil {
@ -749,38 +610,18 @@ func ThreadAbort(ctx * Context, thread Thread, signal GraphSignal) (string, erro
} }
// Default thread action for "stop", sends a signal and returns no error
// Default thread action for "stop", sends a signal and returns no error
func ThreadStop ( ctx * Context , thread Thread , signal GraphSignal ) ( string , error ) { func ThreadStop ( ctx * Context , node ThreadNode , signal GraphSignal ) ( string , error ) {
thread := node . ThreadHandle ( )
context := NewReadContext ( ctx )
context := NewReadContext ( ctx )
err := Signal ( context , thread , thread , NewStatusSignal ( "stopped" , thread . ID ( ) ) )
err := Signal ( context , thread , thread , NewStatusSignal ( "stopped" , thread . ID ( ) ) )
return "finish" , err
return "finish" , err
} }
// Copy the default thread actions to a new ThreadActions map
func NewThreadActions ( ) ThreadActions {
actions := ThreadActions { }
for k , v := range ( BaseThreadActions ) {
actions [ k ] = v
}
return actions
}
// Copy the defult thread handlers to a new ThreadAction map
func NewThreadHandlers ( ) ThreadHandlers {
handlers := ThreadHandlers { }
for k , v := range ( BaseThreadHandlers ) {
handlers [ k ] = v
}
return handlers
}
// Default thread actions
// Default thread actions
var BaseThreadActions = ThreadActions { var BaseThreadActions = ThreadActions {
"wait" : ThreadWait ,
"wait" : ThreadWait ,
"start" : ThreadDefaultStart ,
"start" : ThreadStart ,
"finish" : ThreadDefaultFinish ,
"finish" : ThreadFinish ,
"restore" : ThreadDefaultRestore ,
} }
// Default thread signal handlers
// Default thread signal handlers
