[Udt-java-commits] SF.net SVN: udt-java:[32] udt-java/trunk/src

[<bsc...@us...>]

Revision: 32
          http://udt-java.svn.sourceforge.net/udt-java/?rev=32&view=rev
Author:   bschuller
Date:     2010-04-30 14:02:45 +0000 (Fri, 30 Apr 2010)

Log Message:
-----------


Modified Paths:
--------------
    udt-java/trunk/src/main/java/udt/ClientSession.java
    udt-java/trunk/src/main/java/udt/CongestionControl.java
    udt-java/trunk/src/main/java/udt/ServerSession.java
    udt-java/trunk/src/main/java/udt/UDTCongestionControl.java
    udt-java/trunk/src/main/java/udt/UDTInputStream.java
    udt-java/trunk/src/main/java/udt/UDTReceiver.java
    udt-java/trunk/src/main/java/udt/UDTSender.java
    udt-java/trunk/src/main/java/udt/UDTSession.java
    udt-java/trunk/src/main/java/udt/sender/SenderLossList.java
    udt-java/trunk/src/main/java/udt/util/MeanValue.java
    udt-java/trunk/src/main/java/udt/util/UDTStatistics.java
    udt-java/trunk/src/main/java/udt/util/Util.java
    udt-java/trunk/src/test/java/udt/TestUDTInputStream.java
    udt-java/trunk/src/test/java/udt/TestUDTServerSocket.java
    udt-java/trunk/src/test/java/udt/UDTTestBase.java
    udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java

Added Paths:
-----------
    udt-java/trunk/src/main/java/udt/cc/
    udt-java/trunk/src/main/java/udt/cc/SimpleTCP.java
    udt-java/trunk/src/test/java/udt/performance/TestUDTLargeDataCC1.java

Modified: udt-java/trunk/src/main/java/udt/ClientSession.java
===================================================================
--- udt-java/trunk/src/main/java/udt/ClientSession.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/ClientSession.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -69,17 +69,15 @@
 	
 	public void connect() throws InterruptedException,IOException{
 		int n=0;
-		
 		sendHandShake();
-		
 		while(getState()!=ready){
 			if(getState()==invalid)throw new IOException("Can't connect!");
 			n++;
 			if(getState()!=ready)Thread.sleep(500);
 		}
+		cc.init();
 		logger.info("Connected, "+n+" handshake packets sent");		
 	}
