/*
    * 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.bsp;
  
  import java.io.IOException;
  import java.util.Collection;
  import java.util.List;
  
  import org.apache.giraph.comm.ServerData;
  import org.apache.giraph.comm.WorkerClient;
  import org.apache.giraph.comm.messages.PartitionSplitInfo;
  import org.apache.giraph.graph.AddressesAndPartitionsWritable;
  import org.apache.giraph.graph.FinishedSuperstepStats;
  import org.apache.giraph.graph.GlobalStats;
  import org.apache.giraph.graph.GraphTaskManager;
  import org.apache.giraph.graph.VertexEdgeCount;
  import org.apache.giraph.io.superstep_output.SuperstepOutput;
  import org.apache.giraph.metrics.GiraphTimerContext;
  import org.apache.giraph.partition.PartitionOwner;
  import org.apache.giraph.partition.PartitionStats;
  import org.apache.giraph.partition.PartitionStore;
  import org.apache.giraph.worker.WorkerAggregatorHandler;
  import org.apache.giraph.worker.WorkerContext;
  import org.apache.giraph.worker.WorkerInfo;
  import org.apache.giraph.worker.WorkerInputSplitsHandler;
  import org.apache.giraph.worker.WorkerObserver;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  
  /**
   * All workers should have access to this centralized service to
   * execute the following methods.
   *
   * @param <I> Vertex id
   * @param <V> Vertex value
   * @param <E> Edge value
   */
  @SuppressWarnings("rawtypes")
  public interface CentralizedServiceWorker<I extends WritableComparable,
    V extends Writable, E extends Writable>
    extends CentralizedService<I, V, E>, PartitionSplitInfo<I> {
    /**
     * Setup (must be called prior to any other function)
     *
     * @return Finished superstep stats for the input superstep
     */
    FinishedSuperstepStats setup();
  
    /**
     * Get the worker information
     *
     * @return Worker information
     */
    WorkerInfo getWorkerInfo();
  
    /**
     * Get the worker client (for instantiating WorkerClientRequestProcessor
     * instances.
     *
     * @return Worker client
     */
    WorkerClient<I, V, E> getWorkerClient();
  
    /**
     * Get the worker context.
     *
     * @return worker's WorkerContext
     */
    WorkerContext getWorkerContext();
  
    /**
     * Get the observers for this Worker.
     *
     * @return array of WorkerObservers.
     */
    WorkerObserver[] getWorkerObservers();
  
    /**
     * Get the partition store for this worker.
     * The partitions contain the vertices for
     * this worker and can be used to run compute() for the vertices or do
     * checkpointing.
     *
    * @return The partition store for this worker.
    */
   PartitionStore<I, V, E> getPartitionStore();
 
   /**
    *  Both the vertices and the messages need to be checkpointed in order
    *  for them to be used.  This is done after all messages have been
    *  delivered, but prior to a superstep starting.
    */
   void storeCheckpoint() throws IOException;
 
   /**
    * Load the vertices, edges, messages from the beginning of a superstep.
    * Will load the vertex partitions as designated by the master and set the
    * appropriate superstep.
    *
    * @param superstep which checkpoint to use
    * @return Graph-wide vertex and edge counts
    * @throws IOException
    */
   VertexEdgeCount loadCheckpoint(long superstep) throws IOException;
 
   /**
    * Take all steps prior to actually beginning the computation of a
    * superstep.
    *
    * @return Collection of all the partition owners from the master for this
    *         superstep.
    */
   Collection<? extends PartitionOwner> startSuperstep();
 
   /**
    * Worker is done with its portion of the superstep.  Report the
    * worker level statistics after the computation.
    *
    * @param partitionStatsList All the partition stats for this worker
    * @param superstepTimerContext superstep timer context only given when the
    *      function needs to stop the timer, otherwise null.
    * @return Stats of the superstep completion
    */
   FinishedSuperstepStats finishSuperstep(
       List<PartitionStats> partitionStatsList,
       GiraphTimerContext superstepTimerContext);
 
   /**
    * Get the partition id that a vertex id would belong to.
    *
    * @param vertexId Vertex id
    * @return Partition id
    */
   @Override
   int getPartitionId(I vertexId);
 
   /**
    * Whether a partition with given id exists on this worker.
    *
    * @param partitionId Partition id
    * @return True iff this worker has the specified partition
    */
   boolean hasPartition(Integer partitionId);
 
   /**
    * Every client will need to get a partition owner from a vertex id so that
    * they know which worker to sent the request to.
    *
    * @param vertexId Vertex index to look for
    * @return PartitionOnwer that should contain this vertex if it exists
    */
   PartitionOwner getVertexPartitionOwner(I vertexId);
 
   /**
    * Get all partition owners.
    *
    * @return Iterable through partition owners
    */
   Iterable<? extends PartitionOwner> getPartitionOwners();
 
   /**
    * If desired by the user, vertex partitions are redistributed among
    * workers according to the chosen WorkerGraphPartitioner.
    *
    * @param masterSetPartitionOwners Partition owner info passed from the
    *        master.
    */
   void exchangeVertexPartitions(
       Collection<? extends PartitionOwner> masterSetPartitionOwners);
 
   /**
    * Get the GraphTaskManager that this service is using.  Vertices need to know
    * this.
    *
    * @return the GraphTaskManager instance for this compute node
    */
   GraphTaskManager<I, V, E> getGraphTaskManager();
 
   /**
    * Operations that will be called if there is a failure by a worker.
    */
   void failureCleanup();
 
   /**
    * Get server data
    *
    * @return Server data
    */
   ServerData<I, V, E> getServerData();
 
   /**
    * Get worker aggregator handler
    *
    * @return Worker aggregator handler
    */
   WorkerAggregatorHandler getAggregatorHandler();
 
   /**
    * Final preparation for superstep, called after startSuperstep and
    * potential loading from checkpoint, right before the computation started
    * TODO how to avoid this additional function
    */
   void prepareSuperstep();
 
   /**
    * Get the superstep output class
    *
    * @return SuperstepOutput
    */
   SuperstepOutput<I, V, E> getSuperstepOutput();
 
   /**
    * Clean up the service (no calls may be issued after this)
    *
    * @param finishedSuperstepStats Finished supestep stats
    * @throws IOException
    * @throws InterruptedException
    */
   void cleanup(FinishedSuperstepStats finishedSuperstepStats)
     throws IOException, InterruptedException;
 
   /**
    * Loads Global stats from zookeeper.
    * @return global stats stored in zookeeper for
    * previous superstep.
    */
   GlobalStats getGlobalStats();
 
   /**
    * Get input splits handler used during input
    *
    * @return Input splits handler
    */
   WorkerInputSplitsHandler getInputSplitsHandler();
 
   /**
    * Received addresses and partitions assignments from master.
    *
    * @param addressesAndPartitions Addresses and partitions assignment
    */
   void addressesAndPartitionsReceived(
       AddressesAndPartitionsWritable addressesAndPartitions);
 
   /**
    * Store the counter values in the zookeeper after every superstep
    * and also after all supersteps are done. This is called before closing
    * the zookeeper. We need to call this method after calling cleanup on the
    * worker, since some counters are updated during cleanup
    * @param allSuperstepsDone boolean value whether all the supersteps
    *                          are completed
    */
   void storeCountersInZooKeeper(boolean allSuperstepsDone);
 
   /**
    * Close zookeeper
    */
   void closeZooKeeper();
 }