giraph-core/src/main/java/org/apache/giraph/master/MasterThread.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.master;

import org.apache.giraph.bsp.ApplicationState;
import org.apache.giraph.bsp.BspService;
import org.apache.giraph.bsp.CentralizedServiceMaster;
import org.apache.giraph.bsp.SuperstepState;
import org.apache.giraph.counters.GiraphTimers;
import org.apache.giraph.graph.Computation;
import org.apache.giraph.metrics.GiraphMetrics;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.log4j.Logger;

import java.util.Map;
import java.util.Map.Entry;
import java.util.TreeMap;

import static org.apache.giraph.conf.GiraphConstants.SPLIT_MASTER_WORKER;
import static org.apache.giraph.conf.GiraphConstants.USE_SUPERSTEP_COUNTERS;

/**
 * Master thread that will coordinate the activities of the tasks.  It runs
 * on all task processes, however, will only execute its algorithm if it knows
 * it is the "leader" from ZooKeeper.
 *
 * @param <I> Vertex id
 * @param <V> Vertex value
 * @param <E> Edge value
 */
@SuppressWarnings("rawtypes")
public class MasterThread<I extends WritableComparable, V extends Writable,
    E extends Writable> extends Thread {
  /** Counter group name for the Giraph timers */
  public static final String GIRAPH_TIMERS_COUNTER_GROUP_NAME = "Giraph Timers";
  /** Class logger */
  private static final Logger LOG = Logger.getLogger(MasterThread.class);
  /** Reference to shared BspService */
  private CentralizedServiceMaster<I, V, E> bspServiceMaster = null;
  /** Context (for counters) */
  private final Context context;
  /** Use superstep counters? */
  private final boolean superstepCounterOn;
  /** Are master and worker split or not? */
  private final boolean splitMasterWorker;
  /** Setup seconds */
  private double setupSecs = 0d;
  /** Superstep timer (in seconds) map */
  private final Map<Long, Double> superstepSecsMap =
      new TreeMap<Long, Double>();

  /**
   * Constructor.
   *
   * @param bspServiceMaster Master that already exists and setup() has
   *        been called.
   * @param context Context from the Mapper.
   */
  public MasterThread(CentralizedServiceMaster<I, V, E> bspServiceMaster,
      Context context) {
    super(MasterThread.class.getName());
    this.bspServiceMaster = bspServiceMaster;
    this.context = context;
    GiraphTimers.init(context);
    superstepCounterOn = USE_SUPERSTEP_COUNTERS.get(context.getConfiguration());
    splitMasterWorker = SPLIT_MASTER_WORKER.get(context.getConfiguration());
  }

  /**
   * The master algorithm.  The algorithm should be able to withstand
   * failures and resume as necessary since the master may switch during a
   * job.
   */
  @Override
  public void run() {
    // Algorithm:
    // 1. Become the master
    // 2. If desired, restart from a manual checkpoint
    // 3. Run all supersteps until complete
    try {
      long startMillis = System.currentTimeMillis();
      long initializeMillis = 0;
      long endMillis = 0;
      bspServiceMaster.setup();
      SuperstepState superstepState = SuperstepState.INITIAL;

      if (bspServiceMaster.becomeMaster()) {
        // First call to checkWorkers waits for all pending resources.
        // If these resources are still available at subsequent calls it just
        // reads zookeeper for the list of healthy workers.
        bspServiceMaster.checkWorkers();
        initializeMillis = System.currentTimeMillis();
        GiraphTimers.getInstance().getInitializeMs().increment(
            initializeMillis - startMillis);
        // Attempt to create InputSplits if necessary. Bail out if that fails.
        if (bspServiceMaster.getRestartedSuperstep() !=
            BspService.UNSET_SUPERSTEP ||
            (bspServiceMaster.createMappingInputSplits() != -1 &&
                bspServiceMaster.createVertexInputSplits() != -1 &&
                bspServiceMaster.createEdgeInputSplits() != -1)) {
          long setupMillis = System.currentTimeMillis() - initializeMillis;
          GiraphTimers.getInstance().getSetupMs().increment(setupMillis);
          setupSecs = setupMillis / 1000.0d;
          while (!superstepState.isExecutionComplete()) {
            long startSuperstepMillis = System.currentTimeMillis();
            long cachedSuperstep = bspServiceMaster.getSuperstep();
            // If master and worker are running together, worker will call reset
            if (splitMasterWorker) {
              GiraphMetrics.get().resetSuperstepMetrics(cachedSuperstep);
            }
            Class<? extends Computation> computationClass =
                bspServiceMaster.getMasterCompute().getComputation();
            superstepState = bspServiceMaster.coordinateSuperstep();
            long superstepMillis = System.currentTimeMillis() -
                startSuperstepMillis;
            superstepSecsMap.put(cachedSuperstep,
                superstepMillis / 1000.0d);
            if (LOG.isInfoEnabled()) {
              LOG.info("masterThread: Coordination of superstep " +
                  cachedSuperstep + " took " +
                  superstepMillis / 1000.0d +
                  " seconds ended with state " + superstepState +
                  " and is now on superstep " +
                  bspServiceMaster.getSuperstep());
            }
            if (superstepCounterOn) {
              String computationName = (computationClass == null) ?
                  null : computationClass.getSimpleName();
              GiraphTimers.getInstance().getSuperstepMs(cachedSuperstep,
                  computationName).increment(superstepMillis);
            }
            bspServiceMaster.addGiraphTimersAndSendCounters(cachedSuperstep);

            bspServiceMaster.postSuperstep();

            // If a worker failed, restart from a known good superstep
            if (superstepState == SuperstepState.WORKER_FAILURE) {
              bspServiceMaster.restartFromCheckpoint(
                  bspServiceMaster.getLastGoodCheckpoint());
            }
            endMillis = System.currentTimeMillis();
          }
          bspServiceMaster.setJobState(ApplicationState.FINISHED, -1, -1);
        }
      }
      bspServiceMaster.cleanup(superstepState);
      if (!superstepSecsMap.isEmpty()) {
        GiraphTimers.getInstance().getShutdownMs().
          increment(System.currentTimeMillis() - endMillis);
        if (LOG.isInfoEnabled()) {
          LOG.info("setup: Took " + setupSecs + " seconds.");
        }
        for (Entry<Long, Double> entry : superstepSecsMap.entrySet()) {
          if (LOG.isInfoEnabled()) {
            if (entry.getKey().longValue() ==
                BspService.INPUT_SUPERSTEP) {
              LOG.info("input superstep: Took " +
                  entry.getValue() + " seconds.");
            } else {
              LOG.info("superstep " + entry.getKey() + ": Took " +
                  entry.getValue() + " seconds.");
            }
          }
          context.progress();
        }
        if (LOG.isInfoEnabled()) {
          LOG.info("shutdown: Took " +
              (System.currentTimeMillis() - endMillis) /
              1000.0d + " seconds.");
          LOG.info("total: Took " +
              ((System.currentTimeMillis() - initializeMillis) /
              1000.0d) + " seconds.");
        }
        GiraphTimers.getInstance().getTotalMs().
          increment(System.currentTimeMillis() - initializeMillis);
      }
      bspServiceMaster.addGiraphTimersAndSendCounters(
              bspServiceMaster.getSuperstep());
      bspServiceMaster.postApplication();
      // CHECKSTYLE: stop IllegalCatchCheck
    } catch (Exception e) {
      // CHECKSTYLE: resume IllegalCatchCheck
      LOG.error("masterThread: Master algorithm failed with " +
          e.getClass().getSimpleName(), e);
      bspServiceMaster.failureCleanup(e);
      throw new IllegalStateException(e);
    }
  }
}