-
 	
 	@Override
 	public void received(UDTPacket packet, Destination peer) {

Modified: udt-java/trunk/src/main/java/udt/CongestionControl.java
===================================================================
--- udt-java/trunk/src/main/java/udt/CongestionControl.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/CongestionControl.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -15,45 +15,63 @@
 
 	/**
 	 * set roundtrip time and associated variance
+	 * @param rtt - round trip time in microseconds
+	 * @param rttVar - round trip time variance in microseconds
 	 */
 	public abstract void setRTT(long rtt, long rttVar);
 
 	/**
 	 * update packet arrival rate and link capacity with the
 	 * values received in an ACK packet
-	 * @param rate
-	 * @param linkCapacity
+	 * @param rate - packet rate in packets per second
+	 * @param linkCapacity - estimated link capacity in packets per second
 	 */
 	public abstract void updatePacketArrivalRate(long rate, long linkCapacity);
 
+	/**
+	 * get the current value of the packet arrival 
+	 */
 	public long getPacketArrivalRate();
 	
+	/**
+	 * get the current value of the estimated link capacity 
+	 */
 	public long getEstimatedLinkCapacity();
-
 	
 	/**
-	 * Inter-packet interval in seconds
-	 * @return 
+	 * get the current value of the inter-packet interval in microseconds
 	 */
 	public abstract double getSendInterval();
 
 	/**
 	 * get the congestion window size
 	 */
-	public abstract long getCongestionWindowSize();
+	public abstract double getCongestionWindowSize();
 	
 	/**
+	 * get the ACK interval. If larger than 0, the receiver should acknowledge
+	 * every n'th packet
+	 */
+	public abstract long getAckInterval();
+	
+	/**
+	 * set the ACK interval. If larger than 0, the receiver should acknowledge
+	 * every n'th packet
+	 */
+	public abstract void setAckInterval(long ackInterval);
+	
+	/**
 	 * Callback function to be called when an ACK packet is received.
-	 * @param ackSeqno: the data sequence number acknowledged by this ACK.
+	 * @param ackSeqno - the data sequence number acknowledged by this ACK.
 	 * see spec. page(16-17)
 	 */
 	public abstract void onACK(long ackSeqno);
 
 	/**
-	 *  Callback function to be called when a loss report is received.
-	 * @param lossInfo:list of sequence number of packets, in the format describled in packet.cpp.
+	 * Callback function to be called when a loss report is received.
+	 * @param lossInfo - list of sequence number of packets
 	 */
-	public abstract void onNAK(List<Integer> lossInfo);
+	public abstract void onLoss(List<Integer> lossInfo);
 
 	/**
 	 * Callback function to be called when a timeout event occurs
@@ -62,13 +80,13 @@
 
 	/**
 	 * Callback function to be called when a data packet is sent.
-	 * @param packetSeqNo: the data sequence number.
+	 * @param packetSeqNo - the data packet sequence number
 	 */
 	public abstract void onPacketSend(long packetSeqNo);
 
 	/**
 	 * Callback function to be called when a data packet is received.
-	 * @param packetSeqNo: the data sequence number.
+	 * @param packetSeqNo - the data packet sequence number.
 	 */
 	public abstract void onPacketReceive(long packetSeqNo);
 

Modified: udt-java/trunk/src/main/java/udt/ServerSession.java
===================================================================
--- udt-java/trunk/src/main/java/udt/ServerSession.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/ServerSession.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -80,6 +80,7 @@
 				try{
 					setState(ready);
 					socket=new UDTSocket(endPoint, this);
+					cc.init();
 				}catch(Exception uhe){
 					//session is invalid
 					logger.log(Level.SEVERE,"",uhe);

Modified: udt-java/trunk/src/main/java/udt/UDTCongestionControl.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTCongestionControl.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/UDTCongestionControl.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -16,33 +16,30 @@
 
 	private static final Logger logger=Logger.getLogger(UDTCongestionControl.class.getName());
 
-	private final UDTSession session;
+	protected final UDTSession session;
 
-	private final UDTStatistics statistics;
+	protected final UDTStatistics statistics;
 
 	//round trip time in microseconds
-	private long roundTripTime=0;
+	protected long roundTripTime=0;
 
 	//rate in packets per second
-	private long packetArrivalRate=0;
+	protected long packetArrivalRate=0;
 
 	//link capacity in packets per second
-	private long estimatedLinkCapacity=0;
+	protected long estimatedLinkCapacity=0;
 
 	// Packet sending period = packet send interval, in microseconds
-	private double packetSendingPeriod=1;              
+	protected double packetSendingPeriod=1;              
 
 	// Congestion window size, in packets
-	private long congestionWindowSize=16;
+	protected double congestionWindowSize=16;
 
-	//last rate increase time (microsecond value)
-	long lastRateIncreaseTime=Util.getCurrentTime();
-
 	/*if in slow start phase*/
-	boolean slowStartPhase=true;
+	private boolean slowStartPhase=true;
 
 	/*last ACKed seq no*/
-	long lastAckSeqNumber=-1;
+	private long lastAckSeqNumber=-1;
 
 	/*max packet seq. no. sent out when last decrease happened*/
 	private	long lastDecreaseSeqNo;
@@ -62,11 +59,13 @@
 	//this flag avoids immediate rate increase after a NAK
 	private boolean loss=false;
 
+	//if larger than 0, the receiver should acknowledge every n'th packet
+	protected long ackInterval=-1;
+	
 	public UDTCongestionControl(UDTSession session){
 		this.session=session;
 		this.statistics=session.getStatistics();
 		lastDecreaseSeqNo=session.getInitialSequenceNumber()-1;
-		init();
 	}
 
 	/* (non-Javadoc)
@@ -108,11 +107,21 @@
 		return packetSendingPeriod;
 	}
 
+	public long getAckInterval(){
+		return ackInterval;
+	}
+	public void setAckInterval(long ackInterval){
+		this.ackInterval=ackInterval;
+		if(session.getSocket().getReceiver()!=null){
+			session.getSocket().getReceiver().setAckInterval(ackInterval);
+		}
+	}
+	
 	/**
 	 * congestionWindowSize
 	 * @return
 	 */
-	public long getCongestionWindowSize(){
+	public double getCongestionWindowSize(){
 		return congestionWindowSize;
 	}
 
@@ -186,7 +195,7 @@
 	/* (non-Javadoc)
 	 * @see udt.CongestionControl#onNAK(java.util.List)
 	 */
-	public void onNAK(List<Integer>lossInfo){
+	public void onLoss(List<Integer>lossInfo){
 		loss=true;
 		long firstBiggestlossSeqNo=lossInfo.get(0);
 		nACKCount++;

Modified: udt-java/trunk/src/main/java/udt/UDTInputStream.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTInputStream.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/UDTInputStream.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -78,7 +78,8 @@
 	public UDTInputStream(UDTSocket socket, UDTStatistics statistics)throws IOException{
 		this.socket=socket;
 		this.statistics=statistics;
-		appData=new PriorityBlockingQueue<AppData>(socket.getSession().getFlowWindowSize());
+		int capacity=socket!=null? 4*socket.getSession().getFlowWindowSize() : 64 ;
+		appData=new PriorityBlockingQueue<AppData>(capacity);
 	}
 
 	/**
@@ -178,7 +179,6 @@
 				//check if the data is in-order
 				if(currentChunk.sequenceNumber==highestSequenceNumber+1){
 					highestSequenceNumber++;
-					//statistics.updateReadDataMD5(currentChunk.data);
 					return;
 				}
 				else if(currentChunk.sequenceNumber<=highestSequenceNumber){

Modified: udt-java/trunk/src/main/java/udt/UDTReceiver.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTReceiver.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/UDTReceiver.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -33,8 +33,8 @@
 package udt;
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -118,15 +118,15 @@
 	//to check the ACK, NAK, or EXP timer
 	private long nextACK;
 	//microseconds to next ACK event
-	private long ACK_INTERVAL=Util.getSYNTime();
+	private long ackTimerInterval=Util.getSYNTime();
 
 	private long nextNAK;
 	//microseconds to next NAK event
-	private long NAK_INTERVAL=Util.getSYNTime();
+	private long nakTimerInterval=Util.getSYNTime();
 
 	private long nextEXP;
 	//microseconds to next EXP event
-	private long EXP_INTERVAL=100*Util.getSYNTime();
+	private long expTimerInterval=100*Util.getSYNTime();
 
 	//instant when the session was created (for expiry checking)
 	private final long sessionUpSince;
@@ -143,6 +143,9 @@
 
 	private volatile boolean stopped=false;
 
+	//(optional) ack interval (see CongestionControl interface)
+	private volatile long ackInterval=-1;
+
 	/**
 	 * if set to true connections will not expire, but will only be
 	 * closed by a Shutdown message
@@ -165,10 +168,7 @@
 		packetPairWindow = new PacketPairWindow(16);
 		largestReceivedSeqNumber=session.getInitialSequenceNumber()-1;
 		bufferSize=session.getReceiveBufferSize();
-
-		//incoming packets are ordered by sequence number, with control packets having
-		//preference over data packets
-		handoffQueue=new PriorityBlockingQueue<UDTPacket>(session.getFlowWindowSize());
+		handoffQueue=new ArrayBlockingQueue<UDTPacket>(4*session.getFlowWindowSize());
 		initMetrics();
 		start();
 	}
@@ -194,9 +194,10 @@
 		Runnable r=new Runnable(){
 			public void run(){
 				try{
-					nextACK=Util.getCurrentTime()+ACK_INTERVAL;
-					nextNAK=(long)(Util.getCurrentTime()+1.5*NAK_INTERVAL);
-					nextEXP=Util.getCurrentTime()+2*EXP_INTERVAL;
+					nextACK=Util.getCurrentTime()+ackTimerInterval;
+					nextNAK=(long)(Util.getCurrentTime()+1.5*nakTimerInterval);
+					nextEXP=Util.getCurrentTime()+2*expTimerInterval;
+					ackInterval=session.getCongestionControl().getAckInterval();
 					while(!stopped){
 						receiverAlgorithm();
 					}
@@ -228,18 +229,18 @@
 		//check ACK timer
 		long currentTime=Util.getCurrentTime();
 		if(nextACK<currentTime){
-			nextACK=currentTime+ACK_INTERVAL;
+			nextACK=currentTime+ackTimerInterval;
 			processACKEvent(true);
 		}
 		//check NAK timer
 		if(nextNAK<currentTime){
-			nextNAK=currentTime+NAK_INTERVAL;
+			nextNAK=currentTime+nakTimerInterval;
 			processNAKEvent();
 		}
 
 		//check EXP timer
 		if(nextEXP<currentTime){
-			nextEXP=currentTime+EXP_INTERVAL;
+			nextEXP=currentTime+expTimerInterval;
 			processEXPEvent();
 		}
 		//perform time-bounded UDP receive
@@ -258,7 +259,7 @@
 				}
 			}
 			if(needEXPReset){
-				nextEXP=Util.getCurrentTime()+EXP_INTERVAL;
+				nextEXP=Util.getCurrentTime()+expTimerInterval;
 			}
 			processTime.begin();
 			processUDTPacket(packet);
@@ -367,11 +368,13 @@
 
 	//every nth packet will be discarded... for testing only of course
 	public static int dropRate=0;
+	
 	//number of received data packets
 	private int n=0;
 	
 	protected void onDataPacketReceived(DataPacket dp)throws IOException{
 		long currentSequenceNumber = dp.getPacketSequenceNumber();
+		
 		//check whether to drop this packet
 		n++;
 		if(dropRate>0 && n % dropRate == 0){
@@ -379,6 +382,7 @@
 			return;
 		}
 
+		
 		long currentDataPacketArrivalTime = Util.getCurrentTime();
 
 		/*(4).if the seqNo of the current data packet is 16n+1,record the
@@ -418,6 +422,11 @@
 		if(currentSequenceNumber>largestReceivedSeqNumber){
 			largestReceivedSeqNumber=currentSequenceNumber;
 		}
+
+		//(8) need to send an ACK? Some cc algorithms use this
+		if(ackInterval>0){
+			if(n % ackInterval == 0)processACKEvent(false);
+		}
 	}
 
 	/**
@@ -438,6 +447,7 @@
 			receiverLossList.insert(detectedLossSeqNumber);
 		}
 		endpoint.doSend(nAckPacket);
+		//logger.info("NAK for "+currentSequenceNumber);
 		statistics.incNumberOfNAKSent();
 	}
 
@@ -513,8 +523,8 @@
 			if(roundTripTime>0)roundTripTime = (roundTripTime*7 + rtt)/8;
 			else roundTripTime = rtt;
 			roundTripTimeVar = (roundTripTimeVar* 3 + Math.abs(roundTripTimeVar- rtt)) / 4;
-			ACK_INTERVAL=4*roundTripTime+roundTripTimeVar+Util.getSYNTime();
-			NAK_INTERVAL=ACK_INTERVAL;
+			ackTimerInterval=4*roundTripTime+roundTripTimeVar+Util.getSYNTime();
+			nakTimerInterval=ackTimerInterval;
 			statistics.setRTT(roundTripTime, roundTripTimeVar);
 		}
 	}
@@ -536,7 +546,7 @@
 	private volatile long ackSequenceNumber=0;
 
 	protected void resetEXPTimer(){
-		nextEXP=Util.getCurrentTime()+EXP_INTERVAL;
+		nextEXP=Util.getCurrentTime()+expTimerInterval;
 		expCount=0;
 	}
 
@@ -544,6 +554,10 @@
 		expCount=0;
 	}
 	
+	public void setAckInterval(long ackInterval){
+		this.ackInterval=ackInterval;
+	}
+	
 	protected void onShutdown()throws IOException{
 		stop();
 	}

Modified: udt-java/trunk/src/main/java/udt/UDTSender.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTSender.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/UDTSender.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -246,7 +246,7 @@
 
 		long ackNumber=acknowledgement.getAckNumber();
 		cc.onACK(ackNumber);
-		statistics.setCongestionWindowSize(cc.getCongestionWindowSize());
+		statistics.setCongestionWindowSize((long)cc.getCongestionWindowSize());
 		//need to remove all sequence numbers up the ack number from the sendBuffer
 		boolean removed=false;
 		for(long s=lastAckSequenceNumber;s<ackNumber;s++){
@@ -274,7 +274,7 @@
 		for(Integer i: nak.getDecodedLossInfo()){
 			senderLossList.insert(Long.valueOf(i));
 		}
-		session.getCongestionControl().onNAK(nak.getDecodedLossInfo());
+		session.getCongestionControl().onLoss(nak.getDecodedLossInfo());
 		session.getSocket().getReceiver().resetEXPTimer();
 		statistics.incNumberOfNAKReceived();
 		statistics.storeParameters();
@@ -306,17 +306,16 @@
 	/**
 	 * sender algorithm
 	 */
-	MeanValue v=new MeanValue("",true,128);
+	MeanValue v=new MeanValue("Wait for Ack time: ");
 	public void senderAlgorithm()throws InterruptedException, IOException{
+		statistics.addMetric(v);
 		while(!paused){
 
 			long iterationStart=Util.getCurrentTime(); //last packet send time?
 
 			//if the sender's loss list is not empty 
 			if (!senderLossList.isEmpty()) {
-				v.begin();
 				Long entry=senderLossList.getFirstEntry();
-				v.end();
 				handleResubmit(entry);
 			}
 
@@ -327,7 +326,7 @@
 				int unAcknowledged=unacknowledged.get();
 
 				if(unAcknowledged<session.getCongestionControl().getCongestionWindowSize()
-						&& unAcknowledged<session.getFlowWindowSize()){
+						 && unAcknowledged<session.getFlowWindowSize()){
 					//check for application data
 					DataPacket dp=sendQueue.poll();
 					if(dp!=null){
@@ -342,7 +341,9 @@
 					if(unAcknowledged>=session.getCongestionControl().getCongestionWindowSize()){
 						statistics.incNumberOfCCWindowExceededEvents();
 					}
+					v.begin();
 					waitForAck();
+					v.end();
 				}
 			}
 
@@ -455,7 +456,7 @@
 	 */
 	public void waitForAck()throws InterruptedException{
 		waitForAckLatch.set(new CountDownLatch(1));
-		waitForAckLatch.get().await(1000, TimeUnit.MILLISECONDS);
+		waitForAckLatch.get().await(2, TimeUnit.MILLISECONDS);
 	}
 
 

Modified: udt-java/trunk/src/main/java/udt/UDTSession.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTSession.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/UDTSession.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -34,6 +34,7 @@
 
 import java.net.DatagramPacket;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import udt.packets.Destination;
@@ -72,7 +73,7 @@
 	 * flow window size, i.e. how many data packets are
 	 * in-flight at a single time
 	 */
-	protected int flowWindowSize=128;
+	protected int flowWindowSize=4*128;
 
 	/**
 	 * remote UDT entity (address and socket ID)
@@ -110,21 +111,23 @@
 		mySocketID=nextSocketID.incrementAndGet();
 		this.destination=destination;
 		this.dgPacket=new DatagramPacket(new byte[0],0,destination.getAddress(),destination.getPort());
-		//init configurable CC
 		String clazzP=System.getProperty(CC_CLASS,UDTCongestionControl.class.getName());
 		Object ccObject=null;
 		try{
 			Class<?>clazz=Class.forName(clazzP);
 			ccObject=clazz.getDeclaredConstructor(UDTSession.class).newInstance(this);
 		}catch(Exception e){
+			logger.log(Level.WARNING,"Can't setup congestion control class <"+clazzP+">, using default.",e);
 			ccObject=new UDTCongestionControl(this);
 		}
 		cc=(CongestionControl)ccObject;
-		System.out.println("using "+cc.getClass().getName());
+		logger.info("Using "+cc.getClass().getName());
 	}
 	
+	
 	public abstract void received(UDTPacket packet, Destination peer);
 	
+	
 	public UDTSocket getSocket() {
 		return socket;
 	}

Added: udt-java/trunk/src/main/java/udt/cc/SimpleTCP.java
===================================================================
--- udt-java/trunk/src/main/java/udt/cc/SimpleTCP.java	                        (rev 0)
+++ udt-java/trunk/src/main/java/udt/cc/SimpleTCP.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -0,0 +1,36 @@
+package udt.cc;
+
+import java.util.List;
+
+import udt.UDTCongestionControl;
+import udt.UDTSession;
+
+/**
+ * simple TCP CC algorithm from the paper 
+ * "Optimizing UDP-based Protocol Implementations" by Y. Gu and R. Grossmann
+ */
+public class SimpleTCP extends UDTCongestionControl {
+
+	public SimpleTCP(UDTSession session){
+		super(session);
+	}
+
+	@Override
+	public void init() {
+		packetSendingPeriod=0;
+		congestionWindowSize=2;
+		setAckInterval(2);
+	}
+
+	@Override
+	public void onACK(long ackSeqno) {
+		congestionWindowSize += 1/congestionWindowSize;
+	}
+
+	@Override
+	public void onLoss(List<Integer> lossInfo) {
+		congestionWindowSize *= 0.5;
+	}
+	
+	
+}


Property changes on: udt-java/trunk/src/main/java/udt/cc/SimpleTCP.java
___________________________________________________________________
Added: svn:mime-type
   + text/plain

Modified: udt-java/trunk/src/main/java/udt/sender/SenderLossList.java
===================================================================
--- udt-java/trunk/src/main/java/udt/sender/SenderLossList.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/sender/SenderLossList.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -48,6 +48,7 @@
 	}
 
 	public void insert(Long obj){
+		if(obj==null)System.out.println("!!!");
 		synchronized (backingList) {
 			if(!backingList.contains(obj)){
 				if(backingList.isEmpty())backingList.add(obj);
@@ -62,17 +63,13 @@
 		}
 	}
 
-	public void remove(long seqNo){
-		synchronized (backingList) {
-			backingList.remove(seqNo);	
-		}
-	}
-
 	/**
 	 * retrieves the loss list entry with the lowest sequence number
 	 */
 	public Long getFirstEntry(){
-		return backingList.poll();
+		synchronized(backingList){
+			return backingList.poll();
+		}
 	}
 
 	public boolean isEmpty(){

Modified: udt-java/trunk/src/main/java/udt/util/MeanValue.java
===================================================================
--- udt-java/trunk/src/main/java/udt/util/MeanValue.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/util/MeanValue.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -42,7 +42,7 @@
 	public void addValue(double value){
 		mean=(mean*n+value)/(n+1);
 		n++;
-		if(verbose &&  n % nValue == 1){
+		if(verbose &&  n % nValue == 0){
 			if(msg!=null)System.out.print(msg+" ");
 			System.out.println(getFormattedMean());
 		}

Modified: udt-java/trunk/src/main/java/udt/util/UDTStatistics.java
===================================================================
--- udt-java/trunk/src/main/java/udt/util/UDTStatistics.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/util/UDTStatistics.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -35,8 +35,6 @@
 import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -69,17 +67,10 @@
 	private volatile double sendPeriod;
 	private volatile long congestionWindowSize;
 
-	private MessageDigest digest;
-	
 	private final List<MeanValue>metrics=new ArrayList<MeanValue>();
 		
 	public UDTStatistics(String componentDescription){
 		this.componentDescription=componentDescription;
-		try{
-			digest=MessageDigest.getInstance("MD5");
-		}catch(NoSuchAlgorithmException na){
-			digest=null;
-		}
 	}
 
 	public int getNumberOfSentDataPackets() {
@@ -170,14 +161,6 @@
 		this.congestionWindowSize = congestionWindowSize;
 	}
 
-	public void updateReadDataMD5(byte[]data){
-		digest.update(data);
-	}
-
-	public String getDigest(){
-		return hexString(digest);
-	}
-
 	public long getPacketArrivalRate(){
 		return packetArrivalRate;
 	}
@@ -268,15 +251,4 @@
 		}
 	}
 
-	public static String hexString(MessageDigest digest){
-		byte[] messageDigest = digest.digest();
-		StringBuilder hexString = new StringBuilder();
-		for (int i=0;i<messageDigest.length;i++) {
-			String hex = Integer.toHexString(0xFF & messageDigest[i]); 
-			if(hex.length()==1)hexString.append('0');
-			hexString.append(hex);
-		}
-		return hexString.toString();
-	}
-
 }

Modified: udt-java/trunk/src/main/java/udt/util/Util.java
===================================================================
--- udt-java/trunk/src/main/java/udt/util/Util.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/main/java/udt/util/Util.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -38,6 +38,7 @@
 import java.io.OutputStream;
 import java.net.DatagramPacket;
 import java.net.InetAddress;
+import java.security.MessageDigest;
 
 import udt.UDPEndPoint;
 
@@ -149,5 +150,16 @@
 		p.setPort(clientPort);
 		endpoint.sendRaw(p);
 	}
+
+	public static String hexString(MessageDigest digest){
+		byte[] messageDigest = digest.digest();
+		StringBuilder hexString = new StringBuilder();
+		for (int i=0;i<messageDigest.length;i++) {
+			String hex = Integer.toHexString(0xFF & messageDigest[i]); 
+			if(hex.length()==1)hexString.append('0');
+			hexString.append(hex);
+		}
+		return hexString.toString();
+	}
 	
 }

Modified: udt-java/trunk/src/test/java/udt/TestUDTInputStream.java
===================================================================
--- udt-java/trunk/src/test/java/udt/TestUDTInputStream.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/test/java/udt/TestUDTInputStream.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -3,6 +3,7 @@
 import java.security.MessageDigest;
 
 import udt.util.UDTStatistics;
+import udt.util.Util;
 
 public class TestUDTInputStream extends UDTTestBase{
 
@@ -18,8 +19,8 @@
 		is.haveNewData(3, data3);
 		is.noMoreData();
 		is.setBlocking(false);
-		readAll(is,8);
-		assertEquals(digest,stat.getDigest());
+		String readMD5=readAll(is,8);
+		assertEquals(digest,readMD5);
 	}
 	
 	public void test2()throws Exception{
@@ -34,8 +35,8 @@
 		is.haveNewData(2, data2);
 		is.haveNewData(3, data3);
 		is.noMoreData();
-		readAll(is,5*1024*1024);
-		assertEquals(digest,stat.getDigest());
+		String readMD5=readAll(is,5*1024*1024);
+		assertEquals(digest,readMD5);
 	}
 	
 	public void testInOrder()throws Exception{
@@ -52,8 +53,8 @@
 		}
 		is.noMoreData();
 		
-		readAll(is,1024*999);
-		assertEquals(digest,stat.getDigest());
+		String readMD5 = readAll(is,1024*999);
+		assertEquals(digest,readMD5);
 	}
 	
 	public void testRandomOrder()throws Exception{
@@ -70,9 +71,9 @@
 		for(int i : order){
 			is.haveNewData(i+1, blocks[i]);
 		}
-		readAll(is,512,true);
+		String readMD5=readAll(is,512,true);
 		
-		assertEquals(digest,stat.getDigest());
+		assertEquals(digest,readMD5);
 	}
 	
 	//read and discard data from the given input stream
@@ -89,7 +90,7 @@
 				if(c>0)d.update(buf,0,c);
 			}
 		}
-		return UDTStatistics.hexString(d);
+		return Util.hexString(d);
 	}
 	
 }

