fix: escape death lock when consume after close

docs/architecture.md

@@ -114,8 +114,8 @@ To modify node specifications, use the Command-Line Interface (CLI) to update th
 
 - kind: switch
   name: catch
-  match:
-    - when: self == "invalid argument"
+  matches:
+    - when: self == "unsupported type" || self == "unsupported value"
       port: out[0]
     - when: 'true'
       port: out[1]
@@ -164,9 +164,9 @@ Compiled nodes are converted into symbols and stored in a symbol table, which co
    +--------------------------+   +-------------------+
    |         Database         |   |       Loader      |
    |  +--------------------+  |   |  +-------------+  |
-   |  | Node Specification |  |-->|  |    Scheme   |  |
+   |  | Node Specification |  |   |  |    Scheme   |  |
    |  +--------------------+  |   |  |  +-------+  |  |
-   |  | Node Specification |  |   |  |  | Codec |  |  |--+
+   |  | Node Specification |  |-->|  |  | Codec |  |  |--+
    |  +--------------------+  |   |  |  +-------+  |  |  |
    |  | Node Specification |  |   |  +-------------+  |  |
    |  +--------------------+  |   +-------------------+  |

docs/architecture_kr.md

@@ -114,8 +114,8 @@
 
 - kind: switch
   name: catch
-  match:
-    - when: self == "invalid argument"
+  matches:
+    - when: self == "unsupported type" || self == "unsupported value"
       port: out[0]
     - when: 'true'
       port: out[1]
@@ -164,9 +164,9 @@
    +--------------------------+   +-------------------+
    |         Database         |   |       Loader      |
    |  +--------------------+  |   |  +-------------+  |
-   |  | Node Specification |  |-->|  |    Scheme   |  |
+   |  | Node Specification |  |   |  |    Scheme   |  |
    |  +--------------------+  |   |  |  +-------+  |  |
-   |  | Node Specification |  |   |  |  | Codec |  |  |--+
+   |  | Node Specification |  |-->|  |  | Codec |  |  |--+
    |  +--------------------+  |   |  |  +-------+  |  |  |
    |  | Node Specification |  |   |  +-------------+  |  |
    |  +--------------------+  |   +-------------------+  |

examples/system.yaml

@@ -77,8 +77,8 @@
 
 - kind: switch
   name: catch
-  match:
-    - when: self == "invalid argument"
+  matches:
+    - when: self == "unsupported type" || self == "unsupported value"
       port: out[0]
     - when: 'true'
       port: out[1]

ext/pkg/control/block.go

