/*
 * 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 org.apache.tinkerpop.gremlin.process.computer.traversal;

import org.apache.commons.configuration2.Configuration;
import org.apache.tinkerpop.gremlin.process.computer.Computer;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MemoryComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
import org.apache.tinkerpop.gremlin.process.computer.MessageScope;
import org.apache.tinkerpop.gremlin.process.computer.Messenger;
import org.apache.tinkerpop.gremlin.process.computer.ProgramPhase;
import org.apache.tinkerpop.gremlin.process.computer.VertexComputeKey;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.ComputerResultStep;
import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep;
import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy;
import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.finalization.ComputerFinalizationStrategy;
import org.apache.tinkerpop.gremlin.process.computer.util.AbstractVertexProgramBuilder;
import org.apache.tinkerpop.gremlin.process.computer.util.SingleMessenger;
import org.apache.tinkerpop.gremlin.process.computer.util.VertexProgramHelper;
import org.apache.tinkerpop.gremlin.process.traversal.Operator;
import org.apache.tinkerpop.gremlin.process.traversal.Step;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
import org.apache.tinkerpop.gremlin.process.traversal.step.LocalBarrier;
import org.apache.tinkerpop.gremlin.process.traversal.step.MapReducer;
import org.apache.tinkerpop.gremlin.process.traversal.step.MemoryComputing;
import org.apache.tinkerpop.gremlin.process.traversal.step.map.GraphStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.ProfileSideEffectStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.ProfileStep;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.HaltedTraverserStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.ComputerVerificationStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.IndexedTraverserSet;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
import org.apache.tinkerpop.gremlin.process.traversal.util.DefaultTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.MutableMetrics;
import org.apache.tinkerpop.gremlin.process.traversal.util.PureTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.ScriptTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMetrics;
import org.apache.tinkerpop.gremlin.structure.Direction;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Element;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.util.function.MutableMetricsSupplier;
import org.apache.tinkerpop.gremlin.util.iterator.EmptyIterator;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.Set;

/**
 * {@code TraversalVertexProgram} enables the evaluation of a {@link Traversal} on a {@link GraphComputer}.
 * At the start of the computation, each {@link Vertex} (or {@link Edge}) is assigned a single {@link Traverser}.
 * For each traverser that is local to the vertex, the vertex looks up its current location in the traversal and
 * processes that step. If the outputted traverser of the step references a local structure on the vertex (e.g. the
 * vertex, an incident edge, its properties, or an arbitrary object), then the vertex continues to compute the next
 * traverser. If the traverser references another location in the graph, then the traverser is sent to that location
 * in the graph via a message. The messages of TraversalVertexProgram are traversers. This continues until all
 * traversers in the computation have halted.
 *
 * @author Marko A. Rodriguez (http://markorodriguez.com)
 */
public final class TraversalVertexProgram implements VertexProgram<TraverserSet<Object>> {

    public static final String TRAVERSAL = "gremlin.traversalVertexProgram.traversal";
    public static final String HALTED_TRAVERSERS = "gremlin.traversalVertexProgram.haltedTraversers";
    public static final String ACTIVE_TRAVERSERS = "gremlin.traversalVertexProgram.activeTraversers";
    protected static final String MUTATED_MEMORY_KEYS = "gremlin.traversalVertexProgram.mutatedMemoryKeys";
    private static final String VOTE_TO_HALT = "gremlin.traversalVertexProgram.voteToHalt";
    private static final String COMPLETED_BARRIERS = "gremlin.traversalVertexProgram.completedBarriers";

    // TODO: if not an adjacent traversal, use Local message scope -- a dual messaging system.
    private static final Set<MessageScope> MESSAGE_SCOPES = new HashSet<>(Collections.singletonList(MessageScope.Global.instance()));
    private Set<MemoryComputeKey> memoryComputeKeys = new HashSet<>();
    private static final Set<VertexComputeKey> VERTEX_COMPUTE_KEYS =
            new HashSet<>(Arrays.asList(VertexComputeKey.of(HALTED_TRAVERSERS, false), VertexComputeKey.of(ACTIVE_TRAVERSERS, true)));

