性能优化

.classpath

@@ -1,10 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <classpath>
-	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/zulu19.30.11-ca-jdk19.0.1">
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
-		<attributes>
-			<attribute name="module" value="true"/>
-		</attributes>
-	</classpathentry>
 	<classpathentry kind="src" path="src"/>
 	<classpathentry kind="output" path="bin"/>
 </classpath>

.settings/org.eclipse.jdt.core.prefs

@@ -1,9 +1,9 @@
 eclipse.preferences.version=1
 org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
 org.eclipse.jdt.core.compiler.codegen.methodParameters=do not generate
-org.eclipse.jdt.core.compiler.codegen.targetPlatform=18
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
-org.eclipse.jdt.core.compiler.compliance=18
+org.eclipse.jdt.core.compiler.compliance=1.8
 org.eclipse.jdt.core.compiler.debug.lineNumber=generate
 org.eclipse.jdt.core.compiler.debug.localVariable=generate
 org.eclipse.jdt.core.compiler.debug.sourceFile=generate
@@ -11,5 +11,5 @@ org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
 org.eclipse.jdt.core.compiler.problem.enablePreviewFeatures=disabled
 org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
 org.eclipse.jdt.core.compiler.problem.reportPreviewFeatures=warning
-org.eclipse.jdt.core.compiler.release=enabled
+org.eclipse.jdt.core.compiler.release=disabled
-org.eclipse.jdt.core.compiler.source=18
+org.eclipse.jdt.core.compiler.source=1.8

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

@@ -3,6 +3,7 @@ package org.kne.cloud.network.klalb;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.SocketException;
 import java.util.List;
 import java.util.Vector;
 
