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

SourceForge Headquarters 225 Broadway Suite 1600 San Diego, CA 92101 +1 (858) 422-6466

Revision: 18
          http://udt-java.svn.sourceforge.net/udt-java/?rev=18&view=rev
Author:   bschuller
Date:     2010-04-16 21:05:13 +0000 (Fri, 16 Apr 2010)

Log Message:
-----------
cc made configurable; tried to optimize loss lists

Modified Paths:
--------------
    udt-java/trunk/src/main/java/udt/UDPEndPoint.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/UDTSocket.java
    udt-java/trunk/src/main/java/udt/packets/NegativeAcknowledgement.java
    udt-java/trunk/src/main/java/udt/receiver/ReceiverLossList.java
    udt-java/trunk/src/main/java/udt/receiver/ReceiverLossListEntry.java
    udt-java/trunk/src/main/java/udt/sender/SenderLossList.java
    udt-java/trunk/src/main/java/udt/sender/SenderLossListEntry.java
    udt-java/trunk/src/main/java/udt/util/CircularArray.java
    udt-java/trunk/src/main/java/udt/util/FlowWindow.java
    udt-java/trunk/src/main/java/udt/util/UDTStatistics.java
    udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java
    udt-java/trunk/src/test/java/udt/performance/UDPTest.java

Added Paths:
-----------
    udt-java/trunk/src/main/java/udt/CongestionControl.java
    udt-java/trunk/src/test/java/udt/NullCongestionControl.java

Added: udt-java/trunk/src/main/java/udt/CongestionControl.java
===================================================================

--- udt-java/trunk/src/main/java/udt/CongestionControl.java	                        (rev 0)
+++ udt-java/trunk/src/main/java/udt/CongestionControl.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -0,0 +1,74 @@
+package udt;
+
+import java.util.List;
+
+/**
+ * congestion control interface
+ */
+public interface CongestionControl {
+
+	/**
+	 * Callback function to be called (only) at the start of a UDT connection.
+	 * when the UDT socket is conected 
+	 */
+	public abstract void init();
+
+	/**
+	 * set roundtrip time and associated variance
+	 */
+	public abstract void setRTT(long rtt, long rttVar);
+
+	/**
+	 * set packet arrival rate and link capacity
+	 * @param rate
+	 * @param linkCapacity
+	 */
+	public abstract void setPacketArrivalRate(long rate, long linkCapacity);
+
+	/**
+	 * Inter-packet interval in seconds
+	 * @return 
+	 */
+	public abstract double getSendInterval();
+
+	/**
+	 * get the congestion window size
+	 */
+	public abstract long getCongestionWindowSize();
+	
+	/**
+	 * Callback function to be called when an ACK packet is received.
+	 * @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.
+	 */
+	public abstract void onNAK(List<Integer> lossInfo);
+
+	/**
+	 * Callback function to be called when a timeout event occurs
+	 */
+	public abstract void onTimeout();
+
+	/**
+	 * Callback function to be called when a data is sent.
+	 * @param packetSeqNo: the data sequence number.
+	 */
+	public abstract void onPacketSend(long packetSeqNo);
+
+	/**
+	 * Callback function to be called when a data is received.
+	 * @param packetSeqNo: the data sequence number.
+	 */
+	public abstract void onPacketReceive(long packetSeqNo);
+
+	/**
+	 * Callback function to be called when a UDT connection is closed.
+	 */
+	public abstract void close();
+
+}
\ No newline at end of file


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

