对于以下这段代码，我依然还有个问题，我现在有个需求是我有可能会把一个任务在某一层拆成很多任务，比如我的任务输入data = 10，我需要把输出data + 10，data - 10等多个结果，并以相同的任务ID传给下一层，但是现在并不支持
package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

// Task 定义任务的数据结构
type Task struct {
	ID      int
	Data    interface{}
	Result  interface{}
	Retries int
}

// TaskFunc 定义任务处理函数类型
type TaskFunc func(ctx context.Context, data interface{}) (interface{}, error)

// Layer 定义流水线的每一层
type Layer struct {
	workerCount int
	timeout     time.Duration
	maxRetries  int
	taskFunc    TaskFunc
}

// Pipeline 定义整个流水线结构
type Pipeline struct {
	layers []*Layer
}

// NewPipeline 创建一个新的流水线实例
func NewPipeline() *Pipeline {
	return &Pipeline{
		layers: make([]*Layer, 0),
	}
}

// AddLayer 向流水线中添加一层
func (p *Pipeline) AddLayer(taskFunc TaskFunc, workerCount int, timeout time.Duration, maxRetries int) {
	layer := &Layer{
		workerCount: workerCount,
		timeout:     timeout,
		maxRetries:  maxRetries,
		taskFunc:    taskFunc,
	}
	p.layers = append(p.layers, layer)
}

// Run 启动流水线处理任务
func (p *Pipeline) Run(input []interface{}) ([]interface{}, error) {
	if len(p.layers) == 0 {
		return nil, errors.New("pipeline has no layers")
	}

	// 初始化任务通道
	inCh := make(chan *Task, len(input))

	// 将输入数据封装成任务并发送到第一个通道
	go func() {
		for i, data := range input {
			inCh <- &Task{ID: i, Data: data}
		}
		close(inCh)
	}()

	// 设置初始输入
	currentIn := inCh
	var currentOut chan *Task

	// 用于取消整个流水线的上下文（仅在严重错误时使用）
	ctxPipeline, cancelPipeline := context.WithCancel(context.Background())
	defer cancelPipeline()

	// 遍历每一层，构建流水线
	for _, layer := range p.layers {
		currentOut = make(chan *Task, len(input))
		var wg sync.WaitGroup
		// 启动工作协程
		for w := 0; w < layer.workerCount; w++ {
			wg.Add(1)
			go func(layer *Layer, in <-chan *Task, out chan<- *Task) {
				defer wg.Done()
				for task := range in {
					retries := task.Retries
					for {
						// 创建带超时的上下文
						ctx, cancel := context.WithTimeout(ctxPipeline, layer.timeout)
						result, err := layer.taskFunc(ctx, task.Data)
						cancel()

						if err != nil {
							retries++
							if retries > layer.maxRetries {
								// 超过重试次数，记录失败
								fmt.Printf("Task ID %d failed after %d retries: %v\n", task.ID, retries-1, err)
								task.Result = nil
								break
							}
							task.Retries = retries
							fmt.Printf("Task ID %d retry %d due to error: %v\n", task.ID, retries, err)
							continue
						}
						task.Result = result
						break
					}
					// 将当前层的结果赋值给 Data，以便下一层使用
					task.Data = task.Result
					out <- task
				}
			}(layer, currentIn, currentOut)
		}

		// 等待所有工作协程完成后关闭输出通道
		go func(wg *sync.WaitGroup, out chan *Task) {
			wg.Wait()
			close(out)
		}(&wg, currentOut)

		// 准备下一层的输入
		currentIn = currentOut
	}

	// 收集最终结果
	results := make([]interface{}, len(input))
	done := make(chan struct{})
	var pipelineErr error
	var mu sync.Mutex // 用于保护 pipelineErr

	go func() {
		for task := range currentOut {
			if task.Result == nil {
				// 如果某个任务失败，记录错误信息
				mu.Lock()
				pipelineErr = errors.New("pipeline terminated with task failures")
				mu.Unlock()
			}
			results[task.ID] = task.Result
		}
		close(done)
	}()

	// 等待所有任务完成或被取消
	select {
	case <-done:
		return results, pipelineErr
	case <-ctxPipeline.Done():
		return nil, pipelineErr
	}
}

// 示例任务函数
func exampleTask(ctx context.Context, data interface{}) (interface{}, error) {
	// 模拟任务处理时间
	time.Sleep(100 * time.Millisecond)
	if num, ok := data.(int); ok {
		/*if num%5 == 0 {
		        return nil, fmt.Errorf("error processing data %d", num)
		}*/
		fmt.Println(fmt.Sprintf("Processing task %d with input %v", num, data))
		return num * num, nil
	}
	return nil, errors.New("invalid data type")
}

// 示例任务函数
func exampleTask2(ctx context.Context, data interface{}) (interface{}, error) {
	// 模拟任务处理时间
	time.Sleep(100 * time.Millisecond)
	if num, ok := data.(int); ok {
		/*if num%5 == 0 {
		        return nil, fmt.Errorf("error processing data %d", num)
		}*/
		fmt.Println(fmt.Sprintf("Processing task %d with input %v", num, data))
		return num * 2, nil
	}
	return nil, errors.New("invalid data type")
}

func main() {
	// 创建流水线
	pipeline := NewPipeline()

	// 添加三层，每层配置不同的参数
	pipeline.AddLayer(exampleTask, 5, 1*time.Second, 3)
	pipeline.AddLayer(exampleTask, 3, 1*time.Second, 2)
	pipeline.AddLayer(exampleTask, 2, 2*time.Second, 1)
	pipeline.AddLayer(exampleTask2, 5, 1*time.Second, 3)

	// 输入数据
	inputData := make([]interface{}, 20)
	for i := 0; i < 20; i++ {
		inputData[i] = i
	}

	// 运行流水线
	results, err := pipeline.Run(inputData)
	if err != nil {
		fmt.Printf("Pipeline error: %v\n", err)
		// 即使有错误，您仍然可以选择打印部分或全部结果
	}

	// 输出结果
	for i, res := range results {
		fmt.Printf("Task %d result: %v\n", i, res)
	}
}