connectioncontext.java

本人历尽千辛万苦找的clustream中的jar包

package org.osu.ogsa.stream.util;

import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.nio.charset.*;
import java.net.*;
import java.util.*;
import org.apache.log4j.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class ConnectionContext extends Object {


	public String senderHostName = null;
	public int senderPort = -1;
	public InetAddress senderAddr = null;

	public String rcverHostName = null;
	public int rcverPort = -1;
	public InetAddress rcverAddr = null;

	public String strId = null;

	public int iStep = -1;
	public BufferNotification bufNotification = null;
	public int ioBufferIndex = -1;
	public int in_or_out = -1;
	public String misc_info;
        public double net_bandwidth = -1;
        public double net_util = 3;

//	public Object neighStream = null;
	public String neighStreamHandle = null;
	
	//The following parameters are for the adaptation algorithm

	public int times_overloaded = 0;
	public int times_lightloaded = 0;
	/* The window is used to store the latest 5 loaded messages 
	 * The size of the window is set to 5. Since the order of the messages 
	 * need not be kept, so it is not necessary to declare an array.
	 */
	public int window = 0;
	public double average_loaded = 0;
	public double longTermLoadedFactor  = 0.0;

	//The following parameter is for debugging the adaptation algorithm
	public double loaded;

	//The following two paras are for determining how severely "traffic" is
	public int severity;
	public int numLongTermLoaded;
	private boolean bDegrading = false;
	private boolean bCalled = false;
	
        private static Log log = LogFactory.getLog(ConnectionContext.class.getName());

	public void initAdaptationPara()
	{
		times_overloaded = 0;
		times_lightloaded = 0;
		window = 0;
		average_loaded = 0;
		longTermLoadedFactor = 0;
		severity = DefConstants.LV_NORMAL;
		numLongTermLoaded = 0;
	}

/*	public void fillSimContext(SimConnectionContext simCC)
	{
		simCC.senderHostName = senderHostName;
		simCC.senderPort = senderPort;
		simCC.rcverHostName = rcverHostName;
		simCC.rcverPort = rcverPort;

		simCC.strId = strId;
		simCC.iStep = iStep;
		simCC.in_or_out = in_or_out;
//		simCC.neighStream = neighStream; 
		simCC.neighStreamHandle = neighStreamHandle; 
	} */
	public void changingTrafficInfo()
	{
		if(longTermLoadedFactor >=  DefConstants.LONGTERM_EXTREMELY_OVER_LOADED_THRESHOLD)
			//for determine if the input or output is bottleneck
		{
			numLongTermLoaded ++;
			if(numLongTermLoaded >= DefConstants.BOTTLENECT_SEVERITY_THRESHOLD[severity])
			{
				severity ++;
				if(severity == DefConstants.LV_SEVERE)
					setCalledFlag(false);
				if(severity > DefConstants.LV_CRITICAL)
				{
					severity = DefConstants.LV_CRITICAL;
					numLongTermLoaded =  DefConstants.BOTTLENECT_SEVERITY_THRESHOLD[severity];
				}
				else
					numLongTermLoaded = 0;
			}
		}
		else
		{
			numLongTermLoaded --;

			if(numLongTermLoaded < 0) 
			//degrade
			{
				if(severity == DefConstants.LV_CRITICAL)
				{
					setDegradingFlag(true);
					//setCalledFlag(false);
				}
				severity --;
				if(severity < DefConstants.LV_NORMAL)
				{
					severity = DefConstants.LV_NORMAL;
					numLongTermLoaded = 0;
				}
				else
					numLongTermLoaded = DefConstants.BOTTLENECT_SEVERITY_THRESHOLD[severity];
			}
		}
		log.debug(misc_info + ">longTermLoadedFactor" +longTermLoadedFactor + "|severity:" + severity );
		bufNotification.handleBottleneck(this);
	}


        public double setLongTermLoadedFactor()
        {
                double part1, part2, part3, curLoadedFactor;
		log.info(misc_info + "--->average_loaded:" + average_loaded + " times_overloaded:"+times_overloaded + " times_lightloaded:"+times_lightloaded + " window:" + window);
                //part 1 (t1 - t2)/(t1 + t2);
                if(times_overloaded == 0  && times_lightloaded == 0)
                        part1 = 0; //the previous value is 0
                else
                        part1 = (double)(times_overloaded - times_lightloaded) / (double) (times_overloaded + times_lightloaded);
                                                                                                                             
                //Part 2 t3/|t3| * (e/2)^(|t3| - window_size)
/*		if(window == 0)
                	part2 = (double)(Math.exp(-1 * DefConstants.WINDOW_SIZE))/(double)Math.pow(2, (-1 * DefConstants.WINDOW_SIZE));
		else
	                part2 = (window/Math.abs(window)) *(double)(Math.exp(Math.abs(window)- DefConstants.WINDOW_SIZE))/(double)Math.pow(2, (Math.abs(window)- DefConstants.WINDOW_SIZE));*/
		if(window == 0)
                	part2 = 0;
		else
	                part2 = (window/Math.abs(window)) *(double)(Math.exp(Math.abs(window)- DefConstants.WINDOW_SIZE))/(double)Math.pow(2, (Math.abs(window)- DefConstants.WINDOW_SIZE));
                //Part 3 average -
                if((part3 = average_loaded - DefConstants.FULLY_LOADED) > 0)
                        part3 /= 1 - DefConstants.FULLY_LOADED;
                else
                        part3 /= DefConstants.FULLY_LOADED;
                //combine part1, part2, and part3
                curLoadedFactor = (DefConstants.P1*part1 + DefConstants.P2*part2 + DefConstants.P3*part3);
		log.debug(misc_info + "--->longTermLoadedFactor:" +longTermLoadedFactor+ " part1:" + part1 +" part2:" + part2 + " part3:" + part3 + " curloadedFactor:"+curLoadedFactor);
		longTermLoadedFactor = DefConstants.LEARNING_RATE * curLoadedFactor + (1 - DefConstants.LEARNING_RATE) * longTermLoadedFactor;

                return longTermLoadedFactor;
        }

	public synchronized void setDegradingFlag(boolean flag)
	{
		bDegrading = flag;
	}

	public synchronized boolean getDegradingFlag()
	{
		return bDegrading;
	}

	public synchronized void setCalledFlag(boolean flag)
	{
		bCalled = flag;
	}

	public synchronized boolean getCalledFlag()
	{
		return bCalled;
	}

	public ConnectionContext()
	{
	}

/*	public ConnectionContext(SimConnectionContext simCC)
	public ConnectionContext()
	{
	}

/*	public ConnectionContext(SimConnectionContext simCC)
	{
		fillContext(simCC);
	}



	private void fillContext(SimConnectionContext simCC)
	{
		senderHostName = simCC.senderHostName;
		senderPort = simCC.senderPort;
		rcverHostName = simCC.rcverHostName;
		rcverPort = simCC.rcverPort;

		strId = simCC.strId;
		iStep = simCC.iStep;
		in_or_out = simCC.in_or_out;
//		neighStream = simCC.neighStream; 
		neighStreamHandle = simCC.neighStreamHandle; 
	}*/

}