| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| package com.datatorrent.stram.plan.physical; |
| |
| import java.util.Collection; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| import com.google.common.math.IntMath; |
| |
| import com.datatorrent.api.Context; |
| import com.datatorrent.api.Context.PortContext; |
| import com.datatorrent.api.Operator; |
| import com.datatorrent.api.Partitioner.PartitionKeys; |
| import com.datatorrent.api.StreamCodec; |
| |
| import com.datatorrent.common.util.Pair; |
| import com.datatorrent.stram.plan.logical.LogicalPlan; |
| import com.datatorrent.stram.plan.logical.LogicalPlan.InputPortMeta; |
| import com.datatorrent.stram.plan.logical.LogicalPlan.OperatorMeta; |
| import com.datatorrent.stram.plan.logical.LogicalPlan.StreamMeta; |
| import com.datatorrent.stram.plan.logical.Operators; |
| import com.datatorrent.stram.plan.logical.Operators.PortMappingDescriptor; |
| import com.datatorrent.stram.plan.physical.PTOperator.PTInput; |
| import com.datatorrent.stram.plan.physical.PTOperator.PTOutput; |
| |
| /** |
| * Encapsulates the mapping of input to output operators, including unifiers. Depending on logical plan setting and |
| * number of partitions, unifiers are created as needed and potentially cascaded. |
| * |
| * @since 0.9.0 |
| */ |
| public class StreamMapping implements java.io.Serializable |
| { |
| private static final long serialVersionUID = 8572852828117485193L; |
| |
| private final static Logger LOG = LoggerFactory.getLogger(StreamMapping.class); |
| |
| private final StreamMeta streamMeta; |
| private final PhysicalPlan plan; |
| PTOperator finalUnifier; |
| final Set<PTOperator> cascadingUnifiers = Sets.newHashSet(); |
| final Set<PTOperator> slidingUnifiers = Sets.newHashSet(); |
| private final List<PTOutput> upstream = Lists.newArrayList(); |
| |
| |
| public StreamMapping(StreamMeta streamMeta, PhysicalPlan plan) { |
| this.streamMeta = streamMeta; |
| this.plan = plan; |
| } |
| |
| void addTo(Collection<PTOperator> opers) { |
| if (finalUnifier != null) { |
| opers.add(finalUnifier); |
| } |
| opers.addAll(cascadingUnifiers); |
| opers.addAll(slidingUnifiers); |
| } |
| |
| public void setSources(Collection<PTOperator> partitions) { |
| upstream.clear(); |
| // add existing inputs |
| for (PTOperator uoper : partitions) { |
| for (PTOutput source : uoper.outputs) { |
| if (source.logicalStream == streamMeta) { |
| upstream.add(source); |
| } |
| } |
| } |
| redoMapping(); |
| } |
| |
| public static PTOperator createSlidingUnifier(StreamMeta streamMeta, PhysicalPlan plan, int operatorApplicationWindowCount, int slidingWindowCount) |
| { |
| int gcd = IntMath.gcd(operatorApplicationWindowCount, slidingWindowCount); |
| OperatorMeta um = streamMeta.getSource().getSlidingUnifier(operatorApplicationWindowCount / gcd, gcd, slidingWindowCount / gcd); |
| PTOperator pu = plan.newOperator(um, um.getName()); |
| |
| Operator unifier = um.getOperator(); |
| PortMappingDescriptor mergeDesc = new PortMappingDescriptor(); |
| Operators.describe(unifier, mergeDesc); |
| if (mergeDesc.outputPorts.size() != 1) { |
| throw new AssertionError("Unifier must have a single output port, instead found : " + mergeDesc.outputPorts); |
| } |
| pu.unifiedOperatorMeta = streamMeta.getSource().getOperatorMeta(); |
| pu.outputs.add(new PTOutput(mergeDesc.outputPorts.keySet().iterator().next(), streamMeta, pu)); |
| plan.newOpers.put(pu, unifier); |
| return pu; |
| } |
| |
| public static PTOperator createUnifier(StreamMeta streamMeta, PhysicalPlan plan) |
| { |
| OperatorMeta um = streamMeta.getSource().getUnifierMeta(); |
| PTOperator pu = plan.newOperator(um, um.getName()); |
| |
| Operator unifier = um.getOperator(); |
| PortMappingDescriptor mergeDesc = new PortMappingDescriptor(); |
| Operators.describe(unifier, mergeDesc); |
| if (mergeDesc.outputPorts.size() != 1) { |
| throw new AssertionError("Unifier must have a single output port, instead found : " + mergeDesc.outputPorts); |
| } |
| |
| pu.unifiedOperatorMeta = streamMeta.getSource().getOperatorMeta(); |
| pu.outputs.add(new PTOutput(mergeDesc.outputPorts.keySet().iterator().next(), streamMeta, pu)); |
| plan.newOpers.put(pu, unifier); |
| return pu; |
| } |
| |
| private void addSlidingUnifiers() |
| { |
| OperatorMeta sourceOM = streamMeta.getSource().getOperatorMeta(); |
| if (sourceOM.getAttributes().contains(Context.OperatorContext.SLIDE_BY_WINDOW_COUNT)) { |
| if (sourceOM.getValue(Context.OperatorContext.SLIDE_BY_WINDOW_COUNT) < |
| sourceOM.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT)) { |
| plan.undeployOpers.addAll(slidingUnifiers); |
| slidingUnifiers.clear(); |
| List<PTOutput> newUpstream = Lists.newArrayList(); |
| PTOperator slidingUnifier; |
| for (PTOutput source : upstream) { |
| slidingUnifier = StreamMapping.createSlidingUnifier(streamMeta, plan, |
| sourceOM.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT), |
| sourceOM.getValue(Context.OperatorContext.SLIDE_BY_WINDOW_COUNT)); |
| addInput(slidingUnifier, source, null); |
| this.slidingUnifiers.add(slidingUnifier); |
| newUpstream.add(slidingUnifier.outputs.get(0)); |
| } |
| upstream.clear(); |
| upstream.addAll(newUpstream); |
| } |
| else { |
| LOG.warn("Sliding Window Count {} should be less than APPLICATION WINDOW COUNT {}", sourceOM.getValue(Context.OperatorContext.SLIDE_BY_WINDOW_COUNT), sourceOM.getValue(Context.OperatorContext.APPLICATION_WINDOW_COUNT)); |
| } |
| } |
| } |
| |
| @SuppressWarnings("AssignmentToForLoopParameter") |
| private List<PTOutput> setupCascadingUnifiers(List<PTOutput> upstream, List<PTOperator> pooledUnifiers, int limit, int level) { |
| List<PTOutput> nextLevel = Lists.newArrayList(); |
| PTOperator pu = null; |
| for (int i=0; i<upstream.size(); i++) { |
| if (i % limit == 0) { |
| if (upstream.size() - i < limit) { |
| while(i< upstream.size()) { |
| nextLevel.add(upstream.get(i)); |
| i++; |
| } |
| continue; |
| } |
| if (!pooledUnifiers.isEmpty()) { |
| pu = pooledUnifiers.remove(0); |
| } else { |
| pu = createUnifier(streamMeta, plan); |
| } |
| assert (pu.outputs.size() == 1) : "unifier has single output"; |
| nextLevel.addAll(pu.outputs); |
| this.cascadingUnifiers.add(pu); |
| } |
| |
| PTOutput source = upstream.get(i); |
| addInput(pu, source, null); |
| } |
| |
| if (nextLevel.size() > limit) { |
| return setupCascadingUnifiers(nextLevel, pooledUnifiers, limit, level); |
| } else { |
| return nextLevel; |
| } |
| } |
| |
| /** |
| * rebuild the tree, which may cause more changes to execution layer than need be |
| * TODO: investigate incremental logic |
| */ |
| private void redoMapping() { |
| |
| Set<Pair<PTOperator, InputPortMeta>> downstreamOpers = Sets.newHashSet(); |
| |
| // figure out the downstream consumers |
| for (InputPortMeta ipm : streamMeta.getSinks()) { |
| // gets called prior to all logical operators mapped |
| // skipped for parallel partitions - those are handled elsewhere |
| if (!ipm.getValue(PortContext.PARTITION_PARALLEL) && plan.hasMapping(ipm.getOperatorWrapper())) { |
| List<PTOperator> partitions = plan.getOperators(ipm.getOperatorWrapper()); |
| for (PTOperator doper : partitions) { |
| downstreamOpers.add(new Pair<PTOperator, InputPortMeta>(doper, ipm)); |
| } |
| } |
| } |
| |
| if (!downstreamOpers.isEmpty()) { |
| // unifiers are required |
| for (PTOperator unifier : this.cascadingUnifiers) { |
| detachUnifier(unifier); |
| } |
| if (this.finalUnifier != null) { |
| detachUnifier(finalUnifier); |
| } |
| |
| List<PTOperator> currentUnifiers = Lists.newArrayList(this.cascadingUnifiers); |
| this.cascadingUnifiers.clear(); |
| plan.undeployOpers.addAll(currentUnifiers); |
| addSlidingUnifiers(); |
| |
| int limit = streamMeta.getSource().getValue(PortContext.UNIFIER_LIMIT); |
| |
| boolean separateUnifiers = false; |
| Integer lastId = null; |
| for (InputPortMeta ipm : streamMeta.getSinks()) { |
| StreamCodec<?> streamCodecInfo = ipm.getStreamCodec(); |
| Integer id = plan.getStreamCodecIdentifier(streamCodecInfo); |
| if (lastId == null) { |
| lastId = id; |
| } else if (!id.equals(lastId)) { |
| separateUnifiers = true; |
| break; |
| } |
| } |
| |
| List<PTOutput> unifierSources = this.upstream; |
| Map<StreamCodec<?>, List<PTOutput>> cascadeUnifierSourcesMap = Maps.newHashMap(); |
| |
| if (limit > 1 && this.upstream.size() > limit) { |
| // cascading unifier |
| if (!separateUnifiers) { |
| unifierSources = setupCascadingUnifiers(this.upstream, currentUnifiers, limit, 0); |
| } else { |
| for (InputPortMeta ipm : streamMeta.getSinks()) { |
| StreamCodec<?> streamCodecInfo = ipm.getStreamCodec(); |
| if (!cascadeUnifierSourcesMap.containsKey(streamCodecInfo)) { |
| unifierSources = setupCascadingUnifiers(this.upstream, currentUnifiers, limit, 0); |
| cascadeUnifierSourcesMap.put(streamCodecInfo, unifierSources); |
| } |
| } |
| } |
| } |
| |
| // remove remaining unifiers |
| for (PTOperator oper : currentUnifiers) { |
| plan.removePTOperator(oper); |
| } |
| |
| // Directly getting attribute from map to know if it is set or not as it can be overriden by the input |
| Boolean sourceSingleFinal = streamMeta.getSource().getAttributes().get(PortContext.UNIFIER_SINGLE_FINAL); |
| |
| // link the downstream operators with the unifiers |
| for (Pair<PTOperator, InputPortMeta> doperEntry : downstreamOpers) { |
| |
| Map<LogicalPlan.InputPortMeta, PartitionKeys> partKeys = doperEntry.first.partitionKeys; |
| PartitionKeys pks = partKeys != null ? partKeys.get(doperEntry.second) : null; |
| Boolean sinkSingleFinal = doperEntry.second.getAttributes().get(PortContext.UNIFIER_SINGLE_FINAL); |
| boolean lastSingle = (sinkSingleFinal != null) ? sinkSingleFinal : |
| (sourceSingleFinal != null ? sourceSingleFinal.booleanValue() : PortContext.UNIFIER_SINGLE_FINAL.defaultValue); |
| |
| if (upstream.size() > 1) { |
| if (!separateUnifiers && ((pks == null || pks.mask == 0) || lastSingle)) { |
| if (finalUnifier == null) { |
| finalUnifier = createUnifier(streamMeta, plan); |
| } |
| setInput(doperEntry.first, doperEntry.second, finalUnifier, (pks == null) || (pks.mask == 0) ? null : pks); |
| if (finalUnifier.inputs.isEmpty()) { |
| // set unifier inputs once, regardless how many downstream operators there are |
| for (PTOutput out : unifierSources) { |
| addInput(this.finalUnifier, out, null); |
| } |
| } |
| } else { |
| // MxN partitioning: unifier per downstream partition |
| LOG.debug("MxN unifier for {} {} {}", new Object[] {doperEntry.first, doperEntry.second.getPortName(), pks}); |
| PTOperator unifier = doperEntry.first.upstreamMerge.get(doperEntry.second); |
| if (unifier == null) { |
| unifier = createUnifier(streamMeta, plan); |
| doperEntry.first.upstreamMerge.put(doperEntry.second, unifier); |
| setInput(doperEntry.first, doperEntry.second, unifier, null); |
| } |
| // sources may change dynamically, rebuild inputs (as for cascading unifiers) |
| for (PTInput in : unifier.inputs) { |
| in.source.sinks.remove(in); |
| } |
| unifier.inputs.clear(); |
| List<PTOutput> doperUnifierSources = unifierSources; |
| if (separateUnifiers) { |
| StreamCodec<?> streamCodecInfo = doperEntry.second.getStreamCodec(); |
| List<PTOutput> cascadeSources = cascadeUnifierSourcesMap.get(streamCodecInfo); |
| if (cascadeSources != null) { |
| doperUnifierSources = cascadeSources; |
| } |
| } |
| // add new inputs |
| for (PTOutput out : doperUnifierSources) { |
| addInput(unifier, out, pks); |
| } |
| } |
| } else { |
| // no partitioning |
| PTOperator unifier = doperEntry.first.upstreamMerge.remove(doperEntry.second); |
| if (unifier != null) { |
| plan.removePTOperator(unifier); |
| } |
| setInput(doperEntry.first, doperEntry.second, upstream.get(0).source, pks); |
| } |
| } |
| |
| // Remove the unattached final unifier |
| // Unattached final unifier is from |
| // 1) Upstream operator partitions are scaled down to one. (no unifier needed) |
| // 2) Downstream operators partitions are scaled up from one to multiple. (replaced by merged unifier) |
| if (finalUnifier != null && finalUnifier.inputs.isEmpty()) { |
| plan.removePTOperator(finalUnifier); |
| finalUnifier = null; |
| } |
| |
| } |
| |
| } |
| |
| private void setInput(PTOperator oper, InputPortMeta ipm, PTOperator sourceOper, PartitionKeys pks) { |
| // TODO: see if this can be handled more efficiently |
| for (PTInput in : oper.inputs) { |
| if (in.source.source == sourceOper && in.logicalStream == streamMeta && ipm.getPortName().equals(in.portName)) { |
| return; |
| } |
| } |
| // link to upstream output(s) for this stream |
| for (PTOutput upstreamOut : sourceOper.outputs) { |
| if (upstreamOut.logicalStream == streamMeta) { |
| PTInput input = new PTInput(ipm.getPortName(), streamMeta, oper, pks, upstreamOut, ipm.getValue(LogicalPlan.IS_CONNECTED_TO_DELAY_OPERATOR)); |
| oper.inputs.add(input); |
| } |
| } |
| } |
| |
| public static void addInput(PTOperator target, PTOutput upstreamOut, PartitionKeys pks) |
| { |
| StreamMeta lStreamMeta = upstreamOut.logicalStream; |
| PTInput input = new PTInput("<merge#" + lStreamMeta.getSource().getPortName() + ">", lStreamMeta, target, pks, upstreamOut, false); |
| target.inputs.add(input); |
| } |
| |
| private void detachUnifier(PTOperator unifier) { |
| // remove existing unifiers from downstream inputs |
| for (PTOutput out : unifier.outputs) { |
| for (PTInput input : out.sinks) { |
| input.target.inputs.remove(input); |
| } |
| out.sinks.clear(); |
| } |
| // remove from upstream outputs |
| for (PTInput in : unifier.inputs) { |
| in.source.sinks.remove(in); |
| } |
| unifier.inputs.clear(); |
| } |
| |
| } |
