Merge pull request #8 from memory-overflow/develop

Develop

Author: memory-overflow

actuator/function_actuator.go

@@ -0,0 +1,178 @@
+// 封装好的函数执行器
+
+package actuator
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"runtime/debug"
+	"sync"
+	"time"
+
+	framework "github.com/memory-overflow/light-task-scheduler"
+)
+
+// fucntionActuator 函数执行器，同步任务异步化的示例
+type fucntionActuator struct {
+	runFunc         RunFunction         // 执行函数
+	initFunc        InitFunction        // 初始函数
+	callbackChannel chan framework.Task // 回调队列
+
+	runningTask sync.Map // taskId -> [framework.AsyncTaskStatus, cancel function] 映射
+	datatMap    sync.Map // taskId -> interface{} 映射
+
+}
+
+// RunFunction 执行函数的定义，实现该函数的时候，建议使用传入 ctx 进行超时和退出处理
+// data 任务执行完成返回的数据
+type RunFunction func(ctx context.Context, task *framework.Task) (data interface{}, err error)
+
+// InitFunction 任务执行前的初始工作
+type InitFunction func(ctx context.Context, task *framework.Task) (newTask *framework.Task, err error)
+
+// runFunc 待调度的执行函数，注意实现该函数的时候，需要使用传入 ctx 进行超时和退出处理，框架否则无法控制超时时间
+// callbackChannel 用来执行器进行任务回调，返回已经完成的任务，如果不需要回调，传入 nil 即可
+func MakeFucntionActuator(runFunc RunFunction, initFunc InitFunction) (*fucntionActuator, error) {
+	if runFunc == nil {
+		return nil, fmt.Errorf("runFunc is nil")
+	}
+	return &fucntionActuator{
+		runFunc:  runFunc,
+		initFunc: initFunc,
+	}, nil
+}
+
+// SetCallbackChannel 任务配置回调 channel
+func (fc *fucntionActuator) SetCallbackChannel(callbackChannel chan framework.Task) {
+	fc.callbackChannel = callbackChannel
+}
+
+// Init 任务在被调度前的初始化工作
+func (fc *fucntionActuator) Init(ctx context.Context, task *framework.Task) (
+	newTask *framework.Task, err error) {
+	if fc.initFunc != nil {
+		return fc.initFunc(ctx, task)
+	}
+	return task, nil
+}
+
+// Start 执行任务
+func (fc *fucntionActuator) Start(ctx context.Context, ftask *framework.Task) (
+	newTask *framework.Task, ignoreErr bool, err error) {
+	if st, ok := fc.runningTask.Load(ftask.TaskId); ok {
+		status := st.([]interface{})[0].(framework.AsyncTaskStatus).TaskStatus
+		if status == framework.TASK_STATUS_RUNNING || status == framework.TASK_STATUS_SUCCESS {
+			// 任务已经在执行中，不能重复执行
+			return ftask, false, fmt.Errorf("task %s is running", ftask.TaskId)
+		}
+	}
+	runCtx, cancel := context.WithCancel(ctx)
+
+	ftask.TaskStatus = framework.TASK_STATUS_RUNNING
+	ftask.TaskStartTime = time.Now()
+	fc.datatMap.Delete(ftask.TaskId)
+	fc.runningTask.Store(ftask.TaskId,
+		[]interface{}{
+			framework.AsyncTaskStatus{
+				TaskStatus: framework.TASK_STATUS_RUNNING,
+				Progress:   0.0,
+			}, cancel})
+
+	go func() {
+		data, err := func() (data interface{}, err error) {
+			defer func() {
+				if p := recover(); p != nil {
+					data = nil
+
+					err = fmt.Errorf("panic: %v, stacktrace: %s", p, debug.Stack())
+				}
+			}()
+			return fc.runFunc(runCtx, ftask)
+		}()
+		st, ok := fc.runningTask.Load(ftask.TaskId)
+		if !ok {
+			// 任务可能因为超时被删除，或者手动暂停、不处理
+			return
+		}
+		(st.([]interface{})[1].(context.CancelFunc))() // 任务执行完，也要执行对应的 cancel 函数
+		var newStatus framework.AsyncTaskStatus
+		if err != nil {
+			newStatus = framework.AsyncTaskStatus{
+				TaskStatus:   framework.TASK_STATUS_FAILED,
+				Progress:     0.0,
+				FailedReason: err,
+			}
+		} else {
+			newStatus = framework.AsyncTaskStatus{
+				TaskStatus: framework.TASK_STATUS_SUCCESS,
+				Progress:   100.0,
+			}
+			fc.datatMap.Store(ftask.TaskId, data) // 先存结果
+		}
+		fc.runningTask.Store(ftask.TaskId, []interface{}{newStatus, nil})
+		if fc.callbackChannel != nil {
+			// 如果需要回调
+			callbackTask := *ftask
+			callbackTask.TaskStatus = newStatus.TaskStatus
+			callbackTask.TaskEnbTime = time.Now()
+			if newStatus.FailedReason != nil {
+				callbackTask.FailedReason = newStatus.FailedReason.Error()
+			}
+			fc.callbackChannel <- callbackTask
+		}
+	}()
+
+	return ftask, false, nil
+}
+
+func (fc *fucntionActuator) clear(taskId string) {
+	fc.runningTask.Delete(taskId)
+	fc.datatMap.Delete(taskId)
+}
+
+// Stop 停止任务
+func (fc *fucntionActuator) Stop(ctx context.Context, ftask *framework.Task) error {
+	st, ok := fc.runningTask.Load(ftask.TaskId)
+	if !ok {
+		// 未找到任务
+		fc.datatMap.Delete(ftask.TaskId)
+		return nil
+	}
+	fc.clear(ftask.TaskId)
+	(st.([]interface{})[1].(context.CancelFunc))() // cancel 函数 ctx
+	return nil
+}
+
+// GetAsyncTaskStatus 批量获取任务状态
+func (fc *fucntionActuator) GetAsyncTaskStatus(ctx context.Context, ftasks []framework.Task) (
+	status []framework.AsyncTaskStatus, err error) {
+	for _, ftask := range ftasks {
+		fstatus, ok := fc.runningTask.Load(ftask.TaskId)
+		if !ok {
+			status = append(status, framework.AsyncTaskStatus{
+				TaskStatus:   framework.TASK_STATUS_FAILED,
+				FailedReason: errors.New("同步任务未找到"),
+				Progress:     float32(0.0),
+			})
+		} else {
+			st := fstatus.([]interface{})[0].(framework.AsyncTaskStatus)
+			if st.TaskStatus != framework.TASK_STATUS_RUNNING {
+				fc.runningTask.Delete(ftask.TaskId) // delete task status after query if task finished
+			}
+			status = append(status, st)
+		}
+	}
+	return status, nil
+}
+
+// GetOutput ...
+func (fc *fucntionActuator) GetOutput(ctx context.Context, ftask *framework.Task) (
+	data interface{}, err error) {
+	res, ok := fc.datatMap.Load(ftask.TaskId)
+	if !ok {
+		return nil, fmt.Errorf("not found result for task %s", ftask.TaskId)
+	}
+	fc.clear(ftask.TaskId)
+	return res, nil
+}

