| /* |
| * 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"); |
| } |
| if (gcTimeMetrics != null) { |
| 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; |
| } |
| } |