@@ -11,11 +12,10 @@ public class IOThreadManager {
 	private List<TCPConnection>tcps=new Vector<>();
 	
 	
-	private InputStream in;
+	private TCPConnection local;
-	private OutputStream out;
 	
 
-	private volatile boolean closed=true;
+	private volatile boolean open=true;
 	public void startLocal() {
 
 		Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{
@@ -23,30 +23,32 @@ public class IOThreadManager {
 				
 				while(true) {
 					byte[]b=new byte[Consts.BLOCKSIZE];
-					int size=in.read(b);
+					int size=local.getDin().read(b);
 					if(size==-1)
 						break;
 					klc.packDataBlock(b,size);
 				}
+				klc.waitForRemote();
 			}catch(InterruptedException s) {
+			}catch(SocketException se) {
 			}catch(Exception e) {
 				e.printStackTrace();
 			}finally {
-				closeALL();
+				closeRemote();
 			}
 		});
 		Thread downo=ThreadTool.makeVThreadIfSupport("本地发送线程",()->{
 			try{
 				while(true) {
 
-					out.write(klc.unpackDataBlock());
+					local.getDout().write(klc.unpackDataBlock());
-					out.flush();
+					local.getDout().flush();
 				}
 				}catch(InterruptedException s) {
 				}catch(Exception e) {
 					e.printStackTrace();
 				}finally {
-					closeALL();
+					closeRemote();
 				}
 		});
 		upo.start();
@@ -80,7 +82,6 @@ public class IOThreadManager {
 		up.start();
 		down.start();
 		try {
-			up.join();
 			down.join();
 	} catch (InterruptedException e) {
 			e.printStackTrace();
@@ -89,23 +90,39 @@ public class IOThreadManager {
 		tcps.remove(s);
 		}
 	}
-	public InputStream getIn() {
+	
-		return in;
+	public void closeRemote() {
+		open=false;
+		
+		for (int i = 0; i < tcps.size(); i++) {
+			TCPConnection tll=tcps.get(i);
+			tll.close();
+		}
+		klc.closeRemote();
 	}
 	
-	public void setIn(InputStream in) {
+	public void closeLocal() {
-		this.in = in;
+		open=false;
+		local.close();
+		klc.closeLocal();
 	}
 	
-	public OutputStream getOut() {
+	public TCPConnection getLocal() {
-		return out;
+		return local;
 	}
-
+	public void setLocal(TCPConnection local) {
-	public void setOut(OutputStream out) {
+		this.local = local;
-		this.out = out;
 	}
+	/*
 	public void closeALL() {
-		closed=false;
+		open=false;
+		
+		for (int i = 0; i < tcps.size(); i++) {
+			TCPConnection tll=tcps.get(i);
+			tll.close();
+		}
+		klc.closeRemote();
+		klc.closeLocal();
 		try {
 			in.close();
 			
@@ -117,16 +134,12 @@ public class IOThreadManager {
 		} catch (IOException e) {
 			e.printStackTrace();
 		}
-		for (int i = 0; i < tcps.size(); i++) {
-			TCPConnection tll=tcps.get(i);
-			tll.close();
-		}
-		klc.close();
 	}
+	*/
 	public List<TCPConnection> getTcps() {
 		return tcps;
 	}
-	public boolean isClosed() {
+	public boolean isOpen() {
-		return closed;
+		return open;
 	}
 }

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

@@ -6,6 +6,8 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class KLALBClient {
 		private	List<Tunnel> tls=new ArrayList<>();
@@ -16,8 +18,7 @@ public class KLALBClient {
 					IOThreadManager kcp=new IOThreadManager();
 					UUID uid=UUID.randomUUID();
 					try {
-						kcp.setIn(new BufferedInputStream(s.getInputStream(),8192));
+					kcp.setLocal(new TCPConnection(null, s));
-						kcp.setOut(new BufferedOutputStream(s.getOutputStream(),8192));
 						kcp.startLocal();
 						runProtocol(kcp,uid);
 					} catch (IOException e) {
@@ -27,28 +28,40 @@ public class KLALBClient {
 		});
 	}
 	public void runProtocol(IOThreadManager kcp, UUID uid) {
+		AtomicBoolean b=new AtomicBoolean(true);
+		AtomicInteger aig=new AtomicInteger(0);
 		for (int i = 0; i < tls.size(); i++) {
 			Tunnel tll=tls.get(i);
 			ThreadTool.makeVThreadIfSupport("隧道监视线程",()->{
-				while (kcp.isClosed()) {
+				while (kcp.isOpen()) {
 					try {
 						TCPConnection tc=new TCPConnection(tll);
 						tc.getDout().writeShort(59649);
 						tc.getDout().writeLong(uid.getMostSignificantBits());
 						tc.getDout().writeLong(uid.getLeastSignificantBits());
 						tc.getDout().flush();
-
+						aig.incrementAndGet();
 						System.out.println("隧道"+tll+"已连接，可用线路数量:"+ (kcp.getTcps().size()+1));
 						kcp.handleSocket(tc);
 						
 						int n=kcp.getTcps().size();
 						System.out.println("隧道"+tll+"已断开，可用线路数量:"+ n);
 						if(n<=0) {
-							kcp.closeALL();
+							kcp.closeRemote();
+							kcp.closeLocal();
 							System.out.println("连接已断开");
 							return;
 						}
 					} catch (IOException e) {
+						if(aig.get()<=0) {
+								kcp.closeRemote();
+								kcp.closeLocal();
+								if(b.get()) {
+								System.out.println("连接失败");
+								b.set(false);
+								}
+								return;
+						}
 					}
 					
 						try {

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

@@ -1,139 +0,0 @@
-package org.kne.cloud.network.klalb;
-
-import java.io.*;
-import java.util.Iterator;
-import java.util.List;
-import java.util.UUID;
-import java.util.Vector;
-
-public class KLALBClientProtocol {
-	private KLALBCore klc=new KLALBCore(100);
-	
-	private List<TCPConnection>tcps=new Vector<>();
-	private InputStream in;
-	private OutputStream out;
-	private volatile boolean closed=true;
-
-
-	public void startLocal() {
-		Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{
-			try{
-				
-				while(true) {
-					byte[]b=new byte[Consts.BLOCKSIZE];
-					int size=in.read(b);
-					if(size==-1)
-						break;
-					klc.packDataBlock(b,size);
-				}
-			}catch(InterruptedException s) {
-				
-			}catch(Exception e) {
-				e.printStackTrace();
-			}finally {
-				closeALL();
-			}
-		});
-		Thread downo=ThreadTool.makeVThreadIfSupport("本地发送线程",()->{
-			try{
-				while(true) {
-					out.write(klc.unpackDataBlock());
-					out.flush();
-				}
-				}catch(InterruptedException s) {
-					
-				}catch(Exception e) {
-					e.printStackTrace();
-				}finally {
-					closeALL();
-				}
-		});
-		upo.start();
-		downo.start();
-	}
-	public void handleSocket(TCPConnection tc) throws IOException {
-		tcps.add(tc);
-		Thread up=ThreadTool.makeVThreadIfSupport("远程发送线程",()->{
-			try{
-			while(true) {
-				klc.sendDataBlock(tc);
-			}
-			}catch(InterruptedException s) {
-				
-			}catch(Exception e) {
-				tc.close();
-				
-			}
-		});
-		Thread down=ThreadTool.makeVThreadIfSupport("远程接收线程",()->{
-			try{
-				while(true) {
-					klc.receiveDataBlock(tc);
-				}
-				}catch(InterruptedException s) {
-					
-				}catch(Exception e) {
-					tc.close();
-					up.interrupt();
-				}
-		});
-		up.start();
-		down.start();
-		try {
-			up.join();
-			down.join();
-	} catch (InterruptedException e) {
-			e.printStackTrace();
-			}
-				tcps.remove(tc);
-	}
-	
-
-	
-
-	public InputStream getIn() {
-		return in;
-	}
-
-	public void setIn(InputStream in) {
-		this.in = in;
-	}
-
-	public OutputStream getOut() {
-		return out;
-	}
-
-	public void setOut(OutputStream out) {
-		this.out = out;
-	}
-	public void closeALL() {
-		closed=false;
-		try {
-			in.close();
-			
-		} catch (IOException e) {
-			e.printStackTrace();
-		} 
-		try {
-			out.close();
-		} catch (IOException e) {
-			e.printStackTrace();
-		}
-		synchronized (tcps) {
-			
-		for (int i = 0; i < tcps.size(); i++) {
-			TCPConnection tll=tcps.get(i);
-			tll.close();
-		}
-		}
-		klc.close();
-	}
-	public List<TCPConnection> getTcps() {
-		return tcps;
-	}
-
-	public boolean isClosed() {
-		return closed;
-	}
-	
-}

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

@@ -31,23 +31,24 @@ public class KLALBCore {
 	private Set<KLALBBlock> inputcache = Collections.synchronizedSet(new HashSet<>());
 	private List<KLALBBlock> outputcache = Collections.synchronizedList(new ArrayList<>());
 	
+	private Map<TCPConnection, BlockingQueue<KLALBBlock>>acks=Collections.synchronizedMap(new WeakHashMap<>());
+	
 	private volatile boolean inlocal=true;
 	
-	private volatile boolean close = false;
+	private volatile boolean closeremote = false;
-	
+	private volatile boolean closelocal = false;
 	
 	private volatile int cacheblocks;
 	public KLALBCore(int cachesize) {
 		cacheblocks=cachesize;
 	}
 	
-	private ExecutorService exec=Executors.newCachedThreadPool();
 	public void packDataBlock(byte[] b, int size) throws InterruptedException {
 		
-		if (close)
+		if (closelocal)
 			throw new InterruptedException();
 		while (outputcache.size() > cacheblocks) {
-			if (close)
+			if (closelocal)
 				Thread.currentThread().interrupt();
 			Thread.sleep(1);
 		}
@@ -58,7 +59,7 @@ public class KLALBCore {
 	public byte[] unpackDataBlock() throws InterruptedException {
 		inlocal=false;
 		try {
-		if (close)
+		if (closelocal)
 			throw new InterruptedException();
 		byte[] b = null;
 		while (true) {
@@ -79,7 +80,7 @@ public class KLALBCore {
 				inputcount++;
 				return b;
 			}
-			if (close)
+			if (closelocal)
 				Thread.currentThread().interrupt();
 			Thread.sleep(1);
 
@@ -91,13 +92,20 @@ public class KLALBCore {
 
 	
 	public void sendDataBlock(TCPConnection out) throws IOException, InterruptedException {
-		if (close)
+		if (closeremote)
 			throw new InterruptedException();
-		while ( outputcache.isEmpty()) {
+		BlockingQueue<KLALBBlock> bqk=acks.get(out);
-			if (close)
+		while ((bqk==null||bqk.isEmpty())&& outputcache.isEmpty()) {
+			if (closeremote)
 				Thread.currentThread().interrupt();
 			Thread.sleep(1);
 		}
+	
+		if(bqk!=null&&bqk.size()>0) {
+			KLALBBlock klb=bqk.poll();
+			if(klb!=null)
+			send0(out, klb);
+		}else {
 			KLALBBlock ks = null;
 			synchronized (outputcache) {
 				for (int i = 0; i < outputcache.size(); i++) {
@@ -131,30 +139,41 @@ public class KLALBCore {
 			if (ks != null) {
 				send0(out, ks);
 		//	System.out.println(outputcache.size());
+			}else {
+				Thread.sleep(1);
+			}
 		}
-		
 	}
 
 	public void receiveDataBlock(TCPConnection in) throws IOException, InterruptedException {
-		if (close)
+		if (closeremote)
 			throw new InterruptedException();
 		KLALBBlock x = receive0(in);
 		if (x.number > 0) {
 			
 				KLALBBlock klb=new KLALBBlock(null, 0, -x.number);
-				exec.execute( ()->{
+				BlockingQueue<KLALBBlock> bq=acks.get(in);
+				if(bq==null) {
+					BlockingQueue<KLALBBlock> bqt=new LinkedBlockingQueue<>();
+					bqt.add(klb);
+					acks.put(in,bqt);
+				}else {
+					acks.get(in).add(klb);
+				}
+				/*ThreadTool.makeVThreadIfSupport("TACK", ()->{
 					try {
 						send0(in, klb);
 					} catch (IOException e) {
 						e.printStackTrace();
 					}finally {
 					}
-				});
+				}).start();*/
+				
 				
 				if (x.number >= inputcount) {
 				
 				while(inputcache.size()>(2*cacheblocks)&&inlocal) {
-					if (close)
+					if (closeremote)
 						Thread.currentThread().interrupt();
 						Thread.sleep(1);
 					}
@@ -178,9 +197,8 @@ public class KLALBCore {
 			out.writeInt(kd.size);
 			out.write(kd.data, 0, kd.size);
 		}
+		out.flush();
 		
-		
-		tcp.flush();
 		}
 		 System.out.println("SEND:" + kd);
 	}
@@ -200,12 +218,27 @@ public class KLALBCore {
 		return kb;
 	}
 
-	public void close() {
+	public void closeRemote() {
-		close = true;
+		closeremote = true;
-		exec.shutdown();
+		outputcache.clear();
 	}
-	public void waitForEnding() {
+	public void closeLocal() {
-		while(outputcache.size()>0||inputcache.size()>0) {
+		closelocal = true;
+		inputcache.clear();
+	}
+	
+
+	public void waitForRemote() {
+		while(outputcache.size()>0) {
+			try {
+				Thread.sleep(1);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+	}
+	public void waitForLocal() {
+		while(inputcache.size()>0) {
 			try {
 				Thread.sleep(1);
 			} catch (InterruptedException e) {

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

@@ -28,8 +28,7 @@ public class KLALBServer {
 				}else {
 					nx=new IOThreadManager();
 					Socket soc=new Socket("192.168.1.233",8444);
-					nx.setOut(soc.getOutputStream());
+					nx.setLocal(new TCPConnection(null, soc));
-					nx.setIn(soc.getInputStream());
 					nx.startLocal();
 					whm.put(uid, nx);
 					
@@ -37,7 +36,7 @@ public class KLALBServer {
 				nx.handleSocket(new TCPConnection(null,s));
 				int n=nx.getTcps().size();
 				if(n<=0) {
-					nx.closeALL();
+					nx.closeLocal();
 					System.out.println("连接已关闭");
 				}
 			}catch(ConnectException e) {

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

@@ -1,140 +0,0 @@
-package org.kne.cloud.network.klalb;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.Socket;
-import java.util.*;
-public class KLALBServerProtocol {
-	private KLALBCore klc=new KLALBCore(100);
-	private List<TCPConnection>tcps=new Vector<>();
-	
-	
-	private InputStream in;
-	private OutputStream out;
-	
-
-	private volatile boolean closed=true;
-	public void startLocal() {
-
-		Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{
-			try{
-				
-				while(true) {
-					byte[]b=new byte[Consts.BLOCKSIZE];
-					int size=in.read(b);
-					if(size==-1)
-						break;
-					klc.packDataBlock(b,size);
-				}
-			}catch(InterruptedException s) {
-				
-			}catch(Exception e) {
-				e.printStackTrace();
-			}finally {
-				closeALL();
-			}
-		});
-		Thread downo=ThreadTool.makeVThreadIfSupport("本地发送线程",()->{
-			try{
-				while(true) {
-
-					out.write(klc.unpackDataBlock());
-					out.flush();
-				}
-				}catch(InterruptedException s) {
-					
-				}catch(Exception e) {
-					e.printStackTrace();
-				}finally {
-					closeALL();
-				}
-		});
-		upo.start();
-		downo.start();
-	}
-	public void handleSocket(TCPConnection s) throws IOException {
-		tcps.add(s);
-		Thread up=ThreadTool.makeVThreadIfSupport("远程发送线程",()->{
-			try{
-			while(true) {
-				klc.sendDataBlock(s);
-			}
-			}catch(InterruptedException s1) {
-				
-			}catch(Exception e) {
-			}
-		});
-		Thread down=ThreadTool.makeVThreadIfSupport("远程接收线程",()->{
-			try{
-				while(true) {
-					klc.receiveDataBlock(s);
-				}
-				}catch(InterruptedException s1) {
-					
-				}catch(Exception e) {
-					up.interrupt();
-				}
-		});
-
-		up.start();
-		down.start();
-		try {
-			up.join();
-			down.join();
-	} catch (InterruptedException e) {
-			e.printStackTrace();
-		}
-		tcps.remove(s);
-		int n=tcps.size();
-		if(n<=0) {
-			closeALL();
-			System.out.println("连接已关闭");
-		}
-	}
-	public InputStream getIn() {
-		return in;
-	}
-
-	public void setIn(InputStream in) {
-		this.in = in;
-	}
-
-	public OutputStream getOut() {
-		return out;
-	}
-
-	public void setOut(OutputStream out) {
-		this.out = out;
-	}
-	public void closeALL() {
-		closed=false;
-		try {
-			in.close();
-			
-		} catch (IOException e) {
-			e.printStackTrace();
-		} 
-		try {
-			out.close();
-		} catch (IOException e) {
-			e.printStackTrace();
-		}
-		for (int i = 0; i < tcps.size(); i++) {
-			TCPConnection tll=tcps.get(i);
-			tll.close();
-		}
-		klc.close();
-	}
-	public List<TCPConnection> getTcps() {
-		return tcps;
-	}
-	public boolean isClosed() {
-		return closed;
-	}
-	
-}

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

@@ -34,7 +34,6 @@ public class TCPConnection {
 	}
 
 	public void close() {
-		// TODO 自动生成的方法存根
 		try {
 			if (din != null)
 				din.close();
@@ -70,36 +69,19 @@ public class TCPConnection {
 	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(10);
-					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 boolean isOpen() {
+		if(connect==null) {
+			return false;
+		}
+		return !connect.isClosed();
+	}
 
-	public void flush() throws IOException {
-		flush = true;
-	}
 }

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

@@ -6,8 +6,8 @@ public class ThreadTool {
 	public static Thread makeVThreadIfSupport(String name,Runnable r) {
 		if(forceD)
 			return new Thread(r, name);
-		try {
+		try {	return new Thread(r, name);
-			return Thread.ofVirtual().name(name).unstarted(r);
+			//return Thread.ofVirtual().name(name).unstarted(r);
 		}catch(Throwable e) {
 			if(first) {
 			System.out.println("请使用java19以上版本以提高性能！");