@@ -79,13 +79,13 @@ func (n *BlockNode) In(name string) *port.InPort {
 	n.mu.Lock()
 	defer n.mu.Unlock()
 
-	if p, ok := n.inPorts[name]; ok {
-		return p
+	if inPort, ok := n.inPorts[name]; ok {
+		return inPort
 	}
 	if len(n.nodes) > 0 {
-		if p := n.nodes[0].In(name); p != nil {
-			n.inPorts[name] = p
-			return p
+		if inPort := n.nodes[0].In(name); inPort != nil {
+			n.inPorts[name] = inPort
+			return inPort
 		}
 	}
 	return nil
@@ -96,34 +96,34 @@ func (n *BlockNode) Out(name string) *port.OutPort {
 	n.mu.Lock()
 	defer n.mu.Unlock()
 
-	if p, ok := n.outPorts[name]; ok {
-		return p
+	if outPort, ok := n.outPorts[name]; ok {
+		return outPort
 	}
 	if len(n.nodes) > 0 {
-		if p := n.nodes[len(n.nodes)-1].Out(name); p != nil {
-			n.outPorts[name] = p
-			return p
+		if outPort := n.nodes[len(n.nodes)-1].Out(name); outPort != nil {
+			n.outPorts[name] = outPort
+			return outPort
 		}
 	}
 	return nil
 }
 
 // Close closes all ports associated with the node.
 func (n *BlockNode) Close() error {
-	n.mu.Lock()
-	defer n.mu.Unlock()
+	n.mu.RLock()
+	defer n.mu.RUnlock()
 
-	for _, n := range n.nodes {
-		if err := n.Close(); err != nil {
+	for _, node := range n.nodes {
+		if err := node.Close(); err != nil {
 			return err
 		}
 	}
 
-	for _, p := range n.inPorts {
-		p.Close()
+	for _, inPort := range n.inPorts {
+		inPort.Close()
 	}
-	for _, p := range n.outPorts {
-		p.Close()
+	for _, outPort := range n.outPorts {
+		outPort.Close()
 	}
 
 	return nil

ext/pkg/control/call.go

@@ -1,8 +1,6 @@
 package control
 
 import (
-	"sync"
-
 	"github.com/siyul-park/uniflow/pkg/node"
 	"github.com/siyul-park/uniflow/pkg/packet"
 	"github.com/siyul-park/uniflow/pkg/port"
@@ -18,7 +16,6 @@ type CallNode struct {
 	inPort   *port.InPort
 	outPorts []*port.OutPort
 	errPort  *port.OutPort
-	mu       sync.RWMutex
 }
 
 // CallNodeSpec holds the specifications for creating a CallNode.
@@ -49,23 +46,16 @@ func NewCallNode() *CallNode {
 
 // In returns the input port with the specified name.
 func (n *CallNode) In(name string) *port.InPort {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	switch name {
 	case node.PortIn:
 		return n.inPort
 	default:
 	}
-
 	return nil
 }
 
 // Out returns the output port with the specified name.
 func (n *CallNode) Out(name string) *port.OutPort {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	switch name {
 	case node.PortOut:
 		return n.outPorts[0]
@@ -79,15 +69,11 @@ func (n *CallNode) Out(name string) *port.OutPort {
 			}
 		}
 	}
-
 	return nil
 }
 
 // Close closes all ports associated with the node.
 func (n *CallNode) Close() error {
-	n.mu.Lock()
-	defer n.mu.Unlock()
-
 	n.inPort.Close()
 	for _, outPort := range n.outPorts {
 		outPort.Close()
@@ -99,9 +85,6 @@ func (n *CallNode) Close() error {
 }
 
 func (n *CallNode) forward(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	inReader := n.inPort.Open(proc)
 	outWriter0 := n.outPorts[0].Open(proc)
 	outWriter1 := n.outPorts[1].Open(proc)
@@ -128,9 +111,6 @@ func (n *CallNode) forward(proc *process.Process) {
 }
 
 func (n *CallNode) backward0(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	outWriter0 := n.outPorts[0].Open(proc)
 
 	for {
@@ -144,9 +124,6 @@ func (n *CallNode) backward0(proc *process.Process) {
 }
 
 func (n *CallNode) backward1(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	outWriter1 := n.outPorts[1].Open(proc)
 
 	for {
@@ -160,9 +137,6 @@ func (n *CallNode) backward1(proc *process.Process) {
 }
 
 func (n *CallNode) catch(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	errWriter := n.errPort.Open(proc)
 
 	for {

ext/pkg/control/fork.go

@@ -1,8 +1,6 @@
 package control
 
 import (
-	"sync"
-
 	"github.com/siyul-park/uniflow/pkg/node"
 	"github.com/siyul-park/uniflow/pkg/packet"
 	"github.com/siyul-park/uniflow/pkg/port"
@@ -17,7 +15,6 @@ type ForkNode struct {
 	inPort  *port.InPort
 	outPort *port.OutPort
 	errPort *port.OutPort
-	mu      sync.RWMutex
 }
 
 // ForkNodeSpec holds the specifications for creating a ForkNode.
@@ -46,39 +43,28 @@ func NewForkNode() *ForkNode {
 
 // In returns the input port with the specified name.
 func (n *ForkNode) In(name string) *port.InPort {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	switch name {
 	case node.PortIn:
 		return n.inPort
 	default:
 	}
-
 	return nil
 }
 
 // Out returns the output port with the specified name.
 func (n *ForkNode) Out(name string) *port.OutPort {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	switch name {
 	case node.PortOut:
 		return n.outPort
 	case node.PortErr:
 		return n.errPort
 	default:
 	}
-
 	return nil
 }
 
 // Close closes all ports associated with the node.
 func (n *ForkNode) Close() error {
-	n.mu.Lock()
-	defer n.mu.Unlock()
-
 	n.inPort.Close()
 	n.outPort.Close()
 	n.errPort.Close()
@@ -87,9 +73,6 @@ func (n *ForkNode) Close() error {
 }
 
 func (n *ForkNode) forward(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	inReader := n.inPort.Open(proc)
 
 	for {
@@ -107,9 +90,6 @@ func (n *ForkNode) forward(proc *process.Process) {
 }
 
 func (n *ForkNode) backward(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	outWriter := n.outPort.Open(proc)
 	errWriter := n.errPort.Open(proc)
 
@@ -132,9 +112,6 @@ func (n *ForkNode) backward(proc *process.Process) {
 }
 
 func (n *ForkNode) catch(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	errWriter := n.errPort.Open(proc)
 
 	for {

ext/pkg/control/if.go

@@ -2,7 +2,6 @@ package control
 
 import (
 	"reflect"
-	"sync"
 
 	"github.com/siyul-park/uniflow/ext/pkg/language"
 	"github.com/siyul-park/uniflow/pkg/node"
@@ -21,7 +20,6 @@ type IfNode struct {
 	inPort    *port.InPort
 	outPorts  []*port.OutPort
 	errPort   *port.OutPort
-	mu        sync.RWMutex
 }
 
 // IfNodeSpec holds specifications for creating an IfNode.
@@ -58,9 +56,6 @@ func NewIfNode(condition func(any) (bool, error)) *IfNode {
 
 // In returns the input port with the specified name.
 func (n *IfNode) In(name string) *port.InPort {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	switch name {
 	case node.PortIn:
 		return n.inPort
@@ -72,9 +67,6 @@ func (n *IfNode) In(name string) *port.InPort {
 
 // Out returns the output port with the specified name.
 func (n *IfNode) Out(name string) *port.OutPort {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	switch name {
 	case node.PortOut:
 		return n.outPorts[0]
@@ -93,9 +85,6 @@ func (n *IfNode) Out(name string) *port.OutPort {
 
 // Close closes all ports associated with the node.
 func (n *IfNode) Close() error {
-	n.mu.Lock()
-	defer n.mu.Unlock()
-
 	n.inPort.Close()
 	for _, outPort := range n.outPorts {
 		outPort.Close()
@@ -107,9 +96,6 @@ func (n *IfNode) Close() error {
 }
 
 func (n *IfNode) forward(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	inReader := n.inPort.Open(proc)
 	outWriter0 := n.outPorts[0].Open(proc)
 	outWriter1 := n.outPorts[1].Open(proc)
@@ -138,9 +124,6 @@ func (n *IfNode) forward(proc *process.Process) {
 }
 
 func (n *IfNode) backward(proc *process.Process, index int) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	outWriter := n.outPorts[index].Open(proc)
 
 	for {
@@ -154,9 +137,6 @@ func (n *IfNode) backward(proc *process.Process, index int) {
 }
 
 func (n *IfNode) catch(proc *process.Process) {
-	n.mu.RLock()
-	defer n.mu.RUnlock()
-
 	errWriter := n.errPort.Open(proc)
 
 	for {