KLALB2.3

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

@@ -3,6 +3,7 @@ package org.kne.cloud.network.klalb;
 import java.io.IOException;
 import java.net.BindException;
 import java.net.ConnectException;
+import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -12,10 +13,13 @@ import java.net.Socket;
 import java.net.SocketAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
+import java.time.Clock;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Enumeration;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -23,10 +27,14 @@ import java.util.Map.Entry;
 import java.util.Scanner;
 import java.util.Set;
 import java.util.Timer;
+import java.util.TimerTask;
 import java.util.UUID;
 import java.util.Vector;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiConsumer;
@@ -37,11 +45,14 @@ import javax.management.openmbean.ArrayType;
 import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
 
+import org.kne.cloud.clock.AdjustedNanoClock;
 import org.kne.cloud.network.MultipurposeSocketAddress;
 import org.kne.cloud.network.NetworkService;
 import org.kne.cloud.network.Proxy;
 import org.kne.cloud.network.SocketType;
 import org.kne.cloud.network.ThreadTool;
+import org.kne.cloud.network.monitor.MonitorData;
+import org.kne.cloud.network.monitor.SpeedAndTrafficMonitorDataImpl;
 import org.kne.debug.TimeDebugger;
 
 import javassist.ClassPool;
@@ -52,11 +63,147 @@ import javassist.bytecode.CodeAttribute;
 import javassist.bytecode.CodeIterator;
 
 public class KLALBController {
+	
+	private SpeedAndTrafficMonitorDataImpl linkMonitor=new SpeedAndTrafficMonitorDataImpl();
+	
+	private SpeedAndTrafficMonitorDataImpl datatMonitor=new SpeedAndTrafficMonitorDataImpl();
+
+	private List<MultipurposeSocketAddress>selflineTable=new ArrayList<>();
+	private Timer twk=new Timer("网卡检测扫描计时器", true);
+	{
+		
+		twk.schedule(new TimerTask() {
+			
+			@Override
+			public void run() {
+				try {
+					Enumeration<NetworkInterface>eu= NetworkInterface.getNetworkInterfaces();
+					while (eu.hasMoreElements()) {
+						NetworkInterface networkInterface = (NetworkInterface) eu.nextElement();
+						if(networkInterface.isUp()) {
+						//System.out.println(networkInterface+" "+networkInterface.isUp());
+						Enumeration<InetAddress>ei= networkInterface.getInetAddresses();
+						while (ei.hasMoreElements()) {
+							InetAddress inetAddress = (InetAddress) ei.nextElement();
+							
+							for (Iterator<MultipurposeSocketAddress> iterator = listens.iterator(); iterator.hasNext();) {
+								MultipurposeSocketAddress tcpl = (MultipurposeSocketAddress) iterator.next();
+							
+								try {
+									if(tcpl.getInetAddress().isAnyLocalAddress()||tcpl.getInetAddress().equals(inetAddress)) {
+										MultipurposeSocketAddress bind=new MultipurposeSocketAddress(tcpl.getType(),inetAddress.getHostAddress(),tcpl.getPort());
+										//System.out.println(bind);
+										synchronized (selflineTable) {
+									if(!selflineTable.contains(bind)) {
+										selflineTable.add(bind);
+									}
+}
+									}
+								} catch (UnknownHostException e) {
+									// TODO 自动生成的 catch 块
+									e.printStackTrace();
+								}
+								
+							
+							}
+						}
+						}
+					}
+				} catch (SocketException e) {
+				}
+				try {
+					List<MultipurposeSocketAddress>localaddress=new ArrayList<>();
+					Enumeration<NetworkInterface>eu= NetworkInterface.getNetworkInterfaces();
+					while (eu.hasMoreElements()) {
+						NetworkInterface networkInterface = (NetworkInterface) eu.nextElement();
+						if(networkInterface.isUp()) {
+						//System.out.println(networkInterface+" "+networkInterface.isUp());
+						Enumeration<InetAddress>ei= networkInterface.getInetAddresses();
+						while (ei.hasMoreElements()) {
+							InetAddress inetAddress = (InetAddress) ei.nextElement();
+							MultipurposeSocketAddress bind=new MultipurposeSocketAddress(inetAddress.getHostAddress(),0);
+							localaddress.add(bind);
+							
+						}
+						}
+					}
+					Set<MultipurposeSocketAddress>st=new HashSet<>();
+					for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
+						KLALBRemoteLine multipurposeSocketAddress = (KLALBRemoteLine) iterator.next();
+						if(multipurposeSocketAddress.getSocketAddress()!=null)
+						st.add(multipurposeSocketAddress.getSocketAddress());
+					}
+					
+					
+					for (Iterator<MultipurposeSocketAddress> iterator = st.iterator(); iterator.hasNext();) {
+						MultipurposeSocketAddress multipurposeSocketAddress = (MultipurposeSocketAddress) iterator
+								.next();
+						for (Iterator<MultipurposeSocketAddress> iterator2 = localaddress.iterator(); iterator2.hasNext();) {
+							MultipurposeSocketAddress bind = (MultipurposeSocketAddress) iterator2
+									.next();
+							try {
+								if(multipurposeSocketAddress.getInetAddress()instanceof Inet4Address&&bind.getInetAddress() instanceof Inet6Address) {
+									continue;
+								}
+								if(multipurposeSocketAddress.getInetAddress() instanceof Inet6Address &&bind.getInetAddress() instanceof Inet4Address) {
+									continue;
+								}
+							} catch (UnknownHostException e) {
+							}
+							if(!checkContainsTargetAndBind(multipurposeSocketAddress,bind)) {
+								//System.out.println(target+" "+bind);
+							addRemoteLine( new KLALBRemoteLine(multipurposeSocketAddress,bind));
+							}
+						}
+					
+					}
+					for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
+						KLALBRemoteLine reml = (KLALBRemoteLine) iterator
+								.next();
+						if(reml.getSocketAddress()!=null) {
+							if(reml.getBindAddress()!=null||reml.getRemoteVaddr()!=null)
+						if(!localaddress.contains(reml.getBindAddress())) {
+							reml.close();
+							lines.remove(reml);
+						}
+						}
+					}
+					
+				} catch (SocketException e) {
+				}
+							
+			}
+		}, 2000, 2000);
+	}
+	public SpeedAndTrafficMonitorDataImpl getLinkMonitor() {
+		return linkMonitor;
+	}
+
+	public SpeedAndTrafficMonitorDataImpl getDatatMonitor() {
+		return datatMonitor;
+	}
+	
+	private static ConcurrentHashMap<Inet6Address, AdjustedNanoClock>clocks=new ConcurrentHashMap<>();
+	
+	protected AdjustedNanoClock getAdjustedClockByVaddr(Inet6Address vaddr) {
+		AdjustedNanoClock clk=clocks.get(vaddr);
+		if(clk==null) {
+			clk=new AdjustedNanoClock();
+			clocks.put(vaddr, clk);
+		}
+		return clk;
+	}
+
+	
+	
+	
+
 	private SocketType socketType= new KLALBSocketType();
