autofilloutputbufferarray.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.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

//I should change inArray to HashTable class if I have a spare!!!!
public class AutoFillOutputBufferArray 
{

	private static Log log = LogFactory.getLog(AutoFillOutputBufferArray.class.getName());
	private AutoFillOutputBuffer []outArray = new AutoFillOutputBuffer[DefConstants.MAX_NUM_OUT_BUFFER];
	private boolean []bValidArray = new boolean[DefConstants.MAX_NUM_OUT_BUFFER];
	private int size = 0;
	private AutoFillOutputBuffer validOutBuffer = null;
	private BufferNotification bufNotification = null;
	private Object synValidArray, synValidBuf;
	
	public AutoFillOutputBufferArray()
	{
		synValidArray = new Object();
		synValidBuf = new Object();

	} 

	public void setNotification(BufferNotification bufNot)
	{
		bufNotification = bufNot;
	}

/*	public void setupStreams(Hashtable ht)
	{
		streamNameTable = ht;
	}

    	public SocketChannel acceptClient() throws InterruptedException, IOException
    	{
	     SocketChannel sChannel = this.selectableChannel.accept();
	     if(sChannel == null)
		     throw new InterruptedException();

	     return sChannel;
    	}*/

	public boolean isOutputBufValid(int index)
	{
		boolean temp;
		synchronized(synValidArray)
		{
			temp = bValidArray[index];
		}
		return temp;
	}

	public void setOutputBufStatus(int index, boolean bValid)
	{
		synchronized(synValidArray)
		{
			bValidArray[index] = bValid;
		}
		AutoFillOutputBuffer outBuf = getOutputBuffer(index);
		if(outBuf != null)
			outBuf.setBufferStatus(bValid);

	}

	
	public AutoFillOutputBuffer getOutputBuffer(int index)
	{
		int curSize = howmanyOutputBuffers();
		if(index < 0 || index >= curSize)
			return null;

		return outArray[index];
	}

	public AutoFillOutputBuffer getOutputBuffer()
	{//determine which output buffer should return

		int curSize = howmanyOutputBuffers();
		AutoFillOutputBuffer outBuf;
		for(int index = 0; index < curSize; index ++)
		{
			outBuf = outArray[index];
			if(isOutputBufValid(index))
			{
				return outBuf;
			}
		}
		return null;
	}

/*	public AutoFillOutputBuffer getValidOutputBuffer()
	{
		AutoFillOutputBuffer tempOutBuffer;
		synchronized(synValidBuf)
		{
			tempOutBuffer = validOutBuffer;
		}
		return tempOutBuffer;
	}


	public void setValidOutputBuffer(AutoFillOutputBuffer outBuf)
	{
		synchronized(synValidBuf)
		{
			validOutBuffer = outBuf;
		}
	} */

	public synchronized int howmanyOutputBuffers()
	{
		return size;
	}



	public synchronized int addOutputBuffer(ConnectionContext context)
	{
		if(context.iStep == -1 || context.strId == null)
			log.warn("have to set the values of iStep and strId in ConnectionContext object");

		if(size == DefConstants.MAX_NUM_OUT_BUFFER)
		{
			log.warn("can't allocate more output buffer");
			return -1;
		}


		//***********************************************
		// I need to change the second control statement 
		// to context.URL != null or something like this
		// Because the web service shouldn't be aware of
		// the detail of socket ports etc.
		// **********************************************
		// You have to make sure that the server's port has
		// been set in teh ConnectionContext

		if(context == null)
		{
			log.error("the connect context can't be null");
			return -1;
		}
		else if(context.rcverPort <= 0)
		{
			log.error("A port the socket server listens to doesn't be setup yet");
			return -1;
		} 
		else if(this.bufNotification == null)
		{
			log.error("please setup Notification Object in the web service");
			return -1;
		}
		

		AutoFillOutputBuffer out = new AutoFillOutputBuffer();
		outArray[size] = out;
		bValidArray[size] = true;

		//Setup the connection context, and pass
		//the context object to OutputBufferArray in
		context.bufNotification = this.bufNotification;
		context.ioBufferIndex = size;
		context.in_or_out = DefConstants.OUT_BUFFER;
		out.setConnectionContext(context);

		//run the socket server that listens
		//connections from clients
		/////////////////////////////////////
		//Note: after start(), a thread will be generated automatically
		out.start();
		
		size ++;
		return size - 1;
	}

	public void adjustRtnBufSizeTimes(ConnectionContext cc)
	{
		//adjust the rtnBufSizeTimes 
		//according to comparsions of all buffer size
	//	return isMajorityOverloaded();
	}

	public boolean isMajorityOverloaded()
	{
		int i, number, number_overloaded;
		AutoFillOutputBuffer out_others;

		number = howmanyOutputBuffers();
		if(number == 0)
			return false;
		number_overloaded = 0;

		for(i = 0; i < number; i++)
		{
			out_others = getOutputBuffer(i);
			if(out_others.isBufferOverloaded() == true)
			number_overloaded ++;
		}
		return (((double)number_overloaded / (double)number) > DefConstants.MAJORITY_BUF_THRESHOLD);
	}

        public void initAdaptationPara()
        {
                int i, number;
                AutoFillOutputBuffer out_others;
                                                                                                                             
                number = howmanyOutputBuffers();
                                                                                                                             
                for(i = 0; i < number; i++)
                {
                        out_others = getOutputBuffer(i);
                        out_others.initAdaptationPara();
                }
        }


	public boolean isMajorityLightloaded()
	{
		int i, number, number_lightloaded;
		AutoFillOutputBuffer out_others;

		number = howmanyOutputBuffers();
		if(number == 0)
			return false;
		number_lightloaded = 0;

		for(i = 0; i < number; i++)
		{
			out_others = getOutputBuffer(i);
			if(out_others.isBufferLightloaded() == true)
			number_lightloaded ++;
		}
		return (((double)number_lightloaded / (double)number) > DefConstants.MAJORITY_BUF_THRESHOLD);
	}

	public void setNetBandwidth(String strDownstreamHandle, double [] seqBgTraffic, double util)
	{
		int i;
		ConnectionContext cc;
		int curSize = howmanyOutputBuffers();
		for(i = 0; i < curSize ; i ++)
		{
			cc = outArray[i].getConnectionContext();
			if(strDownstreamHandle.equals(cc.neighStreamHandle))
			{
				outArray[i].setSeqBgTraffic(seqBgTraffic);
				outArray[i].setNetUtil(util);
			}
		}
	}
			
}