src/core/org/apache/hadoop/util/QueueProcessingStatistics.java - hadoop - Git at Google

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

 import org.apache.commons.logging.Log;

 /**
  * Hadoop has several work queues, such as
  * {@link org.apache.hadoop.hdfs.server.namenode.FSNamesystem#neededReplications}
  * With a properly throttled queue, a worker thread cycles repeatedly,
  * doing a chunk of work each cycle then resting a bit, until the queue is
  * empty.  This class is intended to collect statistics about the behavior of
  * such queues and consumers.  It reports the amount of work done and
  * how long it took, for the first cycle after collection starts, and for
  * the total number of cycles needed to flush the queue.  We use a state
  * machine to detect when the queue has been flushed and then we log the
  * stats; see {@link State} for enumeration of the states and their meanings.
  */
 public abstract class QueueProcessingStatistics {
   //All member variables and methods that would normally be access "private"
   //are instead package-private so we can subclass for unit testing.
   State state = State.BEGIN_COLLECTING;
   long startTimeCurrentCycle;
   long startTime;
   long processDuration;
   long clockDuration;
   long workItemCount;
   int  cycleCount;

   String queueName;
   String workItemsName;
   Log LOG;

   /**
    * This enum provides the "states" of a state machine for
    * {@link QueueProcessingStatistics}.
    * The meanings of the states are: <ul>
    * <li> BEGIN_COLLECTING - Ready to begin.
    * <li> IN_FIRST_CYCLE - Started the first cycle.
    * <li> IN_SOLE_CYCLE - Still in first cycle, but already know there will be
    *                      no further cycles because this one will complete all
    *                      needed work. When done, will go straight to
    *                      "DONE_COLLECTING".
    * <li> DONE_FIRST_CYCLE - Done first cycle, doing subsequent cycles.
    * <li> IN_LAST_CYCLE - Started the last cycle.
    * <li> DONE_COLLECTING - Done with last cycle, finishing up.
    * </ul>
    */
   public enum State {
     BEGIN_COLLECTING,
     IN_FIRST_CYCLE,
     IN_SOLE_CYCLE,
     DONE_FIRST_CYCLE,
     IN_LAST_CYCLE,
     DONE_COLLECTING,
   }

   /**
    * @param queueName - Human-readable name of the queue being monitored,
    *                    used as first word in the log messages.
    * @param workItemsName - what kind of work items are being managed
    *                    on the queue?  A plural word is best here, for logging.
    * @param logObject - What log do you want the log messages to be sent to?
    */
   public QueueProcessingStatistics(String queueName, String workItemsName,
       Log logObject) {
     this.queueName = queueName;
     this.workItemsName = workItemsName;
     this.LOG = logObject;
   }

   public void startCycle(int maxWorkToProcess) {
     //only collect stats for one complete flush of the queue
     if (state == State.DONE_COLLECTING) return;

     //regardless of state, record the start of this cycle
     startTimeCurrentCycle = now();
     boolean preDetectLastCycle = preCheckIsLastCycle(maxWorkToProcess);

     switch (state) {
     case BEGIN_COLLECTING:
       startTime = startTimeCurrentCycle;
       state = preDetectLastCycle ? State.IN_SOLE_CYCLE : State.IN_FIRST_CYCLE;
       break;
     default:
       if (preDetectLastCycle)
         state = State.IN_LAST_CYCLE;
       break;
     }
   }

   public void endCycle(int workFound) {
     //only collect stats for first pass through the queue
     if (state == State.DONE_COLLECTING) return;

     //regardless of state, record the end of this cycle
     //and accumulate the cycle's stats
     long endTimeCurrentCycle = now();
     processDuration += endTimeCurrentCycle - startTimeCurrentCycle;
     clockDuration = endTimeCurrentCycle - startTime;
     workItemCount += workFound;
     cycleCount++;
     boolean postDetectLastCycle = postCheckIsLastCycle(workFound);

     switch (state) {
     case BEGIN_COLLECTING:
       logError("endCycle() called before startCycle(), "
           + "exiting stats collection");
       state = State.DONE_COLLECTING;
       break;
     case IN_FIRST_CYCLE:
       if (postDetectLastCycle) {
         state = State.IN_SOLE_CYCLE;
         //and fall through
       } else {
         logEndFirstCycle();
         state = State.DONE_FIRST_CYCLE;
         break;
       }
     case IN_SOLE_CYCLE:
       logEndFirstCycle();
       logEndLastCycle();
       state = State.DONE_COLLECTING;
       break;
     case DONE_FIRST_CYCLE:
       if (postDetectLastCycle) {
         state = State.IN_LAST_CYCLE;
         //and fall through
       } else {
         break;
       }
     case IN_LAST_CYCLE:
       logEndLastCycle();
       state = State.DONE_COLLECTING;
       break;
     default:
       logError("unallowed state reached, exiting stats collection");
       state = State.DONE_COLLECTING;
       break;
     }
   }

   public void checkRestart() {
     switch (state) {
     case BEGIN_COLLECTING:
       //situation normal
       return;
     case DONE_COLLECTING:
       logError("Restarted stats collection after completion of first "
           + "queue flush.");
       initialize();
       break;
     default:
       //for all other cases, we are in the middle of stats collection,
       //so output the stats collected so far before re-initializing
       logErrorWithStats("Restarted stats collection before completion of "
           + "first queue flush.");
       initialize();
       break;
     }
   }

   void initialize() {
     state = State.BEGIN_COLLECTING;
     startTimeCurrentCycle = 0;
     startTime = 0;
     processDuration = 0;
     clockDuration = 0;
     workItemCount = 0;
     cycleCount = 0;
   }

   /**
    * The termination condition is to identify the last cycle that will
    * empty the queue.  Two abstract APIs are called: {@code preCheckIsLastCycle}
    * is called at the beginning of each cycle, and
    * {@link #postCheckIsLastCycle} is called at the end of each cycle.
    * At least one of them must correctly provide the termination
    * condition. The other may always return 'false'.  If either of them
    * returns 'true' in a given cycle, then at the end of that cycle the
    * stats will be output to log, and stats collection will end.
    *
    * @param maxWorkToProcess - if this number is greater than the amount
    *            of work remaining at the start of a cycle, then it will
    *            be the last cycle.
    * @return - true if last cycle detected, else false
    */
   public abstract boolean preCheckIsLastCycle(int maxWorkToProcess);

   /**
    * See {@link #preCheckIsLastCycle}.
    * @param workFound - may not be useful
    * @return - true if remaining work is zero at end of cycle,
    *           else false
    */
   public abstract boolean postCheckIsLastCycle(int workFound);

   String msgEndFirstCycle() {
     return queueName + " QueueProcessingStatistics: First cycle completed "
     + workItemCount + " " + workItemsName + " in " + processDuration
     + " msec";
   }

   void logEndFirstCycle() {
     LOG.info(msgEndFirstCycle());
   }

   String msgEndLastCycle() {
     return queueName
     + " QueueProcessingStatistics: Queue flush completed "
     + workItemCount + " " + workItemsName + " in "
     + processDuration + " msec processing time, "
     + clockDuration + " msec clock time, "
     + cycleCount + " cycles";
   }

   void logEndLastCycle() {
     LOG.info(msgEndLastCycle());
   }

   String msgError(String msg) {
     return queueName
     + " QueueProcessingStatistics - Error: " + msg;
   }

   void logError(String msg) {
     LOG.error(msgError(msg));
   }

   String msgErrorWithStats(String msg) {
     return queueName
     + " QueueProcessingStatistics - Error: " + msg
     + " Completed " + workItemCount + " " + workItemsName + " in "
     + processDuration + " msec processing time, "
     + clockDuration + " msec clock time, "
     + cycleCount + " cycles";
   }

   void logErrorWithStats(String msg) {
     LOG.error(msgErrorWithStats(msg));
   }

   /**
    * Current system time.
    * @return current time in msec.
    */
   static long now() {
     return System.currentTimeMillis();
   }

 }
	/**
	* 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.hadoop.util;

	import org.apache.commons.logging.Log;

	/**
	* Hadoop has several work queues, such as
	* {@link org.apache.hadoop.hdfs.server.namenode.FSNamesystem#neededReplications}
	* With a properly throttled queue, a worker thread cycles repeatedly,
	* doing a chunk of work each cycle then resting a bit, until the queue is
	* empty. This class is intended to collect statistics about the behavior of
	* such queues and consumers. It reports the amount of work done and
	* how long it took, for the first cycle after collection starts, and for
	* the total number of cycles needed to flush the queue. We use a state
	* machine to detect when the queue has been flushed and then we log the
	* stats; see {@link State} for enumeration of the states and their meanings.
	*/
	public abstract class QueueProcessingStatistics {
	//All member variables and methods that would normally be access "private"
	//are instead package-private so we can subclass for unit testing.
	State state = State.BEGIN_COLLECTING;
	long startTimeCurrentCycle;
	long startTime;
	long processDuration;
	long clockDuration;
	long workItemCount;
	int cycleCount;

	String queueName;
	String workItemsName;
	Log LOG;

	/**
	* This enum provides the "states" of a state machine for
	* {@link QueueProcessingStatistics}.
	* The meanings of the states are: <ul>
	* <li> BEGIN_COLLECTING - Ready to begin.
	* <li> IN_FIRST_CYCLE - Started the first cycle.
	* <li> IN_SOLE_CYCLE - Still in first cycle, but already know there will be
	* no further cycles because this one will complete all
	* needed work. When done, will go straight to
	* "DONE_COLLECTING".
	* <li> DONE_FIRST_CYCLE - Done first cycle, doing subsequent cycles.
	* <li> IN_LAST_CYCLE - Started the last cycle.
	* <li> DONE_COLLECTING - Done with last cycle, finishing up.
	* </ul>
	*/
	public enum State {
	BEGIN_COLLECTING,
	IN_FIRST_CYCLE,
	IN_SOLE_CYCLE,
	DONE_FIRST_CYCLE,
	IN_LAST_CYCLE,
	DONE_COLLECTING,
	}

	/**
	* @param queueName - Human-readable name of the queue being monitored,
	* used as first word in the log messages.
	* @param workItemsName - what kind of work items are being managed
	* on the queue? A plural word is best here, for logging.
	* @param logObject - What log do you want the log messages to be sent to?
	*/
	public QueueProcessingStatistics(String queueName, String workItemsName,
	Log logObject) {
	this.queueName = queueName;
	this.workItemsName = workItemsName;
	this.LOG = logObject;
	}

	public void startCycle(int maxWorkToProcess) {
	//only collect stats for one complete flush of the queue
	if (state == State.DONE_COLLECTING) return;

	//regardless of state, record the start of this cycle
	startTimeCurrentCycle = now();
	boolean preDetectLastCycle = preCheckIsLastCycle(maxWorkToProcess);

	switch (state) {
	case BEGIN_COLLECTING:
	startTime = startTimeCurrentCycle;
	state = preDetectLastCycle ? State.IN_SOLE_CYCLE : State.IN_FIRST_CYCLE;
	break;
	default:
	if (preDetectLastCycle)
	state = State.IN_LAST_CYCLE;
	break;
	}
	}

	public void endCycle(int workFound) {
	//only collect stats for first pass through the queue
	if (state == State.DONE_COLLECTING) return;

	//regardless of state, record the end of this cycle
	//and accumulate the cycle's stats
	long endTimeCurrentCycle = now();
	processDuration += endTimeCurrentCycle - startTimeCurrentCycle;
	clockDuration = endTimeCurrentCycle - startTime;
	workItemCount += workFound;
	cycleCount++;
	boolean postDetectLastCycle = postCheckIsLastCycle(workFound);

	switch (state) {
	case BEGIN_COLLECTING:
	logError("endCycle() called before startCycle(), "
	+ "exiting stats collection");
	state = State.DONE_COLLECTING;
	break;
	case IN_FIRST_CYCLE:
	if (postDetectLastCycle) {
	state = State.IN_SOLE_CYCLE;
	//and fall through
	} else {
	logEndFirstCycle();
	state = State.DONE_FIRST_CYCLE;
	break;
	}
	case IN_SOLE_CYCLE:
	logEndFirstCycle();
	logEndLastCycle();
	state = State.DONE_COLLECTING;
	break;
	case DONE_FIRST_CYCLE:
	if (postDetectLastCycle) {
	state = State.IN_LAST_CYCLE;
	//and fall through
	} else {
	break;
	}
	case IN_LAST_CYCLE:
	logEndLastCycle();
	state = State.DONE_COLLECTING;
	break;
	default:
	logError("unallowed state reached, exiting stats collection");
	state = State.DONE_COLLECTING;
	break;
	}
	}

	public void checkRestart() {
	switch (state) {
	case BEGIN_COLLECTING:
	//situation normal
	return;
	case DONE_COLLECTING:
	logError("Restarted stats collection after completion of first "
	+ "queue flush.");
	initialize();
	break;
	default:
	//for all other cases, we are in the middle of stats collection,
	//so output the stats collected so far before re-initializing
	logErrorWithStats("Restarted stats collection before completion of "
	+ "first queue flush.");
	initialize();
	break;
	}
	}

	void initialize() {
	state = State.BEGIN_COLLECTING;
	startTimeCurrentCycle = 0;
	startTime = 0;
	processDuration = 0;
	clockDuration = 0;
	workItemCount = 0;
	cycleCount = 0;
	}

	/**
	* The termination condition is to identify the last cycle that will
	* empty the queue. Two abstract APIs are called: {@code preCheckIsLastCycle}
	* is called at the beginning of each cycle, and
	* {@link #postCheckIsLastCycle} is called at the end of each cycle.
	* At least one of them must correctly provide the termination
	* condition. The other may always return 'false'. If either of them
	* returns 'true' in a given cycle, then at the end of that cycle the
	* stats will be output to log, and stats collection will end.
	*
	* @param maxWorkToProcess - if this number is greater than the amount
	* of work remaining at the start of a cycle, then it will
	* be the last cycle.
	* @return - true if last cycle detected, else false
	*/
	public abstract boolean preCheckIsLastCycle(int maxWorkToProcess);

	/**
	* See {@link #preCheckIsLastCycle}.
	* @param workFound - may not be useful
	* @return - true if remaining work is zero at end of cycle,
	* else false
	*/
	public abstract boolean postCheckIsLastCycle(int workFound);

	String msgEndFirstCycle() {
	return queueName + " QueueProcessingStatistics: First cycle completed "
	+ workItemCount + " " + workItemsName + " in " + processDuration
	+ " msec";
	}

	void logEndFirstCycle() {
	LOG.info(msgEndFirstCycle());
	}

	String msgEndLastCycle() {
	return queueName
	+ " QueueProcessingStatistics: Queue flush completed "
	+ workItemCount + " " + workItemsName + " in "
	+ processDuration + " msec processing time, "
	+ clockDuration + " msec clock time, "
	+ cycleCount + " cycles";
	}

	void logEndLastCycle() {
	LOG.info(msgEndLastCycle());
	}

	String msgError(String msg) {
	return queueName
	+ " QueueProcessingStatistics - Error: " + msg;
	}

	void logError(String msg) {
	LOG.error(msgError(msg));
	}

	String msgErrorWithStats(String msg) {
	return queueName
	+ " QueueProcessingStatistics - Error: " + msg
	+ " Completed " + workItemCount + " " + workItemsName + " in "
	+ processDuration + " msec processing time, "
	+ clockDuration + " msec clock time, "
	+ cycleCount + " cycles";
	}

	void logErrorWithStats(String msg) {
	LOG.error(msgErrorWithStats(msg));
	}

	/**
	* Current system time.
	* @return current time in msec.
	*/
	static long now() {
	return System.currentTimeMillis();
	}

	}