public static <M> JavaPairRDD<Object, ViewIncomingPayload<M>> executeVertexProgramIteration(
            final JavaPairRDD<Object, VertexWritable> graphRDD,
            final JavaPairRDD<Object, ViewIncomingPayload<M>> viewIncomingRDD,
            final GraknSparkMemory memory,
            final Configuration graphComputerConfiguration,    // has the Graph/GraphComputer.configuration() information
            final Configuration vertexProgramConfiguration) { // has the VertexProgram.loadState() information

        boolean partitionedGraphRDD = graphRDD.partitioner().isPresent();

        if (partitionedGraphRDD && null != viewIncomingRDD) // the graphRDD and the viewRDD must have the same partitioner
        {assert graphRDD.partitioner().get().equals(viewIncomingRDD.partitioner().get());}
        final JavaPairRDD<Object, ViewOutgoingPayload<M>> viewOutgoingRDD = ((null == viewIncomingRDD) ?
                graphRDD.mapValues(vertexWritable -> new Tuple2<>(vertexWritable, Optional.<ViewIncomingPayload<M>>absent())) : // first iteration will not have any views or messages
                graphRDD.leftOuterJoin(viewIncomingRDD))                                                   // every other iteration may have views and messages
                // for each partition of vertices emit a view and their outgoing messages
                .mapPartitionsToPair(partitionIterator -> {
                    KryoShimServiceLoader.applyConfiguration(graphComputerConfiguration);

                    // if the partition is empty, return without starting a new VP iteration
                    if (!partitionIterator.hasNext()){
                        return Collections.emptyIterator();}

                    final VertexProgram<M> workerVertexProgram = VertexProgram.createVertexProgram(HadoopGraph.open(graphComputerConfiguration), vertexProgramConfiguration); // each partition(Spark)/worker(TP3) has a local copy of the vertex program (a worker's task)
                    final String[] vertexComputeKeysArray = VertexProgramHelper.vertexComputeKeysAsArray(workerVertexProgram.getVertexComputeKeys()); // the compute keys as an array
                    final SparkMessenger<M> messenger = new SparkMessenger<>();

                    workerVertexProgram.workerIterationStart(memory.asImmutable()); // start the worker
                    return IteratorUtils.map(partitionIterator, vertexViewIncoming -> {
                        final StarGraph.StarVertex vertex = vertexViewIncoming._2()._1().get(); // get the vertex from the vertex writable
                        final boolean hasViewAndMessages = vertexViewIncoming._2()._2().isPresent(); // if this is the first iteration, then there are no views or messages
                        final List<DetachedVertexProperty<Object>> previousView = hasViewAndMessages ? vertexViewIncoming._2()._2().get().getView() : memory.isInitialIteration() ? new ArrayList<>() : Collections.emptyList();
                        // revive compute properties if they already exist
                        if (memory.isInitialIteration() && vertexComputeKeysArray.length > 0){
                            vertex.properties(vertexComputeKeysArray).forEachRemaining(vertexProperty -> previousView.add(DetachedFactory.detach(vertexProperty, true)));}
                        // drop any computed properties that are cached in memory
                        vertex.dropVertexProperties(vertexComputeKeysArray);
                        final List<M> incomingMessages = hasViewAndMessages ? vertexViewIncoming._2()._2().get().getIncomingMessages() : Collections.emptyList();
                        IteratorUtils.removeOnNext(previousView.iterator()).forEachRemaining(property -> property.attach(Attachable.Method.create(vertex)));  // attach the view to the vertex
                        assert previousView.isEmpty();
                        // do the vertex's vertex program iteration
                        messenger.setVertexAndIncomingMessages(vertex, incomingMessages); // set the messenger with the incoming messages
                        workerVertexProgram.execute(ComputerGraph.vertexProgram(vertex, workerVertexProgram), messenger, memory); // execute the vertex program on this vertex for this iteration
                        // assert incomingMessages.isEmpty();  // maybe the program didn't read all the messages
                        incomingMessages.clear();
                        // detached the compute property view from the vertex
                        final List<DetachedVertexProperty<Object>> nextView = vertexComputeKeysArray.length == 0 ?  // not all vertex programs have compute keys
                                Collections.emptyList() :
                                IteratorUtils.list(IteratorUtils.map(vertex.properties(vertexComputeKeysArray), vertexProperty -> DetachedFactory.detach(vertexProperty, true)));
                        // drop compute property view as it has now been detached from the vertex
                        vertex.dropVertexProperties(vertexComputeKeysArray);
                        final List<Tuple2<Object, M>> outgoingMessages = messenger.getOutgoingMessages(); // get the outgoing messages being sent by this vertex
                        if (!partitionIterator.hasNext()) {
                            workerVertexProgram.workerIterationEnd(memory.asImmutable()); // if no more vertices in the partition, end the worker's iteration}
                        }return (nextView.isEmpty() && outgoingMessages.isEmpty()) ?
                                null : // if there is no view nor outgoing messages, emit nothing
                                new Tuple2<>(vertex.id(), new ViewOutgoingPayload<>(nextView, outgoingMessages));  // else, emit the vertex id, its view, and its outgoing messages
                    });
                }, true)  // true means that the partition is preserved
                .filter(tuple -> null != tuple); // if there are no messages or views, then the tuple is null (memory optimization)
        // the graphRDD and the viewRDD must have the same partitioner
        if (partitionedGraphRDD){
            assert graphRDD.partitioner().get().equals(viewOutgoingRDD.partitioner().get());}
        /////////////////////////////////////////////////////////////
        /////////////////////////////////////////////////////////////
        final PairFlatMapFunction<Tuple2<Object, ViewOutgoingPayload<M>>, Object, Payload> messageFunction =
                tuple -> IteratorUtils.concat(
                        IteratorUtils.of(new Tuple2<>(tuple._1(), tuple._2().getView())),      // emit the view payload
                        IteratorUtils.map(tuple._2().getOutgoingMessages().iterator(), message -> new Tuple2<>(message._1(), new MessagePayload<>(message._2()))));
        final MessageCombiner<M> messageCombiner = VertexProgram.<VertexProgram<M>>createVertexProgram(HadoopGraph.open(vertexProgramConfiguration), vertexProgramConfiguration).getMessageCombiner().orElse(null);
        final Function2<Payload, Payload, Payload> reducerFunction = (a, b) -> {      // reduce the view and outgoing messages into a single payload object representing the new view and incoming messages for a vertex
            if (a instanceof ViewIncomingPayload) {
                ((ViewIncomingPayload<M>) a).mergePayload(b, messageCombiner);
                return a;
            } else if (b instanceof ViewIncomingPayload) {
                ((ViewIncomingPayload<M>) b).mergePayload(a, messageCombiner);
                return b;
            } else {
                final ViewIncomingPayload<M> c = new ViewIncomingPayload<>(messageCombiner);
                c.mergePayload(a, messageCombiner);
                c.mergePayload(b, messageCombiner);
                return c;
            }
        };
        /////////////////////////////////////////////////////////////
        /////////////////////////////////////////////////////////////
        // "message pass" by reducing on the vertex object id of the view and message payloads
        final JavaPairRDD<Object, ViewIncomingPayload<M>> newViewIncomingRDD =
                (partitionedGraphRDD ?
                        viewOutgoingRDD.flatMapToPair(messageFunction).reduceByKey(graphRDD.partitioner().get(), reducerFunction) :
                        viewOutgoingRDD.flatMapToPair(messageFunction).reduceByKey(reducerFunction))
                        .mapValues(payload -> { // handle various corner cases of when views don't exist, messages don't exist, or neither exists.
                            if (payload instanceof ViewIncomingPayload) {// this happens if there is a vertex view with incoming messages
                                return (ViewIncomingPayload<M>) payload;}
                            else if (payload instanceof ViewPayload)  {  // this happens if there is a vertex view with no incoming messages
                                return new ViewIncomingPayload<>((ViewPayload) payload);}
                            else  {                                      // this happens when there is a single message to a vertex that has no view or outgoing messages
                                return new ViewIncomingPayload<>((MessagePayload<M>) payload);}
                        });
        // the graphRDD and the viewRDD must have the same partitioner
        if (partitionedGraphRDD){
            assert graphRDD.partitioner().get().equals(newViewIncomingRDD.partitioner().get());}
        newViewIncomingRDD
                .foreachPartition(partitionIterator -> {
                    KryoShimServiceLoader.applyConfiguration(graphComputerConfiguration);
                }); // need to complete a task so its BSP and the memory for this iteration is updated
        return newViewIncomingRDD;
    }