+	
 	public class KLALBSocketType extends SocketType{
 
 		public KLALBSocketType() {
-			super(new KLALBVirtualSocketFactory(KLALBController.this), new KLALBVirtualServerSocketFactory(KLALBController.this));
+			super(new KLALBVirtualSocketFactory(KLALBController.this), new KLALBVirtualServerSocketFactory(KLALBController.this),new KLALBVirtualSocketChannelFactory(KLALBController.this), new KLALBVirtualServerSocketChannelFactory(KLALBController.this));
 		}
 		
 	}
@@ -64,23 +211,30 @@ public class KLALBController {
 		return socketType;
 	}
 
-	private Supplier<String> selflineTableSupplier=()->{return null;};
+	
+	
+	//private Supplier<String> selflineTableSupplier=()->{return null;};
 	
 
-	public Supplier<String> getSelflineTableSupplier() {
+	/*public Supplier<String> getSelflineTableSupplier() {
 		return selflineTableSupplier;
 	}
 
 	public void setSelflineTableSupplier(Supplier<String> selflineTableSupplier) {
 		this.selflineTableSupplier = selflineTableSupplier;
+	}*/
+
+	public List<MultipurposeSocketAddress> getSelflineTable() {
+		return selflineTable;
 	}
 
+
 	private Inet6Address self;
 
 	public Inet6Address getSelf() {
 		return self;
 	}
-	private List<KLALBRemoteLine> lines=new ArrayList<>();
+	private List<KLALBRemoteLine> lines=new CopyOnWriteArrayList<>();
 	//private ReadWriteLock lineslock=new ReentrantReadWriteLock();
 	
 	public List<KLALBRemoteLine> getLines() {
@@ -95,12 +249,10 @@ public class KLALBController {
 
 
 	public void reconnectImmediately() {
-		synchronized (lines) {
 			for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
 				KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
 				klalbRemoteLine.reconnectImmediately();
 			}
-		}
 	}
 	
 	private PacketReceiver prc=new PacketReceiver();
@@ -113,21 +265,26 @@ public class KLALBController {
 					PortPacket pt=(PortPacket) rec;
 					if(!streamPortBinder.distributePacketToConsumer(krs, pt)) {
 						if(!(pt instanceof RSTPacket))
-						sendPacketToAddress(krs.getRemoteVaddr(), new RSTPacket(pt.getDport(), pt.getSport()),
-								0,2);
+						sendPacketToAddress(krs.getRemoteVaddr(), new RSTPacket(pt.getDport(), pt.getSport()),2);
 					}
 				}else {
 				switch (rec.getType()) {
 				case KLALBPacket.ADDLINES:
 					ADDLINESPacket lpt=(ADDLINESPacket) rec;
 					String s=lpt.getLines();
+					ThreadTool.makeVDaemonThreadIfSupport("线路添加任务", ()->{
 					Scanner scn=new Scanner(s);
 					while(scn.hasNext()) {
 						String sn=scn.nextLine();
 						MultipurposeSocketAddress msa=	new MultipurposeSocketAddress(sn);
-						addRemoteLines(msa);
+						try {
+							addRemoteLines(msa);
+						} catch (SocketTimeoutException e) {
+						} catch (SocketException e) {
+						}
 						
 					}
+					}).start();
 					break;
 				default:
 					break;
@@ -142,7 +299,6 @@ public class KLALBController {
 		
 	}
 	public void addRemoteLines(MultipurposeSocketAddress target) throws SocketTimeoutException, SocketException {
-		synchronized (lines) {
 			
 				try {
 					Enumeration<NetworkInterface>eu= NetworkInterface.getNetworkInterfaces();
@@ -154,6 +310,15 @@ public class KLALBController {
 						while (ei.hasMoreElements()) {
 							InetAddress inetAddress = (InetAddress) ei.nextElement();
 							MultipurposeSocketAddress bind=new MultipurposeSocketAddress(inetAddress.getHostAddress(),0);
+							try {
+								if(target.getInetAddress()instanceof Inet4Address&&bind.getInetAddress() instanceof Inet6Address) {
+									continue;
+								}
+								if(target.getInetAddress() instanceof Inet6Address &&bind.getInetAddress() instanceof Inet4Address) {
+									continue;
+								}
+							} catch (UnknownHostException e) {
+							}
 							if(!checkContainsTargetAndBind(target,bind)) {
 								//System.out.println(target+" "+bind);
 							addRemoteLine( new KLALBRemoteLine(target,bind));
@@ -164,21 +329,21 @@ public class KLALBController {
 				} catch (SocketException e) {
 					if(!checkContainsTarget(target))
 					addRemoteLine( new KLALBRemoteLine(target));
-					throw e;
+					//throw e;
 				}
 			
-		}
+		
 	}
 	public Inet6Address getRemoteVaddrBySocketAddress(MultipurposeSocketAddress target) throws SocketTimeoutException {
 		KLALBRemoteLine kr=null;
-		synchronized(lines) {
-		for (Iterator iterator = lines.iterator(); iterator.hasNext();) {
+		for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
 			KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
-			if(target.equals(klalbRemoteLine.getSocketAddress())) {
+			if(target.equals(klalbRemoteLine.getSocketAddress())&&klalbRemoteLine.getBindAddress()==null) {
 				kr=klalbRemoteLine;
+				break;
 			}
 		}
-		}
+		
 		if(kr==null) {
 		kr=new KLALBRemoteLine(target);
 	addRemoteLine(kr);
@@ -213,19 +378,28 @@ public class KLALBController {
 		return b;
 	}
 
-	public void addRemoteLine(KLALBRemoteLine krs) throws SocketTimeoutException {
+	public void addRemoteLine(KLALBRemoteLine krs)  {
 		krs.setPacketReceiver(prc);
+		krs.setKlalbController(this);
 		krs.setLocalVaddrSupplier(()->{return self;});
 		krs.startIO();
-		String selflineTable=selflineTableSupplier.get();
-		if(selflineTable!=null)
-		krs.sendPacket(new ADDLINESPacket(selflineTable), 0);
-		synchronized (lines) {
+		String selflineTable=generateSelfLineTable();
+		if(selflineTable!=null&&!selflineTable.equals(""))
+		krs.sendPacket(new ADDLINESPacket(selflineTable));
 		lines.add(krs);
-		}
 	}
 
 
+	private String generateSelfLineTable() {
+		StringBuilder sbd=new StringBuilder();
+		for (Iterator<MultipurposeSocketAddress> iterator = selflineTable.iterator(); iterator.hasNext();) {
+			MultipurposeSocketAddress klalbRemoteLine = (MultipurposeSocketAddress) iterator.next();
+			sbd.append(klalbRemoteLine.toString());
+			sbd.append('\n');
+		}
+		return sbd.toString();
+	}
+
 	public KLALBController(Inet6Address self) {
 		this.self = self;
 	}
@@ -238,16 +412,51 @@ public class KLALBController {
 		return new KLALBVirtualSocketImpl(this);
 	}
 
+
+	
 	
 
-	protected void sendPacketToAddress(Inet6Address addr, KLALBPacket syntPacket, int priority)
+	protected void sendPacketToAddress(Inet6Address addr, KLALBPacket packet)
 			throws IOException {
-		sendPacketToAddress(addr, syntPacket, priority, 1);
+		sendPacketToAddress(addr, packet, 1);
+	}
+
+private Timer trtr=new Timer("路由表刷新计时器", true);
+{
+	trtr.scheduleAtFixedRate(new TimerTask() {
+		
+		@Override
+		public void run() {
+			Map<Inet6Address,List<KLALBRemoteLine>> lines2 =new ConcurrentHashMap<>();
+			for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
+				KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+				if(klalbRemoteLine.isClosed()) {
+					lines.remove(klalbRemoteLine);
+				}else {
+					if(klalbRemoteLine.getRemoteVaddr()!=null&&klalbRemoteLine.getMonitor().getState()==MonitorData.ONLINE) {
+						if(lines2.containsKey(klalbRemoteLine.getRemoteVaddr())) {
+							lines2.get(klalbRemoteLine.getRemoteVaddr()).add(klalbRemoteLine);
+						}else {
+							ArrayList<KLALBRemoteLine>al1=new ArrayList<>();
+							al1.add(klalbRemoteLine);
+							lines2.put(klalbRemoteLine.getRemoteVaddr(), al1);
+						}
+					}
+				}
+			}
+			for (Iterator<Entry<Inet6Address, List<KLALBRemoteLine>>> iterator = lines2.entrySet().iterator(); iterator.hasNext();) {
+				Entry<Inet6Address, List<KLALBRemoteLine>> klalbRemoteLine = (Entry<Inet6Address, List<KLALBRemoteLine>>) iterator.next();
+				klalbRemoteLine.getValue().forEach((r)->{
+					r.runPredict();
+				});
+				Collections.sort(klalbRemoteLine.getValue());
+			}
+			KLALBController.this.lines2=lines2;
+		}
+	}, 5, 5);
 	}
-	
 			private void updateLines2(Inet6Address addr) throws SocketTimeoutException {
 				List<KLALBRemoteLine> l=new ArrayList();
-				synchronized (lines) {
 				for (int i = 0; i < lines.size(); i++) {
 					KLALBRemoteLine klalbRemoteLine = lines.get(i);
 					if(klalbRemoteLine.isClosed()) {
@@ -255,103 +464,120 @@ public class KLALBController {
 						i--;
 						continue;
 					}
-					if(addr.equals(klalbRemoteLine.getRemoteVaddr())&&klalbRemoteLine.getMonitor().getState()==Monitor.ONLINE) {
+					if(addr.equals(klalbRemoteLine.getRemoteVaddr())&&klalbRemoteLine.getMonitor().getState()==MonitorData.ONLINE) {
 						l.add(klalbRemoteLine);
 					}
 				}
-		}
+				
 				if(l.isEmpty()) {
 					lines2.remove(addr);
 				}else {
+					l.forEach((r)->{
+						r.runPredict();
+					});
+					Collections.sort(l);
 				lines2.put(addr, l);
 				}
 			}
 			private	Map<Inet6Address,List<KLALBRemoteLine>> lines2 = new ConcurrentHashMap<>();
-			private volatile long itm=System.nanoTime();
-	protected void sendPacketToAddress(Inet6Address addr, KLALBPacket packet, int priority, int count)
+			
+			
+	protected void sendPacketToAddress(Inet6Address addr, KLALBPacket packet, int count)
 			throws IOException {
-		/*if(packet instanceof RSTPacket) {
-			new Exception("-RST-").printStackTrace();
-		}*/
+		if(packet==null)
+			throw new NullPointerException("packet is null!");
 		//TimeDebugger tdb=new TimeDebugger();
-		//tdb.putTime("start");
 		
+		
+		//tdb.putTime("start");
+		packet.genseq();
+		loop:while(true) {
+			if(packet.isDisposed())
+				return;
 		List<KLALBRemoteLine> lines2x;
-			//loop:while(true) {
-				long cur=System.nanoTime();
-			if(cur-itm>10000000L) {
-				itm=cur;
-				lines2.clear();
-			}
+				
 				lines2x=lines2.get(addr);
-				if (lines2x == null || lines2x.isEmpty()) {
-					updateLines2(addr);
-					lines2x=lines2.get(addr);
-				}
+				
 				if (lines2x == null || lines2x.isEmpty()) {
 					throw new NoRouteToHostException("address unreachable: " + addr);
 				}
-			//tdb.putTime("selectLines");
-			/*	for (Iterator<KLALBRemoteLine> iterator = lines2x.iterator(); iterator.hasNext();) {
-					KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
-					if(klalbRemoteLine.statLengthBefore(priority)<=65536*10) {
-						break loop;
-					}
-				}
-				try {
-					//System.out.println("slp");
-					Thread.sleep(1);
-				} catch (InterruptedException e) {
-					e.printStackTrace();
-				}
-			}*/
-			List<KLALBRemoteLine> l2 = (List<KLALBRemoteLine>) ((ArrayList<KLALBRemoteLine>) lines2x).clone();
-			l2.removeAll(packet.getSendRecord());
-			if(l2.isEmpty()) {
-				l2 = (List<KLALBRemoteLine>) ((ArrayList<KLALBRemoteLine>) lines2x).clone();
-			}
-
-			//tdb.putTime("findAvaliable");
 			
-			int count0 = Math.min(count, l2.size());
-			l2.forEach((r)->{
-				r.runPredict(packet,priority);
-			});
-			//Collections.shuffle(l2);
-			Collections.sort(l2);
-			//tdb.putTime("makeDecision");
-			//System.out.println(l2);
-			for (int i = 0; i < l2.size(); i++) {
-				KLALBRemoteLine krst =l2.get(i);
-				krst.sendPacket(packet, priority);
+			int count0 = Math.min(count, lines2x.size());
+			
+			for (int i = 0; i < lines2x.size(); i++) {
+				KLALBRemoteLine krst =lines2x.get(i);
+				if(krst.getMonitor().getState()==MonitorData.ONLINE)
+				if(!packet.getSendRecord().contains(krst)) {
+					if(krst.getQueue().size()<3) {
 				packet.getSendRecord().add(krst);
+				krst.sendPacket(packet);
 				count0--;
 				if (count0 <= 0)
-					break;
+					break loop;
+					}
+				}
 			}
-			//tdb.putTime("sendPacket");
-			//tdb.print();
-			/*if(packet instanceof RSTPacket)
-			new Exception().printStackTrace();*/
-	}
-	protected void removeFromSend(Inet6Address addr,KLALBPacket klalbPacket) {
-		synchronized (lines) {
-			for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
-				KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
-				klalbRemoteLine.remoeFromSendQueue(klalbPacket);
+			for (int i = 0; i < lines2x.size(); i++) {
+				KLALBRemoteLine krst =lines2x.get(i);
+				if(krst.getMonitor().getState()==MonitorData.ONLINE)
+				if(packet.getSendRecord().contains(krst)) {
+					if(krst.getQueue().size()<3) {
+				packet.getSendRecord().add(krst);
+				krst.sendPacket(packet);
+				count0--;
+				if (count0 <= 0)
+					break loop;
+					}
+				}
 			}
+			LockSupport.parkNanos(50000);
+			//tdb.putTime("sendfailed");
+			}
+		//tdb.putTime("sendsuccess");
+		//tdb.print();
+
+	/*	List<KLALBRemoteLine> lines2x;
+		lines2x=lines2.get(addr);
+		
+		if (lines2x == null || lines2x.isEmpty()) {
+			throw new NoRouteToHostException("address unreachable: " + addr);
 		}
+		KLALBRemoteLine kr= lines2x.get(0);
+		kr.sendPacket(packet);
+		packet.getSendRecord().add(kr);*/
 	}
-private Timer t=new Timer("数据包发送计时器", true);
-	public Timer getTimer() {
+	/*protected void removeFromSend(Inet6Address addr,KLALBPacket klalbPacket) {
+		for (Iterator iterator = lines.iterator(); iterator.hasNext();) {
+			KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+				klalbRemoteLine.remoeFromSendQueue(klalbPacket);
+			
+		}
+			
+	}*/
+private Timer t=new Timer("数据包重传计时器", true);
+	public Timer getResendTimer() {
 		return t;
 	}
+private Timer t2=new Timer("数据包粘包计时器", true);
+
+private List<MultipurposeSocketAddress> listens=new CopyOnWriteArrayList<>();
+
+	public Timer getNagleTimer() {
+		return t2;
+	}
+	
 	
 	
 	public void registerToProxyTypeAs(String proxyname) {
 		MultipurposeSocketAddress.getSocketTypeRegister().put(proxyname+"_Stream",socketType);
 		//ProxyProfileEntry.getRegister().put(proxyname, this);
 	}
+	
+
+	public List<MultipurposeSocketAddress> getListenSocketAddress() {
+		return listens;
+	}
+
 
 
 	/*@Override