forked from KNEMC/KLALB
UPDATE 2.2
This commit is contained in:
Administrator
@@ -0,0 +1,502 @@
|
||||
package org.kne.cloud.network.klalb;
|
||||
|
||||
import java.io.FileNotFoundException;
|
||||
import java.io.IOException;
|
||||
import java.io.PrintStream;
|
||||
import java.io.PrintWriter;
|
||||
import java.net.BindException;
|
||||
import java.net.Inet6Address;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketTimeoutException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import java.util.Vector;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.concurrent.locks.LockSupport;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.kne.cloud.network.MultipurposeSocketAddress;
|
||||
import org.kne.cloud.network.ThreadTool;
|
||||
|
||||
public class KLALBRemoteLine implements Comparable<KLALBRemoteLine>{
|
||||
|
||||
|
||||
private volatile boolean closed=false;
|
||||
|
||||
private volatile Supplier<Inet6Address> localVaddrSupplier;
|
||||
|
||||
public Supplier<Inet6Address> getLocalVaddrSupplier() {
|
||||
return localVaddrSupplier;
|
||||
}
|
||||
|
||||
public void setLocalVaddrSupplier(Supplier<Inet6Address> localVaddrSupplier) {
|
||||
this.localVaddrSupplier = localVaddrSupplier;
|
||||
}
|
||||
|
||||
|
||||
private volatile Inet6Address remoteVaddr;
|
||||
public Inet6Address getRemoteVaddr() throws SocketTimeoutException {
|
||||
return remoteVaddr;
|
||||
}
|
||||
|
||||
public void waitForRemoteVaddrAvaliable(long timeout) throws SocketTimeoutException {
|
||||
long s=System.nanoTime();
|
||||
while(remoteVaddr==null) {
|
||||
|
||||
try {
|
||||
Thread.sleep(1);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
if(System.nanoTime()-s>timeout*1000000)
|
||||
throw new SocketTimeoutException();
|
||||
}
|
||||
}
|
||||
|
||||
public Monitor getMonitor() {
|
||||
return monitor;
|
||||
}
|
||||
public String toString() {
|
||||
return remoteVaddr+"\t"+monitor.toString()+"\t"+sumNextPacketCount();
|
||||
}
|
||||
|
||||
public String toString2() {
|
||||
return remoteVaddr+"\t"+monitor.toString2();
|
||||
}
|
||||
private Monitor monitor;
|
||||
private volatile KLALBPacketLink kplink;
|
||||
private MultipurposeSocketAddress bindAddress;
|
||||
private MultipurposeSocketAddress socketAddress;
|
||||
|
||||
private Thread lthd;
|
||||
|
||||
|
||||
public void setKplink(KLALBPacketLink kplink) {
|
||||
this.kplink = kplink;
|
||||
}
|
||||
|
||||
public KLALBPacketLink getKplink() {
|
||||
return kplink;
|
||||
}
|
||||
|
||||
|
||||
public MultipurposeSocketAddress getSocketAddress() {
|
||||
return socketAddress;
|
||||
}
|
||||
|
||||
|
||||
|
||||
public MultipurposeSocketAddress getBindAddress() {
|
||||
return bindAddress;
|
||||
}
|
||||
|
||||
protected static KLALBPacketLink createLink(MultipurposeSocketAddress bindAddress,MultipurposeSocketAddress mpa) throws IOException {
|
||||
if(mpa.isStream()) {
|
||||
if(bindAddress!=null) {
|
||||
return new StreamKLALBPacketLink(mpa.connectSocket(InetAddress.getByName( bindAddress.getHost()),bindAddress.getPort()));
|
||||
}else {
|
||||
return new StreamKLALBPacketLink(mpa.connectSocket());
|
||||
}
|
||||
}
|
||||
else {
|
||||
if(bindAddress!=null) {
|
||||
return new DatagramKLALBPacketLink(mpa.connectDatagramSocket(InetAddress.getByName( bindAddress.getHost()),bindAddress.getPort()));
|
||||
}else {
|
||||
return new DatagramKLALBPacketLink(mpa.connectDatagramSocket());
|
||||
}
|
||||
}
|
||||
}
|
||||
public KLALBRemoteLine(MultipurposeSocketAddress mpa) {
|
||||
this(mpa,new Monitor());
|
||||
}
|
||||
public KLALBRemoteLine(MultipurposeSocketAddress mpa,Monitor monitor) {
|
||||
this(mpa,null,monitor);
|
||||
}
|
||||
public KLALBRemoteLine(KLALBPacketLink kpl) {
|
||||
this(kpl,new Monitor());
|
||||
}
|
||||
public KLALBRemoteLine(KLALBPacketLink kpl,Monitor monitor) {
|
||||
this.kplink=kpl;
|
||||
this.monitor=monitor;
|
||||
monitor.setName(kpl.toString());
|
||||
}
|
||||
|
||||
public KLALBRemoteLine(MultipurposeSocketAddress mpa,MultipurposeSocketAddress bindaddr,Monitor monitor ) {
|
||||
this.socketAddress=mpa;
|
||||
this.bindAddress=bindaddr;
|
||||
this.monitor=monitor;
|
||||
monitor.setName(mpa.toString());
|
||||
}
|
||||
|
||||
public KLALBRemoteLine(MultipurposeSocketAddress mpa, MultipurposeSocketAddress bindaddr) {
|
||||
this(mpa,bindaddr,new Monitor());
|
||||
}
|
||||
PrintStream pw;
|
||||
private void startRecord() {
|
||||
try {
|
||||
pw=new PrintStream(UUID.randomUUID()+".csv");
|
||||
|
||||
pw.println("时间,状态,上传速度,下载速度,上传延迟,下载延迟,往返延迟,抖动");
|
||||
ThreadTool.makePDaemonThreadIfSupport("BigData Record", ()->{
|
||||
while(true) {
|
||||
try {
|
||||
Thread.sleep(1000);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
pw.println(System.currentTimeMillis()+","+monitor.getState()+","+monitor.getOutSpeedAvg()+","+monitor.getInSpeedAvg()+","+sndDelayFactor+","+rcvDelayFactor+","+monitor.getLatencyAvg()+","+monitor.getJitter());
|
||||
}
|
||||
}).start();
|
||||
} catch (FileNotFoundException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
private void endRecord() {
|
||||
if(pw!=null)
|
||||
pw.close();
|
||||
}
|
||||
|
||||
protected void startIO() {
|
||||
ThreadTool.makePDaemonThreadIfSupport("远程接收线程", () -> {
|
||||
//startRecord();
|
||||
while(true) {
|
||||
try {
|
||||
|
||||
monitor.setState(Monitor.CONNECTING);
|
||||
if(socketAddress==null) {
|
||||
if(kplink.isClosed()) {
|
||||
monitor.setState(Monitor.OFFLINE);
|
||||
closed=true;
|
||||
break;
|
||||
}
|
||||
}else {
|
||||
try {
|
||||
kplink=createLink(bindAddress,socketAddress);
|
||||
}catch(BindException be){
|
||||
//be.printStackTrace();
|
||||
close();
|
||||
throw be;
|
||||
}
|
||||
}
|
||||
kplink.setSoTimeout(10000);
|
||||
Thread t=ThreadTool.makePDaemonThreadIfSupport("远程发送线程", () -> {
|
||||
try {
|
||||
writePacketToKPL(new VADDRREQPacket());
|
||||
writePacketToKPL(new VADDRREQPacket());
|
||||
writePacketToKPL(new VADDRREQPacket());
|
||||
while ((!kplink.isClosed())&&(!closed)) {
|
||||
if(checkPingTime()) {
|
||||
writePacketToKPL(new PINGPacket(System.nanoTime()));
|
||||
}else {
|
||||
KLALBPacket kpp=getNextPacket();
|
||||
long lth=statLengthBefore(10);
|
||||
if(lth>65536*2) {
|
||||
monitor.updateOutSpeedMax();
|
||||
}
|
||||
if(kpp!=null) {
|
||||
pingInterval= 100000000L;
|
||||
writePacketToKPL(kpp);
|
||||
}else {
|
||||
pingInterval=1000000000L;
|
||||
tlock=Thread.currentThread();
|
||||
LockSupport.parkNanos(1000000);
|
||||
}
|
||||
}
|
||||
Thread.yield();
|
||||
}
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
try {
|
||||
kplink.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
});
|
||||
t.start();
|
||||
boolean fst=true;
|
||||
long stime=5000000000L;
|
||||
while ((!kplink.isClosed())&&(!closed)) {
|
||||
long readStart=System.nanoTime();
|
||||
KLALBPacket kpp = readPacketFromKPL();
|
||||
long readTime=System.nanoTime()-readStart;
|
||||
if(readTime>stime) {
|
||||
stime=readTime;
|
||||
}else {
|
||||
stime=(stime*99+readTime)/100;
|
||||
}
|
||||
if(stime<3000000000L) {
|
||||
stime=3000000000L;
|
||||
}
|
||||
|
||||
//System.out.println(readTime);
|
||||
kplink.setSoTimeout((int) (stime/1000000));
|
||||
if(kpp==null)
|
||||
break;
|
||||
switch (kpp.getType()) {
|
||||
case KLALBPacket.PING:
|
||||
sendPacket(new PONGPacket(((PINGPacket) kpp).getTime(),kpp.getRcvtime(),System.nanoTime()), -1);
|
||||
break;
|
||||
case KLALBPacket.PONG:
|
||||
PONGPacket png=(PONGPacket) kpp;
|
||||
long ul=png.getRcvtime()-( png.getTimepingsnd()+(png.getTimepongsnd()-png.getTimepingrcv()));
|
||||
//System.out.println(toString()+"\t"+ul);
|
||||
monitor.updateLatency (ul);
|
||||
long DsndDelayFactor=png.getTimepingrcv()- png.getTimepingsnd();
|
||||
long DrcvDelayFactor=png.getRcvtime()- png.getTimepongsnd();
|
||||
//sndDelayFactor=DsndDelayFactor;
|
||||
//rcvDelayFactor=DrcvDelayFactor;
|
||||
if(sndDelayFactor==Long.MIN_VALUE) {
|
||||
sndDelayFactor=DsndDelayFactor;
|
||||
}else {
|
||||
sndDelayFactor=(sndDelayFactor+ DsndDelayFactor)/2;
|
||||
}
|
||||
if(rcvDelayFactor==Long.MIN_VALUE) {
|
||||
rcvDelayFactor=DrcvDelayFactor;
|
||||
}else {
|
||||
rcvDelayFactor=(rcvDelayFactor+ DrcvDelayFactor)/2;
|
||||
}
|
||||
LockSupport.unpark(tlock);
|
||||
break;
|
||||
case KLALBPacket.VADDRREQ:
|
||||
sendPacket(new VADDRPacket(localVaddrSupplier.get()), -1);
|
||||
break;
|
||||
case KLALBPacket.VADDR:
|
||||
remoteVaddr=((VADDRPacket) kpp).getVaddr();
|
||||
sendPacket(new VADDRACKPacket(), -1);
|
||||
break;
|
||||
case KLALBPacket.VADDRACK:
|
||||
if(fst) {
|
||||
monitor.resetCoolingTime();
|
||||
monitor.setState(Monitor.ONLINE);
|
||||
fst=false;
|
||||
}
|
||||
break;
|
||||
case KLALBPacket.TEST:
|
||||
break;
|
||||
default:
|
||||
while(rec==null) {
|
||||
Thread.sleep(1);
|
||||
}
|
||||
rec.accept(KLALBRemoteLine.this,kpp);
|
||||
break;
|
||||
}
|
||||
Thread.yield();
|
||||
}
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
monitor.setState(Monitor.OFFLINE);
|
||||
try {
|
||||
if(kplink!=null)
|
||||
kplink.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
if(closed) {
|
||||
break;
|
||||
}
|
||||
//Thread.sleep(monitor.getCoolingTime());
|
||||
lthd=Thread.currentThread();
|
||||
LockSupport.parkNanos(monitor.getCoolingTime()*1000000L);
|
||||
monitor.incCoolingTime();
|
||||
|
||||
}
|
||||
endRecord();
|
||||
|
||||
}).start();
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
private KLALBPacket readPacketFromKPL() throws IOException {
|
||||
KLALBPacket packet=kplink.readPacket();
|
||||
/*if(!(packet instanceof PINGPacket) )
|
||||
if(!(packet instanceof PONGPacket) )
|
||||
System.out.println("RX:" + packet);*/
|
||||
if(packet!=null)
|
||||
monitor.getInTrafficAL().addAndGet(packet.getLength());
|
||||
return packet;
|
||||
}
|
||||
private void writePacketToKPL(KLALBPacket packet) throws IOException {
|
||||
monitor.getOutTrafficAL().addAndGet(packet.getLength());
|
||||
kplink.writePacket(packet);
|
||||
|
||||
/*if(!(packet instanceof PINGPacket) )
|
||||
if(!(packet instanceof PONGPacket) )
|
||||
System.err.println("TX:" + packet);*/
|
||||
}
|
||||
|
||||
public void close() {
|
||||
closed=true;
|
||||
try {
|
||||
if(kplink!=null)
|
||||
kplink.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
public boolean isClosed() {
|
||||
return closed;
|
||||
}
|
||||
|
||||
private volatile long time = System.nanoTime();
|
||||
private volatile long pingInterval=1000000000L;
|
||||
|
||||
private boolean checkPingTime() {
|
||||
long cu = System.nanoTime();
|
||||
if (cu - time > pingInterval) {
|
||||
time = cu;
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
private volatile BiConsumer<KLALBRemoteLine, KLALBPacket> rec;
|
||||
|
||||
//private Object lock=new Object();
|
||||
private volatile Thread tlock;
|
||||
private List<SumQueue<KLALBPacket>> sendDequeList =new ArrayList<SumQueue<KLALBPacket>>();
|
||||
{
|
||||
for(int i=0;i<12;i++) {
|
||||
sendDequeList.add(new SumQueue<KLALBPacket>(new ConcurrentLinkedQueue<>()));
|
||||
}
|
||||
}
|
||||
private KLALBPacket getNextPacket() {
|
||||
for (Iterator<SumQueue<KLALBPacket>> iterator = sendDequeList.iterator(); iterator.hasNext();) {
|
||||
SumQueue<KLALBPacket> queue = (SumQueue<KLALBPacket>) iterator.next();
|
||||
KLALBPacket v=queue.poll();
|
||||
if(v!=null)
|
||||
return v;
|
||||
if(monitor.getLatencyCurr()>(monitor.getLatencyMin()<<3)) {
|
||||
//System.out.println(monitor.getLatencyCurr()+" "+monitor.getLatencyMin());
|
||||
return null;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
private long sumNextPacketSize() {
|
||||
long i=0;
|
||||
for (Iterator<SumQueue<KLALBPacket>> iterator = sendDequeList.iterator(); iterator.hasNext();) {
|
||||
SumQueue<KLALBPacket> queue = (SumQueue<KLALBPacket>) iterator.next();
|
||||
i+=queue.getSumValue();
|
||||
}
|
||||
return i;
|
||||
}
|
||||
private long sumNextPacketCount() {
|
||||
long i=0;
|
||||
for (Iterator<SumQueue<KLALBPacket>> iterator = sendDequeList.iterator(); iterator.hasNext();) {
|
||||
SumQueue<KLALBPacket> queue = (SumQueue<KLALBPacket>) iterator.next();
|
||||
i+=queue.size();
|
||||
}
|
||||
return i;
|
||||
}
|
||||
|
||||
public void sendPacket(KLALBPacket kp) {
|
||||
sendPacket(kp, 5);
|
||||
}
|
||||
|
||||
public void sendPacket(KLALBPacket blk,int prio) {
|
||||
sendDequeList.get(prio+1).add(blk);
|
||||
LockSupport.unpark(tlock);
|
||||
}
|
||||
|
||||
public void setPacketReceiver(BiConsumer<KLALBRemoteLine,KLALBPacket> rec) {
|
||||
this.rec = rec;
|
||||
}
|
||||
|
||||
|
||||
|
||||
public void remoeFromSendQueue(KLALBPacket klalbPacket) {
|
||||
for (Iterator<SumQueue<KLALBPacket>> iterator = sendDequeList.iterator(); iterator.hasNext();) {
|
||||
SumQueue<KLALBPacket> queue = iterator.next();
|
||||
queue.removeIf((x)->{
|
||||
return x.equals(klalbPacket);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
public long statLengthBefore(int priority) {
|
||||
AtomicLong al=new AtomicLong();
|
||||
int n=0;
|
||||
for (Iterator<SumQueue<KLALBPacket>> iterator = sendDequeList.iterator(); iterator.hasNext();n++) {
|
||||
if(n>priority) {
|
||||
break;
|
||||
}
|
||||
SumQueue<KLALBPacket> queue = iterator.next();
|
||||
al.addAndGet(queue.getSumValue());
|
||||
|
||||
}
|
||||
return al.get();
|
||||
}
|
||||
|
||||
private long sndDelayFactor=Long.MIN_VALUE;
|
||||
private long rcvDelayFactor=Long.MIN_VALUE;
|
||||
|
||||
public long getSndDelayFactor() {
|
||||
return sndDelayFactor;
|
||||
}
|
||||
public long getRcvDelayFactor() {
|
||||
return rcvDelayFactor;
|
||||
}
|
||||
private volatile long predictTime;
|
||||
public void runPredict(KLALBPacket curr,int priority) {
|
||||
predictTime=sndDelayFactor;
|
||||
//System.out.println(this+" "+sndDelayFactor);
|
||||
long al=0;
|
||||
al=statLengthBefore(priority);
|
||||
long speed=getMonitor(). getOutSpeed();
|
||||
if(speed==0) {
|
||||
if(al>0) {
|
||||
predictTime=Long.MAX_VALUE;
|
||||
}
|
||||
}else {
|
||||
predictTime+=al*1000000000L/speed;
|
||||
}
|
||||
}
|
||||
@Override
|
||||
public int compareTo(KLALBRemoteLine o2) {
|
||||
long t1=this.predictTime;
|
||||
long t2=o2.predictTime;
|
||||
if(t1>t2) {
|
||||
return 1;
|
||||
}else if(t1<t2){
|
||||
return -1;
|
||||
}else {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
public void reconnectImmediately() {
|
||||
if(lthd!=null) {
|
||||
LockSupport.unpark(lthd);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user