KLALB大更新

新负载均衡算法

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

@@ -2,7 +2,10 @@ package org.kne.cloud.network.klalb;
 
 import java.io.IOException;
 import java.net.Inet6Address;
+import java.net.NoRouteToHostException;
 import java.net.Socket;
+import java.net.SocketException;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
@@ -12,12 +15,35 @@ import org.kne.cloud.network.mport.ThreadTool;
 public class KLALBRemoteSocket extends MonitoredSocket {
 	private Inet6Address remoteVaddr;
 
+	private KLALBController controller;
+	
+	public KLALBController getController() {
+		return controller;
+	}
+	protected void setController(KLALBController controller) {
+		this.controller = controller;
+	}
 	public Inet6Address getRemoteVaddr() {
 		return remoteVaddr;
 	}
-
-	public KLALBRemoteSocket(Socket socket) {
-		super(socket);
+	public String toString() {
+		return super.toString()+ getMonitor().toString();
+	}
+public KLALBRemoteSocket(Socket socket) {
+	this(socket,new Monitor());
+}
+	public KLALBRemoteSocket(Socket socket,Monitor m) {
+		super(socket,m);
+		try {
+			socket.setSendBufferSize(32768);
+			socket.setReceiveBufferSize(65535);
+			//System.out.println(socket.getSendBufferSize()+" "+socket.getReceiveBufferSize());
+			socket.setTcpNoDelay(true);
+			socket.setSoTimeout(10000);
+		} catch (SocketException e1) {
+			// TODO 自动生成的 catch 块
+			e1.printStackTrace();
+		}
 		ThreadTool.makeVDaemonThreadIfSupport("远程接收线程", () -> {
 			try {
 
@@ -25,11 +51,18 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 					KLALBPacket kpp = getInputStream().readPacket();
 					//System.out.println("RX:" + kpp);
 						if(kpp instanceof PINGPacket) {
-						sendPacket(new PONGPacket(), 65540);
+						sendPacket(new PONGPacket(((PINGPacket) kpp).getTime()), 65540);
 					}else if(kpp instanceof PONGPacket) {
-						latency.set(System.nanoTime()-pingstarttime);
-						pinging=false;
+						PONGPacket png=(PONGPacket) kpp;
+						getMonitor().updateLatency (System.nanoTime()- png.getTime());
+						try {
+							socket.setSoTimeout(1100);
+						} catch (SocketException e1) {
+							// TODO 自动生成的 catch 块
+							e1.printStackTrace();
+						}
 					}else {
+					
 						if(kpp instanceof VADDRPacket) {
 							remoteVaddr=((VADDRPacket) kpp).getVaddr();
 						}
@@ -40,7 +73,7 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 					}
 				}
 			} catch (IOException e) {
-				e.printStackTrace();
+				//e.printStackTrace();
 			} catch (InterruptedException e) {
 				e.printStackTrace();
 			} finally {
@@ -55,18 +88,18 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 			try {
 
 				while (!isClosed()) {
-					if(System.nanoTime()-pingstarttime>100000000000L) {
-						pinging=false;
-					}
-					if(checkPingTime()&&!pinging) {
-						getOutputStream().writePacket(new PINGPacket());
-						pinging =true;
-						pingstarttime=System.nanoTime();
+					if(checkPingTime()) {
+						getOutputStream().writePacket(new PINGPacket(System.nanoTime()));
 					}else {
 					PQItem pqi=sendQueue.poll();
 					if(pqi!=null) {
 					KLALBPacket kpp=pqi.getPacket();
+					//if(!(kpp instanceof PONGPacket))
 					//System.out.println("TX:" + kpp);
+					if(kpp instanceof PONGPacket) {
+						PONGPacket pp=(PONGPacket) kpp;
+						pp.redeltaTime(System.nanoTime()-pqi.getAddtime());
+					}
 					getOutputStream().writePacket(kpp);
 					}else {
 							Thread.sleep(1);
@@ -74,7 +107,7 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 					}
 				}
 			} catch (IOException e) {
-				e.printStackTrace();
+			//	e.printStackTrace();
 			} catch (InterruptedException e) {
 			} finally {
 				try {
@@ -87,11 +120,9 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 
 	}
 	
-	private volatile long pingstarttime=System.nanoTime(); 
 
-	private volatile boolean pinging=false;
 	private volatile long time = System.nanoTime();
-	private volatile long pingInterval=1000000000L;
+	private volatile long pingInterval=500000000L;
 
 	public long getPingInterval() {
 		return pingInterval;
@@ -112,11 +143,8 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 	}
 	
 
-	private volatile boolean waitingforPing = false;
 
-	public long getLatency() {
-		return latency.get();
-	}
+
 
 	private KLALBInputStream kis;
 	private KLALBOutputStream kos;
@@ -135,7 +163,7 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 		return kos;
 	}
 
-	private AtomicLong latency = new AtomicLong();
+	
 	private volatile Consumer<KLALBPacket> rec;
 	private Consumer<KLALBRemoteSocket> clo;
 	private PriorityBlockingQueue<PQItem> sendQueue=new PriorityBlockingQueue<PQItem>();
@@ -160,9 +188,22 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 	}
 	@Override
 	public synchronized void close() throws IOException {
-		super.close();
 		if(!isClosed()&&clo!=null)
 			clo.accept(this);
+		cdl.countDown();
+		super.close();
+		sendQueue.forEach((x)->{
+			try {
+				controller.sendPacketToAddress(remoteVaddr, x.getPacket(), x.getPriority()+1);
+			} catch (NoRouteToHostException e) {
+			}
+		});
+		
+	}
+
+	@Override
+	public boolean isClosed() {
+		return cdl.getCount()<=0;
 	}
 
 	private AtomicLong al=new AtomicLong(0);
@@ -170,6 +211,10 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 		private KLALBPacket packet;
 		private long index=al.getAndIncrement();
 		private int priority=5;
+		private long addtime=System.nanoTime();
+		public long getAddtime() {
+			return addtime;
+		}
 		public PQItem(KLALBPacket value, int priority) {
 			super();
 			this.packet = value;
@@ -201,4 +246,17 @@ public class KLALBRemoteSocket extends MonitoredSocket {
 			}
 		}
 	}
+	private CountDownLatch cdl=new CountDownLatch(1);
+	public void waitforlose() {
+		try {
+			cdl.await();
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+		}
+	}
+	public void remoeFromSendQueue(KLALBPacket klalbPacket) {
+		sendQueue.removeIf((x)->{
+			return x.getPacket().equals(klalbPacket);
+		});
+	}
 }