Modified: udt-java/trunk/src/test/java/udt/TestUDTServerSocket.java
===================================================================
--- udt-java/trunk/src/test/java/udt/TestUDTServerSocket.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/test/java/udt/TestUDTServerSocket.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -6,7 +6,7 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-import udt.util.UDTStatistics;
+import udt.util.Util;
 
 public class TestUDTServerSocket extends UDTTestBase{
 	
@@ -115,7 +115,7 @@
 					}
 					System.out.println("Server thread exiting.");
 					serverRunning=false;
-					md5_received=UDTStatistics.hexString(md5);
+					md5_received=Util.hexString(md5);
 					serverSocket.shutDown();
 					System.out.println(s.getSession().getStatistics());
 				}

Modified: udt-java/trunk/src/test/java/udt/UDTTestBase.java
===================================================================
--- udt-java/trunk/src/test/java/udt/UDTTestBase.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/test/java/udt/UDTTestBase.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -5,7 +5,7 @@
 import java.security.MessageDigest;
 import java.util.Random;
 
-import udt.util.UDTStatistics;
+import udt.util.Util;
 
 import junit.framework.TestCase;
 
@@ -60,7 +60,7 @@
 	}
 
 	public static String hexString(MessageDigest digest){
-		return UDTStatistics.hexString(digest);
+		return Util.hexString(digest);
 	}
 
 	