Modified: udt-java/trunk/src/main/java/udt/UDPEndPoint.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDPEndPoint.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/UDPEndPoint.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -55,8 +55,6 @@
 /**
  * the UDPEndpoint takes care of sending and receiving UDP network packets,
  * dispatching them to the correct {@link UDTSession}
- * 
- * 
  */
 public class UDPEndPoint {
 
@@ -83,7 +81,7 @@
 	//has the endpoint been stopped?
 	private volatile boolean stopped=false;
 
-	public static final int DATAGRAM_SIZE=32768;
+	public static final int DATAGRAM_SIZE=1500;
 
 	/**
 	 * bind to any local port on the given host address

Modified: udt-java/trunk/src/main/java/udt/UDTCongestionControl.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTCongestionControl.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/UDTCongestionControl.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -2,12 +2,10 @@
 
 import java.util.List;
 
-import javax.swing.text.Utilities;
-
 import udt.util.UDTStatistics;
 import udt.util.Util;
 
-public class UDTCongestionControl {
+public class UDTCongestionControl implements CongestionControl {
 
 	private final UDTSession session;
 	private final UDTStatistics statistics;
@@ -64,26 +62,30 @@
 		init();
 	}
 
-	/**
-	 * Callback function to be called (only) at the start of a UDT connection.
-	 * when the UDT socket is conected 
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#init()
 	 */
 	public void init() {
 		
 	}
 
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#setRTT(long, long)
+	 */
 	public void setRTT(long rtt, long rttVar){
 		this.roundTripTime=rtt;
 	}
 
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#setPacketArrivalRate(long, long)
+	 */
 	public void setPacketArrivalRate(long rate, long linkCapacity){
 		this.packetArrivalRate=rate;
 		this.estimatedLinkCapacity=linkCapacity;
 	}
 
-	/**
-	 * Inter-packet interval in seconds
-	 * @return 
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#getSendInterval()
 	 */
 	public double getSendInterval(){
 		return packetSendingPeriod ;
@@ -93,14 +95,12 @@
 	 * congestionWindowSize
 	 * @return
 	 */
-	protected long getCongestionWindowSize(){
-		return congestionWindowSize;
+	public long getCongestionWindowSize(){
+		return 2048;//congestionWindowSize;
 	}
 
-	/**
-	 * Callback function to be called when an ACK packet is received.
-	 * @param ackSeqno: the data sequence number acknowledged by this ACK.
-	 * see spec. page(16-17)
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#onACK(long)
 	 */
 	public void onACK(long ackSeqno){
 		//the fixed size of a UDT packet 
@@ -143,9 +143,8 @@
 		return inc;
 	}
 	
-	/**
-	 *  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.
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#onNAK(java.util.List)
 	 */
 	public void onNAK(List<Integer>lossInfo){
 		long firstBiggestlossSeqNo=lossInfo.get(lossInfo.size()-1);
@@ -197,24 +196,22 @@
 		}
 	}
 
-	/**
-	 * Callback function to be called when a timeout event occurs
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#onTimeout()
 	 */
 	public void onTimeout(){}
 
-	/**
-	 * Callback function to be called when a data is sent.
-	 * @param packetSeqNo: the data sequence number.
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#onPacketSend(long)
 	 */
 	public void onPacketSend(long packetSeqNo){}
 
-	/**
-	 * Callback function to be called when a data is received.
-	 * @param packetSeqNo: the data sequence number.
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#onPacketReceive(long)
 	 */
 	public void onPacketReceive(long packetSeqNo){}
-	/**
-	 * Callback function to be called when a UDT connection is closed.
+	/* (non-Javadoc)
+	 * @see udt.CongestionControl#close()
 	 */
 	public void close(){}
 

Modified: udt-java/trunk/src/main/java/udt/UDTInputStream.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTInputStream.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/UDTInputStream.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -60,7 +60,7 @@
 	private final UDTStatistics statistics;
 
 	//the highest sequence number read by the application
-	private long highestSequenceNumber=-1;
+	private volatile long highestSequenceNumber=0;
 
 	//set to 'false' by the receiver when it gets a shutdown signal from the peer
 	//see the noMoreData() method
@@ -88,8 +88,8 @@
 	}
 
 	private int getFlowWindowSize(){
-		if(socket!=null)return socket.getSession().getFlowWindowSize();
-		else return 64;
+		if(socket!=null)return 2*socket.getSession().getFlowWindowSize();
+		else return 128;
 	}
 	/**
 	 * create a new {@link UDTInputStream} connected to the given socket
@@ -221,6 +221,7 @@
 	 * 
 	 */
 	protected boolean haveNewData(long sequenceNumber,byte[]data)throws IOException{
+		if(sequenceNumber<=highestSequenceNumber)return true;
 		return appData.offer(new AppData(sequenceNumber,data));
 	}
 
@@ -270,6 +271,31 @@
 		public String toString(){
 			return sequenceNumber+"["+data.length+"]";
 		}
+
+		@Override
+		public int hashCode() {
+			final int prime = 31;
+			int result = 1;
+			result = prime * result
+					+ (int) (sequenceNumber ^ (sequenceNumber >>> 32));
+			return result;
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (this == obj)
+				return true;
+			if (obj == null)
+				return false;
+			if (getClass() != obj.getClass())
+				return false;
+			AppData other = (AppData) obj;
+			if (sequenceNumber != other.sequenceNumber)
+				return false;
+			return true;
+		}
+		
+		
 	}
 
 }

