Allow multiple parents and fix hanging producer

internal/streams/dot.go

@@ -56,11 +56,11 @@ func humanBytes(i int) string {
 }
 
 type node struct {
-	ID     uint32         `json:"id"`
-	Codec  map[string]any `json:"codec"`
-	Parent uint32         `json:"parent"`
-	Childs []uint32       `json:"childs"`
-	Bytes  int            `json:"bytes"`
+	ID      uint32         `json:"id"`
+	Codec   map[string]any `json:"codec"`
+	Parents []uint32       `json:"parents"`
+	Childs  []uint32       `json:"childs"`
+	Bytes   int            `json:"bytes"`
 	//Packets uint32         `json:"packets"`
 	//Drops   uint32         `json:"drops"`
 }
@@ -124,9 +124,11 @@ func (c *conn) appendDOT(dot []byte, group string) []byte {
 		dot = recv.appendDOT(dot, "node")
 	}
 	for _, send := range c.Senders {
-		dot = fmt.Appendf(dot, "%d -> %d [label=%q];\n", send.Parent, c.ID, humanBytes(send.Bytes))
-		//dot = fmt.Appendf(dot, "%d -> %d [label=%q];\n", send.ID, c.ID, humanBytes(send.Bytes))
-		//dot = send.appendDOT(dot, "node")
+		for _, parentID := range send.Parents {
+			dot = fmt.Appendf(dot, "%d -> %d [label=%q];\n", parentID, c.ID, humanBytes(send.Bytes))
+			//dot = fmt.Appendf(dot, "%d -> %d [label=%q];\n", send.ID, c.ID, humanBytes(send.Bytes))
+			//dot = send.appendDOT(dot, "node")
+		}
 	}
 	return dot
 }

pkg/core/node.go

@@ -27,9 +27,9 @@ type Node struct {
 	Input  HandlerFunc
 	Output HandlerFunc
 
-	id     uint32
-	childs []*Node
-	parent *Node
+	id      uint32
+	childs  []*Node
+	parents []*Node
 
 	mu sync.Mutex
 }
@@ -44,30 +44,41 @@ func (n *Node) AppendChild(child *Node) {
 	n.childs = append(n.childs, child)
 	n.mu.Unlock()
 
-	child.parent = n
+	child.mu.Lock()
+	child.parents = append(child.parents, n)
+	child.mu.Unlock()
 }
 
 func (n *Node) RemoveChild(child *Node) {
 	n.mu.Lock()
-	for i, ch := range n.childs {
-		if ch == child {
-			n.childs = append(n.childs[:i], n.childs[i+1:]...)
-			break
-		}
+	if i := Index(n.childs, child); i != -1 {
+		n.childs = append(n.childs[:i], n.childs[i+1:]...)
 	}
 	n.mu.Unlock()
+
+	child.mu.Lock()
+	if i := Index(child.parents, n); i != -1 {
+		child.parents = append(child.parents[:i], child.parents[i+1:]...)
+	}
+	child.mu.Unlock()
 }
 
 func (n *Node) Close() {
-	if parent := n.parent; parent != nil {
-		parent.RemoveChild(n)
+	n.mu.Lock()
+	if n.parents != nil && len(n.parents) > 0 {
+		parents := n.parents
+		n.mu.Unlock()
 
-		if len(parent.childs) == 0 {
-			parent.Close()
+		for _, parent := range parents {
+			parent.RemoveChild(n)
 		}
 	} else {
-		for _, childs := range n.childs {
-			childs.Close()
+		childs := n.childs
+		n.childs = nil
+		n.mu.Unlock()
+
+		for _, child := range childs {
+			child.Close()
 		}
 	}
 }
@@ -83,6 +94,11 @@ func MoveNode(dst, src *Node) {
 	dst.mu.Unlock()
 
 	for _, child := range childs {
-		child.parent = dst
+		child.mu.Lock()
+		if i := Index(child.parents, dst); i != -1 {
+			child.parents = append(child.parents[:i], child.parents[i+1:]...)
+		}
+		child.parents = append(child.parents, dst)
+		child.mu.Unlock()
 	}
 }

pkg/core/track.go

@@ -132,12 +132,12 @@ func (s *Sender) WithParent(parent *Receiver) *Sender {
 
 func (s *Sender) Start() {
 	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	if s.buf == nil || s.done != nil {
+		s.mu.Unlock()
 		return
 	}
 	s.done = make(chan struct{})
+	s.mu.Unlock()
 
 	go func() {
 		// for range on nil chan is OK
@@ -197,21 +197,21 @@ func (r *Receiver) MarshalJSON() ([]byte, error) {
 
 func (s *Sender) MarshalJSON() ([]byte, error) {
 	v := struct {
-		ID      uint32 `json:"id"`
-		Codec   *Codec `json:"codec"`
-		Parent  uint32 `json:"parent,omitempty"`
-		Bytes   int    `json:"bytes,omitempty"`
-		Packets int    `json:"packets,omitempty"`
-		Drops   int    `json:"drops,omitempty"`
+		ID      uint32    `json:"id"`
+		Codec   *Codec    `json:"codec"`
+		Parents []*uint32 `json:"parents,omitempty"`
+		Bytes   int       `json:"bytes,omitempty"`
+		Packets int       `json:"packets,omitempty"`
+		Drops   int       `json:"drops,omitempty"`
 	}{
 		ID:      s.Node.id,
 		Codec:   s.Node.Codec,
 		Bytes:   s.Bytes,
 		Packets: s.Packets,
 		Drops:   s.Drops,
 	}
-	if s.parent != nil {
-		v.Parent = s.parent.id
+	for _, parent := range s.parents {
+		v.Parents = append(v.Parents, &parent.id)
 	}
 	return json.Marshal(v)
 }