提高程序的稳定性

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

@@ -1,11 +1,12 @@
 package org.kne.cloud.network.klalb;
 
 public class Consts {
-	public static final int BLOCKSIZE=8192;
+	public static final int BLOCKSIZE=16384;
 	public static final long PINGTIMENS=1000000000L;
-	public static final int SO_TIMEOUT = 10000;
+	public static final int SO_TIMEOUT = 5000;
 	public static final long SN_KEEP = 60000000000L;
 	public static final int MAX_RESEND = 100;
-	public static final long UACK_TIME = 10000000000L;
+	public static final long UACK_TIME = 5000000000L;
 	public static final int CONNECT_TIMEOUT=10000;
+	public static final double MAX_QUEUE_TIME=500;
 }

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

@@ -22,7 +22,7 @@ import org.kne.cloud.network.mport.ServiceElement;
 import org.kne.cloud.network.mport.ThreadTool;
 
 public class IOThreadManager {
-	private KLALBCore klc = new KLALBCore(5000);
+	private KLALBCore klc = new KLALBCore(1000);
 	private volatile boolean isopen = true;
 
 	public IOThreadManager() {
@@ -34,6 +34,15 @@ public class IOThreadManager {
 			return;
 		}
 		klc.getLocaltcps().put(tc.getCuid(), tc);
+
+		if(klc.getRemotetcps().isEmpty()) {
+			System.out.println("所有线路已断开，请检查网络");
+			Collection<LocalTCPConnection> c=klc.getLocaltcps().values();
+			for (Iterator<LocalTCPConnection> iterator = c.iterator(); iterator.hasNext();) {
+				LocalTCPConnection object =iterator.next();
+				object.close();
+			}
+		}
 		AtomicBoolean AB=new AtomicBoolean(true);
 		try {
 			if (syn) { 
@@ -57,6 +66,8 @@ public class IOThreadManager {
 
 				}
 			});
+			
+			Thread.sleep(50);
 			Thread ls = ThreadTool.makeVThreadIfSupport("本地发送线程", () -> {
 
 				try {
@@ -69,6 +80,7 @@ public class IOThreadManager {
 					}
 
 				} catch (IOException|InterruptedException e) {
+					e.printStackTrace();
 					AB.set(false);
 					lt.interrupt();
 					tc.getOutputcache().clear();
@@ -95,6 +107,7 @@ public class IOThreadManager {
 						}
 					}
 				} catch (IOException|InterruptedException e) {
+					e.printStackTrace();
 					AB.set(false);
 					ls.interrupt();
 					lt.interrupt();
@@ -149,6 +162,7 @@ public class IOThreadManager {
 					}
 				} catch (InterruptedException e) {
 				} catch (IOException e) {
+					e.printStackTrace();
 				}
 			});
 			Thread rr = ThreadTool.makeVThreadIfSupport("远程接收线程", () -> {
@@ -157,6 +171,7 @@ public class IOThreadManager {
 						klc.remoteReceive(tc);
 					}
 				} catch (Exception e) {
+					e.printStackTrace();
 					rs.interrupt();
 				}
 			});
@@ -177,6 +192,15 @@ public class IOThreadManager {
 				e.printStackTrace();
 			}
 			}
+			
+			if(klc.getRemotetcps().isEmpty()) {
+				System.out.println("所有线路已断开，请检查网络");
+				Collection<LocalTCPConnection> c=klc.getLocaltcps().values();
+				for (Iterator<LocalTCPConnection> iterator = c.iterator(); iterator.hasNext();) {
+					LocalTCPConnection object =iterator.next();
+					object.close();
+				}
+			}
 		}
 	}
 

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