Modified: udt-java/trunk/src/main/java/udt/UDTReceiver.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTReceiver.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/UDTReceiver.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -84,7 +84,7 @@
 	private long lastDataPacketArrivalTime=0;
 	
 	//largest received data packet sequence number(LRSN)
-	private volatile long largestReceivedSeqNumber=-1;
+	private volatile long largestReceivedSeqNumber=0;
 
 	//ACK event related
 
@@ -233,7 +233,6 @@
 			}
 			processUDTPacket(packet);
 		}
-		//else System.out.println("no packet.");
 		Thread.yield();
 	}
 
@@ -337,7 +336,6 @@
 	
 	protected void onDataPacketReceived(DataPacket dp)throws IOException{
 		long currentSequenceNumber = dp.getPacketSequenceNumber();
-		
 		//check whether to drop this packet
 		n++;
 		if(dropRate>0 && n % dropRate == 0){
@@ -412,6 +410,7 @@
 	protected void sendNAK(List<Long>sequenceNumbers)throws IOException{
 		if(sequenceNumbers.size()==0)return;
 		NegativeAcknowledgement nAckPacket= new NegativeAcknowledgement();
+		
 		nAckPacket.addLossInfo(sequenceNumbers);
 		nAckPacket.setSession(session);
 		nAckPacket.setDestinationID(session.getDestination().getSocketID());

Modified: udt-java/trunk/src/main/java/udt/UDTSender.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTSender.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/UDTSender.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -40,6 +40,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -87,7 +88,7 @@
 	private final AtomicInteger unacknowledged=new AtomicInteger(0);
 
 	//for generating data packet sequence numbers
-	private long nextSequenceNumber=-1;
+	private long nextSequenceNumber=0;
 
 	//the largest data packet sequence number that has actually been sent out
 	private volatile long largestSentSequenceNumber=-1;
@@ -100,6 +101,8 @@
 
 	private volatile boolean stopped=false;
 
+	private volatile AtomicReference<CountDownLatch> latchRef=new AtomicReference<CountDownLatch>();
+	
 	public UDTSender(UDTSession session,UDPEndPoint endpoint){
 		this.endpoint= endpoint;
 		this.session=session;
@@ -110,6 +113,8 @@
 		sendBuffer=new ConcurrentHashMap<Long, DataPacket>(MAX_SIZE,0.75f,2); 
 		sendQueue = new LinkedBlockingQueue<DataPacket>(MAX_SIZE);  
 		lastAckSequenceNumber=session.getInitialSequenceNumber();
+		
+		latchRef.set(new CountDownLatch(1));
 		start();
 	}
 
@@ -190,8 +195,8 @@
 	}
 
 	protected void onAcknowledge(Acknowledgement acknowledgement)throws IOException{
-		if(latch!=null)latch.countDown();
-		UDTCongestionControl cc=session.getCongestionControl();
+		latchRef.get().countDown();
+		CongestionControl cc=session.getCongestionControl();
 		if(acknowledgement.getPacketReceiveRate()>0){
 			long rtt=acknowledgement.getRoundTripTime();
 			long rttVar=acknowledgement.getRoundTripTimeVar();
@@ -230,8 +235,6 @@
 		for(Integer i: nak.getDecodedLossInfo()){
 			senderLossList.insert(new SenderLossListEntry(i));
 		}
-		//update SND TODO
-
 		session.getCongestionControl().onNAK(nak.getDecodedLossInfo());
 		//reset EXP. EXP is in the receiver currently.... maybe move to SOCKET?
 		session.getSocket().getReceiver().resetEXPTimer();
@@ -297,7 +300,7 @@
 					statistics.incNumberOfCCSlowDownEvents();
 					return;
 				}
-				DataPacket dp=sendQueue.poll(10,TimeUnit.MILLISECONDS);
+				DataPacket dp=sendQueue.poll(100,TimeUnit.MILLISECONDS);
 				if(dp!=null){
 					lastSentTime=Util.getCurrentTime();
 					send(dp);
@@ -309,8 +312,9 @@
 					statistics.incNumberOfCCWindowExceededEvents();
 				}
 			}
+			Thread.yield();
 		}
-		Thread.yield();
+		
 	}
 
 	/**
@@ -363,16 +367,14 @@
 		return senderLossList.isEmpty();
 	}
 	
-	private volatile CountDownLatch latch=null;
-	
 	/**
 	 * wait for the next acknowledge
 	 * @throws InterruptedException
 	 */
