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);
        }
    }
}
 

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());
}
 

/**
 * The compute statistics base algorithm that is called in every compute statistics query
 *
 * @param <S> The return type of StatisticsMapReduce
 * @return result of compute statistics algorithm, which will be of type S
 */
@Nullable
private <S> S runComputeStatistics(GraqlCompute.Statistics.Value query) {
    AttributeType.ValueType<?> targetValueType = validateAndGetTargetValueType(query);
    if (!targetContainsInstance(query)) return null;

    Set<LabelId> extendedScopeTypes = convertLabelsToIds(extendedScopeTypeLabels(query));
    Set<LabelId> targetTypes = convertLabelsToIds(targetTypeLabels(query));

    VertexProgram program = initStatisticsVertexProgram(query, targetTypes, targetValueType);
    StatisticsMapReduce<?> mapReduce = initStatisticsMapReduce(query, targetTypes, targetValueType);
    ComputerResult computerResult = compute(program, mapReduce, extendedScopeTypes);

    if (query.method().equals(MEDIAN)) {
        Number result = computerResult.memory().get(MedianVertexProgram.MEDIAN);
        LOG.debug("Median = {}", result);
        return (S) result;
    }

    Map<Serializable, S> resultMap = computerResult.memory().get(mapReduce.getClass().getName());
    LOG.debug("Result = {}", resultMap.get(MapReduce.NullObject.instance()));
    return resultMap.get(MapReduce.NullObject.instance());
}
 

@Override
public void storeState(final Configuration configuration) {
    VertexProgram.super.storeState(configuration);
    this.sourceVertexFilterTraversal.storeState(configuration, SOURCE_VERTEX_FILTER);
    this.targetVertexFilterTraversal.storeState(configuration, TARGET_VERTEX_FILTER);
    this.edgeTraversal.storeState(configuration, EDGE_TRAVERSAL);
    this.distanceTraversal.storeState(configuration, DISTANCE_TRAVERSAL);
    configuration.setProperty(INCLUDE_EDGES, this.includeEdges);
    if (this.maxDistance != null)
        configuration.setProperty(MAX_DISTANCE, maxDistance);
    if (this.traversal != null) {
        this.traversal.storeState(configuration, ProgramVertexProgramStep.ROOT_TRAVERSAL);
        configuration.setProperty(ProgramVertexProgramStep.STEP_ID, this.programStep.getId());
    }
    TraversalVertexProgram.storeHaltedTraversers(configuration, this.haltedTraversers);
}
 

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());
        }
    }
}
 

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);
        }
    }
}
 

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());
}
 

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));
    }
}
 

@Override
public final ComputerResult compute(@Nullable VertexProgram<?> program,
                                    @Nullable MapReduce<?, ?, ?, ?, ?> mapReduce,
                                    @Nullable Set<LabelId> scope,
                                    Boolean includesRolePlayerEdges) {

    return new OLAPOperation(hadoopGraph).compute(program, mapReduce, scope, includesRolePlayerEdges);
}
 

@Override
public final ComputerResult compute(@Nullable VertexProgram<?> program,
                                    @Nullable MapReduce<?, ?, ?, ?, ?> mapReduce,
                                    @Nullable Set<LabelId> scope) {

    return new OLAPOperation(hadoopGraph).compute(program, mapReduce, scope);
}
 

private VertexProgramScanJob(IDManager idManager, FulgoraMemory memory,
                            FulgoraVertexMemory vertexMemory, VertexProgram<M> vertexProgram) {
    this.idManager = idManager;
    this.memory = memory;
    this.vertexMemory = vertexMemory;
    this.vertexProgram = vertexProgram;
    this.combiner = FulgoraUtil.getMessageCombiner(vertexProgram);
}
 

public FulgoraVertexMemory(int numVertices, final IDManager idManager, final VertexProgram<M> vertexProgram) {
    Preconditions.checkArgument(numVertices>=0 && vertexProgram!=null && idManager!=null);
    vertexStates = new NonBlockingHashMapLong<>(numVertices);
    partitionVertices = new NonBlockingHashMapLong<>(64);
    this.idManager = idManager;
    this.combiner = FulgoraUtil.getMessageCombiner(vertexProgram);
    this.elementKeyMap = getIdMap(vertexProgram.getElementComputeKeys());
    this.previousScopes = ImmutableMap.of();
}
 

