Several improvements to the codebase

.gitignore

@@ -1,3 +1,3 @@
-testdata.copy
+*.tmp
 qs_*.*
 bak

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")

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++
 	}
 

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")

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")
 }

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")

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
-}