liftIntoParPaths examines seq to see if there's an opportunity to lift operations downstream from a parallel op into its parallel paths to enhance concurrency. If so, we modify the sequential ops in place.
(seq dag.Seq)
| 150 | // lift operations downstream from a parallel op into its parallel paths to |
| 151 | // enhance concurrency. If so, we modify the sequential ops in place. |
| 152 | func (o *Optimizer) liftIntoParPaths(seq dag.Seq) { |
| 153 | if len(seq) < 2 { |
| 154 | // Need a parallel, an optional merge/combine, and something downstream. |
| 155 | return |
| 156 | } |
| 157 | paths, ok := parallelPaths(seq[0]) |
| 158 | if !ok { |
| 159 | return |
| 160 | } |
| 161 | egress := 1 |
| 162 | var merge *dag.MergeOp |
| 163 | switch op := seq[1].(type) { |
| 164 | case *dag.MergeOp: |
| 165 | merge = op |
| 166 | egress = 2 |
| 167 | case *dag.CombineOp: |
| 168 | egress = 2 |
| 169 | } |
| 170 | if egress >= len(seq) { |
| 171 | return |
| 172 | } |
| 173 | switch op := seq[egress].(type) { |
| 174 | case *dag.AggregateOp: |
| 175 | // To decompose the aggregate, we split the flowgraph into |
| 176 | // branches that run up to and including an aggregate, |
| 177 | // followed by a post-merge aggregate that composes the results. |
| 178 | // Copy the aggregator into the tail of the trunk and arrange |
| 179 | // for partials to flow between them. |
| 180 | if op.PartialsIn || op.PartialsOut { |
| 181 | // Need an unmodified aggregate to split into its parials pieces. |
| 182 | return |
| 183 | } |
| 184 | for k := range paths { |
| 185 | partial := dag.CopyOp(op).(*dag.AggregateOp) |
| 186 | partial.PartialsOut = true |
| 187 | paths[k].Append(partial) |
| 188 | } |
| 189 | op.PartialsIn = true |
| 190 | // The upstream aggregators will compute any key expressions |
| 191 | // so the ingress aggregator should simply reference the key |
| 192 | // by its name. This loop updates the ingress to do so. |
| 193 | for k := range op.Keys { |
| 194 | op.Keys[k].RHS = op.Keys[k].LHS |
| 195 | } |
| 196 | case *dag.SortOp: |
| 197 | if len(op.Exprs) == 0 { |
| 198 | return |
| 199 | } |
| 200 | seq[1] = &dag.MergeOp{Kind: "MergeOp", Exprs: op.Exprs} |
| 201 | if egress > 1 { |
| 202 | seq[2] = dag.Pass |
| 203 | } |
| 204 | for k := range paths { |
| 205 | paths[k].Append(dag.CopyOp(op)) |
| 206 | } |
| 207 | case *dag.TopOp: |
| 208 | if len(op.Exprs) == 0 { |
| 209 | return |
no test coverage detected