giraph-block-app/src/main/java/org/apache/giraph/block_app/framework/piece/AbstractPiece.java

/*
 * 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.giraph.block_app.framework.piece;

import java.util.Iterator;
import java.util.List;

import org.apache.giraph.block_app.framework.api.BlockMasterApi;
import org.apache.giraph.block_app.framework.api.BlockWorkerContextReceiveApi;
import org.apache.giraph.block_app.framework.api.BlockWorkerContextSendApi;
import org.apache.giraph.block_app.framework.api.BlockWorkerReceiveApi;
import org.apache.giraph.block_app.framework.api.BlockWorkerSendApi;
import org.apache.giraph.block_app.framework.block.Block;
import org.apache.giraph.block_app.framework.block.PieceCount;
import org.apache.giraph.block_app.framework.piece.interfaces.VertexPostprocessor;
import org.apache.giraph.block_app.framework.piece.interfaces.VertexReceiver;
import org.apache.giraph.block_app.framework.piece.interfaces.VertexSender;
import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
import org.apache.giraph.conf.MessageClasses;
import org.apache.giraph.function.Consumer;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;

import com.google.common.collect.Iterators;

/**
 * Parent of all Pieces, contains comprehensive list of methods Piece
 * can support. Specific subclasses should be extended directly,
 * to simplify usage - most frequently for example Piece class.
 *
 * Single unit of execution, capturing:
 * - sending and then receiving messages from vertices
 * - sending data to be aggregated from workers to master
 * - sending values from master, via aggregators, to workers
 * - sending and receiving worker messages
 *
 *
 * Order of execution is:
 * - On master, once at the start of the application
 * -- registerAggregators (deprecated, use registerReducers instead)
 *
 * - After masterCompute of previous piece, on master:
 * -- registerReducers
 *
 * - Send logic on workers:
 * -- getVertexSender per each worker thread, and on object returned by it:
 * --- vertexSend on each vertex
 * --- postprocess on each worker thread
 * -- workerContextSend per worker
 *
 * - Logic on master:
 * -- masterCompute
 *
 * - Receive logic on workers:
 * -- workerContextReceive per worker
 * -- getVertexReceiver per each worker thread, and on object returned by it:
 * --- vertexReceive on each vertex
 * --- postprocess on each worker thread
 *
 * And before everything, during initialization, registerAggregators.
 *
 * Only masterCompute and registerReducers/registerAggregators should modify
 * the Piece, all of the worker methods should treat Piece as read-only.
 *
 * Each piece should be encapsulated unit of execution. Vertex value should be
 * used as a single implicit "communication" channel between different pieces,
 * all other dependencies should be explicitly defined and passed through
 * constructor, via interfaces (as explained below).
 * I.e. state of the vertex value is invariant that Pieces act upon.
 * Best is not to depend on explicit vertex value class, but on interface that
 * provides all needed functions, so that pieces can be freely combined,
 * as long as vertex value implements appropriate ones.
 * Similarly, use most abstract class you need - if Piece doesn't depend
 * on edge value, don't use NullWritable, but Writable. Or if it doesn't
 * depend on ExecutionStage, use Object for it.
 *
 * All other external dependencies should be explicitly passed through
 * constructor, through interfaces.
 *
 * All Pieces will be created within one context - on the master.
 * They are then going to be replicated across all workers, and across all
 * threads within each worker, and will see everything that happens in global
 * context (masterCompute) before them, including any state master has.
 * Through ObjectHolder/ObjectTransfer, you can pass data between Pieces in
 * global context, and from global context to worker functions of a Piece
 * that happens in the future.
 *
 * VertexReceiver of previous Piece and VertexSender of next Piece live in
 * the same context, and vertexReceive of the next Piece is executed
 * immediately after vertexSend of the previous piece, before vertexSend is
 * called on the next vertex.
 * This detail allows you to have external dependency on each other through
 * memory only mediator objects - like ObjectTransfer.
 *
 * All other logic going to live in different contexts,
 * specifically VertexSender and VertexReceiver of the same Piece,
 * or workerContextSend and VertexSender of the same Piece, and cannot interact
 * with each other outside of changing the state of the graph or using
 * global communication api.
 *
 * All methods on this class (or objects it returns) will be called serially,
 * so there is no need for any Thread synchronization.
 * Each Thread will have a complete deep copy of the Piece, to achieve that,
 * so all static fields must be written to be Thread safe!
 * (i.e. either immutable, or have synchronized/locked access to them)
 *
 * @param <I> Vertex id type
 * @param <V> Vertex value type
 * @param <E> Edge value type
 * @param <M> Message type
 * @param <WV> Worker value type
 * @param <WM> Worker message type
 * @param <S> Execution stage type
 */