    private PureTraversal<?, ?> traversal;
    private TraversalMatrix<?, ?> traversalMatrix;
    private final Set<MapReduce> mapReducers = new HashSet<>();
    private TraverserSet<Object> haltedTraversers;
    private boolean returnHaltedTraversers = false;
    private HaltedTraverserStrategy haltedTraverserStrategy;
    private boolean profile = false;
    // handle current profile metrics if profile is true
    private MutableMetrics iterationMetrics;


    private TraversalVertexProgram() {
    }

    /**
     * Get the {@link PureTraversal} associated with the current instance of the {@link TraversalVertexProgram}.
     *
     * @return the pure traversal of the instantiated program
     */
    public PureTraversal<?, ?> getTraversal() {
        return this.traversal;
    }

    public static <R> TraverserSet<R> loadHaltedTraversers(final Configuration configuration) {
        if (!configuration.containsKey(HALTED_TRAVERSERS))
            return new TraverserSet<>();

        final Object object = configuration.getProperty(HALTED_TRAVERSERS) instanceof String ?
                VertexProgramHelper.deserialize(configuration, HALTED_TRAVERSERS) :
                configuration.getProperty(HALTED_TRAVERSERS);
        if (object instanceof Traverser.Admin)
            return new TraverserSet<>((Traverser.Admin<R>) object);
        else {
            final TraverserSet<R> traverserSet = new TraverserSet<>();
            traverserSet.addAll((Collection) object);
            return traverserSet;
        }
    }

    public static <R> void storeHaltedTraversers(final Configuration configuration, final TraverserSet<R> haltedTraversers) {
        if (null != haltedTraversers && !haltedTraversers.isEmpty()) {
            try {
                VertexProgramHelper.serialize(haltedTraversers, configuration, HALTED_TRAVERSERS);
            } catch (final Exception e) {
                configuration.setProperty(HALTED_TRAVERSERS, haltedTraversers);
            }
        }
    }

    @Override
    public void loadState(final Graph graph, final Configuration configuration) {
        if (!configuration.containsKey(TRAVERSAL))
            throw new IllegalArgumentException("The configuration does not have a traversal: " + TRAVERSAL);
        this.traversal = PureTraversal.loadState(configuration, TRAVERSAL, graph);
        if (!this.traversal.get().isLocked())
            this.traversal.get().applyStrategies();
        /// traversal is compiled and ready to be introspected
        this.traversalMatrix = new TraversalMatrix<>(this.traversal.get());
        // get any master-traversal halted traversers
        this.haltedTraversers = TraversalVertexProgram.loadHaltedTraversers(configuration);
        // if results will be serialized out, don't save halted traversers across the cluster
        this.returnHaltedTraversers =
                (this.traversal.get().getParent().asStep().getNextStep() instanceof ComputerResultStep || // if its just going to stream it out, don't distribute
                        this.traversal.get().getParent().asStep().getNextStep() instanceof EmptyStep ||  // same as above, but if using TraversalVertexProgramStep directly
                        (this.traversal.get().getParent().asStep().getNextStep() instanceof ProfileStep && // same as above, but needed for profiling
                                this.traversal.get().getParent().asStep().getNextStep().getNextStep() instanceof ComputerResultStep));

        // determine how to store halted traversers
        final Iterator<?> itty = IteratorUtils.filter(this.traversal.get().getStrategies(), strategy -> strategy instanceof HaltedTraverserStrategy).iterator();
        this.haltedTraverserStrategy = itty.hasNext() ? (HaltedTraverserStrategy) itty.next() : HaltedTraverserStrategy.reference();

        // register traversal side-effects in memory
        this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
        // register MapReducer memory compute keys
        for (final MapReducer<?, ?, ?, ?, ?> mapReducer : TraversalHelper.getStepsOfAssignableClassRecursively(MapReducer.class, this.traversal.get())) {
            this.mapReducers.add(mapReducer.getMapReduce());
            this.memoryComputeKeys.add(MemoryComputeKey.of(mapReducer.getMapReduce().getMemoryKey(), Operator.assign, false, false));
        }
        // register memory computing steps that use memory compute keys
        for (final MemoryComputing<?> memoryComputing : TraversalHelper.getStepsOfAssignableClassRecursively(MemoryComputing.class, this.traversal.get())) {
            this.memoryComputeKeys.add(memoryComputing.getMemoryComputeKey());
        }
        // register profile steps (TODO: try to hide this)
        for (final ProfileStep profileStep : TraversalHelper.getStepsOfAssignableClassRecursively(ProfileStep.class, this.traversal.get())) {
            this.traversal.get().getSideEffects().register(profileStep.getId(), new MutableMetricsSupplier(profileStep.getPreviousStep()), ProfileStep.ProfileBiOperator.instance());
        }
        // register TraversalVertexProgram specific memory compute keys
        this.memoryComputeKeys.add(MemoryComputeKey.of(VOTE_TO_HALT, Operator.and, false, true));
        this.memoryComputeKeys.add(MemoryComputeKey.of(HALTED_TRAVERSERS, Operator.addAll, false, false));
        this.memoryComputeKeys.add(MemoryComputeKey.of(ACTIVE_TRAVERSERS, Operator.addAll, true, true));
        this.memoryComputeKeys.add(MemoryComputeKey.of(MUTATED_MEMORY_KEYS, Operator.addAll, false, true));
        this.memoryComputeKeys.add(MemoryComputeKey.of(COMPLETED_BARRIERS, Operator.addAll, true, true));

        // does the traversal need profile information
        this.profile = !TraversalHelper.getStepsOfAssignableClassRecursively(ProfileStep.class, this.traversal.get()).isEmpty();
    }

