From 36e117f3137545e3e9c6d331dde713924481d700 Mon Sep 17 00:00:00 2001
From: gazebo <gazeboxu@gmail.com>
Date: Wed, 17 Apr 2019 12:26:51 +0800
Subject: [PATCH] =?UTF-8?q?-=20=E4=BF=AE=E5=A4=8Dtasksch=E4=B8=AD=E4=BB=BB?=
 =?UTF-8?q?=E5=8A=A1panic=E5=AF=BC=E8=87=B4=E7=9A=84BUG?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 business/jxutils/tasksch/parallel_task.go | 10 ++++------
 business/jxutils/tasksch/sequence_task.go |  4 +++-
 business/jxutils/tasksch/task.go          | 11 +++++++++++
 3 files changed, 18 insertions(+), 7 deletions(-)

diff --git a/business/jxutils/tasksch/parallel_task.go b/business/jxutils/tasksch/parallel_task.go
index bab59065a..39a867593 100644
--- a/business/jxutils/tasksch/parallel_task.go
+++ b/business/jxutils/tasksch/parallel_task.go
@@ -123,7 +123,9 @@ func (task *ParallelTask) Run() {
 								chanRetVal = retVal
 								goto end
 							} else {
-								result, err := task.worker(task, job, task.params...)
+								result, err := task.callWorker(func() (retVal interface{}, err error) {
+									return task.worker(task, job, task.params...)
+								})
 								// globals.SugarLogger.Debugf("ParallelTask.Run %s, after call worker result:%v, err:%v", task.Name, result, err)
 								task.finishedOneJob(len(job), err)
 								if err == nil {
@@ -146,11 +148,7 @@ func (task *ParallelTask) Run() {
 				}
 			end:
 				// globals.SugarLogger.Debugf("ParallelTask.Run %s, put to chann chanRetVal:%v", task.Name, chanRetVal)
-				task.locker.RLock()
-				if task.Status < TaskStatusEndBegin {
-					task.subFinishChan <- chanRetVal
-				}
-				task.locker.RUnlock()
+				task.subFinishChan <- chanRetVal
 			})
 		}
 		for _, job := range task.jobList {
diff --git a/business/jxutils/tasksch/sequence_task.go b/business/jxutils/tasksch/sequence_task.go
index a11a957c8..74f719f2a 100644
--- a/business/jxutils/tasksch/sequence_task.go
+++ b/business/jxutils/tasksch/sequence_task.go
@@ -39,7 +39,9 @@ func (task *SeqTask) Run() {
 				goto EndFor
 			default:
 			}
-			result, err := task.worker(task, i, task.params...)
+			result, err := task.callWorker(func() (retVal interface{}, err error) {
+				return task.worker(task, i, task.params...)
+			})
 			task.finishedOneJob(1, err)
 			if taskErr = err; taskErr != nil {
 				globals.SugarLogger.Infof("SeqTask.Run %s step:%d failed with error:%v", task.Name, i, err)
diff --git a/business/jxutils/tasksch/task.go b/business/jxutils/tasksch/task.go
index a7aac15c2..52b991888 100644
--- a/business/jxutils/tasksch/task.go
+++ b/business/jxutils/tasksch/task.go
@@ -352,3 +352,14 @@ func (t *BaseTask) buildTaskErrFromDetail() (err error) {
 	}
 	return nil
 }
+
+func (task *BaseTask) callWorker(worker func() (retVal interface{}, err error)) (retVal interface{}, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			globals.SugarLogger.Errorf("callWorker panic:%v", r)
+			err = fmt.Errorf("panic, r:%v", r)
+		}
+	}()
+	retVal, err = worker()
+	return retVal, err
+}