-	public void waitForAck(long sequenceNumber)throws InterruptedException{
-		latch=new CountDownLatch(1);
+	public synchronized void waitForAck(long sequenceNumber)throws InterruptedException{
 		while(!session.isShutdown() && !haveAcknowledgementFor(sequenceNumber)){
-			latch.await(10, TimeUnit.MILLISECONDS);
+			latchRef.set(new CountDownLatch(1));
+			latchRef.get().await(10, TimeUnit.MILLISECONDS);
 		}
 	}
 

Modified: udt-java/trunk/src/main/java/udt/UDTSession.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTSession.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/UDTSession.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -62,13 +62,13 @@
 	
 	protected int receiveBufferSize=64*32768;
 	
-	protected final UDTCongestionControl cc;
+	protected final CongestionControl cc;
 	
 	/**
 	 * flow window size, i.e. how many data packets are
 	 * in-flight at a single time
 	 */
-	protected int flowWindowSize=32;
+	protected int flowWindowSize=64;
 
 	/**
 	 * remote UDT entity (address and socket ID)
@@ -84,6 +84,12 @@
 	public static final int DEFAULT_DATAGRAM_SIZE=UDPEndPoint.DATAGRAM_SIZE;
 	
 	/**
+	 * key for a system property defining the CC class to be used
+	 * @see CongestionControl
+	 */
+	public static final String CC_CLASS="udt.congestioncontrol.class";
+	
+	/**
 	 * Buffer size (i.e. datagram size)
 	 * This is negotiated during connection setup
 	 */
@@ -99,7 +105,18 @@
 		statistics=new UDTStatistics(description);
 		mySocketID=nextSocketID.incrementAndGet();
 		this.destination=destination;
-		cc=new UDTCongestionControl(this);
+		
+		//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){
+			ccObject=new UDTCongestionControl(this);
+		}
+		cc=(CongestionControl)ccObject;
+		System.out.println("using "+cc.getClass().getName());
 	}
 	
 	public abstract void received(UDTPacket packet, Destination peer);
@@ -108,7 +125,7 @@
 		return socket;
 	}
 
-	public UDTCongestionControl getCongestionControl() {
+	public CongestionControl getCongestionControl() {
 		return cc;
 	}
 

Modified: udt-java/trunk/src/main/java/udt/UDTSocket.java
===================================================================
--- udt-java/trunk/src/main/java/udt/UDTSocket.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/UDTSocket.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -182,7 +182,7 @@
 			packet.setData(chunk);
 			//put the packet into the send queue
 			while(!sender.sendUdtPacket(packet, timeout, units)){
-				System.out.println("WAIT");
+				System.out.println("SOCKET WAIT");
 			}
 		}
 		if(length>0)active=true;

