再次优化调度性能

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

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

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

@@ -8,7 +8,7 @@ import java.util.List;
 import java.util.Vector;
 
 public class IOThreadManager {
-	private KLALBCore klc=new KLALBCore(100);
+	private KLALBCore klc=new KLALBCore(500);
 	private List<TCPConnection>tcps=new Vector<>();
 	
 	

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

@@ -10,6 +10,7 @@ import java.net.URL;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Scanner;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -17,6 +18,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 public class KLALBClient {
 		private	List<Tunnel> tls=new ArrayList<>();
 		private 	TCPListener tcpl;
+		private ServiceElement sel;
 	public KLALBClient(int port) throws IOException {
 	tcpl=new TCPListener(port);
 		tcpl.setCon((s)->{
@@ -43,6 +45,11 @@ public class KLALBClient {
 						TCPConnection tc=new TCPConnection(tll);
 						tc.getDout().writeShort(59649);
 						tc.getDout().write(1);
+						
+						tc.getDout().writeUTF(sel.ipport.getIp().getHostAddress());
+						tc.getDout().writeInt(sel.ipport.getPort());
+						tc.getDout().writeUTF(sel.proc);
+						
 						tc.getDout().writeUTF(tll.getName());
 						tc.getDout().writeUTF(tll.getIp());
 						tc.getDout().writeInt(tll.getPort());
@@ -50,15 +57,15 @@ public class KLALBClient {
 						tc.getDout().writeLong(uid.getLeastSignificantBits());
 						tc.getDout().flush();
 						aig.incrementAndGet();
-						System.out.println("隧道"+tll+"已连接，可用线路数量:"+ (kcp.getTcps().size()+1));
+						//System.out.println("隧道"+tll+"已连接，可用线路数量:"+ (kcp.getTcps().size()+1));
 						kcp.handleSocket(tc);
-						
+						tc.close();
 						int n=kcp.getTcps().size();
-						System.out.println("隧道"+tll+"已断开，可用线路数量:"+ n);
+						//System.out.println("隧道"+tll+"已断开，可用线路数量:"+ n);
 						if(n<=0) {
 							kcp.closeRemote();
 							kcp.closeLocal();
-							System.out.println("连接已断开");
+							//System.out.println("连接已断开");
 							return;
 						}
 					} catch (IOException e) {
@@ -66,7 +73,7 @@ public class KLALBClient {
 								kcp.closeRemote();
 								kcp.closeLocal();
 								if(b.get()) {
-								System.out.println("连接失败");
+								System.out.println("隧道连接失败");
 								b.set(false);
 								}
 								return;
@@ -92,8 +99,11 @@ public class KLALBClient {
 	public List<Tunnel> getTls() {
 		return tls;
 	}
-	public static List<ServiceElement> services=new ArrayList<ServiceElement>();
+	public List<ServiceElement> services=new ArrayList<ServiceElement>();
 
+	public List<ServiceElement> getServices() {
+		return services;
+	}
 	public void searchTunnels(String ipport) throws IOException {
 		IPPort u=new IPPort(ipport);
 		TCPConnection tcpc=null;
@@ -108,19 +118,38 @@ public class KLALBClient {
 		}
 		int s2=tcpc.getDin().readInt();
 		for (int i = 0; i < s2; i++) {
-			tls.add(new Tunnel(tcpc.getDin().readUTF()));
+			tls.add(Tunnel.newTunnel(tcpc.getDin().readUTF()));
 		}
-		System.out.println(services);
+		//System.out.println(services);
-		System.out.println(tls);
+		//System.out.println(tls);
 		}finally {
 			if(tcpc!=null) {
 				tcpc.close();
 			}
 		}
 	}
+	
+	public ServiceElement getSel() {
+		return sel;
+	}
+	public void setSel(ServiceElement sel) {
+		this.sel = sel;
+	}
 	public static void main(String[] args) throws IOException {
+		Scanner scn=new Scanner(System.in);
 		KLALBClient kc=new KLALBClient(4568);
 		kc.searchTunnels("153.36.240.12:65529");
+		kc.setSel(kc.getServices().get(0));
+		kc.open();
+		while(true) {
+			String s=scn.next();
+			switch(s) {
+			case "states":
+				Tunnel.states();
+				break;
+			}
+		}
+	}
 	//Tunnel t=	new Tunnel("Test1", "127.0.0.1", 4569);
 			/*Tunnel t=	new Tunnel("Test1", "cn-hz-bgp-1.openfrp.top", 65529);
 			tls.add(t);
@@ -131,7 +160,7 @@ public class KLALBClient {
 			Tunnel t4=new Tunnel("Test4","cn-sx-xa-bgp-1.openfrp.top",65529);
 			tls.add(t4);
 			*/
-/*		List<Tunnel> tls=kc.getTls();
+	/*		List<Tunnel> tls=kc.getTls();
 			Tunnel t5=new Tunnel("Test5","la.afrps.cn",49966);
 			tls.add(t5);
 			Tunnel t6=new Tunnel("Test6","sg.afrps.cn",49966);
@@ -147,8 +176,5 @@ public class KLALBClient {
 			tls.add(t11);
 			Tunnel t12=new Tunnel("Test12","frp.freefrps.com",49965);
 			tls.add(t12);*/
-		kc.open();
-	}
-
 	
 }

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

@@ -31,8 +31,8 @@ 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 Map<TCPConnection, BlockingQueue<KLALBBlock>>acks=Collections.synchronizedMap(new WeakHashMap<>());
-	
+	private BlockingQueue<KLALBBlock> ackq=new LinkedBlockingQueue<>();
 	private volatile boolean inlocal=true;
 	
 	private volatile boolean closeremote = false;
@@ -94,8 +94,8 @@ public class KLALBCore {
 	public void sendDataBlock(TCPConnection out) throws IOException, InterruptedException {
 		if (closeremote)
 			throw new InterruptedException();
-		BlockingQueue<KLALBBlock> bqk=acks.get(out);
+		//BlockingQueue<KLALBBlock> bqk=acks.get(out);
-		while ((bqk==null||bqk.isEmpty())&& outputcache.isEmpty()) {
+		while (ackq.isEmpty()&& outputcache.isEmpty()) {
 			if (closeremote)
 				Thread.currentThread().interrupt();
 			Thread.sleep(1);
@@ -103,10 +103,8 @@ public class KLALBCore {
 				send0(out, new KLALBBlock(null,0 , 0));
 			}
 		}
-	
+		KLALBBlock klb=ackq.poll();
-		if(bqk!=null&&bqk.size()>0) {
+		if(klb!=null) {
-			KLALBBlock klb=bqk.poll();
-			if(klb!=null)
 			send0(out, klb);
 		}else {
 			KLALBBlock ks = null;
@@ -115,21 +113,21 @@ public class KLALBCore {
 
 					KLALBBlock kd = outputcache.get(i);
 					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);
+							//System.out.println("³¬Ê±ÖØ´«£º"+kd);
 								kd.time = System.nanoTime();
 								kd.thread = Thread.currentThread();
 								ks = kd;
-							}*/
+							}
 						} else {
-							System.out.println("µôÏßÖØ´«£º"+kd);
+							//System.out.println("µôÏßÖØ´«£º"+kd);
-							//kd.time = System.nanoTime();
+							kd.time = System.nanoTime();
 							kd.thread = Thread.currentThread();
 							ks = kd;
 						}
@@ -160,14 +158,15 @@ public class KLALBCore {
 		if (x.number > 0) {
 			
 				KLALBBlock klb=new KLALBBlock(null, 0, -x.number);
-				BlockingQueue<KLALBBlock> bq=acks.get(in);
+				ackq.add(klb);
+				/*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);
-				}
+				}*/
 				
 				
 				if (x.number >= inputcount) {

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

@@ -3,12 +3,14 @@ package org.kne.cloud.network.klalb;
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.Scanner;
 
 public class KLALBSM {
-	public static List<ServiceElement> services=new ArrayList<ServiceElement>();
+	public static Map<String, ServiceElement> services=new HashMap<String, ServiceElement>();
 
 	public static List<Tunnel> tunnels=new ArrayList<Tunnel>();
 	public static void main(String[] args) throws IOException {
@@ -39,10 +41,10 @@ public class KLALBSM {
 			case 0:
 			ServiceElement se=new ServiceElement(s);
 			System.out.println(se);
-			services.add(se);
+			services.put(se.proc,se);
 			break;
 			case 1:
-				Tunnel t=new Tunnel(s);
+				Tunnel t=Tunnel.newTunnel(s);
 				System.out.println(t);
 				tunnels.add(t);
 				break;
@@ -76,10 +78,10 @@ public class KLALBSM {
 							case 0:
 							ServiceElement se=new ServiceElement(s);
 							System.out.println(se);
-							services.add(se);
+							services.put(se.proc,se);
 							break;
 							case 1:
-								Tunnel t=new Tunnel(s);
+								Tunnel t=Tunnel.newTunnel(s);
 								System.out.println(t);
 								tunnels.add(t);
 								break;

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

@@ -4,14 +4,18 @@ import java.io.DataInputStream;
 import java.io.IOException;
 import java.net.ConnectException;
 import java.net.Socket;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.WeakHashMap;
 
 public class KLALBServer {
 	WeakHashMap<UUID, IOThreadManager> whm=new WeakHashMap<>();
-	public KLALBServer(int port, List<ServiceElement> services,List<Tunnel> tunnels) throws IOException {
+	public KLALBServer(int port, Map<String,ServiceElement> services,List<Tunnel> tunnels) throws IOException {
 		TCPListener tcpl=new TCPListener(port);
 		tcpl.setCon((s)->{
 			try { 
@@ -25,10 +29,11 @@ public class KLALBServer {
 				int x=din.read();
 				if(x==0) {
 					 synchronized (services) {
-						int counts=services.size();
+						 tcc.getDout().writeInt(services.size());
-						tcc.getDout().writeInt(counts);
+						 Collection<ServiceElement> cll=services.values();
-						for (int i = 0; i < counts; i++) {
+						 for (Iterator<ServiceElement> iterator = cll.iterator(); iterator.hasNext();) {
-							tcc.getDout().writeUTF(services.get(i).toString());
+							ServiceElement tunnel =  iterator.next();
+							tcc.getDout().writeUTF(tunnel.toString());
 						}
 					}
 					 synchronized (tunnels) {
@@ -41,10 +46,21 @@ public class KLALBServer {
 					 tcc.getDout().flush();
 				return;	
 				}
+				
+				String lip=din.readUTF();
+				int lport=din.readInt();
+				String lname=din.readUTF();
+				ServiceElement eas=services.get(lname);
+				System.out.println(eas);
+				
+				if(eas==null||(!eas.ipport.equals(new IPPort(lip, lport)))){
+					return;
+				}
+				
 				String name = din.readUTF();
 				String ip = din.readUTF();
 				int portx=din.readInt();
-				Tunnel tll=new Tunnel(name, ip, portx);
+				Tunnel tll=Tunnel.newTunnel(name, ip, portx);
 				tcc.setTunnel(tll);
 				
 				UUID uid=new UUID(din.readLong(),din.readLong());
@@ -55,7 +71,7 @@ public class KLALBServer {
 					nx=whm.get(uid);
 				}else {
 					nx=new IOThreadManager();
-					Socket soc=new Socket("192.168.1.233",8444);
+					Socket soc=new Socket("192.168.1.233",eas.ipport.getPort());
 					nx.setLocal(new TCPConnection(null, soc));
 					nx.startLocal();
 					whm.put(uid, nx);

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

@@ -20,11 +20,11 @@ public class TCPConnection {
 	}
 
 	private DataOutputStream dout;
-	private long delay;
+	private long delay=-1;
 
-	public Socket getConnect() {
+	/*public Socket getConnect() {
 		return connect;
-	}
+	}*/
 
 	public DataInputStream getDin() {
 		return din;
@@ -56,6 +56,8 @@ public class TCPConnection {
 			// TODO 自动生成的 catch 块
 			e.printStackTrace();
 		}
+		if(tunnel!=null&&connect!=null)
+		tunnel.getCCount().decrementAndGet();
 		connect = null;
 	}
 
@@ -65,9 +67,12 @@ public class TCPConnection {
 
 	public void setDelay(long delay) {
 		this.delay = delay;
+		if(tunnel!=null) {
+			tunnel.setDelay(delay);
+		}
 		if(connect!=null) {
 			try {
-				connect.setSoTimeout((int) (delay/500000));
+				connect.setSoTimeout((int) (delay/100000));
 			} catch (SocketException e) {
 				e.printStackTrace();
 			}
@@ -84,6 +89,9 @@ public class TCPConnection {
 	public TCPConnection(Tunnel t, Socket soc) throws IOException {
 		connect = soc;
 		tunnel = t;
+		if(t!=null) {
+			t.getCCount().incrementAndGet();
+		}
 		// connect.setSoTimeout(10000);
 		din = new DataInputStream(new BufferedInputStream(connect.getInputStream(), 65536));
 		dout = new DataOutputStream(new BufferedOutputStream(connect.getOutputStream(), 65536));
@@ -108,4 +116,6 @@ public class TCPConnection {
 		}
 	}
 
+	
+	
 }

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

@@ -7,28 +7,57 @@ import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.Socket;
 import java.net.UnknownHostException;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Objects;
+import java.util.Set;
+import java.util.WeakHashMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class Tunnel {
 	private String name;
 	private IPPort ipport;
-	
+	private long delay=-1;
-	public Tunnel(String name, String ip, int port) throws UnknownHostException {
+	public long getDelay() {
+		return delay;
+	}
+	public void setDelay(long delay) {
+		this.delay = delay;
+	}
+	private static Map<IPPort,Tunnel> tls=new ConcurrentHashMap<>();
+	private Tunnel(String name, IPPort ipport) throws UnknownHostException {
 		super();
 		this.name = name;
-		this.ipport=new IPPort(ip,port);
+		this.ipport=ipport;
 	}
-	public Tunnel(String s) throws UnknownHostException {
+	public static Tunnel newTunnel(String name, IPPort ipport) throws UnknownHostException {
+		Tunnel tmp=tls.get(ipport);
+		if(tmp!=null) {
+			return tmp;
+		}
+		tmp=new Tunnel(name, ipport);
+		tls.put(ipport, tmp);
+		return tmp;
+	}
+	public static Tunnel newTunnel(String name, String ipport) throws UnknownHostException {
+		
+		return newTunnel(name, new IPPort(ipport));
+	}
+	public static Tunnel newTunnel(String name, String ip, int port) throws UnknownHostException {
+		
+		return newTunnel(name, new IPPort(ip, port));
+	}
+	
+	public static Tunnel newTunnel(String s) throws UnknownHostException {
 		String[]t=s.split("\\$");
-			name=t[0];
+		return newTunnel(t[0], t[1]);
-			ipport=new IPPort(t[1]);
 	}
 	public String getName() {
 		return name;
 	}
-	public void setName(String name) {
+
-		this.name = name;
-	}
 	public String getIp() {
 		return ipport.getIp().getHostAddress();
 	}
@@ -59,7 +88,18 @@ public class Tunnel {
 		Tunnel other = (Tunnel) obj;
 		return Objects.equals(ipport, other.ipport) && Objects.equals(name, other.name);
 	}
-	
+	public static void states() {
+		Set<Entry<IPPort, Tunnel>>  s=tls.entrySet();
+		for (Iterator iterator = s.iterator(); iterator.hasNext();) {
+			Entry<IPPort, Tunnel> entry = (Entry<IPPort, Tunnel>) iterator.next();
+			Tunnel tll=entry.getValue();
+			System.out.println(tll.getName()+" "+((tll.getDelay()==-1)?"δ֪":tll.getDelay()/1000000+"ms")+" Á¬½ÓÊý"+tll.ati);
+		}
+	}
+	private AtomicInteger ati=new AtomicInteger(0);
+	public AtomicInteger getCCount() {
+		return ati;
+	}