增加延迟检测功能

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

@@ -1,5 +1,6 @@
 package org.kne.cloud.network.klalb;
 
 public class Consts {
-	public static int BLOCKSIZE=65536;
+	public static final int BLOCKSIZE=65536;
+	public static final long PINGTIMENS=500000000;
 }

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

@@ -9,7 +9,7 @@ public class KLALBBlock {
 	 volatile long number;
 	 
 	 
-	 //volatile  long time=-1;
+	 volatile  long time=-1;
 	 volatile  Thread thread;
 	public KLALBBlock(byte[] b, int size,long number) {
 		data=b;
@@ -23,6 +23,10 @@ public class KLALBBlock {
 	public String toString() {
 		if(number>0) {
 			return "DATA"+number+":"+size;
+		}else if(number==0){
+			return "PING";
+		}else if(number==Long.MIN_VALUE){
+			return "PONG";
 		}else {
 			return "ACK"+(-number);
 		}

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

@@ -76,7 +76,7 @@ public class KLALBCore {
 				}
 			}
 			if (b != null) {
-				System.out.println("\tPROCESS:" + inputcache.size());
+				//System.out.println("\tPROCESS:" + inputcache.size());
 				inputcount++;
 				return b;
 			}
@@ -99,6 +99,9 @@ public class KLALBCore {
 			if (closeremote)
 				Thread.currentThread().interrupt();
 			Thread.sleep(1);
+			if(out.checkPingTime()) {
+				send0(out, new KLALBBlock(null,0 , 0));
+			}
 		}
 	
 		if(bqk!=null&&bqk.size()>0) {
@@ -143,12 +146,17 @@ public class KLALBCore {
 				Thread.sleep(1);
 			}
 		}
+		
+		if(out.checkPingTime()) {
+			send0(out, new KLALBBlock(null,0 , 0));
+		}
 	}
 
 	public void receiveDataBlock(TCPConnection in) throws IOException, InterruptedException {
 		if (closeremote)
 			throw new InterruptedException();
 		KLALBBlock x = receive0(in);
+		long rect=System.nanoTime();
 		if (x.number > 0) {
 			
 				KLALBBlock klb=new KLALBBlock(null, 0, -x.number);
@@ -160,14 +168,6 @@ public class KLALBCore {
 				}else {
 					acks.get(in).add(klb);
 				}
-				/*ThreadTool.makeVThreadIfSupport("TACK", ()->{
-					try {
-						send0(in, klb);
-					} catch (IOException e) {
-						e.printStackTrace();
-					}finally {
-					}
-				}).start();*/
 				
 				
 				if (x.number >= inputcount) {
@@ -181,7 +181,24 @@ public class KLALBCore {
 				inputcache.add(x);
 			}
 
-		} else {
+		} else if(x.number==0) { 
+			ThreadTool.makeVThreadIfSupport("TACK", ()->{
+			try {
+				KLALBBlock klk=new KLALBBlock(null,0,Long.MIN_VALUE);
+				long st=System.nanoTime();
+				long tw=st-rect;
+				klk.time=x.time+tw;
+				send0(in, klk);
+			} catch (IOException e) {
+				e.printStackTrace();
+			}finally {
+			}
+		}).start();
+		}else if(x.number==Long.MIN_VALUE){
+			//System.out.println("PING:"+(System.nanoTime()-x.time)/2000000);
+			long del=(System.nanoTime()-x.time)/2;
+			in.setDelay(del);
+		}else{
 			long v = -x.number;
 			outputcache.removeIf((b) -> {
 				return b.number == v;
@@ -196,11 +213,15 @@ public class KLALBCore {
 		if (kd.number > 0) {
 			out.writeInt(kd.size);
 			out.write(kd.data, 0, kd.size);
+		}else if(kd.number==0){
+			out.writeLong(System.nanoTime());
+		}else if(kd.number==Long.MIN_VALUE) {
+			out.writeLong(kd.time);
 		}
 		out.flush();
 		
 		}
-		 System.out.println("SEND:" + kd);
+		 //System.out.println("SEND:" + kd);
 	}
 
 	private KLALBBlock receive0(TCPConnection tcp) throws IOException {
@@ -212,9 +233,13 @@ public class KLALBCore {
 			kb.size = in.readInt();
 			kb.data = new byte[kb.size];
 			in.readFully(kb.data);
+		}else if(kb.number==0){
+			kb.time=in.readLong();
+		}else if(kb.number==Long.MIN_VALUE) {
+			kb.time=in.readLong();
 		}
 		}
-		 System.out.println("RECEIVE:" + kb);
+		 //System.out.println("RECEIVE:" + kb);
 		return kb;
 	}
 

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

@@ -6,6 +6,7 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.Socket;
+import java.net.SocketException;
 import java.net.UnknownHostException;
 
 public class TCPConnection {
@@ -64,6 +65,13 @@ public class TCPConnection {
 
 	public void setDelay(long delay) {
 		this.delay = delay;
+		if(connect!=null) {
+			try {
+				connect.setSoTimeout((int) (delay/500000));
+			} catch (SocketException e) {
+				e.printStackTrace();
+			}
+		}
 	}
 
 	public TCPConnection(Tunnel t) throws UnknownHostException, IOException {
@@ -84,4 +92,16 @@ public class TCPConnection {
 		return !connect.isClosed();
 	}
 
+	
+	private volatile long time=System.nanoTime();
+	public boolean checkPingTime() {
+		long cu=System.nanoTime();
+		if(cu-time>Consts.PINGTIMENS) {
+			time=cu;
+			return true;
+		}else {
+			return false;
+		}
+	}
+
 }