@SuppressWarnings({ "rawtypes" })
public abstract class AbstractPiece<I extends WritableComparable,
    V extends Writable, E extends Writable, M extends Writable, WV,
    WM extends Writable, S> implements Block {

  // Overridable functions

  // registerReducers(CreateReducersApi reduceApi, S executionStage)

  /**
   * Add automatic handling of reducers to registerReducers.
   * Only for internal use.
   */
  public abstract void wrappedRegisterReducers(
      BlockMasterApi masterApi, S executionStage);

  // getVertexSender(BlockWorkerSendApi<I, V, E, M> workerApi, S executionStage)

  /**
   * Add automatic handling of reducers to getVertexSender.
   *
   * Only for Framework internal use.
   */
  public abstract InnerVertexSender getWrappedVertexSender(
      final BlockWorkerSendApi<I, V, E, M> workerApi, S executionStage);

  /**
   * Override to have worker context send computation.
   *
   * Called once per worker, after all vertices have been processed with
   * getVertexSender.
   */
  public void workerContextSend(
      BlockWorkerContextSendApi<I, WM> workerContextApi, S executionStage,
      WV workerValue) {
  }

  /**
   * Function that is called on master, after send phase, before receive phase.
   *
   * It can:
   * - read aggregators sent from worker
   * - do global processing
   * - send data to workers through aggregators
   */
  public void masterCompute(BlockMasterApi masterApi, S executionStage) {
  }

  /**
   * Override to have worker context receive computation.
   *
   * Called once per worker, before all vertices are going to be processed
   * with getVertexReceiver.
   */
  public void workerContextReceive(
      BlockWorkerContextReceiveApi workerContextApi, S executionStage,
      WV workerValue, List<WM> workerMessages) {
  }

  /**
   * Override to do vertex receive processing.
   *
   * Creates handler that defines what should be executed on worker
   * for each vertex during receive phase.
   *
   * This logic executed last.
   * This function is called once on each worker on each thread, in parallel,
   * on their copy of Piece object to create functions handler.
   *
   * If returned object implements Postprocessor interface, then corresponding
   * postprocess() function is going to be called once, after all vertices
   * corresponding thread needed to process are done.
   */
  public VertexReceiver<I, V, E, M> getVertexReceiver(
      BlockWorkerReceiveApi<I> workerApi, S executionStage) {
    return null;
  }

  /**
   * Returns MessageClasses definition for messages being sent by this Piece.
   */
  public abstract MessageClasses<I, M> getMessageClasses(
      ImmutableClassesGiraphConfiguration conf);

  /**
   * Override to provide different next execution stage for
   * Pieces that come after it.
   *
   * Execution stage should be immutable, and this function should be
   * returning a new object, if it needs to return different value.
   *
   * It affects pieces that come after this piece,
   * and isn't applied to execution stage this piece sees.
   */
  public S nextExecutionStage(S executionStage) {
    return executionStage;
  }

  /**
   * Override to register any potential aggregators used by this piece.
   *
   * @deprecated Use registerReducers instead.
   */
  @Deprecated
  public void registerAggregators(BlockMasterApi masterApi)
      throws InstantiationException, IllegalAccessException {
  }

  // Inner classes

  /** Inner class to provide clean use without specifying types */
  public abstract class InnerVertexSender
      implements VertexSender<I, V, E>, VertexPostprocessor {
    @Override
    public void postprocess() { }
  }

  /** Inner class to provide clean use without specifying types */
  public abstract class InnerVertexReceiver
      implements VertexReceiver<I, V, E, M>, VertexPostprocessor {
    @Override
    public void postprocess() { }
  }

  // Internal implementation

  @Override
  public final Iterator<AbstractPiece> iterator() {
    return Iterators.<AbstractPiece>singletonIterator(this);
  }

  @Override
  public void forAllPossiblePieces(Consumer<AbstractPiece> consumer) {
    consumer.apply(this);
  }

  @Override
  public PieceCount getPieceCount() {
    return new PieceCount(1);
  }

  @Override
  public String toString() {
    String name = getClass().getSimpleName();
    if (name.isEmpty()) {
      name = getClass().getName();
    }
    return name;
  }


  // make hashCode and equals final, forcing them to be based on
  // reference identity.
  @Override
  public final int hashCode() {
    return super.hashCode();
  }

  @Override
  public final boolean equals(Object obj) {
    return super.equals(obj);
  }

}