Parallelize tries to parallelize the DAG by splitting each source path as much as possible of the sequence into n parallel branches.
(main *dag.Main, concurrency int)
| 8 | // Parallelize tries to parallelize the DAG by splitting each source |
| 9 | // path as much as possible of the sequence into n parallel branches. |
| 10 | func (o *Optimizer) Parallelize(main *dag.Main, concurrency int) error { |
| 11 | // Compute the number of parallel paths across all input sources to |
| 12 | // achieve the desired level of concurrency. At some point, we should |
| 13 | // use a semaphore here and let each possible path use the max concurrency. |
| 14 | if o.nent == 0 { |
| 15 | return nil |
| 16 | } |
| 17 | concurrency = max(concurrency/o.nent, 2) |
| 18 | seq, err := walkEntries(main.Body, func(seq dag.Seq) (dag.Seq, error) { |
| 19 | if len(seq) == 0 { |
| 20 | return seq, nil |
| 21 | } |
| 22 | var front, parallel dag.Seq |
| 23 | var err error |
| 24 | if lister, slicer, rest := matchSource(seq); lister != nil { |
| 25 | // We parallelize the scanning to achieve the desired concurrency, |
| 26 | // then the step below pulls downstream operators into the parallel |
| 27 | // branches when possible, e.g., to parallelize aggregations etc. |
| 28 | front.Append(lister) |
| 29 | if slicer != nil { |
| 30 | front.Append(slicer) |
| 31 | } |
| 32 | parallel, err = o.parallelizeSeqScan(rest, concurrency) |
| 33 | } else if scan, ok := seq[0].(*dag.FileScan); ok { |
| 34 | if !o.env.UseVAM() { |
| 35 | // Sequence runtime file scan doesn't support parallelism. |
| 36 | return seq, nil |
| 37 | } |
| 38 | front.Append(scan) |
| 39 | parallel, err = o.parallelizeFileScan(seq[1:], concurrency) |
| 40 | } |
| 41 | if err != nil { |
| 42 | return nil, err |
| 43 | } |
| 44 | if parallel == nil { |
| 45 | // Leave the source path unmodified. |
| 46 | return seq, nil |
| 47 | } |
| 48 | // Replace the source path with the parallelized gadget. |
| 49 | return append(front, parallel...), nil |
| 50 | }) |
| 51 | if err != nil { |
| 52 | return err |
| 53 | } |
| 54 | o.optimizeParallels(seq) |
| 55 | main.Body = removePassOps(seq) |
| 56 | return nil |
| 57 | } |
| 58 | |
| 59 | func matchSource(seq dag.Seq) (*dag.ListerScan, *dag.SlicerOp, dag.Seq) { |
| 60 | lister, ok := seq[0].(*dag.ListerScan) |
no test coverage detected