import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void safeExecute(final Vertex vertex, Messenger<PathMessage> messenger, final Memory memory) {
    if (memory.isInitialIteration()) {
        // The type of id will likely have to change as we support more and more vendors
        String id = getVertexId(vertex);
        if (persistentProperties.get(SOURCE).equals(id)) {
            LOGGER.debug("Found source vertex");
            vertex.property(VISITED_IN_ITERATION, 1);
            sendMessage(messenger, PathMessage.of(FROM_SOURCE, id));
        } else if (persistentProperties.get(DESTINATION).equals(id)) {
            LOGGER.debug("Found destination vertex");
            vertex.property(VISITED_IN_ITERATION, -1);
            sendMessage(messenger, PathMessage.of(FROM_DESTINATION, id));
        }
    } else {
        if (memory.<Boolean>get(FOUND_PATH)) {
            if (messenger.receiveMessages().hasNext() && vertex.property(VISITED_IN_ITERATION).isPresent()) {
                recordPredecessors(vertex, messenger, memory);
            }
        } else if (messenger.receiveMessages().hasNext()) {
            updateInstance(vertex, messenger, memory);
        }
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void setup(final Memory memory) {
    LOGGER.debug("MedianVertexProgram Started !!!!!!!!");
    memory.set(COUNT, 0L);
    memory.set(LABEL_SELECTED, memory.getIteration());
    memory.set(NEGATIVE_COUNT, 0L);
    memory.set(POSITIVE_COUNT, 0L);
    memory.set(FOUND, false);
    if (persistentProperties.get(RESOURCE_DATA_TYPE).equals(Schema.VertexProperty.VALUE_LONG.name())) {
        memory.set(MEDIAN, 0L);
        memory.set(PIVOT, 0L);
        memory.set(PIVOT_NEGATIVE, 0L);
        memory.set(PIVOT_POSITIVE, 0L);
    } else {
        memory.set(MEDIAN, 0D);
        memory.set(PIVOT, 0D);
        memory.set(PIVOT_NEGATIVE, 0D);
        memory.set(PIVOT_POSITIVE, 0D);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
private void updateEntityAndAttribute(Vertex vertex, Messenger<String> messenger, Memory memory) {
    if (vertex.property(CLUSTER_LABEL).isPresent()) {
        String id = vertex.value(Schema.VertexProperty.ID.name());
        int messageCount = getMessageCount(messenger, id) +
                (vertex.property(IMPLICIT_MESSAGE_COUNT).isPresent() ?
                        (int) vertex.value(IMPLICIT_MESSAGE_COUNT) : 0);

        if (messageCount >= k) {
            LOGGER.trace("Sending msg from " + id);
            sendMessage(messenger, id);
            memory.add(K_CORE_EXIST, true);
        } else {
            LOGGER.trace("Removing label of " + id);
            vertex.property(CLUSTER_LABEL).remove();
            memory.add(K_CORE_STABLE, false);
        }
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void safeExecute(final Vertex vertex, Messenger<Long> messenger, final Memory memory) {
    switch (memory.getIteration()) {
        case 0:
            degreeStatisticsStepResourceOwner(vertex, messenger, ofLabelIds);
            break;
        case 1:
            degreeStatisticsStepResourceRelation(vertex, messenger, ofLabelIds);
            break;
        case 2:
            degreeStatisticsStepResource(vertex, messenger, ofLabelIds);
            break;
        default:
            throw CommonUtil.unreachableStatement("Exceeded expected maximum number of iterations");
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public <R> R get(final String key) throws IllegalArgumentException {
    if (!this.memoryComputeKeys.containsKey(key)) {
        throw Memory.Exceptions.memoryDoesNotExist(key);
    }
    if (this.inExecute && !this.memoryComputeKeys.get(key).isBroadcast()) {
        throw Memory.Exceptions.memoryDoesNotExist(key);
    }
    final ObjectWritable<R> r = (ObjectWritable<R>) (this.inExecute ?
            this.broadcast.value().get(key) : this.sparkMemory.get(key).value());
    if (null == r || r.isEmpty()) {
        throw Memory.Exceptions.memoryDoesNotExist(key);
    } else {
        return r.get();
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void execute(final Vertex vertex, Messenger<Double> messenger, final Memory memory) {
    if (memory.isInitialIteration()) {
        messenger.sendMessage(this.countMessageScope, 1.0d);
    } else if (1 == memory.getIteration()) {
        double initialPageRank = (null == this.initialRankTraversal ?
                1.0d :
                TraversalUtil.apply(vertex, this.initialRankTraversal.get()).doubleValue()) / this.vertexCountAsDouble;
        double edgeCount = IteratorUtils.reduce(messenger.receiveMessages(), 0.0d, (a, b) -> a + b);
        vertex.property(VertexProperty.Cardinality.single, this.property, initialPageRank);
        vertex.property(VertexProperty.Cardinality.single, EDGE_COUNT, edgeCount);
        if (!this.terminate(memory)) // don't send messages if this is the last iteration
            messenger.sendMessage(this.incidentMessageScope, initialPageRank / edgeCount);
    } else {
        double newPageRank = IteratorUtils.reduce(messenger.receiveMessages(), 0.0d, (a, b) -> a + b);
        newPageRank = (this.alpha * newPageRank) + ((1.0d - this.alpha) / this.vertexCountAsDouble);
        vertex.property(VertexProperty.Cardinality.single, this.property, newPageRank);
        if (!this.terminate(memory)) // don't send messages if this is the last iteration
            messenger.sendMessage(this.incidentMessageScope, newPageRank / vertex.<Double>value(EDGE_COUNT));
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void workerIterationStart(final Memory memory) {
    if (null == graph) {
        graph = GraphFactory.open(configuration.subset(WRITE_GRAPH_CFG_KEY));
        LOGGER.info("Opened Graph instance: {}", graph);
        try {
            listener = new BulkLoadingListener();
            g = graph.traversal().withStrategies(EventStrategy.build().addListener(listener).create());
        } catch (Exception e) {
            try {
                graph.close();
            } catch (Exception e2) {
                LOGGER.warn("Failed to close Graph instance", e2);
            }
            throw e;
        }
    } else {
        LOGGER.warn("Leaked Graph instance: {}", graph);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public TraversalVertexProgram generateProgram(final Graph graph, final Memory memory) {
    final Traversal.Admin<?, ?> computerSpecificTraversal = this.computerTraversal.getPure();
    final TraversalStrategies computerSpecificStrategies = this.getTraversal().getStrategies().clone();
    TraversalStrategies.GlobalCache.getStrategies(graph.getClass())
            .toList()
            .stream()
            .filter(s -> s instanceof TraversalStrategy.ProviderOptimizationStrategy)
            .forEach(computerSpecificStrategies::addStrategies);
    computerSpecificTraversal.setStrategies(computerSpecificStrategies);
    computerSpecificTraversal.setSideEffects(new MemoryTraversalSideEffects(this.getTraversal().getSideEffects()));
    computerSpecificTraversal.setParent(this);
    final TraversalVertexProgram.Builder builder = TraversalVertexProgram.build().traversal(computerSpecificTraversal);
    if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
        builder.haltedTraversers(memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
    return builder.create(graph);
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
protected static void processMemory(final TraversalMatrix<?, ?> traversalMatrix, final Memory memory, final TraverserSet<Object> toProcessTraversers, final Set<String> completedBarriers) {
    // handle traversers and data that were sent from the workers to the master traversal via memory
    if (memory.exists(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
        for (final String key : memory.<Set<String>>get(TraversalVertexProgram.MUTATED_MEMORY_KEYS)) {
            final Step<Object, Object> step = traversalMatrix.getStepById(key);
            assert step instanceof Barrier;
            completedBarriers.add(step.getId());
            if (!(step instanceof LocalBarrier)) {  // local barriers don't do any processing on the master traversal (they just lock on the workers)
                final Barrier<Object> barrier = (Barrier<Object>) step;
                barrier.addBarrier(memory.get(key));
                step.forEachRemaining(toProcessTraversers::add);
                // if it was a reducing barrier step, reset the barrier to its seed value
                if (step instanceof ReducingBarrierStep)
                    memory.set(step.getId(), ((ReducingBarrierStep) step).getSeedSupplier().get());
            }
        }
    }
    memory.set(TraversalVertexProgram.MUTATED_MEMORY_KEYS, new HashSet<>());
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public boolean terminate(final Memory memory) {
    final TraverserGenerator generator = this.traversal.get().getTraverserGenerator();
    MemoryTraversalSideEffects.setMemorySideEffects(this.traversal.get(), memory, ProgramPhase.TERMINATE);
    checkSideEffects();
    if (memory.isInitialIteration()) {
        assertFalse(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
        return false;
    } else {
        ///
        assertTrue(memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS));
        final TraverserSet<String> haltedTraversers = memory.get(TraversalVertexProgram.HALTED_TRAVERSERS);
        haltedTraversers.add(generator.generate("hello", this.programStep, 1l));
        haltedTraversers.add(generator.generate("gremlin", this.programStep, 1l));
        memory.set(TraversalVertexProgram.HALTED_TRAVERSERS, haltedTraversers);
        return true;
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public <R> R get(final String key) throws IllegalArgumentException {
    final R r = (R) this.previousMap.get(key);
    if (null == r)
        throw Memory.Exceptions.memoryDoesNotExist(key);
    else if (this.inExecute && !this.memoryKeys.get(key).isBroadcast())
        throw Memory.Exceptions.memoryDoesNotExist(key);
    else
        return r;
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void set(final String key, final Object value) {
    checkKeyValue(key, value);
    if (this.inExecute)
        throw Memory.Exceptions.memorySetOnlyDuringVertexProgramSetUpAndTerminate(key);
    this.currentMap.put(key, value);
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void add(final String key, final Object value) {
    checkKeyValue(key, value);
    if (!this.inExecute)
        throw Memory.Exceptions.memoryAddOnlyDuringVertexProgramExecute(key);
    this.currentMap.compute(key, (k, v) -> null == v ? value : this.memoryKeys.get(key).getReducer().apply(v, value));
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void setup(final Memory memory) {
    LOGGER.debug("ShortestPathVertexProgram Started !!!!!!!!");

    memory.set(VOTE_TO_HALT_SOURCE, true);
    memory.set(VOTE_TO_HALT_DESTINATION, true);
    memory.set(FOUND_PATH, false);
    memory.set(PATH_HAS_MIDDLE_POINT, true);
    memory.set(ITERATIONS_LEFT, -1);
    memory.set(PREDECESSORS_FROM_SOURCE, new HashMap<String, Set<String>>());
    memory.set(PREDECESSORS_FROM_DESTINATION, new HashMap<String, Set<String>>());
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void set(final String key, final Object value) {
    this.checkKeyValue(key, value);
    if (this.inExecute)
        throw Memory.Exceptions.memorySetOnlyDuringVertexProgramSetUpAndTerminate(key);
    this.setAggregatedValue(key, new ObjectWritable<>(new Pair<>(this.memoryComputeKeys.get(key).getReducer(), value)));
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void execute(Vertex vertex, Messenger<T> messenger, Memory memory) {
    // try to deal with ghost vertex issues by ignoring them
    if (Utility.isAlive(vertex)) {
        safeExecute(vertex, messenger, memory);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public void safeExecute(final Vertex vertex, Messenger<String> messenger, final Memory memory) {
    if (memory.isInitialIteration()) {
        String id = vertex.value(Schema.VertexProperty.ID.name());
        vertex.property(CLUSTER_LABEL, id);
        messenger.sendMessage(messageScopeIn, id);
        messenger.sendMessage(messageScopeOut, id);
    } else {
        updateClusterLabel(vertex, messenger, memory);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
private static void updateClusterLabel(Vertex vertex, Messenger<String> messenger, Memory memory) {
    String currentMax = vertex.value(CLUSTER_LABEL);
    String max = IteratorUtils.reduce(messenger.receiveMessages(), currentMax,
            (a, b) -> a.compareTo(b) > 0 ? a : b);
    if (max.compareTo(currentMax) > 0) {
        vertex.property(CLUSTER_LABEL, max);
        messenger.sendMessage(messageScopeIn, max);
        messenger.sendMessage(messageScopeOut, max);
        memory.add(VOTE_TO_HALT, false);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public boolean terminate(final Memory memory) {
    LOGGER.debug("Finished Iteration " + memory.getIteration());
    if (memory.getIteration() < 2) return false;
    if (memory.<Boolean>get(VOTE_TO_HALT)) {
        return true;
    }
    if (memory.getIteration() == MAX_ITERATION) {
        LOGGER.debug("Reached Max Iteration: " + MAX_ITERATION + " !!!!!!!!");
        throw GraqlQueryException.maxIterationsReached(this.getClass());
    }

    memory.set(VOTE_TO_HALT, true);
    return false;
}
 

import org.apache.tinkerpop.gremlin.process.computer.Memory; //导入依赖的package包/类
@Override
public Set<MessageScope> getMessageScopes(final Memory memory) {
    switch (memory.getIteration()) {
        case 0:
            return Sets.newHashSet(messageScopeShortcutIn, messageScopeResourceOut);
        case 1:
            return Collections.singleton(messageScopeShortcutOut);
        default:
            return Collections.emptySet();
    }
}