Modified: udt-java/trunk/src/main/java/udt/packets/NegativeAcknowledgement.java
===================================================================
--- udt-java/trunk/src/main/java/udt/packets/NegativeAcknowledgement.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/packets/NegativeAcknowledgement.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -109,7 +109,7 @@
 	 */
 	public void addLossInfo(long firstSequenceNumber, long lastSequenceNumber) {
 		//check if we really need an interval
-		if(lastSequenceNumber-firstSequenceNumber==1){
+		if(lastSequenceNumber-firstSequenceNumber==0){
 			addLossInfo(firstSequenceNumber);
 			return;
 		}

Modified: udt-java/trunk/src/main/java/udt/receiver/ReceiverLossList.java
===================================================================
--- udt-java/trunk/src/main/java/udt/receiver/ReceiverLossList.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/receiver/ReceiverLossList.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -32,7 +32,7 @@
 
 package udt.receiver;
 import java.util.ArrayList;
-import java.util.Iterator;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.PriorityBlockingQueue;
 
@@ -50,25 +50,18 @@
 	
 	public ReceiverLossList(){
 		backingList = new PriorityBlockingQueue<ReceiverLossListEntry>(16);
-	} 
+	}
 	
 	public void insert(ReceiverLossListEntry entry){
-		backingList.add(entry);
+		synchronized (backingList) {
+			if(!backingList.contains(entry)){
+				backingList.add(entry);
+			}
+		}
 	}
-	
-	public void remove(ReceiverLossListEntry obj){
-		backingList.remove(obj);
-	}
 
 	public void remove(long seqNo){
-		Iterator<ReceiverLossListEntry>iterator=backingList.iterator();
-		while(iterator.hasNext()){
-			ReceiverLossListEntry e=iterator.next();
-			if(e.getSequenceNumber()==seqNo){
-				iterator.remove();
-				break;
-			}
-		}
+		backingList.remove(new ReceiverLossListEntry(seqNo));
 	}
 	
 	public boolean contains(ReceiverLossListEntry obj){
@@ -102,8 +95,10 @@
 	public List<Long>getFilteredSequenceNumbers(long RTT, boolean doFeedback){
 		List<Long>result=new ArrayList<Long>();
 		long now=Util.getCurrentTime();
-		for(ReceiverLossListEntry e: backingList){
-			if( (now-e.getLastFeedbackTime())>2*RTT){
+		ReceiverLossListEntry[]sorted=backingList.toArray(new ReceiverLossListEntry[0]);
+		Arrays.sort(sorted);
+		for(ReceiverLossListEntry e: sorted){
+			if( (now-e.getLastFeedbackTime())>e.getK()*RTT){
 				result.add(e.getSequenceNumber());
 				if(doFeedback)e.feedback();
 			}

Modified: udt-java/trunk/src/main/java/udt/receiver/ReceiverLossListEntry.java
===================================================================
--- udt-java/trunk/src/main/java/udt/receiver/ReceiverLossListEntry.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/receiver/ReceiverLossListEntry.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -48,7 +48,9 @@
 	 * @param sequenceNumber
 	 */
 	public ReceiverLossListEntry(long sequenceNumber){
-		if(sequenceNumber<=0)throw new IllegalArgumentException();
+		if(sequenceNumber<=0){
+			throw new IllegalArgumentException("Got sequence number "+sequenceNumber);
+		}
 		this.sequenceNumber = sequenceNumber;	
 		this.lastFeedbacktime=Util.getCurrentTime();
 	}
@@ -90,4 +92,30 @@
 		return sequenceNumber+"[k="+k+",time="+lastFeedbacktime+"]";
 	}
 
+
+	@Override
+	public int hashCode() {
+		final int prime = 31;
+		int result = 1;
+		result = prime * result + (int) (k ^ (k >>> 32));
+		result = prime * result
+				+ (int) (sequenceNumber ^ (sequenceNumber >>> 32));
+		return result;
+	}
+
+
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj)
+			return true;
+		if (obj == null)
+			return false;
+		if (getClass() != obj.getClass())
+			return false;
+		ReceiverLossListEntry other = (ReceiverLossListEntry) obj;
+		if (sequenceNumber != other.sequenceNumber)
+			return false;
+		return true;
+	}
+
 }

Modified: udt-java/trunk/src/main/java/udt/sender/SenderLossList.java
===================================================================
--- udt-java/trunk/src/main/java/udt/sender/SenderLossList.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/sender/SenderLossList.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -49,7 +49,9 @@
 	}
 
 	public void insert(SenderLossListEntry obj){
-		backingList.add(obj);
+		synchronized (backingList) {
+			if(!backingList.contains(obj))backingList.add(obj);	
+		}
 	}
 
 	public void remove(long seqNo){
@@ -61,6 +63,7 @@
 				return;
 			}
 		}