callback_receiver.go

@@ -0,0 +1,8 @@
+package lighttaskscheduler
+
+import "context"
+
+// CallbackReceiver 任务状态回调接收器
+type CallbackReceiver interface {
+	GetCallbackChannel(ctx context.Context) (taskChannel chan Task)
+}

container/memory_container/queue_container.go

@@ -3,7 +3,6 @@ package memeorycontainer
 import (
 	"context"
 	"fmt"
-	"log"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -95,8 +94,13 @@ func (q *queueContainer) ToRunningStatus(ctx context.Context, task *lighttasksch
 	newTask *lighttaskscheduler.Task, err error) {
 	task.TaskStartTime = time.Now()
 	task.TaskStatus = lighttaskscheduler.TASK_STATUS_RUNNING
-	if _, ok := q.runningTaskMap.LoadOrStore(task.TaskId, *task); !ok {
+	t, ok := q.runningTaskMap.LoadOrStore(task.TaskId, *task)
+	if !ok {
 		atomic.AddInt32(&q.runningTaskCount, 1)
+	} else {
+		nt := t.(lighttaskscheduler.Task)
+		nt.TaskAttemptsTime = task.TaskAttemptsTime
+		q.runningTaskMap.Store(task.TaskId, nt)
 	}
 	return task, nil
 }
@@ -135,8 +139,8 @@ func (q *queueContainer) ToFailedStatus(ctx context.Context, task *lighttasksche
 	if _, ok := q.runningTaskMap.LoadAndDelete(task.TaskId); ok {
 		atomic.AddInt32(&q.runningTaskCount, -1)
 	}
-	log.Printf("failed task %s, reason %v", task.TaskId, reason)
 	task.TaskStatus = lighttaskscheduler.TASK_STATUS_FAILED
+	task.FailedReason = reason.Error()
 	return task, nil
 }
 
@@ -154,6 +158,9 @@ func (q *queueContainer) ToExportStatus(ctx context.Context, task *lighttasksche
 // ToSuccessStatus 转移到执行成功状态
 func (q *queueContainer) ToSuccessStatus(ctx context.Context, task *lighttaskscheduler.Task) (
 	newTask *lighttaskscheduler.Task, err error) {
+	if _, ok := q.runningTaskMap.LoadAndDelete(task.TaskId); ok {
+		atomic.AddInt32(&q.runningTaskCount, -1)
+	}
 	task.TaskStatus = lighttaskscheduler.TASK_STATUS_SUCCESS
 	return task, nil
 }