src/adaptorlib/Journal.java - plexi - Git at Google

 // Copyright 2011 Google Inc. All Rights Reserved.
 //
 // Licensed 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 adaptorlib;

 import java.util.*;

 /**
  * Contains registers and stats regarding runtime.
  */
 class Journal {
   private HashMap<DocId, Integer> timesPushed
       = new HashMap<DocId, Integer>();
   private long totalPushes;

   private HashMap<DocId, Integer> timesGsaRequested
       = new HashMap<DocId, Integer>();
   private long totalGsaRequests;

   private HashMap<DocId, Integer> timesNonGsaRequested
       = new HashMap<DocId, Integer>();
   private long totalNonGsaRequests;

   private final TimeProvider timeProvider;
   private final long startedAt;
   /**
    * Resolution of {@link System#currentTypeMillis()} to the millisecond.
    */
   private final long timeResolution;

   /**
    * Time-based bookkeeping for charts. Each element in the array is for a
    * different time period.
    */
   private Stats[] timeStats;

   /** Response start time storage until response completion. */
   private ThreadLocal<Long> requestResponseStart = new ThreadLocal<Long>();
   /** Request processing start time storage until processing completion. */
   private ThreadLocal<Long> requestProcessingStart = new ThreadLocal<Long>();

   public Journal() {
     this(new SystemTimeProvider());
   }

   protected Journal(TimeProvider timeProvider) {
     this.timeProvider = timeProvider;
     this.startedAt = timeProvider.currentTimeMillis();
     this.timeResolution = determineTimeResolution();
     // We want data within the Stats to agree with each other, so we provide the
     // same time to each of them.
     long time = startedAt;
     this.timeStats = new Stats[] {
       new Stats(60, 1000,           time), /* one minute, second granularity */
       new Stats(60, 1000 * 60,      time), /* one hour, minute granularity */
       new Stats(48, 1000 * 60 * 30, time), /* one day, half-hour granularity */
     };
   }

   synchronized void recordDocIdPush(List<DocInfo> pushed) {
     for (DocInfo record : pushed) {
       increment(timesPushed, record.getDocId());
     }
     totalPushes += pushed.size();
   }

   synchronized void recordGsaContentRequest(DocId docId) {
     increment(timesGsaRequested, docId);
     totalGsaRequests++;
   }

   synchronized void recordNonGsaContentRequest(DocId requested) {
     increment(timesNonGsaRequested, requested);
     totalNonGsaRequests++;
   }

   /**
    * Record that the response of a request has been started on this thread. This
    * relates to the actual I/O of sending the response.
    */
   void recordRequestResponseStart() {
     requestResponseStart.set(timeProvider.currentTimeMillis());
   }

   /**
    * Record that the response this thread was sending has completed.
    */
   void recordRequestResponseEnd(long responseSize) {
     long time = timeProvider.currentTimeMillis();
     long duration = endDuration(requestResponseStart, time);
     synchronized (this) {
       for (Stats stats : timeStats) {
         Stat stat = stats.getCurrentStat(time);
         stat.requestResponsesCount++;
         stat.requestResponsesDurationSum += duration;
         stat.requestResponsesMaxDuration = Math.max(
             stat.requestResponsesMaxDuration, duration);
         stat.requestResponsesThroughput += responseSize;
       }
     }
   }

   /**
    * Record that the processing of a request has been started on this thread.
    * This relates to internal computation required to satisfy the request.
    */
   void recordRequestProcessingStart() {
     requestProcessingStart.set(timeProvider.currentTimeMillis());
   }

   /**
    * Record that the processing this thread was performing to satify the request
    * has completed.
    */
   void recordRequestProcessingEnd(long responseSize) {
     long time = timeProvider.currentTimeMillis();
     long duration = endDuration(requestProcessingStart, time);
     synchronized (this) {
       for (Stats stats : timeStats) {
         Stat stat = stats.getCurrentStat(time);
         stat.requestProcessingsCount++;
         stat.requestProcessingsDurationSum += duration;
         stat.requestProcessingsMaxDuration = Math.max(
             stat.requestProcessingsMaxDuration, duration);
         stat.requestProcessingsThroughput += responseSize;
       }
     }
   }

   private long endDuration(ThreadLocal<Long> localStartTime, long endTime) {
     Long startTime = localStartTime.get();
     localStartTime.remove();
     if (startTime == null) {
       throw new IllegalStateException("Record start must be called before "
                                       + "record end");
     }
     return endTime - startTime;
   }

   /**
    * Determine the resolution that {@link System#currentTimeMillis} supports, in
    * milliseconds.
    */
   private long determineTimeResolution() {
     long resolution = Long.MAX_VALUE;
     for (int i = 0; i < 5; i++) {
       resolution = Math.min(resolution, determineTimeResolutionOnce());
     }
     return resolution;
   }

   /**
    * One trial of determining the resolution that {@link
    * System#currentTimeMillis} supports, in milliseconds.
    */
   private long determineTimeResolutionOnce() {
     long time = timeProvider.currentTimeMillis();
     long startTime = time;
     while (startTime == time) {
       time = timeProvider.currentTimeMillis();
     }
     return time - startTime;
   }

   private static void increment(HashMap<DocId, Integer> counts, DocId id) {
     if (!counts.containsKey(id)) {
       counts.put(id, 0);
     }
     counts.put(id, 1 + counts.get(id));
   }

   /**
    * Access to the timeStats for use in {@link DashboardHandler} only.
    */
   synchronized JournalSnapshot getSnapshot() {
     long currentTime = timeProvider.currentTimeMillis();
     Stats[] timeStatsClone = new Stats[timeStats.length];
     for (int i = 0; i < timeStats.length; i++) {
       // Cause stats to update its internal structures
       timeStats[i].getCurrentStat(currentTime);
       try {
         timeStatsClone[i] = (Stats) timeStats[i].clone();
       } catch (CloneNotSupportedException ex) {
         throw new IllegalStateException(ex);
       }
     }

     return new JournalSnapshot(this, currentTime, timeStatsClone);
   }

   static class JournalSnapshot {
     final long numUniqueDocIdsPushed;
     final long numTotalDocIdsPushed;
     final long numUniqueGsaRequests;
     final long numTotalGsaRequests;
     final long numUniqueNonGsaRequests;
     final long numTotalNonGsaRequests;
     final long whenStarted;
     final long currentTime;
     final long timeResolution;
     final Stats[] timeStats;

     JournalSnapshot(Journal journal, long currentTime, Stats[] timeStatsClone) {
       this.numUniqueDocIdsPushed = journal.timesPushed.size();
       this.numTotalDocIdsPushed = journal.totalPushes;
       this.numUniqueGsaRequests = journal.timesGsaRequested.size();
       this.numTotalGsaRequests = journal.totalGsaRequests;
       this.numUniqueNonGsaRequests = journal.timesNonGsaRequested.size();
       this.numTotalNonGsaRequests = journal.totalNonGsaRequests;
       this.timeResolution = journal.timeResolution;
       this.whenStarted = journal.startedAt;
       this.currentTime = currentTime;
       this.timeStats = timeStatsClone;
     }
   }

   static class Stats implements Cloneable {
     /**
      * Circular buffer containing all the statistics this object contains. When
      * the current {@link Stat} "expires", {@link #currentStat} is incremented
      * (floor {@code stats.length}) and the new current {@code Stat} object is
      * reset.
      */
     Stat[] stats;
     /**
      * The amount of time each {@link Stat} object's statistics were measured.
      */
     long snapshotDurationMs;
     /**
      * Current {@link Stat} object within {@link #stats}. All incoming
      * statistics should update the current {@code Stat} object.
      */
     int currentStat;
     /**
      * Time that the current {@link Stat} object will expire and {@link
      * #currentStat} will be incremented.
      */
     long pendingStatPeriodEnd;

     public Stats(int statCount, long snapshotDuration, long currentTime) {
       this.snapshotDurationMs = snapshotDuration;

       this.stats = new Stat[statCount];
       // Pre-allocate all Stat objects
       for (int i = 0; i < this.stats.length; i++) {
         this.stats[i] = new Stat();
       }

       // Initialize expiration times for pendingStat objects
       long duration = this.snapshotDurationMs;
       this.pendingStatPeriodEnd = ((currentTime / duration) * duration)
           + duration;
     }

     /**
      * Retrive the current {@code Stat} object that applies to {@code
      * currentTime}. {@code currentTime} is expected to be an actual point in
      * time while the caller was holding the lock on {@link Journal}. It may
      * never be less than the previous call.
      */
     public Stat getCurrentStat(long currentTime) {
       // Check if the current Stat object is still valid to write to
       if (pendingStatPeriodEnd > currentTime) {
         return stats[currentStat];
       }
       // Check if all the Stat objects are invalid. This occurs when the last
       // request was a long time ago.
       if (currentTime - pendingStatPeriodEnd
           > snapshotDurationMs * stats.length) {
         for (int i = 0; i < stats.length; i++) {
           stats[i].reset();
         }
         long duration = snapshotDurationMs;
         pendingStatPeriodEnd = ((currentTime / duration) * duration) + duration;
       }
       // Walk through time to get the current Stat object
       while (pendingStatPeriodEnd <= currentTime) {
         currentStat++;
         currentStat %= stats.length;
         stats[currentStat].reset();
         pendingStatPeriodEnd += snapshotDurationMs;
       }
       return stats[currentStat];
     }

     public Object clone() throws CloneNotSupportedException {
       Stats statsClone = (Stats) super.clone();
       statsClone.stats = new Stat[stats.length];
       for (int i = 0; i < stats.length; i++) {
         statsClone.stats[i] = (Stat) stats[i].clone();
       }
       return statsClone;
     }
   }

   /**
    * Structure for holding statistics data. Data applies to a time period only,
    * as controlled by {@link Stats}.
    */
   static class Stat implements Cloneable {
     /**
      * The number of responses sent.
      */
     long requestResponsesCount;
     /**
      * The total duration of responses sent.
      */
     long requestResponsesDurationSum;
     /**
      * The maximal duration of any one response.
      */
     long requestResponsesMaxDuration;
     /**
      * Number of bytes sent to client.
      */
     long requestResponsesThroughput;

     /**
      * The number of requests processed.
      */
     long requestProcessingsCount;
     /**
      * The total duration of response processings.
      */
     long requestProcessingsDurationSum;
     /**
      * The maximal duration of any one response processing.
      */
     long requestProcessingsMaxDuration;
     /**
      * Number of bytes generated by the adaptor.
      */
     long requestProcessingsThroughput;

     public Stat() {
       reset();
     }

     /**
      * Reset object for reuse.
      */
     private void reset() {
       requestResponsesCount = 0;
       requestResponsesDurationSum = 0;
       requestResponsesMaxDuration = 0;
       requestResponsesThroughput = 0;

       requestProcessingsCount = 0;
       requestProcessingsDurationSum = 0;
       requestProcessingsMaxDuration = 0;
       requestProcessingsThroughput = 0;
     }

     public Object clone() throws CloneNotSupportedException {
       return super.clone();
     }
   }
 }
	// Copyright 2011 Google Inc. All Rights Reserved.
	//
	// Licensed 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 adaptorlib;

	import java.util.*;

	/**
	* Contains registers and stats regarding runtime.
	*/
	class Journal {
	private HashMap<DocId, Integer> timesPushed
	= new HashMap<DocId, Integer>();
	private long totalPushes;

	private HashMap<DocId, Integer> timesGsaRequested
	= new HashMap<DocId, Integer>();
	private long totalGsaRequests;

	private HashMap<DocId, Integer> timesNonGsaRequested
	= new HashMap<DocId, Integer>();
	private long totalNonGsaRequests;

	private final TimeProvider timeProvider;
	private final long startedAt;
	/**
	* Resolution of {@link System#currentTypeMillis()} to the millisecond.
	*/
	private final long timeResolution;

	/**
	* Time-based bookkeeping for charts. Each element in the array is for a
	* different time period.
	*/
	private Stats[] timeStats;

	/** Response start time storage until response completion. */
	private ThreadLocal<Long> requestResponseStart = new ThreadLocal<Long>();
	/** Request processing start time storage until processing completion. */
	private ThreadLocal<Long> requestProcessingStart = new ThreadLocal<Long>();

	public Journal() {
	this(new SystemTimeProvider());
	}

	protected Journal(TimeProvider timeProvider) {
	this.timeProvider = timeProvider;
	this.startedAt = timeProvider.currentTimeMillis();
	this.timeResolution = determineTimeResolution();
	// We want data within the Stats to agree with each other, so we provide the
	// same time to each of them.
	long time = startedAt;
	this.timeStats = new Stats[] {
	new Stats(60, 1000, time), /* one minute, second granularity */
	new Stats(60, 1000 * 60, time), /* one hour, minute granularity */
	new Stats(48, 1000 * 60 * 30, time), /* one day, half-hour granularity */
	};
	}

	synchronized void recordDocIdPush(List<DocInfo> pushed) {
	for (DocInfo record : pushed) {
	increment(timesPushed, record.getDocId());
	}
	totalPushes += pushed.size();
	}

	synchronized void recordGsaContentRequest(DocId docId) {
	increment(timesGsaRequested, docId);
	totalGsaRequests++;
	}

	synchronized void recordNonGsaContentRequest(DocId requested) {
	increment(timesNonGsaRequested, requested);
	totalNonGsaRequests++;
	}

	/**
	* Record that the response of a request has been started on this thread. This
	* relates to the actual I/O of sending the response.
	*/
	void recordRequestResponseStart() {
	requestResponseStart.set(timeProvider.currentTimeMillis());
	}

	/**
	* Record that the response this thread was sending has completed.
	*/
	void recordRequestResponseEnd(long responseSize) {
	long time = timeProvider.currentTimeMillis();
	long duration = endDuration(requestResponseStart, time);
	synchronized (this) {
	for (Stats stats : timeStats) {
	Stat stat = stats.getCurrentStat(time);
	stat.requestResponsesCount++;
	stat.requestResponsesDurationSum += duration;
	stat.requestResponsesMaxDuration = Math.max(
	stat.requestResponsesMaxDuration, duration);
	stat.requestResponsesThroughput += responseSize;
	}
	}
	}

	/**
	* Record that the processing of a request has been started on this thread.
	* This relates to internal computation required to satisfy the request.
	*/
	void recordRequestProcessingStart() {
	requestProcessingStart.set(timeProvider.currentTimeMillis());
	}

	/**
	* Record that the processing this thread was performing to satify the request
	* has completed.
	*/
	void recordRequestProcessingEnd(long responseSize) {
	long time = timeProvider.currentTimeMillis();
	long duration = endDuration(requestProcessingStart, time);
	synchronized (this) {
	for (Stats stats : timeStats) {
	Stat stat = stats.getCurrentStat(time);
	stat.requestProcessingsCount++;
	stat.requestProcessingsDurationSum += duration;
	stat.requestProcessingsMaxDuration = Math.max(
	stat.requestProcessingsMaxDuration, duration);
	stat.requestProcessingsThroughput += responseSize;
	}
	}
	}

	private long endDuration(ThreadLocal<Long> localStartTime, long endTime) {
	Long startTime = localStartTime.get();
	localStartTime.remove();
	if (startTime == null) {
	throw new IllegalStateException("Record start must be called before "
	+ "record end");
	}
	return endTime - startTime;
	}

	/**
	* Determine the resolution that {@link System#currentTimeMillis} supports, in
	* milliseconds.
	*/
	private long determineTimeResolution() {
	long resolution = Long.MAX_VALUE;
	for (int i = 0; i < 5; i++) {
	resolution = Math.min(resolution, determineTimeResolutionOnce());
	}
	return resolution;
	}

	/**
	* One trial of determining the resolution that {@link
	* System#currentTimeMillis} supports, in milliseconds.
	*/
	private long determineTimeResolutionOnce() {
	long time = timeProvider.currentTimeMillis();
	long startTime = time;
	while (startTime == time) {
	time = timeProvider.currentTimeMillis();
	}
	return time - startTime;
	}

	private static void increment(HashMap<DocId, Integer> counts, DocId id) {
	if (!counts.containsKey(id)) {
	counts.put(id, 0);
	}
	counts.put(id, 1 + counts.get(id));
	}

	/**
	* Access to the timeStats for use in {@link DashboardHandler} only.
	*/
	synchronized JournalSnapshot getSnapshot() {
	long currentTime = timeProvider.currentTimeMillis();
	Stats[] timeStatsClone = new Stats[timeStats.length];
	for (int i = 0; i < timeStats.length; i++) {
	// Cause stats to update its internal structures
	timeStats[i].getCurrentStat(currentTime);
	try {
	timeStatsClone[i] = (Stats) timeStats[i].clone();
	} catch (CloneNotSupportedException ex) {
	throw new IllegalStateException(ex);
	}
	}

	return new JournalSnapshot(this, currentTime, timeStatsClone);
	}

	static class JournalSnapshot {
	final long numUniqueDocIdsPushed;
	final long numTotalDocIdsPushed;
	final long numUniqueGsaRequests;
	final long numTotalGsaRequests;
	final long numUniqueNonGsaRequests;
	final long numTotalNonGsaRequests;
	final long whenStarted;
	final long currentTime;
	final long timeResolution;
	final Stats[] timeStats;

	JournalSnapshot(Journal journal, long currentTime, Stats[] timeStatsClone) {
	this.numUniqueDocIdsPushed = journal.timesPushed.size();
	this.numTotalDocIdsPushed = journal.totalPushes;
	this.numUniqueGsaRequests = journal.timesGsaRequested.size();
	this.numTotalGsaRequests = journal.totalGsaRequests;
	this.numUniqueNonGsaRequests = journal.timesNonGsaRequested.size();
	this.numTotalNonGsaRequests = journal.totalNonGsaRequests;
	this.timeResolution = journal.timeResolution;
	this.whenStarted = journal.startedAt;
	this.currentTime = currentTime;
	this.timeStats = timeStatsClone;
	}
	}

	static class Stats implements Cloneable {
	/**
	* Circular buffer containing all the statistics this object contains. When
	* the current {@link Stat} "expires", {@link #currentStat} is incremented
	* (floor {@code stats.length}) and the new current {@code Stat} object is
	* reset.
	*/
	Stat[] stats;
	/**
	* The amount of time each {@link Stat} object's statistics were measured.
	*/
	long snapshotDurationMs;
	/**
	* Current {@link Stat} object within {@link #stats}. All incoming
	* statistics should update the current {@code Stat} object.
	*/
	int currentStat;
	/**
	* Time that the current {@link Stat} object will expire and {@link
	* #currentStat} will be incremented.
	*/
	long pendingStatPeriodEnd;

	public Stats(int statCount, long snapshotDuration, long currentTime) {
	this.snapshotDurationMs = snapshotDuration;

	this.stats = new Stat[statCount];
	// Pre-allocate all Stat objects
	for (int i = 0; i < this.stats.length; i++) {
	this.stats[i] = new Stat();
	}

	// Initialize expiration times for pendingStat objects
	long duration = this.snapshotDurationMs;
	this.pendingStatPeriodEnd = ((currentTime / duration) * duration)
	+ duration;
	}

	/**
	* Retrive the current {@code Stat} object that applies to {@code
	* currentTime}. {@code currentTime} is expected to be an actual point in
	* time while the caller was holding the lock on {@link Journal}. It may
	* never be less than the previous call.
	*/
	public Stat getCurrentStat(long currentTime) {
	// Check if the current Stat object is still valid to write to
	if (pendingStatPeriodEnd > currentTime) {
	return stats[currentStat];
	}
	// Check if all the Stat objects are invalid. This occurs when the last
	// request was a long time ago.
	if (currentTime - pendingStatPeriodEnd
	> snapshotDurationMs * stats.length) {
	for (int i = 0; i < stats.length; i++) {
	stats[i].reset();
	}
	long duration = snapshotDurationMs;
	pendingStatPeriodEnd = ((currentTime / duration) * duration) + duration;
	}
	// Walk through time to get the current Stat object
	while (pendingStatPeriodEnd <= currentTime) {
	currentStat++;
	currentStat %= stats.length;
	stats[currentStat].reset();
	pendingStatPeriodEnd += snapshotDurationMs;
	}
	return stats[currentStat];
	}

	public Object clone() throws CloneNotSupportedException {
	Stats statsClone = (Stats) super.clone();
	statsClone.stats = new Stat[stats.length];
	for (int i = 0; i < stats.length; i++) {
	statsClone.stats[i] = (Stat) stats[i].clone();
	}
	return statsClone;
	}
	}

	/**
	* Structure for holding statistics data. Data applies to a time period only,
	* as controlled by {@link Stats}.
	*/
	static class Stat implements Cloneable {
	/**
	* The number of responses sent.
	*/
	long requestResponsesCount;
	/**
	* The total duration of responses sent.
	*/
	long requestResponsesDurationSum;
	/**
	* The maximal duration of any one response.
	*/
	long requestResponsesMaxDuration;
	/**
	* Number of bytes sent to client.
	*/
	long requestResponsesThroughput;

	/**
	* The number of requests processed.
	*/
	long requestProcessingsCount;
	/**
	* The total duration of response processings.
	*/
	long requestProcessingsDurationSum;
	/**
	* The maximal duration of any one response processing.
	*/
	long requestProcessingsMaxDuration;
	/**
	* Number of bytes generated by the adaptor.
	*/
	long requestProcessingsThroughput;

	public Stat() {
	reset();
	}

	/**
	* Reset object for reuse.
	*/
	private void reset() {
	requestResponsesCount = 0;
	requestResponsesDurationSum = 0;
	requestResponsesMaxDuration = 0;
	requestResponsesThroughput = 0;

	requestProcessingsCount = 0;
	requestProcessingsDurationSum = 0;
	requestProcessingsMaxDuration = 0;
	requestProcessingsThroughput = 0;
	}

	public Object clone() throws CloneNotSupportedException {
	return super.clone();
	}
	}
	}