+		//backingList.remove(new SenderLossListEntry(seqNo));
 	}
 
 	/**

Modified: udt-java/trunk/src/main/java/udt/sender/SenderLossListEntry.java
===================================================================
--- udt-java/trunk/src/main/java/udt/sender/SenderLossListEntry.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/sender/SenderLossListEntry.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -71,4 +71,27 @@
 		return (int)(sequenceNumber-o.sequenceNumber);
 	}
 
+	@Override
+	public int hashCode() {
+		final int prime = 31;
+		int result = 1;
+		result = prime * result
+				+ (int) (sequenceNumber ^ (sequenceNumber >>> 32));
+		return result;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj)
+			return true;
+		if (obj == null)
+			return false;
+		if (getClass() != obj.getClass())
+			return false;
+		SenderLossListEntry other = (SenderLossListEntry) obj;
+		if (sequenceNumber != other.sequenceNumber)
+			return false;
+		return true;
+	}
+
 }

Modified: udt-java/trunk/src/main/java/udt/util/CircularArray.java
===================================================================
--- udt-java/trunk/src/main/java/udt/util/CircularArray.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/util/CircularArray.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -35,18 +35,23 @@
 import java.util.ArrayList;
 import java.util.List;
 
+/**
+ * Circular array: the most recent value overwrites the oldest one if there is no more free 
+ * space in the array
+ */
 public class CircularArray<T>{
 
 	protected int position=0;
+	
 	protected boolean haveOverflow=false;
-	//the maximum number of entries
-	protected int max=1;
 	
-	protected List<T>circularArray;
+	protected final int max;
 	
+	protected final List<T>circularArray;
+	
 	/**
-	 * ArrayList von T(object's type).  The most recent value overwrite the oldest one
-	 * if no more free space in the array
+	 * Create a new circularArray of the given size
+	 * 
 	 * @param size
 	 */
 	public CircularArray(int size){
@@ -55,10 +60,7 @@
 	}
 	
 	/**
-	 * Insert the specified entry at the specified position in this list.
-	 * the most recent value overwrite the oldest one
-	 * if no more free space in the circularArray
-	 * @param entry
+	 * add an entry
 	 */
 	public void add(T entry){
 		if(position>=max){
@@ -74,7 +76,6 @@
 	
 	/**
 	 * Returns the number of elements in this list 
-	 * @return
 	 */
 	public int size(){
 		return circularArray.size();
@@ -83,7 +84,5 @@
 	public String toString(){
 		return circularArray.toString();
 	}
-	
-	
-	
+
 }

Modified: udt-java/trunk/src/main/java/udt/util/FlowWindow.java
===================================================================
--- udt-java/trunk/src/main/java/udt/util/FlowWindow.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/util/FlowWindow.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -74,6 +74,9 @@
 	 */
 	@Override
 	public boolean offer(E e) {
+		if(contains(e)){
+			return true;
+		}
 		if(size()<capacity){
 			return super.offer(e);
 		}else return false;

Modified: udt-java/trunk/src/main/java/udt/util/UDTStatistics.java
===================================================================
--- udt-java/trunk/src/main/java/udt/util/UDTStatistics.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/main/java/udt/util/UDTStatistics.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -33,7 +33,6 @@
 package udt.util;
 
 import java.io.File;
-import java.io.FileOutputStream;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.security.MessageDigest;

Added: udt-java/trunk/src/test/java/udt/NullCongestionControl.java
===================================================================
--- udt-java/trunk/src/test/java/udt/NullCongestionControl.java	                        (rev 0)
+++ udt-java/trunk/src/test/java/udt/NullCongestionControl.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -0,0 +1,48 @@
+package udt;
+
+import java.util.List;
+
+public class NullCongestionControl implements CongestionControl {
+
+	private final UDTSession session;
+	
+	public NullCongestionControl(UDTSession session){
+		this.session=session;
+	}
+	
+	public void close() {
+	}
+
+	public long getCongestionWindowSize() {
+		return Long.MAX_VALUE;
+	}
+
+	public double getSendInterval() {
+		return 0;
+	}
+
+	public void init() {
+	}
+
+	public void onACK(long ackSeqno) {
+	}
+
+	public void onNAK(List<Integer> lossInfo) {
+	}
+
+	public void onPacketReceive(long packetSeqNo) {
+	}
+
+	public void onPacketSend(long packetSeqNo) {
+	}
+
+	public void onTimeout() {
+	}
+
+	public void setPacketArrivalRate(long rate, long linkCapacity) {
+	}
+
+	public void setRTT(long rtt, long rttVar) {
+	}
+
+}


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

Modified: udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java
===================================================================
--- udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/test/java/udt/performance/TestUDTLargeData.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -7,10 +7,12 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import udt.NullCongestionControl;
 import udt.UDTClient;
 import udt.UDTInputStream;
 import udt.UDTReceiver;
 import udt.UDTServerSocket;
+import udt.UDTSession;
 import udt.UDTSocket;
 import udt.UDTTestBase;
 import udt.util.UDTStatistics;
@@ -20,7 +22,8 @@
 	boolean running=false;
 
 	//how many
-	int num_packets=200;
+	int num_packets=100;
+	
 	//how large is a single packet
 	int size=1*1024*1024;
 	
@@ -30,6 +33,7 @@
 
 	public void test1()throws Exception{
 		Logger.getLogger("udt").setLevel(Level.INFO);
+		System.setProperty(UDTSession.CC_CLASS, NullCongestionControl.class.getName());
 		
 		UDTReceiver.dropRate=0;
 		TIMEOUT=Integer.MAX_VALUE;

Modified: udt-java/trunk/src/test/java/udt/performance/UDPTest.java
===================================================================
--- udt-java/trunk/src/test/java/udt/performance/UDPTest.java	2010-04-16 13:15:20 UTC (rev 17)
+++ udt-java/trunk/src/test/java/udt/performance/UDPTest.java	2010-04-16 21:05:13 UTC (rev 18)
@@ -3,20 +3,17 @@
 import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.InetAddress;
-import java.nio.ByteBuffer;
 import java.util.Random;
 
 import junit.framework.TestCase;
 
 /**
  * send some data over a UDP connection and measure performance
- * 
  */
 public class UDPTest extends TestCase {
 
-	final int BUFSIZE=32768;
-	final int num_packets=10*1000;
-	final int packetSize=1024;
+	final int num_packets=100*1000;
+	final int packetSize=1500;
 	
 	public void test1()throws Exception{
 		runServer();
@@ -25,20 +22,15 @@
 
 		//generate a test array with random content
 		N=num_packets*packetSize;
-		byte[]data=new byte[N];
+		byte[]data=new byte[packetSize];
 		new Random().nextBytes(data);
 		long start=System.currentTimeMillis();
-		ByteBuffer bb=ByteBuffer.wrap(data);
-		DatagramPacket dp=new DatagramPacket(new byte[BUFSIZE],BUFSIZE);
+		DatagramPacket dp=new DatagramPacket(new byte[packetSize],packetSize);
 		dp.setAddress(InetAddress.getByName("localhost"));
 		dp.setPort(65321);
-		
-		System.out.println("Sending data block of <"+N+"> bytes");
-		while(bb.remaining()>0){
-			int len=Math.min(bb.remaining(),BUFSIZE);
-			byte[]chunk=new byte[len];
-			bb.get(chunk);
-			dp.setData(chunk);
+		System.out.println("Sending "+num_packets+" data blocks of <"+packetSize+"> bytes");
+		for(int i=0;i<num_packets;i++){
+			dp.setData(data);
 			s.send(dp);
 		}
 		System.out.println("Finished sending.");
@@ -46,7 +38,8 @@
 		System.out.println("Server stopped.");
 		long end=System.currentTimeMillis();
 		System.out.println("Done. Sending "+N/1024/1024+" Mbytes took "+(end-start)+" ms");
-		System.out.println("Rate "+N/(end-start)+" Kbytes/sec");
+		System.out.println("Rate "+N/1000/(end-start)+" Mbytes/sec");
+		System.out.println("Rate "+num_packets+" packets/sec");
 		System.out.println("Server received: "+total);
 	}
 	
@@ -61,7 +54,7 @@
 		Runnable serverProcess=new Runnable(){
 			public void run(){
 				try{
-					byte[]buf=new byte[BUFSIZE];
+					byte[]buf=new byte[packetSize];
 					DatagramPacket dp=new DatagramPacket(buf,buf.length);
 					while(true){
 						serverSocket.receive(dp);


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




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

udt-java-commits