monitormanager.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

/*
 * Copyright (c) 2001 Sun Microsystems, Inc.  All rights
 * reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *       "This product includes software developed by the
 *       Sun Microsystems, Inc. for Project JXTA."
 *    Alternately, this acknowledgment may appear in the software itself,
 *    if and wherever such third-party acknowledgments normally appear.
 *
 * 4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA" must
 *    not be used to endorse or promote products derived from this
 *    software without prior written permission. For written
 *    permission, please contact Project JXTA at http://www.jxta.org.
 *
 * 5. Products derived from this software may not be called "JXTA",
 *    nor may "JXTA" appear in their name, without prior written
 *    permission of Sun.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED.  IN NO EVENT SHALL SUN MICROSYSTEMS OR
 * ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of Project JXTA.  For more
 * information on Project JXTA, please see
 * <http://www.jxta.org/>.
 *
 * This license is based on the BSD license adopted by the Apache Foundation.
 *
 * $Id: MonitorManager.java,v 1.11 2006/05/01 16:37:26 hamada Exp $
 */

package net.jxta.impl.meter;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;
import net.jxta.document.*;
import net.jxta.exception.*;
import net.jxta.id.*;
import net.jxta.impl.endpoint.endpointMeter.*;
import net.jxta.impl.util.TimeUtils;
import net.jxta.meter.*;
import net.jxta.peergroup.*;
import net.jxta.platform.*;
import net.jxta.protocol.*;
import net.jxta.service.*;
import net.jxta.util.*;
import net.jxta.util.documentSerializable.*;

public class MonitorManager implements Service {

	public static final int NOT_PULSING = -1;

	private static final int NO_PRIOR_REPORT = 0;

	private static long supportedReportRates[] = new long [] {
            500, TimeUtils.ASECOND, 
            5 * TimeUtils.ASECOND, 10 * TimeUtils.ASECOND,
            15 * TimeUtils.ASECOND, 30 * TimeUtils.ASECOND,
            TimeUtils.AMINUTE, 5 * TimeUtils.AMINUTE,
            10 * TimeUtils.AMINUTE, 15 * TimeUtils.AMINUTE,
            30 * TimeUtils.AMINUTE, TimeUtils.ANHOUR, 
            3 * TimeUtils.ANHOUR, 6 * TimeUtils.ANHOUR,
            12 * TimeUtils.ANHOUR, TimeUtils.ADAY,
            TimeUtils.AWEEK };

	private int pulsesPerRate[] = new int[supportedReportRates.length];
	private long timeZero = AbstractTimeStampedData.getTimeZero();
	private long startTime = System.currentTimeMillis();
	private long reportCycleBaseTime = -1;	// ie nothing has ever been reported

	private LinkedList monitorListenerInfos = new LinkedList();
	private Hashtable serviceMonitorPulseInfos = new Hashtable();
	private int filtersPerRate[] = new int[supportedReportRates.length];
	private long previousReportTimes[] = new long[supportedReportRates.length];

	private PeerGroup peerGroup;
	private Thread reportThread;

	private long  pulseRate = NOT_PULSING;
	private int pulseRateIndex = NOT_PULSING;
	private int pulseNumber = 0;
	private long nextPulseTime = NO_PRIOR_REPORT;
	private boolean isRunning = true;	// true until monitor is destroyed, triggers termination of report thread

	private ModuleClassID[] supportedModuleClassIDs;
	private Service monitorServiceInterface;
	private ModuleImplAdvertisement implAdvertisement;
	private long lastResetTime = System.currentTimeMillis();

	public Advertisement getImplAdvertisement() {
		return implAdvertisement;
	}
	
	public Service getInterface() {
            // This is good enough. No need to get fancy.
            return this;
	}


          
//	public MonitorManager(PeerGroup peerGroup) {	

	public void init(PeerGroup peerGroup, ID assignedID, Advertisement implAdvertisement) {
		this.implAdvertisement = (ModuleImplAdvertisement) implAdvertisement;
                this.peerGroup = peerGroup;
		createReportThread();

		long currentTime = System.currentTimeMillis();
		
		for (int i = 0; i < previousReportTimes.length; i++) 
			pulsesPerRate[i] = (int) (supportedReportRates[i] / supportedReportRates[0]);
	}
	
	
	public int startApp(java.lang.String[] args) {
		return 0; //fix-me
	}
	
	public void stopApp() {
		destroy();
	}

	private class MonitorListenerInfo {
		MonitorListener monitorListener;
		MonitorFilter monitorFilter;
		long reportRate;
		int reportRateIndex;
		boolean sendCumulativeFirst = false;
		boolean wasCumulativeSent = false;

