KNEMC/KLALB
1
2
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

修复长时间使用导致连接卡死的故障

Browse Source 
优化性能，减少内存占用
提高断线切换的响应速度
This commit is contained in:
Administrator
2023-04-05 08:44:37 +08:00
parent cc1df750ae
commit 0499ecb34d
10 changed files with 508 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/org/kne/cloud/network/klalb/Consts.java
+5 -4
View File
@@ -1,13 +1,14 @@
package org.kne.cloud.network.klalb;
public class Consts {
public static final int BLOCKSIZE=16384;
//public static final int BLOCKSIZE=8192;
public static final long PINGTIMENS=1000000000L;
public static final int SO_TIMEOUT = 5000;
public static final int SO_TIMEOUT = 2000;
public static final long SN_KEEP = 60000000000L;
public static final int MAX_RESEND = 10;
public static final long UACK_TIME = 5000000000L;
public static final int MAX_RESEND = 20;
public static final long UACK_TIME = 2000000000L;
public static final int CONNECT_TIMEOUT=10000;
public static final double MAX_QUEUE_TIME=500;
public static final long KEEP_TIME = 60000000000L;
public static final int PRE_SO_TIMEOUT = 10000;
}
src/org/kne/cloud/network/klalb/KLALBBlock.java
+10 -7
View File
@@ -8,15 +8,15 @@ import java.util.concurrent.atomic.AtomicLong;
import org.kne.cloud.network.mport.IPPort;
public class KLALBBlock implements Comparable<KLALBBlock>{
private static AtomicLong sng=new AtomicLong(1);
//private static AtomicLong sng=new AtomicLong(1);
public volatile long sn;//每个数据包的唯一编号
//public volatile long sn;//每个数据包的唯一编号
public volatile int cuid;//用于识别数据包的stream ID号
public volatile long number;//数据包的编号，用于排序
public volatile byte[]data;//数据内容
public volatile int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST 4=KEEP
public volatile int command;//功能标记 0=PING 1=PONG 2=SYN 3=RST 4=KEEP 5=SYK
public volatile long pingtime;//PING计时器
public volatile int cacheused;
public volatile boolean cacheused;
public volatile String lservice;
public volatile IPPort lipport;
@@ -24,12 +24,12 @@ public class KLALBBlock implements Comparable<KLALBBlock>{
public transient volatile long sendtime;
public transient volatile int resend;
public volatile long sendtimeForRTT;
//public volatile long sendtimeForRTT;
public KLALBBlock() {
/*public KLALBBlock() {
sn=sng.getAndIncrement();
}
}*/
@@ -60,6 +60,9 @@ public class KLALBBlock implements Comparable<KLALBBlock>{
case 3:
sb.append("RST");
break;
case 5:
sb.append("SYK");
break;
}
}
return sb.toString();
src/org/kne/cloud/network/klalb/KLALBClient.java
+2 -2
View File
@@ -36,7 +36,7 @@ import com.google.gson.stream.JsonReader;
public class KLALBClient {
private UUID suid=UUID.randomUUID();
private IOThreadManager iom=new IOThreadManager();
private KLALBController iom=new KLALBController();
private List<Tunnel> tls=new ArrayList<>();
private List<TCPListener> tcpl=new ArrayList<>();
private List<ProxyElement> pet=new ArrayList<>();
@@ -114,7 +114,7 @@ public class KLALBClient {
}
try {
//System.out.println(re);
Thread.sleep(1000*(1<<Math.min(re,7)));
Thread.sleep(5000*(1<<Math.min(re,3)));
} catch (InterruptedException e) {
e.printStackTrace();
}
src/org/kne/cloud/network/klalb/IOThreadManager.java → src/org/kne/cloud/network/klalb/KLALBController.java
+32 -53
View File
@@ -3,6 +3,7 @@ package org.kne.cloud.network.klalb;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.SocketException;
import java.util.Collection;
import java.util.HashMap;
@@ -22,11 +23,11 @@ import java.util.concurrent.atomic.AtomicReference;
import org.kne.cloud.network.mport.ServiceElement;
import org.kne.cloud.network.mport.ThreadTool;
public class IOThreadManager {
private KLALBCore klc = new KLALBCore(1000);
public class KLALBController {
private KLALBCore klc = new KLALBCore();
private volatile boolean isopen = true;
public IOThreadManager() {
public KLALBController() {
}
@@ -34,8 +35,8 @@ public class IOThreadManager {
if(!isopen) {
return;
}
klc.getLocaltcps().put(tc.getCuid(), tc);
//klc.getLocaltcps().put(tc.getCuid(), tc);
klc.openLocal(tc);
// checkRemotes();
AtomicBoolean AB=new AtomicBoolean(true);
@@ -54,18 +55,15 @@ public class IOThreadManager {
}
} catch (InterruptedException e) {
e.printStackTrace();
AB.set(false);
//tn.get().interrupt();
//ltn.get().interrupt();
tc.getOutputcache().clear();
tc.close();
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 3;
klc.submitDataBlockNoDelay(sbk);
} catch (IOException e) {
e.printStackTrace();
AB.set(false);
/*tn.get().interrupt();
ltn.get().interrupt();
tc.getOutputcache().clear();*/
tc.close();
klc.closeLocal(tc);
}
});
@@ -84,14 +82,10 @@ public class IOThreadManager {
} catch (IOException|InterruptedException e) {
e.printStackTrace();
AB.set(false);
lt.interrupt();
tc.getOutputcache().clear();
/*lt.interrupt();
tc.getOutputcache().clear();*/
tc.close();
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 3;
klc.submitDataBlockNoDelay(sbk);
klc.closeLocal(tc);
}finally {
try {
tc.getDout().close();
@@ -114,15 +108,11 @@ public class IOThreadManager {
} catch (IOException|InterruptedException e) {
e.printStackTrace();
AB.set(false);
ls.interrupt();
/*ls.interrupt();
lt.interrupt();
tc.getOutputcache().clear();
tc.getOutputcache().clear();*/
tc.close();
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 3;
klc.submitDataBlockNoDelay(sbk);
klc.closeLocal(tc);
} finally {
try {
tc.getDin().close();
@@ -136,19 +126,14 @@ public class IOThreadManager {
//System.out.println("xxxxxxxxxxxxxxxxxxxxxxxxxx");
//System.exit(123);
AB.set(false);
ls.interrupt();
/*ls.interrupt();
lt.interrupt();
tc.getOutputcache().clear();
tc.getOutputcache().clear();*/
tc.close();
klc.closeLocal(tc);
});
if (syn) {
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 2;
sbk.lservice = tc.getServiceElement().name;
sbk.lipport = tc.getServiceElement().ipport;
klc.submitAckBlockNoDelay(sbk,1);
klc.connectLocal(tc);
}
ls.start();
lr.start();
@@ -161,8 +146,12 @@ public class IOThreadManager {
} catch (InterruptedException e) {
e.printStackTrace();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
} finally {
klc.getLocaltcps().remove(tc.getCuid());
klc.closeLocal(tc);
//klc.getLocaltcps().remove(tc.getCuid());
}
}
@@ -170,7 +159,7 @@ public class IOThreadManager {
if(!isopen) {
return;
}
klc.getRemotetcps().add(tc);
klc.openRemote(tc);
try {
Thread rs = ThreadTool.makeVThreadIfSupport("远程发送线程", () -> {
@@ -200,19 +189,8 @@ public class IOThreadManager {
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
klc.getRemotetcps().remove(tc);
Queue<KLALBBlock> sq=tc.getSendDeque();
KLALBBlock tmp;
while((tmp=sq.poll())!=null) {
try {
klc.submitDataBlock(tmp);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (IOException e) {
// TODO 自动生成的 catch 块
e.printStackTrace();
}
}
klc.closeRemote(tc);
// checkRemotes();
}
@@ -260,4 +238,5 @@ public class IOThreadManager {
return klc;
}
}
src/org/kne/cloud/network/klalb/KLALBCore.java
+203 -60
View File
@@ -6,6 +6,8 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.StreamCorruptedException;
import java.lang.invoke.ConstantCallSite;
import java.net.ConnectException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -33,21 +35,25 @@ import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.zip.DataFormatException;
import org.kne.cloud.network.mport.ThreadTool;
public class KLALBCore {
private volatile long RTT=1000000000;
/*private volatile long RTT=1000000000;
private volatile long RTO=1000000000;
private volatile long DevRTT=0;
private volatile int cacheblocks;
*/
//private volatile int cacheblocks;
private Map<Integer, LocalTCPConnection> localtcps = new ConcurrentHashMap<>();
private Set<Integer>localtcpsconnecting=Collections.synchronizedSet(new HashSet<>());
private List<RemoteTCPConnection> remotetcps = new Vector<>();
private Predicate<KLALBBlock>acceptSYN;
private BiConsumer<KLALBBlock,Consumer<Boolean>>acceptSYN;
/*private volatile long rsntime=System.nanoTime();
@@ -81,11 +87,11 @@ public class KLALBCore {
}*/
public Predicate<KLALBBlock> getAcceptSYN() {
public BiConsumer<KLALBBlock, Consumer<Boolean>> getAcceptSYN() {
return acceptSYN;
}
public void setAcceptSYN(Predicate<KLALBBlock> acceptSYN) {
public void setAcceptSYN(BiConsumer<KLALBBlock, Consumer<Boolean>> acceptSYN) {
this.acceptSYN = acceptSYN;
}
@@ -97,9 +103,6 @@ public class KLALBCore {
return remotetcps;
}
public KLALBCore(int cachesize) {
cacheblocks=cachesize;
}
public void remoteSend(RemoteTCPConnection tc) throws InterruptedException, IOException {
for(;;) {
@@ -112,13 +115,12 @@ public class KLALBCore {
tc.sendBlock(pdb);
tc.updateTraffics();
}
if(!tc.getNDSendDeque().isEmpty()||!tc.getSendDeque().isEmpty())
if(!tc.isEmpty())
break;
synchronized (tc.getSendDeque()) {
tc.getSendDeque().wait(10);
tc.lock(10);
}
}
KLALBBlock k0=tc.getNDSendDeque().poll();
KLALBBlock kbv=tc.getNextBlock();
/*KLALBBlock k0=tc.getNDSendDeque().poll();
if(k0!=null) {
if(k0.cuid==0&&k0.number==0&&k0.command==1) {
k0.pingtime+=System.nanoTime()- k0.sendtime;
@@ -132,6 +134,12 @@ public class KLALBCore {
tc.sendBlock(k);
}
}*/
if(kbv!=null) {
if(kbv.cuid==0&&kbv.number==0&&kbv.command==1) {
kbv.pingtime+=System.nanoTime()- kbv.sendtime;
}
tc.sendBlock(kbv);
}
}
public void remoteReceive(RemoteTCPConnection tc) throws IOException {
@@ -181,11 +189,13 @@ public class KLALBCore {
pdb.command=1;
pdb.pingtime=brc.pingtime;
pdb.sendtime=brc.sendtime;
tc.getNDSendDeque().offer(pdb);
tc.addBlock(pdb, 0);
//tc.getNDSendDeque().offer(pdb);
}else if(brc.command==1) {
long cur=System.nanoTime();
long delay=cur-brc.pingtime;
setDelayAvg(tc,delay);
tc.setSoTimeout(Consts.SO_TIMEOUT);
}
}
}else {
@@ -193,23 +203,53 @@ public class KLALBCore {
if(brc.command==0) {
}else if(brc.command==1) {
}else if(brc.command==2) {
if(!localtcps.containsKey(brc.cuid)) {
if(acceptSYN==null||!acceptSYN.test(brc)) {
if(!localtcps.containsKey(brc.cuid)&&!localtcpsconnecting.contains(brc.cuid)) {
localtcpsconnecting.add(brc.cuid);
KLALBBlock brcx=brc;
ThreadTool.makeVThreadIfSupport("局域网连接监视线程", ()->{
try {
if(acceptSYN==null) {
KLALBBlock rst=new KLALBBlock();
rst.cuid=brc.cuid;
rst.cuid=brcx.cuid;
rst.number=0;
rst.command=3;
submitDataBlockNoDelay(rst);
}else {acceptSYN.accept(brcx,(Consumer<Boolean>)(c)->{
if(c) {
KLALBBlock syk=new KLALBBlock();
syk.cuid=brcx.cuid;
syk.number=0;
syk.command=5;
submitDataBlockNoDelay(syk);
}else {
KLALBBlock rst=new KLALBBlock();
rst.cuid=brcx.cuid;
rst.number=0;
rst.command=3;
submitDataBlockNoDelay(rst);
}
});
}
}finally {
localtcpsconnecting.remove(brcx.cuid);
}
}).start();
}
}else if(brc.command==3) {
LocalTCPConnection ltc=localtcps.get(brc.cuid);
if(ltc!=null) {
//ltc.close();
ltc.setIsrst(true);
ltc.getRSTHook().accept(ltc);
}
}else if(brc.command==5) {
LocalTCPConnection ltc=localtcps.get(brc.cuid);
if(ltc!=null) {
ltc.setIssyk(true);
}
}
}else if(brc.number>0){
LocalTCPConnection ltcs=localtcps.get(brc.cuid);
@@ -217,9 +257,41 @@ public class KLALBCore {
KLALBBlock ack=new KLALBBlock();
ack.cuid=brc.cuid;
ack.number=-brc.number;
ack.cacheused=ltcs.getSendDeque().size();
ack.cacheused=ltcs.getSendDeque().size()>ltcs.getInputcachesize()/ltcs.getBlocksize();
submitAckBlockNoDelay(ack,3);
ltcs.getSendDeque().add(brc);
List<KLALBBlock> ic=ltcs.getInputcache();
synchronized(ic) {
if(brc.number>=ltcs.getInputcount()) {
ic.add(brc);
while(true) {
KLALBBlock kkb=null;
for (int i = 0; i < ic.size(); i++) {
KLALBBlock klalbBlock = ic.get(i);
if(klalbBlock.number==ltcs.getInputcount()) {
ic.remove(i);
i--;
kkb=klalbBlock;
break;
}
}
if(kkb==null)
break;
ltcs.getSendDeque().add(kkb);
ltcs.incInputcount();
}
/*for (Iterator<KLALBBlock> iterator = ltcs.getInputcache().iterator(); iterator.hasNext();) {
KLALBBlock klalbBlock = (KLALBBlock) iterator.next();
if(klalbBlock.number==ltcs.getInputcount()) {
iterator.remove();
ltcs.getSendDeque().add(klalbBlock);
ltcs.incInputcount();
}else {
break;
}
}*/
}
}
}
@@ -227,7 +299,7 @@ public class KLALBCore {
long nx=-brc.number;
LocalTCPConnection ltc=localtcps.get(brc.cuid);
if(ltc!=null) {
List<KLALBBlock> l=ltc.getOutputcache();
/*List<KLALBBlock> l=ltc.getOutputcache();
synchronized (l) {
for (Iterator<KLALBBlock> iterator = l.iterator(); iterator.hasNext();) {
KLALBBlock klalbBlock = (KLALBBlock) iterator.next();
@@ -240,13 +312,13 @@ public class KLALBCore {
//System.out.println(RTO/1000000);
}
}
}
}*/
/*ltc.getOutputcache().removeIf((b) -> {
ltc.getOutputcache().removeIf((b) -> {
return b.number == nx;
});*/
ltc.setPeerCacheUsed(brc.cacheused);
});
ltc.setPeerFull(brc.cacheused);
}else {
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = brc.cuid;
@@ -258,19 +330,18 @@ public class KLALBCore {
synchronized (remotetcps) {
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
Queue<KLALBBlock>sq=remoteTCPConnection.getSendDeque();
sq.removeIf((vb)->{
return vb.number==nx&&vb.cuid==cuid;
});
remoteTCPConnection.ackRemove(nx,cuid);
/*Queue<KLALBBlock>sq=remoteTCPConnection.getSendDeque();
*/
}
}
}
}
}
private void sortRemoteTCPByPrediction() {
private void sortRemoteTCPByPrediction(int prio) {
remotetcps.forEach((x)->{
x.predictLatency();
x.predictLatency(prio);
});
Collections.sort(remotetcps, new Comparator<RemoteTCPConnection>() {
@Override
@@ -324,20 +395,19 @@ public class KLALBCore {
submitDataBlock(ks,1);
}
public void submitDataBlock(KLALBBlock kb,int i) throws InterruptedException, IOException {
submitDataBlock(kb,1,5);
}
public void submitDataBlock(KLALBBlock kb,int i,int prio) throws InterruptedException, IOException {
int ni=Math.min(i, remotetcps.size());
while(true) {
if(remotetcps.isEmpty())
throw new IOException("发送错误");
synchronized (remotetcps) {
sortRemoteTCPByPrediction();
sortRemoteTCPByPrediction(5);
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
RemoteTCPConnection remoteTCPConnection = (RemoteTCPConnection) iterator.next();
RemoteTCPConnection rmt = (RemoteTCPConnection) iterator.next();
//if(remoteTCPConnection.getPredictedLatency()<Consts.MAX_QUEUE_TIME) {
Queue<KLALBBlock> bdq=remoteTCPConnection.getSendDeque();
bdq.add(kb);
synchronized (bdq) {
bdq.notifyAll();
}
rmt.addBlock(kb, prio);
ni--;
if(ni<=0)
return;
@@ -352,11 +422,7 @@ public class KLALBCore {
synchronized (remotetcps) {
for (Iterator<RemoteTCPConnection> iterator = remotetcps.iterator(); iterator.hasNext();) {
RemoteTCPConnection rmt = (RemoteTCPConnection) iterator.next();
Queue<KLALBBlock> sq = rmt.getNDSendDeque();
sq.offer(kb);
synchronized (sq) {
sq.notifyAll();
}
rmt.addBlock(kb, 0);
}
}
}
@@ -369,33 +435,39 @@ public class KLALBCore {
return;
}
RemoteTCPConnection rmt = (RemoteTCPConnection) iterator.next();
Queue<KLALBBlock> sq = rmt.getNDSendDeque();
sq.offer(kb);
synchronized (sq) {
sq.notifyAll();
}
rmt.addBlock(kb, 0);
coun++;
}
}
}
public void putDataBlock(LocalTCPConnection tc, KLALBBlock ks) throws InterruptedException, IOException {
while (true) {
/*while (true) {
if(!localtcps.containsKey(tc.getCuid()))
throw new IOException("本地连接已关闭");
if(remotetcps.isEmpty())
throw new IOException("发送错误");
boolean b;
synchronized (tc.getOutputcache()) {
b=!tc.getOutputcache().isEmpty()&&(tc.getOutputcount()-tc.getOutputcache().get(0).number>cacheblocks);
b=!tc.getOutputcache().isEmpty()&&(tc.getOutputcount()-tc.getOutputcache().get(0).number>tc.getOutputcachesize()/tc.getBlocksize());
}
if(!b) {
break;
}
Thread.sleep(1);
}
/*while(tc.getOutputcache().size()>cacheblocks) {
Thread.sleep(1);
}*/
if(!localtcps.containsKey(tc.getCuid()))
throw new IOException("本地连接已关闭");
if(remotetcps.isEmpty())
throw new IOException("发送错误");
while(tc.getOutputcache().size()>tc.getOutputcachesize()/tc.getBlocksize()) {
if(!localtcps.containsKey(tc.getCuid()))
throw new IOException("本地连接已关闭");
if(remotetcps.isEmpty())
throw new IOException("发送错误");
Thread.sleep(1);
}
ks.sendtime=System.nanoTime();
ks.sendtimeForRTT=ks.sendtime;
// ks.sendtimeForRTT=ks.sendtime;
//System.out.println(tc.getOutputcache().size());
/*if(ks.data==null) {
submitDataBlock(ks);
@@ -408,7 +480,9 @@ public class KLALBCore {
}*/
submitDataBlock(ks);
tc.getOutputcache().add(ks);
while( tc.getPeerCacheUsed()>2*cacheblocks) {
while( tc.isPeerFull()) {
if(!localtcps.containsKey(tc.getCuid()))
throw new IOException("本地连接已关闭");
if(remotetcps.isEmpty())
throw new IOException("发送错误");
Thread.sleep(1);
@@ -435,7 +509,7 @@ public class KLALBCore {
}
}*/
if (timex > (1<<Math.min(4,block.resend))*(i<remotetcps.size()?Consts.MAX_QUEUE_TIME:1000*i)) {
submitDataBlock(block);
submitDataBlock(block,2,4);
block.sendtime = System.nanoTime();
System.out.println("超时重传："+block+" 位置："+i);
block.resend++;
@@ -448,12 +522,12 @@ public class KLALBCore {
}
}
long curr=System.nanoTime();
if(tc.getSendDeque().size()<cacheblocks&&curr-tc.uackt>Consts.UACK_TIME) {
if(curr-tc.uackt>Consts.UACK_TIME) {
tc.uackt=curr;
KLALBBlock ack=new KLALBBlock();
ack.cuid=tc.getCuid();
ack.number=Long.MIN_VALUE;
ack.cacheused=tc.getSendDeque().size();
ack.cacheused=tc.getSendDeque().size()>tc.getInputcachesize()/tc.getBlocksize();
submitAckBlockNoDelay(ack,3);
}
/*if(tc.getSendDeque().size()<cacheblocks&&curr-tc.ukeept>Consts.KEEP_TIME) {
@@ -470,8 +544,10 @@ public class KLALBCore {
public KLALBBlock getDataBlock(LocalTCPConnection tc) throws InterruptedException, IOException {
KLALBBlock recv=null;
while(true) {
if(!localtcps.containsKey(tc.getCuid()))
throw new IOException("本地连接已关闭");
if(remotetcps.isEmpty())
throw new IOException("发送错误");
throw new IOException("接收错误");
recv=tc.getSendDeque().poll();
if(recv!=null)
break;
@@ -482,9 +558,76 @@ public class KLALBCore {
public void waitOutput(LocalTCPConnection tc) throws InterruptedException, IOException {
while(tc.getOutputcache().size()>0) {
if(!localtcps.containsKey(tc.getCuid()))
throw new IOException("本地连接已关闭");
if(remotetcps.isEmpty())
throw new IOException("发送错误");
Thread.sleep(10);
}
}
public void closeLocal(LocalTCPConnection tc) {
tc.getOutputcache().clear();
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 3;
submitDataBlockNoDelay(sbk);
localtcps.remove(tc.getCuid());
}
public void openLocal(LocalTCPConnection tc) {
localtcps.put(tc.getCuid(), tc);
}
public void connectLocal(LocalTCPConnection tc) throws InterruptedException, IOException {
KLALBBlock sbk = new KLALBBlock();
sbk.cuid = tc.getCuid();
sbk.number = 0;
sbk.command = 2;
sbk.lservice = tc.getServiceElement().name;
sbk.lipport = tc.getServiceElement().ipport;
submitDataBlockNoDelay(sbk);
while(!tc.isIssyk()) {
if(!localtcps.containsKey(tc.getCuid()))
throw new IOException("本地连接已关闭");
if(tc.isIsrst()) {
throw new ConnectException("连接被拒绝");
}
if(remotetcps.isEmpty())
throw new IOException("接收错误");
Thread.sleep(2);
}
}
public void openRemote(RemoteTCPConnection tc) {
remotetcps.add(tc);
}
public void closeRemote(RemoteTCPConnection tc) {
remotetcps.remove(tc);
List<Queue<KLALBBlock>> l=tc.getSendDequeList();
synchronized (l) {
for (int i = 0; i < l.size(); i++) {
Queue<KLALBBlock> q=l.get(i);
int ix=i-1;
if(ix<0) {
ix=0;
}
int i2=ix;
q.forEach((kv)->{
try {
submitDataBlock(kv,1,i2);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
});
}
}
}
}
src/org/kne/cloud/network/klalb/KLALBServer.java
+33 -13
View File
@@ -21,12 +21,14 @@ import org.kne.cloud.network.mport.ServiceElement;
import org.kne.cloud.network.mport.ThreadTool;
public class KLALBServer {
Map<UUID, IOThreadManager> whm=Collections.synchronizedMap( new WeakHashMap<UUID, IOThreadManager>());
Map<UUID, KLALBController> whm=new ConcurrentHashMap<>();
//Map<UUID, IOThreadManager> whm=Collections.synchronizedMap( new WeakHashMap<UUID, IOThreadManager>());
public KLALBServer(int port,Supplier<String> gjso) throws IOException {
TCPListener tcpl=new TCPListener(port);
tcpl.setCon((s)->{
RemoteTCPConnection tcc=null;
KLALBController nx=null;
UUID uid=null;
try {
//s.setSoTimeout(10000);
tcc=new RemoteTCPConnection(null,s);
@@ -46,37 +48,50 @@ public class KLALBServer {
Tunnel tll=new Tunnel(name,new IPPort( ip, portx));
tcc.setTunnel(tll);
UUID uid=new UUID(din.readLong(),din.readLong());
uid=new UUID(din.readLong(),din.readLong());
tcc.handshake(10);
IOThreadManager nx=whm.computeIfAbsent(uid, (kuid)->{
IOThreadManager iom=new IOThreadManager();
iom.getCore().setAcceptSYN((b)->{
synchronized (whm) {
nx=whm.computeIfAbsent(uid, (kuid)->{
KLALBController iom=new KLALBController();
iom.getCore().setAcceptSYN((b,bol)->{
try {
STJson stj=new STJson(gjso.get());
if(!stj.checkSafety(b.lipport)) {
System.out.println("未授权的请求:"+b.lipport);
return false;
bol.accept(false);
return;
}
Socket soc=new Socket(b.lipport.getIp(),b.lipport.getPort());
LocalTCPConnection ltc=new LocalTCPConnection(soc, b.cuid);
ThreadTool.makeVThreadIfSupport("局域网连接监视线程", ()->{
LocalTCPConnection ltc=null;
try {
Socket soc=new Socket(b.lipport.getIp(),b.lipport.getPort());
ltc=new LocalTCPConnection(soc, b.cuid);
bol.accept(true);
iom.handleLocal(ltc, false);
return;
} catch (IOException e) {
e.printStackTrace();
}finally {
if(ltc!=null)
ltc.close();
}
}).start();
return true;
bol.accept(false);
return ;
} catch (IOException e) {
System.out.println("连接本地服务"+b.lipport+"失败，请检查你的本地服务");
e.printStackTrace();
}
return false;
bol.accept(false);
return;
});
return iom;
});
}
/*if(whm.containsKey(uid)) {
nx=whm.get(uid);
}else {
@@ -113,6 +128,11 @@ public class KLALBServer {
}finally {
if(tcc!=null)
tcc.close();
if(nx.getCore().getRemotetcps().isEmpty()) {
nx.close();
if(uid!=null)
whm.remove(uid);
}
}
});
tcpl.open();
src/org/kne/cloud/network/klalb/LocalTCPConnection.java
+66 -8
View File
@@ -18,9 +18,14 @@ private int cuid;
private ServiceElement serviceElement;
private Set<KLALBBlock> inputcache = new TreeSet<>();
//private Set<KLALBBlock> inputcache = Collections.synchronizedSet( new TreeSet<>());
private List<KLALBBlock>inputcache=new Vector<>();
private List<KLALBBlock> outputcache = new Vector<>();
private volatile int blocksize=8192;
private volatile int inputcachesize=100*blocksize;
private volatile int outputcachesize=100*blocksize;
private volatile long inputcount = 1;
private volatile long outputcount = 1;
public LocalTCPConnection(Socket s) throws IOException {
@@ -33,6 +38,24 @@ private volatile long outputcount = 1;
s.setKeepAlive(true);
cuid=uid;
}
public int getBlocksize() {
return blocksize;
}
public void setBlocksize(int blocksize) {
this.blocksize = blocksize;
}
public int getInputcachesize() {
return inputcachesize;
}
public void setInputcachesize(int inputcachesize) {
this.inputcachesize = inputcachesize;
}
public int getOutputcachesize() {
return outputcachesize;
}
public void setOutputcachesize(int outputcachesize) {
this.outputcachesize = outputcachesize;
}
public int getCuid() {
return cuid;
}
@@ -47,7 +70,7 @@ private volatile long outputcount = 1;
}
public Set<KLALBBlock> getInputcache() {
public List<KLALBBlock> getInputcache() {
return inputcache;
}
public List<KLALBBlock> getOutputcache() {
@@ -60,7 +83,7 @@ private volatile long outputcount = 1;
return outputcount;
}
public void unpackBlock(KLALBBlock data) throws IOException {
if(data.number<inputcount) {
/*if(data.number<inputcount) {
return;
}
inputcache.add(data);
@@ -74,10 +97,12 @@ private volatile long outputcount = 1;
}else {
break;
}
}
}*/
getDout().write(data.data);
getDout().flush();
}
public KLALBBlock packBlock() throws IOException {
byte[]dat=new byte[Consts.BLOCKSIZE];
byte[]dat=new byte[blocksize];
int len=getDin().read(dat);
KLALBBlock pb=new KLALBBlock();
@@ -96,21 +121,54 @@ private volatile long outputcount = 1;
public BlockingDeque<KLALBBlock> getSendDeque() {
return sendDeque;
}
private volatile int pcu=0;
//private volatile int pcu=0;
private volatile Consumer<LocalTCPConnection> hook;
public volatile long uackt=System.nanoTime();
public volatile long ukeept=System.nanoTime();
public void setPeerCacheUsed(int cacheused) {
/*public void setPeerCacheUsed(int cacheused) {
pcu= cacheused;
}
public int getPeerCacheUsed() {
return pcu;
}
}*/
public Consumer<LocalTCPConnection> getRSTHook() {
return hook;
}
public void setRSTHook(Consumer<LocalTCPConnection> tc) {
this.hook=tc;
}
private volatile boolean pfl=false;
public void setPeerFull(boolean b) {
pfl=b;
}
public boolean isPeerFull() {
return pfl;
}
/*public int getSendDequeBytes() {
AtomicInteger a=new AtomicInteger(0);
getSendDeque().forEach((v)->{
if(v.data!=null)
a.addAndGet(v.data.length);
});
return a.get();
}*/
public void incInputcount() {
inputcount++;
}
private volatile boolean issyk=false;
public boolean isIssyk() {
return issyk;
}
public void setIssyk(boolean issyk) {
this.issyk = issyk;
}
private volatile boolean isrst=false;
public boolean isIsrst() {
return isrst;
}
public void setIsrst(boolean isrst) {
this.isrst = isrst;
}
}
src/org/kne/cloud/network/klalb/RemoteTCPConnection.java
+88 -25
View File
@@ -10,12 +10,17 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.io.StreamCorruptedException;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.UUID;
import java.util.Vector;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
@@ -124,13 +129,18 @@ public class RemoteTCPConnection extends TCPConnection {
public double getPredictedLatency() {
return latency;
}
public void predictLatency() {
public void predictLatency(int prio) {
AtomicLong data=new AtomicLong(0);
sendDeque.forEach((e)->{
for (int i = 0; i <=prio; i++) {
Queue<KLALBBlock> sendDequev=sendDequeList.get(i);
sendDequev.forEach((e)->{
data.addAndGet(12);
if(e.data!=null) {
data.addAndGet(e.data.length);
}
});
}
latency= delay/2000000.0+data.get()*1000.0/upsm;
}
@@ -143,7 +153,7 @@ public class RemoteTCPConnection extends TCPConnection {
super(s);
tunnel=t;
initIOs();
s.setSoTimeout(Consts.SO_TIMEOUT);
s.setSoTimeout(Consts.PRE_SO_TIMEOUT);
}
private long delay = Long.MAX_VALUE;
@@ -209,11 +219,11 @@ public class RemoteTCPConnection extends TCPConnection {
}
} else if (data.number == 0) {
dout.write(data.command);
if(data.command!=0&&data.command!=1) {
/*if(data.command!=0&&data.command!=1) {
dout.writeLong(data.sn);
}else {
}
}*/
switch (data.command) {
case 0:
case 1:
@@ -227,7 +237,7 @@ public class RemoteTCPConnection extends TCPConnection {
break;
}
} else {
dout.writeInt(data.cacheused);
dout.writeBoolean(data.cacheused);
}
if(isflush)
@@ -262,11 +272,11 @@ public class RemoteTCPConnection extends TCPConnection {
}
} else if (klb.number == 0) {
klb.command = din.read();
if(klb.command!=0&&klb.command!=1) {
/*if(klb.command!=0&&klb.command!=1) {
klb.sn = din.readLong();
}else {
klb.sn=0;
}
}*/
switch (klb.command) {
case 0:
case 1:
@@ -280,7 +290,7 @@ public class RemoteTCPConnection extends TCPConnection {
break;
}
} else {
klb.cacheused = din.readInt();
klb.cacheused = din.readBoolean();
}
System.out.println("RECEIVE:" + klb);
@@ -299,29 +309,35 @@ public class RemoteTCPConnection extends TCPConnection {
}
}
private Queue<KLALBBlock> NDsendDeque = new ConcurrentLinkedQueue<KLALBBlock>();
private List<Queue<KLALBBlock>> sendDequeList =new Vector<Queue<KLALBBlock>>();
{
for(int i=0;i<11;i++) {
sendDequeList.add(new ConcurrentLinkedQueue<KLALBBlock>());
}
}
public List<Queue<KLALBBlock>> getSendDequeList() {
return sendDequeList;
}
private Object lock=new Object();
public void addBlock(KLALBBlock blk,int prio) {
sendDequeList.get(prio).add(blk);
synchronized (lock) {
lock.notifyAll();
}
}
/* private Queue<KLALBBlock> NDsendDeque = new ConcurrentLinkedQueue<KLALBBlock>();
public Queue<KLALBBlock> getNDSendDeque() {
return NDsendDeque;
}
//private Queue<KLALBBlock> sendDeque = new ConcurrentLinkedQueue<KLALBBlock>();
private Queue<KLALBBlock> sendDeque = new PriorityBlockingQueue<KLALBBlock>(2, new Comparator<KLALBBlock>() {
@Override
public int compare(KLALBBlock o1, KLALBBlock o2) {
if(o1.resend>o2.resend) {
return -1;
}else if(o1.resend<o2.resend) {
return 1;
}else {
return 0;
}
}
});
private Queue<KLALBBlock> sendDeque = new ConcurrentLinkedQueue<KLALBBlock>();
public Queue<KLALBBlock> getSendDeque() {
return sendDeque;
}
}*/
public void handshake(int x) throws IOException {
for(int i=0;i<x;i++) {
@@ -338,4 +354,51 @@ public class RemoteTCPConnection extends TCPConnection {
}
public void lock(long to) throws InterruptedException {
synchronized (lock) {
lock.wait(to);
}
}
public void ackRemove(long number, int cuid) {
for (Iterator<Queue<KLALBBlock>> iterator = sendDequeList.iterator(); iterator.hasNext();) {
Queue<KLALBBlock> queue = iterator.next();
queue.removeIf((vb)->{
return vb.number==number&&vb.cuid==cuid;
});
}
}
public boolean isEmpty() {
synchronized (sendDequeList) {
for (Iterator<Queue<KLALBBlock>> iterator = sendDequeList.iterator(); iterator.hasNext();) {
Queue<KLALBBlock> queue = (Queue<KLALBBlock>) iterator.next();
if(!queue.isEmpty()) {
return false;
}
}
}
return true;
}
public KLALBBlock getNextBlock() {
synchronized (sendDequeList) {
for (Iterator<Queue<KLALBBlock>> iterator = sendDequeList.iterator(); iterator.hasNext();) {
Queue<KLALBBlock> queue = (Queue<KLALBBlock>) iterator.next();
KLALBBlock v=queue.poll();
if(v!=null)
return v;
}
}
return null;
}
public void setSoTimeout(int soTimeout) {
try {
connect.setSoTimeout(soTimeout);
} catch (SocketException e) {
//e.printStackTrace();
}
}
}
src/org/kne/cloud/network/nathole/TCPNatHoleTeat.java
+67
View File
@@ -0,0 +1,67 @@
package org.kne.cloud.network.nathole;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
public class TCPNatHoleTeat {
public static void main(String[] args) throws IOException {
x("183.198.152.240", 12187);
}
private static void x(String nat, int port) throws IOException {
for (int i = 10000; i < 20000; i++) {
System.out.println("try" + i);
Socket s = null;
ServerSocket srs = null;
try {
s = new Socket();
s.bind(new InetSocketAddress(InetAddress.getByName("0.0.0.0"), i));
try {
s.connect(new InetSocketAddress(InetAddress.getByName("1.1.1.1"), 443), 1);
}catch(SocketTimeoutException e) {
}
s.close();
srs = new ServerSocket();
srs.setReuseAddress(true);
srs.bind(new InetSocketAddress("0.0.0.0", i));
ServerSocket srs2 = srs;
new Thread(() -> {
try {
while (true) {
Socket sa = srs2.accept();
System.out.println(sa.toString());
//sa.close();
// System.exit(0);
}
} catch (IOException e) {
// e.printStackTrace();
}
}).start();
/*Socket ste = new Socket();
try {
ste.connect(new InetSocketAddress(nat, port), 1);
} catch (SocketTimeoutException e) {
}
ste.close();*/
} catch (IOException e) {
e.printStackTrace();
} finally {
if (s != null)
s.close();
/*if (srs != null)
srs.close();*/
}
}
}
}