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/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/connection.go b/core/connection.go index f4ec8ff..25c0452 100644 --- a/core/connection.go +++ b/core/connection.go @@ -3,11 +3,8 @@ package core import ( "fmt" "sync" - "sync/atomic" ) -// Based on https://stackoverflow.com/questions/36857167/how-to-correctly-use-sync-cond - type Connection struct { network *Network pktArray []*Packet @@ -21,52 +18,99 @@ type Connection struct { fullName string array []*Connection downStrProc *Process + + markedLive bool + pendingSends int64 + pendingRecvs int64 } 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() - c.condNE.Broadcast() - for c.nolockIsFull() { // connection is full - atomic.StoreInt32(&p.status, SuspSend) - c.condNF.Wait() - atomic.StoreInt32(&p.status, 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 c.is = (c.is + 1) % len(c.pktArray) pkt.owner = nil p.ownedPkts-- + c.condNE.Signal() return true } 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 + + 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 } - atomic.StoreInt32(&p.status, SuspRecv) - c.condNE.Wait() - atomic.StoreInt32(&p.status, Active) - } + pkt := c.pktArray[c.ir] c.pktArray[c.ir] = nil fmt.Println(p.name, "Received", pkt.Contents) c.ir = (c.ir + 1) % len(c.pktArray) pkt.owner = p p.ownedPkts++ - c.condNF.Broadcast() + c.condNF.Signal() return pkt } @@ -84,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() @@ -130,17 +182,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/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/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..06ee7bb 100644 --- a/core/network.go +++ b/core/network.go @@ -5,17 +5,6 @@ import ( "regexp" "strconv" "sync" - "sync/atomic" - "time" -) - -const ( - Notstarted int32 = iota - Active - Dormant - SuspSend - SuspRecv - Terminated ) type Network struct { @@ -23,6 +12,7 @@ type Network struct { procs map[string]*Process //procList []*Process //driver Process + status networkStatus logFile string wg sync.WaitGroup } @@ -37,24 +27,24 @@ 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) 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 } +func (n *Network) id() string { return fmt.Sprintf("%p", n) } + func (n *Network) NewConnection(cap int) *Connection { conn := &Connection{ network: n, @@ -90,73 +80,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,14 +191,12 @@ 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 conn.fullName = p2.name + "." + in conn.value = initValue - } func (n *Network) Run() { @@ -204,68 +207,17 @@ 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!") - } - - } - }() - - 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/network_status.go b/core/network_status.go new file mode 100644 index 0000000..977e589 --- /dev/null +++ b/core/network_status.go @@ -0,0 +1,98 @@ +package core + +import ( + "fmt" + "sync" +) + +type networkStatus struct { + 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) { + 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.status.liveConnections == 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.status.liveConnections == 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() { + 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()) + } + fmt.Println(network.goroutineTrace()) + panic("Deadlock!") +} diff --git a/core/nulloutport.go b/core/nulloutport.go index 70fb944..ca49966 100644 --- a/core/nulloutport.go +++ b/core/nulloutport.go @@ -2,26 +2,16 @@ 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 +// NullOutPort by default discards the packet. +func (*NullOutPort) send(p *Process, pkt *Packet) bool { + p.Discard(pkt) + return true } -func (c *NullOutPort) SetArrayItem(o *OutPort, i int) {} - -func (c *NullOutPort) ArrayLength() int { - return 0 +func (*NullOutPort) IsConnected() bool { + return false } -func (c *NullOutPort) Close() {} +func (*NullOutPort) Close() {} diff --git a/core/outarrayport.go b/core/outarrayport.go index a543200..1680a96 100644 --- a/core/outarrayport.go +++ b/core/outarrayport.go @@ -1,35 +1,25 @@ 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..081d17d 100644 --- a/core/outport.go +++ b/core/outport.go @@ -1,29 +1,16 @@ 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) + 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) IsConnected() bool { return true } func (o *OutPort) Close() { - o.Conn.decUpstream() + o.conn.decUpstream() } diff --git a/core/outputconn.go b/core/outputconn.go index 1e21d98..e3c58d7 100644 --- a/core/outputconn.go +++ b/core/outputconn.go @@ -1,18 +1,19 @@ package core +type outputCommon interface { + Close() +} + type OutputConn interface { + outputCommon send(*Process, *Packet) bool - //IsEmpty() bool - //IsClosed() bool - SetOptional(b bool) - //GetType() string - Close() + IsConnected() bool } 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..d027be8 100644 --- a/core/process.go +++ b/core/process.go @@ -1,89 +1,69 @@ package core import ( + "context" "fmt" - "sync" + "runtime/pprof" "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 - //done bool - selfStarting bool // process has no non-IIP input ports - //MustRun bool - status int32 - mtx sync.Mutex + + atomicStatus int32 } 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] +func (p *Process) OpenOutPort(name string) OutputConn { + out, ok := p.outPorts[name] + if !ok { + panic(p.name + ": Port name not found (" + name + ")") } - out := p.outPorts[s[0]] + return out.(OutputConn) +} - if len(s) == 2 && s[1] != "opt" { - panic(p.name + ": Invalid 2nd param (" + s[1] + ")") +func (p *Process) OpenOutPortOptional(name string) OutputConn { + out, ok := p.outPorts[name] + if !ok { + 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,116 +79,83 @@ func (p *Process) Receive(c InputConn) *Packet { } func (p *Process) ensureRunning() { - - if !atomic.CompareAndSwapInt32(&p.status, Notstarted, Active) { + if !p.transitionFrom(NotStarted, Dormant) { + if p.status() == Terminated { + panic("tried to send to a terminated process") + } return } - //p.network.wg.Add(1) - go func() { // Process goroutine - defer p.network.wg.Done() - 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 + 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() { - p.mtx.Lock() - defer p.mtx.Unlock() - atomic.StoreInt32(&p.status, Dormant) - defer atomic.StoreInt32(&p.status, Terminated) - defer fmt.Println(p.GetName(), " terminated") - fmt.Println(p.GetName(), " started") +func (p *Process) run() { p.component.Setup(p) - //var allDrained bool - //var hasData bool + runOnce := p.isSelfStarting() + for runOnce || !p.allInputsClosed() { + runOnce = false - allDrained, hasData := p.inputState() - - 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) + p.transition(Active) p.component.Execute(p) // single "activation" - atomic.StoreInt32(&p.status, Dormant) - fmt.Println(p.GetName(), " deactivated") + p.transition(Dormant) if p.ownedPkts > 0 { panic(p.name + " deactivated without disposing of all owned packets") } + } - allDrained, _ := p.inputState() - - if allDrained { - canRun = false - } else { - for _, v := range p.inPorts { - _, b := v.(InitializationConnection) - if b { - v.(*InitializationConnection).resetForNextExecution() - } - } - } + for _, v := range p.outPorts { + v.Close() + } +} +// 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 } - for _, v := range p.outPorts { - _, b := v.(OutputConn) - if b { - v.(OutputConn).Close() - } else { - _, b := v.(OutputArrayConn) - if b { - v.(OutputArrayConn).Close() - } + // 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 } } -} -func (p *Process) isMustRun() bool { - _, hasMustRun := p.component.(ComponentWithMustRun) - return hasMustRun + 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 { @@ -229,6 +176,32 @@ 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 +} + +func (p *Process) status() ProcessStatus { + return ProcessStatus(atomic.LoadInt32(&p.atomicStatus)) +} + +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 +} 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) + } +} 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= 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 c7ff85e..0000000 --- a/testdata.copy +++ /dev/null @@ -1,6 +0,0 @@ -This repo holds the beginning of an FBP implementation in Go - -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