/*
 * 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.graph;

import java.io.IOException;
import java.util.Iterator;

import org.apache.giraph.bsp.CentralizedServiceWorker;
import org.apache.giraph.comm.WorkerClientRequestProcessor;
import org.apache.giraph.edge.Edge;
import org.apache.giraph.edge.OutEdges;
import org.apache.giraph.worker.AllWorkersInfo;
import org.apache.giraph.worker.WorkerAggregatorDelegator;
import org.apache.giraph.worker.WorkerContext;
import org.apache.giraph.worker.WorkerGlobalCommUsage;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * See {@link Computation} for explanation of the interface.
 *
 * This is a abstract class helper for users to implement their computations.
 * It implements all of the methods required by the {@link Computation}
 * interface except for the {@link #compute(Vertex, Iterable)} which we leave
 * to the user to define.
 *
 * In most cases users should inherit from this class when implementing their
 * algorithms with Giraph.
 *
 * @param <I> Vertex id
 * @param <V> Vertex data
 * @param <E> Edge data
 * @param <M1> Incoming message type
 * @param <M2> Outgoing message type
 */
public abstract class AbstractComputation<I extends WritableComparable,
    V extends Writable, E extends Writable, M1 extends Writable,
    M2 extends Writable>
    extends WorkerAggregatorDelegator<I, V, E>
    implements Computation<I, V, E, M1, M2> {
  /** Global graph state **/
  private GraphState graphState;
  /** Handles requests */
  private WorkerClientRequestProcessor<I, V, E> workerClientRequestProcessor;
  /** Service worker */
  private CentralizedServiceWorker<I, V, E> serviceWorker;
  /** Worker context */
  private WorkerContext workerContext;
  /** All workers info */
  private AllWorkersInfo allWorkersInfo;

  /**
   * Must be defined by user to do computation on a single Vertex.
   *
   * @param vertex   Vertex
   * @param messages Messages that were sent to this vertex in the previous
   *                 superstep.  Each message is only guaranteed to have
   *                 a life expectancy as long as next() is not called.
   */
  @Override
  public abstract void compute(Vertex<I, V, E> vertex,
      Iterable<M1> messages) throws IOException;

  /**
   * Prepare for computation. This method is executed exactly once prior to
   * {@link #compute(Vertex, Iterable)} being called for any of the vertices
   * in the partition.
   */
  @Override
  public void preSuperstep() {
  }

  /**
   * Finish computation. This method is executed exactly once after computation
   * for all vertices in the partition is complete.
   */
  @Override
  public void postSuperstep() {
  }

  /**
   * Initialize, called by infrastructure before the superstep starts.
   * Shouldn't be called by user code.
   *
   * @param graphState Graph state
   * @param workerClientRequestProcessor Processor for handling requests
   * @param serviceWorker Graph-wide BSP Mapper for this Vertex
   * @param workerGlobalCommUsage Worker global communication usage
   */
  @Override
  public void initialize(
      GraphState graphState,
      WorkerClientRequestProcessor<I, V, E> workerClientRequestProcessor,
      CentralizedServiceWorker<I, V, E> serviceWorker,
      WorkerGlobalCommUsage workerGlobalCommUsage) {
    this.graphState = graphState;
    this.workerClientRequestProcessor = workerClientRequestProcessor;
    this.setWorkerGlobalCommUsage(workerGlobalCommUsage);
    this.serviceWorker = serviceWorker;
    if (serviceWorker != null) {
      this.workerContext = serviceWorker.getWorkerContext();
      this.allWorkersInfo = new AllWorkersInfo(
          serviceWorker.getWorkerInfoList(), serviceWorker.getWorkerInfo());
    } else {
      this.workerContext = null;
      this.allWorkersInfo = null;
    }
  }

  /**
   * Retrieves the current superstep.
   *
   * @return Current superstep
   */
  @Override
  public long getSuperstep() {
    return graphState.getSuperstep();
  }

  /**
   * Get the total (all workers) number of vertices that
   * existed in the previous superstep.
   *
   * @return Total number of vertices (-1 if first superstep)
   */
  @Override
  public long getTotalNumVertices() {
    return graphState.getTotalNumVertices();
  }

  /**
   * Get the total (all workers) number of edges that
   * existed in the previous superstep.
   *
   * @return Total number of edges (-1 if first superstep)
   */
  @Override
  public long getTotalNumEdges() {
    return graphState.getTotalNumEdges();
  }

  /**
   * Send a message to a vertex id.
   *
   * @param id Vertex id to send the message to
   * @param message Message data to send
   */
  @Override
  public void sendMessage(I id, M2 message) {
    workerClientRequestProcessor.sendMessageRequest(id, message);
  }

  /**
   * Send a message to all edges.
   *
   * @param vertex Vertex whose edges to send the message to.
   * @param message Message sent to all edges.
   */
  @Override
  public void sendMessageToAllEdges(Vertex<I, V, E> vertex, M2 message) {
    workerClientRequestProcessor.sendMessageToAllRequest(vertex, message);
  }

  /**
   * Send a message to multiple target vertex ids in the iterator.
   *
   * @param vertexIdIterator An iterator to multiple target vertex ids.
   * @param message Message sent to all targets in the iterator.
   */
  @Override
  public void sendMessageToMultipleEdges(
      Iterator<I> vertexIdIterator, M2 message) {
    workerClientRequestProcessor.sendMessageToAllRequest(
        vertexIdIterator, message);
  }

  /**
   * Sends a request to create a vertex that will be available during the
   * next superstep.
   *
   * @param id Vertex id
   * @param value Vertex value
   * @param edges Initial edges
   */
  @Override
  public void addVertexRequest(I id, V value,
      OutEdges<I, E> edges) throws IOException {
    Vertex<I, V, E> vertex = getConf().createVertex();
    vertex.initialize(id, value, edges);
    workerClientRequestProcessor.addVertexRequest(vertex);
  }

  /**
   * Sends a request to create a vertex that will be available during the
   * next superstep.
   *
   * @param id Vertex id
   * @param value Vertex value
   */
  @Override
  public void addVertexRequest(I id, V value) throws IOException {
    addVertexRequest(id, value, getConf().createAndInitializeOutEdges());
  }

  /**
   * Request to remove a vertex from the graph
   * (applied just prior to the next superstep).
   *
   * @param vertexId Id of the vertex to be removed.
   */
  @Override
  public void removeVertexRequest(I vertexId) throws IOException {
    workerClientRequestProcessor.removeVertexRequest(vertexId);
  }

  /**
   * Request to add an edge of a vertex in the graph
   * (processed just prior to the next superstep)
   *
   * @param sourceVertexId Source vertex id of edge
   * @param edge Edge to add
   */
  @Override
  public void addEdgeRequest(I sourceVertexId,
      Edge<I, E> edge) throws IOException {
    workerClientRequestProcessor.addEdgeRequest(sourceVertexId, edge);
  }

  /**
   * Request to remove all edges from a given source vertex to a given target
   * vertex (processed just prior to the next superstep).
   *
   * @param sourceVertexId Source vertex id
   * @param targetVertexId Target vertex id
   */
  @Override
  public void removeEdgesRequest(I sourceVertexId,
      I targetVertexId) throws IOException {
    workerClientRequestProcessor.removeEdgesRequest(
        sourceVertexId, targetVertexId);
  }

  /**
   * Get the mapper context
   *
   * @return Mapper context
   */
  @Override
  public Mapper.Context getContext() {
    return graphState.getContext();
  }

  /**
   * Get the worker context
   *
   * @param <W> WorkerContext class
   * @return WorkerContext context
   */
  @SuppressWarnings("unchecked")
  @Override
  public <W extends WorkerContext> W getWorkerContext() {
    return (W) workerContext;
  }

  @Override
  public final int getWorkerCount() {
    return allWorkersInfo.getWorkerCount();
  }

  @Override
  public final int getMyWorkerIndex() {
    return allWorkersInfo.getMyWorkerIndex();
  }

  @Override
  public final int getWorkerForVertex(I vertexId) {
    return allWorkersInfo.getWorkerIndex(
        serviceWorker.getVertexPartitionOwner(vertexId).getWorkerInfo());
  }
}