public FulgoraMemory(final VertexProgram<?> vertexProgram, final Set<MapReduce> mapReducers) {
    this.currentMap = new ConcurrentHashMap<>();
    this.previousMap = new ConcurrentHashMap<>();
    if (null != vertexProgram) {
        for (final String key : vertexProgram.getMemoryComputeKeys()) {
            MemoryHelper.validateKey(key);
            this.memoryKeys.add(key);
        }
    }
    for (final MapReduce mapReduce : mapReducers) {
        this.memoryKeys.add(mapReduce.getMemoryKey());
    }
}
 

public PartitionedVertexProgramExecutor(StandardTitanGraph graph, FulgoraMemory memory,
                             FulgoraVertexMemory vertexMemory, VertexProgram<M> vertexProgram) {
    this.graph=graph;
    this.idManager = graph.getIDManager();
    this.memory = memory;
    this.vertexMemory = vertexMemory;
    this.vertexProgram = vertexProgram;
}
 

@Override
public void storeState(final Configuration configuration) {
    VertexProgram.super.storeState(configuration);
    configuration.setProperty(ALPHA, this.alpha);
    configuration.setProperty(EPSILON, this.epsilon);
    configuration.setProperty(PROPERTY, this.property);
    configuration.setProperty(MAX_ITERATIONS, this.maxIterations);
    if (null != this.edgeTraversal)
        this.edgeTraversal.storeState(configuration, EDGE_TRAVERSAL);
    if (null != this.initialRankTraversal)
        this.initialRankTraversal.storeState(configuration, INITIAL_RANK_TRAVERSAL);
}
 

@Override
public void storeState(final Configuration config) {
    VertexProgram.super.storeState(config);
    if (configuration != null) {
        ConfigurationUtils.copy(configuration, config);
    }
}
 

public ProgramVertexProgramStep(final Traversal.Admin traversal, final VertexProgram vertexProgram) {
    super(traversal);
    this.configuration = new HashMap<>();
    final MapConfiguration base = new MapConfiguration(this.configuration);
    vertexProgram.storeState(base);
    this.toStringOfVertexProgram = vertexProgram.toString();
    this.traverserRequirements = vertexProgram.getTraverserRequirements();
}
 

@Override
public VertexProgram generateProgram(final Graph graph, final Memory memory) {
    final MapConfiguration base = new MapConfiguration(this.configuration);
    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);
}
 

public VertexProgram take() {
    try {
        return this.pool.poll(TIMEOUT_MS, TimeUnit.MILLISECONDS);
    } catch (final InterruptedException e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}
 

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);
    }
}
 

@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));
}
 

@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());
}
 

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));
    }
}
 

public void setVertexProgram(final VertexProgram vertexProgram) {
    this.vertexProgramPool = new VertexProgramPool(vertexProgram, this.numberOfWorkers);
}
 

@Override
public GraphComputer program(final VertexProgram vertexProgram) {
    this.vertexProgram = vertexProgram;
    return this;
}
 

@CheckReturnValue
public ComputerResult compute(@Nullable VertexProgram program, @Nullable MapReduce mapReduce,
                              @Nullable Set<LabelId> types) {
    return compute(program, mapReduce, types, true);
}
 

public static void main(String[] args) throws Exception {
    FileConfiguration configuration = new PropertiesConfiguration(args[0]);
    new GraknSparkComputer(HadoopGraph.open(configuration))
            .program(VertexProgram.createVertexProgram(HadoopGraph.open(configuration), configuration))
            .submit().get();
}
 

ComputerResult compute(@Nullable VertexProgram<?> program,
@Nullable MapReduce<?, ?, ?, ?, ?> mapReduce,
@Nullable Set<LabelId> scope,
Boolean includesRolePlayerEdges);
 

ComputerResult compute(@Nullable VertexProgram<?> program,
@Nullable MapReduce<?, ?, ?, ?, ?> mapReduce,
@Nullable Set<LabelId> scope);
 

public static<M> Executor getVertexProgramScanJob(StandardTitanGraph graph, FulgoraMemory memory,
                                              FulgoraVertexMemory vertexMemory, VertexProgram<M> vertexProgram) {
    VertexProgramScanJob<M> job = new VertexProgramScanJob<M>(graph.getIDManager(),memory,vertexMemory,vertexProgram);
    return new Executor(graph,job);
}