Update add example

Author: memory-overflow

actuator/function_actuator.go

@@ -0,0 +1,158 @@
+// 封装好的函数执行器
+
+package actuator
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	framework "github.com/memory-overflow/light-task-scheduler"
+	lighttaskscheduler "github.com/memory-overflow/light-task-scheduler"
+)
+
+// fucntionActuator 函数执行器，同步任务异步化的示例
+type fucntionActuator struct {
+	runFunc         RunFunction         // 执行函数
+	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)
+
+// runFunc 待调度的执行函数，注意实现该函数的时候，需要使用传入 ctx 进行超时和退出处理，框架否则无法控制超时时间
+// callbackChannel 用来执行器进行任务回调，返回已经完成的任务，如果不需要回调，传入 nil 即可
+func MakeFucntionActuator(runFunc RunFunction, callbackChannel chan framework.Task) (*fucntionActuator, error) {
+	if runFunc == nil {
+		return nil, fmt.Errorf("runFunc is nil")
+	}
+	return &fucntionActuator{
+		runFunc:         runFunc,
+		callbackChannel: callbackChannel,
+	}, nil
+}
+
+// Init 任务在被调度前的初始化工作
+func (fc *fucntionActuator) Init(ctx context.Context, task *framework.Task) (
+	newTask *framework.Task, err error) {
+	// 无初始化工作
+	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 == lighttaskscheduler.TASK_STATUS_RUNNING || status == lighttaskscheduler.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 := 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
+}

container/combination_container.go

@@ -160,15 +160,3 @@ func (c *combinationContainer) UpdateRunningTaskStatus(ctx context.Context,
 	}
 	return nil
 }
-
-// SaveData 导出任务输出，自行处理任务结果
-func (c *combinationContainer) SaveData(ctx context.Context, ftask *lighttaskscheduler.Task,
-	data interface{}) error {
-	if err := c.memeoryContainer.SaveData(ctx, ftask, data); err != nil {
-		return err
-	}
-	if err := c.persistContainer.SaveData(ctx, ftask, data); err != nil {
-		return err
-	}
-	return nil
-}

container/memory_container/queue_container.go

@@ -10,8 +10,6 @@ import (
 	lighttaskscheduler "github.com/memory-overflow/light-task-scheduler"
 )
 
-type SaveDataFunc func(context.Context, *lighttaskscheduler.Task, interface{}) error
-
 // queueContainer 队列型容器，任务无状态，无优先级，先进先出，任务数据，多进程数据无法共享数据
 type queueContainer struct {
 	MemeoryContainer
@@ -22,16 +20,13 @@ type queueContainer struct {
 
 	waitingTasks chan lighttaskscheduler.Task
 	timeout      time.Duration
-	savefunc     SaveDataFunc
 }
 
 // MakeQueueContainer 构造队列型任务容器, size 表示队列的大小, timeout 表示队列读取的超时时间
-// 需要提供一个 savefunc 处理任务结束以后，任务数据如何处理，如何存储，如果不传，默认不处理任务结果
-func MakeQueueContainer(size uint32, timeout time.Duration, savefunc SaveDataFunc) *queueContainer {
+func MakeQueueContainer(size uint32, timeout time.Duration) *queueContainer {
 	return &queueContainer{
 		waitingTasks: make(chan lighttaskscheduler.Task, size),
 		timeout:      timeout,
-		savefunc:     savefunc,
 	}
 }
 
@@ -99,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
 }
@@ -158,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
 }
@@ -167,12 +170,3 @@ func (q *queueContainer) UpdateRunningTaskStatus(ctx context.Context,
 	task *lighttaskscheduler.Task, status lighttaskscheduler.AsyncTaskStatus) error {
 	return nil
 }
-
-// SaveData 保存任务结果
-func (q *queueContainer) SaveData(ctx context.Context, ftask *lighttaskscheduler.Task,
-	data interface{}) error {
-	if q.savefunc != nil {
-		return q.savefunc(ctx, ftask, data)
-	}
-	return nil
-}

example/add_example/add/add_actuator.go

@@ -1,4 +1,4 @@
-package addexample
+package add
 
 import "time"