    @Override
    public void storeState(final Configuration configuration) {
        VertexProgram.super.storeState(configuration);
        this.traversal.storeState(configuration, TRAVERSAL);
        TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
    }

    @Override
    public void setup(final Memory memory) {
        // memory is local
        MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.SETUP);
        ((MemoryTraversalSideEffects) this.traversal.get().getSideEffects()).storeSideEffectsInMemory();
        memory.set(VOTE_TO_HALT, true);
        memory.set(MUTATED_MEMORY_KEYS, new HashSet<>());
        memory.set(COMPLETED_BARRIERS, new HashSet<>());
        // if halted traversers are being sent from a previous VertexProgram in an OLAP chain (non-distributed traversers), get them into the flow
        if (!this.haltedTraversers.isEmpty()) {
            final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
            IteratorUtils.removeOnNext(this.haltedTraversers.iterator()).forEachRemaining(traverser -> {
                traverser.setStepId(this.traversal.get().getStartStep().getId());
                toProcessTraversers.add(traverser);
            });
            assert this.haltedTraversers.isEmpty();
            final IndexedTraverserSet<Object,Vertex> remoteActiveTraversers = new IndexedTraverserSet.VertexIndexedTraverserSet();
            MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, this.haltedTraversers, this.haltedTraverserStrategy);
            memory.set(HALTED_TRAVERSERS, this.haltedTraversers);
            memory.set(ACTIVE_TRAVERSERS, remoteActiveTraversers);
        } else {
            memory.set(HALTED_TRAVERSERS, new TraverserSet<>());
            memory.set(ACTIVE_TRAVERSERS, new IndexedTraverserSet.VertexIndexedTraverserSet());
        }
        // local variable will no longer be used so null it for GC
        this.haltedTraversers = null;
        // does the traversal need profile information
        this.profile = !TraversalHelper.getStepsOfAssignableClassRecursively(ProfileStep.class, this.traversal.get()).isEmpty();
    }

    @Override
    public Set<MessageScope> getMessageScopes(final Memory memory) {
        return MESSAGE_SCOPES;
    }

    @Override
    public void execute(final Vertex vertex, final Messenger<TraverserSet<Object>> messenger, final Memory memory) {
        // if any global halted traversers, simply don't use them as they were handled by master setup()
        // these halted traversers are typically from a previous OLAP job that yielded traversers at the master traversal
        if (null != this.haltedTraversers)
            this.haltedTraversers = null;
        // memory is distributed
        MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.EXECUTE);
        // if a barrier was completed in another worker, it is also completed here (ensure distributed barriers are synchronized)
        final Set<String> completedBarriers = memory.get(COMPLETED_BARRIERS);
        for (final String stepId : completedBarriers) {
            final Step<?, ?> step = this.traversalMatrix.getStepById(stepId);
            if (step instanceof Barrier)
                ((Barrier) this.traversalMatrix.getStepById(stepId)).done();
        }
        // define halted traversers
        final VertexProperty<TraverserSet<Object>> property = vertex.property(HALTED_TRAVERSERS);
        final TraverserSet<Object> haltedTraversers;
        if (property.isPresent()) {
            haltedTraversers = property.value();
        } else {
            haltedTraversers = new TraverserSet<>();
            vertex.property(VertexProperty.Cardinality.single, HALTED_TRAVERSERS, haltedTraversers);
        }
        //////////////////
        if (memory.isInitialIteration()) {    // ITERATION 1
            final TraverserSet<Object> activeTraversers = new TraverserSet<>();
            // if halted traversers are being sent from a previous VertexProgram in an OLAP chain (distributed traversers), get them into the flow
            IteratorUtils.removeOnNext(haltedTraversers.iterator()).forEachRemaining(traverser -> {
                traverser.setStepId(this.traversal.get().getStartStep().getId());
                activeTraversers.add(traverser);
            });
            assert haltedTraversers.isEmpty();
            // for g.V()/E()
            if (this.traversal.get().getStartStep() instanceof GraphStep) {
                final GraphStep<Element, Element> graphStep = (GraphStep<Element, Element>) this.traversal.get().getStartStep();
                graphStep.reset();
                activeTraversers.forEach(traverser -> graphStep.addStart((Traverser.Admin) traverser));
                activeTraversers.clear();
                if (graphStep.returnsVertex())
                    graphStep.setIteratorSupplier(() -> ElementHelper.idExists(vertex.id(), graphStep.getIds()) ? (Iterator) IteratorUtils.of(vertex) : EmptyIterator.instance());
                else
                    graphStep.setIteratorSupplier(() -> (Iterator) IteratorUtils.filter(vertex.edges(Direction.OUT), edge -> ElementHelper.idExists(edge.id(), graphStep.getIds())));
                graphStep.forEachRemaining(traverser -> {
                    if (traverser.isHalted()) {
                        if (this.returnHaltedTraversers)
                            memory.add(HALTED_TRAVERSERS, new TraverserSet<>(this.haltedTraverserStrategy.halt(traverser)));
                        else
                            haltedTraversers.add((Traverser.Admin) traverser.detach());
                    } else
                        activeTraversers.add((Traverser.Admin) traverser);
                });
            }
            memory.add(VOTE_TO_HALT, activeTraversers.isEmpty() || WorkerExecutor.execute(vertex, new SingleMessenger<>(messenger, activeTraversers), this.traversalMatrix, memory, this.returnHaltedTraversers, haltedTraversers, this.haltedTraverserStrategy));
        } else   // ITERATION 1+
            memory.add(VOTE_TO_HALT, WorkerExecutor.execute(vertex, messenger, this.traversalMatrix, memory, this.returnHaltedTraversers, haltedTraversers, this.haltedTraverserStrategy));
        // save space by not having an empty halted traversers property
        if (this.returnHaltedTraversers || haltedTraversers.isEmpty())
            vertex.<TraverserSet>property(HALTED_TRAVERSERS).remove();
    }

    @Override
    public boolean terminate(final Memory memory) {
        // memory is local
        MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.TERMINATE);
        final boolean voteToHalt = memory.<Boolean>get(VOTE_TO_HALT);
        memory.set(VOTE_TO_HALT, true);
        memory.set(ACTIVE_TRAVERSERS, new IndexedTraverserSet.VertexIndexedTraverserSet());
        if (voteToHalt) {
            // local traverser sets to process
            final TraverserSet<Object> toProcessTraversers = new TraverserSet<>();
            // traversers that need to be sent back to the workers (no longer can be processed locally by the master traversal)
            final IndexedTraverserSet<Object,Vertex> remoteActiveTraversers = new IndexedTraverserSet.VertexIndexedTraverserSet();
            // halted traversers that have completed their journey
            final TraverserSet<Object> haltedTraversers = memory.get(HALTED_TRAVERSERS);
            // get all barrier traversers
            final Set<String> completedBarriers = new HashSet<>();
            MasterExecutor.processMemory(this.traversalMatrix, memory, toProcessTraversers, completedBarriers);
            // process all results from barriers locally and when elements are touched, put them in remoteActiveTraversers
            MasterExecutor.processTraversers(this.traversal, this.traversalMatrix, toProcessTraversers, remoteActiveTraversers, haltedTraversers, this.haltedTraverserStrategy);
            // tell parallel barriers that might not have been active in the last round that they are no longer active.
            // accumulate all previously-completed barriers: worker clones start with done=false and need done() called
            // for every barrier ever completed (not just the most recent ones) to prevent stale lazy re-evaluation.
            // see TINKERPOP-3210 for the lazy cap() re-firing that motivated this change.
            final Set<String> allCompletedBarriers = new HashSet<>(memory.get(COMPLETED_BARRIERS));
            allCompletedBarriers.addAll(completedBarriers);
            memory.set(COMPLETED_BARRIERS, allCompletedBarriers);
            if (!remoteActiveTraversers.isEmpty() ||
                    completedBarriers.stream().map(this.traversalMatrix::getStepById).filter(step -> step instanceof LocalBarrier).findAny().isPresent()) {
                // send active traversers back to workers
                memory.set(ACTIVE_TRAVERSERS, remoteActiveTraversers);
                return false;
            } else {
                // finalize locally with any last traversers dangling in the local traversal
                final Step<?, Object> endStep = (Step<?, Object>) this.traversal.get().getEndStep();
                while (endStep.hasNext()) {
                    haltedTraversers.add(this.haltedTraverserStrategy.halt(endStep.next()));
                }
                // the result of a TraversalVertexProgram are the halted traversers
                memory.set(HALTED_TRAVERSERS, haltedTraversers);
                // finalize profile side-effect steps. (todo: try and hide this)
                for (final ProfileSideEffectStep profileStep : TraversalHelper.getStepsOfAssignableClassRecursively(ProfileSideEffectStep.class, this.traversal.get())) {
                    this.traversal.get().getSideEffects().set(profileStep.getSideEffectKey(), profileStep.generateFinalResult(this.traversal.get().getSideEffects().get(profileStep.getSideEffectKey())));
                }
                return true;
            }
        } else {
            return false;
        }
    }

    @Override
    public void workerIterationStart(final Memory memory) {
        // start collecting profile metrics
        if (this.profile) {
            this.iterationMetrics = new MutableMetrics("iteration" + memory.getIteration(), "Worker Iteration " + memory.getIteration());
            this.iterationMetrics.start();
        }
    }

    @Override
    public void workerIterationEnd(final Memory memory) {
        // store profile metrics in proper ProfileStep metrics
        if (this.profile) {
            final List<ProfileStep> profileSteps = TraversalHelper.getStepsOfAssignableClassRecursively(ProfileStep.class, this.traversal.get());
            // guess the profile step to store data
            int profileStepIndex = memory.getIteration();
            // if we guess wrongly write timing into last step
            profileStepIndex = profileStepIndex >= profileSteps.size() ? profileSteps.size() - 1 : profileStepIndex;
            this.iterationMetrics.finish(0);
            // reset counts
            this.iterationMetrics.setCount(TraversalMetrics.TRAVERSER_COUNT_ID,0);
            if (null != MemoryTraversalSideEffects.getMemorySideEffectsPhase(this.traversal.get())) {
                this.traversal.get().getSideEffects().add(profileSteps.get(profileStepIndex).getId(), this.iterationMetrics);
            }
            this.iterationMetrics = null;
        }
    }

    @Override
    public Set<VertexComputeKey> getVertexComputeKeys() {
        return VERTEX_COMPUTE_KEYS;
    }

    @Override
    public Set<MemoryComputeKey> getMemoryComputeKeys() {
        return this.memoryComputeKeys;
    }

    @Override
    public Set<MapReduce> getMapReducers() {
        return this.mapReducers;
    }

    @Override
    public Optional<MessageCombiner<TraverserSet<Object>>> getMessageCombiner() {
        return (Optional) TraversalVertexProgramMessageCombiner.instance();
    }

    @Override
    public TraversalVertexProgram clone() {
        try {
            final TraversalVertexProgram clone = (TraversalVertexProgram) super.clone();
            clone.traversal = this.traversal.clone();
            if (!clone.traversal.get().isLocked())
                clone.traversal.get().applyStrategies();
            clone.traversalMatrix = new TraversalMatrix<>(clone.traversal.get());
            clone.memoryComputeKeys = new HashSet<>();
            for (final MemoryComputeKey memoryComputeKey : this.memoryComputeKeys) {
                clone.memoryComputeKeys.add(memoryComputeKey.clone());
            }
            return clone;
        } catch (final CloneNotSupportedException e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    @Override
    public GraphComputer.ResultGraph getPreferredResultGraph() {
        return GraphComputer.ResultGraph.ORIGINAL;
    }

    @Override
    public GraphComputer.Persist getPreferredPersist() {
        return GraphComputer.Persist.NOTHING;
    }

    @Override
    public String toString() {
        final String traversalString = this.traversal.get().toString().substring(1);
        return StringFactory.vertexProgramString(this, traversalString.substring(0, traversalString.length() - 1));
    }

    @Override
    public Features getFeatures() {
        return new Features() {
            @Override
            public boolean requiresGlobalMessageScopes() {
                return true;
            }

            @Override
            public boolean requiresVertexPropertyAddition() {
                return true;
            }
        };
    }

    //////////////

    public static Builder build() {
        return new Builder();
    }

    public final static class Builder extends AbstractVertexProgramBuilder<Builder> {

        private Builder() {
            super(TraversalVertexProgram.class);
        }

        public Builder haltedTraversers(final TraverserSet<Object> haltedTraversers) {
            TraversalVertexProgram.storeHaltedTraversers(this.configuration, haltedTraversers);
            return this;
        }

        public Builder traversal(final TraversalSource traversalSource, final String scriptEngine, final String traversalScript, final Object... bindings) {
            return this.traversal(new ScriptTraversal<>(traversalSource, scriptEngine, traversalScript, bindings));
        }

        public Builder traversal(Traversal.Admin<?, ?> traversal) {
            // this is necessary if the job was submitted via TraversalVertexProgram.build() instead of TraversalVertexProgramStep.
            if (!(traversal.getParent() instanceof TraversalVertexProgramStep)) {
                final MemoryTraversalSideEffects memoryTraversalSideEffects = new MemoryTraversalSideEffects(traversal.getSideEffects());
                final Traversal.Admin<?, ?> parentTraversal = new DefaultTraversal<>();
                traversal.getGraph().ifPresent(parentTraversal::setGraph);
                final TraversalStrategies strategies = traversal.getStrategies().clone();
                strategies.addStrategies(ComputerFinalizationStrategy.instance(), ComputerVerificationStrategy.instance(), new VertexProgramStrategy(Computer.compute()));
                parentTraversal.setStrategies(strategies);
                traversal.setStrategies(strategies);
                parentTraversal.setSideEffects(memoryTraversalSideEffects);
                parentTraversal.addStep(new TraversalVertexProgramStep(parentTraversal, traversal));
                traversal = ((TraversalVertexProgramStep) parentTraversal.getStartStep()).getGlobalChildren().get(0);
                traversal.setSideEffects(memoryTraversalSideEffects);
            }
            PureTraversal.storeState(this.configuration, TRAVERSAL, traversal);
            return this;
        }
    }

}