修复线程不能正确关闭的问题

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

@@ -10,6 +10,7 @@ import java.util.Map;
 import java.util.UUID;
 import java.util.Vector;
 import java.util.WeakHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.kne.cloud.network.mport.ServiceElement;
 import org.kne.cloud.network.mport.ThreadTool;
@@ -24,6 +25,7 @@ public class IOThreadManager {
 
 	public void handleLocal(LocalTCPConnection tc, boolean syn) {
 		klc.getLocaltcps().put(tc.getCuid(), tc);
+		AtomicBoolean AB=new AtomicBoolean(true);
 		try {
 			if (syn) {
 				KLALBBlock sbk = new KLALBBlock();
@@ -39,7 +41,7 @@ public class IOThreadManager {
 			Thread lt = ThreadTool.makeVThreadIfSupport("数据包计时重发线程", () -> {
 
 				try {
-					while (true) {
+					while (AB.get()) {
 						klc.outputTimer(tc);
 					}
 				} catch (InterruptedException e) {
@@ -49,16 +51,16 @@ public class IOThreadManager {
 			Thread ls = ThreadTool.makeVThreadIfSupport("本地发送线程", () -> {
 
 				try {
-					while (true) {
+					while (AB.get()) {
 						KLALBBlock rd=klc.getDataBlock(tc);
 						if(rd.data==null) {
 							break;
 						}
 						tc.unpackBlock(rd);
 					}
-				} catch (InterruptedException e) {
 
-				} catch (IOException e) {
+				} catch (IOException|InterruptedException e) {
+					AB.set(false);
 					lt.interrupt();
 					tc.getOutputcache().clear();
 					KLALBBlock sbk = new KLALBBlock();
@@ -77,14 +79,15 @@ public class IOThreadManager {
 			});
 			Thread lr = ThreadTool.makeVThreadIfSupport("本地接收线程", () -> {
 				try {
-					while (true) {
+					while (AB.get()) {
 						KLALBBlock ks = tc.packBlock();
 						klc.putDataBlock(tc, ks);
 						if (ks.data == null) {
 							break;
 						}
 					}
-				} catch (IOException e) {
+				} catch (IOException|InterruptedException e) {
+					AB.set(false);
 					ls.interrupt();
 					lt.interrupt();
 					tc.getOutputcache().clear();
@@ -94,7 +97,6 @@ public class IOThreadManager {
 					sbk.command = 3;
 					klc.submitDataBlockNoDelay(sbk);
 					e.printStackTrace();
-				} catch (InterruptedException e) {
 				} finally {
 					try {
 						tc.getDin().close();
@@ -104,7 +106,12 @@ public class IOThreadManager {
 				}
 			});
 			
-
+			tc.setRSTHook((t)->{
+				AB.set(false);
+				ls.interrupt();
+				lt.interrupt();
+				tc.getOutputcache().clear();
+			});
 			ls.start();
 			lr.start();
 			lt.start();

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

@@ -8,7 +8,7 @@ import java.util.concurrent.atomic.AtomicLong;
 import org.kne.cloud.network.mport.IPPort;
 
 public class KLALBBlock implements Comparable<KLALBBlock>{
-	private static AtomicLong sng=new AtomicLong(0);
+	private static AtomicLong sng=new AtomicLong(1);
 	
 	public long sn;//每个数据包的唯一编号
 	public UUID cuid;//用于识别数据包的stream ID号

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

@@ -40,9 +40,9 @@ public class KLALBClient {
 			try {
 				tcc = new LocalTCPConnection(s);
 				tcc.setServiceElement(sel);
-				System.out.println("TCP："+tcc.getCuid()+"已连接");
+				//System.out.println("TCP："+tcc.getCuid()+"已连接");
 					iom.handleLocal(tcc,true);
-					System.out.println("TCP："+tcc.getCuid()+"已关闭");
+					//System.out.println("TCP："+tcc.getCuid()+"已关闭");
 			} catch (IOException e) {
 				e.printStackTrace();
 			}finally {
@@ -55,9 +55,11 @@ public class KLALBClient {
 		for (int i = 0; i < tls.size(); i++) {
 			Tunnel tll=tls.get(i);
 			ThreadTool.makeVThreadIfSupport("隧道监视线程", ()->{
+				int re=0;
 				RemoteTCPConnection tc=null;
 				while(iom.isOpen()) {
 					try {
+						re++;
 						tc=new RemoteTCPConnection(tll);
 						tc.getDout().writeShort(59649);
 						tc.getDout().write(1);
@@ -71,6 +73,7 @@ public class KLALBClient {
 						if(v==-1) {
 							throw new EOFException();
 						}
+						re=0;
 						System.out.println(tll+"：连接成功");
 						iom.handleRemote(tc);
 						System.out.println(tll+"：连接断开");
@@ -85,7 +88,8 @@ public class KLALBClient {
 						tc.close();
 					}
 					try {
-						Thread.sleep(5000);
+						//System.out.println(re);
+						Thread.sleep(500*(1<<Math.min(re,8)));
 					} catch (InterruptedException e) {
 						e.printStackTrace();
 					}

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

@@ -98,6 +98,7 @@ public class KLALBCore {
 		for(;;) {
 			if(tc.checkPingTime()) {
 				KLALBBlock pdb=new KLALBBlock();
+				pdb.sn=0;
 				pdb.cuid=ZERO_UUID;
 				pdb.number=0;
 				pdb.command=0;
@@ -118,7 +119,7 @@ public class KLALBCore {
 		KLALBBlock brc=null;
 		for(;;) {
 		brc=tc.receiveBlock();
-		if(brc.number!=0) {
+		if(brc.sn==0||brc.number!=0) {
 			break;
 		}
 		SN s=new SN(brc.sn);
@@ -145,6 +146,7 @@ public class KLALBCore {
 			if(brc.number==0) {
 				if(brc.command==0) {
 					KLALBBlock pdb=new KLALBBlock();
+					pdb.sn=0;
 					pdb.cuid=ZERO_UUID;
 					pdb.number=0;
 					pdb.command=1;
@@ -172,6 +174,7 @@ public class KLALBCore {
 					LocalTCPConnection ltc=localtcps.get(brc.cuid);
 					if(ltc!=null) {
 						ltc.close();
+						ltc.getRSTHook().accept(ltc);
 					}
 				}
 			}else if(brc.number>0){
@@ -244,9 +247,13 @@ public class KLALBCore {
 		submitDataBlock(ks);
 		ks.sendtime=System.nanoTime();
 		tc.getOutputcache().add(ks);
+		if(tc.getPeerCacheUsed()>cacheblocks) {
 		while(tc.getPeerCacheUsed()>cacheblocks) {
 			Thread.sleep(1);
 		}
+		}else if(tc.getPeerCacheUsed()>cacheblocks/2){
+			Thread.sleep(tc.getPeerCacheUsed()-cacheblocks/2);
+		}
 	}
 
 	private volatile long uackt=System.nanoTime();
@@ -256,7 +263,7 @@ public class KLALBCore {
 			for (int i = 0; i < l.size(); i++) {
 				KLALBBlock block=l.get(i);
 				long timex = (System.nanoTime() - block.sendtime) / 1000000;
-				if (timex > 200+1000*i) {
+				if (timex > 200*(1<<Math.min(8,block.resend))+1000*i) {
 					submitDataBlock(block);
 					block.sendtime = System.nanoTime();
 					block.resend++;

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

@@ -5,6 +5,7 @@ import java.net.Socket;
 import java.util.*;
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.function.Consumer;
 
 import org.kne.cloud.network.mport.ServiceElement;
 
@@ -90,10 +91,17 @@ private volatile long outputcount = 1;
 		return sendDeque;
 	}
 	private volatile int pcu=0;
+	private Consumer<LocalTCPConnection> hook;
 	public void setPeerCacheUsed(int cacheused) {
 		pcu= cacheused;
 	}
 	public int getPeerCacheUsed() {
 		return pcu;
 	}
+	public Consumer<LocalTCPConnection> getRSTHook() {
+		return hook;
+	}
+	public void setRSTHook(Consumer<LocalTCPConnection> tc) {
+		this.hook=tc;
+	}
 }

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

@@ -46,8 +46,10 @@ public class RemoteTCPConnection extends TCPConnection {
 
 	public void sendBlock(KLALBBlock data) throws IOException {
 		dout.writeLong(data.sn);
+		if(data.sn!=0) {
 		dout.writeLong(data.cuid.getMostSignificantBits());
 		dout.writeLong(data.cuid.getLeastSignificantBits());
+		}
 		dout.writeLong(data.number);
 		if (data.number > 0) {
 			if (data.data == null) {
@@ -80,7 +82,11 @@ public class RemoteTCPConnection extends TCPConnection {
 	public KLALBBlock receiveBlock() throws IOException {
 		KLALBBlock klb = new KLALBBlock();
 		klb.sn = din.readLong();
+		if(klb.sn!=0) {
 		klb.cuid = new UUID(din.readLong(), din.readLong());
+		}else {
+			klb.cuid=new UUID(0,0);
+		}
 		klb.number = din.readLong();
 		if (klb.number > 0) {
 			int size = din.readInt();

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

@@ -16,6 +16,8 @@ import java.util.WeakHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import javax.naming.spi.Resolver;
+
 import org.kne.cloud.network.mport.IPPort;
 
 public class Tunnel {