import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public void executeVertexProgram(final TriConsumer<Iterator<Vertex>, VertexProgram, TinkerWorkerMemory> worker) throws InterruptedException {
    for (int i = 0; i < this.numberOfWorkers; i++) {
        final int index = i;
        this.completionService.submit(() -> {
            final VertexProgram vp = this.vertexProgramPool.take();
            final TinkerWorkerMemory workerMemory = this.workerMemoryPool.poll();
            final List<Vertex> vertices = this.workerVertices.get(index);
            worker.accept(vertices.iterator(), vp, workerMemory);
            this.vertexProgramPool.offer(vp);
            this.workerMemoryPool.offer(workerMemory);
            return null;
        });
    }
    for (int i = 0; i < this.numberOfWorkers; i++) {
        try {
            this.completionService.take().get();
        } catch (InterruptedException ie) {
            throw ie;
        } catch (final Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public ComputerResult compute(@Nullable VertexProgram program, @Nullable MapReduce mapReduce,
                              @Nullable Set<LabelId> types, Boolean includesRolePlayerEdges) {
    try {
        graphComputer = getGraphComputer();
        if (program != null) {
            graphComputer.program(program);
        } else {
            filterAllEdges = true;
        }
        if (mapReduce != null) graphComputer.mapReduce(mapReduce);
        applyFilters(types, includesRolePlayerEdges);
        return graphComputer.submit().get();
    } catch (InterruptedException | ExecutionException e) {
        throw asRuntimeException(e.getCause());
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public GraknSparkMemory(final VertexProgram<?> vertexProgram,
                        final Set<MapReduce> mapReducers,
                        final JavaSparkContext sparkContext) {
    if (null != vertexProgram) {
        for (final MemoryComputeKey key : vertexProgram.getMemoryComputeKeys()) {
            this.memoryComputeKeys.put(key.getKey(), key);
        }
    }
    for (final MapReduce mapReduce : mapReducers) {
        this.memoryComputeKeys.put(
                mapReduce.getMemoryKey(),
                MemoryComputeKey.of(mapReduce.getMemoryKey(), Operator.assign, false, false));
    }
    for (final MemoryComputeKey memoryComputeKey : this.memoryComputeKeys.values()) {
        this.sparkMemory.put(
                memoryComputeKey.getKey(),
                sparkContext.accumulator(ObjectWritable.empty(), memoryComputeKey.getKey(),
                        new MemoryAccumulator<>(memoryComputeKey)));
    }
    this.broadcast = sparkContext.broadcast(Collections.emptyMap());
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public static void validateProgramOnComputer(final GraphComputer computer, final VertexProgram vertexProgram) {
    if (vertexProgram.getMemoryComputeKeys().contains(null))
        throw Memory.Exceptions.memoryKeyCanNotBeNull();
    if (vertexProgram.getMemoryComputeKeys().contains(""))
        throw Memory.Exceptions.memoryKeyCanNotBeEmpty();

    final GraphComputer.Features graphComputerFeatures = computer.features();
    final VertexProgram.Features vertexProgramFeatures = vertexProgram.getFeatures();

    for (final Method method : VertexProgram.Features.class.getMethods()) {
        if (method.getName().startsWith("requires")) {
            final boolean supports;
            final boolean requires;
            try {
                supports = (boolean) GraphComputer.Features.class.getMethod(method.getName().replace("requires", "supports")).invoke(graphComputerFeatures);
                requires = (boolean) method.invoke(vertexProgramFeatures);
            } catch (final Exception e) {
                throw new IllegalStateException("A reflection exception has occurred: " + e.getMessage(), e);
            }
            if (requires && !supports)
                throw new IllegalStateException("The vertex program can not be executed on the graph computer: " + method.getName());
        }
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public void executeVertexProgram(final Consumer<VertexProgram> worker) throws InterruptedException {
    for (int i = 0; i < this.numberOfWorkers; i++) {
        this.completionService.submit(() -> {
            final VertexProgram vp = this.vertexProgramPool.take();
            worker.accept(vp);
            this.vertexProgramPool.offer(vp);
            return null;
        });
    }
    for (int i = 0; i < this.numberOfWorkers; i++) {
        try {
            this.completionService.take().get();
        } catch (InterruptedException ie) {
            throw ie;
        } catch (final Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public SparkMemory(final VertexProgram<?> vertexProgram, final Set<MapReduce> mapReducers, final JavaSparkContext sparkContext) {
    if (null != vertexProgram) {
        for (final MemoryComputeKey key : vertexProgram.getMemoryComputeKeys()) {
            this.memoryComputeKeys.put(key.getKey(), key);
        }
    }
    for (final MapReduce mapReduce : mapReducers) {
        this.memoryComputeKeys.put(mapReduce.getMemoryKey(), MemoryComputeKey.of(mapReduce.getMemoryKey(), Operator.assign, false, false));
    }
    for (final MemoryComputeKey memoryComputeKey : this.memoryComputeKeys.values()) {
        this.sparkMemory.put(
                memoryComputeKey.getKey(),
                sparkContext.accumulator(ObjectWritable.empty(), memoryComputeKey.getKey(), new MemoryAccumulator<>(memoryComputeKey)));
    }
    this.broadcast = sparkContext.broadcast(Collections.emptyMap());
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public void executeVertexProgram(Consumer<VertexProgram<M>> worker) {
for (int i = 0; i < numberOfWorkers; i++) {
    completionService.submit(() -> {
	@SuppressWarnings("unchecked")
	VertexProgram<M> vp = vertexProgramPool.take();
	worker.accept(vp);
	vertexProgramPool.offer(vp);
	return null;
    });
}
for (int i = 0; i < numberOfWorkers; i++) {
    try {
	completionService.take().get();
    } catch (Exception e) {
	throw new IllegalStateException(e.getMessage(), e);
    }
}
   }
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public TinkerMemory(final VertexProgram<?> vertexProgram, final Set<MapReduce> mapReducers) {
    this.currentMap = new ConcurrentHashMap<>();
    this.previousMap = new ConcurrentHashMap<>();
    if (null != vertexProgram) {
        for (final MemoryComputeKey memoryComputeKey : vertexProgram.getMemoryComputeKeys()) {
            this.memoryKeys.put(memoryComputeKey.getKey(), memoryComputeKey);
        }
    }
    for (final MapReduce mapReduce : mapReducers) {
        this.memoryKeys.put(mapReduce.getMemoryKey(), MemoryComputeKey.of(mapReduce.getMemoryKey(), Operator.assign, false, false));
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public void storeState(final Configuration configuration) {
    VertexProgram.super.storeState(configuration);
    configuration.setProperty(VERTEX_COUNT, this.vertexCountAsDouble);
    configuration.setProperty(ALPHA, this.alpha);
    configuration.setProperty(TOTAL_ITERATIONS, this.totalIterations);
    configuration.setProperty(PROPERTY, this.property);
    if (null != this.edgeTraversal)
        this.edgeTraversal.storeState(configuration, EDGE_TRAVERSAL);
    if (null != this.initialRankTraversal)
        this.initialRankTraversal.storeState(configuration, INITIAL_RANK_TRAVERSAL);
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public void storeState(final Configuration config) {
    VertexProgram.super.storeState(config);
    if (configuration != null) {
        ConfigurationUtils.copy(configuration, config);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public ProgramVertexProgramStep(final Traversal.Admin traversal, final VertexProgram vertexProgram) {
    super(traversal);
    this.configuration = new HashMap<>();
    final MapConfiguration base = new MapConfiguration(this.configuration);
    base.setDelimiterParsingDisabled(true);
    vertexProgram.storeState(base);
    this.toStringOfVertexProgram = vertexProgram.toString();
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public VertexProgram generateProgram(final Graph graph, final Memory memory) {
    final MapConfiguration base = new MapConfiguration(this.configuration);
    base.setDelimiterParsingDisabled(true);
    PureTraversal.storeState(base, ROOT_TRAVERSAL, TraversalHelper.getRootTraversal(this.getTraversal()).clone());
    base.setProperty(STEP_ID, this.getId());
    if (memory.exists(TraversalVertexProgram.HALTED_TRAVERSERS))
        TraversalVertexProgram.storeHaltedTraversers(base, memory.get(TraversalVertexProgram.HALTED_TRAVERSERS));
    return VertexProgram.createVertexProgram(graph, base);
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public VertexProgram take() {
    try {
        return this.pool.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS);
    } catch (final InterruptedException e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public void offer(final VertexProgram<?> vertexProgram) {
    try {
        this.pool.offer(vertexProgram, TIMEOUT_MS, TimeUnit.MILLISECONDS);
    } catch (final InterruptedException e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public void loadState(final Graph graph, final Configuration configuration) {
    VertexProgram.super.loadState(graph, configuration);
    this.traversal = PureTraversal.loadState(configuration, VertexProgramStep.ROOT_TRAVERSAL, graph);
    this.haltedTraversers = TraversalVertexProgram.loadHaltedTraversers(configuration);
    this.programStep = new TraversalMatrix<>(this.traversal.get()).getStepById(configuration.getString(ProgramVertexProgramStep.STEP_ID));
    this.memoryComputeKeys.addAll(MemoryTraversalSideEffects.getMemoryComputeKeys(this.traversal.get()));
    this.memoryComputeKeys.add(MemoryComputeKey.of(TraversalVertexProgram.HALTED_TRAVERSERS, Operator.addAll, false, false));
    this.memoryComputeKeys.add(MemoryComputeKey.of(TraversalVertexProgram.ACTIVE_TRAVERSERS, Operator.addAll, true, true));
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public void storeState(final Configuration configuration) {
    VertexProgram.super.storeState(configuration);
    this.traversal.storeState(configuration, VertexProgramStep.ROOT_TRAVERSAL);
    TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
    configuration.setProperty(ProgramVertexProgramStep.STEP_ID, this.programStep.getId());
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public void initialize(final TaskAttemptContext context) throws IOException, InterruptedException {
    final Configuration configuration = context.getConfiguration();
    this.recordWriter = ReflectionUtils.newInstance(configuration.getClass(Constants.GREMLIN_HADOOP_GRAPH_WRITER, OutputFormat.class, OutputFormat.class), configuration).getRecordWriter(context);
    this.transientComputeKeys = VertexProgramHelper.vertexComputeKeysAsArray(((VertexProgram<?>) VertexProgram.createVertexProgram(EmptyGraph.instance(), ConfUtil.makeApacheConfiguration(configuration))).getVertexComputeKeys().stream().
            filter(VertexComputeKey::isTransient).
            collect(Collectors.toSet()));
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public void compute(final Vertex<ObjectWritable, VertexWritable, NullWritable> vertex, final Iterable<ObjectWritable> messages) throws IOException {
    final GiraphWorkerContext workerContext = this.getWorkerContext();
    final VertexProgram<?> vertexProgram = workerContext.getVertexProgramPool().take();
    vertexProgram.execute(ComputerGraph.vertexProgram(vertex.getValue().get(), vertexProgram), workerContext.getMessenger((GiraphVertex) vertex, this, messages.iterator()), workerContext.getMemory());
    workerContext.getVertexProgramPool().offer(vertexProgram);
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
@Override
public GraphComputer program(final VertexProgram vertexProgram) {
    super.program(vertexProgram);
    this.memory.addVertexProgramMemoryComputeKeys(this.vertexProgram);
    final BaseConfiguration apacheConfiguration = new BaseConfiguration();
    apacheConfiguration.setDelimiterParsingDisabled(true);
    vertexProgram.storeState(apacheConfiguration);
    IteratorUtils.fill(apacheConfiguration.getKeys(), this.vertexProgramConfigurationKeys);
    ConfUtil.mergeApacheIntoHadoopConfiguration(apacheConfiguration, this.giraphConfiguration);
    this.vertexProgram.getMessageCombiner().ifPresent(combiner -> this.giraphConfiguration.setMessageCombinerClass(GiraphMessageCombiner.class));
    return this;
}
 

import org.apache.tinkerpop.gremlin.process.computer.VertexProgram; //导入依赖的package包/类
public void preApplication() throws InstantiationException, IllegalAccessException {
    final Configuration apacheConfiguration = ConfUtil.makeApacheConfiguration(this.getContext().getConfiguration());
    KryoShimServiceLoader.applyConfiguration(apacheConfiguration);
    final VertexProgram vertexProgram = VertexProgram.createVertexProgram(HadoopGraph.open(apacheConfiguration), apacheConfiguration);
    this.vertexProgramPool = new VertexProgramPool(vertexProgram, this.getContext().getConfiguration().getInt(GiraphConstants.NUM_COMPUTE_THREADS.getKey(), 1));
    this.memory = new GiraphMemory(this, vertexProgram);
}