19两套执行引擎：Pregel vs DAG

func (r *runner) run(ctx context.Context, isStream bool, input any, opts ...Option) (result any, err error) {
  haveOnStart := false // delay triggering onGraphStart until state initialization is complete, so that the state can be accessed within onGraphStart.
  defer func() {
    if !haveOnStart {
      ctx, input = onGraphStart(ctx, input, isStream)
    }
    if err != nil {
      ctx, err = onGraphError(ctx, err)
    } else {
      ctx, result = onGraphEnd(ctx, result, isStream)
    }
  }()

  var runWrapper runnableCallWrapper
  runWrapper = runnableInvoke
  if isStream {
    runWrapper = runnableTransform
  }

  // Initialize channel and task managers.
  cm := r.initChannelManager(isStream)
  tm := r.initTaskManager(runWrapper, getGraphCancel(ctx), opts...)
  maxSteps := r.options.maxRunSteps

  maxSteps, err = r.resolveMaxSteps(maxSteps, opts)
  if err != nil {
    return nil, err
  }

  // Extract and validate options for each node.
  optMap, extractErr := extractOption(r.chanSubscribeTo, opts...)
  if extractErr != nil {
    return nil, newGraphRunError(fmt.Errorf("graph extract option fail: %w", extractErr))
  }

  // Extract CheckPointID
  checkPointID, writeToCheckPointID, stateModifier, forceNewRun := getCheckPointInfo(opts...)
  if checkPointID != nil && r.checkPointer.store == nil {
    return nil, newGraphRunError(fmt.Errorf("receive checkpoint id but have not set checkpoint store"))
  }

  // Extract subgraph
  path, isSubGraph := getNodePath(ctx)

  // load checkpoint from ctx/store or init graph
  initialized := false
  var nextTasks []*task
  if cp := getCheckPointFromCtx(ctx); cp != nil {
// … 省略 204 行；完整声明 L109–360，点击上方「浏览完整文件」

type channel interface {
    reportValues(map[string]any) error       // 上游送来一个值
    reportDependencies([]string)             // 上游"我完成了"
    reportSkip([]string) bool                // 上游"我这条路不走了"
    get(bool, string, *edgeHandlerManager) (any, bool, error)  // 我这一轮就绪了吗?
    // ...
}

type pregelChannel struct {
    Values      map[string]any   // 上游送来的值,攒在这
    mergeConfig FanInMergeConfig
}

func (ch *pregelChannel) get(isStream bool, name string, edgeHandler *edgeHandlerManager) (
  any, bool, error) {
  if len(ch.Values) == 0 {
    return nil, false, nil
  }
  defer func() { ch.Values = map[string]any{} }()
  values := make([]any, len(ch.Values))
  names := make([]string, len(ch.Values))
  i := 0
  for k, v := range ch.Values {
    resolvedV, err := edgeHandler.handle(k, name, v, isStream)
    if err != nil {
      return nil, false, err
    }
    values[i] = resolvedV
    names[i] = k
    i++
  }

  if len(values) == 1 {
    return values[0], true, nil
  }

  // merge
  mergeOpts := &mergeOptions{
    streamMergeWithSourceEOF: ch.mergeConfig.StreamMergeWithSourceEOF,
    names:                    names,
  }
  v, err := mergeValues(values, mergeOpts)
  if err != nil {
    return nil, false, err
  }
  return v, true, nil
}

const (
  // AnyPredecessor means that the node will be triggered when any of its predecessors is included in the previous completed super step.
  // Ref:https://www.cloudwego.io/docs/eino/core_modules/chain_and_graph_orchestration/orchestration_design_principles/#runtime-engine
  AnyPredecessor NodeTriggerMode = "any_predecessor"
  // AllPredecessor means that the current node will only be triggered when all of its predecessor nodes have finished running.
  AllPredecessor NodeTriggerMode = "all_predecessor"
)

func (ch *pregelChannel) reportSkip(_ []string) bool {
  return false
}

// END is the end node of the graph. You can add your last edge with END.
const END = "end"

// graphRunType is a custom type used to control the running mode of the graph.
type graphRunType string

const (
  // runTypePregel is a running mode of the graph that is suitable for large-scale graph processing tasks. Can have cycles in graph. Compatible with NodeTriggerType.AnyPredecessor.
  runTypePregel graphRunType = "Pregel"
  // runTypeDAG is a running mode of the graph that represents the graph as a directed acyclic graph, suitable for tasks that can be represented as a directed acyclic graph. Compatible with NodeTriggerType.AllPredecessor.
  runTypeDAG graphRunType = "DAG"
)

// String returns the string representation of the graph run type.
func (g graphRunType) String() string {
  return string(g)
}

type graph struct {
  nodes        map[string]*graphNode
  controlEdges map[string][]string
  dataEdges    map[string][]string
  branches     map[string][]*GraphBranch
  startNodes   []string
// … 这是 L38–62 的片段（该行不在任何顶层声明内）；点击上方「浏览完整文件」

type dagChannel struct {
    ControlPredecessors map[string]dependencyState  // 每个控制前驱:等待/就绪/跳过
    Values              map[string]any
    DataPredecessors    map[string]bool             // 每个数据前驱:是否到货
    Skipped             bool
    // ...
}

for _, state := range ch.ControlPredecessors {
    if state == dependencyStateWaiting {
        return nil, false, nil    // 还有控制前驱在等 → 不就绪
    }
}
for _, ready := range ch.DataPredecessors {
    if !ready {
        return nil, false, nil    // 还有数据前驱没到 → 不就绪
    }
}

func (ch *dagChannel) reportSkip(keys []string) bool {
  for _, k := range keys {
    if _, ok := ch.ControlPredecessors[k]; ok {
      ch.ControlPredecessors[k] = dependencyStateSkipped
    }
    if _, ok := ch.DataPredecessors[k]; ok {
      ch.DataPredecessors[k] = true
    }
  }

  allSkipped := true
  for _, state := range ch.ControlPredecessors {
    if state != dependencyStateSkipped {
      allSkipped = false
      break
    }
  }
  ch.Skipped = allSkipped

  return allSkipped
}

func (c *channelManager) reportBranch(from string, skippedNodes []string) error {
  var nKeys []string
  for _, node := range skippedNodes {
    skipped := c.channels[node].reportSkip([]string{from})
    if skipped {
      nKeys = append(nKeys, node)
    }
  }

  for i := 0; i < len(nKeys); i++ {
    key := nKeys[i]

    if key == END {
      continue
    }
    if _, ok := c.successors[key]; !ok {
      return fmt.Errorf("unknown node: %s", key)
    }
    for _, successor := range c.successors[key] {
      skipped := c.channels[successor].reportSkip([]string{key})
      if skipped {
        nKeys = appendIfNotExist(nKeys, successor)
      }
      // todo: detect if end node has been skipped?
    }
  }
  return nil
}

func (g *graph) compile(ctx context.Context, opt *graphCompileOptions) (*composableRunnable, error) {
  if g.buildError != nil {
    return nil, g.buildError
  }

  // get run type
  runType := runTypePregel
  cb := pregelChannelBuilder
  if isChain(g.cmp) || isWorkflow(g.cmp) {
    if opt != nil && opt.nodeTriggerMode != "" {
      return nil, errors.New(fmt.Sprintf("%s doesn't support node trigger mode option", g.cmp))
    }
  }
  if (opt != nil && opt.nodeTriggerMode == AllPredecessor) || isWorkflow(g.cmp) {
    runType = runTypeDAG
    cb = dagChannelBuilder
  }

  // get eager type
  eager := false
  if isWorkflow(g.cmp) || runType == runTypeDAG {
    eager = true
  }
  if opt != nil && opt.eagerDisabled {
    eager = false
  }

  if len(g.startNodes) == 0 {
    return nil, errors.New("start node not set")
  }
  if len(g.endNodes) == 0 {
    return nil, errors.New("end node not set")
  }

  // toValidateMap isn't empty means there are nodes that cannot infer type
  for _, v := range g.toValidateMap {
    if len(v) > 0 {
      return nil, fmt.Errorf("some node's input or output types cannot be inferred: %v", g.toValidateMap)
    }
  }

  for key := range g.fieldMappingRecords {
    // not allowed to map multiple fields to the same field
    toMap := make(map[string]bool)
    for _, mapping := range g.fieldMappingRecords[key] {
      if _, ok := toMap[mapping.to]; ok {
        return nil, fmt.Errorf("duplicate mapping target field: %s of node[%s]", mapping.to, key)
      }
// … 省略 171 行；完整声明 L674–892，点击上方「浏览完整文件」

func (r *runner) initChannelManager(isStream bool) *channelManager {
  builder := r.chanBuilder
  if builder == nil {
    builder = pregelChannelBuilder
  }

  chs := make(map[string]channel)
  for ch := range r.chanSubscribeTo {
    chs[ch] = builder(r.controlPredecessors[ch], r.dataPredecessors[ch], r.chanSubscribeTo[ch].action.inputZeroValue, r.chanSubscribeTo[ch].action.inputEmptyStream)
  }

  chs[END] = builder(r.controlPredecessors[END], r.dataPredecessors[END], r.outputZeroValue, r.outputEmptyStream)

  dataPredecessors := make(map[string]map[string]struct{})
  for k, vs := range r.dataPredecessors {
    dataPredecessors[k] = make(map[string]struct{})
    for _, v := range vs {
      dataPredecessors[k][v] = struct{}{}
    }
  }
  controlPredecessors := make(map[string]map[string]struct{})
  for k, vs := range r.controlPredecessors {
    controlPredecessors[k] = make(map[string]struct{})
    for _, v := range vs {
      controlPredecessors[k][v] = struct{}{}
    }
  }

  for k, v := range chs {
    if cfg, ok := r.mergeConfigs[k]; ok {
      v.setMergeConfig(cfg)
    }
  }

  return &channelManager{
    isStream:            isStream,
    channels:            chs,
    successors:          r.successors,
    dataPredecessors:    dataPredecessors,
    controlPredecessors: controlPredecessors,

    edgeHandlerManager:    r.edgeHandlerManager,
    preNodeHandlerManager: r.preNodeHandlerManager,
  }
}