Modified: udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java
===================================================================
--- udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java	2010-04-29 21:33:20 UTC (rev 31)
+++ udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -14,6 +14,7 @@
 import udt.UDTServerSocket;
 import udt.UDTSocket;
 import udt.UDTTestBase;
+import udt.util.Util;
 
 public class TestUDTLargeData extends UDTTestBase{
 	
@@ -87,7 +88,7 @@
 		System.out.println("MD5 hash of data received: "+md5_received);
 		System.out.println(client.getStatistics());
 		
-		//assertEquals(md5_sent,md5_received);
+		assertEquals(md5_sent,md5_received);
 		
 		//store stat history to csv file
 		client.getStatistics().writeParameterHistory(File.createTempFile("/udtstats-",".csv"));
@@ -106,6 +107,7 @@
 		Runnable serverProcess=new Runnable(){
 			public void run(){
 				try{
+					MessageDigest md5=MessageDigest.getInstance("MD5");
 					long start=System.currentTimeMillis();
 					UDTSocket s=serverSocket.accept();
 					assertNotNull(s);
@@ -117,12 +119,13 @@
 						c=is.read(buf);
 						if(c<0)break;
 						else{
+							md5.update(buf,0,c);
 							total+=c;
 						}
 					}
 					System.out.println("Server thread exiting, last received bytes: "+c);
 					serverRunning=false;
-					md5_received=s.getSession().getStatistics().getDigest();
+					md5_received=Util.hexString(md5);
 					serverSocket.shutDown();
 					System.out.println(s.getSession().getStatistics());
 				}

Added: udt-java/trunk/src/test/java/udt/performance/TestUDTLargeDataCC1.java
===================================================================
--- udt-java/trunk/src/test/java/udt/performance/TestUDTLargeDataCC1.java	                        (rev 0)
+++ udt-java/trunk/src/test/java/udt/performance/TestUDTLargeDataCC1.java	2010-04-30 14:02:45 UTC (rev 32)
@@ -0,0 +1,33 @@
+package udt.performance;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import udt.UDTReceiver;
+import udt.UDTSession;
+import udt.cc.SimpleTCP;
+
+//uses different CC algorithm
+public class TestUDTLargeDataCC1 extends TestUDTLargeData{
+	
+	boolean running=false;
+
+	//how many
+	int num_packets=50;
+	
+	//how large is a single packet
+	int size=1*1024*1024;
+	
+	int TIMEOUT=Integer.MAX_VALUE;
+	
+	int READ_BUFFERSIZE=1*1024*1024;
+
+	public void test1()throws Exception{
+		Logger.getLogger("udt").setLevel(Level.INFO);
+		UDTReceiver.dropRate=0;
+		System.setProperty(UDTSession.CC_CLASS, SimpleTCP.class.getName());
+		TIMEOUT=Integer.MAX_VALUE;
+		doTest();
+	}
+
+}


Property changes on: udt-java/trunk/src/test/java/udt/performance/TestUDTLargeDataCC1.java
___________________________________________________________________
Added: svn:mime-type
   + text/plain


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.