autofillinputbufferarray.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 AutoFillInputBufferArray 
{

	//private NonBlockingServer sockServer;
//	private ServerSocketChannel selectableChannel = null;
//	private Selector selector = null;
	
	private AutoFillInputBuffer []inArray = new AutoFillInputBuffer[DefConstants.MAX_NUM_IN_BUFFER];
	private boolean []bValidArray = new boolean[DefConstants.MAX_NUM_IN_BUFFER];
	private Object synValidArray, synValidBuf;

	private int size = 0;
	private BufferNotification bufNotification = null;

	private static Log log = LogFactory.getLog(AutoFillInputBufferArray.class.getName());
/*	Hashtable inBufTable = new Hashtable();
	Hashtable streamNameTable; 
	Hashtable inStreamTable = new Hashtable();
*/
	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 AutoFillInputBuffer getInputBuffer(int index)
	{
		int curSize = howmanyInputBuffers();
		if(index < 0 || index >= curSize)
			return null;

		return inArray[index];
	}

/*	public AutoFillInputBuffer getInputBuffer(ConnectionContext cc)
	{
		int i = 0;
		while( i < size )
		{
			inArray[index].
	}*/
	public void setInputBufStatus(int index, boolean bValid)
	{
                synchronized(synValidArray)
		{
			bValidArray[index] = bValid;
		}

		AutoFillInputBuffer inBuf = getInputBuffer(index);
		if(inBuf != null)
			inBuf.setBufferStatus(bValid);
	}


	public synchronized int howmanyInputBuffers()
	{
		return size;
	}


	public AutoFillInputBufferArray()
	{
		synValidArray = new Object();
		synValidBuf = new Object();
	} 

	public synchronized int addInputBuffer(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_IN_BUFFER)
		{
			log.warn("can't allocate more input 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.

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

		///////////////////////////////////
		//The available port the socket server
		//listens to can be chosen by the socket
		//API if the "port" parameter of bind call
		//is set to 0
		//
		//After binding the socket, we can use getLocalSocketAddress
		//to get port no.

		AutoFillInputBuffer in = new AutoFillInputBuffer();
		inArray[size] = in;
		bValidArray[size]=true;

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

		//run the socket server that listens
		//connections from clients
		/////////////////////////////////////
		//Note: after run_sock, a thread will be generated automatically
		in.start();

		size ++;
		return size - 1;
	}

	public void adjustRtnBufSizeTimes(ConnectionContext cc)
	{
		//adjust the rtnBufSizeTimes 
		//according to comparsions of all buffer size
		int number = howmanyInputBuffers();
		if(number <= 0)
			return;
		
		int size[]= new int[number];
		int temp;
		AutoFillInputBuffer in;
		
		int i, minIndex, minSize;
		minIndex = -1;
		minSize = 100000;
		for(i = 0; i < number ; i++)
		{
			size[i] = getInputBuffer(i).getBufferSize();
			if(minSize > size[i] && size[i] != 0)
			{
				minSize = size[i];
				minIndex = i;
			}
		}

		if(minIndex == -1) 
			return;
		
		for(i = 0; i < number ; i++)
			getInputBuffer(i).setRntBufSizeTimes((int)((float)(size[i])/(float)(minSize)));
		
	//	return isMajorityOverloaded();
	}

	public void initAdaptationPara()
	{
		int i, number;
		AutoFillInputBuffer in_others;

		number = howmanyInputBuffers();

		for(i = 0; i < number; i++)
		{
			in_others = getInputBuffer(i);
			in_others.initAdaptationPara();
		}
	}
		
	public boolean isMajorityOverloaded()
	{
		int i, number, number_overloaded;
		AutoFillInputBuffer in_others;

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

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

	}

	public boolean isMajorityLightloaded()
	{
		int i, number, number_lightloaded;
		AutoFillInputBuffer in_others;

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

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

}