@@ -7,8 +7,8 @@ import java.util.concurrent.atomic.AtomicLong;
 public class InputMetre extends InputStream {
 
 	private InputStream in;
-	private AtomicLong total;
+	private AtomicLong[] total;
-	public InputMetre(InputStream inputStream,AtomicLong v) {
+	public InputMetre(InputStream inputStream,AtomicLong... v) {
 		this.in=inputStream;
 		this.total=v;
 	}
@@ -17,7 +17,8 @@ public class InputMetre extends InputStream {
 	public int read() throws IOException {
 	int v=in.read();
 	if(v!=-1) {
-		total.incrementAndGet();
+		for(AtomicLong x:total)
+		x.incrementAndGet();
 	}
 		return v;
 	}
@@ -26,7 +27,8 @@ public class InputMetre extends InputStream {
 	public int read(byte[] b) throws IOException {
 		int v=in.read(b);
 		if(v!=-1) {
-			total.addAndGet(v);
+			for(AtomicLong x:total)
+			x.addAndGet(v);
 		}
 		return v;
 	}
@@ -35,7 +37,8 @@ public class InputMetre extends InputStream {
 	public int read(byte[] b, int off, int len) throws IOException {
 		int v=in.read(b, off, len);
 		if(v!=-1) {
-			total.addAndGet(v);
+			for(AtomicLong x:total)
+			x.addAndGet(v);
 		}
 		return v;
 	}

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

@@ -10,21 +10,21 @@ import org.kne.cloud.network.mport.IPPort;
 public class KLALBBlock implements Comparable<KLALBBlock>{
 	private static AtomicLong sng=new AtomicLong(1);
 	
-	public long sn;//每个数据包的唯一编号
+	public volatile long sn;//每个数据包的唯一编号
-	public int cuid;//用于识别数据包的stream ID号
+	public volatile int cuid;//用于识别数据包的stream ID号
-	public long number;//数据包的编号，用于排序
+	public volatile long number;//数据包的编号，用于排序
-	public byte[]data;//数据内容
+	public volatile byte[]data;//数据内容
-	public int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST
+	public volatile int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST
-	public long pingtime;//PING计时器
+	public volatile long pingtime;//PING计时器
-	public int cacheused;
+	public volatile int cacheused;
 	
-	public String lservice;
+	public volatile String lservice;
-	public IPPort lipport;
+	public volatile IPPort lipport;
 
 	public transient volatile long sendtime;
 	public transient volatile int resend;
 
-	public long sendtimeForRTT;
+	public volatile long sendtimeForRTT;
 	
 	
 	public KLALBBlock() {

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

@@ -92,10 +92,10 @@ public class KLALBClient {
 						tc.getDout().writeLong(suid.getMostSignificantBits());
 						tc.getDout().writeLong(suid.getLeastSignificantBits());
 						tc.getDout().flush();
-						int v=tc.getDin().read();
+						
-						if(v==-1) {
+						tc.handshake(10);
-							throw new EOFException();
+						
-						}
+						
 						re=0;
 						System.out.println(tll+"：连接成功");
 						tll.setState(true);

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

@@ -176,11 +176,6 @@ public class KLALBClientGUI extends XFrame {
 								wv.add(tp);
 								tunnel.setTlr(new TListener() {
 									
-									@Override
-									public void setDelay(long delay) {
-										tp.setDelay(delay);
-										repaint();
-									}
 
 									@Override
 									public void setState(boolean state) {
@@ -193,8 +188,8 @@ public class KLALBClientGUI extends XFrame {
 									}
 
 									@Override
-									public void setTraffic(long up, long down, long ups, long downs) {
+									public void setTraffic(long up, long down, long ups, long downs,long delay) {
-										tp.setTraffic(up,down,ups,downs);
+										tp.setTraffic(up,down,ups,downs,delay);
 										repaint();
 									}
 									
@@ -219,7 +214,7 @@ public class KLALBClientGUI extends XFrame {
 								synchronized(ts) {
 									for(int i=0;i<ts.size();i++) {
 										Tunnel t=ts.get(i) ;
-										t.updateTraffics(1000);
+										t.updateTraffics();
 									}
 								}
 							}

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

@@ -9,6 +9,7 @@ import java.io.StreamCorruptedException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedList;
@@ -30,6 +31,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 import java.util.zip.DataFormatException;
@@ -43,12 +45,12 @@ public class KLALBCore {
 	
 	private volatile int cacheblocks;
 	
-	private Map<Integer, LocalTCPConnection> localtcps = new ConcurrentHashMap<>();
+	private Map<Integer, LocalTCPConnection> localtcps = Collections.synchronizedMap(new HashMap<>());
 	private List<RemoteTCPConnection> remotetcps = new Vector<>();
 	private Predicate<KLALBBlock>acceptSYN;
 	
 
-	private volatile long rsntime=System.nanoTime();
+	/*private volatile long rsntime=System.nanoTime();
 	private Set<SN> rsns=Collections.synchronizedSet(new HashSet<SN>());
 	private static class SN{
 		volatile long sn;
@@ -77,7 +79,7 @@ public class KLALBCore {
 			return sn == other.sn;
 		}
 		
-	}
+	}*/
 	
 	public Predicate<KLALBBlock> getAcceptSYN() {
 		return acceptSYN;
@@ -108,6 +110,7 @@ public class KLALBCore {
 				pdb.command=0;
 				pdb.pingtime=System.nanoTime();
 				tc.sendBlock(pdb);
+				tc.updateTraffics();
 			}
 			if(!tc.getNDSendDeque().isEmpty()||!tc.getSendDeque().isEmpty())
 				break;
@@ -133,7 +136,7 @@ public class KLALBCore {
 	}
 	public void remoteReceive(RemoteTCPConnection tc) throws IOException {
 		KLALBBlock brc=null;
-		for(;;) {
+		/*for(;;) {
 		try {
 			brc=tc.receiveBlock();
 		}  catch (DataFormatException e) {
@@ -162,7 +165,13 @@ public class KLALBCore {
 				}
 			}
 		}
+		} */
+		try {
+			brc=tc.receiveBlock();
+		}  catch (DataFormatException e) {
+			throw new StreamCorruptedException("ZIP error");
 		}
+		brc.sendtime=System.nanoTime();
 		if(brc.cuid==0) {
 			if(brc.number==0) {
 				if(brc.command==0) {
@@ -176,21 +185,7 @@ public class KLALBCore {
 				}else if(brc.command==1) {
 					long cur=System.nanoTime();
 					long delay=cur-brc.pingtime;
-					tc.setDelayAvg(delay);
+					setDelayAvg(tc,delay);
-					synchronized (remotetcps) {
-						Collections.sort(remotetcps, new Comparator<RemoteTCPConnection>() {
-							@Override
-							public int compare(RemoteTCPConnection o1, RemoteTCPConnection o2) {
-								if(o1.getDelay()>o2.getDelay()) {
-									return 1;
-							}else if(o1.getDelay()<o2.getDelay()){
-								return -1;
-							}else {
-								return 0;
-							}
-							}
-						});
-					}
 				}
 			}
 		}else {
@@ -198,6 +193,9 @@ public class KLALBCore {
 				if(brc.command==0) {
 				}else if(brc.command==1) {
 				}else if(brc.command==2) {
+					if(!localtcps.containsKey(brc.cuid)) {
+						localtcps.put(brc.cuid, null);
+						try {
 					if(acceptSYN==null||!acceptSYN.test(brc)) {
 						KLALBBlock rst=new KLALBBlock();
 						rst.cuid=brc.cuid;
@@ -205,6 +203,10 @@ public class KLALBCore {
 						rst.command=3;
 						submitDataBlockNoDelay(rst);
 					}
+						}finally {
+							localtcps.remove(brc.cuid, null);
+						}
+					}
 				}else if(brc.command==3) {
 					LocalTCPConnection ltc=localtcps.get(brc.cuid);
 					if(ltc!=null) {
@@ -221,7 +223,7 @@ public class KLALBCore {
 					ack.cuid=brc.cuid;
 					ack.number=-brc.number;
 					ack.cacheused=ltcs.getSendDeque().size();
-					submitAckBlockNoDelay(ack,5);
+					submitAckBlockNoDelay(ack,3);
 				}
 					
 				
@@ -244,24 +246,78 @@ public class KLALBCore {
 						}
 					}
 
-					synchronized (remotetcps) {
+					
-						for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
-							RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
-							Queue<KLALBBlock>sq=remoteTCPConnection.getSendDeque();
-							sq.removeIf((vb)->{
-								return vb.number==nx;
-							});
-						}
-					}
 				/*ltc.getOutputcache().removeIf((b) -> {
 					return b.number == nx;
 				});*/
 				ltc.setPeerCacheUsed(brc.cacheused);
 				}
+				int cuid=brc.cuid; 
+				synchronized (remotetcps) {
+					for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
+						RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
+						Queue<KLALBBlock>sq=remoteTCPConnection.getSendDeque();
+						sq.removeIf((vb)->{
+							return vb.number==nx&&vb.cuid==cuid;
+						});
+					}
+				}
 			}
 		}
 
 	}
+	private void sortRemoteTCPByPrediction() {
+			remotetcps.forEach((x)->{
+				x.predictLatency();
+			});
+			Collections.sort(remotetcps, new Comparator<RemoteTCPConnection>() {
+				@Override
+				public int compare(RemoteTCPConnection o1, RemoteTCPConnection o2) {
+					double o1x=o1.getPredictedLatency();
+					double o2x=o2.getPredictedLatency();
+					if(o1x>o2x) {
+						return 1;
+				}else if(o1x<o2x){
+					return -1;
+				}else {
+					return 0;
+				}
+				}
+			});
+		
+	}
+	private void sortRemoteTCPByLatency() {
+		Collections.sort(remotetcps, new Comparator<RemoteTCPConnection>() {
+			@Override
+			public int compare(RemoteTCPConnection o1, RemoteTCPConnection o2) {
+				long o1x=o1.getDelay();
+				long o2x=o2.getDelay();
+				if(o1x>o2x) {
+					return 1;
+			}else if(o1x<o2x){
+				return -1;
+			}else {
+				return 0;
+			}
+			}
+		});
+	
+}
+	private void setDelayAvg(RemoteTCPConnection tc, long delay) {
+		/*if(tc.odelay==Long.MAX_VALUE) {
+			tc.odelay=delay;
+			tc.setDelay(tc.odelay);
+		}else {
+			if(delay<=tc.odelay) {
+				tc.odelay=delay;
+			}else {
+			tc.odelay=(delay+tc.odelay*100)/101;
+			}
+			tc.setDelay(tc.odelay);
+		}*/
+		tc.setDelay((tc.getDelay()*10+delay)/11);
+	}
+
 	public void submitDataBlock(KLALBBlock ks) throws InterruptedException {
 		submitDataBlock(ks,1);
 	}
@@ -269,10 +325,11 @@ public class KLALBCore {
 		int ni=Math.min(i, remotetcps.size());
 			while(true) {
 				synchronized (remotetcps) {
+					sortRemoteTCPByPrediction();
 					for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
 						RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
-						Queue<KLALBBlock> bdq=remoteTCPConnection.getSendDeque();
+						//if(remoteTCPConnection.getPredictedLatency()<Consts.MAX_QUEUE_TIME) {
-						if(bdq.size()<2) {
+							Queue<KLALBBlock> bdq=remoteTCPConnection.getSendDeque();
 							bdq.add(kb);
 							synchronized (bdq) {
 							bdq.notifyAll();
@@ -280,7 +337,7 @@ public class KLALBCore {
 							ni--;
 							if(ni<=0)
 							return;
-						}
+						//}
 					}
 				}
 				Thread.sleep(1);
@@ -301,6 +358,7 @@ public class KLALBCore {
 	}
 	public void submitAckBlockNoDelay(KLALBBlock kb,int limit) {
 		synchronized (remotetcps) {
+			sortRemoteTCPByLatency();
 			int coun=0;
 			for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
 				if(coun>=limit) {
@@ -319,7 +377,7 @@ public class KLALBCore {
 	public void putDataBlock(LocalTCPConnection tc, KLALBBlock ks) throws InterruptedException {
 		while (true) {
 			boolean b;
-			synchronized (ks) {
+			synchronized (tc.getOutputcache()) {
 				b=!tc.getOutputcache().isEmpty()&&(tc.getOutputcount()-tc.getOutputcache().get(0).number>cacheblocks);
 			}
 			if(!b) {
@@ -327,10 +385,13 @@ public class KLALBCore {
 			}
 			Thread.sleep(1);
 		}
+		/*while(tc.getOutputcache().size()>cacheblocks) {
+			Thread.sleep(1);
+		}*/
 		ks.sendtime=System.nanoTime();
 		ks.sendtimeForRTT=ks.sendtime;
 		//System.out.println(tc.getOutputcache().size());
-		if(ks.data==null) {
+		/*if(ks.data==null) {
 			submitDataBlock(ks);
 		}else {
 		if(ks.data.length<Consts.BLOCKSIZE/2) {
@@ -338,16 +399,12 @@ public class KLALBCore {
 		}else {
 			submitDataBlock(ks);
 		}
-		}
+		}*/
+		submitDataBlock(ks);
 		tc.getOutputcache().add(ks);
-		int peerCacheUsed = tc.getPeerCacheUsed();
+		while( tc.getPeerCacheUsed()>5000) {
-		if(peerCacheUsed>cacheblocks) {
-		while( tc.getPeerCacheUsed()>cacheblocks) {
 			Thread.sleep(1);
 		}
-		}else if(peerCacheUsed>cacheblocks/2){
-			Thread.sleep(peerCacheUsed-cacheblocks/2);
-		}
 	}
 
 	private volatile long uackt=System.nanoTime();
@@ -356,8 +413,8 @@ public class KLALBCore {
 		synchronized( l) {
 			for (int i = 0; i < l.size(); i++) {
 				KLALBBlock block=l.get(i);
-				long timex = System.nanoTime() - block.sendtime;
+				long timex = (System.nanoTime() - block.sendtime)/1000000;
-				if(i<10) {
+				/*if(i<10) {
 					if(timex > Math.min(RTO,1000000000L )*(1<<Math.min(8,block.resend))+200*i) {
 						submitDataBlock(block);
 						block.sendtime = System.nanoTime();
@@ -368,8 +425,8 @@ public class KLALBCore {
 							i--;
 						}
 					}
-				}
+				}*/
-				/*if ((i<=10&&timex>500)||timex > (RTO/1000000)*(1<<Math.min(8,block.resend))+2000*i) {
+				if (timex > (1<<Math.min(4,block.resend))*(i<remotetcps.size()?Consts.MAX_QUEUE_TIME:1000*i)) {
 					submitDataBlock(block);
 					block.sendtime = System.nanoTime();
 				System.out.println("超时重传："+block+" 位置："+i);
@@ -378,17 +435,17 @@ public class KLALBCore {
 						l.remove(i);
 						i--;
 					}
-				}*/
+				}
 			}
 		}
 		long curr=System.nanoTime();
-		if(tc.getSendDeque().size()<cacheblocks&&curr-uackt>Consts.UACK_TIME) {
+		if(curr-uackt>Consts.UACK_TIME) {
 			uackt=curr;
 			KLALBBlock ack=new KLALBBlock();
 			ack.cuid=tc.getCuid();
 			ack.number=Long.MIN_VALUE;
 			ack.cacheused=tc.getSendDeque().size();
-			submitDataBlockNoDelay(ack);
+			submitAckBlockNoDelay(ack,3);
 		}
 		Thread.sleep(10);
 	}

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

@@ -47,6 +47,9 @@ public class KLALBServer {
 				tcc.setTunnel(tll);
 				
 				UUID uid=new UUID(din.readLong(),din.readLong());
+
+				tcc.handshake(10);
+				
 				IOThreadManager  nx = null;
 				synchronized (tcpl) {
 				if(whm.containsKey(uid)) {
@@ -80,8 +83,6 @@ public class KLALBServer {
 					whm.put(uid, nx);
 				}
 				}
-				tcc.getDout().write(0);
-				tcc.getDout().flush();
 				nx.handleRemote(tcc);
 			}catch(IOException e) {
 				e.printStackTrace();

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

@@ -7,8 +7,8 @@ import java.util.concurrent.atomic.AtomicLong;
 public class OutputMetre extends OutputStream {
 
 	private OutputStream out;
-	private AtomicLong total;
+	private AtomicLong[] total;
-	public OutputMetre(OutputStream outputStream,AtomicLong v) {
+	public OutputMetre(OutputStream outputStream,AtomicLong ...v) {
 		this.out=outputStream;
 		this.total=v;
 	}
@@ -16,19 +16,22 @@ public class OutputMetre extends OutputStream {
 	@Override
 	public void write(int b) throws IOException {
 		out.write(b);
-		total.incrementAndGet();
+		for(AtomicLong x:total)
+		x.incrementAndGet();
 	}
 
 	@Override
 	public void write(byte[] b) throws IOException {
 		out.write(b);
-		total.addAndGet(b.length);
+		for(AtomicLong x:total)
+		x.addAndGet(b.length);
 	}
 
 	@Override
 	public void write(byte[] b, int off, int len) throws IOException {
 		out.write(b, off, len);
-		total.addAndGet(len);
+		for(AtomicLong x:total)
+		x.addAndGet(len);
 	}
 
 	@Override

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

@@ -4,6 +4,7 @@ import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -21,6 +22,7 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.zip.DataFormatException;
 import java.util.zip.Deflater;
 import java.util.zip.GZIPInputStream;
@@ -39,23 +41,29 @@ public class RemoteTCPConnection extends TCPConnection {
 		t.getRtcs().add(this);
 		}
 	}
-	
+	AtomicLong om=new AtomicLong(0);
+	AtomicLong im=new AtomicLong(0);
+	long om1=0;
+	long im1=0;
+	private volatile long ups,downs;
+	private volatile double upsm,downsm;
 	@Override
 	protected void initIO() throws IOException {
 		
 	
 	}
+	
 	protected void initIOs() throws IOException {
 		connect.setTrafficClass(0x10);
 		connect.setTcpNoDelay(true);
 		OutputStream outm=null;
 		InputStream inm=null;
 		if(tunnel!=null) {
-		outm = new OutputMetre( connect.getOutputStream(),tunnel.getOM());
+		outm = new OutputMetre( connect.getOutputStream(),tunnel.getOM(),om);
-		inm = new InputMetre(connect.getInputStream(),tunnel.getIM());
+		inm = new InputMetre(connect.getInputStream(),tunnel.getIM(),im);
 		}else {
-			outm=connect.getOutputStream();
+			outm = new OutputMetre( connect.getOutputStream(),om);
-			inm=connect.getInputStream();
+			inm = new InputMetre(connect.getInputStream(),im);
 		}
 		
 		new DataOutputStream(outm).writeShort(59649);//59649
@@ -70,6 +78,62 @@ public class RemoteTCPConnection extends TCPConnection {
 		din = new DataInputStream(inm);
 	
 	}
+	public long getUps() {
+		return ups;
+	}
+
+	public long getDowns() {
+		return downs;
+	}
+
+	public AtomicLong getOm() {
+		return om;
+	}
+
+	public AtomicLong getIm() {
+		return im;
+	}
+	private volatile long ptime=System.nanoTime();
+	public void updateTraffics() {
+		long ctime=System.nanoTime();
+		long det=ctime-ptime;
+		ptime=ctime;
+		ups= (om.get()-om1)*1000000000/det;
+		downs= (im.get()-im1)*1000000000/det;
+		/*
+		if(ndbg==null)
+		initcdbg();
+			ndbg.println((om.get()-om1)+ (im.get()-im1)+","+delay);
+		*/
+		om1=om.get();
+		im1=im.get();
+		if(ups>upsm) {
+			upsm=ups;
+		}else {
+			upsm=(upsm*1000+ups)/1001;
+		}
+		if(downs>downsm) {
+			downsm=downs;
+		}else {
+			downsm=(downsm*1000+downs)/1001;
+		}
+		//predictLatency();
+		//System.out.println(tunnel+"//// "+upsm+" "+downsm+" "+getPredictedLatency());
+	}
+	private double latency;
+	public double getPredictedLatency() {
+		return latency;
+	}
+	public void predictLatency() {
+		AtomicLong data=new AtomicLong(0);
+		sendDeque.forEach((e)->{
+			if(e.data!=null) {
+				data.addAndGet(e.data.length);
+			}
+		});
+		latency= delay/2000000.0+data.get()*1000.0/upsm;
+	}
+
 	@Override
 	public String toString() {
 		return "RemoteTCPConnection [tunnel=" + tunnel + "]";
@@ -105,16 +169,6 @@ public class RemoteTCPConnection extends TCPConnection {
 
 	}
 	long odelay=Long.MAX_VALUE;
-	public void setDelayAvg(long delay) {
-		if(odelay==Long.MAX_VALUE) {
-			odelay=delay;
-			setDelay(odelay);
-		}else {
-			odelay=(delay+odelay*100)/101;
-			setDelay(odelay);
-		}
-	}
-	
 	@Override
 	public void close() {
 		if (tunnel != null) {
@@ -141,16 +195,16 @@ public class RemoteTCPConnection extends TCPConnection {
 			if (data.data == null) {
 				dout.writeShort(-1);
 			} else {
-				Deflater def = new Deflater(Deflater.BEST_COMPRESSION, true);
+				/*Deflater def = new Deflater(Deflater.BEST_COMPRESSION, true);
 		        def.setInput(data.data);
 		        def.finish();
 		        byte[] b = new byte[(int)(data.data.length * 1.1D) + 64];
 		        int nsize = def.deflate(b);
 		        def.end();
 		        this.dout.writeShort(nsize);
-		        this.dout.write(b, 0, nsize);
+		        this.dout.write(b, 0, nsize);*/
-				/*dout.writeShort(data.data.length);
+				dout.writeShort(data.data.length);
-				dout.write(data.data);*/
+				dout.write(data.data);
 				//System.out.println(Arrays.toString(data.data));
 			}
 		} else if (data.number == 0) {
@@ -179,7 +233,7 @@ public class RemoteTCPConnection extends TCPConnection {
 		if(isflush)
 		dout.flush();
 
-		//System.out.println("SEND:" + data);
+		System.out.println("SEND:" + data);
 	}
 
 	public KLALBBlock receiveBlock() throws IOException, DataFormatException {
@@ -193,7 +247,7 @@ public class RemoteTCPConnection extends TCPConnection {
 			} else {
 				byte[] d = new byte[size];
 				din.readFully(d);
-				 Inflater in = new Inflater(true);
+				 /*Inflater in = new Inflater(true);
 			        in.setInput(d);
 			        byte[] b = new byte[8192];
 			        int nsize = in.inflate(b);
@@ -202,9 +256,9 @@ public class RemoteTCPConnection extends TCPConnection {
 			          klb.data = b;
 			        } else {
 			          klb.data = Arrays.copyOf(b, nsize);
-			        } 
+			        } */
 				//System.out.println(Arrays.toString(d));
-				//klb.data = d;
+				klb.data = d;
 			}
 		} else if (klb.number == 0) {
 			klb.command = din.read();
@@ -229,7 +283,7 @@ public class RemoteTCPConnection extends TCPConnection {
 			klb.cacheused = din.readInt();
 		}
 
-		//System.out.println("RECEIVE:" + klb);
+		System.out.println("RECEIVE:" + klb);
 		return klb;
 	}
 
@@ -250,7 +304,7 @@ public class RemoteTCPConnection extends TCPConnection {
 	public Queue<KLALBBlock> getNDSendDeque() {
 		return NDsendDeque;
 	}
-
+	//private Queue<KLALBBlock> sendDeque = new ConcurrentLinkedQueue<KLALBBlock>();
 	private Queue<KLALBBlock> sendDeque = new PriorityBlockingQueue<KLALBBlock>(2, new Comparator<KLALBBlock>() {
 
 		@Override
@@ -260,14 +314,8 @@ public class RemoteTCPConnection extends TCPConnection {
 				}else if(o1.resend<o2.resend) {
 					return 1;
 				}else {
-			if(o1.number>o2.number) {
+			return 0;
-			return 1;
-			}else if(o1.number<o2.number) {
-				return -1;
-			}else {
-				return 0;
 			}
-				}
 		}
 	});
 
@@ -275,4 +323,16 @@ public class RemoteTCPConnection extends TCPConnection {
 		return sendDeque;
 	}
 
+	public void handshake(int x) throws IOException {
+		for(int i=0;i<x;i++) {
+			getDout().write(0);
+			getDout().flush();
+			int v=getDin().read();
+			if(v==-1) {
+				throw new EOFException();
+			}
+		}
+		
+	}
+
 }

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

@@ -2,8 +2,6 @@ package org.kne.cloud.network.klalb;
 
 public interface TListener {
 
-	void setDelay(long delay);
-
 	void setState(boolean state);
-	void setTraffic(long up,long down,long ups,long downs) ;
+	void setTraffic(long up,long down,long ups,long downs, long delay) ;
 }

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

@@ -50,12 +50,13 @@ public class TPanel extends JPanel{
 	public long getDelay() {
 		return delay;
 	}
-	public void setDelay(long delay) {
-		this.delay = delay;
-		updateText();
-	}
 	private void updateText() {
-		targ.setText(bytesUnit(up)+"\u2191 "+bytesUnit(down)+"\u2193 "+bytesUnit(ups)+"/s\u2191 "+bytesUnit(downs)+"/s\u2193 "+delay/1000000+"ms");
+		String dl;
+		if(delay==Long.MAX_VALUE)
+			dl="??";
+		else
+		dl=Long.toString(delay/1000000);
+		targ.setText(bytesUnit(up)+"\u2191 "+bytesUnit(down)+"\u2193 "+bytesUnit(ups)+"/s\u2191 "+bytesUnit(downs)+"/s\u2193 "+dl+"ms");
 	}
 	private String bytesUnit(long v) {
 		if(v>=1024L*1024*1024*1024*1024) {
@@ -72,11 +73,12 @@ public class TPanel extends JPanel{
 			return v+"B";
 		}
 	}
-	public void setTraffic(long up2, long down2, long ups, long downs) {
+	public void setTraffic(long up2, long down2, long ups, long downs, long delay2) {
 		up=up2;
 		down=down2;
 		this.ups=ups;
 		this.downs=downs;
+		this.delay=delay2;
 		updateText();
 	}
 	

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

@@ -62,7 +62,7 @@ public class Tunnel{
 			}
 		}
 	}
-	private boolean state=false;
+	private volatile boolean state=false;
 	
 	private TListener tlr;
 	
@@ -91,7 +91,7 @@ public class Tunnel{
 	public void setDelay(long delay) {
 		this.delay = delay;
 		if(tlr!=null)
-		tlr.setDelay(delay);
+			tlr.setTraffic(om.get(), im.get(), ups, downs,delay);
 	}
 
 	public Tunnel() {
@@ -179,6 +179,18 @@ public class Tunnel{
 		return socket;
 	}
 	
+	public long getUp() {
+		return om.get();
+	}
+	public long getDown() {
+		return im.get();
+	}
+	public long getUps() {
+		return ups;
+	}
+	public long getDowns() {
+		return downs;
+	}
 	@Override
 	public int hashCode() {
 		return Objects.hash(getIpport(), getName());
@@ -218,6 +230,7 @@ public class Tunnel{
 	long om1=0;
 	long im1=0;
 	
+	private volatile long ups,downs;
 	
 	public AtomicLong getOM() {
 		return om;
@@ -233,8 +246,13 @@ public class Tunnel{
 			e.printStackTrace();
 		}
 	}*/
-	public void updateTraffics(long timems) {
+	private volatile long ptime=System.nanoTime();
-		tlr.setTraffic(om.get(), im.get(), om.get()-om1, im.get()-im1);
+	public void updateTraffics() {
+		long ctime=System.nanoTime();
+		long det=ctime-ptime;
+		ptime=ctime;
+		ups= (om.get()-om1)*1000000000/det;
+		downs= (im.get()-im1)*1000000000/det;
 		/*
 		if(ndbg==null)
 		initcdbg();
@@ -242,6 +260,9 @@ public class Tunnel{
 		*/
 		om1=om.get();
 		im1=im.get();
+		
+		if(tlr!=null)
+		tlr.setTraffic(om.get(), im.get(), ups, downs,delay);
 	}
 	public void destroyFrpc() {
 		if(p!=null) {

src/org/kne/cloud/network/mport/PortMultiUse.java

@@ -18,7 +18,7 @@ public class PortMultiUse {
 		}
 		if(f.length()<=0) {
 			ept();
-			return;//https://101eduyun.com
+			return;
 		}
 		Scanner scn=new Scanner(f);
 		int remp=scn.nextInt();