fix(executor): hang when task panic

executor.go

@@ -39,11 +39,6 @@ func NewExecutor(concurrency uint) Executor {
 }
 
 func (e *ExecutorImpl) Run(tf *TaskFlow) Executor {
-	// register graceful exit when pinc
-	e.pool.SetPanicHandler(func(ctx *context.Context, i interface{}) {
-
-	})
-
 	tf.graph.setup()
 	for _, node := range tf.graph.entries {
 		e.schedule(node)
@@ -59,12 +54,12 @@ func (e *ExecutorImpl) invokeGraph(g *Graph, parentSpan *span) {
 	ctx := context.Background()
 	for {
 		g.scheCond.L.Lock()
-		for g.JoinCounter() != 0 && e.wq.Len() == 0 {
+		for g.JoinCounter() != 0 && e.wq.Len() == 0 && !g.canceled.Load() {
 			g.scheCond.Wait()
 		}
 		g.scheCond.L.Unlock()
 
-		if g.JoinCounter() == 0 {
+		if g.JoinCounter() == 0 || g.canceled.Load() {
 			break
 		}
 
@@ -86,25 +81,30 @@ func (e *ExecutorImpl) invokeNode(ctx *context.Context, node *Node, parentSpan *
 				typ:  NodeStatic,
 				name: node.name,
 			}, begin: time.Now(), parent: parentSpan}
+
 			defer func() {
 				span.end = time.Now()
 				span.extra.success = true
-				e.profiler.AddSpan(&span)
+				if r := recover(); r != nil {
+					fmt.Println("node", node.name, "recovered ", r)
+					node.g.canceled.Store(true)
+				} else {
+					e.profiler.AddSpan(&span) // remove canceled node span
+				}
+
+				e.wg.Done()
+				node.drop()
+				for _, n := range node.successors {
+					if n.JoinCounter() == 0 {
+						e.schedule(n)
+					}
+				}
+				node.g.scheCond.Signal()
 			}()
 
-			defer e.wg.Done()
 			node.state.Store(kNodeStateRunning)
-			defer node.state.Store(kNodeStateFinished)
-
 			p.handle()
-			node.drop()
-			for _, n := range node.successors {
-				// fmt.Println("put", n.Name)
-				if n.JoinCounter() == 0 {
-					e.schedule(n)
-				}
-			}
-			node.g.scheCond.Signal()
+			node.state.Store(kNodeStateFinished)
 		})
 	case *Subflow:
 		e.pool.Go(func() {
@@ -115,36 +115,45 @@ func (e *ExecutorImpl) invokeNode(ctx *context.Context, node *Node, parentSpan *
 			defer func() {
 				span.end = time.Now()
 				span.extra.success = true
-				e.profiler.AddSpan(&span)
+				if r := recover(); r != nil {
+					fmt.Println("subflow", node.name, "recovered ", r)
+					node.g.canceled.Store(true)
+					p.g.canceled.Store(true)
+				} else {
+					e.profiler.AddSpan(&span) // remove canceled node span
+				}
+				e.wg.Done()
+				e.scheduleGraph(p.g, &span)
+				node.drop()
+
+				for _, n := range node.successors {
+					if n.JoinCounter() == 0 {
+						e.schedule(n)
+					}
+				}
+				node.g.scheCond.Signal()
 			}()
 
-			defer e.wg.Done()
 			node.state.Store(kNodeStateRunning)
-			defer node.state.Store(kNodeStateFinished)
-
 			if !p.g.instancelized {
 				p.handle(p)
 			}
 			p.g.instancelized = true
+			node.state.Store(kNodeStateFinished)
 
-			e.scheduleGraph(p.g, &span)
-			node.drop()
-
-			for _, n := range node.successors {
-				if n.JoinCounter() == 0 {
-					e.schedule(n)
-				}
-			}
-
-			node.g.scheCond.Signal()
 		})
 	default:
-		fmt.Println("exit: ", node.name)
-		panic("do nothing")
+		panic("unsupported node")
 	}
 }
 
 func (e *ExecutorImpl) schedule(node *Node) {
+	if node.g.canceled.Load() {
+		node.g.scheCond.Signal()
+		fmt.Println("node cannot be scheduled, cuz graph canceled", node.name)
+		return
+	}
+
 	e.wg.Add(1)
 	e.wq.Put(node)
 	node.state.Store(kNodeStateWaiting)

graph.go

@@ -2,6 +2,7 @@ package gotaskflow
 
 import (
 	"sync"
+	"sync/atomic"
 
 	"github.com/noneback/go-taskflow/utils"
 )
@@ -13,6 +14,7 @@ type Graph struct {
 	entries       []*Node
 	scheCond      *sync.Cond
 	instancelized bool
+	canceled      atomic.Bool // only changes when task in graph panic
 }
 
 func newGraph(name string) *Graph {

node.go

@@ -12,6 +12,8 @@ const (
 	kNodeStateWaiting  = int32(1)
 	kNodeStateRunning  = int32(2)
 	kNodeStateFinished = int32(3)
+	kNodeStateFailed   = int32(4)
+	kNodeStateCanceled = int32(5)
 )
 
 type NodeType string

taskflow_test.go

@@ -6,6 +6,7 @@ import (
 	_ "net/http/pprof"
 	"os"
 	"testing"
+	"time"
 
 	gotaskflow "github.com/noneback/go-taskflow"
 )
@@ -135,17 +136,69 @@ func TestSubflow(t *testing.T) {
 	// }
 }
 
-// func TestTaskflowPanic(t *testing.T) {
-// 	A, B, C :=
-// 		gotaskflow.NewTask("A", func() {
-// 			fmt.Println("A")
-// 		}),
-// 		gotaskflow.NewTask("B", func() {
-// 			fmt.Println("B")
-// 		}),
-// 		gotaskflow.NewTask("C", func() {
-// 			fmt.Println("C")
-// 			panic("panic C")
-// 		})
-
-// }
+// ERROR robust testing
+func TestTaskflowPanic(t *testing.T) {
+	A, B, C :=
+		gotaskflow.NewTask("A", func() {
+			fmt.Println("A")
+		}),
+		gotaskflow.NewTask("B", func() {
+			fmt.Println("B")
+		}),
+		gotaskflow.NewTask("C", func() {
+			fmt.Println("C")
+			panic("panic C")
+		})
+	A.Precede(B)
+	C.Precede(B)
+	tf := gotaskflow.NewTaskFlow("G")
+	tf.Push(A, B, C)
+
+	exector.Run(tf).Wait()
+}
+
+func TestSubflowPanic(t *testing.T) {
+	A, B, C :=
+		gotaskflow.NewTask("A", func() {
+			fmt.Println("A")
+		}),
+		gotaskflow.NewTask("B", func() {
+			fmt.Println("B")
+		}),
+		gotaskflow.NewTask("C", func() {
+			fmt.Println("C")
+		})
+	A.Precede(B)
+	C.Precede(B)
+
+	subflow := gotaskflow.NewSubflow("sub1", func(sf *gotaskflow.Subflow) {
+		A2, B2, C2 :=
+			gotaskflow.NewTask("A2", func() {
+				fmt.Println("A2")
+				time.Sleep(1 * time.Second)
+			}),
+			gotaskflow.NewTask("B2", func() {
+				fmt.Println("B2")
+			}),
+			gotaskflow.NewTask("C2", func() {
+				fmt.Println("C2")
+				panic("C2 paniced")
+			})
+		A2.Precede(B2)
+		C2.Precede(B2)
+		sf.Push(A2, B2, C2)
+		panic("subflow panic")
+	})
+
+	subflow.Precede(B)
+
+	tf := gotaskflow.NewTaskFlow("G")
+	tf.Push(A, B, C)
+	tf.Push(subflow)
+	exector.Run(tf)
+	exector.Wait()
+	// if err := gotaskflow.Visualizer.Visualize(tf, os.Stdout); err != nil {
+	// 	log.Fatal(err)
+	// }
+	exector.Profile(os.Stdout)
+}

utils/copool.go

@@ -3,7 +3,6 @@ package utils
 import (
 	"context"
 	"fmt"
-	"log"
 	"runtime/debug"
 	"sync"
 )
@@ -59,7 +58,7 @@ func (cp *Copool) CtxGo(ctx *context.Context, f func()) {
 					cp.panicHandler(ctx, r)
 				} else {
 					msg := fmt.Sprintf("[panic] copool: %v: %s", r, debug.Stack())
-					log.Println(msg)
+					fmt.Println(msg)
 				}
 			}
 		}()

utils/utils_test.go

@@ -1,6 +1,7 @@
 package utils
 
 import (
+	"fmt"
 	"testing"
 )
 
@@ -74,3 +75,22 @@ func TestSet(t *testing.T) {
 		t.Errorf("Expected count to be -1, got %d", rc.Value())
 	}
 }
+
+func TestPanic(t *testing.T) {
+	f := func() {
+		defer func() {
+			// 使用 recover 捕获 panic
+			if r := recover(); r != nil {
+				fmt.Println("Recovered in causePanic:", r)
+			}
+			fmt.Println("1")
+		}()
+
+		fmt.Println("result")
+		// panic("Atest")
+	}
+	f()
+}
+// result
+// Recovered in causePanic: Atest
+// 1