		MonitorListenerInfo(MonitorListener monitorListener, long reportRate, MonitorFilter monitorFilter, boolean cumulativeFirst) {
			this.monitorListener = monitorListener;
			this.monitorFilter = monitorFilter;
			this.reportRate = reportRate;
			this.sendCumulativeFirst = cumulativeFirst;
			this.reportRateIndex = getReportRateIndex(reportRate);
		}
	}

	public static long[] getReportRates() {	// return copy so that users can't modify.
		long copy[] = new long[supportedReportRates.length];	
		System.arraycopy(supportedReportRates, 0, copy, 0, supportedReportRates.length);
		return copy;
	}

	public boolean isLocalMonitoringAvailable(ModuleClassID moduleClassID) {
		ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
		return (serviceMonitor != null);
	}

	public PeerGroup getPeerGroup() { return peerGroup; }

        // Cooperate with the code that loaded this module to replace the strong
        // group interface given by init() with a non-counted one.
        private void setPeerGroup(PeerGroup pg) {
            PeerGroup tmp = peerGroup;
            peerGroup = pg;
            tmp.unref();
            tmp = null;
        }

	public PeerMonitorInfo getPeerMonitorInfo() {
		long[] reportRates = getReportRates(); //makes a copy
		ModuleClassID[] moduleClassIDs = getMonitorableServiceTypes(); //ensures that array is initialized.
		long runningTime = System.currentTimeMillis() - lastResetTime;
		
		return new PeerMonitorInfo(MeterBuildSettings.METERING, moduleClassIDs, reportRates, lastResetTime, runningTime);
	}
	
	public int getReportRatesCount() { return supportedReportRates.length; }
	
	public int getReportRateIndex(long reportRate) {
			for (int i=0; i < supportedReportRates.length; i++) {
				if (supportedReportRates[i] == reportRate) 
					return i;
			}

			return -1;
	}

	public boolean isSupportedReportRate(long reportRate) {
		if (getReportRateIndex(reportRate) < 0)
			return false;
		else
			return true;
	}

	public long getReportRate(int index) {
		return supportedReportRates[index];
	}

	public long getBestReportRate(long desiredReportRate) {
		for (int i = 0; i < supportedReportRates.length; i++) {
			if (desiredReportRate <= supportedReportRates[i]) 
				return supportedReportRates[i];
		}

		return supportedReportRates[supportedReportRates.length-1];
	}

	public ServiceMonitor getServiceMonitor(ModuleClassID moduleClassID) {
		ServiceMonitorPulseInfo serviceMonitorPulseInfo = (ServiceMonitorPulseInfo) serviceMonitorPulseInfos.get(moduleClassID);

		if (serviceMonitorPulseInfo != null)
			return serviceMonitorPulseInfo.serviceMonitor;
		else {

			try {
				ModuleImplAdvertisement moduleImplAdvertisement = MonitorResources.getServiceMonitorImplAdvertisement(moduleClassID, implAdvertisement);
				ServiceMonitor serviceMonitor = (ServiceMonitor) peerGroup.loadModule(moduleClassID, moduleImplAdvertisement);
				MonitorResources.registerServiceMonitorModuleImplAdvertisement(moduleImplAdvertisement);
				
				if(serviceMonitor instanceof ServiceMonitorImpl) {
					((ServiceMonitorImpl)serviceMonitor).init(this);

				}
				
				serviceMonitorPulseInfo = new ServiceMonitorPulseInfo(this, serviceMonitor);
				serviceMonitorPulseInfos.put(moduleClassID, serviceMonitorPulseInfo);
				return serviceMonitor;
			} catch (JxtaException e) {
				throw new RuntimeException("Unable to load Service Monitor: " + moduleClassID + "\n\tException: " + e);
			}
		}
	}

