延迟优化

src/org/kne/cloud/network/klalb/KLALBBlock.java

@@ -3,13 +3,13 @@ package org.kne.cloud.network.klalb;
 import java.util.Arrays;
 import java.util.Objects;
 
-public class KLALBBlock implements Comparable<KLALBBlock>{
+public class KLALBBlock {
 	 volatile byte[]data;
 	 volatile int size;
 	 volatile long number;
 	 
 	 
-	 volatile  long time=-1;
+	 //volatile  long time=-1;
 	 volatile  Thread thread;
 	public KLALBBlock(byte[] b, int size,long number) {
 		data=b;
@@ -42,16 +42,6 @@ public class KLALBBlock implements Comparable<KLALBBlock>{
 		KLALBBlock other = (KLALBBlock) obj;
 		return number == other.number;
 	}
-	@Override
-	public int compareTo(KLALBBlock o) {
-		if(time>o.time) {
-			return 1;
-		}else if(time<o.time){
-			return -1;
-		}else {
-			return 0;
-		}
-	}
 	
 	
 }

src/org/kne/cloud/network/klalb/KLALBCore.java

@@ -28,7 +28,6 @@ public class KLALBCore {
 	private Set<KLALBBlock> inputcache = Collections.synchronizedSet(new HashSet<>());
 	private List<KLALBBlock> outputcache = Collections.synchronizedList(new ArrayList<>());
 	
-	private BlockingQueue<KLALBBlock> kack = new LinkedBlockingQueue<>();
 	
 	
 	private volatile boolean inlocal=true;
@@ -91,37 +90,32 @@ public class KLALBCore {
 	public void sendDataBlock(TCPConnection out) throws IOException, InterruptedException {
 		if (close)
 			throw new InterruptedException();
-		while (kack.isEmpty() && outputcache.isEmpty()) {
+		while ( outputcache.isEmpty()) {
 			if (close)
 				Thread.currentThread().interrupt();
 			Thread.sleep(1);
 		}
-		KLALBBlock kx = kack.poll();
-		if (kx != null) {
-			kx.time+=System.nanoTime();
-			send0(out, kx);
-		} else {
 			KLALBBlock ks = null;
 			synchronized (outputcache) {
 				for (int i = 0; i < outputcache.size(); i++) {
 
 					KLALBBlock kd = outputcache.get(i);
-					if (kd.time == -1) {
+					if (kd.thread == null) {
-						kd.time = System.nanoTime();
+						//kd.time = System.nanoTime();
 						kd.thread = Thread.currentThread();
 						ks = kd;
 					} else {
 						if (kd.thread.isAlive()) {
-							long timex = (System.nanoTime() - kd.time) / 1000000;
+							/*long timex = (System.nanoTime() - kd.time) / 1000000;
 							if (timex > 10000) {
 							System.out.println("³¬Ê±ÖØ´«£º"+kd);
 								kd.time = System.nanoTime();
 								kd.thread = Thread.currentThread();
 								ks = kd;
-							}
+							}*/
 						} else {
 							System.out.println("µôÏßÖØ´«£º"+kd);
-							kd.time = System.nanoTime();
+							//kd.time = System.nanoTime();
 							kd.thread = Thread.currentThread();
 							ks = kd;
 						}
@@ -134,7 +128,6 @@ public class KLALBCore {
 			if (ks != null) {
 				send0(out, ks);
 		//	System.out.println(outputcache.size());
-			}
 		}
 		
 	}
@@ -144,14 +137,17 @@ public class KLALBCore {
 			throw new InterruptedException();
 		KLALBBlock x = receive0(in);
 		if (x.number > 0) {
-			try {
+			
 				KLALBBlock klb=new KLALBBlock(null, 0, -x.number);
-				klb.time=x.time-System.nanoTime();
+				ThreadTool.makeVThreadIfSupport("ACK", ()->{
-				kack.put(klb);
+					try {
-			} catch (InterruptedException e) {
+						send0(in, klb);
-				e.printStackTrace();
+					} catch (IOException e) {
-			}
+						e.printStackTrace();
-			if (x.number >= inputcount) {
+					}
+				}).start();
+				
+				if (x.number >= inputcount) {
 				
 				while(inputcache.size()>(2*cacheblocks)&&inlocal) {
 					if (close)
@@ -169,19 +165,18 @@ public class KLALBCore {
 			});
 		}
 	}
-
+	
 	private void send0(TCPConnection tcp, KLALBBlock kd) throws IOException {
 		DataOutputStream out=tcp.getDout();
 		synchronized (out) {
 		out.writeLong(kd.number);
 		if (kd.number > 0) {
-			out.writeLong(System.nanoTime());
 			out.writeInt(kd.size);
 			out.write(kd.data, 0, kd.size);
-		}else {
-			out.writeLong(kd.time);
 		}
-		out.flush();
+		
+		
+		tcp.flush();
 		}
 		 System.out.println("SEND:" + kd);
 	}
@@ -191,17 +186,10 @@ public class KLALBCore {
 		DataInputStream in=tcp.getDin();
 		synchronized (in) {
 		kb.number = in.readLong();
-		kb.time = in.readLong();
-		
-		
-		
 		if (kb.number > 0) {
 			kb.size = in.readInt();
 			kb.data = new byte[kb.size];
 			in.readFully(kb.data);
-		}else{
-			long dela=(System.nanoTime()-kb.time)/2000000;
-		//System.out.println(tcp.getTunnel()+" "+dela);
 		}
 		}
 		 System.out.println("RECEIVE:" + kb);

src/org/kne/cloud/network/klalb/KLALBServer.java

@@ -27,7 +27,7 @@ public class KLALBServer {
 					nx=whm.get(uid);
 				}else {
 					nx=new KLALBServerProtocol();
-					Socket soc=new Socket("192.168.1.233",8444);
+					Socket soc=new Socket("192.168.1.233",3389);
 					nx.setOut(soc.getOutputStream());
 					nx.setIn(soc.getInputStream());
 					nx.startLocal();

src/org/kne/cloud/network/klalb/TCPConnection.java

@@ -12,61 +12,94 @@ public class TCPConnection {
 
 	private Tunnel tunnel;
 	private Socket connect;
-private	DataInputStream din;
+	private DataInputStream din;
-public Tunnel getTunnel() {
+
-	return tunnel;
+	public Tunnel getTunnel() {
-}
+		return tunnel;
-private	DataOutputStream dout;
-private long delay;
-public Socket getConnect() {
-	return connect;
-}
-public DataInputStream getDin() {
-	return din;
-}
-public DataOutputStream getDout() {
-	return dout;
-}
-public void close() {
-	// TODO 自动生成的方法存根
-	try {
-		if(din!=null)
-		din.close();
-	} catch (IOException e1) {
-		// TODO 自动生成的 catch 块
-		e1.printStackTrace();
 	}
-	try {
+
-		if(dout!=null)
+	private DataOutputStream dout;
-		dout.close();
+	private long delay;
-	} catch (IOException e1) {
+
-		// TODO 自动生成的 catch 块
+	public Socket getConnect() {
-		e1.printStackTrace();
+		return connect;
 	}
-	try {
+
-		if(connect!=null)
+	public DataInputStream getDin() {
-		connect.close();
+		return din;
-	} catch (IOException e) {
+	}
-		// TODO 自动生成的 catch 块
+
-		e.printStackTrace();
+	public DataOutputStream getDout() {
+		return dout;
+	}
+
+	public void close() {
+		// TODO 自动生成的方法存根
+		try {
+			if (din != null)
+				din.close();
+		} catch (IOException e1) {
+			// TODO 自动生成的 catch 块
+			e1.printStackTrace();
+		}
+		try {
+			if (dout != null)
+				dout.close();
+		} catch (IOException e1) {
+			// TODO 自动生成的 catch 块
+			e1.printStackTrace();
+		}
+		try {
+			if (connect != null)
+				connect.close();
+		} catch (IOException e) {
+			// TODO 自动生成的 catch 块
+			e.printStackTrace();
+		}
+		connect = null;
+	}
+
+	public long getDelay() {
+		return delay;
+	}
+
+	public void setDelay(long delay) {
+		this.delay = delay;
+	}
+
+	public TCPConnection(Tunnel t) throws UnknownHostException, IOException {
+		this(t, t.connectClientSocket());
+	}
+
+	public TCPConnection(Tunnel t, Socket soc) throws IOException {
+		connect = soc;
+		tunnel = t;
+		// connect.setSoTimeout(10000);
+		din = new DataInputStream(new BufferedInputStream(connect.getInputStream(), 65536));
+		dout = new DataOutputStream(new BufferedOutputStream(connect.getOutputStream(), 65536));
+		ThreadTool.makeVThreadIfSupport("FLUSH", () -> {
+			try {
+				while (true) {
+					Thread.sleep(100);
+					if(flush) {
+					synchronized (dout) {
+						dout.flush();
+
+					}
+					flush=false;
+					}
+				}
+			} catch (IOException e) {
+				e.printStackTrace();
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		}).start();
+	}
+
+	private volatile boolean flush = false;
+
+	public void flush() throws IOException {
+		flush = true;
 	}
-	connect=null;
-}
-public long getDelay() {
-	return delay;
-}
-public void setDelay(long delay) {
-	this.delay = delay;
-}
-public TCPConnection(Tunnel t) throws UnknownHostException, IOException {
-	this(t,t.connectClientSocket());
-}
-public TCPConnection(Tunnel t,Socket soc) throws IOException {
-		connect=soc;
-		tunnel=t;
-		//connect.setSoTimeout(10000);
-		din=new DataInputStream(new BufferedInputStream( connect.getInputStream(),65536));
-		dout=new DataOutputStream(new BufferedOutputStream (connect.getOutputStream(),65536));
-		
-}
 }