From 81223929fdbddc1690908c80e586e69cf409d047 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Fri, 8 Oct 2021 20:20:41 +0300 Subject: [PATCH 01/13] core: port definitions and handle more edge cases in Connect This minimizes port definitions such that the in/out array connection don't share the unneeded methods with in/out connection. This prevents using in array and out array as an argument to .Send or .Receive, preventing some bugs. Handle possible misconnections in Connect. --- core/connection.go | 14 ---- core/inarrayport.go | 16 ++--- core/initializationconnection.go | 14 ---- core/inputconn.go | 16 +++-- core/network.go | 112 +++++++++++++++++-------------- core/nulloutport.go | 16 ----- core/outarrayport.go | 16 ++--- core/outport.go | 16 +---- core/outputconn.go | 16 ++--- core/process.go | 86 +++++++----------------- testdata.copy | 1 - 11 files changed, 118 insertions(+), 205 deletions(-) diff --git a/core/connection.go b/core/connection.go index f4ec8ff..0853d7e 100644 --- a/core/connection.go +++ b/core/connection.go @@ -130,17 +130,3 @@ func (c *Connection) nolockIsFull() bool { } func (c *Connection) resetForNextExecution() {} - -//func (c *Connection) GetType() string { -// return "Connection" -//} - -func (c *Connection) GetArrayItem(i int) *Connection { - return nil -} - -func (c *Connection) SetArrayItem(c2 *Connection, i int) {} - -func (c *Connection) ArrayLength() int { - return 0 -} diff --git a/core/inarrayport.go b/core/inarrayport.go index 698f89d..cda0804 100644 --- a/core/inarrayport.go +++ b/core/inarrayport.go @@ -5,7 +5,7 @@ type InArrayPort struct { portName string fullName string - array []*Connection + array []InputConn } func (c *InArrayPort) isDrained() bool { @@ -26,10 +26,6 @@ func (c *InArrayPort) IsEmpty() bool { return true } -func (c *InArrayPort) receive(p *Process) *Packet { - panic("receive from an array port") -} - func (c *InArrayPort) IsClosed() bool { for _, v := range c.array { if !v.IsClosed() { @@ -41,21 +37,17 @@ func (c *InArrayPort) IsClosed() bool { func (c *InArrayPort) resetForNextExecution() {} -//func (c *InArrayPort) GetType() string { -// return "InArrayPort" -//} - -func (c *InArrayPort) GetArrayItem(i int) *Connection { +func (c *InArrayPort) GetArrayItem(i int) InputConn { if i >= len(c.array) { return nil } return c.array[i] } -func (c *InArrayPort) SetArrayItem(c2 *Connection, i int) { +func (c *InArrayPort) SetArrayItem(c2 InputConn, i int) { if i >= len(c.array) { // add to .array to fit c2 - increaseBy := make([]*Connection, i-len(c.array)+1) + increaseBy := make([]InputConn, i-len(c.array)+1) c.array = append(c.array, increaseBy...) } c.array[i] = c2 diff --git a/core/initializationconnection.go b/core/initializationconnection.go index 85f7b52..97bb538 100644 --- a/core/initializationconnection.go +++ b/core/initializationconnection.go @@ -64,17 +64,3 @@ func (c *InitializationConnection) resetForNextExecution() { c.closed = false } - -//func (c *InitializationConnection) GetType() string { -// return "InitializationConnection" -//} - -func (c *InitializationConnection) GetArrayItem(i int) *Connection { - return nil -} - -func (c *InitializationConnection) SetArrayItem(c2 *Connection, i int) {} - -func (c *InitializationConnection) ArrayLength() int { - return 0 -} diff --git a/core/inputconn.go b/core/inputconn.go index a106ed2..dbd8de3 100644 --- a/core/inputconn.go +++ b/core/inputconn.go @@ -1,18 +1,22 @@ package core -type InputConn interface { - receive(p *Process) *Packet +type inputCommon interface { isDrained() bool resetForNextExecution() IsEmpty() bool IsClosed() bool - //GetType() string +} + +type InputConn interface { + inputCommon + receive(p *Process) *Packet } type InputArrayConn interface { - InputConn - GetArrayItem(i int) *Connection - SetArrayItem(c *Connection, i int) + inputCommon + + GetArrayItem(i int) InputConn + SetArrayItem(c InputConn, i int) ArrayLength() int } diff --git a/core/network.go b/core/network.go index 5a7d492..9b6dcd9 100644 --- a/core/network.go +++ b/core/network.go @@ -49,8 +49,8 @@ func (n *Network) NewProc(nm string, comp Component) *Process { //n.procList = append(n.procList, proc) n.procs[nm] = proc - proc.inPorts = make(map[string]interface{}) - proc.outPorts = make(map[string]interface{}) + proc.inPorts = make(map[string]inputCommon) + proc.outPorts = make(map[string]outputCommon) return proc } @@ -90,73 +90,88 @@ func (n *Network) NewOutArrayPort() *OutArrayPort { } func (n *Network) Connect(p1 *Process, out string, p2 *Process, in string, cap int) { - inPort := parsePort(in) - var connxn *Connection - //var anyInConn InputConn + var conn InputConn if inPort.indexed { - var anyInConn = p2.inPorts[inPort.name] - if anyInConn == nil { - - anyInConn = n.NewInArrayPort() - p2.inPorts[inPort.name] = anyInConn + // try get an existing port + existingConn := p2.inPorts[inPort.name] + if existingConn == nil { + // create one, if it doesn't exist + existingConn = n.NewInArrayPort() + p2.inPorts[inPort.name] = existingConn } - - //anyInConn = anyInConn.(InputArrayConn) - connxn = anyInConn.(InputArrayConn).GetArrayItem(inPort.index) - - if connxn == nil { - connxn = n.NewConnection(cap) - connxn.portName = inPort.name - connxn.fullName = p2.name + "." + inPort.name - connxn.downStrProc = p2 - connxn.network = n - if anyInConn == nil { - p2.inPorts[inPort.name] = connxn - } else { - anyInConn.(InputArrayConn).SetArrayItem(connxn, inPort.index) - } + // enforce it's an array + arrayConn := existingConn.(InputArrayConn) + + // try get an existing connection from array + conn = arrayConn.GetArrayItem(inPort.index) + if conn == nil { + // create one if it doesn't exist + newConn := n.NewConnection(cap) + newConn.portName = inPort.name + newConn.fullName = p2.name + "." + inPort.name + newConn.downStrProc = p2 + newConn.network = n + conn = newConn + arrayConn.SetArrayItem(conn, inPort.index) } } else { - if p2.inPorts[inPort.name] == nil { - connxn = n.NewConnection(cap) - connxn.portName = inPort.name - connxn.fullName = p2.name + "." + inPort.name - connxn.downStrProc = p2 - connxn.network = n - p2.inPorts[inPort.name] = connxn - } else { - connxn = p2.inPorts[inPort.name].(*Connection) + // try get an existing port + existingConn, ok := p2.inPorts[inPort.name] + if !ok { + // create one if it doesn't exist + newConn := n.NewConnection(cap) + newConn.portName = inPort.name + newConn.fullName = p2.name + "." + inPort.name + newConn.downStrProc = p2 + newConn.network = n + existingConn = newConn + p2.inPorts[inPort.name] = newConn } + conn = existingConn.(InputConn) } // connxn built; input port array built if necessary - //var anyOutConn OutputConn + // rest of the code requires that the input is a `*Connection` + // this probably could be unrestricted. + connxn := conn.(*Connection) outPort := parsePort(out) - if outPort.indexed { - var anyOutConn = p1.outPorts[outPort.name] - if anyOutConn == nil { + // try get an existing port + anyOutConn, ok := p1.outPorts[outPort.name] + if !ok { + // create one if it doesn't exist anyOutConn = n.NewOutArrayPort() p1.outPorts[outPort.name] = anyOutConn } + // enforce the output is an array + arrayConn := anyOutConn.(OutputArrayConn) - opt := new(OutPort) - //p1.outPorts[out] = anyOutConn - opt.name = out - anyOutConn.(OutputArrayConn).SetArrayItem(opt, outPort.index) - opt.Conn = connxn + // check that nothing has connected to that item + if existing := arrayConn.GetArrayItem(outPort.index); existing != nil { + panic("output port " + out + " already connected") + } + + // update the array + arrayConn.SetArrayItem(&OutPort{ + name: out, + conn: connxn, + }, outPort.index) } else { - //var opt OutputConn - opt := new(OutPort) - p1.outPorts[out] = opt - opt.name = out - opt.Conn = connxn + // check that nothing has connected already + if _, exists := p1.outPorts[out]; exists { + panic("output port " + out + " already connected") + } + // add the connection + p1.outPorts[out] = &OutPort{ + name: out, + conn: connxn, + } } connxn.incUpstream() @@ -186,7 +201,6 @@ func parsePort(in string) portDefinition { } func (n *Network) Initialize(initValue string, p2 *Process, in string) { - conn := n.NewInitializationConnection() p2.inPorts[in] = conn conn.portName = in diff --git a/core/nulloutport.go b/core/nulloutport.go index 70fb944..6f3b046 100644 --- a/core/nulloutport.go +++ b/core/nulloutport.go @@ -2,26 +2,10 @@ package core type NullOutPort struct { name string - //Conn *Connection - //optional bool } func (c *NullOutPort) SetOptional(b bool) {} func (c *NullOutPort) send(*Process, *Packet) bool { panic("send on null port") } -//func (c *NullOutPort) GetType() string { -// return "NullOutPort" -//} - -func (c *NullOutPort) GetArrayItem(i int) *OutPort { - return nil -} - -func (c *NullOutPort) SetArrayItem(o *OutPort, i int) {} - -func (c *NullOutPort) ArrayLength() int { - return 0 -} - func (c *NullOutPort) Close() {} diff --git a/core/outarrayport.go b/core/outarrayport.go index a543200..34671c0 100644 --- a/core/outarrayport.go +++ b/core/outarrayport.go @@ -1,35 +1,27 @@ package core -//var _ Conn = (*InArrayPort)(nil) - type OutArrayPort struct { network *Network portName string fullName string - array []*OutPort + array []OutputConn closed bool } -func (o *OutArrayPort) send(p *Process, pkt *Packet) bool { panic("send on array port") } - func (o *OutArrayPort) SetOptional(b bool) {} -//func (o *OutArrayPort) GetType() string { -// return "OutArrayPort" -//} - -func (o *OutArrayPort) GetArrayItem(i int) *OutPort { +func (o *OutArrayPort) GetArrayItem(i int) OutputConn { if i >= len(o.array) { return nil } return o.array[i] } -func (o *OutArrayPort) SetArrayItem(o2 *OutPort, i int) { +func (o *OutArrayPort) SetArrayItem(o2 OutputConn, i int) { if i >= len(o.array) { // add to .array to fit c2 - increaseBy := make([]*OutPort, i-len(o.array)+1) + increaseBy := make([]OutputConn, i-len(o.array)+1) o.array = append(o.array, increaseBy...) } o.array[i] = o2 diff --git a/core/outport.go b/core/outport.go index 9d1cfd4..e932669 100644 --- a/core/outport.go +++ b/core/outport.go @@ -2,28 +2,18 @@ package core type OutPort struct { name string - Conn *Connection + conn *Connection optional bool } func (o *OutPort) send(p *Process, pkt *Packet) bool { - return o.Conn.send(p, pkt) + return o.conn.send(p, pkt) } func (o *OutPort) SetOptional(b bool) { o.optional = b } -func (o *OutPort) GetArrayItem(i int) *OutPort { - return nil -} - -func (o *OutPort) SetArrayItem(op *OutPort, i int) {} - -func (o *OutPort) ArrayLength() int { - return 0 -} - func (o *OutPort) Close() { - o.Conn.decUpstream() + o.conn.decUpstream() } diff --git a/core/outputconn.go b/core/outputconn.go index 1e21d98..c4415dd 100644 --- a/core/outputconn.go +++ b/core/outputconn.go @@ -1,18 +1,18 @@ package core +type outputCommon interface { + Close() +} + type OutputConn interface { + outputCommon send(*Process, *Packet) bool - - //IsEmpty() bool - //IsClosed() bool SetOptional(b bool) - //GetType() string - Close() } type OutputArrayConn interface { - OutputConn - GetArrayItem(i int) *OutPort - SetArrayItem(c *OutPort, i int) + outputCommon + GetArrayItem(i int) OutputConn + SetArrayItem(c OutputConn, i int) ArrayLength() int } diff --git a/core/process.go b/core/process.go index 1f2f398..4176ae0 100644 --- a/core/process.go +++ b/core/process.go @@ -6,17 +6,11 @@ import ( "sync/atomic" ) -//import ( -// "github.com/gofbp/core" -//) - type Process struct { - name string - network *Network - //inPorts map[string]InputConn - inPorts map[string]interface{} - //outPorts map[string]OutputConn - outPorts map[string]interface{} + name string + network *Network + inPorts map[string]inputCommon + outPorts map[string]outputCommon logFile string component Component ownedPkts int @@ -31,59 +25,43 @@ func (p *Process) GetName() string { return p.name } -func (p *Process) OpenInPort(s string) InputConn { - if len(p.inPorts) == 0 { - panic(p.name + ": No input ports specified") - } - in := p.inPorts[s] - if in == nil { - panic(p.name + ": Port name not found (" + s + ")") +func (p *Process) OpenInPort(name string) InputConn { + in, ok := p.inPorts[name] + if !ok { + panic(p.name + ": Port name not found (" + name + ")") } return in.(InputConn) } -func (p *Process) OpenInArrayPort(s string) InputArrayConn { - if len(p.inPorts) == 0 { - panic(p.name + ": No input ports specified") - } - in := p.inPorts[s] - if in == nil { - panic(p.name + ": Port name not found (" + s + ")") +func (p *Process) OpenInArrayPort(name string) InputArrayConn { + in, ok := p.inPorts[name] + if !ok { + panic(p.name + ": Port name not found (" + name + ")") } return in.(InputArrayConn) } -func (p *Process) OpenOutPort(s ...string) OutputConn { - if len(p.outPorts) == 0 { - opt := new(NullOutPort) - p.outPorts[s[0]] = opt - opt.name = s[0] - } - out := p.outPorts[s[0]] +func (p *Process) OpenOutPort(name string, opts ...string) OutputConn { + out, ok := p.outPorts[name] + if !ok { + if len(opts) == 0 || opts[0] != "opt" { + panic(p.name + ": Port name not found (" + name + ")") + } - if len(s) == 2 && s[1] != "opt" { - panic(p.name + ": Invalid 2nd param (" + s[1] + ")") + out = &NullOutPort{name: name} + p.outPorts[name] = out } - return out.(OutputConn) - } // not sure it maes sense to allow optional for array ports! -func (p *Process) OpenOutArrayPort(s ...string) OutputArrayConn { - if len(p.outPorts) == 0 { - opt := new(NullOutPort) - p.outPorts[s[0]] = opt - opt.name = s[0] - } - out := p.outPorts[s[0]] - - if len(s) == 2 && s[1] != "opt" { - panic(p.name + ": Invalid 2nd param (" + s[1] + ")") +func (p *Process) OpenOutArrayPort(name string) OutputArrayConn { + out, ok := p.outPorts[name] + if !ok { + panic(p.name + ": Port name not found (" + name + ")") } return out.(OutputArrayConn) - } // Send sends a packet to the output connection. @@ -99,7 +77,6 @@ func (p *Process) Receive(c InputConn) *Packet { } func (p *Process) ensureRunning() { - if !atomic.CompareAndSwapInt32(&p.status, Notstarted, Active) { return } @@ -172,25 +149,14 @@ func (p *Process) Run() { canRun = false } else { for _, v := range p.inPorts { - _, b := v.(InitializationConnection) - if b { - v.(*InitializationConnection).resetForNextExecution() - } + v.resetForNextExecution() } } } for _, v := range p.outPorts { - _, b := v.(OutputConn) - if b { - v.(OutputConn).Close() - } else { - _, b := v.(OutputArrayConn) - if b { - v.(OutputArrayConn).Close() - } - } + v.Close() } } diff --git a/testdata.copy b/testdata.copy index c7ff85e..a83ef5e 100644 --- a/testdata.copy +++ b/testdata.copy @@ -1,4 +1,3 @@ -This repo holds the beginning of an FBP implementation in Go Features include: From fdf8eaf80d3f97a1b38c9262f902965cbad9cdff Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Fri, 8 Oct 2021 20:37:30 +0300 Subject: [PATCH 02/13] core: simplify the process and network startup code When the interfaces are properly aligned, we don't need to do any casting. Similarly, by using `isDrained` simplifies the logic for checking whether the process should be still running. isDrained returns true when the connection is closed and there isn't anything more to read. Adding a method `isSelfStarting` it simplifies the startup code. --- core/connection.go | 2 - core/network.go | 25 +++------- core/process.go | 117 +++++++++++++++++++-------------------------- 3 files changed, 55 insertions(+), 89 deletions(-) diff --git a/core/connection.go b/core/connection.go index 0853d7e..4f7c2b4 100644 --- a/core/connection.go +++ b/core/connection.go @@ -6,8 +6,6 @@ import ( "sync/atomic" ) -// Based on https://stackoverflow.com/questions/36857167/how-to-correctly-use-sync-cond - type Connection struct { network *Network pktArray []*Packet diff --git a/core/network.go b/core/network.go index 9b6dcd9..c34f85d 100644 --- a/core/network.go +++ b/core/network.go @@ -258,28 +258,15 @@ func (n *Network) Run() { } }() - var canRun bool = false + startedAnything := false for _, proc := range n.procs { - proc.mtx.Lock() - defer proc.mtx.Unlock() - proc.selfStarting = true - if proc.inPorts != nil { - for _, conn := range proc.inPorts { - //if conn.GetType() != "InitializationConnection" { - _, b := conn.(*InitializationConnection) - if !b { - proc.selfStarting = false - } - } - } - if !proc.selfStarting { - continue + if proc.isSelfStarting() { + proc.ensureRunning() + startedAnything = true } - - proc.ensureRunning() - canRun = true } - if !canRun { + + if !startedAnything { n.wg.Add(0 - len(n.procs)) panic("No process can start") } diff --git a/core/process.go b/core/process.go index 4176ae0..2d5a381 100644 --- a/core/process.go +++ b/core/process.go @@ -2,7 +2,6 @@ package core import ( "fmt" - "sync" "sync/atomic" ) @@ -14,11 +13,7 @@ type Process struct { logFile string component Component ownedPkts int - //done bool - selfStarting bool // process has no non-IIP input ports - //MustRun bool - status int32 - mtx sync.Mutex + status int32 } func (p *Process) GetName() string { @@ -81,57 +76,29 @@ func (p *Process) ensureRunning() { return } - //p.network.wg.Add(1) - go func() { // Process goroutine + go func() { defer p.network.wg.Done() - p.Run() + p.run() }() } -func (p *Process) inputState() (bool, bool) { - allDrained := true - hasData := false - for _, v := range p.inPorts { - //if v.GetType() == "InArrayPort" { - _, b := v.(*InArrayPort) - if b { - //allClosed = true - for _, w := range v.(*InArrayPort).array { - if !w.isDrained() /* || !w.IsClosed() */ { - allDrained = false - } - hasData = hasData || !w.IsEmpty() - } - } else { - _, b := v.(*Connection) - if b { - if !v.(*Connection).isDrained() /*|| !v.IsClosed() */ { - allDrained = false - } - hasData = hasData || !v.(*Connection).IsEmpty() - } - } - } - return allDrained, hasData -} - -func (p *Process) Run() { - p.mtx.Lock() - defer p.mtx.Unlock() +func (p *Process) run() { atomic.StoreInt32(&p.status, Dormant) defer atomic.StoreInt32(&p.status, Terminated) - defer fmt.Println(p.GetName(), " terminated") + fmt.Println(p.GetName(), " started") - p.component.Setup(p) + defer fmt.Println(p.GetName(), " terminated") - //var allDrained bool - //var hasData bool + p.component.Setup(p) - allDrained, hasData := p.inputState() + runOnce := p.isSelfStarting() + for runOnce || !p.allInputsClosed() { + runOnce = false - canRun := p.selfStarting || hasData || !allDrained || p.isMustRun() + for _, v := range p.inPorts { + v.resetForNextExecution() + } - for canRun { // multiple activations, if necessary! fmt.Println(p.GetName(), " activated") atomic.StoreInt32(&p.status, Active) @@ -143,16 +110,6 @@ func (p *Process) Run() { panic(p.name + " deactivated without disposing of all owned packets") } - allDrained, _ := p.inputState() - - if allDrained { - canRun = false - } else { - for _, v := range p.inPorts { - v.resetForNextExecution() - } - } - } for _, v := range p.outPorts { @@ -160,21 +117,38 @@ func (p *Process) Run() { } } -func (p *Process) isMustRun() bool { - _, hasMustRun := p.component.(ComponentWithMustRun) - return hasMustRun +// isSelfStarting returns whether the process should start at the beginning of the network. +func (p *Process) isSelfStarting() bool { + // start anything that has a MustRun annotation + if isMustRun(p.component) { + return true + } + + // start anything that doesn't have any input ports + if len(p.inPorts) == 0 { + return true + } + + // start anything that has an initialization connection + for _, in := range p.inPorts { + if _, ok := in.(*InitializationConnection); ok { + return true + } + } + + return false } -/* -// create packet containing string -func (p *Process) Create(s string) *Packet { - var pkt *Packet = new(Packet) - pkt.Contents = s - pkt.owner = p - p.ownedPkts++ - return pkt +// allInputsClosed returns whether there are any inbound connections +// that might return data. +func (p *Process) allInputsClosed() bool { + for _, v := range p.inPorts { + if !v.isDrained() { + return false + } + } + return true } -*/ // create packet containing anything! func (p *Process) Create(x interface{}) *Packet { @@ -195,6 +169,13 @@ func (p *Process) CreateBracket(pktType int32, s string) *Packet { return pkt } +// Discard safely deletes the packet. func (p *Process) Discard(pkt *Packet) { p.ownedPkts-- } + +// isMustRun checks whether component has MustRun annotation. +func isMustRun(comp Component) bool { + _, hasMustRun := comp.(ComponentWithMustRun) + return hasMustRun +} From 0a57bd7fbe0a57a93931c74572785012dee05b27 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Fri, 8 Oct 2021 20:42:17 +0300 Subject: [PATCH 03/13] core: move deadlock detection to separate func --- core/network.go | 75 ++++++++++++++++++++++++------------------------- 1 file changed, 37 insertions(+), 38 deletions(-) diff --git a/core/network.go b/core/network.go index c34f85d..002949f 100644 --- a/core/network.go +++ b/core/network.go @@ -6,7 +6,6 @@ import ( "strconv" "sync" "sync/atomic" - "time" ) const ( @@ -218,45 +217,9 @@ func (n *Network) Run() { // reactivated many times during the process "run" n.wg.Add(len(n.procs)) - defer n.wg.Wait() - go func() { - for { - time.Sleep(200 * time.Millisecond) - allTerminated := true - deadlockDetected := true - for _, proc := range n.procs { - //proc.mtx.Lock() - //defer proc.mtx.Unlock() - status := atomic.LoadInt32(&proc.status) - if status != Terminated { - allTerminated = false - if status == Active { - deadlockDetected = false - } - } - } - if allTerminated { - // fmt.Println("Run terminated") - return - } - if deadlockDetected { - fmt.Println("\nDeadlock detected!") - for key, proc := range n.procs { - fmt.Println(key, " Status: ", - []string{"notStarted", - "active", - "dormant", - "suspSend", - "suspRecv", - "terminated"}[proc.status]) - } - panic("Deadlock!") - } - - } - }() + go n.deadlockDetection() startedAnything := false for _, proc := range n.procs { @@ -271,3 +234,39 @@ func (n *Network) Run() { panic("No process can start") } } + +func (n *Network) deadlockDetection() { + for { + allTerminated := true + deadlockDetected := true + for _, proc := range n.procs { + //proc.mtx.Lock() + //defer proc.mtx.Unlock() + status := atomic.LoadInt32(&proc.status) + if status != Terminated { + allTerminated = false + if status == Active { + deadlockDetected = false + } + } + } + if allTerminated { + // fmt.Println("Run terminated") + return + } + if deadlockDetected { + fmt.Println("\nDeadlock detected!") + for key, proc := range n.procs { + fmt.Println(key, " Status: ", + []string{"notStarted", + "active", + "dormant", + "suspSend", + "suspRecv", + "terminated"}[proc.status]) + } + panic("Deadlock!") + } + + } +} From 10b634bc3c5ce0a46e4d233f6821c7ba1714085a Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Fri, 8 Oct 2021 20:54:53 +0300 Subject: [PATCH 04/13] core: move Process status changes to a separate method We need to track all transitions at the network level to detect deadlocks. --- core/connection.go | 10 ++++------ core/network.go | 27 ++++----------------------- core/process.go | 25 +++++++++++++++++++------ core/process_status.go | 33 +++++++++++++++++++++++++++++++++ 4 files changed, 60 insertions(+), 35 deletions(-) create mode 100644 core/process_status.go diff --git a/core/connection.go b/core/connection.go index 4f7c2b4..0589e2f 100644 --- a/core/connection.go +++ b/core/connection.go @@ -3,7 +3,6 @@ package core import ( "fmt" "sync" - "sync/atomic" ) type Connection struct { @@ -31,9 +30,9 @@ func (c *Connection) send(p *Process, pkt *Packet) bool { c.downStrProc.ensureRunning() c.condNE.Broadcast() for c.nolockIsFull() { // connection is full - atomic.StoreInt32(&p.status, SuspSend) + p.transition(SuspendedSend) c.condNF.Wait() - atomic.StoreInt32(&p.status, Active) + p.transition(Active) } fmt.Println(p.name, "Sent", pkt.Contents) c.pktArray[c.is] = pkt @@ -53,10 +52,9 @@ func (c *Connection) receive(p *Process) *Packet { c.condNF.Broadcast() return nil } - atomic.StoreInt32(&p.status, SuspRecv) + p.transition(SuspendedRecv) c.condNE.Wait() - atomic.StoreInt32(&p.status, Active) - + p.transition(Active) } pkt := c.pktArray[c.ir] c.pktArray[c.ir] = nil diff --git a/core/network.go b/core/network.go index 002949f..1709965 100644 --- a/core/network.go +++ b/core/network.go @@ -5,16 +5,7 @@ import ( "regexp" "strconv" "sync" - "sync/atomic" -) - -const ( - Notstarted int32 = iota - Active - Dormant - SuspSend - SuspRecv - Terminated + "time" ) type Network struct { @@ -36,13 +27,11 @@ func NewNetwork(name string) *Network { } func (n *Network) NewProc(nm string, comp Component) *Process { - proc := &Process{ name: nm, network: n, logFile: "", component: comp, - status: Notstarted, } //n.procList = append(n.procList, proc) @@ -237,12 +226,11 @@ func (n *Network) Run() { func (n *Network) deadlockDetection() { for { + time.Sleep(200 * time.Millisecond) allTerminated := true deadlockDetected := true for _, proc := range n.procs { - //proc.mtx.Lock() - //defer proc.mtx.Unlock() - status := atomic.LoadInt32(&proc.status) + status := proc.status() if status != Terminated { allTerminated = false if status == Active { @@ -251,19 +239,12 @@ func (n *Network) deadlockDetection() { } } if allTerminated { - // fmt.Println("Run terminated") return } if deadlockDetected { fmt.Println("\nDeadlock detected!") for key, proc := range n.procs { - fmt.Println(key, " Status: ", - []string{"notStarted", - "active", - "dormant", - "suspSend", - "suspRecv", - "terminated"}[proc.status]) + fmt.Println(key, " Status: ", proc.status()) } panic("Deadlock!") } diff --git a/core/process.go b/core/process.go index 2d5a381..58ced07 100644 --- a/core/process.go +++ b/core/process.go @@ -13,7 +13,8 @@ type Process struct { logFile string component Component ownedPkts int - status int32 + + atomicStatus int32 } func (p *Process) GetName() string { @@ -72,7 +73,7 @@ func (p *Process) Receive(c InputConn) *Packet { } func (p *Process) ensureRunning() { - if !atomic.CompareAndSwapInt32(&p.status, Notstarted, Active) { + if !p.compareSwapTransition(NotStarted, Active) { return } @@ -83,8 +84,8 @@ func (p *Process) ensureRunning() { } func (p *Process) run() { - atomic.StoreInt32(&p.status, Dormant) - defer atomic.StoreInt32(&p.status, Terminated) + p.transition(Dormant) + defer p.transition(Terminated) fmt.Println(p.GetName(), " started") defer fmt.Println(p.GetName(), " terminated") @@ -101,9 +102,9 @@ func (p *Process) run() { // multiple activations, if necessary! fmt.Println(p.GetName(), " activated") - atomic.StoreInt32(&p.status, Active) + p.transition(Active) p.component.Execute(p) // single "activation" - atomic.StoreInt32(&p.status, Dormant) + p.transition(Dormant) fmt.Println(p.GetName(), " deactivated") if p.ownedPkts > 0 { @@ -179,3 +180,15 @@ func isMustRun(comp Component) bool { _, hasMustRun := comp.(ComponentWithMustRun) return hasMustRun } + +func (p *Process) status() ProcessStatus { + return ProcessStatus(atomic.LoadInt32(&p.atomicStatus)) +} + +func (p *Process) transition(targetStatus ProcessStatus) { + atomic.StoreInt32(&p.atomicStatus, int32(targetStatus)) +} + +func (p *Process) compareSwapTransition(existingStatus, targetStatus ProcessStatus) bool { + return atomic.CompareAndSwapInt32(&p.atomicStatus, int32(existingStatus), int32(targetStatus)) +} diff --git a/core/process_status.go b/core/process_status.go new file mode 100644 index 0000000..9b972b6 --- /dev/null +++ b/core/process_status.go @@ -0,0 +1,33 @@ +package core + +import "fmt" + +type ProcessStatus int32 + +const ( + NotStarted ProcessStatus = iota + Active + Dormant + SuspendedSend + SuspendedRecv + Terminated +) + +func (status ProcessStatus) String() string { + switch status { + case NotStarted: + return "NotStarted" + case Active: + return "Active" + case Dormant: + return "Dormant" + case SuspendedSend: + return "SuspendedSend" + case SuspendedRecv: + return "SuspendedRecv" + case Terminated: + return "Terminated" + default: + return fmt.Sprintf("ProcessStatus(%d)", status) + } +} From ea31fa7345ace201ded8007317d9e0410c79208c Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Fri, 8 Oct 2021 21:34:15 +0300 Subject: [PATCH 05/13] core: consolidate process transition printing --- core/network.go | 1 - core/process.go | 25 +++++++++++-------------- 2 files changed, 11 insertions(+), 15 deletions(-) diff --git a/core/network.go b/core/network.go index 1709965..c79f7d4 100644 --- a/core/network.go +++ b/core/network.go @@ -195,7 +195,6 @@ func (n *Network) Initialize(initValue string, p2 *Process, in string) { conn.fullName = p2.name + "." + in conn.value = initValue - } func (n *Network) Run() { diff --git a/core/process.go b/core/process.go index 58ced07..b63d19e 100644 --- a/core/process.go +++ b/core/process.go @@ -73,23 +73,18 @@ func (p *Process) Receive(c InputConn) *Packet { } func (p *Process) ensureRunning() { - if !p.compareSwapTransition(NotStarted, Active) { + if !p.transitionFrom(NotStarted, Dormant) { return } go func() { defer p.network.wg.Done() + defer p.transition(Terminated) p.run() }() } func (p *Process) run() { - p.transition(Dormant) - defer p.transition(Terminated) - - fmt.Println(p.GetName(), " started") - defer fmt.Println(p.GetName(), " terminated") - p.component.Setup(p) runOnce := p.isSelfStarting() @@ -101,16 +96,13 @@ func (p *Process) run() { } // multiple activations, if necessary! - fmt.Println(p.GetName(), " activated") p.transition(Active) p.component.Execute(p) // single "activation" p.transition(Dormant) - fmt.Println(p.GetName(), " deactivated") if p.ownedPkts > 0 { panic(p.name + " deactivated without disposing of all owned packets") } - } for _, v := range p.outPorts { @@ -185,10 +177,15 @@ func (p *Process) status() ProcessStatus { return ProcessStatus(atomic.LoadInt32(&p.atomicStatus)) } -func (p *Process) transition(targetStatus ProcessStatus) { - atomic.StoreInt32(&p.atomicStatus, int32(targetStatus)) +func (p *Process) transition(next ProcessStatus) { + previous := ProcessStatus(atomic.SwapInt32(&p.atomicStatus, int32(next))) + fmt.Printf("%s %s -> %s\n", p.GetName(), previous, next) } -func (p *Process) compareSwapTransition(existingStatus, targetStatus ProcessStatus) bool { - return atomic.CompareAndSwapInt32(&p.atomicStatus, int32(existingStatus), int32(targetStatus)) +func (p *Process) transitionFrom(current, next ProcessStatus) bool { + ok := atomic.CompareAndSwapInt32(&p.atomicStatus, int32(current), int32(next)) + if ok { + fmt.Printf("%s %s -> %s\n", p.GetName(), current, next) + } + return ok } From 80fc189767081c0e9aea3c74f9c8462bf1f0f8a8 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Fri, 8 Oct 2021 21:36:05 +0300 Subject: [PATCH 06/13] core: partial implementation of deadlock detection Currently does not handle connections properly when they are pending a handoff of work items. --- core/network.go | 32 +-------------- core/network_status.go | 90 ++++++++++++++++++++++++++++++++++++++++++ core/process.go | 2 + 3 files changed, 93 insertions(+), 31 deletions(-) create mode 100644 core/network_status.go diff --git a/core/network.go b/core/network.go index c79f7d4..70361bd 100644 --- a/core/network.go +++ b/core/network.go @@ -5,7 +5,6 @@ import ( "regexp" "strconv" "sync" - "time" ) type Network struct { @@ -13,6 +12,7 @@ type Network struct { procs map[string]*Process //procList []*Process //driver Process + status networkStatus logFile string wg sync.WaitGroup } @@ -207,8 +207,6 @@ func (n *Network) Run() { n.wg.Add(len(n.procs)) defer n.wg.Wait() - go n.deadlockDetection() - startedAnything := false for _, proc := range n.procs { if proc.isSelfStarting() { @@ -222,31 +220,3 @@ func (n *Network) Run() { panic("No process can start") } } - -func (n *Network) deadlockDetection() { - for { - time.Sleep(200 * time.Millisecond) - allTerminated := true - deadlockDetected := true - for _, proc := range n.procs { - status := proc.status() - if status != Terminated { - allTerminated = false - if status == Active { - deadlockDetected = false - } - } - } - if allTerminated { - return - } - if deadlockDetected { - fmt.Println("\nDeadlock detected!") - for key, proc := range n.procs { - fmt.Println(key, " Status: ", proc.status()) - } - panic("Deadlock!") - } - - } -} diff --git a/core/network_status.go b/core/network_status.go new file mode 100644 index 0000000..c1a5397 --- /dev/null +++ b/core/network_status.go @@ -0,0 +1,90 @@ +package core + +import ( + "fmt" + "sync" +) + +const deadlockDetectionEnabled = false + +type networkStatus struct { + mu sync.Mutex + deadlocked bool + running int32 + suspended int32 +} + +func (network *Network) processTransitioned(from, to ProcessStatus) { + if from == to { + return + } + + type tx [2]ProcessStatus + + switch (tx{from, to}) { + case tx{NotStarted, Dormant}: + network.status.mu.Lock() + network.status.running++ + network.status.mu.Unlock() + + // these have no impact on deadlock detection + case tx{Dormant, Active}: + case tx{Active, Dormant}: + + case tx{Active, SuspendedSend}, tx{Active, SuspendedRecv}: + network.status.mu.Lock() + network.status.suspended++ + network.status.running-- + leftRunning := network.status.running + network.status.mu.Unlock() + + if leftRunning <= 0 { + network.deadlockDetected() + } + + case tx{SuspendedSend, Active}, tx{SuspendedRecv, Active}: + network.status.mu.Lock() + network.status.suspended-- + network.status.running++ + network.status.mu.Unlock() + + case tx{Dormant, Terminated}, tx{Active, Terminated}: + network.status.mu.Lock() + network.status.running-- + running, suspended := network.status.running, network.status.suspended + network.status.mu.Unlock() + + if running == 0 && suspended > 0 { + network.deadlockDetected() + } + + default: + network.status.mu.Lock() + defer network.status.mu.Unlock() + + if !network.status.deadlocked { + panic(fmt.Sprintf("unhandled transition %s -> %s", from, to)) + } + } +} + +func (network *Network) deadlockDetected() { + if !deadlockDetectionEnabled { + return + } + + network.status.mu.Lock() + defer network.status.mu.Unlock() + + if network.status.deadlocked { + // avoid printing status twice + return + } + network.status.deadlocked = true + + fmt.Println("\nDeadlock detected!") + for _, proc := range network.procs { + fmt.Printf("\t%-15s\t%s\n", proc.name, proc.status()) + } + panic("Deadlock!") +} diff --git a/core/process.go b/core/process.go index b63d19e..9288d50 100644 --- a/core/process.go +++ b/core/process.go @@ -180,12 +180,14 @@ func (p *Process) status() ProcessStatus { func (p *Process) transition(next ProcessStatus) { previous := ProcessStatus(atomic.SwapInt32(&p.atomicStatus, int32(next))) fmt.Printf("%s %s -> %s\n", p.GetName(), previous, next) + p.network.processTransitioned(previous, next) } func (p *Process) transitionFrom(current, next ProcessStatus) bool { ok := atomic.CompareAndSwapInt32(&p.atomicStatus, int32(current), int32(next)) if ok { fmt.Printf("%s %s -> %s\n", p.GetName(), current, next) + p.network.processTransitioned(current, next) } return ok } From 718c3b1829c7d130b90628f4cd0db2202ce188e2 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Fri, 8 Oct 2021 22:53:55 +0300 Subject: [PATCH 07/13] core: implement Optional out ports more cleanly. By default sending to an optional port will discard the packet. To check whether port is connected to something, it's possible to use if opt.IsConnected() { --- components/io/writefile.go | 10 ++-------- components/testrtn/counter.go | 12 ++---------- components/testrtn/discard.go | 3 +-- components/testrtn/writetoconsnl.go | 11 ++--------- components/testrtn/writetoconsole.go | 10 ++-------- core/nulloutport.go | 12 +++++++++--- core/outarrayport.go | 2 -- core/outport.go | 9 +++------ core/outputconn.go | 3 ++- core/process.go | 12 ++++++++---- 10 files changed, 31 insertions(+), 53 deletions(-) diff --git a/components/io/writefile.go b/components/io/writefile.go index 70201a7..4dc47a8 100644 --- a/components/io/writefile.go +++ b/components/io/writefile.go @@ -16,7 +16,7 @@ type WriteFile struct { func (writeFile *WriteFile) Setup(p *core.Process) { writeFile.iptIp = p.OpenInPort("FILENAME") writeFile.ipt = p.OpenInPort("IN") - writeFile.opt = p.OpenOutPort("OUT", "opt") + writeFile.opt = p.OpenOutPortOptional("OUT") } func (WriteFile) MustRun() {} @@ -47,13 +47,7 @@ func (writeFile *WriteFile) Execute(p *core.Process) { panic("Unable to write file: " + fname) } - _, b := writeFile.opt.(*core.OutPort) - if b { - //if writeFile.opt.GetType() == "OutPort" { - p.Send(writeFile.opt, pkt) - } else { - p.Discard(pkt) - } + p.Send(writeFile.opt, pkt) } fmt.Println(p.GetName()+": File", fname, "written") diff --git a/components/testrtn/counter.go b/components/testrtn/counter.go index 67eb338..97b30db 100644 --- a/components/testrtn/counter.go +++ b/components/testrtn/counter.go @@ -16,7 +16,7 @@ type Counter struct { func (counter *Counter) Setup(p *core.Process) { counter.ipt = p.OpenInPort("IN") counter.cnt = p.OpenOutPort("COUNT") - counter.opt = p.OpenOutPort("OUT", "opt") + counter.opt = p.OpenOutPortOptional("OUT") } func (Counter) MustRun() {} @@ -31,15 +31,7 @@ func (counter *Counter) Execute(p *core.Process) { if pkt == nil { break } - //fmt.Println(pkt.Contents) - //if counter.opt.GetType() == "OutPort" { - _, b := counter.opt.(*core.OutPort) - if b { - p.Send(counter.opt, pkt) - } else { - p.Discard(pkt) - } - + p.Send(counter.opt, pkt) count++ } diff --git a/components/testrtn/discard.go b/components/testrtn/discard.go index 726afee..9be7cc5 100644 --- a/components/testrtn/discard.go +++ b/components/testrtn/discard.go @@ -13,7 +13,7 @@ type Discard struct { func (discard *Discard) Setup(p *core.Process) { discard.ipt = p.OpenInPort("IN") - discard.opt = p.OpenOutPort("OUT", "opt") + discard.opt = p.OpenOutPortOptional("OUT") } //func (Discard) MustRun() {} @@ -29,7 +29,6 @@ func (discard *Discard) Execute(p *core.Process) { //fmt.Println(pkt.Contents) p.Discard(pkt) - } //fmt.Println(p.GetName() + " ended") diff --git a/components/testrtn/writetoconsnl.go b/components/testrtn/writetoconsnl.go index b45f464..81833f8 100644 --- a/components/testrtn/writetoconsnl.go +++ b/components/testrtn/writetoconsnl.go @@ -15,7 +15,7 @@ type WriteToConsNL struct { func (writeToConsole *WriteToConsNL) Setup(p *core.Process) { writeToConsole.ipt = p.OpenInPort("IN") - writeToConsole.opt = p.OpenOutPort("OUT", "opt") + writeToConsole.opt = p.OpenOutPortOptional("OUT") } //func (WriteToConsNL) MustRun() {} @@ -30,14 +30,7 @@ func (writeToConsole *WriteToConsNL) Execute(p *core.Process) { return } fmt.Println(pkt.Contents) - //if writeToConsole.opt.GetType() == "OutPort" { - _, b := writeToConsole.opt.(*core.OutPort) - if b { - p.Send(writeToConsole.opt, pkt) - } else { - p.Discard(pkt) - } - //} + p.Send(writeToConsole.opt, pkt) //fmt.Println(p.GetName() + " deactivated") } diff --git a/components/testrtn/writetoconsole.go b/components/testrtn/writetoconsole.go index 76894ed..59a797d 100644 --- a/components/testrtn/writetoconsole.go +++ b/components/testrtn/writetoconsole.go @@ -13,7 +13,7 @@ type WriteToConsole struct { func (writeToConsole *WriteToConsole) Setup(p *core.Process) { writeToConsole.ipt = p.OpenInPort("IN") - writeToConsole.opt = p.OpenOutPort("OUT", "opt") + writeToConsole.opt = p.OpenOutPortOptional("OUT") } func (WriteToConsole) MustRun() {} @@ -27,13 +27,7 @@ func (writeToConsole *WriteToConsole) Execute(p *core.Process) { break } fmt.Println(pkt.Contents) - //if writeToConsole.opt.GetType() == "OutPort" { - _, b := writeToConsole.opt.(*core.OutPort) - if b { - p.Send(writeToConsole.opt, pkt) - } else { - p.Discard(pkt) - } + p.Send(writeToConsole.opt, pkt) } //fmt.Println(p.GetName() + " ended") diff --git a/core/nulloutport.go b/core/nulloutport.go index 6f3b046..ca49966 100644 --- a/core/nulloutport.go +++ b/core/nulloutport.go @@ -4,8 +4,14 @@ type NullOutPort struct { name string } -func (c *NullOutPort) SetOptional(b bool) {} +// NullOutPort by default discards the packet. +func (*NullOutPort) send(p *Process, pkt *Packet) bool { + p.Discard(pkt) + return true +} -func (c *NullOutPort) send(*Process, *Packet) bool { panic("send on null port") } +func (*NullOutPort) IsConnected() bool { + return false +} -func (c *NullOutPort) Close() {} +func (*NullOutPort) Close() {} diff --git a/core/outarrayport.go b/core/outarrayport.go index 34671c0..1680a96 100644 --- a/core/outarrayport.go +++ b/core/outarrayport.go @@ -9,8 +9,6 @@ type OutArrayPort struct { closed bool } -func (o *OutArrayPort) SetOptional(b bool) {} - func (o *OutArrayPort) GetArrayItem(i int) OutputConn { if i >= len(o.array) { return nil diff --git a/core/outport.go b/core/outport.go index e932669..081d17d 100644 --- a/core/outport.go +++ b/core/outport.go @@ -1,18 +1,15 @@ package core type OutPort struct { - name string - conn *Connection - optional bool + name string + conn *Connection } func (o *OutPort) send(p *Process, pkt *Packet) bool { return o.conn.send(p, pkt) } -func (o *OutPort) SetOptional(b bool) { - o.optional = b -} +func (o *OutPort) IsConnected() bool { return true } func (o *OutPort) Close() { o.conn.decUpstream() diff --git a/core/outputconn.go b/core/outputconn.go index c4415dd..e3c58d7 100644 --- a/core/outputconn.go +++ b/core/outputconn.go @@ -7,7 +7,8 @@ type outputCommon interface { type OutputConn interface { outputCommon send(*Process, *Packet) bool - SetOptional(b bool) + + IsConnected() bool } type OutputArrayConn interface { diff --git a/core/process.go b/core/process.go index 9288d50..1ee9ca2 100644 --- a/core/process.go +++ b/core/process.go @@ -37,13 +37,17 @@ func (p *Process) OpenInArrayPort(name string) InputArrayConn { return in.(InputArrayConn) } -func (p *Process) OpenOutPort(name string, opts ...string) OutputConn { +func (p *Process) OpenOutPort(name string) OutputConn { out, ok := p.outPorts[name] if !ok { - if len(opts) == 0 || opts[0] != "opt" { - panic(p.name + ": Port name not found (" + name + ")") - } + panic(p.name + ": Port name not found (" + name + ")") + } + return out.(OutputConn) +} +func (p *Process) OpenOutPortOptional(name string) OutputConn { + out, ok := p.outPorts[name] + if !ok { out = &NullOutPort{name: name} p.outPorts[name] = out } From 9c9ee28e6fe827ce577ae474ce2050789935c9cd Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Sat, 9 Oct 2021 16:13:36 +0300 Subject: [PATCH 08/13] core: ensure broadcast happens after writing to array --- core/connection.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/connection.go b/core/connection.go index 0589e2f..6720d2d 100644 --- a/core/connection.go +++ b/core/connection.go @@ -28,7 +28,6 @@ func (c *Connection) send(p *Process, pkt *Packet) bool { defer c.condNF.L.Unlock() fmt.Println(p.name, "Sending", pkt.Contents) c.downStrProc.ensureRunning() - c.condNE.Broadcast() for c.nolockIsFull() { // connection is full p.transition(SuspendedSend) c.condNF.Wait() @@ -39,6 +38,7 @@ func (c *Connection) send(p *Process, pkt *Packet) bool { c.is = (c.is + 1) % len(c.pktArray) pkt.owner = nil p.ownedPkts-- + c.condNE.Broadcast() return true } From 83d9c8097f3227572a3df2980174ede0db48bfa3 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Sat, 9 Oct 2021 16:19:04 +0300 Subject: [PATCH 09/13] core: use c.mtx to Lock instead of condition By using the mtx for lock/unlock it makes it more clear that the send and recv path are using the same lock. Also, slightly better performance. --- core/connection.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/core/connection.go b/core/connection.go index 6720d2d..1797203 100644 --- a/core/connection.go +++ b/core/connection.go @@ -24,8 +24,8 @@ func (c *Connection) send(p *Process, pkt *Packet) bool { if pkt.owner != p { panic("Sending packet not owned by this process") } - c.condNF.L.Lock() - defer c.condNF.L.Unlock() + c.mtx.Lock() + defer c.mtx.Unlock() fmt.Println(p.name, "Sending", pkt.Contents) c.downStrProc.ensureRunning() for c.nolockIsFull() { // connection is full @@ -43,8 +43,8 @@ func (c *Connection) send(p *Process, pkt *Packet) bool { } func (c *Connection) receive(p *Process) *Packet { - c.condNE.L.Lock() - defer c.condNE.L.Unlock() + c.mtx.Lock() + defer c.mtx.Unlock() fmt.Println(p.name, "Receiving") for c.nolockIsEmpty() { // connection is empty From 514796a9cb05ed2309e2c08f6b6f4df816fc9a44 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Tue, 12 Oct 2021 10:13:00 +0300 Subject: [PATCH 10/13] core: fix spurious wakeups --- core/connection.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/connection.go b/core/connection.go index 1797203..edef1c1 100644 --- a/core/connection.go +++ b/core/connection.go @@ -38,7 +38,7 @@ func (c *Connection) send(p *Process, pkt *Packet) bool { c.is = (c.is + 1) % len(c.pktArray) pkt.owner = nil p.ownedPkts-- - c.condNE.Broadcast() + c.condNE.Signal() return true } @@ -62,7 +62,7 @@ func (c *Connection) receive(p *Process) *Packet { c.ir = (c.ir + 1) % len(c.pktArray) pkt.owner = p p.ownedPkts++ - c.condNF.Broadcast() + c.condNF.Signal() return pkt } From cd11d652c756a4cd7c7fc7c32fedcec659d0925e Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Tue, 12 Oct 2021 10:30:31 +0300 Subject: [PATCH 11/13] core: implement deadlock detection Tracking per process status for deadlock detection is trivial, however we need to account for connection where there are suspended receivers and senders. To properly handle this scenario we introduce a concept of "live connections". Connection is considered live when there could be further progress. Connection becomes live when: 1. There are both pending receives and sends or 2. Connection is closed and there is a pending receiver. Connection becomes unlive when: 1. There are no more receivers or 2. There are no more pending sends and the connection is empty. --- core/connection.go | 80 +++++++++++++++++++++++++++++++++++------- core/network_status.go | 31 +++++++++------- core/process.go | 3 ++ 3 files changed, 90 insertions(+), 24 deletions(-) diff --git a/core/connection.go b/core/connection.go index edef1c1..25c0452 100644 --- a/core/connection.go +++ b/core/connection.go @@ -18,6 +18,10 @@ type Connection struct { fullName string array []*Connection downStrProc *Process + + markedLive bool + pendingSends int64 + pendingRecvs int64 } func (c *Connection) send(p *Process, pkt *Packet) bool { @@ -28,10 +32,29 @@ func (c *Connection) send(p *Process, pkt *Packet) bool { defer c.mtx.Unlock() fmt.Println(p.name, "Sending", pkt.Contents) c.downStrProc.ensureRunning() - for c.nolockIsFull() { // connection is full - p.transition(SuspendedSend) - c.condNF.Wait() - p.transition(Active) + if c.nolockIsFull() { + // Any connection that has a pair of sends and receives + // should be considered live. + if !c.markedLive && c.pendingRecvs > 0 { + c.markedLive = true + c.network.incLiveConnection() + } + c.pendingSends++ + + for c.nolockIsFull() { // connection is full + p.transition(SuspendedSend) + c.condNF.Wait() + p.transition(Active) + } + + // We unmark liveness when there are no receivers. + // Otherwise there's a possibility that the sender terminates + // and the receiver is still suspended. + c.pendingSends-- + if c.markedLive && c.pendingRecvs == 0 { + c.markedLive = false + c.network.decLiveConnection() + } } fmt.Println(p.name, "Sent", pkt.Contents) c.pktArray[c.is] = pkt @@ -47,15 +70,40 @@ func (c *Connection) receive(p *Process) *Packet { defer c.mtx.Unlock() fmt.Println(p.name, "Receiving") - for c.nolockIsEmpty() { // connection is empty + + if c.nolockIsEmpty() { + // Any connection that has a pair of sends and receives + // should be considered live. It might take some time to + // propagate those items through. + if !c.markedLive && c.pendingSends > 0 { + c.markedLive = true + c.network.incLiveConnection() + } + c.pendingRecvs++ + + for c.nolockIsEmpty() && !c.closed { + p.transition(SuspendedRecv) + c.condNE.Wait() + p.transition(Active) + } + + // Once there are no more receivers we can unmark the connection. + c.pendingRecvs-- + if c.markedLive && c.pendingSends == 0 && c.nolockIsEmpty() { + c.markedLive = false + c.network.decLiveConnection() + } + if c.markedLive && c.pendingRecvs == 0 { + c.markedLive = false + c.network.decLiveConnection() + } + if c.closed { c.condNF.Broadcast() return nil } - p.transition(SuspendedRecv) - c.condNE.Wait() - p.transition(Active) } + pkt := c.pktArray[c.ir] c.pktArray[c.ir] = nil fmt.Println(p.name, "Received", pkt.Contents) @@ -80,16 +128,24 @@ func (c *Connection) decUpstream() { c.upStrmCnt-- if c.upStrmCnt == 0 { - c.closed = true - c.condNE.Broadcast() - c.downStrProc.ensureRunning() - + c.nolockClose() } } func (c *Connection) Close() { c.mtx.Lock() defer c.mtx.Unlock() + c.nolockClose() +} + +func (c *Connection) nolockClose() { + // Mark connection live when there are pending receives, + // otherwise the receivers could be marked as deadlocked + // although they have to receive the close signal. + if !c.markedLive && c.pendingRecvs > 0 { + c.markedLive = true + c.network.incLiveConnection() + } c.closed = true c.condNE.Broadcast() diff --git a/core/network_status.go b/core/network_status.go index c1a5397..b3346f8 100644 --- a/core/network_status.go +++ b/core/network_status.go @@ -5,13 +5,24 @@ import ( "sync" ) -const deadlockDetectionEnabled = false - type networkStatus struct { - mu sync.Mutex - deadlocked bool - running int32 - suspended int32 + mu sync.Mutex + deadlocked bool + running int32 + liveConnections int32 + suspended int32 +} + +func (network *Network) incLiveConnection() { + network.status.mu.Lock() + defer network.status.mu.Unlock() + network.status.liveConnections++ +} + +func (network *Network) decLiveConnection() { + network.status.mu.Lock() + defer network.status.mu.Unlock() + network.status.liveConnections-- } func (network *Network) processTransitioned(from, to ProcessStatus) { @@ -38,7 +49,7 @@ func (network *Network) processTransitioned(from, to ProcessStatus) { leftRunning := network.status.running network.status.mu.Unlock() - if leftRunning <= 0 { + if leftRunning <= 0 && network.status.liveConnections == 0 { network.deadlockDetected() } @@ -54,7 +65,7 @@ func (network *Network) processTransitioned(from, to ProcessStatus) { running, suspended := network.status.running, network.status.suspended network.status.mu.Unlock() - if running == 0 && suspended > 0 { + if running == 0 && suspended > 0 && network.status.liveConnections == 0 { network.deadlockDetected() } @@ -69,10 +80,6 @@ func (network *Network) processTransitioned(from, to ProcessStatus) { } func (network *Network) deadlockDetected() { - if !deadlockDetectionEnabled { - return - } - network.status.mu.Lock() defer network.status.mu.Unlock() diff --git a/core/process.go b/core/process.go index 1ee9ca2..12de363 100644 --- a/core/process.go +++ b/core/process.go @@ -78,6 +78,9 @@ func (p *Process) Receive(c InputConn) *Packet { func (p *Process) ensureRunning() { if !p.transitionFrom(NotStarted, Dormant) { + if p.status() == Terminated { + panic("tried to send to a terminated process") + } return } From c2913ff4c87bdd408136b3bb6e075efd79bb7543 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Tue, 12 Oct 2021 10:42:33 +0300 Subject: [PATCH 12/13] move testdata to appropriate folder * Go uses `testdata` folder to keep testing data. * Use correct filepath separators, otherwise code will have problems on Linux/Mac. * Filepaths are relative to the tests, so there's no need for Getwd. * Remove the temporary file. --- .gitignore | 2 +- gofbp_test.go | 20 +++++--------------- testdata.copy | 5 ----- testdata.txt => testdata/testdata.txt | 0 4 files changed, 6 insertions(+), 21 deletions(-) delete mode 100644 testdata.copy rename testdata.txt => testdata/testdata.txt (100%) diff --git a/.gitignore b/.gitignore index e4413b7..c697517 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,3 @@ -testdata.copy +*.tmp qs_*.* bak \ No newline at end of file diff --git a/gofbp_test.go b/gofbp_test.go index 4ab697f..4cf153b 100644 --- a/gofbp_test.go +++ b/gofbp_test.go @@ -1,7 +1,7 @@ package main import ( - "os" + "path/filepath" "testing" "github.com/jpaulm/gofbp/components/io" @@ -72,13 +72,8 @@ func TestCopyFile(t *testing.T) { proc2 := net.NewProc("WriteFile", &io.WriteFile{}) - path, err := os.Getwd() - if err != nil { - panic("Can't find workspace directory") - } - - net.Initialize(path+"\\testdata.txt", proc1, "FILENAME") - net.Initialize(path+"\\testdata.copy", proc2, "FILENAME") + net.Initialize(filepath.Join("testdata", "testdata.txt"), proc1, "FILENAME") + net.Initialize(filepath.Join("testdata", "copy-file.tmp"), proc2, "FILENAME") net.Connect(proc1, "OUT", proc2, "IN", 6) net.Run() @@ -92,14 +87,9 @@ func TestDoSelect(t *testing.T) { proc3a := net.NewProc("WriteFile", &io.WriteFile{}) proc3b := net.NewProc("WriteToConsole", &testrtn.WriteToConsole{}) - path, err := os.Getwd() - if err != nil { - panic("Can't find workspace directory") - } - - net.Initialize(path+"\\testdata.txt", proc1, "FILENAME") + net.Initialize(filepath.Join("testdata", "testdata.txt"), proc1, "FILENAME") net.Initialize("X", proc2, "PARAM") - net.Initialize(path+"\\testdata.copy", proc3a, "FILENAME") + net.Initialize(filepath.Join("testdata", "do-select.tmp"), proc3a, "FILENAME") net.Connect(proc1, "OUT", proc2, "IN", 6) net.Connect(proc2, "ACC", proc3a, "IN", 6) net.Connect(proc2, "REJ", proc3b, "IN", 6) diff --git a/testdata.copy b/testdata.copy deleted file mode 100644 index a83ef5e..0000000 --- a/testdata.copy +++ /dev/null @@ -1,5 +0,0 @@ - -Features include: - -- delayed start of goroutines (FBP processes), unless `MustRun` attribute is specified or the process has no non-IIP inputs (same as JavaFBP delayed start feature) -- optional output ports - see https://github.com/jpaulm/gofbp/blob/master/components/testrtn/writetoconsole.go diff --git a/testdata.txt b/testdata/testdata.txt similarity index 100% rename from testdata.txt rename to testdata/testdata.txt From f33c3678b6373bf32686be84e674afd28d2f3cb3 Mon Sep 17 00:00:00 2001 From: Egon Elbre Date: Tue, 26 Oct 2021 16:44:54 +0300 Subject: [PATCH 13/13] core: add printing deadlock traces --- core/deadlock.go | 111 +++++++++++++++++++++++++++++++++++++++++ core/network.go | 2 + core/network_status.go | 1 + core/process.go | 17 +++++-- go.mod | 2 + go.sum | 7 +++ 6 files changed, 135 insertions(+), 5 deletions(-) create mode 100644 core/deadlock.go create mode 100644 go.sum diff --git a/core/deadlock.go b/core/deadlock.go new file mode 100644 index 0000000..9ba9859 --- /dev/null +++ b/core/deadlock.go @@ -0,0 +1,111 @@ +package core + +import ( + "bytes" + "fmt" + "runtime/pprof" + "sort" + "strings" + + "github.com/google/pprof/profile" +) + +// goroutineTrace returns summary of the goroutines. +func (n *Network) goroutineTrace() (string, error) { + var pb bytes.Buffer + profiler := pprof.Lookup("goroutine") + if profiler == nil { + return "", fmt.Errorf("unable to find profile") + } + err := profiler.WriteTo(&pb, 0) + if err != nil { + return "", fmt.Errorf("failed to write profile: %w", err) + } + + p, err := profile.ParseData(pb.Bytes()) + if err != nil { + return "", fmt.Errorf("failed to parse profile: %w", err) + } + + return n.summarizeProfile(p, n.id()) +} + +func (n *Network) summarizeProfile(p *profile.Profile, networkID string) (string, error) { + var b strings.Builder + + for _, sample := range p.Sample { + if !existsInSlice(sample.Label["network"], networkID) { + continue + } + + fmt.Fprintf(&b, "count %d @", sample.Value[0]) + + // stack trace summary + + if len(sample.Label)+len(sample.NumLabel) > 0 { + if len(sample.Label) > 0 { + keys := []string{} + for k := range sample.Label { + if k == "network" { + continue + } + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + values := sample.Label[k] + fmt.Fprintf(&b, " %s:", k) + switch len(values) { + case 0: + case 1: + fmt.Fprintf(&b, "%q", values[0]) + default: + fmt.Fprintf(&b, "%q", values) + } + } + } + if len(sample.NumLabel) > 0 { + keys := []string{} + for k := range sample.NumLabel { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + fmt.Fprintf(&b, "%s:%v", k, sample.NumLabel[k]) + } + } + } + fmt.Fprintf(&b, "\n") + + // each line + for _, loc := range sample.Location { + for i, ln := range loc.Line { + if i == 0 { + fmt.Fprintf(&b, "# %#8x", loc.Address) + if loc.IsFolded { + fmt.Fprint(&b, " [F]") + } + } else { + fmt.Fprint(&b, "# ") + } + if fn := ln.Function; fn != nil { + fmt.Fprintf(&b, " %-50s %s:%d", fn.Name, fn.Filename, ln.Line) + } else { + fmt.Fprintf(&b, " ???") + } + fmt.Fprintf(&b, "\n") + } + } + fmt.Fprintf(&b, "\n") + } + return b.String(), nil +} + +func existsInSlice(xs []string, v string) bool { + for _, x := range xs { + if x == v { + return true + } + } + return false +} diff --git a/core/network.go b/core/network.go index 70361bd..06ee7bb 100644 --- a/core/network.go +++ b/core/network.go @@ -43,6 +43,8 @@ func (n *Network) NewProc(nm string, comp Component) *Process { return proc } +func (n *Network) id() string { return fmt.Sprintf("%p", n) } + func (n *Network) NewConnection(cap int) *Connection { conn := &Connection{ network: n, diff --git a/core/network_status.go b/core/network_status.go index b3346f8..977e589 100644 --- a/core/network_status.go +++ b/core/network_status.go @@ -93,5 +93,6 @@ func (network *Network) deadlockDetected() { for _, proc := range network.procs { fmt.Printf("\t%-15s\t%s\n", proc.name, proc.status()) } + fmt.Println(network.goroutineTrace()) panic("Deadlock!") } diff --git a/core/process.go b/core/process.go index 12de363..d027be8 100644 --- a/core/process.go +++ b/core/process.go @@ -1,7 +1,9 @@ package core import ( + "context" "fmt" + "runtime/pprof" "sync/atomic" ) @@ -84,11 +86,16 @@ func (p *Process) ensureRunning() { return } - go func() { - defer p.network.wg.Done() - defer p.transition(Terminated) - p.run() - }() + pprof.Do(context.TODO(), pprof.Labels( + "network", p.network.id(), + "process", p.name, + ), func(c context.Context) { + go func() { + defer p.network.wg.Done() + defer p.transition(Terminated) + p.run() + }() + }) } func (p *Process) run() { diff --git a/go.mod b/go.mod index dfc619e..e6dc086 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,5 @@ module github.com/jpaulm/gofbp go 1.17 + +require github.com/google/pprof v0.0.0-20211008130755-947d60d73cc0 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..416a569 --- /dev/null +++ b/go.sum @@ -0,0 +1,7 @@ +github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI= +github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI= +github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU= +github.com/google/pprof v0.0.0-20211008130755-947d60d73cc0 h1:zHs+jv3LO743/zFGcByu2KmpbliCU2AhjcGgrdTwSG4= +github.com/google/pprof v0.0.0-20211008130755-947d60d73cc0/go.mod h1:KgnwoLYCZ8IQu3XUZ8Nc/bM9CCZFOyjUNOSygVozoDg= +github.com/ianlancetaylor/demangle v0.0.0-20210905161508-09a460cdf81d/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w= +golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=