	private void resetPulseRate() {
		int oldPulseRateIndex = pulseRateIndex;
		pulseRateIndex = NOT_PULSING;
		pulseRate = NOT_PULSING;
		
		for (int i=0; i < filtersPerRate.length; i++) {
			if (filtersPerRate[i] != 0) {
				pulseRateIndex = i;
				pulseRate = getReportRate(pulseRateIndex);
				break;
			}
		}

		if (oldPulseRateIndex == pulseRateIndex)
			return;	// nothing changed

		long now = System.currentTimeMillis();
	

		if (oldPulseRateIndex == NOT_PULSING) { 					// 	case 1: No pulse to pulse			
			for (int i = 0; i < filtersPerRate.length; i++) {
				if (filtersPerRate[i] != 0)
					previousReportTimes[i] = now;
				else
					previousReportTimes[i] = NO_PRIOR_REPORT;
			}

			pulseNumber = 0;
			nextPulseTime = now + pulseRate;
		} else if (pulseRateIndex == NOT_PULSING) { 				//			case 2: pulse to No pulse
			// Do nothing
		} else if (pulseRateIndex < oldPulseRateIndex) {			//			case 3: pulse going to a faster pulse
			for (int i = pulseRateIndex; i < (oldPulseRateIndex-1); i++) {
				if (filtersPerRate[i] != 0)
					previousReportTimes[i] = now;
				else
					previousReportTimes[i] = NO_PRIOR_REPORT;
			}

			long timeToNextPulse = nextPulseTime - now;

			if (pulseRate < timeToNextPulse) {
				int numPulsesToNow = (int) (timeToNextPulse/pulseRate);
				int numNewToOldPulses = (int) (supportedReportRates[oldPulseRateIndex] / supportedReportRates[pulseRateIndex]);

				pulseNumber += (numNewToOldPulses - numPulsesToNow) * pulsesPerRate[pulseRateIndex];
				timeToNextPulse = now - (numPulsesToNow * pulseRate);
			} else 
				pulseNumber += (pulsesPerRate[oldPulseRateIndex] - pulsesPerRate[pulseRateIndex]);

		} else if (pulseRateIndex > oldPulseRateIndex) {			//			case 3: pulse going to a  slower pulse
			int nextPulseNumber = pulseNumber + pulsesPerRate[oldPulseRateIndex];
			pulseNumber = ((nextPulseNumber-1) / pulsesPerRate[pulseRateIndex]) * pulsesPerRate[pulseRateIndex];
			nextPulseTime += (nextPulseNumber - pulseNumber) * supportedReportRates[0];

			for(int i=0; i < pulseRateIndex; i++)
				previousReportTimes[i] = NO_PRIOR_REPORT;
		}

		reportThread.interrupt();
	}

	private MonitorReport getMonitorReport(MonitorFilter monitorFilter, long reportRate, long previousDeltaTime, long beginReportTime) {
		MonitorReport monitorReport = new MonitorReport(previousDeltaTime, beginReportTime, false);

		for (Iterator i = monitorFilter.getModuleClassIDs(); i.hasNext(); ) {
			ModuleClassID moduleClassID = (ModuleClassID)i.next();

			ServiceMonitorFilter serviceMonitorFilter = monitorFilter.getServiceMonitorFilter(moduleClassID);	

			ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
			
			if (serviceMonitorFilter != null) {
				ServiceMetric serviceMetric = serviceMonitor.getServiceMetric(serviceMonitorFilter, previousDeltaTime, beginReportTime, getReportRateIndex(reportRate), reportRate);

				if (serviceMetric != null)
					monitorReport.addServiceMetric(serviceMetric);
			}
		}
		return monitorReport;
	}

	public void validateCumulativeMonitorFilter(MonitorFilter monitorFilter) throws MonitorFilterException{
		boolean isAnyServiceFilters = false;
		
		for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext(); ) {
			ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter)i.next();

			ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
			ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);

			if (serviceMonitor == null)
				throw new MonitorFilterException(MonitorFilterException.SERVICE_NOT_SUPPORTED, moduleClassID);


			serviceMonitor.validateCumulativeServiceMonitorFilter(serviceMonitorFilter);
			isAnyServiceFilters = true;
		}

		if (!isAnyServiceFilters)
			throw new MonitorFilterException("Empty Monitor Filter");
			

	}

	public void validateMonitorFilter(MonitorFilter monitorFilter, long reportRate) throws MonitorFilterException{

		if (!isSupportedReportRate(reportRate)) 
			throw new MonitorFilterException(MonitorFilterException.REPORT_RATE_NOT_SUPPORTED, reportRate);

		boolean isAnyServiceFilters = false;

		for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext(); ) {
			ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter)i.next();
			
			ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
			ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);

			if(serviceMonitor == null)
				throw new MonitorFilterException(MonitorFilterException.SERVICE_NOT_SUPPORTED, moduleClassID);		

			
			serviceMonitor.validateServiceMonitorFilter(serviceMonitorFilter, reportRate);
			isAnyServiceFilters = true;
		}
		
		if (!isAnyServiceFilters)
			throw new MonitorFilterException("Empty Monitor Filter");	
	}

	public MonitorFilter createSupportedCumulativeMonitorFilter(MonitorFilter monitorFilter) throws MonitorFilterException{
		MonitorFilter newMonitorFilter = new MonitorFilter(monitorFilter.getDescription());
		boolean anythingAdded = false;

		for (Iterator i = monitorFilter.getServiceMonitorFilters(); i.hasNext(); ) {
			ServiceMonitorFilter serviceMonitorFilter = (ServiceMonitorFilter)i.next();
			
			ModuleClassID moduleClassID = serviceMonitorFilter.getModuleClassID();
			ServiceMonitor serviceMonitor = getServiceMonitor(moduleClassID);
			
			if (serviceMonitor == null)
				continue;
			
			ServiceMonitorFilter newServiceMonitorFilter = serviceMonitor.createSupportedCumulativeServiceMonitorFilter(serviceMonitorFilter);