forked from KNEMC/KLALB
TCP连接池功能
This commit is contained in:
Administrator
@@ -1,7 +1,9 @@
|
||||
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=10000000000L;
|
||||
public static final double A = 0.125;
|
||||
public static final int SO_TIMEOUT = 20000;
|
||||
public static final long SN_KEEP = 60000000000L;
|
||||
}
|
||||
|
||||
@@ -4,149 +4,90 @@ import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
import java.net.SocketException;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.Vector;
|
||||
import java.util.WeakHashMap;
|
||||
|
||||
import org.kne.cloud.network.mport.ServiceElement;
|
||||
import org.kne.cloud.network.mport.ThreadTool;
|
||||
|
||||
public class IOThreadManager {
|
||||
private KLALBCore klc=new KLALBCore(10000);
|
||||
private List<TCPConnection>tcps=new Vector<>();
|
||||
|
||||
|
||||
private TCPConnection local;
|
||||
|
||||
private KLALBCore klc = new KLALBCore(5000);
|
||||
private boolean isopen = true;
|
||||
|
||||
private volatile boolean open=true;
|
||||
public void startLocal() {
|
||||
|
||||
Thread upo=ThreadTool.makeVThreadIfSupport("本地接收线程",()->{
|
||||
try{
|
||||
|
||||
while(true) {
|
||||
byte[]b=new byte[Consts.BLOCKSIZE];
|
||||
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 {
|
||||
closeRemote();
|
||||
}
|
||||
});
|
||||
Thread downo=ThreadTool.makeVThreadIfSupport("本地发送线程",()->{
|
||||
try{
|
||||
while(true) {
|
||||
|
||||
local.getDout().write(klc.unpackDataBlock());
|
||||
local.getDout().flush();
|
||||
}
|
||||
}catch(InterruptedException s) {
|
||||
}catch(Exception e) {
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
}
|
||||
});
|
||||
upo.start();
|
||||
downo.start();
|
||||
}
|
||||
public void handleSocket(TCPConnection s) throws IOException {
|
||||
tcps.add(s);
|
||||
public void handleLocal(LocalTCPConnection tc, ServiceElement se,boolean syn) {
|
||||
klc.getLocaltcps().add(tc);
|
||||
try {
|
||||
Thread up=ThreadTool.makeVThreadIfSupport("远程发送线程",()->{
|
||||
try{
|
||||
while(true) {
|
||||
if(s.getTunnel().getName().contains("Openfrp")) {
|
||||
klc.sendDataBlockControlOnly(s);
|
||||
}else {
|
||||
klc.sendDataBlock(s);
|
||||
}
|
||||
if(syn) {
|
||||
KLALBBlock sbk=new KLALBBlock();
|
||||
sbk.cuid=tc.getCuid();
|
||||
sbk.number=0;
|
||||
sbk.command=2;
|
||||
sbk.lservice=se.proc;
|
||||
sbk.lipport=se.ipport;
|
||||
klc.submitDataBlockNoDelay(sbk);
|
||||
}
|
||||
}catch(InterruptedException s1) {
|
||||
Thread ls = ThreadTool.makeVThreadIfSupport("本地发送线程", () -> {
|
||||
|
||||
}catch(Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
});
|
||||
Thread down=ThreadTool.makeVThreadIfSupport("远程接收线程",()->{
|
||||
try{
|
||||
while(true) {
|
||||
klc.receiveDataBlock(s);
|
||||
}
|
||||
}catch(InterruptedException s1) {
|
||||
|
||||
}catch(Exception e) {
|
||||
up.interrupt();
|
||||
});
|
||||
Thread lr = ThreadTool.makeVThreadIfSupport("本地接收线程", () -> {
|
||||
|
||||
});
|
||||
ls.start();
|
||||
lr.start();
|
||||
ls.join();
|
||||
lr.join();
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
klc.getLocaltcps().remove(tc);
|
||||
}
|
||||
}
|
||||
|
||||
public void handleRemote(RemoteTCPConnection tc) {
|
||||
klc.getRemotetcps().add(tc);
|
||||
try {
|
||||
Thread rs = ThreadTool.makeVThreadIfSupport("远程发送线程", () -> {
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
klc.remoteSend(tc);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
});
|
||||
});
|
||||
Thread rr = ThreadTool.makeVThreadIfSupport("远程接收线程", () -> {
|
||||
try {
|
||||
while (true) {
|
||||
klc.remoteReceive(tc);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
});
|
||||
rs.start();
|
||||
rr.start();
|
||||
rs.join();
|
||||
rr.join();
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
} finally {
|
||||
klc.getRemotetcps().remove(tc);
|
||||
}
|
||||
}
|
||||
|
||||
up.start();
|
||||
down.start();
|
||||
try {
|
||||
down.join();
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}finally {
|
||||
tcps.remove(s);
|
||||
}
|
||||
}
|
||||
|
||||
public void closeRemote() {
|
||||
open=false;
|
||||
|
||||
for (int i = 0; i < tcps.size(); i++) {
|
||||
TCPConnection tll=tcps.get(i);
|
||||
tll.close();
|
||||
}
|
||||
klc.closeRemote();
|
||||
}
|
||||
|
||||
public void closeLocal() {
|
||||
open=false;
|
||||
local.close();
|
||||
klc.closeLocal();
|
||||
}
|
||||
|
||||
public TCPConnection getLocal() {
|
||||
return local;
|
||||
}
|
||||
public void setLocal(TCPConnection local) {
|
||||
this.local = local;
|
||||
}
|
||||
/*
|
||||
public void closeALL() {
|
||||
open=false;
|
||||
|
||||
for (int i = 0; i < tcps.size(); i++) {
|
||||
TCPConnection tll=tcps.get(i);
|
||||
tll.close();
|
||||
}
|
||||
klc.closeRemote();
|
||||
klc.closeLocal();
|
||||
try {
|
||||
in.close();
|
||||
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
try {
|
||||
out.close();
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
*/
|
||||
public List<TCPConnection> getTcps() {
|
||||
return tcps;
|
||||
public void close() {
|
||||
isopen = false;
|
||||
}
|
||||
|
||||
public boolean isOpen() {
|
||||
return open;
|
||||
return isopen;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -2,50 +2,58 @@ package org.kne.cloud.network.klalb;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Objects;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
import org.kne.cloud.network.mport.IPPort;
|
||||
|
||||
public class KLALBBlock {
|
||||
volatile byte[]data;
|
||||
volatile int size;
|
||||
volatile long number;
|
||||
|
||||
|
||||
volatile long time=-1;
|
||||
volatile TCPConnection connect;
|
||||
public KLALBBlock(byte[] b, int size,long number) {
|
||||
data=b;
|
||||
this.size=size;
|
||||
this.number=number;
|
||||
}
|
||||
private static AtomicLong sng=new AtomicLong(0);
|
||||
|
||||
public long sn;//每个数据包的唯一编号
|
||||
public UUID cuid;//用于识别数据包的stream ID号
|
||||
public long number;//数据包的编号，用于排序
|
||||
public byte[]data;//数据内容
|
||||
public int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST
|
||||
public long pingtime;//PING计时器
|
||||
|
||||
public String lservice;
|
||||
public IPPort lipport;
|
||||
|
||||
|
||||
|
||||
public KLALBBlock() {
|
||||
// TODO 自动生成的构造函数存根
|
||||
}
|
||||
sn=sng.getAndIncrement();
|
||||
}
|
||||
|
||||
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
StringBuilder sb=new StringBuilder();
|
||||
sb.append(cuid);
|
||||
sb.append(' ');
|
||||
if(number>0) {
|
||||
return "DATA"+number+":"+size;
|
||||
}else if(number==0){
|
||||
return "PING";
|
||||
}else if(number==Long.MIN_VALUE){
|
||||
return "PONG";
|
||||
sb.append("DATA").append(number).append(':').append(data.length);
|
||||
}else if(number<0) {
|
||||
sb.append("ACK").append(-number);
|
||||
}else {
|
||||
return "ACK"+(-number);
|
||||
switch(command) {
|
||||
case 0:
|
||||
sb.append("PING");
|
||||
break;
|
||||
case 1:
|
||||
sb.append("PONG");
|
||||
break;
|
||||
case 2:
|
||||
sb.append("SYN:").append(lservice).append(" ").append(lipport);
|
||||
break;
|
||||
case 3:
|
||||
sb.append("RST");
|
||||
break;
|
||||
}
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(number);
|
||||
}
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (this == obj)
|
||||
return true;
|
||||
if (obj == null)
|
||||
return false;
|
||||
if (getClass() != obj.getClass())
|
||||
return false;
|
||||
KLALBBlock other = (KLALBBlock) obj;
|
||||
return number == other.number;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import java.io.BufferedInputStream;
|
||||
import java.io.BufferedOutputStream;
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.net.ConnectException;
|
||||
import java.net.MalformedURLException;
|
||||
import java.net.Socket;
|
||||
import java.net.URI;
|
||||
@@ -21,111 +22,68 @@ import org.kne.cloud.network.mport.ServiceElement;
|
||||
import org.kne.cloud.network.mport.ThreadTool;
|
||||
|
||||
public class KLALBClient {
|
||||
private static Object olock=new Object();
|
||||
static {
|
||||
new Thread(()->{
|
||||
while(true) {
|
||||
try {
|
||||
Thread.sleep(10000);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
synchronized(olock) {
|
||||
olock.notifyAll();
|
||||
}
|
||||
}
|
||||
}).start();
|
||||
}
|
||||
|
||||
|
||||
|
||||
UUID suid=UUID.randomUUID();
|
||||
IOThreadManager iom=new IOThreadManager();
|
||||
private List<Tunnel> tls=new ArrayList<>();
|
||||
private TCPListener tcpl;
|
||||
private ServiceElement sel;
|
||||
|
||||
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.isOpen()) {
|
||||
try {
|
||||
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());
|
||||
tc.getDout().writeLong(uid.getMostSignificantBits());
|
||||
tc.getDout().writeLong(uid.getLeastSignificantBits());
|
||||
tc.getDout().flush();
|
||||
aig.incrementAndGet();
|
||||
System.out.println("隧道"+tll+"已连接，可用线路数量:"+ (kcp.getTcps().size()+1));
|
||||
try {
|
||||
kcp.handleSocket(tc);
|
||||
|
||||
}catch(IOException e){
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
tc.close();
|
||||
}
|
||||
int n=kcp.getTcps().size();
|
||||
System.out.println("隧道"+tll+"已断开，可用线路数量:"+ n);
|
||||
if(n<=0) {
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
synchronized(olock) {
|
||||
try {
|
||||
olock.wait();
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
}).start();
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
public void open(int port) throws IOException {
|
||||
tcpl=new TCPListener(port);
|
||||
tcpl.setCon((s)->{
|
||||
IOThreadManager kcp=new IOThreadManager();
|
||||
UUID uid=UUID.randomUUID();
|
||||
LocalTCPConnection tcc = null;
|
||||
try {
|
||||
tcc = new LocalTCPConnection(s);
|
||||
System.out.println("TCP："+tcc.getCuid()+"已连接");
|
||||
iom.handleLocal(tcc,sel,true);
|
||||
System.out.println("TCP："+tcc.getCuid()+"已关闭");
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
if(tcc!=null) {
|
||||
tcc.close();
|
||||
}
|
||||
}
|
||||
});
|
||||
tcpl.open();
|
||||
for (int i = 0; i < tls.size(); i++) {
|
||||
Tunnel tll=tls.get(i);
|
||||
ThreadTool.makeVThreadIfSupport("隧道监视线程", ()->{
|
||||
RemoteTCPConnection tc=null;
|
||||
while(iom.isOpen()) {
|
||||
try {
|
||||
kcp.setLocal(new TCPConnection(null, s));
|
||||
kcp.startLocal();
|
||||
runProtocol(kcp,uid);
|
||||
tc=new RemoteTCPConnection(tll);
|
||||
tc.getDout().writeShort(59649);
|
||||
tc.getDout().write(1);
|
||||
tc.getDout().writeUTF(tll.getName());
|
||||
tc.getDout().writeUTF(tll.getIp());
|
||||
tc.getDout().writeInt(tll.getPort());
|
||||
tc.getDout().writeLong(suid.getMostSignificantBits());
|
||||
tc.getDout().writeLong(suid.getLeastSignificantBits());
|
||||
tc.getDout().flush();
|
||||
System.out.println(tll+"：连接成功");
|
||||
iom.handleRemote(tc);
|
||||
System.out.println(tll+"：连接断开");
|
||||
} catch (UnknownHostException e) {
|
||||
e.printStackTrace();
|
||||
}catch(ConnectException e) {
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
if(tc!=null)
|
||||
tc.close();
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
|
||||
tcpl.open();
|
||||
try {
|
||||
Thread.sleep(5000);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}).start();
|
||||
}
|
||||
}
|
||||
public List<Tunnel> getTls() {
|
||||
return tls;
|
||||
@@ -139,7 +97,7 @@ public class KLALBClient {
|
||||
IPPort u=new IPPort(ipport);
|
||||
TCPConnection tcpc=null;
|
||||
try {
|
||||
tcpc=new TCPConnection(null, new Socket(u.getIp(),u.getPort()));
|
||||
tcpc=new RemoteTCPConnection(null, new Socket(u.getIp(),u.getPort()));
|
||||
tcpc.getDout().writeShort(59649);
|
||||
tcpc.getDout().write(0);
|
||||
tcpc.getDout().flush();
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
package org.kne.cloud.network.klalb;
|
||||
|
||||
import java.awt.Color;
|
||||
|
||||
import org.kne.ui.XFrame;
|
||||
|
||||
public class KLALBClientGUI extends XFrame{
|
||||
public KLALBClientGUI() {
|
||||
setResizable(false);
|
||||
setTitleColor(new Color(0,0,255,128));
|
||||
getContentPane().setBackground(new Color(128,128,255,128));
|
||||
setTitle("KNE云网络负载均衡客户端");
|
||||
setSize(600,370);
|
||||
setLocationRelativeTo(null);
|
||||
setVisible(true);
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
new KLALBClientGUI();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -13,9 +13,11 @@ import java.util.Iterator;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Queue;
|
||||
import java.util.Set;
|
||||
import java.util.TreeMap;
|
||||
import java.util.UUID;
|
||||
import java.util.Vector;
|
||||
import java.util.WeakHashMap;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
@@ -25,294 +27,154 @@ import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.Predicate;
|
||||
|
||||
import org.kne.cloud.network.mport.ThreadTool;
|
||||
|
||||
public class KLALBCore {
|
||||
private volatile long inputcount = 1;
|
||||
private volatile long outputcount = 1;
|
||||
private Set<KLALBBlock> inputcache = Collections.synchronizedSet(new HashSet<>());
|
||||
private List<KLALBBlock> outputcache = new Vector<>();
|
||||
|
||||
private BlockingQueue<KLALBBlock> ackp=new LinkedBlockingQueue<>();
|
||||
private BlockingQueue<KLALBBlock> ackq=new LinkedBlockingQueue<>();
|
||||
private volatile boolean inlocal=true;
|
||||
|
||||
private volatile boolean closeremote = false;
|
||||
private volatile boolean closelocal = false;
|
||||
|
||||
private static final UUID ZERO_UUID=new UUID(0,0);
|
||||
private volatile int cacheblocks;
|
||||
|
||||
|
||||
private List<LocalTCPConnection> localtcps = new Vector<>();
|
||||
private List<RemoteTCPConnection> remotetcps = new Vector<>();
|
||||
private Predicate<KLALBBlock>acceptSYN;
|
||||
|
||||
|
||||
private volatile long rsntime=System.nanoTime();
|
||||
private Set<SN> rsns=Collections.synchronizedSet(new HashSet<SN>());
|
||||
private static class SN{
|
||||
volatile long sn;
|
||||
volatile long time;
|
||||
public SN(long sn) {
|
||||
super();
|
||||
this.sn = sn;
|
||||
this.time=System.nanoTime();
|
||||
}
|
||||
public boolean isVaild() {
|
||||
return System.nanoTime()-time<=Consts.SN_KEEP;
|
||||
}
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(sn);
|
||||
}
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
if (this == obj)
|
||||
return true;
|
||||
if (obj == null)
|
||||
return false;
|
||||
if (getClass() != obj.getClass())
|
||||
return false;
|
||||
SN other = (SN) obj;
|
||||
return sn == other.sn;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public Predicate<KLALBBlock> getAcceptSYN() {
|
||||
return acceptSYN;
|
||||
}
|
||||
|
||||
public void setAcceptSYN(Predicate<KLALBBlock> acceptSYN) {
|
||||
this.acceptSYN = acceptSYN;
|
||||
}
|
||||
|
||||
public List<LocalTCPConnection> getLocaltcps() {
|
||||
return localtcps;
|
||||
}
|
||||
|
||||
public List<RemoteTCPConnection> getRemotetcps() {
|
||||
return remotetcps;
|
||||
}
|
||||
|
||||
public KLALBCore(int cachesize) {
|
||||
cacheblocks=cachesize;
|
||||
}
|
||||
|
||||
public void packDataBlock(byte[] b, int size) throws InterruptedException {
|
||||
|
||||
if (closelocal)
|
||||
throw new InterruptedException();
|
||||
while (!outputcache.isEmpty()&&(outputcount-outputcache.get(0).number>cacheblocks)) {
|
||||
if (closelocal)
|
||||
Thread.currentThread().interrupt();
|
||||
public void remoteSend(RemoteTCPConnection tc) throws InterruptedException, IOException {
|
||||
for(;;) {
|
||||
if(tc.checkPingTime()) {
|
||||
KLALBBlock pdb=new KLALBBlock();
|
||||
pdb.cuid=ZERO_UUID;
|
||||
pdb.number=0;
|
||||
pdb.command=0;
|
||||
pdb.pingtime=System.nanoTime();
|
||||
tc.sendBlock(pdb);
|
||||
}
|
||||
if(!tc.getSendDeque().isEmpty())
|
||||
break;
|
||||
Thread.sleep(1);
|
||||
}
|
||||
outputcache.add(new KLALBBlock(b, size, outputcount++));
|
||||
|
||||
KLALBBlock k=tc.getSendDeque().poll();
|
||||
if(k!=null) {
|
||||
tc.sendBlock(k);
|
||||
}
|
||||
}
|
||||
|
||||
public byte[] unpackDataBlock() throws InterruptedException {
|
||||
inlocal=false;
|
||||
try {
|
||||
if (closelocal)
|
||||
throw new InterruptedException();
|
||||
byte[] b = null;
|
||||
while (true) {
|
||||
synchronized (inputcache) {
|
||||
Iterator<KLALBBlock> klb = inputcache.iterator();
|
||||
while (klb.hasNext()) {
|
||||
KLALBBlock klalbBlock = (KLALBBlock) klb.next();
|
||||
if (klalbBlock.number < inputcount) {
|
||||
klb.remove();
|
||||
} else if (klalbBlock.number == inputcount) {
|
||||
b = klalbBlock.data;
|
||||
klb.remove();
|
||||
}
|
||||
public void remoteReceive(RemoteTCPConnection tc) throws IOException {
|
||||
KLALBBlock brc=null;
|
||||
for(;;) {
|
||||
brc=tc.receiveBlock();
|
||||
SN s=new SN(brc.sn);
|
||||
synchronized (rsns) {
|
||||
if(!rsns.contains(s)) {
|
||||
rsns.add(s);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
synchronized (rsns) {
|
||||
long ctime=System.nanoTime();
|
||||
if(ctime-rsntime>10000000000L) {
|
||||
rsntime=ctime;
|
||||
for (Iterator<SN> iterator = rsns.iterator(); iterator.hasNext();) {
|
||||
SN sn = (SN) iterator.next();
|
||||
if(!sn.isVaild()) {
|
||||
iterator.remove();
|
||||
}
|
||||
}
|
||||
if (b != null) {
|
||||
System.out.println("\tSORT:"+inputcount+" " + inputcache.size());
|
||||
inputcount++;
|
||||
return b;
|
||||
}
|
||||
if (closelocal)
|
||||
Thread.currentThread().interrupt();
|
||||
Thread.sleep(1);
|
||||
//System.err.println(inputcount);
|
||||
|
||||
}
|
||||
}finally {
|
||||
inlocal=true;
|
||||
}
|
||||
}
|
||||
|
||||
private void makeAck(KLALBBlock klalbBlock) {
|
||||
KLALBBlock klb1=new KLALBBlock(null, 0, -klalbBlock.number);
|
||||
klb1.time=klalbBlock.time;
|
||||
ackq.add(klb1);
|
||||
if(ackp.size()<100) {
|
||||
KLALBBlock klb2=new KLALBBlock(null, 0, -klalbBlock.number);
|
||||
klb2.time=klalbBlock.time;
|
||||
ackp.add(klb2);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public void sendDataBlock(TCPConnection out) throws IOException, InterruptedException {
|
||||
if (closeremote)
|
||||
throw new InterruptedException();
|
||||
//BlockingQueue<KLALBBlock> bqk=acks.get(out);
|
||||
while (ackq.isEmpty()&& outputcache.isEmpty()) {
|
||||
if (closeremote)
|
||||
Thread.currentThread().interrupt();
|
||||
Thread.sleep(1);
|
||||
if(out.checkPingTime()) {
|
||||
send0(out, new KLALBBlock(null,0 , 0));
|
||||
}
|
||||
}
|
||||
KLALBBlock klb=ackq.poll();
|
||||
if(klb!=null) {
|
||||
send0(out, klb);
|
||||
}else {
|
||||
KLALBBlock ks = null;
|
||||
synchronized (outputcache) {
|
||||
for (int i = 0; i < outputcache.size(); i++) {
|
||||
|
||||
KLALBBlock kd = outputcache.get(i);
|
||||
if (kd.connect == null) {
|
||||
kd.time = System.nanoTime();
|
||||
kd.connect = out;
|
||||
ks = kd;
|
||||
} else {
|
||||
if (kd.connect.isOpen()) {
|
||||
long timex = (System.nanoTime() - kd.time) / 1000000;
|
||||
if (timex > 100+1000*i) {
|
||||
kd.time = System.nanoTime();
|
||||
kd.connect = out;
|
||||
ks = kd;
|
||||
System.out.println("³¬Ê±ÖØ´«£º"+kd);
|
||||
}
|
||||
} else {
|
||||
kd.time = System.nanoTime();
|
||||
kd.connect = out;
|
||||
ks = kd;
|
||||
System.out.println("µôÏßÖØ´«£º"+kd);
|
||||
}
|
||||
}
|
||||
if (ks != null) {
|
||||
break;
|
||||
}
|
||||
if(brc.cuid.equals(ZERO_UUID)) {
|
||||
if(brc.number==0) {
|
||||
if(brc.command==0) {
|
||||
KLALBBlock pdb=new KLALBBlock();
|
||||
pdb.cuid=ZERO_UUID;
|
||||
pdb.number=0;
|
||||
pdb.command=1;
|
||||
pdb.pingtime=brc.pingtime;
|
||||
tc.getSendDeque().addFirst(pdb);
|
||||
}else if(brc.command==1) {
|
||||
long cur=System.nanoTime();
|
||||
long delay=(cur-brc.pingtime)/2;
|
||||
tc.setDelay(delay);
|
||||
}
|
||||
}
|
||||
if (ks != null) {
|
||||
send0(out, ks);
|
||||
ks.time = System.nanoTime();
|
||||
}else {
|
||||
Thread.sleep(1);
|
||||
}
|
||||
}
|
||||
|
||||
if(out.checkPingTime()) {
|
||||
send0(out, new KLALBBlock(null,0 , 0));
|
||||
}
|
||||
}
|
||||
|
||||
public void sendDataBlockControlOnly(TCPConnection out) throws IOException, InterruptedException {
|
||||
if (closeremote)
|
||||
throw new InterruptedException();
|
||||
//BlockingQueue<KLALBBlock> bqk=acks.get(out);
|
||||
while (ackp.isEmpty()) {
|
||||
if (closeremote)
|
||||
Thread.currentThread().interrupt();
|
||||
Thread.sleep(1);
|
||||
if(out.checkPingTime()) {
|
||||
send0(out, new KLALBBlock(null,0 , 0));
|
||||
}
|
||||
}
|
||||
KLALBBlock klb=ackp.poll();
|
||||
if(klb!=null) {
|
||||
send0(out, klb);
|
||||
//System.out.println(klb+" "+ackp.size());
|
||||
}
|
||||
if(out.checkPingTime()) {
|
||||
send0(out, new KLALBBlock(null,0 , 0));
|
||||
}
|
||||
}
|
||||
|
||||
public void receiveDataBlock(TCPConnection in) throws IOException, InterruptedException {
|
||||
if (closeremote)
|
||||
throw new InterruptedException();
|
||||
KLALBBlock x = receive0(in);
|
||||
long rect=System.nanoTime();
|
||||
if (x.number > 0) {
|
||||
|
||||
/*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) {
|
||||
|
||||
while(inputcache.size()>(2*cacheblocks)&&inlocal) {
|
||||
if (closeremote)
|
||||
Thread.currentThread().interrupt();
|
||||
Thread.sleep(1);
|
||||
}else {
|
||||
if(brc.number==0) {
|
||||
if(brc.command==0) {
|
||||
}else if(brc.command==1) {
|
||||
}else if(brc.command==2) {
|
||||
if(acceptSYN==null||!acceptSYN.test(brc)) {
|
||||
KLALBBlock rst=new KLALBBlock();
|
||||
rst.cuid=brc.cuid;
|
||||
rst.number=0;
|
||||
rst.command=3;
|
||||
submitDataBlockNoDelay(rst);
|
||||
}
|
||||
|
||||
inputcache.add(x);
|
||||
}else if(brc.command==3) {
|
||||
}
|
||||
makeAck(x);
|
||||
|
||||
} else if(x.number==0) {
|
||||
ThreadTool.makeVThreadIfSupport("TACK", ()->{
|
||||
try {
|
||||
KLALBBlock klk=new KLALBBlock(null,0,Long.MIN_VALUE);
|
||||
long st=System.nanoTime();
|
||||
long tw=st-rect;
|
||||
klk.time=x.time+tw;
|
||||
send0(in, klk);
|
||||
} catch (IOException e) {
|
||||
}finally {
|
||||
}
|
||||
}).start();
|
||||
}else if(x.number==Long.MIN_VALUE){
|
||||
//System.out.println("PING:"+(System.nanoTime()-x.time)/2000000);
|
||||
long del=(System.nanoTime()-x.time)/2;
|
||||
in.setDelay(del);
|
||||
}else{
|
||||
long v = -x.number;
|
||||
outputcache.removeIf((b) -> {
|
||||
return b.number == v;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private void send0(TCPConnection tcp, KLALBBlock kd) throws IOException {
|
||||
DataOutputStream out=tcp.getDout();
|
||||
synchronized (out) {
|
||||
|
||||
out.writeLong(kd.number);
|
||||
if (kd.number > 0) {
|
||||
out.writeLong(System.nanoTime());
|
||||
out.writeInt(kd.size);
|
||||
out.write(kd.data, 0, kd.size);
|
||||
}else if(kd.number==0){
|
||||
out.writeLong(System.nanoTime());
|
||||
}else if(kd.number==Long.MIN_VALUE) {
|
||||
out.writeLong(kd.time);
|
||||
}else {
|
||||
out.writeLong(kd.time);
|
||||
}
|
||||
out.flush();
|
||||
public void submitDataBlock(KLALBBlock kb) {
|
||||
|
||||
}
|
||||
System.out.println(tcp.getTunnel().getName()+" SEND:" + kd);
|
||||
}
|
||||
|
||||
private KLALBBlock receive0(TCPConnection tcp) throws IOException {
|
||||
KLALBBlock kb = new KLALBBlock();
|
||||
DataInputStream in=tcp.getDin();
|
||||
synchronized (in) {
|
||||
|
||||
kb.number = in.readLong();
|
||||
if (kb.number > 0) {
|
||||
kb.time=in.readLong();
|
||||
kb.size = in.readInt();
|
||||
if(kb.size<0||kb.size>65536) {
|
||||
throw new StreamCorruptedException(tcp+"block size error:"+kb.size);
|
||||
}
|
||||
kb.data = new byte[kb.size];
|
||||
in.readFully(kb.data);
|
||||
}else if(kb.number==0){
|
||||
kb.time=in.readLong();
|
||||
}else if(kb.number==Long.MIN_VALUE) {
|
||||
kb.time=in.readLong();
|
||||
}else {
|
||||
kb.time=in.readLong();
|
||||
//tcp.nextRTT(System.nanoTime()-kb.time);
|
||||
}
|
||||
}
|
||||
System.out.println(tcp.getTunnel().getName()+" RECEIVE:" + kb);
|
||||
return kb;
|
||||
}
|
||||
|
||||
public void closeRemote() {
|
||||
closeremote = true;
|
||||
outputcache.clear();
|
||||
}
|
||||
public void closeLocal() {
|
||||
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) {
|
||||
e.printStackTrace();
|
||||
public void submitDataBlockNoDelay(KLALBBlock kb) {
|
||||
synchronized (remotetcps) {
|
||||
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
|
||||
RemoteTCPConnection rmt = (RemoteTCPConnection) iterator.next();
|
||||
rmt.getSendDeque().addFirst(kb);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,14 +18,15 @@ import org.kne.cloud.network.mport.IPPort;
|
||||
import org.kne.cloud.network.mport.ServiceElement;
|
||||
|
||||
public class KLALBServer {
|
||||
|
||||
WeakHashMap<UUID, IOThreadManager> whm=new WeakHashMap<>();
|
||||
public KLALBServer(int port, Map<String,ServiceElement> services,List<Tunnel> tunnels) throws IOException {
|
||||
TCPListener tcpl=new TCPListener(port);
|
||||
tcpl.setCon((s)->{
|
||||
TCPConnection tcc=null;
|
||||
RemoteTCPConnection tcc=null;
|
||||
try {
|
||||
//s.setSoTimeout(10000);
|
||||
tcc=new TCPConnection(null,s);
|
||||
tcc=new RemoteTCPConnection(null,s);
|
||||
DataInputStream din=tcc.getDin();
|
||||
int val=din.readShort()&0xffff;
|
||||
if(val!=59649) {
|
||||
@@ -52,16 +53,6 @@ public class KLALBServer {
|
||||
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();
|
||||
@@ -69,28 +60,16 @@ public class KLALBServer {
|
||||
tcc.setTunnel(tll);
|
||||
|
||||
UUID uid=new UUID(din.readLong(),din.readLong());
|
||||
|
||||
System.out.println(new IPPort((InetSocketAddress)s.getRemoteSocketAddress())+"->"+tll.getIpport()+"->"+eas.ipport);
|
||||
IOThreadManager nx = null;
|
||||
synchronized (tcpl) {
|
||||
if(whm.containsKey(uid)) {
|
||||
nx=whm.get(uid);
|
||||
}else {
|
||||
nx=new IOThreadManager();
|
||||
Socket soc=new Socket(eas.ipport.getIp(),eas.ipport.getPort());
|
||||
nx.setLocal(new TCPConnection(null, soc));
|
||||
nx.startLocal();
|
||||
whm.put(uid, nx);
|
||||
|
||||
}
|
||||
nx.handleSocket(tcc);
|
||||
int n=nx.getTcps().size();
|
||||
System.out.println(nx.getTcps());
|
||||
if(n<=0) {
|
||||
nx.closeLocal();
|
||||
System.out.println("连接已关闭");
|
||||
}
|
||||
}catch(ConnectException e) {
|
||||
System.out.println("连接本地服务失败，请检查你的服务程序");
|
||||
nx.handleRemote(tcc);
|
||||
}catch(IOException e) {
|
||||
e.printStackTrace();
|
||||
}finally {
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
package org.kne.cloud.network.klalb;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.Socket;
|
||||
import java.net.UnknownHostException;
|
||||
import java.util.UUID;
|
||||
|
||||
public class LocalTCPConnection extends TCPConnection {
|
||||
private UUID cuid;
|
||||
public LocalTCPConnection(Socket s) throws IOException {
|
||||
super(s);
|
||||
cuid=UUID.randomUUID();
|
||||
}
|
||||
public LocalTCPConnection(Socket s,UUID uid) throws IOException {
|
||||
super(s);
|
||||
cuid=uid;
|
||||
}
|
||||
public UUID getCuid() {
|
||||
return cuid;
|
||||
}
|
||||
public void setCuid(UUID cuid) {
|
||||
this.cuid = cuid;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
package org.kne.cloud.network.klalb;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.Socket;
|
||||
import java.net.UnknownHostException;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.BlockingDeque;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
|
||||
import org.kne.cloud.network.mport.IPPort;
|
||||
|
||||
public class RemoteTCPConnection extends TCPConnection {
|
||||
|
||||
public RemoteTCPConnection(Tunnel t) throws UnknownHostException, IOException {
|
||||
this(t,t.connectClientSocket());
|
||||
}
|
||||
public RemoteTCPConnection(Tunnel t,Socket s) throws UnknownHostException, IOException {
|
||||
super(s);
|
||||
this.tunnel=t;
|
||||
s.setSoTimeout(Consts.SO_TIMEOUT);
|
||||
}
|
||||
private long delay=-1;
|
||||
private Tunnel tunnel;
|
||||
public Tunnel getTunnel() {
|
||||
return tunnel;
|
||||
}
|
||||
|
||||
public void setTunnel(Tunnel tunnel) {
|
||||
this.tunnel = tunnel;
|
||||
}
|
||||
public long getDelay() {
|
||||
return delay;
|
||||
}
|
||||
|
||||
public void setDelay(long delay) {
|
||||
this.delay = delay;
|
||||
if(tunnel!=null) {
|
||||
tunnel.setDelay(delay);
|
||||
}
|
||||
|
||||
}
|
||||
public void sendBlock(KLALBBlock data) throws IOException {
|
||||
dout.writeLong(data.sn);
|
||||
dout.writeLong(data.cuid.getMostSignificantBits());
|
||||
dout.writeLong(data.cuid.getLeastSignificantBits());
|
||||
dout.writeLong(data.number);
|
||||
if(data.number>0) {
|
||||
dout.writeInt(data.data.length);
|
||||
dout.write(data.data);
|
||||
}else if(data.number==0) {
|
||||
dout.write(data.command);
|
||||
switch(data.command) {
|
||||
case 0:
|
||||
case 1:
|
||||
dout.writeLong(data.pingtime);
|
||||
break;
|
||||
case 2:
|
||||
dout.writeUTF(data.lservice);
|
||||
dout.writeUTF(data.lipport.toString());
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
dout.flush();
|
||||
|
||||
System.out.println("SEND:"+data);
|
||||
}
|
||||
public KLALBBlock receiveBlock() throws IOException {
|
||||
KLALBBlock klb=new KLALBBlock();
|
||||
klb.sn=din.readLong();
|
||||
klb.cuid=new UUID(din.readLong(), din.readLong());
|
||||
klb.number=din.readLong();
|
||||
if(klb.number>0) {
|
||||
int size=din.readInt();
|
||||
byte[]d=new byte[size];
|
||||
din.readFully(d);
|
||||
klb.data=d;
|
||||
}else if(klb.number==0) {
|
||||
klb.command=din.read();
|
||||
switch(klb.command) {
|
||||
case 0:
|
||||
case 1:
|
||||
klb.pingtime=din.readLong();
|
||||
break;
|
||||
case 2:
|
||||
klb.lservice=din.readUTF();
|
||||
klb.lipport=new IPPort(din.readUTF());
|
||||
}
|
||||
}
|
||||
|
||||
System.out.println("RECEIVE:"+klb);
|
||||
return klb;
|
||||
}
|
||||
|
||||
private volatile long time=System.nanoTime();
|
||||
public boolean checkPingTime() {
|
||||
long cu=System.nanoTime();
|
||||
if(cu-time>Consts.PINGTIMENS) {
|
||||
time=cu;
|
||||
return true;
|
||||
}else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private BlockingDeque<KLALBBlock>sendDeque=new LinkedBlockingDeque<>();
|
||||
public BlockingDeque<KLALBBlock> getSendDeque() {
|
||||
return sendDeque;
|
||||
}
|
||||
}
|
||||
@@ -8,28 +8,16 @@ import java.io.IOException;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketException;
|
||||
import java.net.UnknownHostException;
|
||||
import java.util.zip.GZIPInputStream;
|
||||
import java.util.zip.GZIPOutputStream;
|
||||
|
||||
public class TCPConnection {
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "TCPConnection [tunnel=" + tunnel + ", connect=" + connect + ", delay=" + delay + "]";
|
||||
}
|
||||
|
||||
private Tunnel tunnel;
|
||||
|
||||
private Socket connect;
|
||||
private DataInputStream din;
|
||||
|
||||
public Tunnel getTunnel() {
|
||||
return tunnel;
|
||||
}
|
||||
|
||||
private DataOutputStream dout;
|
||||
private long delay=-1;
|
||||
|
||||
/*public Socket getConnect() {
|
||||
return connect;
|
||||
}*/
|
||||
protected DataInputStream din;
|
||||
protected DataOutputStream dout;
|
||||
|
||||
public DataInputStream getDin() {
|
||||
return din;
|
||||
@@ -61,44 +49,19 @@ public class TCPConnection {
|
||||
// TODO 自动生成的 catch 块
|
||||
e.printStackTrace();
|
||||
}
|
||||
if(tunnel!=null&&connect!=null)
|
||||
tunnel.getCCount().decrementAndGet();
|
||||
connect = null;
|
||||
}
|
||||
|
||||
public long getDelay() {
|
||||
return delay;
|
||||
}
|
||||
|
||||
public void setDelay(long delay) {
|
||||
this.delay = delay;
|
||||
if(tunnel!=null) {
|
||||
tunnel.setDelay(delay);
|
||||
}
|
||||
if(connect!=null) {
|
||||
try {
|
||||
connect.setSoTimeout(10000);
|
||||
} catch (SocketException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void setTunnel(Tunnel tunnel) {
|
||||
this.tunnel = tunnel;
|
||||
}
|
||||
|
||||
public TCPConnection(Tunnel t) throws UnknownHostException, IOException {
|
||||
this(t, t.connectClientSocket());
|
||||
}
|
||||
public TCPConnection(Tunnel t, Socket soc) throws IOException {
|
||||
public TCPConnection( Socket soc) throws IOException {
|
||||
connect = soc;
|
||||
tunnel = t;
|
||||
if(t!=null) {
|
||||
t.getCCount().incrementAndGet();
|
||||
}
|
||||
initIO();
|
||||
}
|
||||
|
||||
|
||||
|
||||
protected void initIO() throws IOException {
|
||||
dout = new DataOutputStream( connect.getOutputStream());
|
||||
din = new DataInputStream(connect.getInputStream());
|
||||
dout = new DataOutputStream(connect.getOutputStream());
|
||||
}
|
||||
|
||||
public boolean isOpen() {
|
||||
@@ -107,24 +70,6 @@ public class TCPConnection {
|
||||
}
|
||||
return !connect.isClosed();
|
||||
}
|
||||
|
||||
|
||||
private volatile long time=System.nanoTime();
|
||||
public boolean checkPingTime() {
|
||||
long cu=System.nanoTime();
|
||||
if(cu-time>Consts.PINGTIMENS) {
|
||||
time=cu;
|
||||
return true;
|
||||
}else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
private volatile long RTT=10000000;
|
||||
public void nextRTT(long NRTT) {
|
||||
RTT=(long) ((1.0-Consts.A)*RTT+Consts.A*NRTT);
|
||||
System.out.println(RTT);
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -76,7 +76,7 @@ public class Tunnel {
|
||||
}
|
||||
public Socket connectClientSocket() throws UnknownHostException, IOException {
|
||||
Socket socket=new Socket();
|
||||
socket.connect(ipport.getSocketAddress(), 5000);
|
||||
socket.connect(ipport.getSocketAddress(), 10000);
|
||||
return socket;
|
||||
}
|
||||
|
||||
@@ -109,5 +109,4 @@ public class Tunnel {
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user