public static < M > JavaPairRDD < Object , ViewIncomingPayload < M > > executeVertexProgramIteration ( final JavaPairRDD < Object , VertexWritable > graphRDD , final JavaPairRDD < Object , ViewIncomingPayload < M > > viewIncomingRDD , final GraknSparkMemory memory , final Configuration graphComputerConfiguration , final Configuration vertexProgramConfiguration ) { boolean partitionedGraphRDD = graphRDD . partitioner ( ) . isPresent ( ) ; if ( partitionedGraphRDD && null != viewIncomingRDD ) { assert graphRDD . partitioner ( ) . get ( ) . equals ( viewIncomingRDD . partitioner ( ) . get ( ) ) ; } final JavaPairRDD < Object , ViewOutgoingPayload < M > > viewOutgoingRDD = ( ( null == viewIncomingRDD ) ? graphRDD . mapValues ( vertexWritable -> new Tuple2 < > ( vertexWritable , Optional . < ViewIncomingPayload < M > > absent ( ) ) ) : graphRDD . leftOuterJoin ( viewIncomingRDD ) ) . mapPartitionsToPair ( partitionIterator -> { KryoShimServiceLoader . applyConfiguration ( graphComputerConfiguration ) ; if ( ! partitionIterator . hasNext ( ) ) { return Collections . emptyIterator ( ) ; } final VertexProgram < M > workerVertexProgram = VertexProgram . createVertexProgram ( HadoopGraph . open ( graphComputerConfiguration ) , vertexProgramConfiguration ) ; final String [ ] vertexComputeKeysArray = VertexProgramHelper . vertexComputeKeysAsArray ( workerVertexProgram . getVertexComputeKeys ( ) ) ; final SparkMessenger < M > messenger = new SparkMessenger < > ( ) ; workerVertexProgram . workerIterationStart ( memory . asImmutable ( ) ) ; return IteratorUtils . map ( partitionIterator , vertexViewIncoming -> { final StarGraph . StarVertex vertex = vertexViewIncoming . _2 ( ) . _1 ( ) . get ( ) ; final boolean hasViewAndMessages = vertexViewIncoming . _2 ( ) . _2 ( ) . isPresent ( ) ; final List < DetachedVertexProperty < Object > > previousView = hasViewAndMessages ? vertexViewIncoming . _2 ( ) . _2 ( ) . get ( ) . getView ( ) : memory . isInitialIteration ( ) ? new ArrayList < > ( ) : Collections . emptyList ( ) ; if ( memory . isInitialIteration ( ) && vertexComputeKeysArray . length > 0 ) { vertex . properties ( vertexComputeKeysArray ) . forEachRemaining ( vertexProperty -> previousView . add ( DetachedFactory . detach ( vertexProperty , true ) ) ) ; } vertex . dropVertexProperties ( vertexComputeKeysArray ) ; final List < M > incomingMessages = hasViewAndMessages ? vertexViewIncoming . _2 ( ) . _2 ( ) . get ( ) . getIncomingMessages ( ) : Collections . emptyList ( ) ; IteratorUtils . removeOnNext ( previousView . iterator ( ) ) . forEachRemaining ( property -> property . attach ( Attachable . Method . create ( vertex ) ) ) ; assert previousView . isEmpty ( ) ; messenger . setVertexAndIncomingMessages ( vertex , incomingMessages ) ; workerVertexProgram . execute ( ComputerGraph . vertexProgram ( vertex , workerVertexProgram ) , messenger , memory ) ; incomingMessages . clear ( ) ; final List < DetachedVertexProperty < Object > > nextView = vertexComputeKeysArray . length == 0 ? Collections . emptyList ( ) : IteratorUtils . list ( IteratorUtils . map ( vertex . properties ( vertexComputeKeysArray ) , vertexProperty -> DetachedFactory . detach ( vertexProperty , true ) ) ) ; vertex . dropVertexProperties ( vertexComputeKeysArray ) ; final List < Tuple2 < Object , M > > outgoingMessages = messenger . getOutgoingMessages ( ) ; if ( ! partitionIterator . hasNext ( ) ) { workerVertexProgram . workerIterationEnd ( memory . asImmutable ( ) ) ; } return ( nextView . isEmpty ( ) && outgoingMessages . isEmpty ( ) ) ? null : new Tuple2 < > ( vertex . id ( ) , new ViewOutgoingPayload < > ( nextView , outgoingMessages ) ) ; } ) ; } , true ) . filter ( tuple -> null != tuple ) ; if ( partitionedGraphRDD ) { assert graphRDD . partitioner ( ) . get ( ) . equals ( viewOutgoingRDD . partitioner ( ) . get ( ) ) ; } final PairFlatMapFunction < Tuple2 < Object , ViewOutgoingPayload < M > > , Object , Payload > messageFunction = tuple -> IteratorUtils . concat ( IteratorUtils . of ( new Tuple2 < > ( tuple . _1 ( ) , tuple . _2 ( ) . getView ( ) ) ) , IteratorUtils . map ( tuple . _2 ( ) . getOutgoingMessages ( ) . iterator ( ) , message -> new Tuple2 < > ( message . _1 ( ) , new MessagePayload < > ( message . _2 ( ) ) ) ) ) ; final MessageCombiner < M > messageCombiner = VertexProgram . < VertexProgram < M > > createVertexProgram ( HadoopGraph . open ( vertexProgramConfiguration ) , vertexProgramConfiguration ) . getMessageCombiner ( ) . orElse ( null ) ; final Function2 < Payload , Payload , Payload > reducerFunction = ( a , b ) -> { if ( a instanceof ViewIncomingPayload ) { ( ( ViewIncomingPayload < M > ) a ) . mergePayload ( b , messageCombiner ) ; return a ; } else if ( b instanceof ViewIncomingPayload ) { ( ( ViewIncomingPayload < M > ) b ) . mergePayload ( a , messageCombiner ) ; return b ; } else { final ViewIncomingPayload < M > c = new ViewIncomingPayload < > ( messageCombiner ) ; c . mergePayload ( a , messageCombiner ) ; c . mergePayload ( b , messageCombiner ) ; return c ; } } ; final JavaPairRDD < Object , ViewIncomingPayload < M > > newViewIncomingRDD = ( partitionedGraphRDD ? viewOutgoingRDD . flatMapToPair ( messageFunction ) . reduceByKey ( graphRDD . partitioner ( ) . get ( ) , reducerFunction ) : viewOutgoingRDD . flatMapToPair ( messageFunction ) . reduceByKey ( reducerFunction ) ) . mapValues ( payload -> { if ( payload instanceof ViewIncomingPayload ) { return ( ViewIncomingPayload < M > ) payload ; } else if ( payload instanceof ViewPayload ) { return new ViewIncomingPayload < > ( ( ViewPayload ) payload ) ; } else { return new ViewIncomingPayload < > ( ( MessagePayload < M > ) payload ) ; } } ) ; if ( partitionedGraphRDD ) { assert graphRDD . partitioner ( ) . get ( ) . equals ( newViewIncomingRDD . partitioner ( ) . get ( ) ) ; } newViewIncomingRDD . foreachPartition ( partitionIterator -> { KryoShimServiceLoader . applyConfiguration ( graphComputerConfiguration ) ; } ) ; return newViewIncomingRDD ; }
//////////////////