/*
    * 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.lang.management.GarbageCollectorMXBean;
  import java.lang.management.ManagementFactory;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Enumeration;
  import java.util.List;
  import java.util.concurrent.Callable;
  import java.util.concurrent.TimeUnit;
  
  import com.sun.management.GarbageCollectionNotificationInfo;
  import com.yammer.metrics.core.Counter;
  
  import org.apache.commons.lang3.exception.ExceptionUtils;
  import org.apache.giraph.bsp.BspService;
  import org.apache.giraph.bsp.CentralizedServiceMaster;
  import org.apache.giraph.bsp.CentralizedServiceWorker;
  import org.apache.giraph.bsp.checkpoints.CheckpointStatus;
  import org.apache.giraph.comm.messages.MessageStore;
  import org.apache.giraph.conf.ClassConfOption;
  import org.apache.giraph.conf.GiraphConstants;
  import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
  import org.apache.giraph.job.JobProgressTracker;
  import org.apache.giraph.master.BspServiceMaster;
  import org.apache.giraph.master.MasterThread;
  import org.apache.giraph.metrics.GiraphMetrics;
  import org.apache.giraph.metrics.GiraphMetricsRegistry;
  import org.apache.giraph.metrics.GiraphTimer;
  import org.apache.giraph.metrics.GiraphTimerContext;
  import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;
  import org.apache.giraph.metrics.SuperstepMetricsRegistry;
  import org.apache.giraph.ooc.OutOfCoreEngine;
  import org.apache.giraph.partition.PartitionOwner;
  import org.apache.giraph.partition.PartitionStats;
  import org.apache.giraph.partition.PartitionStore;
  import org.apache.giraph.scripting.ScriptLoader;
  import org.apache.giraph.utils.CallableFactory;
  import org.apache.giraph.utils.GcObserver;
  import org.apache.giraph.utils.MemoryUtils;
  import org.apache.giraph.utils.ProgressableUtils;
  import org.apache.giraph.worker.BspServiceWorker;
  import org.apache.giraph.worker.InputSplitsCallable;
  import org.apache.giraph.worker.WorkerContext;
  import org.apache.giraph.worker.WorkerObserver;
  import org.apache.giraph.worker.WorkerProgress;
  import org.apache.giraph.writable.kryo.KryoWritableWrapper;
  import org.apache.giraph.zk.ZooKeeperManager;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.io.Writable;
  import org.apache.hadoop.io.WritableComparable;
  import org.apache.hadoop.mapreduce.Mapper;
  import org.apache.log4j.Appender;
  import org.apache.log4j.Level;
  import org.apache.log4j.LogManager;
  import org.apache.log4j.Logger;
  import org.apache.log4j.PatternLayout;
  
  import javax.management.Notification;
  import javax.management.NotificationEmitter;
  import javax.management.NotificationListener;
  import javax.management.openmbean.CompositeData;
  
  /**
   * The Giraph-specific business logic for a single BSP
   * compute node in whatever underlying type of cluster
   * our Giraph job will run on. Owning object will provide
   * the glue into the underlying cluster framework
   * and will call this object to perform Giraph work.
   *
   * @param <I> Vertex id
   * @param <V> Vertex data
   * @param <E> Edge data
   */
  @SuppressWarnings("rawtypes")
  public class GraphTaskManager<I extends WritableComparable, V extends Writable,
    E extends Writable> implements
    ResetSuperstepMetricsObserver {
  /*if_not[PURE_YARN]
   static { // Eliminate this? Even MRv1 tasks should not need it here.
     Configuration.addDefaultResource("giraph-site.xml");
   }
 end[PURE_YARN]*/
   /**
    * Class which checks if an exception on some thread should cause worker
    * to fail
    */
   public static final ClassConfOption<CheckerIfWorkerShouldFailAfterException>
   CHECKER_IF_WORKER_SHOULD_FAIL_AFTER_EXCEPTION_CLASS = ClassConfOption.create(
       "giraph.checkerIfWorkerShouldFailAfterExceptionClass",
       FailWithEveryExceptionOccurred.class,
       CheckerIfWorkerShouldFailAfterException.class,
       "Class which checks if an exception on some thread should cause worker " +
           "to fail, by default all exceptions cause failure");
   /** Name of metric for superstep time in msec */
   public static final String TIMER_SUPERSTEP_TIME = "superstep-time-ms";
   /** Name of metric for compute on all vertices in msec */
   public static final String TIMER_COMPUTE_ALL = "compute-all-ms";
   /** Name of metric for time from begin compute to first message sent */
   public static final String TIMER_TIME_TO_FIRST_MSG =
       "time-to-first-message-ms";
   /** Name of metric for time from first message till last message flushed */
   public static final String TIMER_COMMUNICATION_TIME = "communication-time-ms";
   /** Name of metric for time spent doing GC per superstep in msec */
   public static final String TIMER_SUPERSTEP_GC_TIME = "superstep-gc-time-ms";
 
   /** Class logger */
   private static final Logger LOG = Logger.getLogger(GraphTaskManager.class);
   /** Coordination service worker */
   private CentralizedServiceWorker<I, V, E> serviceWorker;
   /** Coordination service master */
   private CentralizedServiceMaster<I, V, E> serviceMaster;
   /** Coordination service master thread */
   private Thread masterThread = null;
   /** The worker should be run exactly once, or else there is a problem. */
   private boolean alreadyRun = false;
   /** Manages the ZooKeeper servers if necessary (dynamic startup) */
   private ZooKeeperManager zkManager;
   /** Configuration */
   private ImmutableClassesGiraphConfiguration<I, V, E> conf;
   /** Already complete? */
   private boolean done = false;
   /** What kind of functions is this mapper doing? */
   private GraphFunctions graphFunctions = GraphFunctions.UNKNOWN;
   /** Superstep stats */
   private FinishedSuperstepStats finishedSuperstepStats =
       new FinishedSuperstepStats(0, false, 0, 0, false, CheckpointStatus.NONE);
   /** Job progress tracker */
   private JobProgressTrackerClient jobProgressTracker;
 
   // Per-Job Metrics
   /** Timer for WorkerContext#preApplication() */
   private GiraphTimer wcPreAppTimer;
   /** Timer for WorkerContext#postApplication() */
   private GiraphTimer wcPostAppTimer;
 
   // Per-Superstep Metrics
   /** Time for how long superstep took */
   private GiraphTimer superstepTimer;
   /** Time for all compute() calls in a superstep */
   private GiraphTimer computeAll;
   /** Time from starting compute to sending first message */
   private GiraphTimer timeToFirstMessage;
   /** Context for timing time to first message above */
   private GiraphTimerContext timeToFirstMessageTimerContext;
   /** Time from first sent message till last message flushed. */
   private GiraphTimer communicationTimer;
   /** Context for timing communication time above */
   private GiraphTimerContext communicationTimerContext;
   /** Timer for WorkerContext#preSuperstep() */
   private GiraphTimer wcPreSuperstepTimer;
   /** Timer to keep aggregated time spent in GC in a superstep */
   private Counter gcTimeMetric;
   /** The Hadoop Mapper#Context for this job */
   private final Mapper<?, ?, ?, ?>.Context context;
   /** is this GraphTaskManager the master? */
   private boolean isMaster;
   /** Mapper observers */
   private MapperObserver[] mapperObservers;
 
   /**
    * Default constructor for GiraphTaskManager.
    * @param context a handle to the underlying cluster framework.
    *                For Hadoop clusters, this is a Mapper#Context.
    */
   public GraphTaskManager(Mapper<?, ?, ?, ?>.Context context) {
     this.context = context;
     this.isMaster = false;
   }
 
   /**
    * Run the user's input checking code.
    */
   private void checkInput() {
     if (conf.hasEdgeInputFormat()) {
       conf.createWrappedEdgeInputFormat().checkInputSpecs(conf);
     }
     if (conf.hasVertexInputFormat()) {
       conf.createWrappedVertexInputFormat().checkInputSpecs(conf);
     }
   }
 
   /**
    * In order for job client to know which ZooKeeper the job is using,
    * we create a counter with server:port as its name inside of
    * ZOOKEEPER_SERVER_PORT_COUNTER_GROUP.
    *
    * @param serverPortList Server:port list for ZooKeeper used
    */
   private void createZooKeeperCounter(String serverPortList) {
     // Getting the counter will actually create it.
     context.getCounter(GiraphConstants.ZOOKEEPER_SERVER_PORT_COUNTER_GROUP,
         serverPortList);
   }
 
   /**
    * Called by owner of this GraphTaskManager on each compute node
    *
    * @param zkPathList the path to the ZK jars we need to run the job
    */
   public void setup(Path[] zkPathList)
     throws IOException, InterruptedException {
     context.setStatus("setup: Beginning worker setup.");
     Configuration hadoopConf = context.getConfiguration();
     conf = new ImmutableClassesGiraphConfiguration<I, V, E>(hadoopConf);
     initializeJobProgressTracker();
     // Setting the default handler for uncaught exceptions.
     Thread.setDefaultUncaughtExceptionHandler(createUncaughtExceptionHandler());
     setupMapperObservers();
     // Write user's graph types (I,V,E,M) back to configuration parameters so
     // that they are set for quicker access later. These types are often
     // inferred from the Computation class used.
     conf.getGiraphTypes().writeIfUnset(conf);
     // configure global logging level for Giraph job
     initializeAndConfigureLogging();
     // init the metrics objects
     setupAndInitializeGiraphMetrics();
     // Check input
     checkInput();
     // Load any scripts that were deployed
     ScriptLoader.loadScripts(conf);
     // One time setup for computation factory
     conf.createComputationFactory().initialize(conf);
     // Do some task setup (possibly starting up a Zookeeper service)
     context.setStatus("setup: Initializing Zookeeper services.");
     String serverPortList = conf.getZookeeperList();
     if (serverPortList.isEmpty()) {
       if (startZooKeeperManager()) {
         return; // ZK connect/startup failed
       }
     } else {
       createZooKeeperCounter(serverPortList);
     }
     if (zkManager != null && zkManager.runsZooKeeper()) {
       if (LOG.isInfoEnabled()) {
         LOG.info("setup: Chosen to run ZooKeeper...");
       }
     }
     context
         .setStatus("setup: Connected to Zookeeper service " + serverPortList);
     this.graphFunctions = determineGraphFunctions(conf, zkManager);
     if (zkManager != null && this.graphFunctions.isMaster()) {
       zkManager.cleanupOnExit();
     }
     try {
       instantiateBspService();
     } catch (IOException e) {
       LOG.error("setup: Caught exception just before end of setup", e);
       if (zkManager != null) {
         zkManager.offlineZooKeeperServers(ZooKeeperManager.State.FAILED);
       }
       throw new RuntimeException(
         "setup: Offlining servers due to exception...", e);
     }
     context.setStatus(getGraphFunctions().toString() + " starting...");
   }
 
   /**
    * Create and connect a client to JobProgressTrackerService,
    * or no-op implementation if progress shouldn't be tracked or something
    * goes wrong
    */
   private void initializeJobProgressTracker() {
     if (!conf.trackJobProgressOnClient()) {
       jobProgressTracker = new JobProgressTrackerClientNoOp();
     } else {
       jobProgressTracker =
         GiraphConstants.JOB_PROGRESS_TRACKER_CLIENT_CLASS.newInstance(conf);
       try {
         jobProgressTracker.init(conf);
         // CHECKSTYLE: stop IllegalCatch
       } catch (Exception e) {
         // CHECKSTYLE: resume IllegalCatch
         throw new RuntimeException(
           "Failed to initialize JobProgressTrackerClient", e);
       }
     }
     jobProgressTracker.mapperStarted();
   }
 
   /**
   * Perform the work assigned to this compute node for this job run.
   * 1) Run checkpoint per frequency policy.
   * 2) For every vertex on this mapper, run the compute() function
   * 3) Wait until all messaging is done.
   * 4) Check if all vertices are done.  If not goto 2).
   * 5) Dump output.
   */
   public void execute() throws IOException, InterruptedException {
     if (checkTaskState()) {
       return;
     }
     preLoadOnWorkerObservers();
     GiraphTimerContext superstepTimerContext = superstepTimer.time();
     finishedSuperstepStats = serviceWorker.setup();
     superstepTimerContext.stop();
     if (collectInputSuperstepStats(finishedSuperstepStats)) {
       return;
     }
     prepareGraphStateAndWorkerContext();
     List<PartitionStats> partitionStatsList = new ArrayList<PartitionStats>();
     int numComputeThreads = conf.getNumComputeThreads();
 
     // main superstep processing loop
     while (!finishedSuperstepStats.allVerticesHalted()) {
       final long superstep = serviceWorker.getSuperstep();
       superstepTimerContext = getTimerForThisSuperstep(superstep);
       GraphState graphState = new GraphState(superstep,
           finishedSuperstepStats.getVertexCount(),
           finishedSuperstepStats.getEdgeCount(),
           context);
       Collection<? extends PartitionOwner> masterAssignedPartitionOwners =
         serviceWorker.startSuperstep();
       if (LOG.isDebugEnabled()) {
         LOG.debug("execute: " + MemoryUtils.getRuntimeMemoryStats());
       }
       context.progress();
       serviceWorker.exchangeVertexPartitions(masterAssignedPartitionOwners);
       context.progress();
       boolean hasBeenRestarted = checkSuperstepRestarted(superstep);
 
       GlobalStats globalStats = serviceWorker.getGlobalStats();
 
       if (hasBeenRestarted) {
         graphState = new GraphState(superstep,
             finishedSuperstepStats.getVertexCount(),
             finishedSuperstepStats.getEdgeCount(),
             context);
       } else if (storeCheckpoint(globalStats.getCheckpointStatus())) {
         break;
       }
       serviceWorker.getServerData().prepareResolveMutations();
       context.progress();
       prepareForSuperstep(graphState);
       context.progress();
       MessageStore<I, Writable> messageStore =
           serviceWorker.getServerData().getCurrentMessageStore();
       int numPartitions = serviceWorker.getPartitionStore().getNumPartitions();
       int numThreads = Math.min(numComputeThreads, numPartitions);
       if (LOG.isInfoEnabled()) {
         LOG.info("execute: " + numPartitions + " partitions to process with " +
           numThreads + " compute thread(s), originally " +
           numComputeThreads + " thread(s) on superstep " + superstep);
       }
       partitionStatsList.clear();
       // execute the current superstep
       if (numPartitions > 0) {
         processGraphPartitions(context, partitionStatsList, graphState,
           messageStore, numThreads);
       }
       finishedSuperstepStats = completeSuperstepAndCollectStats(
         partitionStatsList, superstepTimerContext);
 
       // END of superstep compute loop
     }
 
     if (LOG.isInfoEnabled()) {
       LOG.info("execute: BSP application done (global vertices marked done)");
     }
     updateSuperstepGraphState();
     postApplication();
   }
 
   /**
    * Handle post-application callbacks.
    */
   private void postApplication() throws IOException, InterruptedException {
     GiraphTimerContext postAppTimerContext = wcPostAppTimer.time();
     serviceWorker.getWorkerContext().postApplication();
     serviceWorker.getSuperstepOutput().postApplication();
     postAppTimerContext.stop();
     context.progress();
 
     for (WorkerObserver obs : serviceWorker.getWorkerObservers()) {
       obs.postApplication();
       context.progress();
     }
   }
 
   /**
    * Sets the "isMaster" flag for final output commit to happen on master.
    * @param im the boolean input to set isMaster. Applies to "pure YARN only"
    */
   public void setIsMaster(final boolean im) {
     this.isMaster = im;
   }
 
   /**
    * Get "isMaster" status flag -- we need to know if we're the master in the
    * "finally" block of our GiraphYarnTask#execute() to commit final job output.
    * @return true if this task IS the master.
    */
   public boolean isMaster() {
     return isMaster;
   }
 
   /**
    * Produce a reference to the "start" superstep timer for the current
    * superstep.
    * @param superstep the current superstep count
    * @return a GiraphTimerContext representing the "start" of the supestep
    */
   private GiraphTimerContext getTimerForThisSuperstep(long superstep) {
     GiraphMetrics.get().resetSuperstepMetrics(superstep);
     return superstepTimer.time();
   }
 
   /**
    * Utility to encapsulate Giraph metrics setup calls
    */
   private void setupAndInitializeGiraphMetrics() {
     GiraphMetrics.init(conf);
     GiraphMetrics.get().addSuperstepResetObserver(this);
     initJobMetrics();
     MemoryUtils.initMetrics();
     InputSplitsCallable.initMetrics();
   }
 
   /**
    * Instantiate and configure ZooKeeperManager for this job. This will
    * result in a Giraph-owned Zookeeper instance, a connection to an
    * existing quorum as specified in the job configuration, or task failure
    * @return true if this task should terminate
    */
   private boolean startZooKeeperManager()
     throws IOException, InterruptedException {
     zkManager = new ZooKeeperManager(context, conf);
     context.setStatus("setup: Setting up Zookeeper manager.");
     zkManager.setup();
     if (zkManager.computationDone()) {
       done = true;
       return true;
     }
     zkManager.onlineZooKeeperServer();
     String serverPortList = zkManager.getZooKeeperServerPortString();
     conf.setZookeeperList(serverPortList);
     createZooKeeperCounter(serverPortList);
     return false;
   }
 
   /**
    * Utility to place a new, updated GraphState object into the serviceWorker.
    */
   private void updateSuperstepGraphState() {
     serviceWorker.getWorkerContext().setGraphState(
         new GraphState(serviceWorker.getSuperstep(),
             finishedSuperstepStats.getVertexCount(),
             finishedSuperstepStats.getEdgeCount(), context));
   }
 
   /**
    * Utility function for boilerplate updates and cleanup done at the
    * end of each superstep processing loop in the <code>execute</code> method.
    * @param partitionStatsList list of stas for each superstep to append to
    * @param superstepTimerContext for job metrics
    * @return the collected stats at the close of the current superstep.
    */
   private FinishedSuperstepStats completeSuperstepAndCollectStats(
     List<PartitionStats> partitionStatsList,
     GiraphTimerContext superstepTimerContext) {
 
     // the superstep timer is stopped inside the finishSuperstep function
     // (otherwise metrics are not available at the end of the computation
     //  using giraph.metrics.enable=true).
     finishedSuperstepStats =
       serviceWorker.finishSuperstep(partitionStatsList, superstepTimerContext);
     if (conf.metricsEnabled()) {
       GiraphMetrics.get().perSuperstep().printSummary(System.err);
     }
     return finishedSuperstepStats;
   }
 
   /**
    * Utility function to prepare various objects managing BSP superstep
    * operations for the next superstep.
    * @param graphState graph state metadata object
    */
   private void prepareForSuperstep(GraphState graphState) {
     serviceWorker.prepareSuperstep();
 
     serviceWorker.getWorkerContext().setGraphState(graphState);
     serviceWorker.getWorkerContext().setupSuperstep(serviceWorker);
     GiraphTimerContext preSuperstepTimer = wcPreSuperstepTimer.time();
     serviceWorker.getWorkerContext().preSuperstep();
     preSuperstepTimer.stop();
     context.progress();
 
     for (WorkerObserver obs : serviceWorker.getWorkerObservers()) {
       obs.preSuperstep(graphState.getSuperstep());
       context.progress();
     }
   }
 
   /**
    * Prepare graph state and worker context for superstep cycles.
    */
   private void prepareGraphStateAndWorkerContext() {
     updateSuperstepGraphState();
     workerContextPreApp();
   }
 
   /**
     * Get the worker function enum.
     *
     * @return an enum detailing the roles assigned to this
     *         compute node for this Giraph job.
     */
   public GraphFunctions getGraphFunctions() {
     return graphFunctions;
   }
 
   public final WorkerContext getWorkerContext() {
     return serviceWorker.getWorkerContext();
   }
 
   public JobProgressTracker getJobProgressTracker() {
     return jobProgressTracker;
   }
 
   /**
    * Figure out what roles this BSP compute node should take on in the job.
    * Basic logic is as follows:
    * 1) If not split master, everyone does the everything and/or running
    *    ZooKeeper.
    * 2) If split master/worker, masters also run ZooKeeper
    *
    * 3) If split master/worker == true and <code>giraph.zkList</code> is
    *    externally provided, the master will not instantiate a ZK instance, but
    *    will assume a quorum is already active on the cluster for Giraph to use.
    *
    * @param conf Configuration to use
    * @param zkManager ZooKeeper manager to help determine whether to run
    *        ZooKeeper.
    * @return Functions that this mapper should do.
    */
   private static GraphFunctions determineGraphFunctions(
       ImmutableClassesGiraphConfiguration conf,
       ZooKeeperManager zkManager) {
     boolean splitMasterWorker = conf.getSplitMasterWorker();
     int taskPartition = conf.getTaskPartition();
     boolean zkAlreadyProvided = conf.isZookeeperExternal();
     GraphFunctions functions = GraphFunctions.UNKNOWN;
     // What functions should this mapper do?
     if (!splitMasterWorker) {
       if ((zkManager != null) && zkManager.runsZooKeeper()) {
         functions = GraphFunctions.ALL;
       } else {
         functions = GraphFunctions.ALL_EXCEPT_ZOOKEEPER;
       }
     } else {
       if (zkAlreadyProvided) {
         if (taskPartition == 0) {
           functions = GraphFunctions.MASTER_ONLY;
         } else {
           functions = GraphFunctions.WORKER_ONLY;
         }
       } else {
         if ((zkManager != null) && zkManager.runsZooKeeper()) {
           functions = GraphFunctions.MASTER_ZOOKEEPER_ONLY;
         } else {
           functions = GraphFunctions.WORKER_ONLY;
         }
       }
     }
     return functions;
   }
 
   /**
    * Instantiate the appropriate BspService object (Master or Worker)
    * for this compute node.
    */
   private void instantiateBspService()
     throws IOException, InterruptedException {
     if (graphFunctions.isMaster()) {
       if (LOG.isInfoEnabled()) {
         LOG.info("setup: Starting up BspServiceMaster " +
           "(master thread)...");
       }
       serviceMaster = new BspServiceMaster<I, V, E>(context, this);
       masterThread = new MasterThread<I, V, E>(serviceMaster, context);
       masterThread.setUncaughtExceptionHandler(
           createUncaughtExceptionHandler());
       masterThread.start();
     }
     if (graphFunctions.isWorker()) {
       if (LOG.isInfoEnabled()) {
         LOG.info("setup: Starting up BspServiceWorker...");
       }
       serviceWorker = new BspServiceWorker<I, V, E>(context, this);
       installGCMonitoring();
       if (LOG.isInfoEnabled()) {
         LOG.info("setup: Registering health of this worker...");
       }
     }
   }
 
   /**
    * Install GC monitoring. This method intercepts all GC, log the gc, and
    * notifies an out-of-core engine (if any is used) about the GC.
    */
   private void installGCMonitoring() {
     final GcObserver[] gcObservers = conf.createGcObservers(context);
     List<GarbageCollectorMXBean> mxBeans = ManagementFactory
         .getGarbageCollectorMXBeans();
     final OutOfCoreEngine oocEngine =
         serviceWorker.getServerData().getOocEngine();
     for (GarbageCollectorMXBean gcBean : mxBeans) {
       NotificationEmitter emitter = (NotificationEmitter) gcBean;
       NotificationListener listener = new NotificationListener() {
         @Override
         public void handleNotification(Notification notification,
                                        Object handle) {
           if (notification.getType().equals(GarbageCollectionNotificationInfo
               .GARBAGE_COLLECTION_NOTIFICATION)) {
             GarbageCollectionNotificationInfo info =
                 GarbageCollectionNotificationInfo.from(
                     (CompositeData) notification.getUserData());
 
             if (LOG.isInfoEnabled()) {
               LOG.info("installGCMonitoring: name = " + info.getGcName() +
                   ", action = " + info.getGcAction() + ", cause = " +
                   info.getGcCause() + ", duration = " +
                   info.getGcInfo().getDuration() + "ms");
             }
             gcTimeMetric.inc(info.getGcInfo().getDuration());
             GiraphMetrics.get().getGcTracker().gcOccurred(info.getGcInfo());
             for (GcObserver gcObserver : gcObservers) {
               gcObserver.gcOccurred(info);
             }
             if (oocEngine != null) {
               oocEngine.gcCompleted(info);
             }
           }
         }
       };
       //Add the listener
       emitter.addNotificationListener(listener, null, null);
     }
   }
 
   /**
    * Initialize the root logger and appender to the settings in conf.
    */
   private void initializeAndConfigureLogging() {
     // Set the log level
     String logLevel = conf.getLocalLevel();
     if (!Logger.getRootLogger().getLevel().equals(Level.toLevel(logLevel))) {
       Logger.getRootLogger().setLevel(Level.toLevel(logLevel));
       if (LOG.isInfoEnabled()) {
         LOG.info("setup: Set log level to " + logLevel);
       }
     } else {
       if (LOG.isInfoEnabled()) {
         LOG.info("setup: Log level remains at " + logLevel);
       }
     }
     // Sets pattern layout for all appenders
     if (conf.useLogThreadLayout()) {
       PatternLayout layout =
         new PatternLayout("%-7p %d [%t] %c %x - %m%n");
       Enumeration<Appender> appenderEnum =
         Logger.getRootLogger().getAllAppenders();
       while (appenderEnum.hasMoreElements()) {
         appenderEnum.nextElement().setLayout(layout);
       }
     }
     // Change ZooKeeper logging level to error (info is quite verbose) for
     // testing only
     if (conf.getLocalTestMode()) {
       LogManager.getLogger(org.apache.zookeeper.server.PrepRequestProcessor.
           class.getName()).setLevel(Level.ERROR);
     }
   }
 
   /**
    * Initialize job-level metrics used by this class.
    */
   private void initJobMetrics() {
     GiraphMetricsRegistry jobMetrics = GiraphMetrics.get().perJobOptional();
     wcPreAppTimer = new GiraphTimer(jobMetrics, "worker-context-pre-app",
         TimeUnit.MILLISECONDS);
     wcPostAppTimer = new GiraphTimer(jobMetrics, "worker-context-post-app",
         TimeUnit.MILLISECONDS);
   }
 
   @Override
   public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {
     superstepTimer = new GiraphTimer(superstepMetrics,
         TIMER_SUPERSTEP_TIME, TimeUnit.MILLISECONDS);
     computeAll = new GiraphTimer(superstepMetrics,
         TIMER_COMPUTE_ALL, TimeUnit.MILLISECONDS);
     timeToFirstMessage = new GiraphTimer(superstepMetrics,
         TIMER_TIME_TO_FIRST_MSG, TimeUnit.MICROSECONDS);
     communicationTimer = new GiraphTimer(superstepMetrics,
         TIMER_COMMUNICATION_TIME, TimeUnit.MILLISECONDS);
     gcTimeMetric = superstepMetrics.getCounter(TIMER_SUPERSTEP_GC_TIME);
     wcPreSuperstepTimer = new GiraphTimer(superstepMetrics,
         "worker-context-pre-superstep", TimeUnit.MILLISECONDS);
   }
 
   /**
    * Notification from Vertex that a message has been sent.
    */
   public void notifySentMessages() {
     // We are tracking the time between when the compute started and the first
     // message get sent. We use null to flag that we have already recorded it.
     GiraphTimerContext tmp = timeToFirstMessageTimerContext;
     if (tmp != null) {
       synchronized (timeToFirstMessage) {
         if (timeToFirstMessageTimerContext != null) {
           timeToFirstMessageTimerContext.stop();
           timeToFirstMessageTimerContext = null;
           communicationTimerContext = communicationTimer.time();
         }
       }
     }
   }
 
   /**
    * Notification of last message flushed. Comes when we finish the superstep
    * and are done waiting for all messages to send.
    */
   public void notifyFinishedCommunication() {
     GiraphTimerContext tmp = communicationTimerContext;
     if (tmp != null) {
       synchronized (communicationTimer) {
         if (communicationTimerContext != null) {
           communicationTimerContext.stop();
           communicationTimerContext = null;
         }
       }
     }
   }
 
   /**
    * Process graph data partitions active in this superstep.
    * @param context handle to the underlying cluster framework
    * @param partitionStatsList to pick up this superstep's processing stats
    * @param graphState the BSP graph state
    * @param messageStore the messages to be processed in this superstep
    * @param numThreads number of concurrent threads to do processing
    */
   private void processGraphPartitions(final Mapper<?, ?, ?, ?>.Context context,
       List<PartitionStats> partitionStatsList,
       final GraphState graphState,
       final MessageStore<I, Writable> messageStore,
       int numThreads) {
     PartitionStore<I, V, E> partitionStore = serviceWorker.getPartitionStore();
     long verticesToCompute = 0;
     for (Integer partitionId : partitionStore.getPartitionIds()) {
       verticesToCompute += partitionStore.getPartitionVertexCount(partitionId);
     }
     WorkerProgress.get().startSuperstep(
         serviceWorker.getSuperstep(), verticesToCompute,
         serviceWorker.getPartitionStore().getNumPartitions());
     partitionStore.startIteration();
 
     GiraphTimerContext computeAllTimerContext = computeAll.time();
     timeToFirstMessageTimerContext = timeToFirstMessage.time();
 
     CallableFactory<Collection<PartitionStats>> callableFactory =
       new CallableFactory<Collection<PartitionStats>>() {
         @Override
         public Callable<Collection<PartitionStats>> newCallable(
             int callableId) {
           return new ComputeCallable<I, V, E, Writable, Writable>(
               context,
               graphState,
               messageStore,
               conf,
               serviceWorker);
         }
       };
     List<Collection<PartitionStats>> results =
         ProgressableUtils.getResultsWithNCallables(callableFactory, numThreads,
             "compute-%d", context);
 
     for (Collection<PartitionStats> result : results) {
       partitionStatsList.addAll(result);
     }
 
     computeAllTimerContext.stop();
   }
 
   /**
    * Handle the event that this superstep is a restart of a failed one.
    * @param superstep current superstep
    * @return the graph state, updated if this is a restart superstep
    */
   private boolean checkSuperstepRestarted(long superstep) throws IOException {
     // Might need to restart from another superstep
     // (manually or automatic), or store a checkpoint
     if (serviceWorker.getRestartedSuperstep() == superstep) {
       if (LOG.isInfoEnabled()) {
         LOG.info("execute: Loading from checkpoint " + superstep);
       }
       VertexEdgeCount vertexEdgeCount = serviceWorker.loadCheckpoint(
         serviceWorker.getRestartedSuperstep());
       finishedSuperstepStats = new FinishedSuperstepStats(0, false,
           vertexEdgeCount.getVertexCount(), vertexEdgeCount.getEdgeCount(),
           false, CheckpointStatus.NONE);
       return true;
     }
     return false;
   }
 
   /**
    * Check if it's time to checkpoint and actually does checkpointing
    * if it is.
    * @param checkpointStatus master's decision
    * @return true if we need to stop computation after checkpoint
    * @throws IOException
    */
   private boolean storeCheckpoint(CheckpointStatus checkpointStatus)
     throws IOException {
     if (checkpointStatus != CheckpointStatus.NONE) {
       serviceWorker.storeCheckpoint();
     }
     return checkpointStatus == CheckpointStatus.CHECKPOINT_AND_HALT;
   }
 
   /**
    * Attempt to collect the final statistics on the graph data
    * processed in this superstep by this compute node
    * @param inputSuperstepStats the final graph data stats object for the
    *                            input superstep
    * @return true if the graph data has no vertices (error?) and
    *         this node should terminate
    */
   private boolean collectInputSuperstepStats(
     FinishedSuperstepStats inputSuperstepStats) {
     if (inputSuperstepStats.getVertexCount() == 0 &&
         !inputSuperstepStats.mustLoadCheckpoint()) {
       LOG.warn("map: No vertices in the graph, exiting.");
       return true;
     }
     if (conf.metricsEnabled()) {
       GiraphMetrics.get().perSuperstep().printSummary(System.err);
     }
     return false;
   }
 
   /**
    * Did the state of this compute node change?
    * @return true if the processing of supersteps should terminate.
    */
   private boolean checkTaskState() {
     if (done) {
       return true;
     }
     GiraphMetrics.get().resetSuperstepMetrics(BspService.INPUT_SUPERSTEP);
     if (graphFunctions.isNotAWorker()) {
       if (LOG.isInfoEnabled()) {
         LOG.info("map: No need to do anything when not a worker");
       }
       return true;
     }
     if (alreadyRun) {
       throw new RuntimeException("map: In BSP, map should have only been" +
         " run exactly once, (already run)");
     }
     alreadyRun = true;
     return false;
   }
 
   /**
    * Call to the WorkerContext before application begins.
    */
   private void workerContextPreApp() {
     GiraphTimerContext preAppTimerContext = wcPreAppTimer.time();
     try {
       serviceWorker.getWorkerContext().preApplication();
     } catch (InstantiationException e) {
       LOG.fatal("execute: preApplication failed in instantiation", e);
       throw new RuntimeException(
           "execute: preApplication failed in instantiation", e);
     } catch (IllegalAccessException e) {
       LOG.fatal("execute: preApplication failed in access", e);
       throw new RuntimeException(
           "execute: preApplication failed in access", e);
     }
     preAppTimerContext.stop();
     context.progress();
 
     for (WorkerObserver obs : serviceWorker.getWorkerObservers()) {
       obs.preApplication();
       context.progress();
     }
   }
 
   /**
    * Setup mapper observers
    */
   public void setupMapperObservers() {
     mapperObservers = conf.createMapperObservers(context);
     for (MapperObserver mapperObserver : mapperObservers) {
       mapperObserver.setup();
     }
   }
 
   /**
    * Executes preLoad() on worker observers.
    */
   private void preLoadOnWorkerObservers() {
     for (WorkerObserver obs : serviceWorker.getWorkerObservers()) {
       obs.preLoad();
       context.progress();
     }
   }
 
   /**
    * Executes postSave() on worker observers.
    */
   private void postSaveOnWorkerObservers() {
     for (WorkerObserver obs : serviceWorker.getWorkerObservers()) {
       obs.postSave();
       context.progress();
     }
   }
 
   /**
    * Called by owner of this GraphTaskManager object on each compute node
    */
   public void cleanup()
     throws IOException, InterruptedException {
     if (LOG.isInfoEnabled()) {
       LOG.info("cleanup: Starting for " + getGraphFunctions());
     }
     jobProgressTracker.cleanup();
     if (done) {
       return;
     }
 
     if (serviceWorker != null) {
       serviceWorker.cleanup(finishedSuperstepStats);
     }
   }
 
   /**
    * Method to send the counter values from the worker to the master,
    * after all supersteps are done, and finish cleanup
    */
   public void sendWorkerCountersAndFinishCleanup() {
     if (serviceWorker != null) {
       postSaveOnWorkerObservers();
       serviceWorker.storeCountersInZooKeeper(true);
       serviceWorker.closeZooKeeper();
     }
     try {
       if (masterThread != null) {
         masterThread.join();
         LOG.info("cleanup: Joined with master thread");
       }
     } catch (InterruptedException e) {
       // cleanup phase -- just log the error
       LOG.error("cleanup: Master thread couldn't join");
     }
     if (zkManager != null) {
       LOG.info("cleanup: Offlining ZooKeeper servers");
       try {
         zkManager.offlineZooKeeperServers(ZooKeeperManager.State.FINISHED);
         // We need this here cause apparently exceptions are eaten by Hadoop
         // when they come from the cleanup lifecycle and it's useful to know
         // if something is wrong.
         //
         // And since it's cleanup nothing too bad should happen if we don't
         // propagate and just allow the job to finish normally.
         // CHECKSTYLE: stop IllegalCatch
       } catch (Throwable e) {
         // CHECKSTYLE: resume IllegalCatch
         LOG.error("cleanup: Error offlining zookeeper", e);
       }
     }
 
     // Stop tracking metrics
     GiraphMetrics.get().shutdown();
   }
 
   /**
    * Cleanup a ZooKeeper instance managed by this
    * GiraphWorker upon job run failure.
    */
   public void zooKeeperCleanup() {
     if (graphFunctions.isZooKeeper()) {
       // ZooKeeper may have had an issue
       if (zkManager != null) {
         zkManager.cleanup();
       }
     }
   }
 
   /**
    * Cleanup all of Giraph's framework-agnostic resources
    * regardless of which type of cluster Giraph is running on.
    */
   public void workerFailureCleanup() {
     try {
       if (graphFunctions.isWorker()) {
         serviceWorker.failureCleanup();
       }
       // Stop tracking metrics
       GiraphMetrics.get().shutdown();
     // Checkstyle exception due to needing to get the original
     // exception on failure
     // CHECKSTYLE: stop IllegalCatch
     } catch (RuntimeException e1) {
     // CHECKSTYLE: resume IllegalCatch
       LOG.error("run: Worker failure failed on another RuntimeException, " +
           "original expection will be rethrown", e1);
     }
   }
 
   /**
    * Creates exception handler that will terminate process gracefully in case
    * of any uncaught exception.
    * @return new exception handler object.
    */
   public Thread.UncaughtExceptionHandler createUncaughtExceptionHandler() {
     return new OverrideExceptionHandler(
         CHECKER_IF_WORKER_SHOULD_FAIL_AFTER_EXCEPTION_CLASS.newInstance(
             getConf()), getJobProgressTracker());
   }
 
   /**
    * Creates exception handler with the passed implementation of
    * {@link CheckerIfWorkerShouldFailAfterException}.
    *
    * @param checker Instance that checks whether the job should fail.
    * @return Exception handler.
    */
   public Thread.UncaughtExceptionHandler createUncaughtExceptionHandler(
     CheckerIfWorkerShouldFailAfterException checker) {
     return new OverrideExceptionHandler(checker, getJobProgressTracker());
   }
 
   public ImmutableClassesGiraphConfiguration<I, V, E> getConf() {
     return conf;
   }
 
   /**
    * @return Time spent in GC recorder by the GC listener
    */
   public long getSuperstepGCTime() {
     return (gcTimeMetric == null) ? 0 : gcTimeMetric.count();
   }
 
   /**
    * Returns a list of zookeeper servers to connect to.
    * If the port is set to 0 and Giraph is starting a single
    * ZooKeeper server, then Zookeeper will pick its own port.
    * Otherwise, the ZooKeeper port set by the user will be used.
    * @return host:port,host:port for each zookeeper
    */
   public String getZookeeperList() {
     if (zkManager != null) {
       return zkManager.getZooKeeperServerPortString();
     } else {
       return conf.getZookeeperList();
     }
   }
 
   /**
    * Default handler for uncaught exceptions.
    * It will do the best to clean up and then will terminate current giraph job.
    */
   class OverrideExceptionHandler implements Thread.UncaughtExceptionHandler {
     /** Checker if worker should fail after a thread gets an exception */
     private final CheckerIfWorkerShouldFailAfterException checker;
     /** JobProgressTracker to log problems to */
     private final JobProgressTracker jobProgressTracker;
 
     /**
      * Constructor
      *
      * @param checker Checker if worker should fail after a thread gets an
      *                exception
      * @param jobProgressTracker JobProgressTracker to log problems to
      */
     public OverrideExceptionHandler(
         CheckerIfWorkerShouldFailAfterException checker,
         JobProgressTracker jobProgressTracker) {
       this.checker = checker;
       this.jobProgressTracker = jobProgressTracker;
     }
 
     @Override
     public void uncaughtException(final Thread t, final Throwable e) {
       if (!checker.checkIfWorkerShouldFail(t, e)) {
         LOG.error(
           "uncaughtException: OverrideExceptionHandler on thread " +
             t.getName() + ", msg = " +  e.getMessage(), e);
         return;
       }
       try {
         LOG.fatal(
             "uncaughtException: OverrideExceptionHandler on thread " +
                 t.getName() + ", msg = " +  e.getMessage() + ", exiting...", e);
         byte [] exByteArray = KryoWritableWrapper.convertToByteArray(e);
         jobProgressTracker.logError(ExceptionUtils.getStackTrace(e),
                 exByteArray);
         zooKeeperCleanup();
         workerFailureCleanup();
       } finally {
         System.exit(1);
       }
     }
   }
 
   /**
    * Interface to check if worker should fail after a thread gets an exception
    */
   public interface CheckerIfWorkerShouldFailAfterException {
     /**
      * Check if worker should fail after a thread gets an exception
      *
      * @param thread Thread which raised the exception
      * @param exception Exception which occurred
      * @return True iff worker should fail after this exception
      */
     boolean checkIfWorkerShouldFail(Thread thread, Throwable exception);
   }
 
   /**
    * Class to use by default, where each exception causes job failure
    */
   public static class FailWithEveryExceptionOccurred
       implements CheckerIfWorkerShouldFailAfterException {
     @Override
     public boolean checkIfWorkerShouldFail(Thread thread, Throwable exception) {
       return true;
     }
   }
 
   /**
    * Checks the message of a throwable, and checks whether it is a
    * "connection reset by peer" type of exception.
    *
    * @param throwable Throwable
    * @return True if the throwable is a "connection reset by peer",
    * false otherwise.
    */
   public static boolean isConnectionResetByPeer(Throwable throwable) {
     return throwable.getMessage().startsWith(
       "Connection reset by peer") ? true : false;
   }
 }